完善自动投放创意审核与素材召回

examples/auto_put_ad_mini/config.py

@@ -9,8 +9,10 @@
 """
 import os
 import logging
+from datetime import datetime
 from pathlib import Path
 from typing import Optional
+from zoneinfo import ZoneInfo
 from agent.core.runner import RunConfig, KnowledgeConfig
 
 # 初始化 logger（必须在使用前定义）
@@ -77,11 +79,16 @@ LOG_LEVEL = "INFO"
 LOG_FILE = None
 
 # ═══════════════════════════════════════════
-# 时区配置（海外部署）
+# 时区配置
 # ═══════════════════════════════════════════
-TIMEZONE = os.getenv("TZ", "UTC")
+TIMEZONE = os.getenv("TZ", "Asia/Shanghai")
 logger.info(f"运行时区：{TIMEZONE}")
 
+
+def now_in_timezone() -> datetime:
+    """返回项目统一业务时区的 aware datetime。"""
+    return datetime.now(ZoneInfo(TIMEZONE))
+
 # ═══════════════════════════════════════════
 # V3 数据窗口配置
 # ═══════════════════════════════════════════
@@ -947,10 +954,10 @@ VIDEO_RISK_MAX_ALLOWED_LEVEL = int(os.getenv("VIDEO_RISK_MAX_ALLOWED_LEVEL", "1"
 VIDEO_RISK_API_TIMEOUT_SECONDS = int(os.getenv("VIDEO_RISK_API_TIMEOUT_SECONDS", "10"))
 
 # --- 素材召回质量过滤(2026-06-10 用户确认,batchByText 升级)---
-# 服务端 ranking 参数:simThreshold=0.7(语义相关),alpha=0(不要 sim 加权)
+# 服务端 ranking 参数:simThreshold=0.8(语义相关),alpha=0(不要 sim 加权)
 # wCtr=1, wCvr=wRoi=wOpenRate=wFissionRate=0 → qualityScore 由 ctr 主导 → 等价"取 CTR 最高"
 # 客户端 impressions > 阈值 二次筛 → 保证"语义相关 + 曝光足够 + CTR 最高"
-RECALL_SIM_THRESHOLD = 0.7
+RECALL_SIM_THRESHOLD = 0.8
 RECALL_ALPHA = 0
 RECALL_W_CTR = 1
 RECALL_W_CVR = 0

examples/auto_put_ad_mini/db/schema.sql

@@ -168,6 +168,99 @@ CREATE TABLE IF NOT EXISTS decision_history (
     KEY idx_created_at (created_at)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='决策历史记录';
 
+-- =====================================================
+-- 8. 创意创建与审核扫描任务表
+-- =====================================================
+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_last_review_checked_at (last_review_checked_at),
+    KEY idx_account_status (account_id, review_status)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='创意创建与审核扫描任务';
+
+-- =====================================================
+-- 9. 腾讯动态创意审核结果表
+-- =====================================================
+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='腾讯动态创意审核结果';
+
+-- =====================================================
+-- 10. 创意审核拒绝事实表
+-- =====================================================
+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='创意审核拒绝事实';
+
+-- =====================================================
+-- 11. 视频元素特征缓存表（ODPS 最新分区缓存）
+-- =====================================================
+CREATE TABLE IF NOT EXISTS video_element_feature_cache (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
+    video_id BIGINT NOT NULL COMMENT '业务视频ID，对应 ODPS vid',
+    dt VARCHAR(8) NOT NULL COMMENT 'ODPS 分区日期 YYYYMMDD',
+    point_type VARCHAR(50) NOT NULL COMMENT '点类型',
+    standard_element VARCHAR(255) NOT NULL COMMENT '标准化元素',
+    element_dimension VARCHAR(50) NOT NULL DEFAULT '实质' COMMENT '元素维度',
+    contribution_score DOUBLE NOT NULL DEFAULT 0 COMMENT '贡献分',
+    is_miss BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否为无特征负缓存',
+    raw_element VARCHAR(255) DEFAULT NULL COMMENT '原始元素名称',
+    element_id VARCHAR(100) DEFAULT NULL COMMENT '标准化元素ID',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+
+    UNIQUE KEY uk_dt_video_point_element (dt, video_id, point_type, standard_element),
+    KEY idx_video_dt (video_id, dt),
+    KEY idx_dt (dt),
+    KEY idx_updated_at (updated_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='视频元素特征缓存';
+
 -- =====================================================
 -- 初始化数据
 -- =====================================================
@@ -180,6 +273,7 @@ INSERT INTO system_config (config_key, config_value, value_type, description) VA
 ('run_on_startup', 'false', 'boolean', '服务启动时是否立即执行一次'),
 ('max_concurrent_tasks', '1', 'int', '最大并发任务数'),
 ('approval_timeout_minutes', '30', 'int', '飞书审批超时时间（分钟）'),
+('creative_review_scan_interval_hours', '2', 'int', '创意正式审核结果扫描间隔（小时）'),
 ('roi_low_factor', '0.75', 'string', '关停线系数（ROI < 渠道P50 * 0.75 触发关停）'),
 ('bid_down_roi_factor', '0.90', 'string', '降价线系数（ROI < 渠道P50 * 0.90 触发降价）'),
 ('bid_up_roi_factor', '1.05', 'string', '提价线系数（ROI > 渠道P50 * 1.05 触发提价）')

examples/auto_put_ad_mini/execute_creation_apply.py

@@ -16,7 +16,6 @@ import asyncio
 import json
 import logging
 import sys
-from datetime import datetime, timezone
 from pathlib import Path
 
 _HERE = Path(__file__).parent
@@ -26,7 +25,7 @@ sys.path.insert(0, str(_HERE))
 from dotenv import load_dotenv  # noqa: E402
 load_dotenv(_HERE / ".env")
 
-from config import FEISHU_OPERATOR_CHAT_ID  # noqa: E402
+from config import FEISHU_OPERATOR_CHAT_ID, now_in_timezone  # noqa: E402
 
 # 强占 sys.path[0],绕过 config import 副作用(im-client/tools.py 同名冲突)
 while str(_HERE) in sys.path:
@@ -34,6 +33,10 @@ while str(_HERE) in sys.path:
 sys.path.insert(0, str(_HERE))
 
 from tools.creative_creation import post_creative_with_prepared_body  # noqa: E402
+from tools.creative_review import (  # noqa: E402
+    mark_creation_submit_failed,
+    record_creation_submission,
+)
 
 logger = logging.getLogger("execute_creation_apply")
 
@@ -166,7 +169,7 @@ def apply_pending_records(records: list[dict]) -> dict:
           "records": [...]  # 每条含 action + dynamic_creative_id(成功时) + error(失败时)
         }
     """
-    run_started = datetime.now(timezone.utc).isoformat()
+    run_started = now_in_timezone().isoformat()
     posted_ok = 0
     posted_failed = 0
     approved_total = 0
@@ -197,12 +200,17 @@ def apply_pending_records(records: list[dict]) -> dict:
         if cid:
             rec["dynamic_creative_id"] = str(cid)
             posted_ok += 1
+            try:
+                record_creation_submission(rec, int(cid))
+            except Exception as e:
+                logger.warning("[apply] 记录创意审核扫描任务失败 cid=%s: %s", cid, e)
         else:
             rec["error"] = "post_failed"
             posted_failed += 1
+            mark_creation_submit_failed(rec, "post_failed")
         out_records.append(rec)
 
-    run_finished = datetime.now(timezone.utc).isoformat()
+    run_finished = now_in_timezone().isoformat()
     return {
         "run_started": run_started,
         "run_finished": run_finished,
@@ -235,8 +243,9 @@ def _strip_body_for_json(records: list[dict]) -> list[dict]:
 
 def write_summary(summary: dict, output_dir: Path) -> Path:
     output_dir.mkdir(parents=True, exist_ok=True)
-    date_str = datetime.now(timezone.utc).strftime("%Y%m%d")
-    ts = datetime.now(timezone.utc).strftime("%H%M%S")
+    now = now_in_timezone()
+    date_str = now.strftime("%Y%m%d")
+    ts = now.strftime("%H%M%S")
     out_path = output_dir / f"creation_run_{date_str}_{ts}.json"
     persisted = dict(summary)
     persisted["records"] = _strip_body_for_json(summary["records"])

examples/auto_put_ad_mini/execute_creation_once.py

@@ -25,7 +25,6 @@ import json
 import logging
 import sys
 import time
-from datetime import datetime, timezone
 from pathlib import Path
 
 _HERE = Path(__file__).parent
@@ -42,6 +41,7 @@ from config import (  # noqa: E402
     TARGET_CREATIVES_PER_AD,
     WHITELIST_ACCOUNTS,
     get_creation_account_ids,
+    now_in_timezone,
 )
 
 # config import 副作用会改 sys.path(把 im-client / agent/tools/builtin 顶到前面),
@@ -169,8 +169,6 @@ def phase0_create_ads(target_ads: int = ADS_PER_ACCOUNT) -> list[dict]:
 
     Returns: 本次新建成功的广告 list[dict],每项含 {account_id, adgroup_id, adgroup_name, wechat_position}
     """
-    from datetime import datetime, timezone
-
     from tools.ad_api import _get
     from tools.im_approval_ad_creation import run_ad_approval_workflow
 
@@ -240,13 +238,13 @@ def phase0_create_ads(target_ads: int = ADS_PER_ACCOUNT) -> list[dict]:
             logger.warning("[phase0]   enumerate 返回 0 条候选,跳过")
             continue
 
-        today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
+        today = now_in_timezone().strftime("%Y-%m-%d")
         for c in candidates:
             age_range = ",".join(
                 f"{a.get('min')}-{a.get('max')}" for a in c.age
             )
-            # 2026-06-10 修复:Phase 0 经过飞书审批后,广告应立即 NORMAL 生效
-            # build_ad_request_body 的 default SUSPEND 是模块 A dry-run 时代的旧 SOP
+            # 新建广告默认开启;实际投放仍受创意审核/账户/预算控制。
+            # 这里显式传 NORMAL,避免未来默认值变化影响端到端创建语义。
             body = build_ad_request_body(c, configured_status="AD_STATUS_NORMAL")
             rec = {
                 "approval_date": today,
@@ -446,8 +444,9 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
 def _write_pending_records(records: list[dict], output_dir: Path) -> Path:
     """落 Phase 1 产物 JSON(供 Phase 3 独立 apply 用 / 追溯)。"""
     output_dir.mkdir(parents=True, exist_ok=True)
-    date_str = datetime.now(timezone.utc).strftime("%Y%m%d")
-    ts = datetime.now(timezone.utc).strftime("%H%M%S")
+    now = now_in_timezone()
+    date_str = now.strftime("%Y%m%d")
+    ts = now.strftime("%H%M%S")
     out_path = output_dir / f"creation_pending_{date_str}_{ts}.json"
     with open(out_path, "w", encoding="utf-8") as f:
         json.dump(records, f, ensure_ascii=False, indent=2)
@@ -475,7 +474,7 @@ def phase2_approval(records: list[dict], xlsx_output_dir: Path) -> dict[int, str
 
 def run_once() -> dict:
     """完整主循环:Phase 0 → Phase 1 → Phase 2(若开关 True)→ Phase 3。"""
-    run_started = datetime.now(timezone.utc).isoformat()
+    run_started = now_in_timezone().isoformat()
     sync_stats: dict = {}
 
     # Phase -1:每日主流程启动时先同步飞书「自动化账户」配置表。
@@ -490,7 +489,7 @@ def run_once() -> dict:
         logger.exception("[main] Phase -1 同步飞书配置失败,本轮停止:%s", e)
         return {
             "run_started": run_started,
-            "run_finished": datetime.now(timezone.utc).isoformat(),
+            "run_finished": now_in_timezone().isoformat(),
             "approval_required": CREATION_APPROVAL_REQUIRED,
             "sync_stats": {"error": str(e)},
             "phase0_created_ads": 0,
@@ -512,7 +511,7 @@ def run_once() -> dict:
         logger.info("[main] Phase 1 无 pending records,主循环退出")
         return {
             "run_started": run_started,
-            "run_finished": datetime.now(timezone.utc).isoformat(),
+            "run_finished": now_in_timezone().isoformat(),
             "approval_required": CREATION_APPROVAL_REQUIRED,
             "sync_stats": sync_stats,
             "phase0_created_ads": len(created_ads),
@@ -547,7 +546,7 @@ def run_once() -> dict:
     _send_apply_summary_to_feishu(summary)
 
     summary["run_started"] = run_started
-    summary["run_finished"] = datetime.now(timezone.utc).isoformat()
+    summary["run_finished"] = now_in_timezone().isoformat()
     summary["sync_stats"] = sync_stats
     summary["phase0_created_ads"] = len(created_ads)
     summary["phase0_ads"] = created_ads

examples/auto_put_ad_mini/k8s/configmap.yaml

@@ -24,5 +24,10 @@ data:
   RUN_ON_STARTUP: "false"
   PORT: "8080"
 
+  # 创意正式审核结果扫描配置
+  CREATIVE_REVIEW_LOOKBACK_HOURS: "72"
+  CREATIVE_REVIEW_SCAN_LIMIT: "1000"
+  CREATIVE_REVIEW_BATCH_SIZE: "100"
+
   # 日志配置
   LOG_LEVEL: "INFO"

examples/auto_put_ad_mini/k8s/cronjob_creative_review.yaml

@@ -0,0 +1,47 @@
+# 腾讯动态创意正式审核结果扫描
+# 每 2 小时扫描一次已提交创意,拉取拒绝原因/感知审核结果并写入 DB。
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: auto-put-ad-mini-creative-review
+  namespace: ad-automation
+  labels:
+    app: auto-put-ad-mini
+    component: creative-review
+spec:
+  schedule: "0 */2 * * *"
+  concurrencyPolicy: Forbid
+  successfulJobsHistoryLimit: 3
+  failedJobsHistoryLimit: 3
+  jobTemplate:
+    spec:
+      backoffLimit: 1
+      activeDeadlineSeconds: 1800
+      template:
+        metadata:
+          labels:
+            app: auto-put-ad-mini
+            component: creative-review
+        spec:
+          restartPolicy: Never
+          containers:
+          - name: creative-review-scanner
+            image: your-registry/auto-put-ad-mini:latest
+            imagePullPolicy: Always
+            command: ["python", "scan_creative_reviews.py"]
+            envFrom:
+            - configMapRef:
+                name: ad-config
+            - secretRef:
+                name: ad-secrets
+            resources:
+              requests:
+                memory: "512Mi"
+                cpu: "250m"
+              limits:
+                memory: "1Gi"
+                cpu: "500m"
+            securityContext:
+              runAsNonRoot: true
+              runAsUser: 1000
+              allowPrivilegeEscalation: false

examples/auto_put_ad_mini/scan_creative_reviews.py

@@ -0,0 +1,68 @@
+"""Scan Tencent official dynamic creative review results.
+
+Run every 2 hours in production. This scanner only reads submitted creatives
+from DB and writes official review results/rejection facts back to DB.
+"""
+
+import json
+import logging
+import os
+import sys
+from pathlib import Path
+
+_HERE = Path(__file__).parent
+sys.path.insert(0, str(_HERE.parent.parent))
+sys.path.insert(0, str(_HERE))
+
+from dotenv import load_dotenv  # noqa: E402
+load_dotenv(_HERE / ".env")
+
+while str(_HERE) in sys.path:
+    sys.path.remove(str(_HERE))
+sys.path.insert(0, str(_HERE))
+
+from tools.creative_review import scan_pending_reviews  # noqa: E402
+
+logger = logging.getLogger("scan_creative_reviews")
+
+
+def _setup_logging() -> None:
+    logging.basicConfig(
+        level=os.getenv("LOG_LEVEL", "INFO"),
+        format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
+        datefmt="%H:%M:%S",
+    )
+    for noisy in ("httpx", "httpcore", "config", "db.config"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
+    try:
+        from tools.sls_setup import attach_sls_handler
+        attach_sls_handler()
+    except Exception as e:
+        logger.warning("[sls] 挂载异常(降级为本地 only):%s", e)
+
+
+def main() -> int:
+    _setup_logging()
+    lookback_hours = int(os.getenv("CREATIVE_REVIEW_LOOKBACK_HOURS", "72"))
+    limit = int(os.getenv("CREATIVE_REVIEW_SCAN_LIMIT", "1000"))
+    batch_size = int(os.getenv("CREATIVE_REVIEW_BATCH_SIZE", "100"))
+
+    logger.info(
+        "[creative_review] 扫描启动 lookback_hours=%d limit=%d batch_size=%d",
+        lookback_hours, limit, batch_size,
+    )
+    summary = scan_pending_reviews(
+        lookback_hours=lookback_hours,
+        limit=limit,
+        batch_size=batch_size,
+    )
+    logger.info(
+        "[creative_review] 扫描完成 %s",
+        json.dumps(summary, ensure_ascii=False),
+    )
+    return 0 if not summary.get("errors") else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

examples/auto_put_ad_mini/test_ad_creation_status.py

@@ -0,0 +1,44 @@
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parent
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+PROJECT_ROOT = ROOT.parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from tools.ad_creation import AdCandidate, build_ad_request_body
+
+
+def _candidate() -> AdCandidate:
+    return AdCandidate(
+        account_id=123,
+        adgroup_name="test-ad",
+        site_set=["SITE_SET_WECHAT"],
+        custom_audience=None,
+        bid_amount_fen=35,
+        audience_tier_label="泛人群",
+        age=[{"min": 45, "max": 66}],
+        location_types=["LIVE_IN"],
+        region_ids=[1],
+        daily_budget_fen=10000,
+        time_series="1" * 48,
+        delivery_version="miniapp_user_growth_v1",
+        fingerprint="fp",
+        wechat_position=None,
+    )
+
+
+class AdCreationStatusTest(unittest.TestCase):
+    def test_ad_request_defaults_to_normal_status(self):
+        with patch("tools.ad_creation.get_account_feedback_id", return_value=6700001):
+            body = build_ad_request_body(_candidate(), begin_date="2026-07-01")
+
+        self.assertEqual(body["configured_status"], "AD_STATUS_NORMAL")
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/auto_put_ad_mini/test_creative_review_scan.py

@@ -0,0 +1,72 @@
+import unittest
+
+
+class CreativeReviewScanTests(unittest.TestCase):
+    def test_parse_rejected_result_collects_reasons_and_locations(self):
+        from tools.creative_review import parse_review_result
+
+        parsed = parse_review_result({
+            "dynamic_creative_id": 123,
+            "reject_message_list": ["整体拒绝"],
+            "site_set_result_list": [
+                {
+                    "site_set": "SITE_SET_MOMENTS",
+                    "system_status": "DYNAMIC_CREATIVE_STATUS_DENIED",
+                    "reject_message": "朋友圈拒绝",
+                }
+            ],
+            "element_result_list": [
+                {
+                    "element_name": "image",
+                    "element_type": "ELEMENT_TYPE_IMAGE",
+                    "review_status": "REVIEW_STATUS_REJECTED",
+                    "element_reject_detail_info": [
+                        {
+                            "reason": "图片含违规信息",
+                            "reject_info_location": [
+                                {"x": 1, "y": 2, "width": 3, "height": 4}
+                            ],
+                        }
+                    ],
+                }
+            ],
+        })
+
+        self.assertEqual(parsed.review_status, "rejected")
+        self.assertIn("整体拒绝", parsed.reject_messages)
+        self.assertIn("朋友圈拒绝", parsed.reject_messages)
+        self.assertIn("图片含违规信息", parsed.reject_messages)
+        self.assertEqual(len(parsed.rejection_facts), 2)
+
+    def test_parse_pending_result(self):
+        from tools.creative_review import parse_review_result
+
+        parsed = parse_review_result({
+            "dynamic_creative_id": 456,
+            "is_all_component_compose_pending": True,
+            "delay_message_list": ["审核延迟"],
+        })
+
+        self.assertEqual(parsed.review_status, "pending")
+        self.assertIn("审核延迟", parsed.delay_messages)
+
+    def test_chunk_ids_by_account_limits_to_100(self):
+        from tools.creative_review import group_review_tasks
+
+        tasks = [
+            {"account_id": 1, "dynamic_creative_id": i}
+            for i in range(250)
+        ] + [{"account_id": 2, "dynamic_creative_id": 999}]
+
+        groups = list(group_review_tasks(tasks, batch_size=100))
+
+        self.assertEqual(len(groups), 4)
+        self.assertEqual(groups[0][0], 1)
+        self.assertEqual(len(groups[0][1]), 100)
+        self.assertEqual(len(groups[1][1]), 100)
+        self.assertEqual(len(groups[2][1]), 50)
+        self.assertEqual(groups[3], (2, [999]))
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/auto_put_ad_mini/test_hot_video_feature_enrichment.py

@@ -0,0 +1,183 @@
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parent
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+PROJECT_ROOT = ROOT.parents[1]
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from tools.material_recall import _pick_query_strategies_for_batch
+from tools.creative_creation import _is_landing_candidate
+from tools.video_recall import LandingVideo
+
+
+def _landing(**kwargs):
+    base = dict(
+        video_id=1,
+        title="",
+        cover_url="",
+        video_url="",
+        score=0.0,
+        rov=0.0,
+        sim=0.0,
+        visit_uv=0,
+        category="",
+        standard_element="",
+        category_name="",
+        demand_content_title="",
+        demand_content_topic="",
+        demand_content_id="",
+        demand_type="",
+        point_type="",
+        dimension="",
+        experiment_id="",
+        raw={},
+    )
+    base.update(kwargs)
+    return LandingVideo(**base)
+
+
+class HotVideoFeatureEnrichmentTest(unittest.TestCase):
+    def test_hot_video_element_features_expand_into_multiple_recall_strategies(self):
+        landing = _landing(
+            raw={
+                "element_features": [
+                    {"point_type": "关键点", "standard_element": "孝道观念"},
+                    {"point_type": "灵感点", "standard_element": "养老院"},
+                ]
+            },
+        )
+
+        strategies = _pick_query_strategies_for_batch(landing)
+
+        self.assertEqual(
+            strategies[:2],
+            [
+                ("孝道观念", "KEYPOINT_SUBSTANCE", "标准化元素-关键点"),
+                ("养老院", "INSPIRATION_SUBSTANCE", "标准化元素-灵感点"),
+            ],
+        )
+
+    def test_landing_candidate_requires_cluster_feature_point_demand_type(self):
+        self.assertTrue(_is_landing_candidate(_landing(
+            demand_type="聚类特征点",
+            point_type="灵感点",
+            standard_element="退休书",
+        )))
+        self.assertFalse(_is_landing_candidate(_landing(
+            demand_type="tree泛化特征点",
+            point_type="灵感点",
+            standard_element="奇幻视觉",
+        )))
+
+    def test_hot_landing_candidate_allows_odps_enriched_element_features_without_demand_type(self):
+        self.assertTrue(_is_landing_candidate(_landing(
+            demand_type="",
+            point_type="关键点",
+            standard_element="孝道观念",
+            raw={
+                "source": "hot",
+                "element_features": [
+                    {"point_type": "关键点", "standard_element": "孝道观念"},
+                ],
+            },
+        )))
+
+    def test_video_feature_query_uses_latest_dt_cache_miss_substance_dimension_and_score_threshold(self):
+        from tools import video_feature_query
+
+        captured_sql = []
+
+        class FakeClient:
+            def query(self, sql):
+                captured_sql.append(sql)
+                return [
+                    {
+                        "vid": "70242543",
+                        "point_type": "关键点",
+                        "standard_element": "孝道观念",
+                        "contribution_score": 0.91,
+                        "dt": "20260701",
+                    },
+                    {
+                        "vid": "70242543",
+                        "point_type": "目的点",
+                        "standard_element": "家庭矛盾",
+                        "contribution_score": 0.86,
+                        "dt": "20260701",
+                    },
+                ]
+
+        with (
+            patch.object(video_feature_query, "_get_odps_client", return_value=FakeClient()),
+            patch.object(video_feature_query, "_latest_partition_dt", return_value="20260701"),
+            patch.object(video_feature_query, "_ensure_cache_table"),
+            patch.object(video_feature_query, "_read_cache", return_value=({}, set())),
+            patch.object(video_feature_query, "_write_cache") as write_cache,
+        ):
+            features = video_feature_query.fetch_video_element_features([70242543])
+
+        self.assertEqual(len(features[70242543]), 2)
+        write_cache.assert_called_once_with(features, [], "20260701")
+        sql = captured_sql[0]
+        self.assertIn("dt = '20260701'", sql)
+        self.assertIn("`元素维度` = '实质'", sql)
+        self.assertIn("`贡献分` >= 0.8", sql)
+
+    def test_video_feature_query_cache_hit_skips_odps_element_query(self):
+        from tools import video_feature_query
+        from tools.video_feature_query import VideoElementFeature
+
+        class FakeClient:
+            def query(self, sql):
+                raise AssertionError("cache hit should not query ODPS elements")
+
+        cached = {
+            70242543: [
+                VideoElementFeature(
+                    video_id=70242543,
+                    point_type="关键点",
+                    standard_element="孝道观念",
+                    contribution_score=0.91,
+                    dt="20260701",
+                )
+            ]
+        }
+        with (
+            patch.object(video_feature_query, "_get_odps_client", return_value=FakeClient()),
+            patch.object(video_feature_query, "_latest_partition_dt", return_value="20260701"),
+            patch.object(video_feature_query, "_ensure_cache_table"),
+            patch.object(video_feature_query, "_read_cache", return_value=(cached, {70242543})),
+            patch.object(video_feature_query, "_write_cache") as write_cache,
+        ):
+            features = video_feature_query.fetch_video_element_features([70242543])
+
+        self.assertEqual(features, cached)
+        write_cache.assert_not_called()
+
+    def test_video_feature_query_writes_negative_cache_for_missing_ids(self):
+        from tools import video_feature_query
+
+        class FakeClient:
+            def query(self, sql):
+                return []
+
+        with (
+            patch.object(video_feature_query, "_get_odps_client", return_value=FakeClient()),
+            patch.object(video_feature_query, "_latest_partition_dt", return_value="20260701"),
+            patch.object(video_feature_query, "_ensure_cache_table"),
+            patch.object(video_feature_query, "_read_cache", return_value=({}, set())),
+            patch.object(video_feature_query, "_write_cache") as write_cache,
+        ):
+            features = video_feature_query.fetch_video_element_features([70242543])
+
+        self.assertEqual(features, {})
+        write_cache.assert_called_once_with({}, [70242543], "20260701")
+
+
+if __name__ == "__main__":
+    unittest.main()

examples/auto_put_ad_mini/tools/ad_creation.py

@@ -260,12 +260,13 @@ def enumerate_new_ad_candidates(
 def build_ad_request_body(
     candidate: AdCandidate,
     begin_date: Optional[str] = None,
-    configured_status: str = "AD_STATUS_SUSPEND",
+    configured_status: str = "AD_STATUS_NORMAL",
 ) -> dict:
     """生成 /v3.0/adgroups/add 的完整请求 body。
 
     关键点:
-    - configured_status 默认 SUSPEND(创建后默认暂停,运营 review 后启用)
+    - configured_status 默认 NORMAL:新广告先开启,实际投放受创意审核/账户/预算控制
+      如需暂停创建,调用方显式传 AD_STATUS_SUSPEND
     - 不传 conversion_id(可选 + 不支持朋友圈,我们有朋友圈版位)
     - 不传顶层 marketing_target_type,只通过 marketing_asset_outer_spec 传
       (待 dry run 验证;若腾讯要求,加上)

examples/auto_put_ad_mini/tools/audience_grant.py

@@ -20,6 +20,19 @@ DEFAULT_AUDIENCE_SOURCE_ACCOUNT_ID = int(
 AUDIENCE_GRANT_BUSINESS_ID = int(os.getenv("TENCENT_AUDIENCE_GRANT_BUSINESS_ID", "0") or 0)
 TARGET_PERMISSION = "GRANT_PERMISSION_TYPE_TARGET"
 
+AUDIENCE_NAME_ALIASES = {
+    # 业务配置名保留给内容服务 crowdPackage 使用;腾讯侧真实人群包命名是 R_330+。
+    "回流330以上人群": ["R_330+", "R330+"],
+}
+
+
+def _audience_lookup_names(audience_name: str) -> list[str]:
+    names = [audience_name]
+    for alias in AUDIENCE_NAME_ALIASES.get(audience_name, []):
+        if alias and alias not in names:
+            names.append(alias)
+    return names
+
 
 def _fetch_custom_audiences(account_id: int) -> list[dict]:
     out: list[dict] = []
@@ -53,32 +66,36 @@ def _pick_source_audience(source_account_id: int, audience_name: str) -> dict:
         if item.get("status") == "SUCCESS"
         and item.get("online_status") == "ONLINE"
     ]
+    lookup_names = _audience_lookup_names(audience_name)
+
+    def score_one(item_name: str, lookup_name: str) -> int:
+        if item_name == lookup_name:
+            return 3
+        if item_name.startswith(f"{lookup_name}_"):
+            return 2
+        if lookup_name in item_name:
+            return 1
+        return 0
 
     def score(item: dict) -> tuple[int, str]:
         name = item.get("name") or ""
-        if name == audience_name:
-            rank = 3
-        elif name.startswith(f"{audience_name}_"):
-            rank = 2
-        elif audience_name in name:
-            rank = 1
-        else:
-            rank = 0
+        rank = max(score_one(name, lookup_name) for lookup_name in lookup_names)
         return rank, str(item.get("created_time") or "")
 
     matches = [item for item in usable if score(item)[0] > 0]
     if not matches:
         raise RuntimeError(
             f"source_account={source_account_id} 未找到 ONLINE/SUCCESS 人群包:"
-            f" audience_name={audience_name!r}"
+            f" audience_name={audience_name!r}, lookup_names={lookup_names!r}"
         )
 
     matches.sort(key=score, reverse=True)
     chosen = matches[0]
     logger.info(
-        "[audience_grant] source=%d audience_name=%r → audience_id=%s full_name=%r",
+        "[audience_grant] source=%d audience_name=%r lookup_names=%r → audience_id=%s full_name=%r",
         source_account_id,
         audience_name,
+        lookup_names,
         chosen.get("audience_id"),
         chosen.get("name"),
     )

examples/auto_put_ad_mini/tools/creative_creation.py

@@ -57,6 +57,13 @@ def _landing_category_values(v: LandingVideo) -> set[str]:
 
 def _is_landing_candidate(v: LandingVideo) -> bool:
     """承接视频基础预筛:语义字段齐全,且内容品类不在黑名单。"""
+    has_hot_element_features = bool(v.raw.get("element_features"))
+    if v.demand_type != "聚类特征点" and not has_hot_element_features:
+        logger.info(
+            "[creative_creation] landing=%d demandType=%r 非聚类特征点,跳过",
+            v.video_id, v.demand_type,
+        )
+        return False
     if not (v.point_type and v.standard_element):
         return False
     excluded = _landing_category_values(v) & LANDING_EXCLUDED_CATEGORIES

examples/auto_put_ad_mini/tools/creative_review.py

@@ -0,0 +1,577 @@
+"""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_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:
+    upper = (status or "").upper()
+    return "REJECT" in upper or "DENIED" in upper
+
+
+def _status_is_pending(status: str) -> bool:
+    upper = (status or "").upper()
+    return "PENDING" in upper or "REVIEWING" in upper
+
+
+def _status_is_approved(status: str) -> bool:
+    upper = (status or "").upper()
+    return "PASS" in upper or "APPROVED" in upper or "NORMAL" in upper
+
+
+def _component_type(detail: dict) -> str:
+    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:
+    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]:
+    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]:
+    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:
+    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

examples/auto_put_ad_mini/tools/im_approval_ad_creation.py

@@ -25,6 +25,7 @@ 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,
@@ -282,10 +283,9 @@ def run_ad_approval_workflow(
     timeout_minutes: int = CREATION_APPROVAL_TIMEOUT_MINUTES,
 ) -> tuple[dict, dict[int, str]]:
     """完整模块 A 审批工作流。"""
-    from datetime import datetime, timezone
-
-    date_str = datetime.now(timezone.utc).strftime("%Y%m%d")
-    ts = datetime.now(timezone.utc).strftime("%H%M%S")
+    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)

examples/auto_put_ad_mini/tools/im_approval_creation.py

@@ -30,6 +30,7 @@ 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,
@@ -359,10 +360,9 @@ def run_approval_workflow(
           sheet_meta:{url, sheet_token, sheet_id, im_sent, im_message_id}
           actions:{row_idx: action},row_idx 1-based 对应 records[row_idx-1]
     """
-    from datetime import datetime, timezone
-
-    date_str = datetime.now(timezone.utc).strftime("%Y%m%d")
-    ts = datetime.now(timezone.utc).strftime("%H%M%S")
+    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)

examples/auto_put_ad_mini/tools/material_recall.py

@@ -75,6 +75,11 @@ POINT_TYPE_TO_VIDEO_CONFIG = {
     "关键点": "VIDEO_KEYPOINT",
     "目的点": "VIDEO_PURPOSE",
 }
+SUBSTANCE_TO_VIDEO_CONFIG = {
+    "INSPIRATION_SUBSTANCE": "VIDEO_INSPIRATION",
+    "KEYPOINT_SUBSTANCE": "VIDEO_KEYPOINT",
+    "PURPOSE_SUBSTANCE": "VIDEO_PURPOSE",
+}
 
 # fallback(动态接口失败时用)
 MATERIAL_EFFECTIVE_CONFIG_CODES = ["VIDEO_TOPIC"] + list(POINT_TYPE_TO_SUBSTANCE.values())
@@ -195,10 +200,27 @@ def _pick_query_strategies_for_batch(landing: LandingVideo) -> List[tuple]:
     prepare 阶段:维度 1 失败 → 试维度 2 → 仍失败 → 换 landing。
     """
     strategies = []
+    seen = set()
+    for feature in landing.raw.get("element_features") or []:
+        if not isinstance(feature, dict):
+            continue
+        standard_element = (feature.get("standard_element") or "").strip()
+        point_type = (feature.get("point_type") or "").strip()
+        if standard_element in PLACEHOLDER_VALUES:
+            continue
+        cc = POINT_TYPE_TO_SUBSTANCE.get(point_type)
+        if not cc:
+            continue
+        key = (standard_element, cc)
+        if key in seen:
+            continue
+        seen.add(key)
+        strategies.append((standard_element, cc, f"标准化元素-{point_type}"))
+
     # 维度 1: 标准化元素(优先)
     if landing.standard_element and landing.standard_element not in PLACEHOLDER_VALUES:
         cc = POINT_TYPE_TO_SUBSTANCE.get(landing.point_type)
-        if cc:
+        if cc and (landing.standard_element, cc) not in seen:
             strategies.append((landing.standard_element, cc, f"标准化元素-{landing.point_type}"))
     # 维度 2: 选题
     if landing.demand_content_topic and landing.demand_content_topic not in PLACEHOLDER_VALUES:
@@ -213,7 +235,7 @@ def _fallback_config_codes_for_strategy(config_code: str, landing: LandingVideo)
     VIDEO_KEYPOINT/VIDEO_INSPIRATION 等历史向量字段命中素材。
     """
     out = [config_code]
-    video_cc = POINT_TYPE_TO_VIDEO_CONFIG.get(landing.point_type)
+    video_cc = SUBSTANCE_TO_VIDEO_CONFIG.get(config_code) or POINT_TYPE_TO_VIDEO_CONFIG.get(landing.point_type)
     if video_cc and video_cc not in out:
         out.append(video_cc)
     if config_code != "VIDEO_TOPIC" and "VIDEO_TOPIC" not in out:
@@ -420,11 +442,14 @@ def recall_materials_for_video(
             fmats, fn_bl, fn_low_imp, fn_low_ctr = _items_to_materials(
                 fallback_items, RECALL_MIN_IMPRESSIONS, RECALL_MIN_CTR,
             )
+            fn_low_score = sum(1 for m in fmats if (m.score or 0.0) < RECALL_SIM_THRESHOLD)
+            fmats = [m for m in fmats if (m.score or 0.0) >= RECALL_SIM_THRESHOLD]
             logger.info(
-                "[material_recall]     fallback=%s 返回 %d 条,⊘ 黑名单 %d,⊘ imp<=%d %d,⊘ ctr<%.2f %d → 保留 %d",
+                "[material_recall]     fallback=%s 返回 %d 条,⊘ 黑名单 %d,⊘ imp<=%d %d,⊘ ctr<%.2f %d,⊘ score<%.2f %d → 保留 %d",
                 fallback_cc, len(fallback_items), fn_bl,
                 RECALL_MIN_IMPRESSIONS, fn_low_imp,
                 RECALL_MIN_CTR, fn_low_ctr,
+                RECALL_SIM_THRESHOLD, fn_low_score,
                 len(fmats),
             )
             if fmats:

examples/auto_put_ad_mini/tools/video_feature_query.py

@@ -0,0 +1,312 @@
+"""ODPS video feature lookup for hot landing-video fallback.
+
+Hot videos from videoContentList may not include pointType / standardElement.
+This module enriches them from loghubods.dwd_video_element_contribution_analysis.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable
+
+logger = logging.getLogger(__name__)
+
+_MINI_DIR = Path(__file__).resolve().parent.parent
+
+
+@dataclass(frozen=True)
+class VideoElementFeature:
+    video_id: int
+    point_type: str
+    standard_element: str
+    contribution_score: float
+    dt: str
+
+
+CREATE_CACHE_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS video_element_feature_cache (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
+    video_id BIGINT NOT NULL COMMENT '业务视频ID，对应 ODPS vid',
+    dt VARCHAR(8) NOT NULL COMMENT 'ODPS 分区日期 YYYYMMDD',
+    point_type VARCHAR(50) NOT NULL COMMENT '点类型',
+    standard_element VARCHAR(255) NOT NULL COMMENT '标准化元素',
+    element_dimension VARCHAR(50) NOT NULL DEFAULT '实质' COMMENT '元素维度',
+    contribution_score DOUBLE NOT NULL DEFAULT 0 COMMENT '贡献分',
+    is_miss BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否为无特征负缓存',
+    raw_element VARCHAR(255) DEFAULT NULL COMMENT '原始元素名称',
+    element_id VARCHAR(100) DEFAULT NULL COMMENT '标准化元素ID',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+    UNIQUE KEY uk_dt_video_point_element (dt, video_id, point_type, standard_element),
+    KEY idx_video_dt (video_id, dt),
+    KEY idx_dt (dt),
+    KEY idx_updated_at (updated_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='视频元素特征缓存'
+"""
+
+
+def _quote_sql_string(value: str) -> str:
+    return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"
+
+
+def _chunks(items: list[int], size: int) -> Iterable[list[int]]:
+    for i in range(0, len(items), size):
+        yield items[i:i + size]
+
+
+def _get_odps_client():
+    import sys
+
+    tools_dir = _MINI_DIR / "tools"
+    if str(tools_dir) not in sys.path:
+        sys.path.insert(0, str(tools_dir))
+    from odps_module import get_odps_client
+
+    return get_odps_client(project="loghubods")
+
+
+def _latest_partition_dt(client) -> str:
+    table_name = "dwd_video_element_contribution_analysis"
+    try:
+        table = client._odps.get_table(table_name)
+        partition = table.get_max_partition()
+        if partition is not None:
+            name = getattr(partition, "name", "") or str(partition)
+            for part in name.split(","):
+                if part.startswith("dt="):
+                    return part.split("=", 1)[1].strip("'\"")
+    except Exception as e:
+        logger.warning("[video_feature] get max partition by metadata failed:%s", e)
+
+    rows = client.query(
+        "SELECT MAX_PT('loghubods.dwd_video_element_contribution_analysis') AS dt",
+        limit=1,
+    )
+    if rows and rows[0].get("dt"):
+        return str(rows[0]["dt"])
+    raise RuntimeError("无法获取 dwd_video_element_contribution_analysis 最新分区")
+
+
+def _query_odps_rows(client, sql: str, columns: list[str]) -> list[dict]:
+    """Read ODPS rows without tunnel when possible.
+
+    The configured tunnel endpoint can be unreachable in some runtime
+    environments. Small feature lookups are better served through the instance
+    reader without tunnel.
+    """
+    odps = getattr(client, "_odps", None)
+    if odps is None:
+        return client.query(sql)
+
+    instance = odps.execute_sql(sql)
+    instance.wait_for_success()
+    rows: list[dict] = []
+    with instance.open_reader(tunnel=False) as reader:
+        for record in reader:
+            rows.append({col: record[i] for i, col in enumerate(columns)})
+    return rows
+
+
+def _ensure_cache_table() -> None:
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(CREATE_CACHE_TABLE_SQL)
+            cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'is_miss'")
+            if not cur.fetchone():
+                cur.execute(
+                    """
+ALTER TABLE video_element_feature_cache
+ADD COLUMN is_miss BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否为无特征负缓存'
+AFTER contribution_score
+"""
+                )
+    finally:
+        conn.close()
+
+
+def _read_cache(video_ids: list[int], dt: str) -> tuple[dict[int, list[VideoElementFeature]], set[int]]:
+    if not video_ids:
+        return {}, set()
+    from db.connection import get_connection
+
+    out: dict[int, list[VideoElementFeature]] = {}
+    cached_ids: set[int] = set()
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            for batch in _chunks(video_ids, 500):
+                placeholders = ",".join(["%s"] * len(batch))
+                cur.execute(
+                    f"""
+SELECT video_id, dt, point_type, standard_element, contribution_score, is_miss
+FROM video_element_feature_cache
+WHERE dt = %s
+AND video_id IN ({placeholders})
+ORDER BY video_id, contribution_score DESC
+""",
+                    [dt, *batch],
+                )
+                for row in cur.fetchall():
+                    video_id = int(row["video_id"])
+                    cached_ids.add(video_id)
+                    if row.get("is_miss"):
+                        continue
+                    out.setdefault(video_id, []).append(VideoElementFeature(
+                        video_id=video_id,
+                        point_type=str(row["point_type"] or ""),
+                        standard_element=str(row["standard_element"] or ""),
+                        contribution_score=float(row["contribution_score"] or 0.0),
+                        dt=str(row["dt"] or ""),
+                    ))
+    finally:
+        conn.close()
+    return out, cached_ids
+
+
+def _write_cache(
+    features: dict[int, list[VideoElementFeature]],
+    miss_video_ids: Iterable[int],
+    dt: str,
+) -> None:
+    rows = [
+        (
+            feature.video_id,
+            feature.dt,
+            feature.point_type,
+            feature.standard_element,
+            "实质",
+            feature.contribution_score,
+            False,
+        )
+        for feature_list in features.values()
+        for feature in feature_list
+    ]
+    miss_rows = [
+        (int(video_id), dt, "", "", "实质", 0.0, True)
+        for video_id in miss_video_ids
+    ]
+    rows.extend(miss_rows)
+    if not rows:
+        return
+
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.executemany(
+                """
+INSERT INTO video_element_feature_cache
+    (video_id, dt, point_type, standard_element, element_dimension, contribution_score, is_miss)
+VALUES (%s, %s, %s, %s, %s, %s, %s)
+ON DUPLICATE KEY UPDATE
+    contribution_score = VALUES(contribution_score),
+    element_dimension = VALUES(element_dimension),
+    is_miss = VALUES(is_miss),
+    updated_at = CURRENT_TIMESTAMP
+""",
+                rows,
+            )
+    finally:
+        conn.close()
+
+
+def fetch_video_element_features(
+    video_ids: Iterable[int],
+    chunk_size: int = 100,
+) -> dict[int, list[VideoElementFeature]]:
+    """Fetch point type and standard element rows per video id from latest ODPS partition.
+
+    One video can have multiple usable element rows. Keep all rows where:
+    - 元素维度 = 实质
+    - 贡献分 >= 0.8
+    - 点类型 and 标准化元素 are non-empty
+    """
+    ids: list[int] = []
+    seen: set[int] = set()
+    for raw in video_ids:
+        try:
+            vid = int(raw)
+        except (TypeError, ValueError):
+            continue
+        if vid in seen:
+            continue
+        seen.add(vid)
+        ids.append(vid)
+
+    if not ids:
+        return {}
+
+    client = _get_odps_client()
+    if client is None:
+        logger.warning("[video_feature] ODPS client unavailable, skip enrichment")
+        return {}
+
+    dt = _latest_partition_dt(client)
+    _ensure_cache_table()
+    out, cached_ids = _read_cache(ids, dt)
+    missing_ids = [video_id for video_id in ids if video_id not in cached_ids]
+    if not missing_ids:
+        logger.info(
+            "[video_feature] cache hit %d/%d videos dt=%s",
+            len(cached_ids), len(ids), dt,
+        )
+        return out
+
+    fetched: dict[int, list[VideoElementFeature]] = {}
+    for batch in _chunks(missing_ids, max(1, chunk_size)):
+        vid_list = ",".join(_quote_sql_string(str(v)) for v in batch)
+        sql = f"""
+SELECT  vid
+        ,`点类型` AS point_type
+        ,`标准化元素` AS standard_element
+        ,`贡献分` AS contribution_score
+        ,dt
+FROM    loghubods.dwd_video_element_contribution_analysis
+WHERE   dt = {_quote_sql_string(dt)}
+AND     vid IN ({vid_list})
+AND     `元素维度` = '实质'
+AND     `贡献分` >= 0.8
+AND     `点类型` IS NOT NULL
+AND     `点类型` <> ''
+AND     `标准化元素` IS NOT NULL
+AND     `标准化元素` <> ''
+ORDER BY vid, `贡献分` DESC
+"""
+        rows = _query_odps_rows(
+            client,
+            sql,
+            ["vid", "point_type", "standard_element", "contribution_score", "dt"],
+        )
+        for row in rows:
+            try:
+                video_id = int(row.get("vid"))
+            except (TypeError, ValueError):
+                continue
+            point_type = str(row.get("point_type") or "").strip()
+            standard_element = str(row.get("standard_element") or "").strip()
+            if not point_type or not standard_element:
+                continue
+            fetched.setdefault(video_id, []).append(VideoElementFeature(
+                video_id=video_id,
+                point_type=point_type,
+                standard_element=standard_element,
+                contribution_score=float(row.get("contribution_score") or 0.0),
+                dt=str(row.get("dt") or ""),
+            ))
+
+    fetched_ids = set(fetched.keys())
+    miss_ids = [video_id for video_id in missing_ids if video_id not in fetched_ids]
+    _write_cache(fetched, miss_ids, dt)
+    out.update(fetched)
+    logger.info(
+        "[video_feature] dt=%s cache_hit=%d odps_hit=%d odps_miss=%d/%d element_rows=%d",
+        dt, len(cached_ids), len(fetched), len(miss_ids), len(missing_ids),
+        sum(len(v) for v in out.values()),
+    )
+    return out

examples/auto_put_ad_mini/tools/video_recall.py

@@ -56,6 +56,43 @@ def _normalize_category(raw) -> str:
     return str(raw).strip()
 
 
+def _enrich_hot_video_elements(videos: List["LandingVideo"]) -> None:
+    """Fill pointType / standardElement for hot videos from ODPS latest partition."""
+    if not videos:
+        return
+    try:
+        from tools.video_feature_query import fetch_video_element_features
+    except Exception as e:
+        logger.warning("[video_recall] hot 元素补全模块导入失败,跳过:%s", e)
+        return
+
+    features_by_vid = fetch_video_element_features(v.video_id for v in videos)
+    enriched = 0
+    for v in videos:
+        features = features_by_vid.get(v.video_id) or []
+        if not features:
+            continue
+        feature_payload = [
+            {
+                "point_type": f.point_type,
+                "standard_element": f.standard_element,
+                "contribution_score": f.contribution_score,
+                "dt": f.dt,
+            }
+            for f in features
+        ]
+        v.raw["element_features"] = feature_payload
+        # Keep the top contribution element on first-class fields so existing
+        # candidate checks continue to work.
+        v.point_type = v.point_type or features[0].point_type
+        v.standard_element = v.standard_element or features[0].standard_element
+        enriched += 1
+    logger.info(
+        "[video_recall] hot 元素补全完成 videos=%d/%d element_rows=%d",
+        enriched, len(videos), sum(len(v) for v in features_by_vid.values()),
+    )
+
+
 @dataclass
 class LandingVideo:
     """承接视频(landing video)— 用户进小程序后看到的视频。
@@ -173,6 +210,8 @@ def fetch_landing_videos(
         "[video_recall] 返回 %d 条承接视频(总池 %d / 当前页 %d)",
         len(videos), payload.get("totalSize") or 0, payload.get("currentPage") or 1,
     )
+    if source == PIAOQUANTV_HOT_FALLBACK_SOURCE:
+        _enrich_hot_video_elements(videos)
     return videos