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