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",
    ),
}
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():
    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 not in CONFIG:
        raise SystemExit("usage: python3 run_first_layer_source_metrics.py [realtime|offline]")
    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()