data_query_miniprogram/run_first_layer_source_metrics.py

#!/usr/bin/env python3
from pathlib import Path
import sys

import pandas as pd

from odps_module import ODPSClient


BASE = Path(__file__).resolve().parent
CONFIG = {
    "realtime": (
        BASE / "first_layer_source_metrics_realtime_20260721_apptype4_version1578.sql",
        BASE / "first_layer_source_metrics_realtime_20260721_apptype4_version1578.csv",
    ),
    "offline": (
        BASE / "first_layer_source_metrics_offline_20260716_20260720_apptype4_all_versions.sql",
        BASE / "first_layer_source_metrics_offline_20260716_20260720_apptype4_all_versions.csv",
    ),
    "offline_20260611_1507": (
        BASE / "first_layer_source_metrics_offline_20260611_apptype0_version1507.sql",
        BASE / "first_layer_source_metrics_offline_20260611_apptype0_version1507.csv",
    ),
    "offline_20260717_20260720_special_layer": (
        BASE / "first_layer_source_metrics_offline_20260717_20260720_apptype4_all_versions_no_qywx_filter_special_layer.sql",
        BASE / "first_layer_source_metrics_offline_20260717_20260720_apptype4_all_versions_no_qywx_filter_special_layer.csv",
    ),
    "offline_20260721_1578_special_layer": (
        BASE / "first_layer_source_metrics_offline_20260721_apptype4_version1578_no_qywx_filter_special_layer.sql",
        BASE / "first_layer_source_metrics_offline_20260721_apptype4_version1578_no_qywx_filter_special_layer.csv",
    ),
    "realtime_20260722_special_layer": (
        BASE / "first_layer_source_metrics_realtime_20260722_apptype4_version1578_no_qywx_filter_special_layer.sql",
        BASE / "first_layer_source_metrics_realtime_20260722_apptype4_version1578_no_qywx_filter_special_layer.csv",
    ),
    "offline_20260716_20260721_20260722_all_special_layer": (
        BASE / "first_layer_source_metrics_offline_20260716_20260721_20260722_apptype4_all_versions_no_qywx_filter_special_layer.sql",
        BASE / "first_layer_source_metrics_offline_20260716_20260721_20260722_apptype4_all_versions_no_qywx_filter_special_layer.csv",
    ),
    "realtime_20260723_all_special_layer": (
        BASE / "first_layer_source_metrics_realtime_20260723_apptype4_all_versions_no_qywx_filter_special_layer.sql",
        BASE / "first_layer_source_metrics_realtime_20260723_apptype4_all_versions_no_qywx_filter_special_layer.csv",
    ),
}
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",
]
SPECIAL_LAYER_OUTPUTS = [
    CONFIG["realtime_20260722_special_layer"][1],
    CONFIG["offline_20260721_1578_special_layer"][1],
    CONFIG["offline_20260717_20260720_special_layer"][1],
]
SPECIAL_LAYER_COMBINED = BASE / "first_layer_source_metrics_20260717_20260720_offline_all_20260721_offline_1578_20260722_realtime_1578_no_qywx_filter_special_layer.csv"
SPECIAL_LAYER_DAILY = BASE / "first_layer_source_metrics_daily_compare_20260717_20260722_no_qywx_filter_special_layer.csv"
ALL_VERSION_SPECIAL_LAYER_OUTPUTS = [
    CONFIG["realtime_20260723_all_special_layer"][1],
    CONFIG["offline_20260716_20260721_20260722_all_special_layer"][1],
    CONFIG["offline_20260717_20260720_special_layer"][1],
]
ALL_VERSION_SPECIAL_LAYER_COMBINED = BASE / "first_layer_source_metrics_20260716_20260722_offline_20260723_realtime_apptype4_all_versions_no_qywx_filter_special_layer.csv"
ALL_VERSION_SPECIAL_LAYER_DAILY = BASE / "first_layer_source_metrics_daily_compare_20260716_20260723_apptype4_all_versions_no_qywx_filter_special_layer.csv"


def column_mapping():
    result = {
        "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():
            result[f"{key}_{field}"] = f"{source}{label}"
    return result


def group_name(bucket):
    return "实验组（0、1）" if bucket in {"0", "1"} else "对照组（其余14桶）"


def format_percent(value):
    return "" if pd.isna(value) else f"{value * 100:.4f}%"


def add_rates(data):
    result = data.copy()
    first_layer_uv = result["首层UV"].replace(0, pd.NA)
    for source in SOURCES:
        exposure_pv = 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_pv
        result[f"{source}T0裂变率（裂变层UV/首层UV）"] = result[f"{source}裂变层UV"] / first_layer_uv
        result[f"{source}裂变层UV/曝光PV"] = result[f"{source}裂变层UV"] / exposure_pv
    return result


def add_relative_changes(data):
    result = data.copy()
    result["首层UV相对对照组变化率"] = ""
    exp = result.index[(result["行类型"] == "分组聚合") & (result["分组"] == "实验组（0、1）")]
    ctrl = result.index[(result["行类型"] == "分组聚合") & (result["分组"] == "对照组（其余14桶）")]
    if len(exp) == 1 and len(ctrl) == 1:
        exp_mean = result.loc[exp[0], "首层UV"] / 2
        ctrl_mean = result.loc[ctrl[0], "首层UV"] / 14
        if ctrl_mean:
            change = format_percent(exp_mean / ctrl_mean - 1)
            mask = (result["分组"] == "实验组（0、1）") & result["行类型"].isin(["分组聚合", "每桶均值"])
            result.loc[mask, "首层UV相对对照组变化率"] = change

    for source in SOURCES:
        for suffix in RATE_SUFFIXES:
            metric = f"{source}{suffix}"
            change_column = f"{metric}相对对照组变化率"
            result[change_column] = ""
            for row_type in ["分组聚合", "每桶均值"]:
                exp_idx = result.index[(result["行类型"] == row_type) & (result["分组"] == "实验组（0、1）")]
                ctrl_idx = result.index[(result["行类型"] == row_type) & (result["分组"] == "对照组（其余14桶）")]
                if len(exp_idx) == 1 and len(ctrl_idx) == 1:
                    control = result.loc[ctrl_idx[0], metric]
                    if pd.notna(control) and control != 0:
                        result.loc[exp_idx[0], change_column] = format_percent(result.loc[exp_idx[0], metric] / control - 1)
    return result


def format_output(data):
    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 build_daily(data, stat_date):
    daily = data[data["日期"] == stat_date].copy()
    daily["尾号"] = daily["尾号"].astype(str)
    daily["行类型"] = "尾号明细"
    daily["分组"] = daily["尾号"].map(group_name)
    detail = daily[["日期", "产品类型", "版本号", "行类型", "分组", "尾号", *FACT_COLUMNS]]

    aggregates, means = [], []
    for group, buckets in [("实验组（0、1）", {"0", "1"}), ("对照组（其余14桶）", set("23456789abcdef"))]:
        selected = detail[detail["尾号"].isin(buckets)]
        common = {"日期": stat_date, "产品类型": detail["产品类型"].iloc[0], "版本号": detail["版本号"].iloc[0], "分组": group}
        aggregates.append({
            **common, "行类型": "分组聚合", "尾号": "0、1" if len(buckets) == 2 else "其余14桶",
            **selected[FACT_COLUMNS].sum().to_dict(),
        })
        means.append({
            **common, "行类型": "每桶均值", "尾号": "2桶均值" if len(buckets) == 2 else "14桶均值",
            **selected[FACT_COLUMNS].mean().to_dict(),
        })

    result = pd.concat([detail, pd.DataFrame(aggregates), pd.DataFrame(means)], ignore_index=True)
    return format_output(add_relative_changes(add_rates(result)))


def main():
    mode = sys.argv[1] if len(sys.argv) > 1 else "realtime"
    if mode == "combine_all_version_special_layer":
        result = pd.concat([pd.read_csv(path, dtype=str) for path in ALL_VERSION_SPECIAL_LAYER_OUTPUTS], ignore_index=True)
        result = result.sort_values("日期", ascending=False, kind="stable").reset_index(drop=True)
        if len(result) != 160 or result["日期"].nunique() != 8:
            raise ValueError(f"合并结果异常：rows={len(result)}, dates={result['日期'].nunique()}")
        result.to_csv(ALL_VERSION_SPECIAL_LAYER_COMBINED, index=False, encoding="utf-8-sig")
        daily = result[result["行类型"] == "分组聚合"].copy()
        daily.to_csv(ALL_VERSION_SPECIAL_LAYER_DAILY, index=False, encoding="utf-8-sig")
        print(f"[CSV] {ALL_VERSION_SPECIAL_LAYER_COMBINED} ({len(result)} rows)")
        print(f"[CSV] {ALL_VERSION_SPECIAL_LAYER_DAILY} ({len(daily)} rows)")
        return
    if mode == "combine_special_layer":
        result = pd.concat([pd.read_csv(path, dtype=str) for path in SPECIAL_LAYER_OUTPUTS], ignore_index=True)
        result.to_csv(SPECIAL_LAYER_COMBINED, index=False, encoding="utf-8-sig")
        daily = result[result["行类型"] == "分组聚合"].copy()
        daily.to_csv(SPECIAL_LAYER_DAILY, index=False, encoding="utf-8-sig")
        print(f"[CSV] {SPECIAL_LAYER_COMBINED} ({len(result)} rows)")
        print(f"[CSV] {SPECIAL_LAYER_DAILY} ({len(daily)} rows)")
        return
    if mode not in CONFIG:
        raise SystemExit(f"usage: python3 run_first_layer_source_metrics.py [{'|'.join(CONFIG)}|combine_special_layer|combine_all_version_special_layer]")
    sql_file, output_file = CONFIG[mode]
    data = ODPSClient().execute_sql(sql_file.read_text(encoding="utf-8")).rename(columns=column_mapping())
    data["日期"] = data["日期"].astype(str)
    reports = [build_daily(data, stat_date) for stat_date in sorted(data["日期"].unique(), reverse=True)]
    result = pd.concat(reports, ignore_index=True)
    result.to_csv(output_file, index=False, encoding="utf-8-sig")
    print(f"[CSV] {output_file}", flush=True)
    summary = result[result["行类型"] == "分组聚合"][[
        "日期", "版本号", "分组", "首层UV", "首层UV相对对照组变化率",
        "全部曝光PV/首层UV", "全部播放PV/首层UV", "全部分享PV/首层UV",
        "全部T0裂变率（裂变层UV/首层UV）", "全部裂变层UV/曝光PV",
    ]]
    print(summary.to_string(index=False), flush=True)


if __name__ == "__main__":
    main()