"""Three-day ROI Excel report generation with hidden audit columns.""" from __future__ import annotations import json import re from pathlib import Path from typing import Mapping, Sequence import numpy as np import pandas as pd from openpyxl import Workbook from openpyxl.formatting.rule import ColorScaleRule from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter from openpyxl.worksheet.datavalidation import DataValidation from .fission_multiplier import ( DISPLAY_MULTIPLIER_COLUMN, DISPLAY_TOTAL_TO_FIRST_COLUMN, ) from .metrics import ENTITY_GZH, ENTITY_SELF, ENTITY_SELF_AD HEADER_FILL = PatternFill("solid", fgColor="1F4E78") HEADER_FONT = Font(color="FFFFFF", bold=True) OBSERVE_FILL = PatternFill("solid", fgColor="FFF2CC") APPROVAL_FILL = PatternFill("solid", fgColor="FFD966") APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000") REPORT_VERSION = "roi_report_v41" REPORT_RUN_SUFFIX = "r41" AGENCY_REPORT_VERSION = "roi_agency_advice_v9" T0_FISSION_MULTIPLIER_COLUMN = "裂变系数-总裂变UV/T0裂变UV" TOTAL_FISSION_TO_FIRST_UV_COLUMN = DISPLAY_TOTAL_TO_FIRST_COLUMN FINAL_ROI_COLUMN = "三日加权平均效率ROI" SUMMARY_SHEETS = { ENTITY_SELF: "小程序创意级三日汇总", ENTITY_SELF_AD: "小程序广告级三日汇总", ENTITY_GZH: "公众号三日汇总", } AGENCY_SUMMARY_SHEETS = { ENTITY_SELF: "小程序创意调控建议", ENTITY_SELF_AD: SUMMARY_SHEETS[ENTITY_SELF_AD], } DAILY_SHEETS = { ENTITY_SELF: "小程序创意级每日明细", ENTITY_SELF_AD: "小程序广告级每日明细", ENTITY_GZH: "公众号每日明细", } SHEET_TO_ENTITY = { **{value: key for key, value in SUMMARY_SHEETS.items()}, **{value: key for key, value in DAILY_SHEETS.items()}, } ENTITY_DIMENSIONS = { ENTITY_SELF: ( "渠道", "代理名称", "账号id", "账号名称", "广告id", "广告名称", "包名", "广告优化目标", "创意id", "广告age", ), ENTITY_SELF_AD: ( "渠道", "代理名称", "账号id", "账号名称", "广告id", "广告名称", "包名", "广告优化目标", "广告age", ), ENTITY_GZH: ("渠道", "合作方名", "公众号名"), } AGENCY_ENTITY_DIMENSIONS = { ENTITY_SELF: ( "渠道", "代理名称", "账号id", "账号名称", "广告id", "广告名称", "广告优化目标", "创意id", ), ENTITY_SELF_AD: ( "渠道", "代理名称", "账号id", "账号名称", "广告id", "广告名称", "广告优化目标", ), } SUMMARY_METRICS = ( "日均首层UV", "日均T0裂变人数", "日均T0裂变率", "日均首层效率收入", "日均T0裂变效率收入", "日均总预估效率收入", "日均成本", T0_FISSION_MULTIPLIER_COLUMN, TOTAL_FISSION_TO_FIRST_UV_COLUMN, "三日均值ROI", "当日效率ROI", "预测总效率ROI", "关停线", "扩量线（P80）", "建议动作", "建议说明", ) DAILY_METRICS = ( "首层UV", "T0裂变人数", "T0裂变率", "首层效率收入", "T0裂变效率收入", "预测总效率收入", "成本", "当日效率ROI", "预测总效率ROI", T0_FISSION_MULTIPLIER_COLUMN, TOTAL_FISSION_TO_FIRST_UV_COLUMN, ) AGENCY_SUMMARY_METRICS = ( "日均成本", "评分", "建议动作", ) REPORT_EXCLUDED_COLUMNS = {"关停线分位点"} def _visible_columns( sheet_name: str, expected_dates: Sequence[str] | None = None, stop_quantile: float = 0.20, ) -> list[str]: entity_type = SHEET_TO_ENTITY[sheet_name] if sheet_name in SUMMARY_SHEETS.values(): columns = list(ENTITY_DIMENSIONS[entity_type]) + list(SUMMARY_METRICS) if entity_type == ENTITY_SELF: columns += ["当前创意状态"] return columns return [ "dt", "渠道", *[c for c in ENTITY_DIMENSIONS[entity_type] if c != "渠道"], *DAILY_METRICS, ] def _ensure_columns(frame: pd.DataFrame, columns: Sequence[str]) -> pd.DataFrame: result = frame.copy() for column in columns: if column not in result: result[column] = "" return result def _canonical_agency_name(value: object) -> str: if pd.isna(value): return "" return re.sub(r"\s*-\s*", "-", str(value).strip()) def _safe_filename_component(value: str) -> str: return re.sub(r'[\\/:*?"<>|]', "_", value).strip(" .") or "未命名代理" def _agency_summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame: frame = _summary_frame(rows, entity_type) frame["评分"] = pd.to_numeric(frame["当日效率ROI"], errors="coerce") columns = list(AGENCY_ENTITY_DIMENSIONS[entity_type]) + list( AGENCY_SUMMARY_METRICS ) if entity_type == ENTITY_SELF: columns.append("当前创意状态") return _ensure_columns(frame, columns)[columns].copy() def _summary_frame(rows: pd.DataFrame, entity_type: str) -> pd.DataFrame: subset = rows[rows["entity_type"].eq(entity_type)].copy() subset = subset.rename(columns={"channel": "渠道"}) subset["日均首层UV"] = pd.to_numeric( subset.get("日均首层UV"), errors="coerce" ).round().astype("Int64") subset["日均T0裂变人数"] = pd.to_numeric( subset.get("日均T0裂变数"), errors="coerce" ) subset["日均首层效率收入"] = pd.to_numeric( subset.get("日均效率收入"), errors="coerce" ) subset["日均T0裂变效率收入"] = pd.to_numeric( subset.get("日均T0裂变效率收入"), errors="coerce" ) subset["日均总预估效率收入"] = pd.to_numeric( subset.get("日均总预估效率收入"), errors="coerce" ) subset["日均T0裂变率"] = pd.to_numeric( subset.get("三日加权平均T0裂变率"), errors="coerce" ) subset["三日均值ROI"] = pd.to_numeric( subset.get("三日加权平均实际ROI"), errors="coerce" ) subset["当日效率ROI"] = pd.to_numeric( subset.get("最新日实际ROI"), errors="coerce" ) subset["预测总效率ROI"] = pd.to_numeric( subset.get("三日加权平均效率ROI"), errors="coerce" ) subset[T0_FISSION_MULTIPLIER_COLUMN] = pd.to_numeric( subset.get(DISPLAY_MULTIPLIER_COLUMN), errors="coerce" ) subset[TOTAL_FISSION_TO_FIRST_UV_COLUMN] = pd.to_numeric( subset.get(DISPLAY_TOTAL_TO_FIRST_COLUMN), errors="coerce" ) subset["关停线"] = pd.to_numeric( subset.get("t_stop"), errors="coerce" ) subset["关停线分位点"] = pd.to_numeric( subset.get("关停线分位点"), errors="coerce" ) subset["扩量线（P80）"] = ( pd.to_numeric(subset.get("t_up"), errors="coerce") if entity_type == ENTITY_SELF else np.nan ) subset["建议动作"] = subset.get("动作", "") if entity_type == ENTITY_SELF: subset["建议动作"] = subset["建议动作"].replace( {"关停": "关停创意"} ) elif entity_type == ENTITY_SELF_AD: subset["建议动作"] = subset["建议动作"].replace( {"关停": "关停广告"} ) subset["建议说明"] = subset.get("动作原因", "") visible = _visible_columns(SUMMARY_SHEETS[entity_type]) required = list(visible) if entity_type in (ENTITY_SELF, ENTITY_SELF_AD): required.append("审批选择") subset = _ensure_columns(subset, required) if not subset.empty: subset["_观察排序"] = subset["阈值样本状态"].eq( "补充观察_昨日UV>200" ).astype(int) subset["_动作排序"] = ( subset["建议动作"] .fillna("") .map( { "关停": 0, "关停创意": 0, "关停广告": 0, "扩量": 1, "": 2, "观察": 3, } ) .fillna(4) .astype(int) ) reason = subset["建议说明"].fillna("").astype(str) subset["_说明排序"] = np.select( [ reason.str.contains("三日加权平均预测总效率ROI", regex=False), reason.str.contains("同时满足ROI≤", regex=False), ], [0, 1], default=2, ) roi_sort = pd.to_numeric( subset["三日加权平均效率ROI"], errors="coerce" ).fillna(np.inf) subset["_ROI排序"] = np.where( subset["建议动作"].eq("扩量"), -roi_sort, roi_sort ) subset["_成本排序"] = pd.to_numeric( subset["日均成本"], errors="coerce" ).fillna(0) subset["_UV排序"] = pd.to_numeric( subset["最新日首层UV"], errors="coerce" ).fillna(0) observation = subset["_观察排序"].eq(1) subset.loc[observation, "_ROI排序"] = np.inf subset.loc[observation, "_成本排序"] = 0 subset = subset.sort_values( [ "_观察排序", "_动作排序", "_说明排序", "_ROI排序", "_成本排序", "_UV排序", ], ascending=[True, True, True, True, False, False], kind="stable", ).drop( columns=[ "_观察排序", "_动作排序", "_说明排序", "_ROI排序", "_成本排序", "_UV排序", ] ) neutral = subset["建议动作"].fillna("").eq("") subset.loc[neutral, "建议动作"] = "观察" if entity_type == ENTITY_SELF: neutral_reason = ( "预测总效率ROI位于动态关停线与扩量线（P80）之间，" "当前无需关停或扩量" ) elif entity_type == ENTITY_SELF_AD: neutral_reason = "预测总效率ROI高于小程序动态关停线，当前无需关停整个广告" else: neutral_reason = "预测总效率ROI高于公众号独立关停线，当前仅观察" subset.loc[neutral, "建议说明"] = neutral_reason hidden = [ column for column in subset.columns if column not in visible and column not in REPORT_EXCLUDED_COLUMNS ] return subset[visible + hidden] def _daily_frame( rows: pd.DataFrame, entity_type: str, expected_dates: Sequence[str], ) -> pd.DataFrame: summary = _summary_frame(rows, entity_type) daily_frames: list[pd.DataFrame] = [] for dt in expected_dates: daily = summary.copy() daily["dt"] = dt daily["首层UV"] = daily.get(f"首层UV_{dt}") daily["T0裂变人数"] = daily.get(f"T0裂变数_{dt}") uv = pd.to_numeric(daily["首层UV"], errors="coerce") fission = pd.to_numeric(daily["T0裂变人数"], errors="coerce") daily["T0裂变率"] = np.where(uv.gt(0), fission / uv, np.nan) daily["首层效率收入"] = daily.get(f"效率收入_{dt}") daily["T0裂变效率收入"] = daily.get(f"裂变效率收入_{dt}") daily["预测总效率收入"] = daily.get(f"预测全链路效率收入_{dt}") daily["成本"] = daily.get(f"成本_{dt}") daily["当日效率ROI"] = daily.get(f"实际ROI_{dt}") daily["预测总效率ROI"] = daily.get(f"ROI_{dt}") daily_frames.append(daily) result = pd.concat(daily_frames, ignore_index=True) if daily_frames else summary visible = _visible_columns(DAILY_SHEETS[entity_type]) result = _ensure_columns(result, visible) if not result.empty: entity_dimensions = [c for c in ENTITY_DIMENSIONS[entity_type] if c in result] result["_实体序"] = result.groupby(entity_dimensions, dropna=False, sort=False).ngroup() result["_日期序"] = result["dt"].map( {dt: i for i, dt in enumerate(reversed(expected_dates))} ) result = result.sort_values(["_日期序", "_实体序"], kind="stable").drop( columns=["_实体序", "_日期序"] ) hidden = [column for column in result.columns if column not in visible] return result[visible + hidden] def _sheet_frame( rows: pd.DataFrame, sheet_name: str, expected_dates: Sequence[str] | None = None, stop_quantile: float = 0.20, ) -> pd.DataFrame: entity_type = SHEET_TO_ENTITY[sheet_name] if sheet_name in SUMMARY_SHEETS.values(): return _summary_frame(rows, entity_type) if expected_dates is None: raise ValueError("每日明细必须提供 expected_dates") return _daily_frame(rows, entity_type, expected_dates) def _write_dataframe(ws, frame: pd.DataFrame) -> None: ws.append(list(frame.columns)) for values in frame.itertuples(index=False, name=None): ws.append( [ None if isinstance(value, (float, np.floating)) and np.isnan(value) else value for value in values ] ) def _number_format_for_header(header: object) -> str: name = re.sub(r"_\d{8}$", "", str(header or "")) if "占比" in name or "分位点" in name: return "0.00%" if name in { T0_FISSION_MULTIPLIER_COLUMN, TOTAL_FISSION_TO_FIRST_UV_COLUMN, }: return "0.00" lower = name.lower() is_count = ( name == "dt" or lower.endswith("id") or "uv" in lower or "人数" in name or "数量" in name or name.endswith("age") or name.endswith("天数") or (name.endswith("数") and not name.endswith("系数")) ) return "0" if is_count else "0.00" def _apply_number_formats(ws, headers: Mapping[object, int]) -> None: for header, column_index in headers.items(): number_format = _number_format_for_header(header) for row_number in range(2, ws.max_row + 1): cell = ws.cell(row_number, column_index) if isinstance( cell.value, (int, float, np.integer, np.floating) ) and not isinstance(cell.value, bool): cell.number_format = number_format def _format_sheet( ws, visible_columns: Sequence[str], *, approval: bool = False, freeze_panes: str = "A2", ) -> None: max_column = max(ws.max_column, 1) max_row = max(ws.max_row, 1) ws.freeze_panes = freeze_panes ws.auto_filter.ref = f"A1:{get_column_letter(max_column)}{max_row}" ws.row_dimensions[1].height = 28 headers = {cell.value: cell.column for cell in ws[1]} for cell in ws[1]: cell.fill = HEADER_FILL cell.font = HEADER_FONT cell.alignment = Alignment(horizontal="center", vertical="center") for column_index in range(1, max_column + 1): header = ws.cell(1, column_index).value ws.column_dimensions[get_column_letter(column_index)].hidden = ( header not in visible_columns ) if header in visible_columns: ws.column_dimensions[get_column_letter(column_index)].width = min( max(12, len(str(header)) * 2 + 2), 34 ) _apply_number_formats(ws, headers) action_column = headers.get("建议动作") or headers.get("动作") status_column = headers.get("阈值样本状态") for row_number in range(2, max_row + 1): action = ws.cell(row_number, action_column).value if action_column else "" status = ws.cell(row_number, status_column).value if status_column else "" fill = ( OBSERVE_FILL if status == "补充观察_昨日UV>200" or action == "观察" else None ) if fill: for column_index in range(1, len(visible_columns) + 1): ws.cell(row_number, column_index).fill = fill if action_column: def add_top_separator(row_number: int) -> None: separator = Side(style="medium", color="1F1F1F") for column_index in range(1, max_column + 1): if ws.cell(1, column_index).value in visible_columns: ws.cell(row_number, column_index).border = Border( top=separator ) first_scale_row = next( ( row_number for row_number in range(2, max_row + 1) if ws.cell(row_number, action_column).value == "扩量" ), None, ) if first_scale_row: add_top_separator(first_scale_row) first_observe_row = next( ( row_number for row_number in range(first_scale_row + 1, max_row + 1) if ws.cell(row_number, action_column).value == "观察" ), None, ) if first_observe_row: add_top_separator(first_observe_row) for roi_header in ("三日均值ROI", "当日效率ROI", "预测总效率ROI"): roi_column = headers.get(roi_header) if roi_header not in visible_columns or not roi_column or max_row < 2: continue roi_letter = get_column_letter(roi_column) ws.conditional_formatting.add( f"{roi_letter}2:{roi_letter}{max_row}", ColorScaleRule( start_type="min", start_color="C00000", mid_type="percentile", mid_value=50, mid_color="FFEB84", end_type="max", end_color="00B050", ), ) if approval and "审批选择" in headers and max_row >= 2: column_index = headers["审批选择"] letter = get_column_letter(column_index) ws.cell(1, column_index).fill = APPROVAL_HEADER_FILL validation = DataValidation( type="list", formula1='"批准,拒绝"', allow_blank=True ) ws.add_data_validation(validation) for row_number in range(2, max_row + 1): cell = ws.cell(row_number, column_index) if cell.value in (None, ""): validation.add(cell) cell.fill = APPROVAL_FILL def _write_summary_sheet( workbook: Workbook, thresholds: pd.DataFrame, expected_dates: Sequence[str], config: Mapping[str, object], ) -> None: ws = workbook.create_sheet("运行摘要") by_type = ( thresholds.set_index("entity_type").to_dict("index") if not thresholds.empty else {} ) self_threshold = by_type.get(ENTITY_SELF, {}) gzh_threshold = by_type.get(ENTITY_GZH, {}) rows = [ ("报表版本", REPORT_VERSION), ("统计窗口", f"{expected_dates[0]} 至 {expected_dates[-1]}"), ("数据深度口径", "usersharedepth<=1，与最新业务SQL一致"), ("小程序昨日总成本", self_threshold.get("小程序昨日总成本")), ("小程序目标关停成本", self_threshold.get("目标关停成本")), ("小程序实际关停成本", self_threshold.get("实际关停成本")), ("小程序实际关停成本占比", self_threshold.get("实际关停成本占比")), ("小程序关停成本预算状态", self_threshold.get("关停成本预算状态")), ("小程序三日关停线", self_threshold.get("三日关停线")), ("小程序三日关停线分位点", self_threshold.get("三日关停线分位点")), ("小程序单日P10线", self_threshold.get("单日P10线")), ("小程序单日合并资格线", self_threshold.get("单日合并资格线")), ("小程序单日实际关停线", self_threshold.get("单日实际关停线")), ("小程序单日实际关停线分位点", self_threshold.get("单日实际关停线分位点")), ("小程序三日实际关停成本", self_threshold.get("三日实际关停成本")), ("小程序单日实际关停成本", self_threshold.get("单日实际关停成本")), ("小程序单日候选池样本数", self_threshold.get("单日候选池样本数")), ("小程序单日合并候选数", self_threshold.get("单日合并候选数")), ("小程序阈值样本数", self_threshold.get("阈值样本数")), ("创意扩量线（P80）", self_threshold.get("t_up")), ("扩量样本数", self_threshold.get("扩量样本数")), ("公众号独立关停线", gzh_threshold.get("t_stop")), ("公众号关停线分位点", gzh_threshold.get("关停线分位点")), ("公众号阈值样本数", gzh_threshold.get("阈值样本数")), ( "阈值样本", "小程序与公众号按渠道独立；连续三天每天首层UV>200、成本>0且ROI有效", ), ("关停年龄门槛", "所有小程序创意级和广告级关停均要求广告age>3天"), ("广告级", "直接按广告去重计算，复用小程序三日动态关停线但不进入成本预算样本池；低于关停线且广告age>3天时，审批后暂停整个广告"), ("日均字段", "三日总量/3，缺失日按0"), ("ROI与裂变率", "三日汇总分子/三日汇总分母的加权口径"), ("小程序成本软预算", "使用T-1实际成本，目标占小程序创意级昨日总成本5%；三日持续低ROI与单日补充规则按6:4基础额度分配，未用额度只在低质候选间流转"), ("单日补充规则", "严格排除已进入三日正式判断的小程序创意；剩余创意最新日UV>200、成本>0、ROI有效且广告age>3天时进入单日池，同时满足预测总效率ROI≤0.20和实体等权P10才成为关停候选"), ("补充观察", "其余非正式样本中最新日首层UV>200，置于汇总表末尾且不执行"), ("配置快照", json.dumps(dict(config), ensure_ascii=False, default=str)), ] for row in rows: ws.append(row) ws.column_dimensions["A"].width = 28 ws.column_dimensions["B"].width = 110 for cell in ws[1]: cell.font = Font(bold=True) for row_number in range(1, ws.max_row + 1): label = str(ws.cell(row_number, 1).value or "") value_cell = ws.cell(row_number, 2) if isinstance(value_cell.value, (int, float)) and not isinstance( value_cell.value, bool ): if "占比" in label or "分位点" in label: value_cell.number_format = "0.00%" else: value_cell.number_format = "0" if label.endswith("数") else "0.00" def write_workbook( rows: pd.DataFrame, thresholds: pd.DataFrame, expected_dates: Sequence[str], output_path: Path, config: Mapping[str, object], fission_match_summary: pd.DataFrame | None = None, ) -> None: workbook = Workbook() workbook.remove(workbook.active) for entity_type in (ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH): sheet_name = SUMMARY_SHEETS[entity_type] frame = _summary_frame(rows, entity_type) ws = workbook.create_sheet(sheet_name) _write_dataframe(ws, frame) visible = _visible_columns(sheet_name) _format_sheet( ws, visible, approval=entity_type in (ENTITY_SELF, ENTITY_SELF_AD), freeze_panes="H2", ) if entity_type == ENTITY_SELF_AD: ws.sheet_state = "hidden" for entity_type in (ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH): sheet_name = DAILY_SHEETS[entity_type] frame = _daily_frame(rows, entity_type, expected_dates) ws = workbook.create_sheet(sheet_name) _write_dataframe(ws, frame) _format_sheet( ws, _visible_columns(sheet_name), freeze_panes="H2", ) if entity_type == ENTITY_SELF_AD: ws.sheet_state = "hidden" if fission_match_summary is not None: ws = workbook.create_sheet("传播裂变系数匹配") _write_dataframe(ws, fission_match_summary) _format_sheet(ws, list(fission_match_summary.columns)) ws.sheet_state = "hidden" _write_summary_sheet(workbook, thresholds, expected_dates, config) output_path.parent.mkdir(parents=True, exist_ok=True) workbook.save(output_path) def write_agency_workbooks( rows: pd.DataFrame, output_dir: Path, report_date: str, agency_names: set[str] | None = None, ) -> list[dict[str, object]]: """Create one miniapp control-advice workbook per agency.""" miniapp = rows[ rows["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD]) & rows["channel"].astype(str).str.startswith("小程序投流") ].copy() miniapp["_代理规范名"] = miniapp["代理名称"].map(_canonical_agency_name) agencies = sorted(name for name in miniapp["_代理规范名"].unique() if name) if agency_names is not None: canonical_names = {_canonical_agency_name(name) for name in agency_names} agencies = [name for name in agencies if name in canonical_names] output_dir.mkdir(parents=True, exist_ok=True) outputs: list[dict[str, object]] = [] for agency_name in agencies: agency_rows = miniapp[miniapp["_代理规范名"].eq(agency_name)].drop( columns=["_代理规范名"] ) workbook = Workbook() workbook.remove(workbook.active) for entity_type in (ENTITY_SELF, ENTITY_SELF_AD): summary_name = AGENCY_SUMMARY_SHEETS[entity_type] summary = _agency_summary_frame(agency_rows, entity_type) summary_ws = workbook.create_sheet(summary_name) _write_dataframe(summary_ws, summary) _format_sheet(summary_ws, list(summary.columns), freeze_panes="H2") if entity_type == ENTITY_SELF_AD: summary_ws.sheet_state = "hidden" filename = f"{report_date}_{_safe_filename_component(agency_name)}_调控建议.xlsx" output_path = output_dir / filename workbook.save(output_path) outputs.append( { "agency_name": agency_name, "report_version": AGENCY_REPORT_VERSION, "report": str(output_path), "creative_rows": int( agency_rows["entity_type"].eq(ENTITY_SELF).sum() ), "ad_rows": int( agency_rows["entity_type"].eq(ENTITY_SELF_AD).sum() ), } ) return outputs