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

#!/usr/bin/env python3
# coding=utf-8
"""单用户实时行为时间线：当前可用实时/5分钟日志。"""
import argparse
from datetime import datetime
from pathlib import Path

import pandas as pd

from odps_module import ODPSClient
from user_timeline import EXCLUDED_BUSINESSTYPES, add_device_info, behavior_definition, build_apptype_filter, safe_name, sql_text


SQL_REALTIME = """
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, videoid AS 视频id,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid, CAST(NULL AS STRING) AS creativeCode,
           CAST(NULL AS STRING) AS rootPageSource, CAST(NULL AS STRING) AS objectId, CAST(NULL AS STRING) AS targetUid,
           CAST(NULL AS STRING) AS networkType, CAST(NULL AS STRING) AS operationOptions, CAST(NULL AS STRING) AS operationExtParams,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.video_action_log_per5min
    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}'{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, eventid, GET_JSON_OBJECT(machineinfo, '$.system') AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype, headvideoid AS 视频id,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid, creativecode AS creativeCode,
           CAST(NULL AS STRING) AS rootPageSource, CAST(NULL AS STRING) AS objectId, CAST(NULL AS STRING) AS targetUid,
           CAST(NULL AS STRING) AS networkType, CAST(NULL AS STRING) AS operationOptions, CAST(NULL AS STRING) AS operationExtParams,
           hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.ad_action_log_own_per5min
    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}'{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, eventid, machineinfo_system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype, videoid AS 视频id,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid, CAST(NULL AS STRING) AS creativeCode,
           CAST(NULL AS STRING) AS rootPageSource, CAST(NULL AS STRING) AS objectId, CAST(NULL AS STRING) AS targetUid,
           CAST(NULL AS STRING) AS networkType, CAST(NULL AS STRING) AS operationOptions, CAST(NULL AS STRING) AS operationExtParams,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.video_play_log_per5min
    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}'{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, eventid, system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, objecttype, videoid AS 视频id,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid, CAST(NULL AS STRING) AS creativeCode,
           CAST(NULL AS STRING) AS rootPageSource, CAST(NULL AS STRING) AS objectId, CAST(NULL AS STRING) AS targetUid,
           CAST(NULL AS STRING) AS networkType, CAST(NULL AS STRING) AS operationOptions, CAST(NULL AS STRING) AS operationExtParams,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.simpleevent_log_flow
    WHERE year='{year}' AND month='{month}' AND day='{day_of_month}'{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, eventid, system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype, CAST(NULL AS STRING) AS 视频id,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid, CAST(NULL AS STRING) AS creativeCode,
           CAST(NULL AS STRING) AS rootPageSource, CAST(NULL AS STRING) AS objectId, CAST(NULL AS STRING) AS targetUid,
           CAST(NULL AS STRING) AS networkType, CAST(NULL AS STRING) AS operationOptions, CAST(NULL AS STRING) AS operationExtParams,
           CAST(NULL AS STRING) AS hotsencetype, path, subsessionid, sessionid
    FROM loghubods.useractive_log_per5min
    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}'{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, eventid, CAST(NULL AS STRING) AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype, CAST(NULL AS STRING) AS 视频id,
           topic, shareid, CAST(NULL AS STRING) AS creativeCode,
           rootpagesource AS rootPageSource, shareobjectid AS objectId, CAST(NULL AS STRING) AS targetUid,
           CAST(NULL AS STRING) AS networkType, CAST(NULL AS STRING) AS operationOptions, CAST(NULL AS STRING) AS operationExtParams,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.user_share_log_per5min
    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}'{apptype_filter} AND machinecode='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
o AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, apptype AS 产品apptype, 'operation' AS 来源,
           CAST(NULL AS STRING) AS businesstype, eventid, machineinfo_system AS 系统, pagesource, CAST(NULL AS STRING) AS endRoutePath, CAST(NULL AS STRING) AS objecttype, CAST(NULL AS STRING) AS 视频id,
           CAST(NULL AS STRING) AS topic, CAST(NULL AS STRING) AS shareid, CAST(NULL AS STRING) AS creativeCode,
           rootpagesource AS rootPageSource, objectid AS objectId, targetuid AS targetUid,
           networktype AS networkType, options AS operationOptions, extparams AS operationExtParams,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.operation_log_per5min
    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}'{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 UNION ALL SELECT * FROM o
)
SELECT t.ts, t.产品apptype, t.来源, t.businesstype, t.eventid, t.系统, t.endRoutePath, t.objecttype, t.pagesource, t.topic, t.shareid, t.creativeCode, t.视频id, b.title AS 视频标题,
       t.rootPageSource, t.objectId, t.targetUid, t.networkType, t.operationOptions, t.operationExtParams,
       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 main():
    parser = argparse.ArgumentParser(description="查询单用户实时行为时间线")
    parser.add_argument("user_id", help="machinecode/mid")
    parser.add_argument("date", nargs="?", default=datetime.now().strftime("%Y%m%d"), help="yyyyMMdd")
    parser.add_argument("apptype", nargs="?", default=None, help="可选；不传则查询全部产品")
    parser.add_argument("--output-dir", type=Path, default=Path("."))
    args = parser.parse_args()
    query_date = datetime.strptime(args.date, "%Y%m%d")
    sql_args = {"mc": sql_text(args.user_id), "apptype_filter": build_apptype_filter(args.apptype), "dt_start": f"{args.date}000000", "dt_end": f"{args.date}235959", "year": query_date.strftime("%Y"), "month": query_date.strftime("%m"), "day_of_month": query_date.strftime("%d")}
    df = ODPSClient().execute_sql(SQL_REALTIME.format(**sql_args))
    df = df[~df["businesstype"].isin(EXCLUDED_BUSINESSTYPES)].sort_values("ts", kind="stable").reset_index(drop=True)
    df = df.rename(columns={"creativecode": "creativeCode", "endroutepath": "endRoutePath", "rootpagesource": "rootPageSource", "objectid": "objectId", "targetuid": "targetUid", "networktype": "networkType", "operationoptions": "operationOptions", "operationextparams": "operationExtParams"})
    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_realtime_{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"
    print("注意：simpleevent_log_flow 仅保留当前短实时窗口。")
    print(f"[XLSX] 日期={args.date} 事件数={len(df)} -> {out.resolve()}", flush=True)


if __name__ == "__main__":
    main()