data_query_skill/skills/query-user-behavior-path/scripts/user_timeline.py

#!/usr/bin/env python
# coding=utf-8
"""单用户完整行为时间线：视频、广告、播放、用户行为、活跃路径五类离线日志。

五类日志在同一个 ODPS SQL 中 UNION ALL；Python 端补充中文行为定义并输出 Excel。

用法：python3 user_timeline.py <machinecode/mid值> <日期yyyyMMdd> [apptype] [--output-dir 目录]
"""
import argparse
from datetime import datetime
from pathlib import Path
import re

import pandas as pd

from odps_module import ODPSClient


PAGESTATUS_MEAN = {
    "1": "头部未起播(封面/广告覆盖)",
    "2": "feed正常态",
    "3": "feed_m初始态",
    "4": "沉浸式页",
    "5": "feed-从沉浸/feed_m返回(粘)",
    "7": "首页category曝光/分享",
    "8": "feed_m-从沉浸返回(粘)",
}

EXCLUDED_BUSINESSTYPES = {"deviceId", "openGIdSuccess", "buttonView"}


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


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, 'video' AS 来源,
           businesstype, pagesource, CAST(NULL AS STRING) AS endRoutePath,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, videoid AS 视频id,
           GET_JSON_OBJECT(extparams, '$.auto_enter') AS auto_enter,
           GET_JSON_OBJECT(extparams, '$.newPage') AS newPage,
           GET_JSON_OBJECT(extparams, '$.pageStatus') AS pageStatus,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.video_action_log_applet
    WHERE dt='{day}' AND apptype='{apptype}' AND mid='{mc}'
      AND businesstype <> 'videoPreView'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
a AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, 'ad' AS 来源,
           businesstype, pagesource, CAST(NULL AS STRING) AS endRoutePath,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, headvideoid AS 视频id,
           CAST(NULL AS STRING) AS auto_enter, CAST(NULL AS STRING) AS newPage, CAST(NULL AS STRING) AS pageStatus,
           hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.ad_action_log_own
    WHERE dt='{day}' AND apptype='{apptype}' AND machinecode='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
p AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, 'play' AS 来源,
           businesstype, pagesource, CAST(NULL AS STRING) AS endRoutePath,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, videoid AS 视频id,
           GET_JSON_OBJECT(extparams, '$.auto_enter') AS auto_enter,
           GET_JSON_OBJECT(extparams, '$.newPage') AS newPage,
           GET_JSON_OBJECT(extparams, '$.pageStatus') AS pageStatus,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.video_play_log
    WHERE dt='{day}' AND apptype='{apptype}' AND mid='{mc}'
      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
),
s AS (
    SELECT CAST(clienttimestamp AS BIGINT) AS ts, 'simpleevent' AS 来源,
           businesstype, pagesource, endroutepath AS endRoutePath,
           GET_JSON_OBJECT(extparams, '$.isAdPlaying') AS isAdPlaying,
           GET_JSON_OBJECT(extparams, '$.creativeCode') AS creativeCode, videoid AS 视频id,
           CAST(NULL AS STRING) AS auto_enter, CAST(NULL AS STRING) AS newPage, CAST(NULL AS STRING) AS pageStatus,
           CAST(NULL AS STRING) AS hotsencetype, CAST(NULL AS STRING) AS path, subsessionid, sessionid
    FROM loghubods.simpleevent_log
    WHERE dt='{day}' AND apptype='{apptype}' 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, 'useractive' AS 来源,
           businesstype, pagesource, CAST(NULL AS STRING) AS endRoutePath,
           CAST(NULL AS STRING) AS isAdPlaying, CAST(NULL AS STRING) AS creativeCode, CAST(NULL AS STRING) AS 视频id,
           CAST(NULL AS STRING) AS auto_enter, CAST(NULL AS STRING) AS newPage, CAST(NULL AS STRING) AS pageStatus,
           CAST(NULL AS STRING) AS hotsencetype, path, subsessionid, sessionid
    FROM loghubods.useractive_log
    WHERE dt='{day}' AND apptype='{apptype}' 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
)
SELECT t.ts, t.来源, t.businesstype, t.pagesource, t.endRoutePath, t.isAdPlaying, t.creativeCode, t.视频id, b.title AS 视频标题,
       t.auto_enter, t.newPage, t.pageStatus, 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 behavior_definition(row):
    source = text(row["来源"])
    businesstype = text(row["businesstype"])
    pagesource = text(row["pagesource"])
    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" and is_head_video_page:
        return {
            "adRequest": "头部视频页广告请求",
            "adLoaded": "头部视频页广告加载",
            "adView": "头部视频页广告曝光",
            "adPlay": "头部视频页广告播放",
        }.get(businesstype, "")

    if source == "simpleevent":
        if businesstype == "pageView" and pagesource.endswith("category_55"):
            return "首页分类页曝光（分类55）"
        if is_head_video_page:
            return {
                "pageView": "视频分享页曝光",
                "detailRequest": "视频分享页详情接口请求成功",
                "userCaptureScreen": "用户在视频分享页截图",
            }.get(businesstype, "")
        return {
            "userPause": "视频暂停",
            "userActiveEnd": "小程序进入后台",
        }.get(businesstype, "")

    return ""


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="0")
    parser.add_argument("--output-dir", type=Path, default=Path("."))
    args = parser.parse_args()

    datetime.strptime(args.date, "%Y%m%d")
    cli = ODPSClient()
    sql_args = {
        "mc": sql_text(args.user_id),
        "day": args.date,
        "apptype": sql_text(args.apptype),
    }
    df = cli.execute_sql(SQL_ALL.format(**sql_args))
    df = df[~df["businesstype"].isin(EXCLUDED_BUSINESSTYPES)]
    df = df.sort_values("ts", kind="stable").reset_index(drop=True)
    df = df.rename(
        columns={
            "pagestatus": "pageStatus",
            "newpage": "newPage",
            "endroutepath": "endRoutePath",
            "isadplaying": "isAdPlaying",
            "creativecode": "creativeCode",
        }
    )
    df["pageStatus"] = df["pageStatus"].map(
        lambda value: f"{text(value)}_{PAGESTATUS_MEAN[text(value)]}" if text(value) in PAGESTATUS_MEAN else text(value)
    )
    df.insert(0, "北京时间", pd.to_datetime(df["ts"], unit="ms", utc=True).dt.tz_convert("Asia/Shanghai").dt.tz_localize(None))
    df = df.drop(columns="ts")
    df.insert(df.columns.get_loc("businesstype") + 1, "中文行为定义", df.apply(behavior_definition, axis=1))

    output_dir = args.output_dir.expanduser()
    output_dir.mkdir(parents=True, exist_ok=True)
    out = 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["行为路径"]["A"][1:]:
            cell.number_format = "yyyy/mm/dd hh:mm:ss"
    counts = df["来源"].value_counts().to_dict()
    for source in ["video", "ad", "play", "simpleevent", "useractive"]:
        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()