#!/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()