纯技术干货|Python 自动化获取 APP 日活数据 + 钉钉推送(附完整代码+逐段讲解)

#!/usr/local/python39/bin/python3import timefrom datetime import datetimeimport requestsimport jsonimport hmacimport hashlibimport base64import urllib.parseimport datetimefrom decimal import Decimal# -------------------------- 钉钉消息推送（加密签名） --------------------------def send_dd_message(msg):    timestamp = str(round(time.time() * 1000))    secret = '你的钉钉密钥'  # 替换成自己的钉钉机器人密钥    secret_enc = secret.encode('utf-8')    string_to_sign = '{}\n{}'.format(timestamp, secret)    string_to_sign_enc = string_to_sign.encode('utf-8')    # HMAC-SHA256 加密 + Base64 编码 + URL 转义    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))    # 拼接钉钉推送地址（替换自己的 access_token）    url = 'https://oapi.dingtalk.com/robot/send?access_token=你的access_token&timestamp=' + timestamp + '&sign=' + sign    HEADERS = {"Content-Type": "application/json;charset=utf-8"}    # 构造文本消息体    String_textMsg = {        "msgtype": "text",        "text": {"content": msg}    }    # 转JSON格式并推送    String_textMsg = json.dumps(String_textMsg).encode('utf-8')    requests.post(url, data=String_textMsg, headers=HEADERS)# -------------------------- 获取 APP 日活、版本数据 --------------------------def get_ylxf_app():    # 导入友盟 API SDK 并配置    import aop    import aop.api    aop.set_default_server('gateway.open.umeng.com')  # 友盟 API 网关地址（固定）    aop.set_default_appinfo(8****55, "***UMno")  # 替换成自己的友盟 appId 和密钥    # 配置 Android + iOS 双端 appkey（替换成自己的）    datas = [        ['Android', '安卓appkey'],        ['ios', 'iOSappkey'],    ]    today_activityUsers = 0  # 今日总活跃数    yesterday_activityUsers = 0  # 昨日总活跃数    today = datetime.datetime.now()    now = today.strftime('%Y-%m-%d %H:%M:%S')  # 当前时间戳，用于推送文案    # 1. 获取双端今日/昨日活跃用户    req = aop.api.UmengUappGetTodayYesterdayDataRequest()    for data in datas:        try:            # 调用友盟接口，获取单端数据            resp = req.get_response(None, appkey=data[1])            # 累加双端今日、昨日活跃数            today_activityUsers += int(resp['todayData']['activityUsers'])            yesterday_activityUsers += int(resp['yesterdayData']['activityUsers'])        except Exception as e:            print(e)  # 异常捕获，避免脚本崩溃    # 2. 高精度计算环比涨跌（保留2位小数）    activityUsers_per = (Decimal(today_activityUsers - yesterday_activityUsers) / Decimal(yesterday_activityUsers)).quantize(Decimal('0.00'))    # 3. 拼接推送文案    text = f"截止：{now}"    text += f"\nAPP日活跃数 {today_activityUsers} 个（昨日：{yesterday_activityUsers} 个），"    # 自动判断增长/下降    if activityUsers_per > Decimal(0):        text += f"活跃用户环比增长 {activityUsers_per} %；"    else:        text += f"活跃用户环比下降 {abs(activityUsers_per)} %；"    # 4. 统计指定版本累计用户（可自定义版本号）    versions = ['2.4.0', '3.0.0']  # 替换成自己要监控的版本    req = aop.api.UmengUappGetVersionDataRequest()    for version in versions:        version_total_user = 0  # 该版本双端累计用户        for data in datas:            try:                # 调用友盟版本分布接口，获取当日版本数据                resp = req.get_response(None, appkey=data[1], date=f'{today.date()}')                # 匹配目标版本，累加用户数                for versioninfo in resp['versionInfos']:                    if versioninfo['version'] == version:                        version_total_user += int(versioninfo['totalUser'])            except Exception as e:                print(e)        text += f"\nAPP {version} 版本累计用户：{version_total_user} 人；"    return text# -------------------------- 主执行入口 --------------------------if __name__ == "__main__":    # 获取推送文案    msg = f'\n\n1.{get_ylxf_app()}'    msg += f'\n（按添加时间统计）'    # 推送到钉钉    send_dd_message(msg)    # 控制台打印日志，方便调试    print(msg)

def send_dd_message(msg):    timestamp = str(round(time.time() * 1000))    secret = '你的钉钉密钥'    secret_enc = secret.encode('utf-8')    string_to_sign = '{}\n{}'.format(timestamp, secret)    string_to_sign_enc = string_to_sign.encode('utf-8')    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))    url = 'https://oapi.dingtalk.com/robot/send?access_token=你的access_token&timestamp=' + timestamp + '&sign=' + sign    HEADERS = {"Content-Type": "application/json;charset=utf-8"}    String_textMsg = {        "msgtype": "text",        "text": {"content": msg}    }    String_textMsg = json.dumps(String_textMsg).encode('utf-8')    requests.post(url, data=String_textMsg, headers=HEADERS)

import aopimport aop.apiaop.set_default_server('gateway.open.umeng.com')aop.set_default_appinfo(8689055, "zwXyKehUMno")datas = [    ['Android', '安卓appkey'],    ['ios', 'iOSappkey'],]

req = aop.api.UmengUappGetTodayYesterdayDataRequest()for data in datas:    try:        resp = req.get_response(None, appkey=data[1])        today_activityUsers += int(resp['todayData']['activityUsers'])        yesterday_activityUsers += int(resp['yesterdayData']['activityUsers'])    except Exception as e:        print(e)

activityUsers_per = (Decimal(today_activityUsers - yesterday_activityUsers) / Decimal(yesterday_activityUsers)).quantize(Decimal('0.00'))text = f"截止：{now}"text += f"\nAPP日活跃数 {today_activityUsers} 个（昨日：{yesterday_activityUsers} 个），"if activityUsers_per > Decimal(0):    text += f"活跃用户环比增长 {activityUsers_per} %；"else:    text += f"活跃用户环比下降 {abs(activityUsers_per)} %；"

versions = ['2.4.0', '3.0.0']req = aop.api.UmengUappGetVersionDataRequest()for version in versions:    version_total_user = 0    for data in datas:        try:            resp = req.get_response(None, appkey=data[1], date=f'{today.date()}')            for versioninfo in resp['versionInfos']:                if versioninfo['version'] == version:                    version_total_user += int(versioninfo['totalUser'])        except Exception as e:            print(e)    text += f"\nAPP {version} 版本累计用户：{version_total_user} 人；"

if __name__ == "__main__":    msg = f'\n\n1.{get_ylxf_app()}'    msg += f'\n（按添加时间统计）'    send_dd_message(msg)    print(msg)