Python爬取阿拉丁统计信息过程图解

背景

目前项目在移动端上，首推使用微信小程序。各项目的小程序访问数据有必要进行采集入库，方便后续做统计分析。虽然阿拉丁后台也提供了趋势分析等功能，但一个个的获取数据做数据分析是很痛苦的事情。通过将数据转换成sql持久化到数据库上，为后面的数据分析和展示提供了基础。

实现思路

阿拉丁产品分开放平台和统计平台两个产品线，目前开放平台有api及配套的文档。统计平台api需要收费，而且贼贵。既然没有现成的api可以获取数据，那么我们尝试一下用python抓取页面上的数据，毕竟python擅长干这种事情。

获取数据流程

1、首先登陆阿拉丁的统计平台，如下图

发现实际需要获取的关键数据主要是“新用户数”、“访问人数”、“访问次数”、“打开次数”、“次均停留时长”、“跳出率”、“累计用户”，而且这里有现成的昨日数据，那么我们一天爬一次这个页面然后格式化昨日数据入库就可以了。

2、打开浏览器F12，发现这个页面的数据主要来自链接如下图

这里隐去敏感信息token，默认是20条数据进行分页显示，那么问题来了，这里的token从哪里获取的呢？猜测这个token应该是用户登录之后生成的，只要有正确的token，就可以获取数据。

在preview标签中可以看到实际返回了50条的数据

展开详细的数据，这里隐去敏感信息app_key和app_name

展开第1条数据，可以看到详细的统计数据信息，现在我们的目标就是如何通过python获取这些信息。

登录过程

下面我们来看一下登录过程。浏览器退出账号之后，重新登录（账密方式登录），打开浏览器F12，发现实际登录请求的url地址如下图

注意content-type和user-agent参数，这里隐去敏感数据用户名、密码和secretkey, 这个secretkey又是哪里冒出来的呢？

不管三七二十一，先用postman请求一下这个地址，的确获取到了token。

前面登录url的payload中有code字段，实际用postman请求发现可以忽略。通过F12发现，验证码的url是通过拼接实现的，具体如下图。

如果验证码是强制需要的，我们可以先拼接好url获取图片，通过pytesseract进行识别，当然这里面可能存在准确率问题，不过既然这个目前是非必须的，咋们就忽略吧。

网页上退出登录，发现页面请求了token地址如下

直接用浏览器访问，可以看到secretKey

梳理一下：

1、目前我们已经有了登录的url、获取统计数据的url、用户名密码、secretKey等信息，code可忽略

2、登录完可以获取到token信息

3、可以通过token请求获取统计数据的url来获取数据

代码实现

1、login.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import requests
import json

#钉钉告警函数
def dingtalk(content):
  dingtalk_url = \'https://oapi.dingtalk.com/robot/send?access_token=\"请输入钉钉token\"\'
  dingtalk_header ={\"Content-Type\": \"application/json\"}
  dingtalk_payload = {\"msgtype\": \"text\",\"text\": {\"content\": \"%s\" %content}}
  requests.post(dingtalk_url, data=json.dumps(dingtalk_payload), headers=dingtalk_header)

#获取secretkey函数
def get_secretkey():
  token_url = \'http://betaapi.aldwx.com/m/Login_reg/Login/token\'
  header = {\"Content-Type\": \"application/x-www-form-urlencoded; charset=UTF-8\",
       \"user-agent\": \"User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36\"}
  req = requests.post(token_url,headers=header).text
  return json.loads(req).get(\"secretKey\")

#获取token函数
def get_token(secretkey):
  s = requests.Session()
  login_url = \'https://betaapi.aldwx.com/Main/action/Login_reg/Login/login\'
  header = {\"Content-Type\": \"application/x-www-form-urlencoded; charset=UTF-8\",
       \"user-agent\": \"User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36\"}

  payload = {\"phone\": \"用户名\",
        \"password\": \"密码\",
        \"source\": \"0\",
        \"plan\": \"0\",
        \"creative\": \"0\",
        \"keyword1\": \"0\",
        \"secretKey\": secretkey}
  req = s.post(login_url, data=payload, headers=header).text
  if json.loads(req).get(\"code\") == 200:
    return json.loads(req).get(\"data\").get(\"token\")
  else:
    dingtalk(\"获取阿拉丁登录token失败，请检查!\")
    return None

2、aldwx.py

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import requests
import json
from common.mysql_conn import DBAPI
from conf import settings
from login import get_token,get_secretkey,dingtalk

#定义写数据库函数
def exec_sql(sql):
  my_conn = DBAPI(settings.Params[\'host\'], settings.Params[\'user\'], settings.Params[\'password\'], int(settings.Params[\'port\']), settings.Params[\'database\'], settings.Params[\'charset\'])
  my_conn.conn_dml(sql)

#获取统计数据拼接成列表
def get_data(token):
  header = {\"Content-Type\": \"application/x-www-form-urlencoded\",
       \"user-agent\": \"User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36\"}
  url = \'http://betaapi.aldwx.com/upgrade/api/applet_homepage\'
  payload = {\'currentpage\': \'1\', \'total\': \'80\', \'token\': token, \'appkey\': \'\', \'is_demo\': \'0\'}
  try:
      req = requests.post(url, data=payload, headers=header).text
      data = (json.loads(req).get(\"data\"))
      sql_value = []
      for i in data:
        app_name = i.get(\'app_name\')
        yesterday_data = i.get(\'countList\')[1]
        create_time = yesterday_data.get(\'day\')
        new_comer_count = yesterday_data.get(\'new_comer_count\')
        visitor_count = yesterday_data.get(\'visitor_count\')
        open_count = yesterday_data.get(\'open_count\')
        total_page_count = yesterday_data.get(\'total_page_count\')
        secondary_avg_stay_time = yesterday_data.get(\'secondary_avg_stay_time\')
        bounce_rate = yesterday_data.get(\'bounce_rate\')
        total_visitor_count = yesterday_data.get(\'total_visitor_count\')
        value = \"(\\\"%s\\\",\\\"%s\\\",\\\"%s\\\",\\\"%s\\\",\\\"%s\\\",\\\"%s\\\",\\\"%s\\\",\\\"%s\\\",\\\"%s\\\")\" % (app_name, create_time, new_comer_count, visitor_count, open_count, total_page_count,secondary_avg_stay_time,bounce_rate, total_visitor_count)
        sql_value.append(value)
      if sql_value != \'None\':
        return sql_value
      else:
        dingtalk(\"获取微信小程序统计信息失败,token: %s\" %token)
  except Exception as e:
    dingtalk(\"获取微信小程序%s统计信息失败，%s\" %e)

if __name__ == \'__main__\':
  secretkey=get_secretkey()
  token = get_token(secretkey)
  sql = \"\"\"INSERT INTO operations_db.aldwx_stat (APP_NAME,CREATE_TIME,NEW_COMER_COUNT,VISITOR_COUNT,OPEN_COUNT,TOTAL_PAGE_COUNT,SECONDARY_AVG_STAY_TIME,BOUNCE_RATE,TOTAL_VISITOR_COUNT) VALUES\"\"\"
  value = get_data(token)
  #拼接并转换sql
  value = str(value).strip(\"\'[\").strip(\"]\'\").strip().replace(\"\', \'\",\",\")
  sql = \"%s%s;\" % (sql, value)
  print(sql)
  exec_sql(sql)

3、入库效果