"""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