data-query-agent/.agents/skills/query-user-behavior-path/scripts/user_timeline.py

#!/usr/bin/env python3
# coding=utf-8
"""单用户完整行为时间线：六张离线日志合并为 Excel。"""
import argparse
from datetime import datetime
from pathlib import Path
import re

import pandas as pd

from odps_module import ODPSClient


EXCLUDED_BUSINESSTYPES = {
    "deviceId", "openGIdSuccess", "buttonView", "systemInfo", "videoPlayCancel", "windowView"
}


def sql_text(value):
    return str(value).replace("'", "''")


def build_apptype_filter(value):
    return "" if value is None else f" AND apptype='{sql_text(value)}'"


def safe_name(value):
    return re.sub(r"[^A-Za-z0-9_-]", "_", str(value))


SQL_ALL = """
WITH v AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'video' AS 来源,
           businesstype, CAST(NULL AS STRING) AS eventid, machineinfo_system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, videoid AS 视频id,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.video_action_log_applet
    WHERE dt='{day}'{apptype_filter} AND mid='{mc}'
      AND businesstype <> 'videoPreView'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
a AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'ad' AS 来源,
           businesstype, CAST(eventid AS STRING) AS eventid, GET_JSON_OBJECT(machineinfo, '$.system') AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid,
           CAST(NULL AS STRING) AS isAdPlaying, creativecode AS creativeCode, headvideoid AS 视频id,
           hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.ad_action_log_own
    WHERE dt='{day}'{apptype_filter} AND machinecode='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
p AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'play' AS 来源,
           businesstype, CAST(eventid AS STRING) AS eventid, machineinfo_system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, videoid AS 视频id,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.video_play_log
    WHERE dt='{day}'{apptype_filter} AND mid='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
s AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'simpleevent' AS 来源,
           businesstype, CAST(eventid AS STRING) AS eventid, system AS 系统, pagesource, endroutepath AS endRoutePath, objecttype,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid,
           GET_JSON_OBJECT(extparams, '$.isAdPlaying') AS isAdPlaying,
           GET_JSON_OBJECT(extparams, '$.creativeCode') AS creativeCode, videoid AS 视频id,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.simpleevent_log
    WHERE dt='{day}'{apptype_filter} AND machinecode='{mc}'
      AND (businesstype IS NULL OR businesstype <> 'openGIdError')
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
u AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'useractive' AS 来源,
           businesstype, CAST(eventid AS STRING) AS eventid, system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, CAST(NULL AS STRING) AS 视频id,
           CAST(NULL AS STRING) AS hotsencetype, path, subsessionid, sessionid
    FROM loghubods.useractive_log
    WHERE dt='{day}'{apptype_filter} AND machinecode='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
r AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'share' AS 来源,
           type AS businesstype, CAST(eventid AS STRING) AS eventid, CAST(NULL AS STRING) AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype,
           topic, shareid,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, CAST(NULL AS STRING) AS 视频id,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.user_share_log
    WHERE dt='{day}'{apptype_filter} AND machinecode='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
t AS (
    SELECT * FROM v UNION ALL SELECT * FROM a UNION ALL SELECT * FROM p
    UNION ALL SELECT * FROM s UNION ALL SELECT * FROM u UNION ALL SELECT * FROM r
)
SELECT t.ts, t.产品apptype, t.来源, t.businesstype, t.eventid, t.系统, t.endRoutePath, t.objecttype, t.pagesource, t.topic, t.shareid,
       t.isAdPlaying, t.creativeCode, t.视频id, b.title AS 视频标题,
       t.hotsencetype, t.path, t.subsessionid, t.sessionid
FROM t
LEFT JOIN videoods.dim_video b ON t.视频id = b.videoid
ORDER BY t.ts
"""


def text(value):
    return "" if pd.isna(value) else str(value)


def normalize_device_system(value):
    value = text(value).lower()
    if "ios" in value or "iphone" in value or "ipad" in value:
        return "iOS"
    if "android" in value:
        return "Android"
    return ""


def add_device_info(df, group_column=None):
    df["机型信息"] = df["系统"].map(normalize_device_system)
    if group_column is None:
        known = df.loc[df["机型信息"] != "", "机型信息"].drop_duplicates()
        if len(known) == 1:
            df.loc[df["机型信息"] == "", "机型信息"] = known.iloc[0]
    else:
        known = df.loc[df["机型信息"] != "", [group_column, "机型信息"]].drop_duplicates()
        systems = known.groupby(group_column)["机型信息"].agg(lambda values: values.iloc[0] if len(values) == 1 else "")
        df.loc[df["机型信息"] == "", "机型信息"] = df.loc[df["机型信息"] == "", group_column].map(systems).fillna("")
    df = df.drop(columns="系统")
    df.insert(df.columns.get_loc("来源") + 1, "机型信息", df.pop("机型信息"))
    return df


EVENTID_BEHAVIOR = {
    "107001": "视频封面加载完成", "107002": "视频封面加载失败", "130010": "广告组件加入页面",
    "22022221": "详情接口请求成功", "22022222": "详情接口请求失败", "550001": "用户截图",
}


def behavior_definition(row):
    source = text(row["来源"])
    businesstype = text(row["businesstype"])
    pagesource = text(row["pagesource"])
    eventid = text(row.get("eventid", ""))
    objecttype = text(row.get("objecttype", ""))
    topic = text(row.get("topic", ""))
    is_head_video_page = pagesource.endswith("user-videos-share")
    if source == "video" and is_head_video_page:
        return {"videoView": "头部视频曝光", "videoPlay": "头部视频播放"}.get(businesstype, "")
    if source == "ad":
        return {"adRequest": "广告请求", "adLoaded": "广告加载", "adView": "广告曝光", "adPlay": "广告播放", "adCloseBtnTap": "广告关闭"}.get(businesstype, "")
    if source == "play":
        return {"videoPlaySuccess": "播放成功", "videoPlaySlow": "播放卡顿", "videoRealPlay": "有效播放"}.get(businesstype, "")
    if source == "share" and topic == "click":
        return "点击卡片"
    if source == "useractive" and businesstype == "path":
        return "打开应用"
    if source == "simpleevent":
        if businesstype == "buttonClick":
            definition = {"videoBackIcon": "点击视频页返回图标", "weapp_quitbutton": "从分享视频页返回首页"}.get(objecttype, "")
            if definition:
                return definition
        if businesstype == "pageView" and pagesource.endswith("category_55"):
            return "首页分类页曝光（分类55）"
        if is_head_video_page:
            definition = {"pageView": "视频分享页曝光", "detailRequest": "视频分享页详情接口请求成功", "userCaptureScreen": "用户在视频分享页截图"}.get(businesstype, "")
            if definition:
                return definition
        definition = {"userPause": "视频暂停", "userActiveEnd": "小程序进入后台", "userActiveStart": "小程序进入前台", "decideSharePageJump": "准备跳转", "jumpSwiperPage": "跳转到沉浸式"}.get(businesstype, "")
        if definition:
            return definition
    return EVENTID_BEHAVIOR.get(eventid, "")


def main():
    parser = argparse.ArgumentParser(description="查询单用户离线行为时间线")
    parser.add_argument("user_id", help="machinecode/mid")
    parser.add_argument("date", help="yyyyMMdd")
    parser.add_argument("apptype", nargs="?", default=None, help="可选；不传则查询全部产品")
    parser.add_argument("--output-dir", type=Path, default=Path("."))
    args = parser.parse_args()
    datetime.strptime(args.date, "%Y%m%d")
    df = ODPSClient().execute_sql(SQL_ALL.format(mc=sql_text(args.user_id), day=args.date, apptype_filter=build_apptype_filter(args.apptype)))
    df = df[~df["businesstype"].isin(EXCLUDED_BUSINESSTYPES)].sort_values("ts", kind="stable").reset_index(drop=True)
    df = df.rename(columns={"endroutepath": "endRoutePath", "isadplaying": "isAdPlaying", "creativecode": "creativeCode"})
    df.insert(0, "北京时间", pd.to_datetime(df["ts"], unit="ms", utc=True).dt.tz_convert("Asia/Shanghai").dt.tz_localize(None))
    df = add_device_info(df.drop(columns="ts"))
    df.insert(df.columns.get_loc("businesstype") + 1, "行为", df.apply(behavior_definition, axis=1))
    df = df.rename(columns={"businesstype": "事件类型", "eventid": "事件ID"})
    df.insert(0, "用户ID", args.user_id)
    df.insert(0, "产品apptype", df.pop("产品apptype"))
    args.output_dir.mkdir(parents=True, exist_ok=True)
    out = args.output_dir / f"timeline_{safe_name(args.user_id[-12:])}_{args.date}.xlsx"
    with pd.ExcelWriter(out, engine="openpyxl", datetime_format="yyyy/mm/dd hh:mm:ss") as writer:
        df.to_excel(writer, sheet_name="行为路径", index=False)
        for cell in writer.sheets["行为路径"]["C"][1:]:
            cell.number_format = "yyyy/mm/dd hh:mm:ss"
    counts = df["来源"].value_counts().to_dict()
    for source in ["video", "ad", "play", "simpleevent", "useractive", "share"]:
        print(f"[ROWS] {source}={counts.get(source, 0)}", flush=True)
    print(f"[XLSX] 日期={args.date} 事件数={len(df)} -> {out.resolve()}", flush=True)


if __name__ == "__main__":
    main()