feat: package reusable ODPS query skills

.env.example

@@ -0,0 +1,4 @@
+export ODPS_ACCESS_ID=''
+export ODPS_ACCESS_SECRET=''
+export ODPS_PROJECT=''
+export ODPS_ENDPOINT=''

.gitignore

@@ -0,0 +1,15 @@
+.DS_Store
+__pycache__/
+*.py[cod]
+
+.env
+.env.*
+!.env.example
+
+*.csv
+*.xlsx
+*.log
+
+task_plan.md
+findings.md
+progress.md

README.md

@@ -0,0 +1,63 @@
+# Data Query Skills
+
+面向团队私有仓库分发的 4 个 ODPS 数据查询 Skill：
+
+- `odps-ad-risk-analysis`
+- `odps-growth-fission-report`
+- `odps-product-efficiency-report`
+- `query-user-behavior-path`
+
+每个 Skill 都可以独立安装，运行时只读取当前使用者自己的环境变量，不依赖作者机器上的项目目录或凭证文件。
+
+## 环境准备
+
+```bash
+python3 -m pip install -r requirements.txt
+```
+
+在仓库外配置以下环境变量：
+
+```bash
+export ODPS_ACCESS_ID='个人 Access ID'
+export ODPS_ACCESS_SECRET='个人 Access Secret'
+export ODPS_PROJECT='ODPS 项目名'
+export ODPS_ENDPOINT='ODPS Endpoint'
+```
+
+可以复制 `.env.example` 到个人私有配置目录，例如 `~/.config/data-query-skills/odps.env`，填写后执行：
+
+```bash
+chmod 600 ~/.config/data-query-skills/odps.env
+source ~/.config/data-query-skills/odps.env
+```
+
+不要把填有真实值的配置文件复制回仓库。每位同事应使用自己的账号和最小权限凭证。
+
+## 从私有 GitHub 仓库安装
+
+```bash
+python3 ~/.codex/skills/.system/skill-installer/scripts/install-skill-from-github.py \
+  --repo <组织>/<仓库> \
+  --path \
+    skills/odps-ad-risk-analysis \
+    skills/odps-growth-fission-report \
+    skills/odps-product-efficiency-report \
+    skills/query-user-behavior-path
+```
+
+私有仓库需要本机已有 Git 凭证，或者配置 `GH_TOKEN`。安装完成后，Skill 会在下一轮 Codex 对话中可用。
+
+## 从其他私有 Git 服务安装
+
+```bash
+git clone <私有仓库地址> data-query-skills
+mkdir -p ~/.codex/skills
+cp -R data-query-skills/skills/* ~/.codex/skills/
+```
+
+## 安全边界
+
+- Skill 和 Git 仓库不得包含 Access ID、Access Secret、临时 Token 或带 Token 的 LogView URL。
+- `scripts/odps_module.py` 只打印 ODPS Instance ID，不打印 LogView URL。
+- 查询生成的 CSV、Excel、日志和个人 `.env` 默认不提交。
+- 表名、指标口径和风控规则仍属于内部资料，只应放在授权的私有仓库。

requirements.txt

@@ -0,0 +1,3 @@
+pyodps
+pandas
+openpyxl

skills/odps-ad-risk-analysis/SKILL.md

@@ -0,0 +1,48 @@
+---
+name: odps-ad-risk-analysis
+description: 查询和维护 ODPS 广告风险用户策略，覆盖广告播放页截图、落地页截图、广告曝光后切后台、落地页隐藏且无打开转化、动态黑名单分层、DAU 交集，以及按 mid 或 uid 定向诊断。用于生成、执行、审查或调整 loghubods 广告风控 SQL。
+---
+
+# ODPS 广告风险分析
+
+将 `SKILL_DIR` 解析为当前 `SKILL.md` 所在的安装目录。所有资源都相对该目录解析；不得假定当前工作目录，更不得引用作者机器上的项目路径。
+
+## 运行环境
+
+需要执行查询时，使用随 Skill 分发的 `scripts/run_sql.py`。它只读取以下环境变量：
+
+- `ODPS_ACCESS_ID`
+- `ODPS_ACCESS_SECRET`
+- `ODPS_PROJECT`
+- `ODPS_ENDPOINT`
+
+依赖 `pyodps` 和 `pandas`。不得把凭证复制到 SQL、脚本、Skill 文件或回复中。只输出 ODPS Instance ID，不输出 LogView URL 或 Token。
+
+## 工作流程
+
+1. 阅读 [risk-strategies.md](references/risk-strategies.md) 和 [sql-templates.md](references/sql-templates.md) 中对应的完整模板。
+2. 确认风险场景，以及用户需要用户清单、UV、事件对、DAU 交集、黑名单层级还是共享表更新。
+3. 在用户指定的输出目录生成 SQL。除非明确要求实时数据，否则默认使用离线表。
+4. 保留场景定义中的事件键、时间窗口、事件粒度和输出字段。
+5. 指定 `mid` 或 `uid` 时，先执行窄范围命中查询，不向终端输出大批用户清单。
+6. 需要执行时运行：
+
+   `python3 "$SKILL_DIR/scripts/run_sql.py" <query.sql> <result.csv>`
+
+7. 输出前校验事件粒度、日期覆盖、行数和黑名单层级行为。
+
+## 不可变规则
+
+- `mid` 对应 `machinecode`。
+- 历史风险清单默认使用 `20260401` 至 `${bizdate}`，除非用户确认其他时间段。
+- 保留最新事件的原始 `loginuid`；最新值为空时不得用旧事件补齐。
+- 共享表中的 3 次及以上永久黑名单必须保留；只动态替换当前 1/2 层级。
+- 只按文档规定的事件对粒度计数，不得擅自按 session 去重。
+- 只有用户明确要求宽松实时检查时才能使用 `simpleevent_log_flow`，并说明短保留周期限制。
+
+## 资源
+
+- [risk-strategies.md](references/risk-strategies.md)：策略定义、键、字段、类型、表和输出规则。
+- [sql-templates.md](references/sql-templates.md)：标准完整 SQL 模板及诊断变体。
+- `scripts/odps_module.py`：基于环境变量的 ODPS 连接。
+- `scripts/run_sql.py`：可移植的 SQL 转 CSV 执行器。

skills/odps-ad-risk-analysis/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "ODPS 广告风控查询"
+  short_description: "广告风险用户、黑名单分层与 ODPS SQL 模板"
+  default_prompt: "使用 $odps-ad-risk-analysis 统计广告风险用户、分层黑名单或生成可运行的 ODPS SQL。"

skills/odps-ad-risk-analysis/references/risk-strategies.md

@@ -0,0 +1,57 @@
+# Risk Strategy Catalog
+
+## Shared identifiers and outputs
+
+- `mid`: `machinecode`.
+- `uid`: the scenario's raw `loginuid`; a null latest value stays null.
+- Standard user output: ``type, mid, uid, risk_level``; `risk_level=0`.
+- Historical start: `20260401`; end: `${bizdate}`.
+- Use offline tables unless the user explicitly requests real-time tables.
+
+## S1: Advertising-playback screenshot
+
+- Event chain: `ad_action_log_own.adView` -> `simpleevent_log.userCaptureScreen`.
+- Key: same `event_dt + mid + subsessionid`; capture timestamp is at or after adView.
+- Exclusion: no `adCloseBtnTap` between adView and capture.
+- User type: `adPlay_capture`.
+- No blacklist tier: one deduplicated `mid` row.
+
+## S2: Advertising landing-page screenshot
+
+- Event: `ad_action_log_own.adUserCaptureScreen`.
+- User type: `adlanding_capture`.
+- No additional landing view or open-conversion condition unless requested.
+- No blacklist tier: one deduplicated `mid` row.
+
+## S3: Self-owned ad backgrounding then return
+
+- Event chain: own-platform `adView` -> within 10 seconds `userActiveEnd` on `pages/swiper/index` -> same-session return to `pages/swiper/index`.
+- Key before background: same `event_dt + mid + sessionid + subsessionid`.
+- Exclusions: no `adCloseBtnTap`, no homepage `pageView` whose `pagesource` ends in `category_55`, both between adView and active end.
+- Count grain: each distinct `mid, sessionid, subsessionid, adview_ts, active_end_ts` pair is one risk event.
+- Types: `adview10s_hide_return_risk_1_forbidden_3`, `adview10s_hide_return_risk_2_forbidden_30`, `adview10s_hide_return_risk_3_forbidden_forever`.
+
+## S4: Self-owned ad landing hide, no open conversion, then return
+
+- Event chain: own-platform `adSelfLandingView` -> within 10 seconds `adSelfLandingHide` -> same-session return to one of the four landing paths.
+- View/hide key: same `event_dt + mid + sessionid + subsessionid + pqtid`.
+- Open-conversion exclusion: `pqtid` must not exist in `ad_own_open_conv`.
+- Return paths: `pages/ad-self-landing/index`, `pages/marketing-landing/index`, `pages/promo-page/index`, `pages/event-landing/index`.
+- Count grain: each distinct `mid, sessionid, subsessionid, pqtid, view_ts, hide_ts` pair is one risk event.
+- Types: `adlanding10s_hide_no_open_return_risk_1_forbidden_3`, `adlanding10s_hide_no_open_return_risk_2_forbidden_30`, `adlanding10s_hide_no_open_return_risk_3_forbidden_forever`.
+
+## Dynamic blacklist tiers
+
+For S3 and S4, tiers are mutually exclusive.
+
+- 1 event: blacklist only when `last_event_dt` is from `${bizdate}-2 days` through `${bizdate}`.
+- 2 events: blacklist only when `last_event_dt` is from `${bizdate}-29 days` through `${bizdate}`.
+- 3+ events: permanent blacklist.
+- Compute dynamic starts with `TO_CHAR(DATEADD(TO_DATE('${bizdate}', 'yyyymmdd'), -N, 'dd'), 'yyyymmdd')`.
+- When writing to a shared target table, retain historical permanent rows, replace current 1/2 rows, and preserve unrelated types.
+
+## DAU and real-time diagnostics
+
+- Confirmed 0709 Path DAU: `useractive_log`, `dt='20260709'`, `businesstype='path'`, distinct `machinecode`.
+- `simpleevent_log_flow` has fields `year, month, day, hour` and no `endRoutePath`.
+- Real-time checks using the flow table must explicitly say that removing `endRoutePath` creates a relaxed metric and only covers the available short retention window.

skills/odps-ad-risk-analysis/references/sql-templates.md

@@ -0,0 +1,87 @@
+# Canonical SQL Templates
+
+Use `${bizdate}` as `yyyyMMdd`. Do not mix these templates' CTEs across scenarios.
+
+## S1: Playback screenshot user list
+
+```sql
+WITH adview_events AS (
+  SELECT dt event_dt, machinecode mid, subsessionid, CAST(clienttimestamp AS BIGINT) adview_ts
+  FROM loghubods.ad_action_log_own
+  WHERE dt BETWEEN '20260401' AND '${bizdate}' AND businesstype='adView'
+    AND machinecode IS NOT NULL AND machinecode<>'' AND subsessionid IS NOT NULL AND subsessionid<>'' AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+), capture_events AS (
+  SELECT dt event_dt, machinecode mid, loginuid uid, subsessionid, CAST(clienttimestamp AS BIGINT) capture_ts
+  FROM loghubods.simpleevent_log
+  WHERE dt BETWEEN '20260401' AND '${bizdate}' AND businesstype='userCaptureScreen'
+    AND machinecode IS NOT NULL AND machinecode<>'' AND subsessionid IS NOT NULL AND subsessionid<>'' AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+), close_events AS (
+  SELECT dt event_dt, machinecode mid, subsessionid, CAST(clienttimestamp AS BIGINT) close_ts
+  FROM loghubods.ad_action_log_own
+  WHERE dt BETWEEN '20260401' AND '${bizdate}' AND businesstype='adCloseBtnTap'
+    AND machinecode IS NOT NULL AND machinecode<>'' AND subsessionid IS NOT NULL AND subsessionid<>'' AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+), matched AS (
+  SELECT DISTINCT a.mid,c.uid,c.capture_ts
+  FROM adview_events a JOIN capture_events c ON a.event_dt=c.event_dt AND a.mid=c.mid AND a.subsessionid=c.subsessionid AND c.capture_ts>=a.adview_ts
+  WHERE NOT EXISTS (SELECT 1 FROM close_events x WHERE x.event_dt=a.event_dt AND x.mid=a.mid AND x.subsessionid=a.subsessionid AND x.close_ts>a.adview_ts AND x.close_ts<=c.capture_ts)
+), latest_uid AS (
+  SELECT mid,uid FROM (SELECT mid,uid,ROW_NUMBER() OVER(PARTITION BY mid ORDER BY capture_ts DESC) rn FROM matched) t WHERE rn=1
+)
+SELECT 'adPlay_capture' AS `type`, mid, uid, 0 AS risk_level FROM latest_uid;
+```
+
+For the count, keep all CTEs and replace the final select with `SELECT COUNT(*) FROM latest_uid`.
+
+## S2: Landing screenshot user list
+
+```sql
+WITH landing_capture_events AS (
+  SELECT machinecode mid, loginuid uid, CAST(clienttimestamp AS BIGINT) capture_ts
+  FROM loghubods.ad_action_log_own
+  WHERE dt BETWEEN '20260401' AND '${bizdate}' AND businesstype='adUserCaptureScreen'
+    AND machinecode IS NOT NULL AND machinecode<>'' AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+), latest_uid AS (
+  SELECT mid,uid FROM (SELECT mid,uid,ROW_NUMBER() OVER(PARTITION BY mid ORDER BY capture_ts DESC) rn FROM landing_capture_events) t WHERE rn=1
+)
+SELECT 'adlanding_capture' AS `type`, mid, uid, 0 AS risk_level FROM latest_uid;
+```
+
+## S3/S4: Dynamic-tier output contract
+
+Build `risk_events(mid, uid, event_dt, event_ts, ...)` with the scenario-specific keys documented in `risk-strategies.md`, then use this exact suffix:
+
+```sql
+, user_risk AS (
+  SELECT mid, COUNT(*) behavior_cnt, MAX(event_dt) last_event_dt FROM risk_events GROUP BY mid
+), latest_uid AS (
+  SELECT mid,uid FROM (
+    SELECT mid,uid,ROW_NUMBER() OVER(PARTITION BY mid ORDER BY event_ts DESC) rn FROM risk_events
+  ) t WHERE rn=1
+), current_risk_users AS (
+  SELECT r.mid,u.uid,
+    CASE
+      WHEN r.behavior_cnt>=3 THEN '${type_3}'
+      WHEN r.behavior_cnt=2 AND r.last_event_dt BETWEEN TO_CHAR(DATEADD(TO_DATE('${bizdate}','yyyymmdd'),-29,'dd'),'yyyymmdd') AND '${bizdate}' THEN '${type_2}'
+      WHEN r.behavior_cnt=1 AND r.last_event_dt BETWEEN TO_CHAR(DATEADD(TO_DATE('${bizdate}','yyyymmdd'),-2,'dd'),'yyyymmdd') AND '${bizdate}' THEN '${type_1}'
+    END AS risk_type
+  FROM user_risk r LEFT JOIN latest_uid u ON r.mid=u.mid
+)
+SELECT risk_type AS `type`,mid,uid,0 AS risk_level
+FROM current_risk_users WHERE risk_type IS NOT NULL;
+```
+
+### S3 risk_events builder
+
+Use `ad_action_log_own.adView` with `ownAdSystemType='ownPlatform'`; join `simpleevent_log.userActiveEnd` on same day, mid, sessionid, and subsessionid; require `0 <= active_end_ts-adview_ts <= 10000`; exclude close and homepage pageview in that interval; require later same-session `useractive_log.path='pages/swiper/index'`. Set `event_ts=active_end_ts`.
+
+Types: `${type_1}=adview10s_hide_return_risk_1_forbidden_3`, `${type_2}=adview10s_hide_return_risk_2_forbidden_30`, `${type_3}=adview10s_hide_return_risk_3_forbidden_forever`.
+
+### S4 risk_events builder
+
+Use own-platform `adSelfLandingView` and `adSelfLandingHide` on same day, mid, sessionid, subsessionid, pqtid; require `0 <= hide_ts-view_ts <= 10000`; exclude pqtid present in `ad_own_open_conv`; require later same-session `useractive_log.path` in the four documented landing paths. Set `event_ts=hide_ts`.
+
+Types: `${type_1}=adlanding10s_hide_no_open_return_risk_1_forbidden_3`, `${type_2}=adlanding10s_hide_no_open_return_risk_2_forbidden_30`, `${type_3}=adlanding10s_hide_no_open_return_risk_3_forbidden_forever`.
+
+## Shared-table write behavior
+
+For a target table with `type, mid, uid, risk_level`, use `INSERT OVERWRITE` only after unioning: unrelated existing types, existing permanent type-3 users, newly computed type-3 users, and current dynamic type-1/type-2 users excluding permanent mids. Do not retain expired type-1/type-2 rows.

skills/odps-ad-risk-analysis/scripts/odps_module.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Minimal PyODPS client configured only through environment variables."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from odps import ODPS, options
+
+
+ENV_ACCESS_ID = "ODPS_ACCESS_ID"
+ENV_ACCESS_SECRET = "ODPS_ACCESS_SECRET"
+ENV_PROJECT = "ODPS_PROJECT"
+ENV_ENDPOINT = "ODPS_ENDPOINT"
+
+
+def require_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(f"Missing required environment variable: {name}")
+    return value
+
+
+class ODPSClient:
+    def __init__(self, project: str | None = None, endpoint: str | None = None):
+        access_id = require_env(ENV_ACCESS_ID)
+        access_secret = require_env(ENV_ACCESS_SECRET)
+        project_name = project or require_env(ENV_PROJECT)
+        endpoint_url = endpoint or require_env(ENV_ENDPOINT)
+
+        options.connect_timeout = 60
+        options.read_timeout = 1200
+        options.retry_times = 3
+
+        self.odps = ODPS(
+            access_id,
+            access_secret,
+            project=project_name,
+            endpoint=endpoint_url,
+        )
+
+    def execute_sql(self, sql: str):
+        if not sql.strip():
+            raise ValueError("SQL must not be empty")
+
+        instance = self.odps.run_sql(
+            sql,
+            hints={"odps.sql.submit.mode": "script"},
+        )
+        print(f"[ODPS] InstanceId: {instance.id}", flush=True)
+        instance.wait_for_success()
+        with instance.open_reader(tunnel=True) as reader:
+            return reader.to_pandas()
+
+    def execute_sql_result_save_file(self, sql: str, output_file: str | Path):
+        output_path = Path(output_file).expanduser()
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        data = self.execute_sql(sql)
+        data.to_csv(output_path, index=False, encoding="utf-8-sig")
+        return output_path

skills/odps-ad-risk-analysis/scripts/run_sql.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Execute an explicit SQL file with the bundled environment-based ODPS client."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from odps_module import ODPSClient
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run an ODPS SQL file and save UTF-8 CSV")
+    parser.add_argument("sql_file", type=Path)
+    parser.add_argument("output_file", type=Path)
+    parser.add_argument("--project", help="Override ODPS_PROJECT for this query")
+    args = parser.parse_args()
+
+    sql = args.sql_file.expanduser().read_text(encoding="utf-8")
+    output_path = ODPSClient(project=args.project).execute_sql_result_save_file(
+        sql,
+        args.output_file,
+    )
+    print(f"[CSV] {output_path.resolve()}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

skills/odps-growth-fission-report/SKILL.md

@@ -0,0 +1,52 @@
+---
+name: odps-growth-fission-report
+description: 生成、执行、校验并解释参数化 ODPS 首层增长和 T0 裂变实验报表，包含首层 UV、曝光、播放、分享、裂变 UV、STR、T0 裂变率和单位曝光裂变 UV，并按头部、推荐、全部流量拆分。用于产品类型、离线或实时模式、日期、分桶位置、实验尾号、版本、首层规则或企微排除条件变化的查询。
+---
+
+# ODPS 增长裂变报表
+
+将 `SKILL_DIR` 解析为当前 `SKILL.md` 所在的安装目录。所有脚本和参考资料都相对该目录解析；不得假定当前工作目录或作者仓库。
+
+## 运行环境
+
+依赖 `pyodps` 和 `pandas`。随 Skill 分发的 ODPS 客户端读取 `ODPS_ACCESS_ID`、`ODPS_ACCESS_SECRET`、`ODPS_PROJECT` 和 `ODPS_ENDPOINT`。不得把这些值写入请求 JSON、SQL、输出文件或回复。
+
+## 工作流程
+
+1. 阅读 [metrics.md](references/metrics.md) 和 [raw-output-contract.md](references/raw-output-contract.md)。
+2. 在用户指定输出目录把参数写入 `request.json`。
+3. 规范化并校验参数：
+
+   `python3 "$SKILL_DIR/scripts/normalize_request.py" request.json > normalized.json`
+
+4. 明确回显日期、首层规则、模式、数据表、分桶位置和分组、版本策略及企微处理方式。
+5. 生成一个符合原始事实契约的参数化 SQL。不得搜索或依赖外部 `AGENTS.md`、特定日期 SQL 或本地 runner。
+6. 把所有参数一致应用到首层身份、视频行为、源分享和 click 归因，SQL 必须为每天生成完整 16 个桶。
+7. 预检人群口径、同日身份关联、来源归因以及版本和渠道过滤。
+8. 只提交一个 ODPS 实例：
+
+   `python3 "$SKILL_DIR/scripts/run_sql.py" query.sql raw_facts.csv`
+
+9. 生成标准完整报表和聚合报表：
+
+   `python3 "$SKILL_DIR/scripts/format_report.py" normalized.json raw_facts.csv full_report.csv --aggregate-output aggregate_report.csv`
+
+10. 执行 [metrics.md](references/metrics.md) 中的全部校验，并分别解释头部、推荐和全部流量。
+
+## 参数规则
+
+- 必须提供 `app_type`、`date_from`、`date_to`、`data_mode`、`bucket_position_from_end`、`experiment_buckets` 和 `version`。
+- 未明确给出对照桶时，自动使用 `0-f` 中实验桶的补集。
+- `first_layer_rule` 默认 `special-layer-compatible`；只有用户明确要求旧口径时才使用 `depth-zero`。
+- `version: "all"` 表示不限制版本；指定版本时限制首层用户、其视频行为和源分享，不限制回流 click 用户。
+- 实时模式只支持一个自然日；已完成的历史日期使用离线数据源。
+- `exclude_qywx` 默认 false；为 true 时必须贯穿符合条件的来源链路。
+- 裂变只归因于首层用户同日发出的源分享及相同 root session 带回的同日 click 用户。
+- 对照组只有一个桶时，必须说明单桶波动风险。
+
+## 资源
+
+- `scripts/normalize_request.py`：参数规范化与校验。
+- `scripts/odps_module.py`：基于环境变量的 ODPS 连接。
+- `scripts/run_sql.py`：可移植的 SQL 转 CSV 执行器。
+- `scripts/format_report.py`：确定性的 65 列格式器及聚合输出器。

skills/odps-growth-fission-report/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "增长裂变实验报表"
+  short_description: "按产品日期数据源分桶位置实验尾号和版本生成首层增长裂变实验报表"
+  default_prompt: "使用 $odps-growth-fission-report 查询指定产品、日期、分桶和版本的增长裂变数据。"

skills/odps-growth-fission-report/references/metrics.md

@@ -0,0 +1,55 @@
+# Growth-fission metric contract
+
+## Source selection
+
+| Mode | First-layer identity | Video actions | Source share and click |
+|---|---|---|---|
+| offline | `loghubods.useractive_log` | `loghubods.video_action_log_applet` | `loghubods.user_share_log` |
+| realtime | `loghubods.useractive_log_per5min` | `loghubods.video_action_log_flow` | `loghubods.user_share_log_per5min` |
+
+## Population and attribution
+
+- Start from `businesstype='path'` useractive records.
+- `special-layer-compatible`: `(userShareDepth='0' AND COALESCE(isSpecialLayer,'0')='0') OR (userShareDepth='1' AND isSpecialLayer='1')`.
+- `depth-zero`: `userShareDepth='0'`.
+- First-layer UV: distinct eligible `machinecode` per date and tail.
+- Join eligible video events by date, `mid=machinecode`, and exact `rootSessionId`.
+- Source shares must be made by the same first-layer `machinecode + rootSessionId` on the same date.
+- Fission layer UV: distinct same-day click `machinecode` returned by those source-share root sessions.
+- Returned click users do not need a version or layer field unless explicitly requested.
+
+## Bucket and source rules
+
+- For position `n`, bucket with `LOWER(SUBSTR(rootSessionId, LENGTH(rootSessionId) - n + 1, 1))`; keep only `0-f`.
+- Head: `pagesource RLIKE 'user-videos-share$'`.
+- Recommendation: `pagesource RLIKE '(detail|category|recommend)$'`.
+- All: no pagesource restriction.
+- Source-split fission is attributed by the source share's pagesource.
+
+## Facts and rates
+
+- Facts: first-layer UV; exposure/play/share PV and UV; fission-layer UV.
+- Exposure/play/share use `videoView`, `videoPlay`, and `videoShareFriend`.
+- Exposure, play, share per first-layer user: respective PV / first-layer UV.
+- STR: share PV / exposure PV.
+- T0 fission rate: fission-layer UV / first-layer UV.
+- Fission yield: fission-layer UV / exposure PV.
+- Relative change: experiment aggregate rate / control aggregate rate - 1.
+- First-layer UV relative change uses experiment per-tail mean versus control per-tail mean.
+
+## Version and channel policy
+
+- Specific version: restrict first-layer useractive, eligible video, and source-share rows. Do not restrict returned click rows.
+- All versions: omit version filters and label output `全部`.
+- `exclude_qywx=true`: exclude records with `rootSourceId/rootsourceid` starting `dyyqw` at each source stage.
+- `exclude_qywx=false`: do not add channel exclusions.
+
+## Output validation
+
+- Retain the established Chinese 65-column order for head, recommendation, and all.
+- Every date contains 16 detail tails, two aggregates, and two per-tail means.
+- Facts are integers; aggregates equal member-tail sums.
+- Use actual experiment/control tail counts in means and first-layer UV comparison.
+- Sort dates descending and tails ascending.
+- Cross-check a T0-only result, when generated, against full-report first-layer UV and all fission UV tail by tail.
+- Call out first-layer rule, channel exclusion, and version policy in the final response; these materially change the population.

skills/odps-growth-fission-report/references/raw-output-contract.md

@@ -0,0 +1,37 @@
+# Growth-fission raw output contract
+
+Generate one row per `stat_date + bucket`. Every date must contain exactly the 16 lowercase buckets `0-f`; create a bucket spine and left join zero facts when necessary.
+
+## Identity columns
+
+- `stat_date`: `yyyyMMdd`
+- `app_type`: requested product/app type
+- `version_code`: requested version or the literal `all`
+- `bucket`: lowercase `0-f`
+- `first_layer_uv`: distinct eligible first-layer `machinecode`
+
+## Source fact columns
+
+For each prefix `head`, `recommend`, and `all`, emit:
+
+- `<prefix>_exposure_pv`
+- `<prefix>_exposure_uv`
+- `<prefix>_play_pv`
+- `<prefix>_play_uv`
+- `<prefix>_share_pv`
+- `<prefix>_share_uv`
+- `<prefix>_fission_uv`
+
+The complete raw result therefore has 26 columns and 16 rows per date.
+
+## SQL construction rules
+
+1. Select offline or realtime tables from normalized parameters.
+2. Build first-layer identity from `businesstype='path'` using the normalized first-layer rule.
+3. Normalize `rootSessionId`, derive the requested bucket, and keep the first-layer identity at date + machinecode + root session grain.
+4. Join video behavior to first-layer identity by same date, `mid=machinecode`, and exact root session.
+5. Attribute fission from same-day source shares by first-layer users to same-day click users returned by the same root session.
+6. Attribute head/recommendation fission using the source share's `pagesource`; globally deduplicate all-source fission users per date and bucket.
+7. Apply the requested app type, dates, version policy, and enterprise-WeChat exclusion throughout eligible source stages. Do not restrict returned click users by version.
+8. Join all facts to a `0-f` bucket spine and `COALESCE` missing counts to zero.
+9. Order by date descending and bucket ascending.

skills/odps-growth-fission-report/scripts/format_report.py

@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+"""Build the standard 65-column growth-fission report from raw ODPS facts."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+import pandas as pd
+
+
+HEX = "0123456789abcdef"
+SOURCES = ["头部", "推荐", "全部"]
+SOURCE_KEYS = {"头部": "head", "推荐": "recommend", "全部": "all"}
+FACT_SUFFIXES = ["曝光PV", "曝光UV", "播放PV", "播放UV", "分享PV", "分享UV", "裂变层UV"]
+FACT_COLUMNS = ["首层UV", *[f"{source}{suffix}" for source in SOURCES for suffix in FACT_SUFFIXES]]
+RATE_SUFFIXES = [
+    "曝光PV/首层UV",
+    "播放PV/首层UV",
+    "分享PV/首层UV",
+    "STR（分享PV/曝光PV）",
+    "T0裂变率（裂变层UV/首层UV）",
+    "裂变层UV/曝光PV",
+]
+
+
+def column_mapping() -> dict[str, str]:
+    mapping = {
+        "stat_date": "日期",
+        "app_type": "产品类型",
+        "version_code": "版本号",
+        "bucket": "尾号",
+        "first_layer_uv": "首层UV",
+    }
+    fields = {
+        "exposure_pv": "曝光PV",
+        "exposure_uv": "曝光UV",
+        "play_pv": "播放PV",
+        "play_uv": "播放UV",
+        "share_pv": "分享PV",
+        "share_uv": "分享UV",
+        "fission_uv": "裂变层UV",
+    }
+    for source, key in SOURCE_KEYS.items():
+        for field, label in fields.items():
+            mapping[f"{key}_{field}"] = f"{source}{label}"
+    return mapping
+
+
+def compact_buckets(buckets: list[str]) -> str:
+    indexes = sorted(HEX.index(bucket) for bucket in buckets)
+    groups: list[str] = []
+    start = previous = indexes[0]
+    for current in indexes[1:] + [None]:
+        if current is not None and current == previous + 1:
+            previous = current
+            continue
+        groups.append(HEX[start] if start == previous else f"{HEX[start]}-{HEX[previous]}")
+        if current is not None:
+            start = previous = current
+    return ",".join(groups)
+
+
+def format_percent(value) -> str:
+    return "" if pd.isna(value) else f"{value * 100:.4f}%"
+
+
+def add_rates(data: pd.DataFrame) -> pd.DataFrame:
+    result = data.copy()
+    first_layer_uv = result["首层UV"].replace(0, pd.NA)
+    for source in SOURCES:
+        exposure = result[f"{source}曝光PV"].replace(0, pd.NA)
+        result[f"{source}曝光PV/首层UV"] = result[f"{source}曝光PV"] / first_layer_uv
+        result[f"{source}播放PV/首层UV"] = result[f"{source}播放PV"] / first_layer_uv
+        result[f"{source}分享PV/首层UV"] = result[f"{source}分享PV"] / first_layer_uv
+        result[f"{source}STR（分享PV/曝光PV）"] = result[f"{source}分享PV"] / exposure
+        result[f"{source}T0裂变率（裂变层UV/首层UV）"] = result[f"{source}裂变层UV"] / first_layer_uv
+        result[f"{source}裂变层UV/曝光PV"] = result[f"{source}裂变层UV"] / exposure
+    return result
+
+
+def add_relative_changes(
+    data: pd.DataFrame,
+    experiment_label: str,
+    control_label: str,
+    experiment_bucket_count: int,
+    control_bucket_count: int,
+) -> pd.DataFrame:
+    result = data.copy()
+    result["首层UV相对对照组变化率"] = ""
+    for stat_date in result["日期"].unique():
+        daily = result["日期"] == stat_date
+        exp = result.index[daily & (result["行类型"] == "分组聚合") & (result["分组"] == experiment_label)]
+        ctrl = result.index[daily & (result["行类型"] == "分组聚合") & (result["分组"] == control_label)]
+        if len(exp) == len(ctrl) == 1:
+            exp_mean = result.loc[exp[0], "首层UV"] / experiment_bucket_count
+            ctrl_mean = result.loc[ctrl[0], "首层UV"] / control_bucket_count
+            if ctrl_mean:
+                change = format_percent(exp_mean / ctrl_mean - 1)
+                mask = daily & (result["分组"] == experiment_label) & result["行类型"].isin(["分组聚合", "每桶均值"])
+                result.loc[mask, "首层UV相对对照组变化率"] = change
+
+        for source in SOURCES:
+            for suffix in RATE_SUFFIXES:
+                metric = f"{source}{suffix}"
+                change_column = f"{metric}相对对照组变化率"
+                if change_column not in result:
+                    result[change_column] = ""
+                for row_type in ["分组聚合", "每桶均值"]:
+                    exp_row = result.index[daily & (result["行类型"] == row_type) & (result["分组"] == experiment_label)]
+                    ctrl_row = result.index[daily & (result["行类型"] == row_type) & (result["分组"] == control_label)]
+                    if len(exp_row) == len(ctrl_row) == 1:
+                        control = result.loc[ctrl_row[0], metric]
+                        if pd.notna(control) and control != 0:
+                            result.loc[exp_row[0], change_column] = format_percent(
+                                result.loc[exp_row[0], metric] / control - 1
+                            )
+    return result
+
+
+def format_output(data: pd.DataFrame) -> pd.DataFrame:
+    result = data.copy()
+    for source in SOURCES:
+        for suffix in ["曝光PV/首层UV", "播放PV/首层UV", "分享PV/首层UV"]:
+            result[f"{source}{suffix}"] = result[f"{source}{suffix}"].round(4)
+        for suffix in ["STR（分享PV/曝光PV）", "T0裂变率（裂变层UV/首层UV）", "裂变层UV/曝光PV"]:
+            result[f"{source}{suffix}"] = result[f"{source}{suffix}"].map(format_percent)
+    result[FACT_COLUMNS] = result[FACT_COLUMNS].round(0).astype("Int64")
+
+    ordered = ["日期", "产品类型", "版本号", "行类型", "分组", "尾号", "首层UV", "首层UV相对对照组变化率"]
+    for source in SOURCES:
+        ordered.extend(
+            [
+                f"{source}曝光PV",
+                f"{source}曝光UV",
+                f"{source}曝光PV/首层UV",
+                f"{source}曝光PV/首层UV相对对照组变化率",
+                f"{source}播放PV",
+                f"{source}播放UV",
+                f"{source}播放PV/首层UV",
+                f"{source}播放PV/首层UV相对对照组变化率",
+                f"{source}分享PV",
+                f"{source}分享UV",
+                f"{source}分享PV/首层UV",
+                f"{source}分享PV/首层UV相对对照组变化率",
+                f"{source}STR（分享PV/曝光PV）",
+                f"{source}STR（分享PV/曝光PV）相对对照组变化率",
+                f"{source}裂变层UV",
+                f"{source}T0裂变率（裂变层UV/首层UV）",
+                f"{source}T0裂变率（裂变层UV/首层UV）相对对照组变化率",
+                f"{source}裂变层UV/曝光PV",
+                f"{source}裂变层UV/曝光PV相对对照组变化率",
+            ]
+        )
+    return result[ordered]
+
+
+def validate_raw(data: pd.DataFrame) -> None:
+    expected = set(column_mapping())
+    missing = sorted(expected - set(data.columns))
+    if missing:
+        raise ValueError(f"raw facts are missing columns: {missing}")
+
+    duplicates = data.duplicated(["stat_date", "bucket"], keep=False)
+    if duplicates.any():
+        raise ValueError("raw facts contain duplicate date/bucket rows")
+
+    for stat_date, daily in data.groupby("stat_date"):
+        buckets = set(daily["bucket"].astype(str).str.lower())
+        if buckets != set(HEX):
+            raise ValueError(f"{stat_date} must contain exactly buckets 0-f; got {sorted(buckets)}")
+        if daily["app_type"].nunique() != 1 or daily["version_code"].nunique() != 1:
+            raise ValueError(f"{stat_date} contains multiple app types or versions")
+
+
+def build_report(raw: pd.DataFrame, config: dict) -> pd.DataFrame:
+    validate_raw(raw)
+    experiment = [str(item) for item in config["experiment_buckets"]]
+    control = [str(item) for item in config["control_buckets"]]
+    exp_label = f"实验组（{compact_buckets(experiment)}）"
+    ctrl_label = f"对照组（{compact_buckets(control)}）"
+
+    data = raw.rename(columns=column_mapping()).copy()
+    for column in FACT_COLUMNS:
+        data[column] = pd.to_numeric(data[column], errors="raise")
+    for column in ["日期", "产品类型", "版本号", "尾号"]:
+        data[column] = data[column].astype(str)
+    data["尾号"] = data["尾号"].str.lower()
+    data["行类型"] = "尾号明细"
+    data["分组"] = data["尾号"].map(lambda bucket: exp_label if bucket in experiment else ctrl_label)
+
+    reports = []
+    for stat_date in sorted(data["日期"].unique(), reverse=True):
+        detail = data[data["日期"] == stat_date].copy()
+        detail["_bucket_order"] = detail["尾号"].map(HEX.index)
+        detail = detail.sort_values("_bucket_order").drop(columns="_bucket_order")
+        detail = detail[["日期", "产品类型", "版本号", "行类型", "分组", "尾号", *FACT_COLUMNS]]
+
+        aggregate_rows = []
+        mean_rows = []
+        for group, buckets in [(exp_label, experiment), (ctrl_label, control)]:
+            selected = detail[detail["尾号"].isin(buckets)]
+            common = {
+                "日期": stat_date,
+                "产品类型": detail["产品类型"].iloc[0],
+                "版本号": detail["版本号"].iloc[0],
+                "分组": group,
+            }
+            aggregate_rows.append(
+                {
+                    **common,
+                    "行类型": "分组聚合",
+                    "尾号": compact_buckets(buckets),
+                    **selected[FACT_COLUMNS].sum().to_dict(),
+                }
+            )
+            mean_rows.append(
+                {
+                    **common,
+                    "行类型": "每桶均值",
+                    "尾号": f"{len(buckets)}桶均值",
+                    **selected[FACT_COLUMNS].mean().to_dict(),
+                }
+            )
+        reports.append(pd.concat([detail, pd.DataFrame(aggregate_rows), pd.DataFrame(mean_rows)], ignore_index=True))
+
+    result = pd.concat(reports, ignore_index=True)
+    result = add_rates(result)
+    result = add_relative_changes(result, exp_label, ctrl_label, len(experiment), len(control))
+    return format_output(result)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Format raw growth-fission facts")
+    parser.add_argument("normalized_request", type=Path)
+    parser.add_argument("raw_facts_csv", type=Path)
+    parser.add_argument("full_report_csv", type=Path)
+    parser.add_argument("--aggregate-output", type=Path)
+    args = parser.parse_args()
+
+    config = json.loads(args.normalized_request.read_text(encoding="utf-8"))
+    if config.get("report_kind") != "growth_fission":
+        raise ValueError("normalized request is not a growth-fission request")
+
+    raw = pd.read_csv(
+        args.raw_facts_csv,
+        dtype={"stat_date": str, "app_type": str, "version_code": str, "bucket": str},
+    )
+    report = build_report(raw, config)
+
+    args.full_report_csv.parent.mkdir(parents=True, exist_ok=True)
+    report.to_csv(args.full_report_csv, index=False, encoding="utf-8-sig")
+    print(f"[CSV] {args.full_report_csv.resolve()} ({len(report)} rows)", flush=True)
+
+    if args.aggregate_output:
+        aggregate = report[report["行类型"] == "分组聚合"].copy()
+        args.aggregate_output.parent.mkdir(parents=True, exist_ok=True)
+        aggregate.to_csv(args.aggregate_output, index=False, encoding="utf-8-sig")
+        print(f"[CSV] {args.aggregate_output.resolve()} ({len(aggregate)} rows)", flush=True)
+
+
+if __name__ == "__main__":
+    main()

skills/odps-growth-fission-report/scripts/normalize_request.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import re
+from datetime import datetime, timedelta
+from pathlib import Path
+
+
+HEX = "0123456789abcdef"
+
+
+def parse_date(value):
+    try:
+        return datetime.strptime(str(value), "%Y%m%d").date()
+    except ValueError as exc:
+        raise ValueError(f"invalid YYYYMMDD date: {value}") from exc
+
+
+def parse_buckets(value):
+    if isinstance(value, list):
+        tokens = [str(item).lower() for item in value]
+    else:
+        raw = str(value).lower().replace("、", ",").replace(" ", "")
+        tokens = []
+        for item in raw.split(","):
+            if not item:
+                continue
+            if re.fullmatch(r"[0-9a-f]-[0-9a-f]", item):
+                start, end = HEX.index(item[0]), HEX.index(item[2])
+                if start > end:
+                    raise ValueError(f"descending bucket range: {item}")
+                tokens.extend(HEX[start:end + 1])
+            else:
+                tokens.append(item)
+    invalid = sorted(set(tokens) - set(HEX))
+    if invalid:
+        raise ValueError(f"invalid buckets: {invalid}")
+    return [bucket for bucket in HEX if bucket in set(tokens)]
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Normalize ODPS growth-fission report parameters")
+    parser.add_argument("request", type=Path)
+    args = parser.parse_args()
+    data = json.loads(args.request.read_text(encoding="utf-8"))
+
+    required = ["app_type", "date_from", "date_to", "data_mode", "bucket_position_from_end", "experiment_buckets", "version"]
+    missing = [key for key in required if key not in data]
+    if missing:
+        raise ValueError(f"missing required parameters: {missing}")
+
+    start, end = parse_date(data["date_from"]), parse_date(data["date_to"])
+    if start > end:
+        raise ValueError("date_from must not exceed date_to")
+    mode = str(data["data_mode"]).lower()
+    if mode not in {"offline", "realtime"}:
+        raise ValueError("data_mode must be offline or realtime")
+    if mode == "realtime" and start != end:
+        raise ValueError("realtime mode supports one calendar date")
+
+    position = int(data["bucket_position_from_end"])
+    if position < 1:
+        raise ValueError("bucket_position_from_end must be positive")
+    experiment = parse_buckets(data["experiment_buckets"])
+    if not experiment:
+        raise ValueError("experiment_buckets must not be empty")
+    control = parse_buckets(data["control_buckets"]) if data.get("control_buckets") is not None else [b for b in HEX if b not in experiment]
+    if not control or set(experiment) & set(control):
+        raise ValueError("experiment and control buckets must be non-empty and disjoint")
+    if set(experiment) | set(control) != set(HEX):
+        raise ValueError("experiment and control buckets must cover 0-f")
+
+    rule = data.get("first_layer_rule", "special-layer-compatible")
+    if rule not in {"special-layer-compatible", "depth-zero"}:
+        raise ValueError("first_layer_rule must be special-layer-compatible or depth-zero")
+    version = str(data["version"])
+    version_code = None if version.lower() == "all" else version
+    dates = []
+    current = start
+    while current <= end:
+        dates.append(current.strftime("%Y%m%d"))
+        current += timedelta(days=1)
+
+    tables = {
+        "offline": {"useractive": "loghubods.useractive_log", "video": "loghubods.video_action_log_applet", "share": "loghubods.user_share_log"},
+        "realtime": {"useractive": "loghubods.useractive_log_per5min", "video": "loghubods.video_action_log_flow", "share": "loghubods.user_share_log_per5min"},
+    }[mode]
+    normalized = {
+        "report_kind": "growth_fission",
+        "app_type": str(data["app_type"]),
+        "date_from": dates[0],
+        "date_to": dates[-1],
+        "dates": dates,
+        "date_count": len(dates),
+        "data_mode": mode,
+        "tables": tables,
+        "bucket_position_from_end": position,
+        "bucket_substr_offset": position - 1,
+        "experiment_buckets": experiment,
+        "control_buckets": control,
+        "experiment_bucket_count": len(experiment),
+        "control_bucket_count": len(control),
+        "version": "all" if version_code is None else "specific",
+        "version_code": version_code,
+        "first_layer_rule": rule,
+        "exclude_qywx": bool(data.get("exclude_qywx", False)),
+        "output_dir": data.get("output_dir", "."),
+    }
+    print(json.dumps(normalized, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

skills/odps-growth-fission-report/scripts/odps_module.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Minimal PyODPS client configured only through environment variables."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from odps import ODPS, options
+
+
+ENV_ACCESS_ID = "ODPS_ACCESS_ID"
+ENV_ACCESS_SECRET = "ODPS_ACCESS_SECRET"
+ENV_PROJECT = "ODPS_PROJECT"
+ENV_ENDPOINT = "ODPS_ENDPOINT"
+
+
+def require_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(f"Missing required environment variable: {name}")
+    return value
+
+
+class ODPSClient:
+    def __init__(self, project: str | None = None, endpoint: str | None = None):
+        access_id = require_env(ENV_ACCESS_ID)
+        access_secret = require_env(ENV_ACCESS_SECRET)
+        project_name = project or require_env(ENV_PROJECT)
+        endpoint_url = endpoint or require_env(ENV_ENDPOINT)
+
+        options.connect_timeout = 60
+        options.read_timeout = 1200
+        options.retry_times = 3
+
+        self.odps = ODPS(
+            access_id,
+            access_secret,
+            project=project_name,
+            endpoint=endpoint_url,
+        )
+
+    def execute_sql(self, sql: str):
+        if not sql.strip():
+            raise ValueError("SQL must not be empty")
+
+        instance = self.odps.run_sql(
+            sql,
+            hints={"odps.sql.submit.mode": "script"},
+        )
+        print(f"[ODPS] InstanceId: {instance.id}", flush=True)
+        instance.wait_for_success()
+        with instance.open_reader(tunnel=True) as reader:
+            return reader.to_pandas()
+
+    def execute_sql_result_save_file(self, sql: str, output_file: str | Path):
+        output_path = Path(output_file).expanduser()
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        data = self.execute_sql(sql)
+        data.to_csv(output_path, index=False, encoding="utf-8-sig")
+        return output_path

skills/odps-growth-fission-report/scripts/run_sql.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Execute an explicit SQL file with the bundled environment-based ODPS client."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from odps_module import ODPSClient
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run an ODPS SQL file and save UTF-8 CSV")
+    parser.add_argument("sql_file", type=Path)
+    parser.add_argument("output_file", type=Path)
+    parser.add_argument("--project", help="Override ODPS_PROJECT for this query")
+    args = parser.parse_args()
+
+    sql = args.sql_file.expanduser().read_text(encoding="utf-8")
+    output_path = ODPSClient(project=args.project).execute_sql_result_save_file(
+        sql,
+        args.output_file,
+    )
+    print(f"[CSV] {output_path.resolve()}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

skills/odps-product-efficiency-report/SKILL.md

@@ -0,0 +1,50 @@
+---
+name: odps-product-efficiency-report
+description: 生成、执行、校验并解释参数化 ODPS 产品效率实验报表，包含 DAU、曝光、播放、分享、回流、STR 和 ROV，并按头部、推荐、全部流量拆分。用于产品类型、离线或实时模式、日期范围、rootSessionId 分桶位置、实验/对照尾号、版本过滤或企微排除条件发生变化的查询。
+---
+
+# ODPS 产品效率报表
+
+将 `SKILL_DIR` 解析为当前 `SKILL.md` 所在的安装目录。所有脚本和参考资料都相对该目录解析；不得假定当前工作目录或作者仓库。
+
+## 运行环境
+
+依赖 `pyodps` 和 `pandas`。随 Skill 分发的 ODPS 客户端读取 `ODPS_ACCESS_ID`、`ODPS_ACCESS_SECRET`、`ODPS_PROJECT` 和 `ODPS_ENDPOINT`。不得把这些值写入请求 JSON、SQL、输出文件或回复。
+
+## 工作流程
+
+1. 阅读 [metrics.md](references/metrics.md) 和 [raw-output-contract.md](references/raw-output-contract.md)。
+2. 在用户指定输出目录把参数写入 `request.json`。
+3. 规范化并校验参数：
+
+   `python3 "$SKILL_DIR/scripts/normalize_request.py" request.json > normalized.json`
+
+4. 明确回显日期、模式、数据表、分桶位置和分组、版本策略及企微处理方式。
+5. 生成一个符合原始事实契约的参数化 SQL。不得搜索或依赖外部 `AGENTS.md`、特定日期 SQL 或本地 runner。
+6. 预检全部日期、数据表、产品类型、分桶表达式、版本条件、分组标签和渠道排除条件。SQL 必须为每天生成完整 16 个桶。
+7. 只提交一个 ODPS 实例：
+
+   `python3 "$SKILL_DIR/scripts/run_sql.py" query.sql raw_facts.csv`
+
+8. 生成标准完整报表和聚合报表：
+
+   `python3 "$SKILL_DIR/scripts/format_report.py" normalized.json raw_facts.csv full_report.csv --aggregate-output aggregate_report.csv`
+
+9. 执行 [metrics.md](references/metrics.md) 中的全部校验，并分别解释头部、推荐和全部流量。
+
+## 参数规则
+
+- 必须提供 `app_type`、`date_from`、`date_to`、`data_mode`、`bucket_position_from_end`、`experiment_buckets` 和 `version`。
+- 未明确给出对照桶时，自动使用 `0-f` 中实验桶的补集。
+- `version: "all"` 表示不限制版本；指定版本时限制 DAU、视频行为和源分享，不限制回流接收用户。
+- 实时模式只支持一个自然日；已完成的历史日期使用离线数据源。
+- `exclude_qywx` 默认 false；为 true 时必须贯穿所有适用来源。
+- DAU 每桶比较必须使用真实实验桶数和对照桶数。
+- 对照组只有一个桶时，必须说明单桶波动风险。
+
+## 资源
+
+- `scripts/normalize_request.py`：参数规范化与校验。
+- `scripts/odps_module.py`：基于环境变量的 ODPS 连接。
+- `scripts/run_sql.py`：可移植的 SQL 转 CSV 执行器。
+- `scripts/format_report.py`：确定性的 65 列格式器及聚合输出器。

skills/odps-product-efficiency-report/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "产品效率实验报表"
+  short_description: "按产品日期数据源分桶位置实验尾号和版本生成整体效率实验对比报表"
+  default_prompt: "使用 $odps-product-efficiency-report 查询指定产品、日期、分桶和版本的整体效率数据。"

skills/odps-product-efficiency-report/references/metrics.md

@@ -0,0 +1,46 @@
+# Product-efficiency metric contract
+
+## Source selection
+
+| Mode | DAU | Video actions | Share and click |
+|---|---|---|---|
+| offline | `loghubods.useractive_log` | `loghubods.video_action_log_applet` | `loghubods.user_share_log` |
+| realtime | `loghubods.useractive_log_per5min` | `loghubods.video_action_log_flow` | `loghubods.user_share_log_per5min` |
+
+For realtime video-flow partitions, set year, month, and day consistently. For offline tables, group every fact by `dt`; never deduplicate across dates.
+
+## Bucket and source definitions
+
+- Read `rootSessionId` from the applicable physical column or `extparams`.
+- For position `n` from the end, use `LOWER(SUBSTR(rootSessionId, LENGTH(rootSessionId) - n + 1, 1))` and retain only `^[0-9a-f]$`.
+- Head: `pagesource RLIKE 'user-videos-share$'`.
+- Recommendation: `pagesource RLIKE '(detail|category|recommend)$'`.
+- All: no pagesource restriction.
+
+## Facts and rates
+
+- DAU: distinct `machinecode` from useractive.
+- Exposure: `businesstype='videoView'`; PV and distinct `mid` UV.
+- Play: `businesstype='videoPlay'`; PV and distinct `mid` UV.
+- Share: `businesstype='videoShareFriend'`; PV and distinct `mid` UV.
+- Return: source share and same-day `topic='click'` joined by `shareid`; distinct click `machinecode` UV.
+- STR: share PV / exposure PV.
+- ROV: return UV / exposure PV.
+- Per-user metrics: PV or return UV / same-tail DAU.
+- DAU relative change: experiment per-tail mean / control per-tail mean - 1.
+- Other relative changes: experiment aggregate rate / control aggregate rate - 1.
+
+## Version policy
+
+When a version is specified, filter DAU, video actions, and source-share rows to that version. Do not restrict recipient click rows by version. When version is `all`, omit all version filters and output version label `全部`.
+
+## Output and validation
+
+- Chinese facts: DAU plus exposure/play/share PV/UV and return UV for head, recommendation, and all.
+- Interleave each fact group with its rates and experiment-relative changes; retain the established 65-column order.
+- Fact cells must be integers. Per-user numeric rates use four decimals; percentages use four decimal places.
+- Per date: exactly 16 distinct detail tails `0-f`, two aggregates, and two per-tail means, for 20 rows.
+- Aggregate facts must equal the sum of their member detail tails.
+- Per-tail means must use the actual number of tails in each group.
+- Sort dates descending and detail tails ascending.
+- Treat a one-tail control as more volatile and say so in the analysis.

skills/odps-product-efficiency-report/references/raw-output-contract.md

@@ -0,0 +1,36 @@
+# Product-efficiency raw output contract
+
+Generate one row per `stat_date + bucket`. Every date must contain exactly the 16 lowercase buckets `0-f`; create a bucket spine and left join zero facts when necessary.
+
+## Identity columns
+
+- `stat_date`: `yyyyMMdd`
+- `app_type`: requested product/app type
+- `version_code`: requested version or the literal `all`
+- `bucket`: lowercase `0-f`
+- `dau`: distinct eligible `machinecode`
+
+## Source fact columns
+
+For each prefix `head`, `recommend`, and `all`, emit:
+
+- `<prefix>_exposure_pv`
+- `<prefix>_exposure_uv`
+- `<prefix>_play_pv`
+- `<prefix>_play_uv`
+- `<prefix>_share_pv`
+- `<prefix>_share_uv`
+- `<prefix>_return_uv`
+
+The complete raw result therefore has 26 columns and 16 rows per date.
+
+## SQL construction rules
+
+1. Select offline or realtime tables from normalized parameters.
+2. Normalize `rootSessionId` from the documented physical column or `extparams`.
+3. Derive the bucket using the normalized `bucket_substr_offset`.
+4. Aggregate every fact independently by date and bucket before joining.
+5. Apply the requested app type, dates, version policy, and enterprise-WeChat exclusion to every applicable source.
+6. Attribute return users by same-day source share and click joined by `shareid`; do not restrict returned users by version.
+7. Join all facts to a `0-f` bucket spine and `COALESCE` missing counts to zero.
+8. Order by date descending and bucket ascending.

skills/odps-product-efficiency-report/scripts/format_report.py

@@ -0,0 +1,264 @@
+#!/usr/bin/env python3
+"""Build the standard 65-column product-efficiency report from raw ODPS facts."""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+import pandas as pd
+
+
+HEX = "0123456789abcdef"
+SOURCES = ["头部", "推荐", "全部"]
+SOURCE_KEYS = {"头部": "head", "推荐": "recommend", "全部": "all"}
+FACT_SUFFIXES = ["曝光PV", "曝光UV", "播放PV", "播放UV", "分享PV", "分享UV", "回流UV"]
+FACT_COLUMNS = ["DAU", *[f"{source}{suffix}" for source in SOURCES for suffix in FACT_SUFFIXES]]
+RATE_SUFFIXES = [
+    "曝光PV/DAU",
+    "播放PV/DAU",
+    "分享PV/DAU",
+    "回流UV/DAU",
+    "STR（分享PV/曝光PV）",
+    "ROV（回流UV/曝光PV）",
+]
+
+
+def column_mapping() -> dict[str, str]:
+    mapping = {
+        "stat_date": "日期",
+        "app_type": "产品类型",
+        "version_code": "版本号",
+        "bucket": "尾号",
+        "dau": "DAU",
+    }
+    fields = {
+        "exposure_pv": "曝光PV",
+        "exposure_uv": "曝光UV",
+        "play_pv": "播放PV",
+        "play_uv": "播放UV",
+        "share_pv": "分享PV",
+        "share_uv": "分享UV",
+        "return_uv": "回流UV",
+    }
+    for source, key in SOURCE_KEYS.items():
+        for field, label in fields.items():
+            mapping[f"{key}_{field}"] = f"{source}{label}"
+    return mapping
+
+
+def compact_buckets(buckets: list[str]) -> str:
+    indexes = sorted(HEX.index(bucket) for bucket in buckets)
+    groups: list[str] = []
+    start = previous = indexes[0]
+    for current in indexes[1:] + [None]:
+        if current is not None and current == previous + 1:
+            previous = current
+            continue
+        groups.append(HEX[start] if start == previous else f"{HEX[start]}-{HEX[previous]}")
+        if current is not None:
+            start = previous = current
+    return ",".join(groups)
+
+
+def format_percent(value) -> str:
+    return "" if pd.isna(value) else f"{value * 100:.4f}%"
+
+
+def add_rates(data: pd.DataFrame) -> pd.DataFrame:
+    result = data.copy()
+    dau = result["DAU"].replace(0, pd.NA)
+    for source in SOURCES:
+        exposure = result[f"{source}曝光PV"].replace(0, pd.NA)
+        result[f"{source}曝光PV/DAU"] = result[f"{source}曝光PV"] / dau
+        result[f"{source}播放PV/DAU"] = result[f"{source}播放PV"] / dau
+        result[f"{source}分享PV/DAU"] = result[f"{source}分享PV"] / dau
+        result[f"{source}回流UV/DAU"] = result[f"{source}回流UV"] / dau
+        result[f"{source}STR（分享PV/曝光PV）"] = result[f"{source}分享PV"] / exposure
+        result[f"{source}ROV（回流UV/曝光PV）"] = result[f"{source}回流UV"] / exposure
+    return result
+
+
+def add_relative_changes(
+    data: pd.DataFrame,
+    experiment_label: str,
+    control_label: str,
+    experiment_bucket_count: int,
+    control_bucket_count: int,
+) -> pd.DataFrame:
+    result = data.copy()
+    result["DAU相对对照组变化率"] = ""
+    for stat_date in result["日期"].unique():
+        daily = result["日期"] == stat_date
+        exp = result.index[daily & (result["行类型"] == "分组聚合") & (result["分组"] == experiment_label)]
+        ctrl = result.index[daily & (result["行类型"] == "分组聚合") & (result["分组"] == control_label)]
+        if len(exp) == len(ctrl) == 1:
+            exp_mean = result.loc[exp[0], "DAU"] / experiment_bucket_count
+            ctrl_mean = result.loc[ctrl[0], "DAU"] / control_bucket_count
+            if ctrl_mean:
+                change = format_percent(exp_mean / ctrl_mean - 1)
+                mask = daily & (result["分组"] == experiment_label) & result["行类型"].isin(["分组聚合", "每桶均值"])
+                result.loc[mask, "DAU相对对照组变化率"] = change
+
+        for source in SOURCES:
+            for suffix in RATE_SUFFIXES:
+                metric = f"{source}{suffix}"
+                change_column = f"{metric}相对对照组变化率"
+                if change_column not in result:
+                    result[change_column] = ""
+                for row_type in ["分组聚合", "每桶均值"]:
+                    exp_row = result.index[daily & (result["行类型"] == row_type) & (result["分组"] == experiment_label)]
+                    ctrl_row = result.index[daily & (result["行类型"] == row_type) & (result["分组"] == control_label)]
+                    if len(exp_row) == len(ctrl_row) == 1:
+                        control = result.loc[ctrl_row[0], metric]
+                        if pd.notna(control) and control != 0:
+                            result.loc[exp_row[0], change_column] = format_percent(
+                                result.loc[exp_row[0], metric] / control - 1
+                            )
+    return result
+
+
+def format_output(data: pd.DataFrame) -> pd.DataFrame:
+    result = data.copy()
+    for source in SOURCES:
+        for suffix in ["曝光PV/DAU", "播放PV/DAU", "分享PV/DAU"]:
+            result[f"{source}{suffix}"] = result[f"{source}{suffix}"].round(4)
+        for suffix in ["回流UV/DAU", "STR（分享PV/曝光PV）", "ROV（回流UV/曝光PV）"]:
+            result[f"{source}{suffix}"] = result[f"{source}{suffix}"].map(format_percent)
+    result[FACT_COLUMNS] = result[FACT_COLUMNS].round(0).astype("Int64")
+
+    ordered = ["日期", "产品类型", "版本号", "行类型", "分组", "尾号", "DAU", "DAU相对对照组变化率"]
+    for source in SOURCES:
+        ordered.extend(
+            [
+                f"{source}曝光PV",
+                f"{source}曝光UV",
+                f"{source}曝光PV/DAU",
+                f"{source}曝光PV/DAU相对对照组变化率",
+                f"{source}播放PV",
+                f"{source}播放UV",
+                f"{source}播放PV/DAU",
+                f"{source}播放PV/DAU相对对照组变化率",
+                f"{source}分享PV",
+                f"{source}分享UV",
+                f"{source}分享PV/DAU",
+                f"{source}分享PV/DAU相对对照组变化率",
+                f"{source}STR（分享PV/曝光PV）",
+                f"{source}STR（分享PV/曝光PV）相对对照组变化率",
+                f"{source}回流UV",
+                f"{source}回流UV/DAU",
+                f"{source}回流UV/DAU相对对照组变化率",
+                f"{source}ROV（回流UV/曝光PV）",
+                f"{source}ROV（回流UV/曝光PV）相对对照组变化率",
+            ]
+        )
+    return result[ordered]
+
+
+def validate_raw(data: pd.DataFrame) -> None:
+    expected = set(column_mapping())
+    missing = sorted(expected - set(data.columns))
+    if missing:
+        raise ValueError(f"raw facts are missing columns: {missing}")
+
+    duplicates = data.duplicated(["stat_date", "bucket"], keep=False)
+    if duplicates.any():
+        raise ValueError("raw facts contain duplicate date/bucket rows")
+
+    for stat_date, daily in data.groupby("stat_date"):
+        buckets = set(daily["bucket"].astype(str).str.lower())
+        if buckets != set(HEX):
+            raise ValueError(f"{stat_date} must contain exactly buckets 0-f; got {sorted(buckets)}")
+        if daily["app_type"].nunique() != 1 or daily["version_code"].nunique() != 1:
+            raise ValueError(f"{stat_date} contains multiple app types or versions")
+
+
+def build_report(raw: pd.DataFrame, config: dict) -> pd.DataFrame:
+    validate_raw(raw)
+    experiment = [str(item) for item in config["experiment_buckets"]]
+    control = [str(item) for item in config["control_buckets"]]
+    exp_label = f"实验组（{compact_buckets(experiment)}）"
+    ctrl_label = f"对照组（{compact_buckets(control)}）"
+
+    data = raw.rename(columns=column_mapping()).copy()
+    for column in FACT_COLUMNS:
+        data[column] = pd.to_numeric(data[column], errors="raise")
+    for column in ["日期", "产品类型", "版本号", "尾号"]:
+        data[column] = data[column].astype(str)
+    data["尾号"] = data["尾号"].str.lower()
+    data["行类型"] = "尾号明细"
+    data["分组"] = data["尾号"].map(lambda bucket: exp_label if bucket in experiment else ctrl_label)
+
+    reports = []
+    for stat_date in sorted(data["日期"].unique(), reverse=True):
+        detail = data[data["日期"] == stat_date].copy()
+        detail["_bucket_order"] = detail["尾号"].map(HEX.index)
+        detail = detail.sort_values("_bucket_order").drop(columns="_bucket_order")
+        detail = detail[["日期", "产品类型", "版本号", "行类型", "分组", "尾号", *FACT_COLUMNS]]
+
+        aggregate_rows = []
+        mean_rows = []
+        for group, buckets in [(exp_label, experiment), (ctrl_label, control)]:
+            selected = detail[detail["尾号"].isin(buckets)]
+            common = {
+                "日期": stat_date,
+                "产品类型": detail["产品类型"].iloc[0],
+                "版本号": detail["版本号"].iloc[0],
+                "分组": group,
+            }
+            aggregate_rows.append(
+                {
+                    **common,
+                    "行类型": "分组聚合",
+                    "尾号": compact_buckets(buckets),
+                    **selected[FACT_COLUMNS].sum().to_dict(),
+                }
+            )
+            mean_rows.append(
+                {
+                    **common,
+                    "行类型": "每桶均值",
+                    "尾号": f"{len(buckets)}桶均值",
+                    **selected[FACT_COLUMNS].mean().to_dict(),
+                }
+            )
+        reports.append(pd.concat([detail, pd.DataFrame(aggregate_rows), pd.DataFrame(mean_rows)], ignore_index=True))
+
+    result = pd.concat(reports, ignore_index=True)
+    result = add_rates(result)
+    result = add_relative_changes(result, exp_label, ctrl_label, len(experiment), len(control))
+    return format_output(result)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Format raw product-efficiency facts")
+    parser.add_argument("normalized_request", type=Path)
+    parser.add_argument("raw_facts_csv", type=Path)
+    parser.add_argument("full_report_csv", type=Path)
+    parser.add_argument("--aggregate-output", type=Path)
+    args = parser.parse_args()
+
+    config = json.loads(args.normalized_request.read_text(encoding="utf-8"))
+    if config.get("report_kind") != "product_efficiency":
+        raise ValueError("normalized request is not a product-efficiency request")
+
+    raw = pd.read_csv(
+        args.raw_facts_csv,
+        dtype={"stat_date": str, "app_type": str, "version_code": str, "bucket": str},
+    )
+    report = build_report(raw, config)
+
+    args.full_report_csv.parent.mkdir(parents=True, exist_ok=True)
+    report.to_csv(args.full_report_csv, index=False, encoding="utf-8-sig")
+    print(f"[CSV] {args.full_report_csv.resolve()} ({len(report)} rows)", flush=True)
+
+    if args.aggregate_output:
+        aggregate = report[report["行类型"] == "分组聚合"].copy()
+        args.aggregate_output.parent.mkdir(parents=True, exist_ok=True)
+        aggregate.to_csv(args.aggregate_output, index=False, encoding="utf-8-sig")
+        print(f"[CSV] {args.aggregate_output.resolve()} ({len(aggregate)} rows)", flush=True)
+
+
+if __name__ == "__main__":
+    main()

skills/odps-product-efficiency-report/scripts/normalize_request.py

@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+import argparse
+import json
+import re
+from datetime import datetime, timedelta
+from pathlib import Path
+
+
+HEX = "0123456789abcdef"
+
+
+def parse_date(value):
+    try:
+        return datetime.strptime(str(value), "%Y%m%d").date()
+    except ValueError as exc:
+        raise ValueError(f"invalid YYYYMMDD date: {value}") from exc
+
+
+def parse_buckets(value):
+    if isinstance(value, list):
+        tokens = [str(item).lower() for item in value]
+    else:
+        raw = str(value).lower().replace("、", ",").replace(" ", "")
+        tokens = []
+        for item in raw.split(","):
+            if not item:
+                continue
+            if re.fullmatch(r"[0-9a-f]-[0-9a-f]", item):
+                start, end = HEX.index(item[0]), HEX.index(item[2])
+                if start > end:
+                    raise ValueError(f"descending bucket range: {item}")
+                tokens.extend(HEX[start:end + 1])
+            else:
+                tokens.append(item)
+    invalid = sorted(set(tokens) - set(HEX))
+    if invalid:
+        raise ValueError(f"invalid buckets: {invalid}")
+    return [bucket for bucket in HEX if bucket in set(tokens)]
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Normalize ODPS product-efficiency report parameters")
+    parser.add_argument("request", type=Path)
+    args = parser.parse_args()
+    data = json.loads(args.request.read_text(encoding="utf-8"))
+
+    required = ["app_type", "date_from", "date_to", "data_mode", "bucket_position_from_end", "experiment_buckets", "version"]
+    missing = [key for key in required if key not in data]
+    if missing:
+        raise ValueError(f"missing required parameters: {missing}")
+
+    start, end = parse_date(data["date_from"]), parse_date(data["date_to"])
+    if start > end:
+        raise ValueError("date_from must not exceed date_to")
+    mode = str(data["data_mode"]).lower()
+    if mode not in {"offline", "realtime"}:
+        raise ValueError("data_mode must be offline or realtime")
+    if mode == "realtime" and start != end:
+        raise ValueError("realtime mode supports one calendar date")
+
+    position = int(data["bucket_position_from_end"])
+    if position < 1:
+        raise ValueError("bucket_position_from_end must be positive")
+    experiment = parse_buckets(data["experiment_buckets"])
+    if not experiment:
+        raise ValueError("experiment_buckets must not be empty")
+    control = parse_buckets(data["control_buckets"]) if data.get("control_buckets") is not None else [b for b in HEX if b not in experiment]
+    if not control or set(experiment) & set(control):
+        raise ValueError("experiment and control buckets must be non-empty and disjoint")
+    if set(experiment) | set(control) != set(HEX):
+        raise ValueError("experiment and control buckets must cover 0-f")
+
+    version = str(data["version"])
+    version_code = None if version.lower() == "all" else version
+    dates = []
+    current = start
+    while current <= end:
+        dates.append(current.strftime("%Y%m%d"))
+        current += timedelta(days=1)
+
+    tables = {
+        "offline": {"dau": "loghubods.useractive_log", "video": "loghubods.video_action_log_applet", "share": "loghubods.user_share_log"},
+        "realtime": {"dau": "loghubods.useractive_log_per5min", "video": "loghubods.video_action_log_flow", "share": "loghubods.user_share_log_per5min"},
+    }[mode]
+    normalized = {
+        "report_kind": "product_efficiency",
+        "app_type": str(data["app_type"]),
+        "date_from": dates[0],
+        "date_to": dates[-1],
+        "dates": dates,
+        "date_count": len(dates),
+        "data_mode": mode,
+        "tables": tables,
+        "bucket_position_from_end": position,
+        "bucket_substr_offset": position - 1,
+        "experiment_buckets": experiment,
+        "control_buckets": control,
+        "experiment_bucket_count": len(experiment),
+        "control_bucket_count": len(control),
+        "version": "all" if version_code is None else "specific",
+        "version_code": version_code,
+        "exclude_qywx": bool(data.get("exclude_qywx", False)),
+        "output_dir": data.get("output_dir", "."),
+    }
+    print(json.dumps(normalized, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

skills/odps-product-efficiency-report/scripts/odps_module.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Minimal PyODPS client configured only through environment variables."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from odps import ODPS, options
+
+
+ENV_ACCESS_ID = "ODPS_ACCESS_ID"
+ENV_ACCESS_SECRET = "ODPS_ACCESS_SECRET"
+ENV_PROJECT = "ODPS_PROJECT"
+ENV_ENDPOINT = "ODPS_ENDPOINT"
+
+
+def require_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(f"Missing required environment variable: {name}")
+    return value
+
+
+class ODPSClient:
+    def __init__(self, project: str | None = None, endpoint: str | None = None):
+        access_id = require_env(ENV_ACCESS_ID)
+        access_secret = require_env(ENV_ACCESS_SECRET)
+        project_name = project or require_env(ENV_PROJECT)
+        endpoint_url = endpoint or require_env(ENV_ENDPOINT)
+
+        options.connect_timeout = 60
+        options.read_timeout = 1200
+        options.retry_times = 3
+
+        self.odps = ODPS(
+            access_id,
+            access_secret,
+            project=project_name,
+            endpoint=endpoint_url,
+        )
+
+    def execute_sql(self, sql: str):
+        if not sql.strip():
+            raise ValueError("SQL must not be empty")
+
+        instance = self.odps.run_sql(
+            sql,
+            hints={"odps.sql.submit.mode": "script"},
+        )
+        print(f"[ODPS] InstanceId: {instance.id}", flush=True)
+        instance.wait_for_success()
+        with instance.open_reader(tunnel=True) as reader:
+            return reader.to_pandas()
+
+    def execute_sql_result_save_file(self, sql: str, output_file: str | Path):
+        output_path = Path(output_file).expanduser()
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        data = self.execute_sql(sql)
+        data.to_csv(output_path, index=False, encoding="utf-8-sig")
+        return output_path

skills/odps-product-efficiency-report/scripts/run_sql.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Execute an explicit SQL file with the bundled environment-based ODPS client."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from odps_module import ODPSClient
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run an ODPS SQL file and save UTF-8 CSV")
+    parser.add_argument("sql_file", type=Path)
+    parser.add_argument("output_file", type=Path)
+    parser.add_argument("--project", help="Override ODPS_PROJECT for this query")
+    args = parser.parse_args()
+
+    sql = args.sql_file.expanduser().read_text(encoding="utf-8")
+    output_path = ODPSClient(project=args.project).execute_sql_result_save_file(
+        sql,
+        args.output_file,
+    )
+    print(f"[CSV] {output_path.resolve()}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

skills/query-user-behavior-path/SKILL.md

@@ -0,0 +1,55 @@
+---
+name: query-user-behavior-path
+description: 从 loghubods 视频、广告、播放、简单事件和用户活跃日志中查询、合并并解释单个用户的完整离线或明确指定的实时行为时间线。用于按 mid 或 machinecode 检查完整行为路径、生成按时间排序的 Excel、补充 businesstype 和 pagesource 中文定义，或诊断用户旅程。
+---
+
+# 查询用户完整行为路径
+
+将 `SKILL_DIR` 解析为当前 `SKILL.md` 所在的安装目录。所有命令都通过该目录调用随 Skill 分发的脚本；不得假定作者的工作区路径。
+
+## 运行环境
+
+依赖 `pyodps`、`pandas` 和 `openpyxl`。随 Skill 分发的 ODPS 客户端读取 `ODPS_ACCESS_ID`、`ODPS_ACCESS_SECRET`、`ODPS_PROJECT` 和 `ODPS_ENDPOINT`。不得把凭证复制到 Skill、查询输出或回复中。
+
+## 必要输入
+
+- 用户标识：视频/播放日志使用 `mid`，广告/简单事件/用户活跃日志使用 `machinecode`，值相同。
+- 日期：`yyyyMMdd`。
+- 可选 `apptype`，默认 `0`。
+- 输出目录，默认当前目录。
+
+## 离线流程
+
+1. 阅读 [logs-and-definitions.md](references/logs-and-definitions.md)。
+2. 运行：
+
+   `python3 "$SKILL_DIR/scripts/user_timeline.py" <mid_or_machinecode> <yyyyMMdd> [apptype] --output-dir <目录>`
+
+3. 报告输出路径、总行数、各来源行数以及行数为零的来源。
+4. 校验时间顺序、Excel 时间格式、技术事件排除和中文行为定义。
+
+## 实时流程
+
+只有用户明确要求实时数据时才运行：
+
+`python3 "$SKILL_DIR/scripts/user_timeline_realtime.py" <mid_or_machinecode> [yyyyMMdd] [apptype] --output-dir <目录>`
+
+必须说明结果只是当前可用实时数据。`simpleevent_log_flow` 保留周期短，不能代替完整离线日回灌。
+
+## 批量实时流程
+
+输入 CSV 必须包含 `用户标识`，可以包含 `命中策略`。运行：
+
+`python3 "$SKILL_DIR/scripts/user_timeline_realtime_batch.py" <用户清单.csv> [yyyyMMdd] [apptype] --output-dir <目录>`
+
+输出包含可筛选的 `实时行为路径` 工作表和 `用户摘要` 工作表；摘要必须保留当前事件数为零的输入用户。
+
+## 安全与校验
+
+- 除非用户明确要求实时数据，否则只使用离线表。
+- 每张表都必须按日期、`apptype`、用户标识和非空 `clienttimestamp` 过滤。
+- 各表事件使用 `UNION ALL` 保留，不得跨表去重。
+- 先按原始时间戳稳定排序，再转换为北京时间。
+- 不向终端打印完整用户事件表。
+- 只有参考资料或已确认源码支持时才补充中文行为定义。
+- 执行前运行 `python3 -m py_compile "$SKILL_DIR"/scripts/*.py`。

skills/query-user-behavior-path/agents/openai.yaml

@@ -0,0 +1,4 @@
+interface:
+  display_name: "查询用户完整行为路径"
+  short_description: "查询并解释跨视频广告播放与活跃日志的完整用户行为时间线"
+  default_prompt: "使用 $query-user-behavior-path 查询并解释指定用户的完整行为时间线。"

skills/query-user-behavior-path/references/logs-and-definitions.md

@@ -0,0 +1,71 @@
+# Logs and definitions
+
+## Offline logs
+
+| Source label | Table | Identifier | Time field | Role |
+|---|---|---|---|---|
+| `video` | `loghubods.video_action_log_applet` | `mid` | `clienttimestamp` | Video exposure and playback actions |
+| `ad` | `loghubods.ad_action_log_own` | `machinecode` | `clienttimestamp` | Ad lifecycle actions |
+| `play` | `loghubods.video_play_log` | `mid` | `clienttimestamp` | Playback logs |
+| `simpleevent` | `loghubods.simpleevent_log` | `machinecode` | `clienttimestamp` | Independent user and app events |
+| `useractive` | `loghubods.useractive_log` | `machinecode` | `clienttimestamp` | User-active path records |
+
+The same supplied identifier is used as both `mid` and `machinecode`.
+
+## Explicit real-time table map
+
+| Source label | Real-time table | Date filter | Notes |
+|---|---|---|---|
+| `video` | `loghubods.video_action_log_per5min` | `dt` from `YYYYMMDD000000` to `YYYYMMDD235959` | There is no `video_action_log_applet_per5min`. |
+| `ad` | `loghubods.ad_action_log_own_per5min` | Same 5-minute `dt` range | |
+| `play` | `loghubods.video_play_log_per5min` | Same 5-minute `dt` range | |
+| `simpleevent` | `loghubods.simpleevent_log_flow` | `year/month/day` (optionally hour) | No `_per5min` table; short real-time retention. |
+| `useractive` | `loghubods.useractive_log_per5min` | Same 5-minute `dt` range | |
+
+Use this map only when the user explicitly asks for real-time data. The result is the currently available real-time window; it is not a substitute for a complete offline daily path.
+
+For a multi-user real-time output, the input CSV must include `用户标识`; `命中策略` is optional. The batch script deduplicates users, merges their strategy labels, and writes both a filterable event sheet and a user-count summary sheet.
+
+## Output fields
+
+Offline output additionally includes `endRoutePath`, `isAdPlaying`, and `creativeCode` after `pagesource`. Real-time output keeps its existing fields unless the user explicitly requests the same extension.
+
+`pagesource` is the scene where the behavior occurred. Order timestamps using raw millisecond `clienttimestamp`; convert to China Standard Time only for output.
+
+For offline `simpleevent_log`, output `endRoutePath` and parse `isAdPlaying` and `creativeCode` from `extparams`; force these exact camel-case column names before writing Excel because ODPS aliases can be returned in lowercase.
+
+## Exclude
+
+Do not output rows where `businesstype` is one of:
+
+- `deviceId`
+- `openGIdSuccess`
+- `buttonView`
+
+For source `simpleevent` only, also exclude `openGIdError`.
+
+## Confirmed Chinese definitions
+
+| Source | businesstype | pagesource condition | 中文行为定义 | Evidence |
+|---|---|---|---|---|
+| `video` | `videoView` | ends with `user-videos-share` | 头部视频曝光 | Player marks this path as `headVideo` |
+| `video` | `videoPlay` | ends with `user-videos-share` | 头部视频播放 | Same scene rule |
+| `ad` | `adRequest` | ends with `user-videos-share` | 头部视频页广告请求 | Ad component reporting |
+| `ad` | `adLoaded` | ends with `user-videos-share` | 头部视频页广告加载 | Ad component reporting |
+| `ad` | `adView` | ends with `user-videos-share` | 头部视频页广告曝光 | Ad component reporting |
+| `ad` | `adPlay` | ends with `user-videos-share` | 头部视频页广告播放 | Ad component reporting |
+| `simpleevent` | `pageView` | ends with `category_55` | 首页分类页曝光（分类55） | `pages/category.js` |
+| `simpleevent` | `pageView` | ends with `user-videos-share` | 视频分享页曝光 | `PageSource.videoShare` |
+| `simpleevent` | `detailRequest` | ends with `user-videos-share` | 视频分享页详情接口请求成功 | Event ID `22022221` |
+| `simpleevent` | `userCaptureScreen` | ends with `user-videos-share` | 用户在视频分享页截图 | Event ID `550001` |
+| `simpleevent` | `userPause` | any | 视频暂停 | Player `reportPaused` |
+| `simpleevent` | `userActiveEnd` | any | 小程序进入后台 | App `onHide` |
+
+Leave any other definition blank unless the user confirms it or source code supplies evidence.
+
+## Mini-program evidence
+
+- `longVideoFactory/src/utils/Enum.js`: `PageSource.videoShare` is `vlog-pages/user-videos-share`.
+- `longVideoFactory/src/components/video/videoPlayerNew/pqVideoPlayer.js`: this path ending marks a head video.
+- `longVideoFactory/src/logcenter/logEnum.js`: page, detail, capture-screen event IDs.
+- `longVideoFactory/src/app.js`: `userActiveEnd` is reported in `onHide`.

skills/query-user-behavior-path/scripts/odps_module.py

@@ -0,0 +1,61 @@
+#!/usr/bin/env python3
+"""Minimal PyODPS client configured only through environment variables."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from odps import ODPS, options
+
+
+ENV_ACCESS_ID = "ODPS_ACCESS_ID"
+ENV_ACCESS_SECRET = "ODPS_ACCESS_SECRET"
+ENV_PROJECT = "ODPS_PROJECT"
+ENV_ENDPOINT = "ODPS_ENDPOINT"
+
+
+def require_env(name: str) -> str:
+    value = os.getenv(name)
+    if not value:
+        raise RuntimeError(f"Missing required environment variable: {name}")
+    return value
+
+
+class ODPSClient:
+    def __init__(self, project: str | None = None, endpoint: str | None = None):
+        access_id = require_env(ENV_ACCESS_ID)
+        access_secret = require_env(ENV_ACCESS_SECRET)
+        project_name = project or require_env(ENV_PROJECT)
+        endpoint_url = endpoint or require_env(ENV_ENDPOINT)
+
+        options.connect_timeout = 60
+        options.read_timeout = 1200
+        options.retry_times = 3
+
+        self.odps = ODPS(
+            access_id,
+            access_secret,
+            project=project_name,
+            endpoint=endpoint_url,
+        )
+
+    def execute_sql(self, sql: str):
+        if not sql.strip():
+            raise ValueError("SQL must not be empty")
+
+        instance = self.odps.run_sql(
+            sql,
+            hints={"odps.sql.submit.mode": "script"},
+        )
+        print(f"[ODPS] InstanceId: {instance.id}", flush=True)
+        instance.wait_for_success()
+        with instance.open_reader(tunnel=True) as reader:
+            return reader.to_pandas()
+
+    def execute_sql_result_save_file(self, sql: str, output_file: str | Path):
+        output_path = Path(output_file).expanduser()
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        data = self.execute_sql(sql)
+        data.to_csv(output_path, index=False, encoding="utf-8-sig")
+        return output_path

skills/query-user-behavior-path/scripts/run_sql.py

@@ -0,0 +1,28 @@
+#!/usr/bin/env python3
+"""Execute an explicit SQL file with the bundled environment-based ODPS client."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+from odps_module import ODPSClient
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Run an ODPS SQL file and save UTF-8 CSV")
+    parser.add_argument("sql_file", type=Path)
+    parser.add_argument("output_file", type=Path)
+    parser.add_argument("--project", help="Override ODPS_PROJECT for this query")
+    args = parser.parse_args()
+
+    sql = args.sql_file.expanduser().read_text(encoding="utf-8")
+    output_path = ODPSClient(project=args.project).execute_sql_result_save_file(
+        sql,
+        args.output_file,
+    )
+    print(f"[CSV] {output_path.resolve()}", flush=True)
+
+
+if __name__ == "__main__":
+    main()

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

@@ -0,0 +1,198 @@
+#!/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()

skills/query-user-behavior-path/scripts/user_timeline_realtime.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""单用户实时行为时间线：基于实时/5分钟日志输出 Excel。
+
+用法：python3 user_timeline_realtime.py <machinecode/mid值> [日期yyyyMMdd] [apptype] [--output-dir 目录]
+
+注意：simpleevent_log_flow 仅保留短实时窗口，因此结果代表当前可用实时数据，
+不等同于离线表的整日完整回灌数据。
+"""
+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, PAGESTATUS_MEAN, behavior_definition, safe_name, sql_text, text
+
+
+SQL_REALTIME = """
+WITH v AS (
+    SELECT CAST(clienttimestamp AS BIGINT) AS ts, 'video' AS 来源,
+           businesstype, pagesource, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' 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, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' 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, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' 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, 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_flow
+    WHERE year='{year}' AND month='{month}' AND day='{day_of_month}'
+      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 视频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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' 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.视频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 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="0")
+    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": sql_text(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)]
+    df = df.sort_values("ts", kind="stable").reset_index(drop=True)
+    df = df.rename(columns={"pagestatus": "pageStatus", "newpage": "newPage"})
+    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_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["实时行为路径"]["A"][1:]:
+            cell.number_format = "yyyy/mm/dd hh:mm:ss"
+
+    print("注意：simpleevent_log_flow 仅保留当前短实时窗口。")
+    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()

skills/query-user-behavior-path/scripts/user_timeline_realtime_batch.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python
+# coding=utf-8
+"""多个实时风险用户的行为时间线，输出可按用户筛选的单一 Excel。
+
+用法：python3 user_timeline_realtime_batch.py <用户清单.csv> [日期yyyyMMdd] [apptype] [--output-dir 目录]
+
+清单必须包含“用户标识”列，可选“命中策略”列。相同用户会只查询一次，
+并将多条策略标签合并到输出的“命中策略”列。
+"""
+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, PAGESTATUS_MEAN, behavior_definition, safe_name, sql_text, text
+
+
+SQL_REALTIME_BATCH = """
+WITH v AS (
+    SELECT CAST(clienttimestamp AS BIGINT) AS ts, mid AS 用户标识, 'video' AS 来源,
+           businesstype, pagesource, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' AND apptype='{apptype}' AND mid IN ({mc_list})
+      AND businesstype <> 'videoPreView'
+      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+),
+a AS (
+    SELECT CAST(clienttimestamp AS BIGINT) AS ts, machinecode AS 用户标识, 'ad' AS 来源,
+           businesstype, pagesource, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' AND apptype='{apptype}' AND machinecode IN ({mc_list})
+      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+),
+p AS (
+    SELECT CAST(clienttimestamp AS BIGINT) AS ts, mid AS 用户标识, 'play' AS 来源,
+           businesstype, pagesource, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' AND apptype='{apptype}' AND mid IN ({mc_list})
+      AND clienttimestamp IS NOT NULL AND clienttimestamp<>''
+),
+s AS (
+    SELECT CAST(clienttimestamp AS BIGINT) AS ts, machinecode AS 用户标识, 'simpleevent' AS 来源,
+           businesstype, pagesource, 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_flow
+    WHERE year='{year}' AND month='{month}' AND day='{day_of_month}'
+      AND apptype='{apptype}' 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) AS ts, machinecode AS 用户标识, 'useractive' AS 来源,
+           businesstype, pagesource, 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_per5min
+    WHERE dt >= '{dt_start}' AND dt <= '{dt_end}' AND apptype='{apptype}' 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
+)
+SELECT t.ts, t.用户标识, t.来源, t.businesstype, t.pagesource, 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 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 not user:
+            continue
+        strategies.setdefault(user, [])
+        strategy = row[strategy_column].strip() if strategy_column else ""
+        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="0")
+    parser.add_argument("--output-dir", type=Path, default=Path("."))
+    args = parser.parse_args()
+
+    user_strategies = load_user_strategies(args.users_file)
+    query_date = datetime.strptime(args.date, "%Y%m%d")
+    sql_args = {
+        "mc_list": sql_literals(user_strategies),
+        "apptype": sql_text(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_BATCH.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"})
+    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(2, "命中策略", df["用户标识"].map(user_strategies).fillna(""))
+    df.insert(df.columns.get_loc("businesstype") + 1, "中文行为定义", df.apply(behavior_definition, axis=1))
+    event_counts = df["用户标识"].value_counts()
+    summary = pd.DataFrame(
+        {
+            "用户标识": list(user_strategies),
+            "命中策略": list(user_strategies.values()),
+            "实时事件数": [event_counts.get(user, 0) for user in user_strategies],
+        }
+    )
+
+    output_dir = args.output_dir.expanduser()
+    output_dir.mkdir(parents=True, exist_ok=True)
+    cohort = safe_name(args.users_file.stem)
+    out = output_dir / f"timeline_realtime_{cohort}_{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["A"][1:]:
+            cell.number_format = "yyyy/mm/dd hh:mm:ss"
+        summary.to_excel(writer, sheet_name="用户摘要", index=False)
+        summary_sheet = writer.sheets["用户摘要"]
+        summary_sheet.freeze_panes = "A2"
+        summary_sheet.auto_filter.ref = summary_sheet.dimensions
+
+    print("注意：simpleevent_log_flow 仅保留当前短实时窗口。")
+    print(f"[XLSX] 用户数={len(user_strategies)} 日期={args.date} 事件数={len(df)} -> {out.resolve()}", flush=True)
+
+
+if __name__ == "__main__":
+    main()