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