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

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

import requests
import json

#钉钉告警函数
def dingtalk(content):
  dingtalk_url = "https://oapi.dingtalk.com/robot/send&＃63;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

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