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、入庫效果