Agent/examples/auto_put_ad_mini/tools/im_approval_creation.py

"""模块 B 创意搭建审批 — 飞书 sheet 生成 + 发送 + 轮询(P0-C,2026-06-09)。

数据流:
  Phase 1 准备 pending records (list[dict])
    → generate_approval_xlsx          生成 30 列 xlsx(hyperlink/下拉/颜色/冻结)
    → send_approval_to_feishu         上传转飞书 sheet + 发链接到群
    → poll_approval_actions           轮询读 sheet "决策"列,拿 approve/reject/hold
    → 调用方按 row_idx 匹配回 records 写决策

设计决策(2026-06-09 用户确认):
  - 30 列:A(灰)5 + B(紫)3 + C(橙)21 + D(绿)1,header 颜色分组
  - 素材预览用 HYPERLINK 公式(=IMAGE() 不被飞书支持,write-images API 留作 P3-5)
  - 落地视频列也用 HYPERLINK(点 video_id 跳转 video_url)
  - 决策列下拉:approve / reject / hold
  - 冻结首行(A2 起冻结上方)
  - 飞书 sheet 不支持 =IMAGE,实测 #VALUE!,所以 hyperlink 兜底
"""

import asyncio
import json
import logging
import os
import tempfile
import time
from pathlib import Path
from typing import Optional

import httpx
from openpyxl import Workbook
from openpyxl.drawing.image import Image as OpenpyxlImage
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.worksheet.datavalidation import DataValidation
from PIL import Image as PilImage

from config import (
    CREATION_APPROVAL_TIMEOUT_MINUTES,
    FEISHU_OPERATOR_CHAT_ID,
    now_in_timezone,
)

logger = logging.getLogger(__name__)

FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
EXTERNAL_RECALL_UV_WINDOW_DAYS = int(
    os.getenv("EXTERNAL_RECALL_UV_WINDOW_DAYS", "90")
)

# 30 列(中文)— 外部素材额外展示原图与配置窗口访问UV。
HEADERS = [
    # A 浅灰 5 列
    "日期", "账户ID", "人群包", "广告ID", "广告名称",
    # B 浅紫 3 列
    "出价(元)", "投放版位", "年龄定向",
    # C 浅橙 21 列(落地视频 + 风险审核 + 素材来源 + 素材质量)
    "落地视频", "落地视频标题", "风险等级", "风险标签", "风险原因",
    "素材来源", "素材预览", "素材链接", "原始素材",
    f"{EXTERNAL_RECALL_UV_WINDOW_DAYS}天访问UV", "创意文案",
    "成本(元)", "ROI", "CTR", "曝光数", "相似度",
    "召回维度", "召回点类型", "召回元素", "命中维度明细",
    "创意名(归因)",
    # D 浅绿 1 列
    "决策",
]

GROUP_COLORS = [
    ((1, 5), "FFD9D9D9"),    # A 浅灰
    ((6, 8), "FFD9C8E8"),    # B 浅紫
    ((9, 29), "FFFCD8B4"),   # C 浅橙
    ((30, 30), "FFC6E0B4"),  # D 浅绿
]

COL_WIDTHS = {
    "A": 12, "B": 14, "C": 16, "D": 16, "E": 28,
    "F": 10, "G": 28, "H": 12,
    "I": 14, "J": 26, "K": 10, "L": 24, "M": 32,
    "N": 14, "O": 18, "P": 24, "Q": 18, "R": 14, "S": 30,
    "T": 12, "U": 10, "V": 10, "W": 10, "X": 10,
    "Y": 14, "Z": 14, "AA": 18, "AB": 42,
    "AC": 38,                     # 创意名(归因)
    "AD": 14,                     # 决策
}

DECISION_COL_LETTER = "AD"
VALID_ACTIONS = ("approve", "reject", "hold")
EMBED_MATERIAL_PREVIEW_IMAGES = os.getenv(
    "CREATION_APPROVAL_EMBED_IMAGES", "1"
).strip().lower() in {"1", "true", "yes", "y", "on"}
MATERIAL_PREVIEW_COL_IDX = 15
MATERIAL_PREVIEW_COL_LETTER = "O"


def _color_for_col(col_idx: int) -> str:
    for (lo, hi), color in GROUP_COLORS:
        if lo <= col_idx <= hi:
            return color
    return "FFFFFFFF"


def _format_record_to_row(rec: dict) -> list:
    """把 pending record 字典转成 xlsx 一行。

    record 必含字段:
        approval_date, account_id, audience_tier, adgroup_id, adgroup_name,
        bid_amount_yuan, site_set, age_range,
        landing_video_id, landing_video_url, landing_title,
        material_cover_url, creative_name
    """
    # 创意文案:N 条 description 用 \n 拼接(N=1 时单条)
    descriptions = rec.get("_description_contents") or []
    descriptions_str = "\n".join(descriptions) if descriptions else ""

    # 素材质量数据(2026-06-10 batchByText 升级 — 来自 materialDetail.quality)
    ctr = rec.get("material_ctr")
    cost = rec.get("material_cost")
    roi = rec.get("material_roi")
    imp = rec.get("material_impressions")
    score = rec.get("material_score")
    hit_queries = rec.get("material_recall_hit_queries") or []
    hit_detail = json.dumps(hit_queries, ensure_ascii=False) if hit_queries else ""

    return [
        rec["approval_date"],
        str(rec["account_id"]),
        rec["audience_tier"],
        str(rec["adgroup_id"]),
        rec["adgroup_name"],
        rec["bid_amount_yuan"],
        rec["site_set"],
        rec["age_range"],
        # 落地视频:HYPERLINK(video_url, video_id)
        f'=HYPERLINK("{rec["landing_video_url"]}","{rec["landing_video_id"]}")',
        rec["landing_title"][:30],
        rec.get("landing_risk_level", ""),
        rec.get("landing_risk_tag_ids", ""),
        rec.get("landing_risk_reason", ""),
        rec.get("material_source", "history"),
        # 素材预览:HYPERLINK(cover_url, "查看素材")— 见模块顶部说明
        f'=HYPERLINK("{rec["material_cover_url"]}","查看素材")',
        f'=HYPERLINK("{rec["material_cover_url"]}","打开素材")',
        (
            f'=HYPERLINK("{rec["_external_source_image_url"]}","打开原图")'
            if rec.get("_external_source_image_url") else ""
        ),
        (
            str(int(rec["material_visit_uv_30d"]))
            if rec.get("material_visit_uv_30d") is not None else ""
        ),
        # 创意文案(2026-06-09 加列):description 换行展示
        descriptions_str,
        f"{cost:.2f}" if cost is not None else "",
        f"{roi:.4f}" if roi is not None else "",
        f"{ctr:.4f}" if ctr is not None else "",
        str(imp) if imp is not None else "",
        f"{score:.4f}" if score is not None else "",
        rec.get("material_recall_element_dimension", ""),
        rec.get("material_recall_point_type", ""),
        rec.get("material_recall_standard_element", ""),
        hit_detail,
        rec["creative_name"],
        "",  # 决策列空,等运营填
    ]


def _download_preview_image(url: str, output_path: Path) -> bool:
    try:
        resp = httpx.get(url, timeout=20, follow_redirects=True)
        resp.raise_for_status()
        output_path.write_bytes(resp.content)
        return True
    except Exception as e:
        logger.warning("[im_approval_creation] 素材图下载失败 url=%s error=%s", url, e)
        return False


def _prepare_thumbnail(src_path: Path, dst_path: Path) -> bool:
    try:
        with PilImage.open(src_path) as img:
            img = img.convert("RGB")
            img.thumbnail((180, 100))
            img.save(dst_path, format="JPEG", quality=85)
        return True
    except Exception as e:
        logger.warning(
            "[im_approval_creation] 素材图缩略图生成失败 path=%s error=%s",
            src_path, e,
        )
        return False


def _embed_material_preview_images(ws, records: list[dict]) -> None:
    """Embed thumbnails into the material preview column, keeping hyperlinks as fallback."""
    if not EMBED_MATERIAL_PREVIEW_IMAGES:
        return

    tmp_dir = tempfile.TemporaryDirectory(prefix="creative_preview_")
    tmp_root = Path(tmp_dir.name)
    # Keep the tempdir alive until workbook.save() finishes.
    ws._creative_preview_tmp_dir = tmp_dir  # type: ignore[attr-defined]
    embedded = 0
    for row_idx, rec in enumerate(records, start=2):
        url = str(rec.get("material_cover_url") or "").strip()
        if not url:
            continue
        raw_path = tmp_root / f"raw_{row_idx}"
        thumb_path = tmp_root / f"thumb_{row_idx}.jpg"
        if not _download_preview_image(url, raw_path):
            continue
        if not _prepare_thumbnail(raw_path, thumb_path):
            continue
        try:
            img = OpenpyxlImage(str(thumb_path))
            img.width = 180
            img.height = 100
            ws.add_image(img, f"{MATERIAL_PREVIEW_COL_LETTER}{row_idx}")
            ws.cell(row=row_idx, column=MATERIAL_PREVIEW_COL_IDX).value = "查看素材"
            ws.cell(row=row_idx, column=MATERIAL_PREVIEW_COL_IDX).hyperlink = url
            ws.row_dimensions[row_idx].height = 88
            embedded += 1
        except Exception as e:
            logger.warning(
                "[im_approval_creation] 素材图嵌入失败 row=%d url=%s error=%s",
                row_idx, url, e,
            )

    logger.info(
        "[im_approval_creation] 素材图嵌入完成 embedded=%d/%d",
        embedded, len(records),
    )


def generate_approval_xlsx(records: list[dict], output_path: Path) -> Path:
    """生成待审批 xlsx,含 hyperlink/下拉/颜色/冻结。"""
    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"
    dv.errorTitle = "无效选项"
    dv.prompt = "点击单元格选择决策"
    dv.promptTitle = "审批决策"
    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

    _embed_material_preview_images(ws, records)

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


def _query_first_sheet_id(sheet_token: str) -> str:
    """调 sheets/v3 查 spreadsheet 的 sheets 列表,返回第一个 sheet_id。

    飞书 sheet 内每个 sheet(tab)有独立 sheet_id,后续读 cell 需要用它。
    """
    from tools.feishu_doc import _auth_headers, _get_tenant_token

    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 失败 sheet_token={sheet_token} data={data}"
        )
    sheets = (data.get("data") or {}).get("sheets") or []
    if not sheets:
        raise RuntimeError(f"sheet_token={sheet_token} 无 sheet")
    first = sheets[0]
    sheet_id = first.get("sheet_id", "")
    logger.info(
        "[im_approval_creation] sheet_token=%s 首个 sheet_id=%s",
        sheet_token, sheet_id,
    )
    return sheet_id


async def send_approval_to_feishu(
    xlsx_path: Path,
    chat_id: str = "",
    preamble: str = "",
) -> dict:
    """上传 xlsx → 转飞书 sheet → 发链接到群,返回 sheet 元数据。

    Returns:
        {url, sheet_token, sheet_id, im_sent, im_message_id}
    """
    chat_id = chat_id or FEISHU_OPERATOR_CHAT_ID
    if not preamble:
        preamble = (
            "【创意搭建审批】下方在线表格列出本次准备挂的创意。\n"
            "系统已在创建落地计划前过滤高风险视频,表格中的【风险等级/风险标签/风险原因】为通过项的审核摘要。\n"
            "历史和外部召回素材按相似度≥0.8准入;历史素材按消耗倒序,"
            f"外部素材按最近{EXTERNAL_RECALL_UV_WINDOW_DAYS}天访问UV倒序并经AI清理。\n"
            "请在最右侧【决策】列下拉框选择 approve / reject / hold。\n"
            "我们会在 2 小时内或所有行决策完成后,执行 approve 项,跳过 reject / hold。\n"
            "提示:点击【素材预览】或【落地视频】可在浏览器打开查看(若 403 请右键新标签或复制 URL)。"
        )

    from tools.feishu_doc import import_to_feishu

    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_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]:
    """读 sheet 的"决策"列,返回 {row_idx: action}。

    Args:
        row_count: pending records 数(不含 header)

    Returns:
        {row_idx: action}  — row_idx 是 1-based 数据行编号(1=第一条 record)
        未填或非法值的 row 不在 dict 中,视为"未决策"。
    """
    from tools.feishu_doc import _auth_headers, _get_tenant_token

    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 = None
    last_error: Exception | None = None
    for attempt, delay_seconds in enumerate((0, 2, 5, 10), start=1):
        if delay_seconds:
            time.sleep(delay_seconds)
        try:
            resp = httpx.get(url, headers=_auth_headers(token), timeout=30)
            resp.raise_for_status()
            break
        except httpx.HTTPStatusError as e:
            last_error = e
            status = e.response.status_code
            if status < 500 and status != 429:
                raise
            logger.warning(
                "[poll_approval] 读 sheet HTTP %s,第 %d 次重试 range=%s",
                status, attempt, range_str,
            )
        except httpx.RequestError as e:
            last_error = e
            logger.warning(
                "[poll_approval] 读 sheet 网络异常,第 %d 次重试 range=%s error=%s",
                attempt, range_str, e,
            )
    else:
        assert last_error is not None
        raise last_error
    assert resp is not None
    data = resp.json()
    if data.get("code") != 0:
        raise RuntimeError(f"读 sheet 失败 range={range_str} 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_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]:
    """轮询读 sheet "决策"列,直到所有 row 决策填完或超时。

    Args:
        expected_row_count: 等待决策的总条数
        timeout_minutes: 最长等待(配 config.CREATION_APPROVAL_TIMEOUT_MINUTES,默认 120)
        poll_interval_seconds: 轮询间隔

    Returns:
        {row_idx: action}  — 完整 dict;超时未填的 row 不在 dict
    """
    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_approval] %d/%d 已决策",
                len(actions), expected_row_count,
            )
            last_count = len(actions)
        if len(actions) >= expected_row_count:
            logger.info(
                "[poll_approval] 全部完成 %d/%d, 提前退出轮询",
                len(actions), expected_row_count,
            )
            return actions
        time.sleep(poll_interval_seconds)

    logger.warning(
        "[poll_approval] 超时 %d 分钟, 仅 %d/%d 已决策 — 未决策视为跳过",
        timeout_minutes, last_count, expected_row_count,
    )
    return actions


def run_approval_workflow(
    records: list[dict],
    xlsx_output_dir: Path,
    timeout_minutes: int = CREATION_APPROVAL_TIMEOUT_MINUTES,
) -> tuple[dict, dict[int, str]]:
    """完整审批工作流:生成 xlsx → 发飞书 → 轮询拿决策。

    Args:
        records: Phase 1 准备好的待审批 records(每个 dict 见 _format_record_to_row)
        xlsx_output_dir: xlsx 输出目录
        timeout_minutes: 审批超时

    Returns:
        (sheet_meta, actions)
          sheet_meta:{url, sheet_token, sheet_id, im_sent, im_message_id}
          actions:{row_idx: action},row_idx 1-based 对应 records[row_idx-1]
    """
    now = now_in_timezone()
    date_str = now.strftime("%Y%m%d")
    ts = now.strftime("%H%M%S")
    xlsx_path = xlsx_output_dir / f"creation_approval_{date_str}_{ts}.xlsx"

    generate_approval_xlsx(records, xlsx_path)
    sheet_meta = asyncio.run(send_approval_to_feishu(xlsx_path))
    actions = poll_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