data_query_miniprogram/run_bucket_source_offline_5day_report.py

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

import pandas as pd

from odps_module import ODPSClient
from run_bucket_source_pv_uv_return_realtime import (
    FACT_COLUMNS,
    add_rates,
    add_relative_changes,
    column_mapping,
    format_fact_columns_as_integers,
    format_rate_columns,
    reorder_columns,
)


BASE = Path(__file__).resolve().parent
CONFIGS = {
    "offline": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260716_20260720_apptype4_all_versions.sql",
        "output": BASE / "bucket_source_full_report_offline_20260716_20260720_apptype4_all_versions.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260713_20260716": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260713_20260716_apptype4_all_versions.sql",
        "output": BASE / "bucket_source_full_report_offline_20260713_20260716_apptype4_all_versions.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260612_1508": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260612_apptype0_version1508.sql",
        "output": BASE / "bucket_source_full_report_offline_20260612_apptype0_version1508.csv",
        "experiment_buckets": set("0123456789ab"),
        "experiment_label": "实验组（0-b）",
        "control_label": "对照组（c-f）",
        "experiment_tail_label": "0-b",
        "control_tail_label": "c-f",
    },
    "offline_20260617_all": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260617_apptype0_all_versions.sql",
        "output": BASE / "bucket_source_full_report_offline_20260617_apptype0_all_versions.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260617_1512": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260617_apptype0_version1512.sql",
        "output": BASE / "bucket_source_full_report_offline_20260617_apptype0_version1512.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260721_1578": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260721_apptype4_version1578.sql",
        "output": BASE / "bucket_source_full_report_offline_20260721_apptype4_version1578.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260722_all": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260722_apptype4_all_versions.sql",
        "output": BASE / "bucket_source_full_report_offline_20260722_apptype4_all_versions.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260717_20260721_exclude_qywx": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260717_20260721_apptype4_all_versions_exclude_qywx.sql",
        "output": BASE / "bucket_source_full_report_offline_20260717_20260721_apptype4_all_versions_exclude_qywx.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
    "offline_20260721_1578_exclude_qywx": {
        "sql": BASE / "bucket_source_pv_uv_return_offline_20260721_apptype4_version1578_exclude_qywx.sql",
        "output": BASE / "bucket_source_full_report_offline_20260721_apptype4_version1578_exclude_qywx.csv",
        "experiment_buckets": set("01"),
        "experiment_label": "实验组（0、1）",
        "control_label": "对照组（其余14桶）",
        "experiment_tail_label": "0、1",
        "control_tail_label": "其余14桶",
    },
}
MODE = next((arg for arg in sys.argv[1:] if not arg.startswith("--")), "offline")
if MODE not in CONFIGS:
    raise SystemExit("usage: python3 run_bucket_source_offline_5day_report.py [offline|offline_20260713_20260716|offline_20260612_1508|offline_20260617_all|offline_20260617_1512|offline_20260721_1578|offline_20260722_all|offline_20260717_20260721_exclude_qywx|offline_20260721_1578_exclude_qywx] [--rebuild-with-dau]")
CONFIG = CONFIGS[MODE]
SQL_FILE, OUTPUT_FILE = CONFIG["sql"], CONFIG["output"]
DAU_SQL_FILE = BASE / "offline_dau_by_bucket_20260716_20260720_apptype4.sql"


def build_daily_report(data, stat_date):
    daily = data[data["日期"] == stat_date].copy()
    app_types = daily["产品类型"].astype(str).unique()
    versions = daily["版本号"].astype(str).unique()
    if len(app_types) != 1 or len(versions) != 1:
        raise ValueError(f"日期{stat_date}的产品或版本不唯一：apptype={app_types}, versions={versions}")
    app_type, version = app_types[0], versions[0]
    daily["尾号"] = daily["尾号"].astype(str)
    daily["行类型"] = "尾号明细"
    experiment_buckets = CONFIG["experiment_buckets"]
    control_buckets = set("0123456789abcdef") - experiment_buckets
    experiment_label = CONFIG["experiment_label"]
    control_label = CONFIG["control_label"]
    tail_labels = {
        experiment_label: CONFIG["experiment_tail_label"],
        control_label: CONFIG["control_tail_label"],
    }
    daily["分组"] = daily["尾号"].map(lambda bucket: experiment_label if bucket in experiment_buckets else control_label)
    detail = daily[["日期", "产品类型", "版本号", "行类型", "分组", "尾号", *FACT_COLUMNS]]

    aggregate_rows, mean_rows = [], []
    for group, buckets in [(experiment_label, experiment_buckets), (control_label, control_buckets)]:
        selected = detail[detail["尾号"].isin(buckets)]
        common = {"日期": stat_date, "产品类型": app_type, "版本号": version, "分组": group}
        aggregate_rows.append({
            **common,
            "行类型": "分组聚合",
            "尾号": tail_labels[group],
            **selected[FACT_COLUMNS].sum().to_dict(),
        })
        mean_rows.append({
            **common,
            "行类型": "每桶均值",
            "尾号": f"{len(buckets)}桶均值",
            **selected[FACT_COLUMNS].mean().to_dict(),
        })

    result = pd.concat([detail, pd.DataFrame(aggregate_rows), pd.DataFrame(mean_rows)], ignore_index=True)
    result = add_relative_changes(
        add_rates(result),
        experiment_label=experiment_label,
        control_label=control_label,
        experiment_bucket_count=len(experiment_buckets),
        control_bucket_count=len(control_buckets),
    )
    return reorder_columns(format_fact_columns_as_integers(format_rate_columns(result)))


def main():
    if "--rebuild-with-dau" in sys.argv:
        if MODE != "offline":
            raise ValueError("--rebuild-with-dau 仅适用于五天全版本离线报表")
        existing = pd.read_csv(OUTPUT_FILE, dtype={"日期": str, "尾号": str})
        detail = existing[existing["行类型"] == "尾号明细"][[
            "日期", "产品类型", "版本号", "尾号", *FACT_COLUMNS
        ]].copy()
        detail = detail.drop(columns=["DAU"])

        dau = ODPSClient().execute_sql(DAU_SQL_FILE.read_text(encoding="utf-8")).rename(
            columns={"stat_date": "日期", "bucket": "尾号", "dau": "DAU"}
        )
        dau["日期"] = dau["日期"].astype(str)
        dau["尾号"] = dau["尾号"].astype(str)
        facts = detail.merge(dau, on=["日期", "尾号"], how="left", validate="one_to_one")
        if len(facts) != 80 or facts["DAU"].isna().any():
            raise ValueError(f"DAU合并异常：rows={len(facts)}, missing_dau={facts['DAU'].isna().sum()}")

        reports = [build_daily_report(facts, stat_date) for stat_date in sorted(facts["日期"].unique(), reverse=True)]
        result = pd.concat(reports, ignore_index=True)
        result.to_csv(OUTPUT_FILE, index=False, encoding="utf-8-sig")
        print(f"[CSV REBUILT WITH OFFLINE DAU] {OUTPUT_FILE}", flush=True)
        print(f"[ROWS] {len(result)}", flush=True)
        print(result[result["行类型"] == "分组聚合"][["日期", "分组", "DAU", "DAU相对对照组变化率", "全部曝光PV/DAU", "全部分享PV/DAU", "全部回流UV/DAU", "全部ROV（回流UV/曝光PV）"]].to_string(index=False), flush=True)
        return

    facts = ODPSClient().execute_sql(SQL_FILE.read_text(encoding="utf-8")).rename(columns=column_mapping())
    facts["日期"] = facts["日期"].astype(str)
    reports = [build_daily_report(facts, stat_date) for stat_date in sorted(facts["日期"].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)
    print(f"[ROWS] {len(result)}", flush=True)
    print(result[result["行类型"] == "分组聚合"][["日期", "分组", "DAU", "DAU相对对照组变化率", "全部曝光PV/DAU", "全部分享PV/DAU", "全部ROV（回流UV/曝光PV）"]].to_string(index=False), flush=True)


if __name__ == "__main__":
    main()