Agent/examples/auto_put_ad_mini/tools/im_approval_ad_creation.py

"""模块 A 广告新建审批 — 飞书 sheet 生成 + 发送 + 轮询(P1-G,2026-06-09)。

数据流(Phase 0 模块 A):
  ad_creation.enumerate_new_ad_candidates 准备 N 条 AdCandidate (无定投 + 有定投)
  → generate_ad_approval_xlsx          生成 9 列 xlsx(hyperlink/下拉/颜色/冻结)
  → send_ad_approval_to_feishu         上传转飞书 sheet + 发卡片到群
  → poll_ad_approval_actions           轮询读"决策"列 → 拿 approve/reject/hold
  → 调用方按 row_idx 匹配 candidate 写决策

跟创意审批 (im_approval_creation) 对偶,但维度是"广告"而非"创意":
  - 列展示广告级元数据(版位 / 定投场景 / 出价 / 年龄)
  - 中文场景名通过 scene_spec 缓存查询
"""

import asyncio
import logging
import time
from pathlib import Path

import httpx
from openpyxl import Workbook
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.worksheet.datavalidation import DataValidation

from config import (
    CREATION_APPROVAL_TIMEOUT_MINUTES,
    FEISHU_OPERATOR_CHAT_ID,
    now_in_timezone,
)
from tools.feishu_doc import (
    _auth_headers,
    _get_tenant_token,
    import_to_feishu,
)
from tools.scene_spec import describe_position_ids
from tools.delivery_config import BID_MODE_MAX_CONVERSION

logger = logging.getLogger(__name__)

FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"

# 11 列(中文)
HEADERS = [
    # A 浅灰 4 列
    "日期", "账户ID", "人群包", "广告名称",
    # B 浅紫 6 列(广告维度)
    "投放版位", "版位定投场景", "出价方式", "出价(元)", "智能版位", "年龄定向",
    # C 浅绿 1 列
    "决策",
]

GROUP_COLORS = [
    ((1, 4), "FFD9D9D9"),    # A 浅灰
    ((5, 10), "FFD9C8E8"),   # B 浅紫
    ((11, 11), "FFC6E0B4"),  # C 浅绿
]

COL_WIDTHS = {
    "A": 12, "B": 14, "C": 16, "D": 32,
    "E": 36, "F": 50, "G": 18, "H": 10,
    "I": 12, "J": 12, "K": 14,
}

DECISION_COL_LETTER = "K"  # 第 11 列
VALID_ACTIONS = ("approve", "reject", "hold")


def _color_for_col(col_idx: int) -> str:
    """按列号返回所属分组的 header 背景色（ARGB），未匹配时返回白色。"""
    for (lo, hi), color in GROUP_COLORS:
        if lo <= col_idx <= hi:
            return color
    return "FFFFFFFF"


def _format_site_set_cn(site_set: list) -> str:
    """把版位枚举列表转成中文名称，用顿号拼接（未知枚举原样保留）。"""
    cn_map = {
        "SITE_SET_WECHAT": "微信公众号",
        "SITE_SET_WECHAT_PLUGIN": "微信插件",
        "SITE_SET_SEARCH_SCENE": "搜索场景",
        "SITE_SET_MOMENTS": "朋友圈",
        "SITE_SET_MINI_GAME_WECHAT": "小游戏",
        "SITE_SET_MINI_PROGRAM_WECHAT": "小程序",
    }
    return "、".join(cn_map.get(s, s) for s in (site_set or []))


def _format_record_to_row(rec: dict) -> list:
    """rec 是 ad_creation.AdCandidate 的 dict 形式 + 加 approval_date / audience_tier / age_range 等"""
    automatic_site_enabled = bool(rec.get("automatic_site_enabled"))
    site_set_cn = "AIM+" if automatic_site_enabled else _format_site_set_cn(rec.get("site_set") or [])
    wp_ids = rec.get("wechat_position") or []
    wp_cn = describe_position_ids(int(rec["account_id"]), wp_ids)
    bid_yuan = f"{rec['bid_amount_fen'] / 100:.2f}" if rec.get("bid_amount_fen") else ""
    bid_scene_cn = (
        "最大转化量"
        if rec.get("bid_scene") == BID_MODE_MAX_CONVERSION
        else "平均成本"
    )

    return [
        rec["approval_date"],
        str(rec["account_id"]),
        rec.get("audience_tier_label", ""),
        rec.get("adgroup_name", ""),
        site_set_cn,
        wp_cn,
        bid_scene_cn,
        bid_yuan,
        "是" if automatic_site_enabled else "否",
        rec.get("age_range", ""),
        "",  # 决策列空,等运营填
    ]


def generate_ad_approval_xlsx(records: list[dict], output_path: Path) -> Path:
    """生成 9 列模块 A 广告新建待审批 xlsx。"""
    wb = Workbook()
    ws = wb.active
    ws.title = "广告新建审批"

    # header
    for col_idx, name in enumerate(HEADERS, start=1):
        cell = ws.cell(row=1, column=col_idx, value=name)
        cell.font = Font(bold=True, size=11)
        cell.alignment = Alignment(horizontal="center", vertical="center")
        cell.fill = PatternFill(
            start_color=_color_for_col(col_idx),
            end_color=_color_for_col(col_idx),
            fill_type="solid",
        )

    # 数据行
    for row_idx, rec in enumerate(records, start=2):
        row = _format_record_to_row(rec)
        for col_idx, val in enumerate(row, start=1):
            cell = ws.cell(row=row_idx, column=col_idx, value=val)
            cell.alignment = Alignment(
                horizontal="center", vertical="center", wrap_text=True,
            )

    # 决策列下拉
    dv = DataValidation(
        type="list",
        formula1='"approve,reject,hold"',
        allow_blank=True,
    )
    dv.error = "请选择 approve / reject / hold"
    ws.add_data_validation(dv)
    last_row = max(100, len(records) + 1)
    dv.add(f"{DECISION_COL_LETTER}2:{DECISION_COL_LETTER}{last_row}")

    # 冻结首行
    ws.freeze_panes = "A2"

    # 列宽 + 行高
    for col, w in COL_WIDTHS.items():
        ws.column_dimensions[col].width = w
    ws.row_dimensions[1].height = 28
    for r in range(2, 2 + len(records)):
        ws.row_dimensions[r].height = 80

    output_path.parent.mkdir(parents=True, exist_ok=True)
    wb.save(output_path)
    logger.info(
        "[im_approval_ad_creation] xlsx 已生成 records=%d path=%s",
        len(records), output_path,
    )
    return output_path


def _query_first_sheet_id(sheet_token: str) -> str:
    """查 sheet_id(对偶 im_approval_creation 同名函数)"""
    token = _get_tenant_token()
    url = f"{FEISHU_BASE_URL}/sheets/v3/spreadsheets/{sheet_token}/sheets/query"
    resp = httpx.get(url, headers=_auth_headers(token), timeout=30)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"查 sheet_id 失败: {data}")
    sheets = (data.get("data") or {}).get("sheets") or []
    if not sheets:
        raise RuntimeError(f"sheet_token={sheet_token} 无 sheet")
    return sheets[0]["sheet_id"]


async def send_ad_approval_to_feishu(
    xlsx_path: Path,
    chat_id: str = "",
    preamble: str = "",
) -> dict:
    """上传 xlsx → 转飞书 sheet → 发卡片到群。"""
    chat_id = chat_id or FEISHU_OPERATOR_CHAT_ID
    if not preamble:
        preamble = (
            "【广告新建审批】下方在线表格列出本次准备建的广告候选。\n"
            "请在最右侧【决策】列下拉框选择 approve / reject / hold。\n"
            "我们会在 2 小时内或所有行决策完成后,执行 approve 项,跳过 reject / hold。\n"
            "差异化机制:无定投 + 有定投(版位定投场景) → 腾讯唯一性绕过。"
        )

    result = await import_to_feishu(
        xlsx_path=str(xlsx_path),
        send_im=True,
        chat_id=chat_id,
        preamble=preamble,
    )
    meta = result.metadata or {}
    sheet_token = meta.get("sheet_token", "")
    if not sheet_token:
        raise RuntimeError(
            f"send_ad_approval_to_feishu: 未拿到 sheet_token result={result}"
        )

    sheet_id = _query_first_sheet_id(sheet_token)
    return {
        "url": meta.get("url", ""),
        "sheet_token": sheet_token,
        "sheet_id": sheet_id,
        "im_sent": meta.get("im_sent", False),
        "im_message_id": meta.get("im_message_id", ""),
    }


def read_actions_from_sheet(
    sheet_token: str, sheet_id: str, row_count: int,
) -> dict[int, str]:
    """读决策列,返回 {row_idx: action}。"""
    token = _get_tenant_token()
    last_row = row_count + 1
    range_str = f"{sheet_id}!{DECISION_COL_LETTER}2:{DECISION_COL_LETTER}{last_row}"
    url = (
        f"{FEISHU_BASE_URL}/sheets/v2/spreadsheets/{sheet_token}"
        f"/values/{range_str}?valueRenderOption=ToString"
    )
    resp = httpx.get(url, headers=_auth_headers(token), timeout=30)
    resp.raise_for_status()
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"读 sheet 失败 data={data}")

    values = (data.get("data") or {}).get("valueRange", {}).get("values", []) or []

    actions: dict[int, str] = {}
    for i, row in enumerate(values, start=1):
        if not row:
            continue
        cell_raw = row[0]
        if cell_raw is None:
            continue
        cell = str(cell_raw).strip().lower()
        if cell in VALID_ACTIONS:
            actions[i] = cell
    return actions


def poll_ad_approval_actions(
    sheet_token: str, sheet_id: str, expected_row_count: int,
    timeout_minutes: int = CREATION_APPROVAL_TIMEOUT_MINUTES,
    poll_interval_seconds: int = 60,
) -> dict[int, str]:
    """轮询读决策列,完成或超时退出。"""
    deadline = time.time() + timeout_minutes * 60
    last_count = -1
    actions: dict[int, str] = {}

    while time.time() < deadline:
        actions = read_actions_from_sheet(sheet_token, sheet_id, expected_row_count)
        if len(actions) != last_count:
            logger.info(
                "[poll_ad_approval] %d/%d 已决策",
                len(actions), expected_row_count,
            )
            last_count = len(actions)
        if len(actions) >= expected_row_count:
            logger.info(
                "[poll_ad_approval] 全部完成 %d/%d, 提前退出轮询",
                len(actions), expected_row_count,
            )
            return actions
        time.sleep(poll_interval_seconds)

    logger.warning(
        "[poll_ad_approval] 超时 %d 分钟, 仅 %d/%d 已决策",
        timeout_minutes, last_count, expected_row_count,
    )
    return actions


def run_ad_approval_workflow(
    records: list[dict],
    xlsx_output_dir: Path,
    timeout_minutes: int = CREATION_APPROVAL_TIMEOUT_MINUTES,
) -> tuple[dict, dict[int, str]]:
    """完整模块 A 审批工作流。"""
    now = now_in_timezone()
    date_str = now.strftime("%Y%m%d")
    ts = now.strftime("%H%M%S")
    xlsx_path = xlsx_output_dir / f"ad_creation_approval_{date_str}_{ts}.xlsx"

    generate_ad_approval_xlsx(records, xlsx_path)
    sheet_meta = asyncio.run(send_ad_approval_to_feishu(xlsx_path))
    actions = poll_ad_approval_actions(
        sheet_token=sheet_meta["sheet_token"],
        sheet_id=sheet_meta["sheet_id"],
        expected_row_count=len(records),
        timeout_minutes=timeout_minutes,
    )
    return sheet_meta, actions