Agent/examples/auto_put_ad_mini/tools/creative_review.py

"""Tencent dynamic creative review result scanner.

This module tracks creatives submitted by the creation pipeline and polls
Tencent's official review-result API. It does not perform prereview.
"""

from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Iterable, Iterator, Optional

logger = logging.getLogger(__name__)


CREATE_REVIEW_TABLES_SQL = [
    """
CREATE TABLE IF NOT EXISTS creative_creation_task (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
    adgroup_id BIGINT DEFAULT NULL COMMENT '广告ID',
    dynamic_creative_id BIGINT NOT NULL COMMENT '动态创意ID',
    dynamic_creative_name VARCHAR(200) DEFAULT NULL COMMENT '动态创意名称',
    landing_video_id BIGINT DEFAULT NULL COMMENT '承接视频ID',
    material_id VARCHAR(100) DEFAULT NULL COMMENT '内部素材ID',
    material_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
    review_status VARCHAR(50) NOT NULL DEFAULT 'submitted' COMMENT '审核状态',
    submit_status VARCHAR(50) NOT NULL DEFAULT 'submitted' COMMENT '提交状态',
    submitted_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '提交时间',
    last_review_checked_at TIMESTAMP NULL DEFAULT NULL COMMENT '最近审核扫描时间',
    review_finished_at TIMESTAMP NULL DEFAULT NULL COMMENT '审核完成时间',
    error TEXT DEFAULT NULL COMMENT '提交或扫描错误',
    raw_record MEDIUMTEXT DEFAULT NULL COMMENT '创建记录JSON',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    UNIQUE KEY uk_account_creative (account_id, dynamic_creative_id),
    KEY idx_review_status (review_status),
    KEY idx_submitted_at (submitted_at),
    KEY idx_landing_submitted (landing_video_id, submitted_at),
    KEY idx_last_review_checked_at (last_review_checked_at),
    KEY idx_account_status (account_id, review_status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='创意创建与审核扫描任务'
""",
    """
CREATE TABLE IF NOT EXISTS creative_review_result (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
    dynamic_creative_id BIGINT NOT NULL COMMENT '动态创意ID',
    review_status VARCHAR(50) NOT NULL COMMENT '解析后的审核状态',
    reject_messages MEDIUMTEXT DEFAULT NULL COMMENT '拒绝原因JSON数组',
    delay_messages MEDIUMTEXT DEFAULT NULL COMMENT '延迟审核原因JSON数组',
    raw_result MEDIUMTEXT NOT NULL COMMENT '腾讯审核结果原文JSON',
    checked_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '扫描时间',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
    UNIQUE KEY uk_account_creative (account_id, dynamic_creative_id),
    KEY idx_review_status (review_status),
    KEY idx_checked_at (checked_at)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='腾讯动态创意审核结果'
""",
    """
CREATE TABLE IF NOT EXISTS creative_rejection_fact (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
    dynamic_creative_id BIGINT NOT NULL COMMENT '动态创意ID',
    fact_type VARCHAR(50) NOT NULL COMMENT '拒绝事实类型:site/element/compose',
    fact_key VARCHAR(255) DEFAULT NULL COMMENT '元素/版位/组件标识',
    reason TEXT NOT NULL COMMENT '拒绝原因',
    site_set VARCHAR(100) DEFAULT NULL COMMENT '影响版位',
    element_type VARCHAR(100) DEFAULT NULL COMMENT '元素类型',
    component_type VARCHAR(100) DEFAULT NULL COMMENT '组件类型',
    raw_detail MEDIUMTEXT DEFAULT NULL COMMENT '原始明细JSON',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
    KEY idx_account_creative (account_id, dynamic_creative_id),
    KEY idx_fact_type (fact_type),
    KEY idx_fact_key (fact_key(100))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='创意审核拒绝事实'
""",
]

FINAL_REVIEW_STATUSES = {"approved", "rejected"}
PENDING_REVIEW_STATUSES = {"submitted", "pending", "unknown"}


@dataclass
class RejectionFact:
    fact_type: str
    reason: str
    fact_key: str = ""
    site_set: str = ""
    element_type: str = ""
    component_type: str = ""
    raw_detail: dict = field(default_factory=dict)


@dataclass
class ParsedReviewResult:
    dynamic_creative_id: Optional[int]
    review_status: str
    reject_messages: list[str] = field(default_factory=list)
    delay_messages: list[str] = field(default_factory=list)
    rejection_facts: list[RejectionFact] = field(default_factory=list)


def _dedupe(values: Iterable[str]) -> list[str]:
    """字符串列表去重，保留首次出现顺序，跳过空字符串。"""
    out: list[str] = []
    seen: set[str] = set()
    for raw in values:
        value = str(raw or "").strip()
        if not value or value in seen:
            continue
        seen.add(value)
        out.append(value)
    return out


def _status_is_rejected(status: str) -> bool:
    """判断审核状态是否为拒绝（REJECT 或 DENIED）。"""
    upper = (status or "").upper()
    return "REJECT" in upper or "DENIED" in upper


def _status_is_pending(status: str) -> bool:
    """判断审核状态是否为待审（PENDING 或 REVIEWING）。"""
    upper = (status or "").upper()
    return "PENDING" in upper or "REVIEWING" in upper


def _status_is_approved(status: str) -> bool:
    """判断审核状态是否为通过（PASS 或 APPROVED 或 NORMAL）。"""
    upper = (status or "").upper()
    return "PASS" in upper or "APPROVED" in upper or "NORMAL" in upper


def _component_type(detail: dict) -> str:
    """从审核明细中提取组件类型（component_type）。"""
    component = detail.get("component_info") or {}
    if isinstance(component, dict):
        return str(component.get("component_type") or "")
    return ""


def _collect_element_facts(result: dict) -> tuple[list[str], list[RejectionFact], bool, bool]:
    """从审核结果中提取元素级拒绝事实，返回 (拒绝消息列表, 拒绝事实列表, 是否有待审, 是否有通过)。"""
    messages: list[str] = []
    facts: list[RejectionFact] = []
    has_pending = False
    has_approved = False

    for element in result.get("element_result_list") or []:
        if not isinstance(element, dict):
            continue
        status = str(element.get("review_status") or "")
        has_pending = has_pending or _status_is_pending(status)
        has_approved = has_approved or _status_is_approved(status)
        if not _status_is_rejected(status):
            continue
        details = element.get("element_reject_detail_info") or []
        if not details:
            details = [{"reason": element.get("reason") or ""}]
        for detail in details:
            if not isinstance(detail, dict):
                continue
            reason = detail.get("reason") or element.get("reason") or ""
            if reason:
                messages.append(str(reason))
                facts.append(RejectionFact(
                    fact_type="element",
                    fact_key=str(
                        element.get("image_id")
                        or element.get("video_id")
                        or element.get("element_name")
                        or ""
                    ),
                    reason=str(reason),
                    element_type=str(element.get("element_type") or ""),
                    component_type=_component_type(element),
                    raw_detail={**element, "element_reject_detail": detail},
                ))
    return messages, facts, has_pending, has_approved


def _collect_site_facts(result: dict) -> tuple[list[str], list[RejectionFact], bool, bool]:
    """从审核结果中提取版位级拒绝事实，返回 (拒绝消息列表, 拒绝事实列表, 是否有待审, 是否有通过)。"""
    messages: list[str] = []
    facts: list[RejectionFact] = []
    has_pending = False
    has_approved = False

    for site in result.get("site_set_result_list") or []:
        if not isinstance(site, dict):
            continue
        status = str(site.get("system_status") or site.get("review_status") or "")
        has_pending = has_pending or _status_is_pending(status)
        has_approved = has_approved or _status_is_approved(status)
        reject_message = site.get("reject_message") or ""
        if _status_is_rejected(status) or reject_message:
            if reject_message:
                messages.append(str(reject_message))
                facts.append(RejectionFact(
                    fact_type="site",
                    fact_key=str(site.get("site_set") or ""),
                    site_set=str(site.get("site_set") or ""),
                    reason=str(reject_message),
                    raw_detail=site,
                ))
        for detail in site.get("element_reject_detail_info") or []:
            if not isinstance(detail, dict):
                continue
            reason = detail.get("reason") or ""
            if reason:
                messages.append(str(reason))
                facts.append(RejectionFact(
                    fact_type="element",
                    fact_key=str(detail.get("element_name") or ""),
                    site_set=str(site.get("site_set") or ""),
                    reason=str(reason),
                    element_type=str(detail.get("element_type") or ""),
                    component_type=_component_type(detail),
                    raw_detail={**detail, "site_set": site.get("site_set")},
                ))
    return messages, facts, has_pending, has_approved


def _collect_compose_facts(result: dict) -> tuple[list[str], list[RejectionFact]]:
    """从审核结果中提取组件组合级拒绝事实，返回 (拒绝消息列表, 拒绝事实列表)。"""
    messages: list[str] = []
    facts: list[RejectionFact] = []
    for item in result.get("reject_component_compose_info_list") or []:
        if not isinstance(item, dict):
            continue
        reason = item.get("reject_message") or ""
        if reason:
            messages.append(str(reason))
            facts.append(RejectionFact(
                fact_type="compose",
                fact_key="component_compose",
                reason=str(reason),
                raw_detail=item,
            ))
    return messages, facts


def parse_review_result(result: dict) -> ParsedReviewResult:
    """Parse Tencent dynamic creative review result into stable internal status."""
    reject_messages: list[str] = []
    delay_messages = [str(v) for v in (result.get("delay_message_list") or []) if v]
    facts: list[RejectionFact] = []
    has_pending = bool(result.get("is_all_component_compose_pending"))
    has_approved = False

    reject_messages.extend(str(v) for v in (result.get("reject_message_list") or []) if v)
    if result.get("reject_component_compose_count"):
        reject_messages.append("组件组合审核拒绝")
    if result.get("pass_component_compose_count"):
        has_approved = True
    if result.get("total_component_compose_count") and result.get("reject_component_compose_count") == 0:
        has_approved = True

    for collector in (_collect_site_facts, _collect_element_facts):
        messages, new_facts, pending, approved = collector(result)
        reject_messages.extend(messages)
        facts.extend(new_facts)
        has_pending = has_pending or pending
        has_approved = has_approved or approved

    messages, new_facts = _collect_compose_facts(result)
    reject_messages.extend(messages)
    facts.extend(new_facts)

    reject_messages = _dedupe(reject_messages)
    delay_messages = _dedupe(delay_messages)

    if reject_messages or facts or int(result.get("reject_component_compose_count") or 0) > 0:
        status = "rejected"
    elif has_pending or delay_messages:
        status = "pending"
    elif has_approved:
        status = "approved"
    else:
        status = "unknown"

    dynamic_creative_id = result.get("dynamic_creative_id")
    try:
        dynamic_creative_id = int(dynamic_creative_id) if dynamic_creative_id is not None else None
    except (TypeError, ValueError):
        dynamic_creative_id = None

    return ParsedReviewResult(
        dynamic_creative_id=dynamic_creative_id,
        review_status=status,
        reject_messages=reject_messages,
        delay_messages=delay_messages,
        rejection_facts=facts,
    )


def group_review_tasks(
    tasks: Iterable[dict],
    batch_size: int = 100,
) -> Iterator[tuple[int, list[int]]]:
    """Yield (account_id, dynamic_creative_ids) batches for Tencent API."""
    by_account: dict[int, list[int]] = {}
    for task in tasks:
        try:
            account_id = int(task["account_id"])
            creative_id = int(task["dynamic_creative_id"])
        except (KeyError, TypeError, ValueError):
            continue
        by_account.setdefault(account_id, []).append(creative_id)

    for account_id in sorted(by_account):
        ids = by_account[account_id]
        for i in range(0, len(ids), batch_size):
            yield account_id, ids[i:i + batch_size]


def ensure_review_tables() -> None:
    """确保审核相关的数据库表（creative_creation_task / creative_review_result / creative_rejection_fact）已创建。"""
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            for sql in CREATE_REVIEW_TABLES_SQL:
                cur.execute(sql)
    finally:
        conn.close()


def record_creation_submission(record: dict, dynamic_creative_id: int) -> None:
    """Persist a successfully submitted creative as pending review scan."""
    ensure_review_tables()
    from db.connection import get_connection

    body = record.get("_request_body") or {}
    raw_record = json.dumps(record, ensure_ascii=False, default=str)
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
INSERT INTO creative_creation_task
    (account_id, adgroup_id, dynamic_creative_id, dynamic_creative_name,
     landing_video_id, material_id, material_image_id,
     review_status, submit_status, submitted_at, raw_record)
VALUES (%s, %s, %s, %s, %s, %s, %s, 'submitted', 'submitted', NOW(), %s)
ON DUPLICATE KEY UPDATE
    adgroup_id=VALUES(adgroup_id),
    dynamic_creative_name=VALUES(dynamic_creative_name),
    landing_video_id=VALUES(landing_video_id),
    material_id=VALUES(material_id),
    material_image_id=VALUES(material_image_id),
    submit_status='submitted',
    review_status=IF(review_status IN ('approved','rejected'), review_status, 'submitted'),
    raw_record=VALUES(raw_record),
    updated_at=CURRENT_TIMESTAMP
""",
                (
                    int(record["account_id"]),
                    int(record.get("adgroup_id") or body.get("adgroup_id") or 0) or None,
                    int(dynamic_creative_id),
                    record.get("creative_name") or body.get("dynamic_creative_name"),
                    record.get("landing_video_id"),
                    str(record.get("_material_id") or ""),
                    str(record.get("_material_image_id") or ""),
                    raw_record,
                ),
            )
    finally:
        conn.close()


def mark_creation_submit_failed(record: dict, error: str) -> None:
    """Persist submit failure for observability."""
    ensure_review_tables()
    from db.connection import get_connection

    body = record.get("_request_body") or {}
    raw_record = json.dumps(record, ensure_ascii=False, default=str)
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
INSERT INTO creative_creation_task
    (account_id, adgroup_id, dynamic_creative_id, dynamic_creative_name,
     landing_video_id, material_id, material_image_id,
     review_status, submit_status, error, raw_record)
VALUES (%s, %s, 0, %s, %s, %s, %s, 'submit_failed', 'failed', %s, %s)
""",
                (
                    int(record["account_id"]),
                    int(record.get("adgroup_id") or body.get("adgroup_id") or 0) or None,
                    record.get("creative_name") or body.get("dynamic_creative_name"),
                    record.get("landing_video_id"),
                    str(record.get("_material_id") or ""),
                    str(record.get("_material_image_id") or ""),
                    error[:2000],
                    raw_record,
                ),
            )
    except Exception as e:
        logger.warning("[creative_review] 记录提交失败状态异常:%s", e)
    finally:
        conn.close()


def load_pending_review_tasks(lookback_hours: int = 72, limit: int = 1000) -> list[dict]:
    """从数据库中加载待审核的创意任务列表（最近 N 小时内提交且状态为 submitted/pending/unknown）。"""
    ensure_review_tables()
    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
SELECT account_id, dynamic_creative_id
FROM creative_creation_task
WHERE dynamic_creative_id > 0
  AND review_status IN ('submitted','pending','unknown')
  AND submitted_at >= DATE_SUB(NOW(), INTERVAL %s HOUR)
ORDER BY COALESCE(last_review_checked_at, submitted_at) ASC
LIMIT %s
""",
                (int(lookback_hours), int(limit)),
            )
            return list(cur.fetchall())
    finally:
        conn.close()


def fetch_dynamic_creative_review_results(
    account_id: int,
    creative_ids: list[int],
) -> list[dict]:
    """调腾讯 API 批量查询动态创意的审核结果。"""
    if not creative_ids:
        return []
    from tools.ad_api import _check, _get

    resp = _get(
        "/dynamic_creative_review_results/get",
        {
            "account_id": account_id,
            "dynamic_creative_id_list": [int(v) for v in creative_ids],
        },
    )
    data = _check(resp, "dynamic_creative_review_results/get")
    return data.get("list") or []


def save_review_result(account_id: int, raw_result: dict) -> ParsedReviewResult:
    """解析并持久化一条创意的审核结果，同步更新 creation_task 状态与 rejection_fact 明细。"""
    ensure_review_tables()
    parsed = parse_review_result(raw_result)
    if parsed.dynamic_creative_id is None:
        raise ValueError("审核结果缺 dynamic_creative_id")

    raw_json = json.dumps(raw_result, ensure_ascii=False, default=str)
    reject_json = json.dumps(parsed.reject_messages, ensure_ascii=False)
    delay_json = json.dumps(parsed.delay_messages, ensure_ascii=False)
    finished_at_sql = "NOW()" if parsed.review_status in FINAL_REVIEW_STATUSES else "NULL"

    from db.connection import get_connection

    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
INSERT INTO creative_review_result
    (account_id, dynamic_creative_id, review_status, reject_messages, delay_messages, raw_result, checked_at)
VALUES (%s, %s, %s, %s, %s, %s, NOW())
ON DUPLICATE KEY UPDATE
    review_status=VALUES(review_status),
    reject_messages=VALUES(reject_messages),
    delay_messages=VALUES(delay_messages),
    raw_result=VALUES(raw_result),
    checked_at=NOW(),
    updated_at=CURRENT_TIMESTAMP
""",
                (
                    int(account_id),
                    int(parsed.dynamic_creative_id),
                    parsed.review_status,
                    reject_json,
                    delay_json,
                    raw_json,
                ),
            )
            cur.execute(
                f"""
UPDATE creative_creation_task
SET review_status=%s,
    last_review_checked_at=NOW(),
    review_finished_at={finished_at_sql},
    updated_at=CURRENT_TIMESTAMP
WHERE account_id=%s AND dynamic_creative_id=%s
""",
                (parsed.review_status, int(account_id), int(parsed.dynamic_creative_id)),
            )
            cur.execute(
                """
DELETE FROM creative_rejection_fact
WHERE account_id=%s AND dynamic_creative_id=%s
""",
                (int(account_id), int(parsed.dynamic_creative_id)),
            )
            if parsed.rejection_facts:
                cur.executemany(
                    """
INSERT INTO creative_rejection_fact
    (account_id, dynamic_creative_id, fact_type, fact_key, reason,
     site_set, element_type, component_type, raw_detail)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
""",
                    [
                        (
                            int(account_id),
                            int(parsed.dynamic_creative_id),
                            fact.fact_type,
                            fact.fact_key[:255] if fact.fact_key else None,
                            fact.reason,
                            fact.site_set or None,
                            fact.element_type or None,
                            fact.component_type or None,
                            json.dumps(fact.raw_detail, ensure_ascii=False, default=str),
                        )
                        for fact in parsed.rejection_facts
                    ],
                )
    finally:
        conn.close()

    return parsed


def scan_pending_reviews(
    lookback_hours: int = 72,
    limit: int = 1000,
    batch_size: int = 100,
) -> dict:
    """Scan pending submitted creatives and persist official review results."""
    tasks = load_pending_review_tasks(lookback_hours=lookback_hours, limit=limit)
    summary = {
        "run_started": datetime.now(timezone.utc).isoformat(),
        "tasks": len(tasks),
        "batches": 0,
        "results": 0,
        "approved": 0,
        "rejected": 0,
        "pending": 0,
        "unknown": 0,
        "errors": [],
    }
    for account_id, creative_ids in group_review_tasks(tasks, batch_size=batch_size):
        summary["batches"] += 1
        try:
            results = fetch_dynamic_creative_review_results(account_id, creative_ids)
        except Exception as e:
            err = f"account={account_id} ids={len(creative_ids)} error={e}"
            logger.exception("[creative_review] 查询审核结果失败:%s", err)
            summary["errors"].append(err)
            continue
        for raw in results:
            try:
                parsed = save_review_result(account_id, raw)
            except Exception as e:
                err = f"account={account_id} creative={raw.get('dynamic_creative_id')} error={e}"
                logger.exception("[creative_review] 保存审核结果失败:%s", err)
                summary["errors"].append(err)
                continue
            summary["results"] += 1
            summary[parsed.review_status] = summary.get(parsed.review_status, 0) + 1
            logger.info(
                "[creative_review] account=%d creative=%s status=%s reject=%d delay=%d",
                account_id,
                parsed.dynamic_creative_id,
                parsed.review_status,
                len(parsed.reject_messages),
                len(parsed.delay_messages),
            )

    summary["run_finished"] = datetime.now(timezone.utc).isoformat()
    return summary