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

 #!/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-urle<b style="color:transparent">来源gao@dai!ma.com搞$代^码网</b>ncoded; 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

 #!/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)