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

#!/usr/bin/env python3
# coding=utf-8
"""多个实时用户的行为时间线，输出可筛选的单一 Excel。"""
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_BATCH = """
WITH v AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,mid 用户标识,'video' 来源,businesstype,CAST(NULL AS STRING) eventid,machineinfo_system 系统,pagesource,CAST(NULL AS STRING) endRoutePath,CAST(NULL AS STRING) objecttype,CAST(NULL AS STRING) topic,CAST(NULL AS STRING) shareid,CAST(NULL AS STRING) creativeCode,videoid 视频id,CAST(NULL AS STRING) rootPageSource,CAST(NULL AS STRING) objectId,CAST(NULL AS STRING) targetUid,CAST(NULL AS STRING) networkType,CAST(NULL AS STRING) operationOptions,CAST(NULL AS STRING) operationExtParams,CAST(NULL AS STRING) hotsencetype,CAST(NULL AS STRING) path,subsessionid,sessionid
 FROM loghubods.video_action_log_per5min WHERE dt>='{dt_start}' AND dt<='{dt_end}'{apptype_filter} AND mid IN ({mc_list}) AND businesstype<>'videoPreView' AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),a AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,machinecode 用户标识,'ad' 来源,businesstype,eventid,GET_JSON_OBJECT(machineinfo,'$.system') 系统,pagesource,CAST(NULL AS STRING) endRoutePath,CAST(NULL AS STRING) objecttype,CAST(NULL AS STRING) topic,CAST(NULL AS STRING) shareid,creativecode creativeCode,headvideoid 视频id,CAST(NULL AS STRING) rootPageSource,CAST(NULL AS STRING) objectId,CAST(NULL AS STRING) targetUid,CAST(NULL AS STRING) networkType,CAST(NULL AS STRING) operationOptions,CAST(NULL AS STRING) operationExtParams,hotsencetype,CAST(NULL AS STRING) path,subsessionid,sessionid
 FROM loghubods.ad_action_log_own_per5min WHERE dt>='{dt_start}' AND dt<='{dt_end}'{apptype_filter} AND machinecode IN ({mc_list}) AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),p AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,mid 用户标识,'play' 来源,businesstype,eventid,machineinfo_system 系统,pagesource,CAST(NULL AS STRING) endRoutePath,CAST(NULL AS STRING) objecttype,CAST(NULL AS STRING) topic,CAST(NULL AS STRING) shareid,CAST(NULL AS STRING) creativeCode,videoid 视频id,CAST(NULL AS STRING) rootPageSource,CAST(NULL AS STRING) objectId,CAST(NULL AS STRING) targetUid,CAST(NULL AS STRING) networkType,CAST(NULL AS STRING) operationOptions,CAST(NULL AS STRING) operationExtParams,CAST(NULL AS STRING) hotsencetype,CAST(NULL AS STRING) path,subsessionid,sessionid
 FROM loghubods.video_play_log_per5min WHERE dt>='{dt_start}' AND dt<='{dt_end}'{apptype_filter} AND mid IN ({mc_list}) AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),s AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,machinecode 用户标识,'simpleevent' 来源,businesstype,eventid,system 系统,pagesource,CAST(NULL AS STRING) endRoutePath,objecttype,CAST(NULL AS STRING) topic,CAST(NULL AS STRING) shareid,CAST(NULL AS STRING) creativeCode,videoid 视频id,CAST(NULL AS STRING) rootPageSource,CAST(NULL AS STRING) objectId,CAST(NULL AS STRING) targetUid,CAST(NULL AS STRING) networkType,CAST(NULL AS STRING) operationOptions,CAST(NULL AS STRING) operationExtParams,CAST(NULL AS STRING) hotsencetype,CAST(NULL AS STRING) path,subsessionid,sessionid
 FROM loghubods.simpleevent_log_flow WHERE year='{year}' AND month='{month}' AND day='{day_of_month}'{apptype_filter} AND machinecode IN ({mc_list}) AND (businesstype IS NULL OR businesstype<>'openGIdError') AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),u AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,machinecode 用户标识,'useractive' 来源,businesstype,eventid,system 系统,pagesource,CAST(NULL AS STRING) endRoutePath,CAST(NULL AS STRING) objecttype,CAST(NULL AS STRING) topic,CAST(NULL AS STRING) shareid,CAST(NULL AS STRING) creativeCode,CAST(NULL AS STRING) 视频id,CAST(NULL AS STRING) rootPageSource,CAST(NULL AS STRING) objectId,CAST(NULL AS STRING) targetUid,CAST(NULL AS STRING) networkType,CAST(NULL AS STRING) operationOptions,CAST(NULL AS STRING) operationExtParams,CAST(NULL AS STRING) hotsencetype,path,subsessionid,sessionid
 FROM loghubods.useractive_log_per5min WHERE dt>='{dt_start}' AND dt<='{dt_end}'{apptype_filter} AND machinecode IN ({mc_list}) AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),r AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,machinecode 用户标识,'share' 来源,type businesstype,eventid,CAST(NULL AS STRING) 系统,pagesource,CAST(NULL AS STRING) endRoutePath,CAST(NULL AS STRING) objecttype,topic,shareid,CAST(NULL AS STRING) creativeCode,CAST(NULL AS STRING) 视频id,rootpagesource rootPageSource,shareobjectid objectId,CAST(NULL AS STRING) targetUid,CAST(NULL AS STRING) networkType,CAST(NULL AS STRING) operationOptions,CAST(NULL AS STRING) operationExtParams,CAST(NULL AS STRING) hotsencetype,CAST(NULL AS STRING) path,subsessionid,sessionid
 FROM loghubods.user_share_log_per5min WHERE dt>='{dt_start}' AND dt<='{dt_end}'{apptype_filter} AND machinecode IN ({mc_list}) AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),o AS (
 SELECT CAST(clienttimestamp AS BIGINT) ts,apptype 产品apptype,machinecode 用户标识,'operation' 来源,CAST(NULL AS STRING) businesstype,eventid,machineinfo_system 系统,pagesource,CAST(NULL AS STRING) endRoutePath,CAST(NULL AS STRING) objecttype,CAST(NULL AS STRING) topic,CAST(NULL AS STRING) shareid,CAST(NULL AS STRING) creativeCode,CAST(NULL AS STRING) 视频id,rootpagesource rootPageSource,objectid objectId,targetuid targetUid,networktype networkType,options operationOptions,extparams operationExtParams,CAST(NULL AS STRING) hotsencetype,CAST(NULL AS STRING) path,subsessionid,sessionid
 FROM loghubods.operation_log_per5min WHERE dt>='{dt_start}' AND dt<='{dt_end}'{apptype_filter} AND machinecode IN ({mc_list}) 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.来源,t.businesstype,t.eventid,t.系统,t.endRoutePath,t.objecttype,t.pagesource,t.topic,t.shareid,t.creativeCode,t.视频id,b.title 视频标题,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 load_user_strategies(path):
    users = pd.read_csv(path, dtype=str).fillna("")
    if "用户标识" not in users.columns:
        raise ValueError("用户清单必须包含“用户标识”列")
    strategy_column = "命中策略" if "命中策略" in users.columns else None
    strategies = {}
    for _, row in users.iterrows():
        user = row["用户标识"].strip()
        if user:
            strategy = row[strategy_column].strip() if strategy_column else ""
            strategies.setdefault(user, [])
            if strategy and strategy not in strategies[user]:
                strategies[user].append(strategy)
    if not strategies:
        raise ValueError("用户清单中没有有效的用户标识")
    return {user: "；".join(tags) for user, tags in strategies.items()}


def sql_literals(values):
    return ",".join("'" + sql_text(value) + "'" for value in values)


def main():
    parser = argparse.ArgumentParser(description="批量查询实时用户行为时间线")
    parser.add_argument("users_file", type=Path)
    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()
    users = load_user_strategies(args.users_file)
    date = datetime.strptime(args.date, "%Y%m%d")
    sql_args = {"mc_list": sql_literals(users), "apptype_filter": build_apptype_filter(args.apptype), "dt_start": f"{args.date}000000", "dt_end": f"{args.date}235959", "year": date.strftime("%Y"), "month": date.strftime("%m"), "day_of_month": date.strftime("%d")}
    df = ODPSClient().execute_sql(SQL_REALTIME_BATCH.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"), group_column="用户标识")
    df.insert(2, "命中策略", df["用户标识"].map(users).fillna(""))
    df.insert(df.columns.get_loc("businesstype") + 1, "行为", df.apply(behavior_definition, axis=1))
    df = df.rename(columns={"businesstype": "事件类型", "eventid": "事件ID"})
    df.insert(0, "产品apptype", df.pop("产品apptype"))
    df.insert(1, "用户标识", df.pop("用户标识"))
    df.insert(2, "命中策略", df.pop("命中策略"))
    counts = df["用户标识"].value_counts()
    summary = pd.DataFrame({"用户标识": list(users), "命中策略": list(users.values()), "实时事件数": [counts.get(user, 0) for user in users]})
    args.output_dir.mkdir(parents=True, exist_ok=True)
    out = args.output_dir / f"timeline_realtime_{safe_name(args.users_file.stem)}_{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)
        worksheet = writer.sheets["实时行为路径"]
        worksheet.freeze_panes = "A2"
        worksheet.auto_filter.ref = worksheet.dimensions
        for cell in worksheet["D"][1:]:
            cell.number_format = "yyyy/mm/dd hh:mm:ss"
        summary.to_excel(writer, sheet_name="用户摘要", index=False)
        writer.sheets["用户摘要"].freeze_panes = "A2"
        writer.sheets["用户摘要"].auto_filter.ref = writer.sheets["用户摘要"].dimensions
    print("注意：simpleevent_log_flow 仅保留当前短实时窗口。")
    print(f"[XLSX] 用户数={len(users)} 日期={args.date} 事件数={len(df)} -> {out.resolve()}", flush=True)


if __name__ == "__main__":
    main()