Agent/examples/auto_put_ad_mini/roi_control/reporting.py

"""调控结果 Excel 报告生成。"""

from __future__ import annotations

from pathlib import Path
from typing import Dict, Iterable, 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, Font, PatternFill
from openpyxl.utils import get_column_letter
from .metrics import ENTITY_GZH, ENTITY_QIWEI, ENTITY_SELF


HEADER_FILL = PatternFill("solid", fgColor="1F4E78")
HEADER_FONT = Font(color="FFFFFF", bold=True)
STOP_FILL = PatternFill("solid", fgColor="F4CCCC")
UP_FILL = PatternFill("solid", fgColor="D9EAD3")
ADJUST_FILL = PatternFill("solid", fgColor="FFF2CC")


BASE_COLUMNS: Dict[str, Sequence[str]] = {
    "小程序投流": (
        "渠道",
        "代理名称",
        "账号id",
        "账号名称",
        "广告id",
        "广告名称",
        "包名",
        "创意id",
        "广告age",
        "日均首层UV",
        "三日首层UV",
        "三日T0裂变数",
        "T0裂变率",
        "效率收入",
        "多层裂变收入",
        "全链路效率收入",
        "成本",
        "三日最小单日成本",
        "三日聚合ROI",
        "关停线",
        "扩量线",
        "动作",
        "执行模式",
        "执行说明",
    ),
    "公众号即转": (
        "渠道",
        "合作方名",
        "公众号名",
        "日均首层UV",
        "三日首层UV",
        "三日T0裂变数",
        "T0裂变率",
        "效率收入",
        "多层裂变收入",
        "全链路效率收入",
        "成本",
        "三日最小单日成本",
        "三日聚合ROI",
        "关停线",
        "扩量线",
        "渠道内ROI排名百分位",
        "动作",
        "执行模式",
        "执行说明",
    ),
    "企微群合作": (
        "渠道",
        "合作方名",
        "公众号名",
        "日均首层UV",
        "三日首层UV",
        "三日T0裂变数",
        "T0裂变率",
        "效率收入",
        "多层裂变收入",
        "全链路效率收入",
        "成本",
        "三日最小单日成本",
        "三日聚合ROI",
        "关停线",
        "扩量线",
        "动作",
        "执行模式",
        "执行说明",
    ),
}

ENTITY_TO_SHEET = {
    ENTITY_SELF: "小程序投流",
    ENTITY_GZH: "公众号即转",
    ENTITY_QIWEI: "企微群合作",
}

FREEZE_PANES = {
    "小程序投流": "F2",
    "公众号即转": "D2",
    "企微群合作": "C2",
}


def _date_detail_columns(columns: Iterable[str]) -> list[str]:
    result = [
        column
        for column in columns
        if column.startswith(("首层UV_20", "成本_20", "ROI_20"))
    ]
    return sorted(result, key=lambda value: (value.rsplit("_", 1)[-1], value.split("_", 1)[0]))


def _sheet_frame(candidates: pd.DataFrame, sheet_name: str) -> pd.DataFrame:
    entity_type = next(
        key for key, value in ENTITY_TO_SHEET.items() if value == sheet_name
    )
    subset = candidates[candidates["entity_type"].eq(entity_type)].copy()
    subset = subset.rename(columns={"channel": "渠道"})
    subset["三日聚合ROI"] = subset["ROI"]
    subset["关停线"] = subset["t_stop"]
    subset["扩量线"] = subset["t_up"]
    date_columns = _date_detail_columns(subset.columns)
    ordered = list(BASE_COLUMNS[sheet_name]) + date_columns
    for column in ordered:
        if column not in subset:
            subset[column] = ""
    subset = subset[ordered]
    if not subset.empty:
        subset = subset.sort_values(["动作", "三日聚合ROI"], ascending=[True, False])
    return subset


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) and np.isnan(value) else value
                for value in values
            ]
        )


def _format_sheet(ws, sheet_name: str) -> None:
    max_column = max(ws.max_column, 1)
    max_row = max(ws.max_row, 1)
    ws.freeze_panes = FREEZE_PANES[sheet_name]
    ws.auto_filter.ref = f"A1:{get_column_letter(max_column)}{max_row}"
    ws.row_dimensions[1].height = 28

    headers = {}
    for cell in ws[1]:
        cell.fill = HEADER_FILL
        cell.font = HEADER_FONT
        cell.alignment = Alignment(horizontal="center", vertical="center")
        headers[cell.value] = cell.column

    for column_index in range(1, max_column + 1):
        header = ws.cell(1, column_index).value or ""
        values = [str(ws.cell(row, column_index).value or "") for row in range(1, max_row + 1)]
        width = min(max(max(map(len, values)), len(str(header))) + 2, 42)
        ws.column_dimensions[get_column_letter(column_index)].width = max(width, 11)

        if "ROI" in str(header) or "裂变率" in str(header) or "百分位" in str(header):
            for row in range(2, max_row + 1):
                ws.cell(row, column_index).number_format = "0.000"
        elif "成本" in str(header) or header in {
            "效率收入",
            "多层裂变收入",
            "全链路效率收入",
        }:
            for row in range(2, max_row + 1):
                ws.cell(row, column_index).number_format = "#,##0.00"
        elif "UV" in str(header) or "裂变数" in str(header):
            for row in range(2, max_row + 1):
                ws.cell(row, column_index).number_format = "#,##0.00"

        if "ROI" in str(header) and max_row >= 2:
            letter = get_column_letter(column_index)
            ws.conditional_formatting.add(
                f"{letter}2:{letter}{max_row}",
                ColorScaleRule(
                    start_type="min",
                    start_color="F8696B",
                    mid_type="percentile",
                    mid_value=50,
                    mid_color="FFEB84",
                    end_type="max",
                    end_color="63BE7B",
                ),
            )

    action_column = headers.get("动作")
    if action_column:
        for row in range(2, max_row + 1):
            action = ws.cell(row, action_column).value
            fill = {
                "关停": STOP_FILL,
                "扩量": UP_FILL,
                "调整封面&落地页视频": ADJUST_FILL,
            }.get(action)
            if fill:
                ws.cell(row, action_column).fill = fill

def _write_summary(
    workbook: Workbook,
    thresholds: pd.DataFrame,
    candidates: pd.DataFrame,
    expected_dates: Sequence[str],
    qiwei_fission_is_zero: bool,
    rule_config: Mapping[str, object] | None,
) -> None:
    ws = workbook.active
    ws.title = "运行摘要"
    rows = [
        ("统计窗口", f"{expected_dates[0]} 至 {expected_dates[-1]}"),
        ("首层口径", "usersharedepth='0'"),
        ("UV口径", "COUNT(DISTINCT mid)"),
        ("ROI口径", "三日聚合：(三日效率收入 + 三日多层t0裂变收入) / 三日成本"),
        ("阈值口径", "三类合格实体混入同一全局池：三日聚合ROI的P10=关停线、P80=扩量线"),
        (
            "合格池口径",
            "完整覆盖3天；UV、单日成本、广告age与分位门槛按本批次规则配置执行",
        ),
        ("动作优先级", "关停 > 扩量 > 调整封面&落地页视频"),
        ("审批方式", "飞书消息整批确认或拒绝；表格仅用于查看，不做逐行审批"),
    ]
    for label, value in rows:
        ws.append([label, value])

    if rule_config:
        ws.append([])
        ws.append(["规则配置", "值"])
        for key in (
            "self_stop_min_age",
            "self_up_min_age",
            "self_min_avg_uv",
            "partner_min_avg_uv",
            "min_daily_cost",
            "stop_quantile",
            "up_quantile",
            "scale_ratio",
            "scale_cooldown_days",
            "max_base_ratio",
        ):
            ws.append([key, rule_config.get(key)])

    ws.append([])
    display_thresholds = thresholds.rename(
        columns={"t_stop": "关停线", "t_up": "扩量线"}
    )
    threshold_columns = [
        "统计窗口",
        "关停线",
        "扩量线",
        "阈值样本数",
        "小程序样本数",
        "公众号样本数",
        "企微样本数",
    ]
    ws.append(threshold_columns)
    for values in display_thresholds[threshold_columns].itertuples(
        index=False,
        name=None,
    ):
        ws.append(list(values))

    ws.append([])
    ws.append(["渠道", "建议动作数"])
    for entity_type, sheet_name in ENTITY_TO_SHEET.items():
        count = int(candidates["entity_type"].eq(entity_type).sum())
        ws.append([sheet_name, count])

    if qiwei_fission_is_zero:
        ws.append([])
        ws.append(
            [
                "数据质量告警",
                "企微群合作在统计窗口内的多层t0裂变收入合计为0；本报告严格按附件公式计算，未使用ARPU估算。",
            ]
        )

    ws.column_dimensions["A"].width = 22
    ws.column_dimensions["B"].width = 92
    for row in ws.iter_rows():
        if row[0].value in {"统计窗口", "渠道"}:
            for cell in row:
                cell.font = Font(bold=True)
    ws.freeze_panes = "A2"


def write_workbook(
    candidates: pd.DataFrame,
    thresholds: pd.DataFrame,
    expected_dates: Sequence[str],
    output_path: Path,
    qiwei_fission_is_zero: bool,
    rule_config: Mapping[str, object] | None = None,
) -> Path:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    workbook = Workbook()
    _write_summary(
        workbook,
        thresholds,
        candidates,
        expected_dates,
        qiwei_fission_is_zero,
        rule_config,
    )

    for sheet_name in BASE_COLUMNS:
        ws = workbook.create_sheet(sheet_name)
        frame = _sheet_frame(candidates, sheet_name)
        _write_dataframe(ws, frame)
        _format_sheet(ws, sheet_name)

    workbook.save(output_path)
    return output_path