完善小程序自动投放创意召回流程

AGENTS.md

@@ -26,19 +26,37 @@
 ## 承接视频安全和质量
 
 - 创建落地计划或腾讯创意前,必须先做视频风险审核。
-- 默认风险阈值是 1。风险等级 2-10 应拦截。
+- 默认风险阈值是 5。风险等级 6-10 应拦截。
 - 风险接口异常时要保守处理,不能创建落地计划。
 - 使用 `videoContentList.category` 做内容品类过滤。
-- 默认过滤品类是 `早中晚好`、`祝福音乐`。
+- 默认过滤品类是 `早中晚好`、`祝福音乐`、`历史名人`。
 - 品类过滤必须发生在风险审核、素材召回和 `xcx/save` 之前。
 - 不要把 `categoryName` 当作内容品类使用;它是召回逻辑使用的语义/视觉特征字段。
 
 ## 素材召回规则
 
-- 除非用户明确要求,不要随意降低素材质量门槛。
-- 当前素材门槛:`impressions > 2000`、`ctr >= 0.10`、统计窗口 180 天。
-- 视频可以通过风险审核,但仍然因为没有素材满足质量门槛而不能被选中。
-- 不能为了凑满创意数量而静默降低素材质量阈值。
+- `videoContentList` 只负责选择承接视频,素材召回特征必须通过 `video_id` 查询 ODPS 表 `loghubods.dwd_video_element_contribution_analysis`。
+- primary 和 hot 视频都必须走同一套 ODPS 特征查询和素材召回逻辑。
+- 当前只使用 ODPS 特征中的 `解构选题` 列和 `实质` 维度;`形式`、`意图` 不参与素材召回。
+- 当前召回映射:
+  - `解构选题` -> `VIDEO_TOPIC`
+  - `实质 + 灵感点` -> `INSPIRATION_SUBSTANCE`
+  - `实质 + 关键点` -> `KEYPOINT_SUBSTANCE`
+  - `实质 + 目的点` -> `PURPOSE_SUBSTANCE`
+- 同一视频的多个召回 query 可以并行调用 `batchByText`,再按 `material_id` 合并去重。
+- 当前素材硬筛只看相似度:`score >= 0.8`。
+- 曝光、CTR、ROI 只作为审批展示和兜底排序参考,不作为硬筛。
+- 素材通过相似度筛选后,默认按历史消耗 `cost` 倒序选择。
+- 落地页视频不进历史排重库;当前运行内同一广告的同一 `landing_video_id` 默认最多进入本轮审批候选 2 次。
+- 素材需要跨账户/跨天排重,默认按同人群包下近期使用过的 `material_id` 排除。
+- 当前运行内的审批候选需要做轻量展示去重:同一 `crowd_package + material_id` 只进入本轮审批表一次;该去重只存在内存,不写入历史排重库。
+
+## 视频召回人群包映射
+
+- 腾讯投放、人群包授权、落地计划仍使用账户配置的人群包。
+- 内容服务 `videoContentList` 的 `crowdPackage` 可以有独立映射。
+- 当前默认映射:`cell*year*商业` 获取视频时映射为 `wx*商业`。
+- `source=hot` 只能改变内容服务的视频来源,不能改变映射后的召回人群包。
 
 ## 腾讯人群包和资产规则
 
@@ -62,4 +80,3 @@
 - 端到端执行前,必须意识到可能产生外部副作用:腾讯广告/创意创建、图片上传、DataNexus 监测链接创建、人群包授权、`xcx/save` 落地计划创建。
 - 端到端流程中断时,已经完成的外部副作用不会自动回滚。
 - 行为、环境变量或运营流程变化时,要同步更新面向生产的文档。
-

examples/auto_put_ad_mini/PRODUCTION_AUTOMATION.md

@@ -63,7 +63,15 @@ hot source: 最多 100 条
 3. hot 池同样最多 100 条
 4. hot 池同样走风险审核、品类过滤、素材质量过滤
 
-注意: `source=hot` 只改变内容服务的视频来源,`crowdPackage` 仍然传当前账户配置的人群包,不会固定成其他人群包。
+注意: `source=hot` 只改变内容服务的视频来源。默认情况下 `crowdPackage` 使用当前账户配置的人群包;如果配置了视频召回映射,则 primary/hot 都使用映射后的召回人群包。
+
+当前默认视频召回映射:
+
+```text
+cell*year*商业 -> wx*商业
+```
+
+该映射只影响 `videoContentList` 获取视频,不影响腾讯投放定向、人群包授权和 `xcx/save` 落地计划。
 
 ## 视频过滤规则
 
@@ -72,7 +80,7 @@ hot source: 最多 100 条
 默认过滤:
 
 ```text
-LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐
+LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐,历史名人
 ```
 
 命中后,该视频会在以下步骤前被跳过:
@@ -84,7 +92,7 @@ LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐
 可通过 Docker/K8s 环境变量覆盖:
 
 ```bash
-LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐
+LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐,历史名人
 ```
 
 多个品类用英文逗号分隔。
@@ -100,32 +108,61 @@ POST https://longvideoapi.piaoquantv.com/longvideoapi/openapi/video/getVideoTagI
 当前阈值:
 
 ```text
-VIDEO_RISK_MAX_ALLOWED_LEVEL=1
+VIDEO_RISK_MAX_ALLOWED_LEVEL=5
 ```
 
 规则:
 
-- 风险等级 0/1 通过
-- 风险等级 2-10 拦截
+- 风险等级 0-5 通过
+- 风险等级 6-10 拦截
 - 风险接口异常时按不通过处理,避免风险未知的视频进入投放
 
-## 素材召回质量门槛
+## 素材召回筛选与排序
+
+`videoContentList` 只负责提供承接视频候选。素材召回特征统一通过 `video_id`
+查询 ODPS 表:
+
+```text
+loghubods.dwd_video_element_contribution_analysis
+```
+
+当前使用的召回特征:
 
-素材召回仍使用现有质量阈值:
+- `解构选题` 列 -> `VIDEO_TOPIC`
+- `元素维度=实质, 点类型=灵感点` -> `INSPIRATION_SUBSTANCE`
+- `元素维度=实质, 点类型=关键点` -> `KEYPOINT_SUBSTANCE`
+- `元素维度=实质, 点类型=目的点` -> `PURPOSE_SUBSTANCE`
+
+`形式` 和 `意图` 不参与素材召回。primary 和 hot 视频都走同一套 ODPS
+查询、风险审核、素材召回和审批流程。
+
+素材召回只使用相似度做硬筛:
 
 ```text
-RECALL_MIN_IMPRESSIONS=2000
-RECALL_MIN_CTR=0.10
+RECALL_SIM_THRESHOLD=0.8
 RECALL_DAYS=180
 RECALL_DISPLAY_K=30
 ```
 
-素材必须同时满足:
+素材必须满足:
 
-- 曝光 `impressions > 2000`
-- CTR `>= 10%`
+- 相似度 `score >= 0.8`
 - 封面 URL 不命中已知黑名单
 
+通过筛选后按历史消耗 `cost` 倒序选择。ROI、CTR、曝光数和相似度会进入创意审批报表,但不再作为硬筛。
+
+排重分两层:
+
+- 历史排重只读取已有明确结果的素材使用记录,按同人群包下的 `material_id` 排除。
+- 本轮审批候选做内存展示去重,同一 `crowd_package + material_id` 只进入当前审批表一次。该规则不写入历史排重库,进程结束即失效。
+- 落地页视频不进入历史排重库,但同一广告在本轮准备中同一个 `landing_video_id` 默认最多出现 2 次。
+
+可通过环境变量调整同广告落地页视频本轮限频:
+
+```bash
+MAX_SAME_LANDING_PER_AD_IN_RUN=2
+```
+
 ## 热门兜底配置
 
 热门兜底由以下环境变量控制:
@@ -154,8 +191,10 @@ PIAOQUANTV_VIDEO_TYPE=5
 PIAOQUANTV_VIDEO_SOURCE=prior
 PIAOQUANTV_HOT_FALLBACK_ENABLED=true
 PIAOQUANTV_HOT_FALLBACK_SOURCE=hot
-LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐
-VIDEO_RISK_MAX_ALLOWED_LEVEL=1
+LANDING_EXCLUDED_CATEGORIES=早中晚好,祝福音乐,历史名人
+VIDEO_RISK_MAX_ALLOWED_LEVEL=5
+VIDEO_RECALL_CROWD_PACKAGE_MAP={"cell*year*商业":"wx*商业"}
+MAX_SAME_LANDING_PER_AD_IN_RUN=2
 TENCENT_AUDIENCE_SOURCE_ACCOUNT_ID=55615440
 TENCENT_AUDIENCE_GRANT_BUSINESS_ID=12312
 ```
@@ -164,7 +203,7 @@ TENCENT_AUDIENCE_GRANT_BUSINESS_ID=12312
 
 - `R330` 账户如果源账户无法解析到 ONLINE/SUCCESS 人群包,Phase 0 会跳过该账户。
 - `泛人群` 主池可能返回 0,需要依赖 `source=hot` 兜底。
-- 当前素材质量门槛较高,部分安全视频会因素材曝光/CTR 不达标而无法产出创意。
+- 当前素材只按相似度准入并按消耗排序;低曝光/低 CTR 素材可能进入审批表,需要运营结合报表判断。
 - 每次 pending creative 准备会调用 `xcx/save` 生成落地计划;端到端测试中断不会自动清理已生成的落地计划。
 
 ## 上线检查
@@ -178,4 +217,3 @@ TENCENT_AUDIENCE_GRANT_BUSINESS_ID=12312
 - 品牌图和监测链接模板可自动初始化
 - Docker/K8s 环境变量与本文件一致
 - 手动跑一次 `execute_creation_once.py` 能生成飞书审批表
-

examples/auto_put_ad_mini/config.py

@@ -925,7 +925,7 @@ MAX_MATERIAL_PER_LANDING = 10
 # 内容服务返回的内容品类黑名单。为空则不过滤;多个品类用英文逗号分隔。
 LANDING_EXCLUDED_CATEGORIES = {
     v.strip()
-    for v in os.getenv("LANDING_EXCLUDED_CATEGORIES", "早中晚好,祝福音乐").split(",")
+    for v in os.getenv("LANDING_EXCLUDED_CATEGORIES", "早中晚好,祝福音乐,历史名人").split(",")
     if v.strip()
 }
 
@@ -949,27 +949,29 @@ VIDEO_RISK_TAG_LEVELS = {
     "85869": 9,
     "85870": 10,
 }
-# 默认仅允许 0/1,拦截 2-10。可通过环境变量临时调整。
-VIDEO_RISK_MAX_ALLOWED_LEVEL = int(os.getenv("VIDEO_RISK_MAX_ALLOWED_LEVEL", "1"))
+# 默认允许 0-5,拦截 6-10。可通过环境变量临时调整。
+VIDEO_RISK_MAX_ALLOWED_LEVEL = int(os.getenv("VIDEO_RISK_MAX_ALLOWED_LEVEL", "5"))
 VIDEO_RISK_API_TIMEOUT_SECONDS = int(os.getenv("VIDEO_RISK_API_TIMEOUT_SECONDS", "10"))
 
 # --- 素材召回质量过滤(2026-06-10 用户确认,batchByText 升级)---
-# 服务端 ranking 参数:simThreshold=0.8(语义相关),alpha=0(不要 sim 加权)
-# wCtr=1, wCvr=wRoi=wOpenRate=wFissionRate=0 → qualityScore 由 ctr 主导 → 等价"取 CTR 最高"
-# 客户端 impressions > 阈值 二次筛 → 保证"语义相关 + 曝光足够 + CTR 最高"
+# 服务端 ranking 参数只保留 simThreshold=0.8(语义相关),其余加权全部置 0。
+# 客户端只用 score >= simThreshold 做硬筛,再按历史消耗 cost 倒序排序。
 RECALL_SIM_THRESHOLD = 0.8
 RECALL_ALPHA = 0
-RECALL_W_CTR = 1
+RECALL_W_CTR = 0
 RECALL_W_CVR = 0
 RECALL_W_ROI = 0
 RECALL_W_OPEN_RATE = 0
 RECALL_W_FISSION_RATE = 0
-RECALL_DECONSTRUCT_BOOST = 0.4
+RECALL_DECONSTRUCT_BOOST = 0
 RECALL_DAYS = 180                                 # 投放统计天数(2026-06-10 修正:30→180,跟 admin 后台一致,冷门素材需长周期)
 RECALL_DISPLAY_K = 30                             # 服务端展示条数
-RECALL_MIN_IMPRESSIONS = 2000                     # 客户端 impressions 过滤阈值(2026-06-10 用户:曝光 > 2000)
-RECALL_MIN_CTR = 0.10                             # 客户端 CTR 阈值(2026-06-10 用户:CTR ≥ 10%)
+RECALL_PARALLEL_MAX_WORKERS = int(os.getenv("RECALL_PARALLEL_MAX_WORKERS", "4"))
+RECALL_QUERY_LIMIT_PER_VIDEO = int(os.getenv("RECALL_QUERY_LIMIT_PER_VIDEO", "12"))
+RECALL_MIN_IMPRESSIONS = 0                        # 保留兼容配置;当前不作为硬筛。
+RECALL_MIN_CTR = 0.0                              # 保留兼容配置;当前不作为硬筛。
 RECALL_SOURCE_LABELS = ["内部素材"]               # 只要内部
+MAX_SAME_LANDING_PER_AD_IN_RUN = int(os.getenv("MAX_SAME_LANDING_PER_AD_IN_RUN", "2"))
 RECALL_CONFIG_CODES_FULL = [
     "VIDEO_TOPIC", "VIDEO_INSPIRATION", "VIDEO_PURPOSE",
     "VIDEO_KEYPOINT", "VIDEO_TITLE",

examples/auto_put_ad_mini/db/schema.sql

@@ -261,6 +261,30 @@ CREATE TABLE IF NOT EXISTS video_element_feature_cache (
     KEY idx_updated_at (updated_at)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='视频元素特征缓存';
 
+-- =====================================================
+-- 12. 创意素材使用历史/占位表
+-- =====================================================
+CREATE TABLE IF NOT EXISTS creative_material_usage (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
+    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
+    adgroup_id BIGINT DEFAULT NULL COMMENT '广告ID',
+    crowd_package VARCHAR(200) NOT NULL COMMENT '投放人群包名称',
+    landing_video_id BIGINT DEFAULT NULL COMMENT '承接视频ID,仅审计不默认排重',
+    material_id VARCHAR(100) NOT NULL COMMENT '内部素材ID',
+    material_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
+    dynamic_creative_id BIGINT DEFAULT NULL COMMENT '腾讯动态创意ID',
+    status VARCHAR(50) NOT NULL DEFAULT 'prepared' COMMENT 'prepared/submitted/failed',
+    source VARCHAR(50) DEFAULT NULL COMMENT 'primary/hot',
+    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 '更新时间',
+
+    KEY idx_crowd_material_created (crowd_package, material_id, created_at),
+    KEY idx_crowd_account_ad_material (crowd_package, account_id, adgroup_id, material_id),
+    KEY idx_account_created (account_id, created_at),
+    KEY idx_status_created (status, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='创意素材使用历史/占位';
+
 -- =====================================================
 -- 初始化数据
 -- =====================================================

examples/auto_put_ad_mini/execute_creation_apply.py

@@ -13,6 +13,7 @@
 """
 
 import asyncio
+import argparse
 import json
 import logging
 import sys
@@ -25,7 +26,11 @@ sys.path.insert(0, str(_HERE))
 from dotenv import load_dotenv  # noqa: E402
 load_dotenv(_HERE / ".env")
 
-from config import FEISHU_OPERATOR_CHAT_ID, now_in_timezone  # noqa: E402
+from config import (  # noqa: E402
+    CREATION_APPROVAL_TIMEOUT_MINUTES,
+    FEISHU_OPERATOR_CHAT_ID,
+    now_in_timezone,
+)
 
 # 强占 sys.path[0],绕过 config import 副作用(im-client/tools.py 同名冲突)
 while str(_HERE) in sys.path:
@@ -37,6 +42,7 @@ from tools.creative_review import (  # noqa: E402
     mark_creation_submit_failed,
     record_creation_submission,
 )
+from tools.creative_material_usage import update_material_usage_status  # noqa: E402
 
 logger = logging.getLogger("execute_creation_apply")
 
@@ -181,6 +187,13 @@ def apply_pending_records(records: list[dict]) -> dict:
         rec["action"] = action
 
         if action != "approve":
+            try:
+                update_material_usage_status(rec, action)
+            except Exception as e:
+                logger.warning(
+                    "[apply] 更新素材 usage 状态失败 action=%s material=%s: %s",
+                    action, rec.get("_material_id"), e,
+                )
             out_records.append(rec)
             continue
 
@@ -189,6 +202,10 @@ def apply_pending_records(records: list[dict]) -> dict:
         if not body:
             rec["error"] = "missing _request_body"
             posted_failed += 1
+            try:
+                update_material_usage_status(rec, "post_failed", error=rec["error"])
+            except Exception as e:
+                logger.warning("[apply] 更新素材 usage 状态失败: %s", e)
             out_records.append(rec)
             continue
 
@@ -200,6 +217,10 @@ def apply_pending_records(records: list[dict]) -> dict:
         if cid:
             rec["dynamic_creative_id"] = str(cid)
             posted_ok += 1
+            try:
+                update_material_usage_status(rec, "posted_ok", dynamic_creative_id=cid)
+            except Exception as e:
+                logger.warning("[apply] 更新素材 usage 状态失败 cid=%s: %s", cid, e)
             try:
                 record_creation_submission(rec, int(cid))
             except Exception as e:
@@ -208,6 +229,10 @@ def apply_pending_records(records: list[dict]) -> dict:
             rec["error"] = "post_failed"
             posted_failed += 1
             mark_creation_submit_failed(rec, "post_failed")
+            try:
+                update_material_usage_status(rec, "post_failed", error=rec["error"])
+            except Exception as e:
+                logger.warning("[apply] 更新素材 usage 状态失败: %s", e)
         out_records.append(rec)
 
     run_finished = now_in_timezone().isoformat()
@@ -271,11 +296,21 @@ def main() -> int:
     except Exception as e:
         logger.warning("[sls] 挂载异常(降级为本地 only):%s", e)
 
-    if len(sys.argv) < 2:
-        logger.error("用法: python execute_creation_apply.py <pending_records.json>")
-        return 1
+    parser = argparse.ArgumentParser(
+        description="从 pending records 执行创意创建;可选择读取已有飞书审批表决策。",
+    )
+    parser.add_argument("pending_records_json")
+    parser.add_argument("--sheet-token", default="", help="已有创意审批飞书表 token")
+    parser.add_argument("--sheet-id", default="", help="已有创意审批飞书表 sheet_id")
+    parser.add_argument(
+        "--approval-timeout-minutes",
+        type=int,
+        default=CREATION_APPROVAL_TIMEOUT_MINUTES,
+        help="读取已有审批表时等待决策的分钟数",
+    )
+    args = parser.parse_args()
 
-    pending_path = Path(sys.argv[1])
+    pending_path = Path(args.pending_records_json)
     if not pending_path.exists():
         logger.error(f"文件不存在: {pending_path}")
         return 1
@@ -283,6 +318,23 @@ def main() -> int:
     with open(pending_path, encoding="utf-8") as f:
         records = json.load(f)
 
+    if args.sheet_token and args.sheet_id:
+        from tools.im_approval_creation import poll_approval_actions
+
+        logger.info(
+            "读取已有审批表决策 sheet_token=%s sheet_id=%s records=%d",
+            args.sheet_token, args.sheet_id, len(records),
+        )
+        actions = poll_approval_actions(
+            sheet_token=args.sheet_token,
+            sheet_id=args.sheet_id,
+            expected_row_count=len(records),
+            timeout_minutes=args.approval_timeout_minutes,
+        )
+        for i, rec in enumerate(records, start=1):
+            rec["action"] = actions.get(i, "skip")
+        logger.info("审批决策读取完成: %d/%d", len(actions), len(records))
+
     logger.info(f"读到 {len(records)} 条 pending records,开始 Phase 3 执行")
     summary = apply_pending_records(records)
     out_path = write_summary(summary, _HERE / "outputs" / "data")

examples/auto_put_ad_mini/execute_creation_once.py

@@ -38,6 +38,7 @@ from config import (  # noqa: E402
     ADS_PER_ACCOUNT,
     CREATION_APPROVAL_REQUIRED,
     CREATION_APPROVAL_TIMEOUT_MINUTES,
+    MAX_SAME_LANDING_PER_AD_IN_RUN,
     TARGET_CREATIVES_PER_AD,
     WHITELIST_ACCOUNTS,
     get_creation_account_ids,
@@ -62,6 +63,8 @@ from tools.creative_creation import (  # noqa: E402
     load_excluded_ad_ids_from_adjustment,
     prepare_one_creative_for_ad,
 )
+from tools.creative_material_usage import record_prepared_material_usage  # noqa: E402
+from tools.video_recall import get_account_crowd_package  # noqa: E402
 
 from execute_creation_apply import (  # noqa: E402
     apply_pending_records,
@@ -374,11 +377,12 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
     logger.info("[phase1] 关联点过滤集合 size=%d", len(excluded_ad_ids))
 
     pending_records: list[dict] = []
-    # 全局 used 集合(2026-06-10 P0-NEW-3 修复:跨账户 landing/material 也互斥)
-    # 起因:实测 70073686 跨"泛人群"+"回流330以上人群"两个 crowd_package 池子重复,
-    # account 级 set 漏 — 提到 run_once 一份。
-    used_material_ids_global: set = set()
-    used_landing_ids_global: set = set()
+    # 素材排重只读取已有明确结果的 DB 记录。
+    # pending/prepared 没有审批或投放结果,不进入排重。
+    # 但同一轮审批候选需要做展示去重,避免表格里同 crowd_package 反复出现同一素材。
+    # 这个 set 只存在内存里,不写历史排重库;进程结束即失效。
+    display_used_material_ids_by_crowd: dict[str, set[str]] = {}
+    # landing_video 不进历史排重库,只在同一广告本轮候选内做限频。
 
     for account_id in creation_accounts:
         logger.info("=" * 60)
@@ -402,21 +406,39 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
             logger.info("[phase1] account=%d 无广告需补创意", account_id)
             continue
 
+        try:
+            crowd_package = get_account_crowd_package(account_id)
+        except Exception as e:
+            logger.exception(
+                "[phase1] account=%d 读取 crowd_package 失败,跳过:%s",
+                account_id, e,
+            )
+            continue
+        display_excluded_material_ids = display_used_material_ids_by_crowd.setdefault(
+            crowd_package, set(),
+        )
+
         for ad in ads_after_filter:
             adgroup_id = ad["adgroup_id"]
             already_have = ad["creative_count"]
             to_add = max(0, target_creatives - already_have)
+            landing_counts_for_ad: dict[int, int] = {}
             logger.info(
                 "[phase1]   adgroup=%d(have=%d need=%d)",
                 adgroup_id, already_have, to_add,
             )
 
             for _ in range(to_add):
+                landing_excluded_for_ad = {
+                    vid
+                    for vid, count in landing_counts_for_ad.items()
+                    if count >= MAX_SAME_LANDING_PER_AD_IN_RUN
+                }
                 try:
                     rec = prepare_one_creative_for_ad(
                         account_id, adgroup_id,
-                        excluded_material_ids=used_material_ids_global,
-                        excluded_landing_ids=used_landing_ids_global,
+                        excluded_material_ids=display_excluded_material_ids,
+                        excluded_landing_ids=landing_excluded_for_ad,
                     )
                 except Exception as e:
                     logger.exception(
@@ -426,8 +448,33 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
 
                 if rec:
                     pending_records.append(rec)
-                    used_material_ids_global.add(rec["_material_id"])
-                    used_landing_ids_global.add(rec["landing_video_id"])
+                    material_id = rec.get("_material_id")
+                    if material_id:
+                        display_excluded_material_ids.add(str(material_id))
+                        logger.info(
+                            "[phase1] 本轮展示去重登记 crowd=%r material=%s size=%d",
+                            crowd_package, material_id,
+                            len(display_excluded_material_ids),
+                        )
+                    landing_video_id = rec.get("landing_video_id")
+                    if landing_video_id is not None:
+                        landing_video_id = int(landing_video_id)
+                        landing_counts_for_ad[landing_video_id] = (
+                            landing_counts_for_ad.get(landing_video_id, 0) + 1
+                        )
+                        logger.info(
+                            "[phase1] 本轮同广告 landing 计数 adgroup=%d landing=%d count=%d limit=%d",
+                            adgroup_id, landing_video_id,
+                            landing_counts_for_ad[landing_video_id],
+                            MAX_SAME_LANDING_PER_AD_IN_RUN,
+                        )
+                    try:
+                        record_prepared_material_usage(rec)
+                    except Exception as e:
+                        logger.warning(
+                            "[phase1] material usage 记录失败 account=%d adgroup=%d material=%s:%s",
+                            account_id, adgroup_id, rec.get("_material_id"), e,
+                        )
                 else:
                     # 2026-06-10 用户要求:单条 prepare 失败 → continue 不 break
                     # 同广告剩余 to_add 创意还能继续试,不被一次失败拖累

examples/auto_put_ad_mini/tools/creative_creation.py

@@ -38,13 +38,19 @@ from config import (
 from tools.ad_api import _check, _get, _post, images_add
 from tools.landing_plan import LandingPlanResult, create_landing_plan
 from tools.material_recall import Material, recall_materials_for_video
+from tools.creative_material_usage import (
+    load_recent_used_material_ids,
+    merge_used_material_ids,
+)
 from tools.video_recall import (
     LandingVideo,
     PIAOQUANTV_HOT_FALLBACK_SOURCE,
     PIAOQUANTV_VIDEO_SOURCE,
     fetch_landing_videos_for_account,
     get_account_crowd_package,
+    map_crowd_package_for_video_recall,
 )
+from tools.video_feature_query import fetch_video_element_features
 from tools.video_risk import VideoRiskResult, check_video_risk
 
 logger = logging.getLogger(__name__)
@@ -56,16 +62,11 @@ 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
+    """承接视频基础预筛:内容品类不在黑名单。
+
+    素材召回特征统一通过 video_id 查 ODPS,不再依赖 videoContentList 返回的
+    pointType / standardElement。
+    """
     excluded = _landing_category_values(v) & LANDING_EXCLUDED_CATEGORIES
     if excluded:
         logger.info(
@@ -531,6 +532,7 @@ def prepare_one_creative_for_ad(
     Args:
         excluded_material_ids: 同广告内已挂的 material_id 集合(2026-06-09 N=3 时去重必需)。
                                召回后过滤掉这些,避免同广告挂重复素材被腾讯模型降权曝光。
+        excluded_landing_ids: 兼容旧调用保留,当前不再使用。落地页视频允许重复。
 
     Returns:
         pending record dict(飞书表格字段 + Phase 3 POST 用的完整 body + 元数据);
@@ -540,6 +542,26 @@ def prepare_one_creative_for_ad(
 
     excluded_material_ids = excluded_material_ids or set()
     excluded_landing_ids = excluded_landing_ids or set()
+    crowd_package = get_account_crowd_package(account_id)
+    video_crowd_package = map_crowd_package_for_video_recall(crowd_package)
+    try:
+        recent_material_ids = load_recent_used_material_ids(crowd_package)
+    except Exception as e:
+        logger.warning(
+            "[prepare_one_creative] account=%d crowd=%r 读取素材使用历史失败,仅使用本轮排重:%s",
+            account_id, crowd_package, e,
+        )
+        recent_material_ids = set()
+    effective_excluded_material_ids = merge_used_material_ids(
+        excluded_material_ids,
+        recent_material_ids,
+    )
+    logger.info(
+        "[prepare_one_creative] account=%d adgroup=%d crowd=%r video_crowd=%r material_dedupe run=%d recent=%d effective=%d",
+        account_id, adgroup_id, crowd_package, video_crowd_package,
+        len(excluded_material_ids), len(recent_material_ids),
+        len(effective_excluded_material_ids),
+    )
 
     chosen_landing = None
     chosen_material = None
@@ -559,16 +581,21 @@ def prepare_one_creative_for_ad(
             enable_hot_fallback=False,
         )
         valid = [v for v in videos if _is_landing_candidate(v)]
+        features_by_vid = fetch_video_element_features(v.video_id for v in valid)
         logger.info(
-            "[prepare_one_creative] account=%d adgroup=%d source=%s valid landing=%d/100 excl_mat=%d excl_landing=%d",
+            "[prepare_one_creative] account=%d adgroup=%d source=%s valid landing=%d/100 feature_videos=%d feature_rows=%d excl_mat=%d",
             account_id, adgroup_id, source_label, len(valid),
-            len(excluded_material_ids), len(excluded_landing_ids),
+            len(features_by_vid), sum(len(v) for v in features_by_vid.values()),
+            len(effective_excluded_material_ids),
         )
 
         attempts = 0
         for v in valid:
-            # 2026-06-10:跨广告 landing 去重(全局共享 used set)
             if v.video_id in excluded_landing_ids:
+                logger.info(
+                    "[prepare_one_creative]   landing=%d 本轮同广告 landing 限频,跳过",
+                    v.video_id,
+                )
                 continue
             attempts += 1
             if attempts > max_landings:
@@ -584,14 +611,42 @@ def prepare_one_creative_for_ad(
                 )
                 continue
 
-            materials = recall_materials_for_video(v, final_top_n=max_materials_per_landing)
+            element_features = features_by_vid.get(v.video_id) or []
+            if not element_features:
+                logger.info(
+                    "[prepare_one_creative]   landing=%d 无 ODPS 召回特征,跳过",
+                    v.video_id,
+                )
+                continue
+
+            materials = recall_materials_for_video(
+                v,
+                final_top_n=max_materials_per_landing,
+                element_features=element_features,
+            )
             # material_id 去重(2026-06-09):跳过已用素材(账户层 set,跨广告也共享)
-            fresh = [m for m in materials if m.material_id not in excluded_material_ids]
+            fresh = [
+                m for m in materials
+                if m.material_id not in effective_excluded_material_ids
+            ]
             if fresh:
                 chosen_landing = v
                 chosen_material = fresh[0]
                 chosen_risk = risk
                 chosen_landing_source = source_label
+                logger.info(
+                    "[prepare_one_creative]   选中 landing=%d source=%s category=%r material=%s recall=%s/%s/%s cost=%s roi=%s ctr=%s imp=%s score=%s policy=score>=0.8,cost_desc",
+                    v.video_id, source_label, v.category,
+                    chosen_material.material_id,
+                    chosen_material.recall_element_dimension,
+                    chosen_material.recall_point_type,
+                    chosen_material.recall_standard_element,
+                    chosen_material.cost,
+                    chosen_material.roi,
+                    chosen_material.ctr,
+                    chosen_material.impressions,
+                    chosen_material.score,
+                )
                 break
             if materials:
                 logger.info(
@@ -608,7 +663,7 @@ def prepare_one_creative_for_ad(
     if not chosen_landing or not chosen_material:
         logger.error(
             "[prepare_one_creative] account=%d adgroup=%d 穷尽 landing 后无可用素材(excluded=%d)",
-            account_id, adgroup_id, len(excluded_material_ids),
+            account_id, adgroup_id, len(effective_excluded_material_ids),
         )
         return None
 
@@ -624,7 +679,6 @@ def prepare_one_creative_for_ad(
     material_image_id = images_add(account_id, image_url)
 
     # 注册落地计划(xcx/save)
-    crowd_package = get_account_crowd_package(account_id)
     plan = create_landing_plan(crowd_package, chosen_landing)
 
     # 读账户 brand
@@ -654,6 +708,7 @@ def prepare_one_creative_for_ad(
         "approval_date": today,
         "account_id": account_id,
         "audience_tier": crowd_package,
+        "video_crowd_package": video_crowd_package,
         "adgroup_id": adgroup_id,
         "adgroup_name": ad_info.get("adgroup_name", ""),
         "bid_amount_yuan": ad_info.get("bid_amount_yuan", ""),
@@ -668,8 +723,22 @@ def prepare_one_creative_for_ad(
         # 素材质量字段(2026-06-10 batchByText 升级 — 给 Task 25 飞书表展示用)
         "material_ctr": chosen_material.ctr,
         "material_cost": chosen_material.cost,
+        "material_roi": chosen_material.roi,
         "material_impressions": chosen_material.impressions,
         "material_quality_score": chosen_material.quality_score,
+        "material_score": chosen_material.score,
+        "material_selection_policy": "score>=0.8,cost_desc",
+        "material_recall_strategy": chosen_material.recall_strategy,
+        "material_recall_query_text": chosen_material.recall_query_text,
+        "material_recall_config_code": chosen_material.recall_config_code,
+        "material_recall_element_dimension": chosen_material.recall_element_dimension,
+        "material_recall_point_type": chosen_material.recall_point_type,
+        "material_recall_standard_element": chosen_material.recall_standard_element,
+        "material_recall_hit_queries": chosen_material.recall_hit_queries,
+        "material_dedupe_recent_count": len(recent_material_ids),
+        "material_dedupe_run_count": len(excluded_material_ids),
+        "material_dedupe_effective_count": len(effective_excluded_material_ids),
+        "landing_dedupe_run_count": len(excluded_landing_ids),
         "creative_name": plan.root_source_id,
         # === Phase 3 POST 用 ===
         "_request_body": body,

examples/auto_put_ad_mini/tools/creative_material_usage.py

@@ -0,0 +1,199 @@
+"""Creative material usage reservation for duplicate control.
+
+Usage is written when a pending creative is prepared, before Tencent POST.
+Review/task tables remain responsible for submitted Tencent creatives.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from typing import Iterable
+
+logger = logging.getLogger(__name__)
+
+CREATIVE_MATERIAL_DEDUPE_LOOKBACK_DAYS = int(
+    os.getenv("CREATIVE_MATERIAL_DEDUPE_LOOKBACK_DAYS", "7")
+)
+
+CREATE_USAGE_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS creative_material_usage (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
+    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
+    adgroup_id BIGINT DEFAULT NULL COMMENT '广告ID',
+    crowd_package VARCHAR(200) NOT NULL COMMENT '投放人群包名称',
+    landing_video_id BIGINT DEFAULT NULL COMMENT '承接视频ID,仅审计不默认排重',
+    material_id VARCHAR(100) NOT NULL COMMENT '内部素材ID',
+    material_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
+    dynamic_creative_id BIGINT DEFAULT NULL COMMENT '腾讯动态创意ID',
+    status VARCHAR(50) NOT NULL DEFAULT 'prepared' COMMENT 'prepared/submitted/failed',
+    source VARCHAR(50) DEFAULT NULL COMMENT 'primary/hot',
+    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 '更新时间',
+    KEY idx_crowd_material_created (crowd_package, material_id, created_at),
+    KEY idx_crowd_account_ad_material (crowd_package, account_id, adgroup_id, material_id),
+    KEY idx_account_created (account_id, created_at),
+    KEY idx_status_created (status, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='创意素材使用历史/占位'
+"""
+
+
+def ensure_usage_table() -> None:
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(CREATE_USAGE_TABLE_SQL)
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def load_recent_used_material_ids(
+    crowd_package: str,
+    lookback_days: int = CREATIVE_MATERIAL_DEDUPE_LOOKBACK_DAYS,
+) -> set[str]:
+    """Load recent material reservations for the same crowd package."""
+    if not crowd_package:
+        return set()
+    ensure_usage_table()
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT DISTINCT material_id
+                FROM (
+                    SELECT material_id
+                    FROM creative_material_usage
+                    WHERE crowd_package=%s
+                      AND material_id IS NOT NULL
+                      AND material_id <> ''
+                      AND status IN ('posted_ok', 'rejected')
+                      AND created_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
+
+                    UNION
+
+                    SELECT material_id
+                    FROM creative_creation_task
+                    WHERE material_id IS NOT NULL
+                      AND material_id <> ''
+                      AND review_status IN ('approved', 'rejected')
+                      AND submitted_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
+                      AND JSON_UNQUOTE(JSON_EXTRACT(raw_record, '$.audience_tier'))=%s
+                ) t
+                """,
+                (crowd_package, int(lookback_days), int(lookback_days), crowd_package),
+            )
+            rows = cur.fetchall() or []
+    finally:
+        conn.close()
+    return {str(row["material_id"]) for row in rows if row.get("material_id")}
+
+
+def record_prepared_material_usage(record: dict, status: str = "prepared") -> None:
+    """Reserve a material once Phase 1 has produced a pending creative."""
+    material_id = str(record.get("_material_id") or "").strip()
+    crowd_package = str(record.get("audience_tier") or "").strip()
+    if not material_id or not crowd_package:
+        return
+    ensure_usage_table()
+    from db.connection import get_connection
+
+    raw_record = json.dumps(record, ensure_ascii=False, default=str)
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                INSERT INTO creative_material_usage
+                    (account_id, adgroup_id, crowd_package, landing_video_id,
+                     material_id, material_image_id, dynamic_creative_id,
+                     status, source, raw_record)
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+                """,
+                (
+                    int(record["account_id"]),
+                    int(record.get("adgroup_id") or 0) or None,
+                    crowd_package,
+                    record.get("landing_video_id"),
+                    material_id,
+                    str(record.get("_material_image_id") or ""),
+                    record.get("dynamic_creative_id"),
+                    status,
+                    record.get("landing_source"),
+                    raw_record,
+                ),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def update_material_usage_status(
+    record: dict,
+    status: str,
+    dynamic_creative_id: int | str | None = None,
+    error: str = "",
+) -> None:
+    """Update the latest usage reservation for this crowd_package + material_id."""
+    material_id = str(record.get("_material_id") or "").strip()
+    crowd_package = str(record.get("audience_tier") or "").strip()
+    if not material_id or not crowd_package:
+        return
+    ensure_usage_table()
+    from db.connection import get_connection
+
+    normalized_status = {
+        "reject": "rejected",
+        "approve": "approved",
+        "hold": "no_result",
+        "skip": "no_result",
+    }.get(status, status)
+    raw_record = json.dumps(record, ensure_ascii=False, default=str)
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                UPDATE creative_material_usage
+                SET status=%s,
+                    dynamic_creative_id=COALESCE(%s, dynamic_creative_id),
+                    raw_record=%s,
+                    updated_at=CURRENT_TIMESTAMP
+                WHERE crowd_package=%s
+                  AND material_id=%s
+                  AND account_id=%s
+                  AND adgroup_id <=> %s
+                ORDER BY id DESC
+                LIMIT 1
+                """,
+                (
+                    normalized_status,
+                    int(dynamic_creative_id) if dynamic_creative_id else None,
+                    raw_record if not error else json.dumps(
+                        {**record, "usage_error": error[:2000]},
+                        ensure_ascii=False,
+                        default=str,
+                    ),
+                    crowd_package,
+                    material_id,
+                    int(record["account_id"]),
+                    int(record.get("adgroup_id") or 0) or None,
+                ),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def merge_used_material_ids(*sets: Iterable[str]) -> set[str]:
+    out: set[str] = set()
+    for values in sets:
+        out.update(str(v) for v in values if str(v))
+    return out

examples/auto_put_ad_mini/tools/im_approval_creation.py

@@ -17,6 +17,7 @@
 """
 
 import asyncio
+import json
 import logging
 import time
 from pathlib import Path
@@ -32,17 +33,12 @@ from config import (
     FEISHU_OPERATOR_CHAT_ID,
     now_in_timezone,
 )
-from tools.feishu_doc import (
-    _auth_headers,
-    _get_tenant_token,
-    import_to_feishu,
-)
 
 logger = logging.getLogger(__name__)
 
 FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
 
-# 20 列(中文)— 2026-06-29 加落地视频风险审核 3 列
+# 26 列(中文)— 素材排序改为 score 准入 + cost 倒序,报表同步展示召回依据。
 HEADERS = [
     # A 浅灰 5 列
     "日期", "账户ID", "人群包", "广告ID", "广告名称",
@@ -51,7 +47,8 @@ HEADERS = [
     # C 浅橙 11 列(落地视频 + 风险审核 + 素材质量)
     "落地视频", "落地视频标题", "风险等级", "风险标签", "风险原因",
     "素材预览", "创意文案",
-    "CTR", "成本(元)", "曝光数",
+    "成本(元)", "ROI", "CTR", "曝光数", "相似度",
+    "召回维度", "召回点类型", "召回元素", "命中维度明细",
     "创意名(归因)",
     # D 浅绿 1 列
     "决策",
@@ -60,8 +57,8 @@ HEADERS = [
 GROUP_COLORS = [
     ((1, 5), "FFD9D9D9"),    # A 浅灰
     ((6, 8), "FFD9C8E8"),    # B 浅紫
-    ((9, 19), "FFFCD8B4"),   # C 浅橙
-    ((20, 20), "FFC6E0B4"),  # D 浅绿(第 20 列)
+    ((9, 25), "FFFCD8B4"),   # C 浅橙
+    ((26, 26), "FFC6E0B4"),  # D 浅绿
 ]
 
 COL_WIDTHS = {
@@ -69,12 +66,13 @@ COL_WIDTHS = {
     "F": 10, "G": 28, "H": 12,
     "I": 14, "J": 26, "K": 10, "L": 24, "M": 32,
     "N": 18, "O": 30,
-    "P": 10, "Q": 12, "R": 10,   # CTR / 成本 / 曝光数
-    "S": 38,                      # 创意名(归因)
-    "T": 14,                      # 决策
+    "P": 12, "Q": 10, "R": 10, "S": 10, "T": 10,
+    "U": 14, "V": 14, "W": 18, "X": 42,
+    "Y": 38,                      # 创意名(归因)
+    "Z": 14,                      # 决策
 }
 
-DECISION_COL_LETTER = "T"  # 第 20 列(风险审核列新增后后移)
+DECISION_COL_LETTER = "Z"
 VALID_ACTIONS = ("approve", "reject", "hold")
 
 
@@ -86,7 +84,7 @@ def _color_for_col(col_idx: int) -> str:
 
 
 def _format_record_to_row(rec: dict) -> list:
-    """把 pending record 字典转成 xlsx 一行(20 列)。
+    """把 pending record 字典转成 xlsx 一行。
 
     record 必含字段:
         approval_date, account_id, audience_tier, adgroup_id, adgroup_name,
@@ -101,7 +99,11 @@ def _format_record_to_row(rec: dict) -> list:
     # 素材质量数据(2026-06-10 batchByText 升级 — 来自 materialDetail.quality)
     ctr = rec.get("material_ctr")
     cost = rec.get("material_cost")
+    roi = rec.get("material_roi")
     imp = rec.get("material_impressions")
+    score = rec.get("material_score")
+    hit_queries = rec.get("material_recall_hit_queries") or []
+    hit_detail = json.dumps(hit_queries, ensure_ascii=False) if hit_queries else ""
 
     return [
         rec["approval_date"],
@@ -122,17 +124,22 @@ def _format_record_to_row(rec: dict) -> list:
         f'=HYPERLINK("{rec["material_cover_url"]}","查看素材")',
         # 创意文案(2026-06-09 加列):description 换行展示
         descriptions_str,
-        # CTR / 成本 / 曝光数(2026-06-10 新加 3 列)
-        f"{ctr:.4f}" if ctr is not None else "",
         f"{cost:.2f}" if cost is not None else "",
+        f"{roi:.4f}" if roi is not None else "",
+        f"{ctr:.4f}" if ctr is not None else "",
         str(imp) if imp is not None else "",
+        f"{score:.4f}" if score is not None else "",
+        rec.get("material_recall_element_dimension", ""),
+        rec.get("material_recall_point_type", ""),
+        rec.get("material_recall_standard_element", ""),
+        hit_detail,
         rec["creative_name"],
         "",  # 决策列空,等运营填
     ]
 
 
 def generate_approval_xlsx(records: list[dict], output_path: Path) -> Path:
-    """生成 20 列待审批 xlsx,含 hyperlink/下拉/颜色/冻结。"""
+    """生成待审批 xlsx,含 hyperlink/下拉/颜色/冻结。"""
     wb = Workbook()
     ws = wb.active
     ws.title = "创意审批"
@@ -157,7 +164,7 @@ def generate_approval_xlsx(records: list[dict], output_path: Path) -> Path:
                 horizontal="center", vertical="center", wrap_text=True,
             )
 
-    # 决策列下拉(M 列)
+    # 决策列下拉
     dv = DataValidation(
         type="list",
         formula1='"approve,reject,hold"',
@@ -195,6 +202,8 @@ def _query_first_sheet_id(sheet_token: str) -> str:
 
     飞书 sheet 内每个 sheet(tab)有独立 sheet_id,后续读 cell 需要用它。
     """
+    from tools.feishu_doc import _auth_headers, _get_tenant_token
+
     token = _get_tenant_token()
     url = f"{FEISHU_BASE_URL}/sheets/v3/spreadsheets/{sheet_token}/sheets/query"
     resp = httpx.get(url, headers=_auth_headers(token), timeout=30)
@@ -231,11 +240,14 @@ async def send_approval_to_feishu(
         preamble = (
             "【创意搭建审批】下方在线表格列出本次准备挂的创意。\n"
             "系统已在创建落地计划前过滤高风险视频,表格中的【风险等级/风险标签/风险原因】为通过项的审核摘要。\n"
+            "素材按相似度≥0.8准入,再按历史消耗倒序选择;ROI/CTR/曝光仅供审批参考。\n"
             "请在最右侧【决策】列下拉框选择 approve / reject / hold。\n"
             "我们会在 2 小时内或所有行决策完成后,执行 approve 项,跳过 reject / hold。\n"
             "提示:点击【素材预览】或【落地视频】可在浏览器打开查看(若 403 请右键新标签或复制 URL)。"
         )
 
+    from tools.feishu_doc import import_to_feishu
+
     result = await import_to_feishu(
         xlsx_path=str(xlsx_path),
         send_im=True,
@@ -262,7 +274,7 @@ async def send_approval_to_feishu(
 def read_actions_from_sheet(
     sheet_token: str, sheet_id: str, row_count: int,
 ) -> dict[int, str]:
-    """读 sheet 的"决策"列(M2:M{row_count+1}),返回 {row_idx: action}。
+    """读 sheet 的"决策"列,返回 {row_idx: action}。
 
     Args:
         row_count: pending records 数(不含 header)
@@ -271,6 +283,8 @@ def read_actions_from_sheet(
         {row_idx: action}  — row_idx 是 1-based 数据行编号(1=第一条 record)
         未填或非法值的 row 不在 dict 中,视为"未决策"。
     """
+    from tools.feishu_doc import _auth_headers, _get_tenant_token
+
     token = _get_tenant_token()
     last_row = row_count + 1
     range_str = f"{sheet_id}!{DECISION_COL_LETTER}2:{DECISION_COL_LETTER}{last_row}"
@@ -278,8 +292,34 @@ def read_actions_from_sheet(
         f"{FEISHU_BASE_URL}/sheets/v2/spreadsheets/{sheet_token}"
         f"/values/{range_str}?valueRenderOption=ToString"
     )
-    resp = httpx.get(url, headers=_auth_headers(token), timeout=30)
-    resp.raise_for_status()
+    resp = None
+    last_error: Exception | None = None
+    for attempt, delay_seconds in enumerate((0, 2, 5, 10), start=1):
+        if delay_seconds:
+            time.sleep(delay_seconds)
+        try:
+            resp = httpx.get(url, headers=_auth_headers(token), timeout=30)
+            resp.raise_for_status()
+            break
+        except httpx.HTTPStatusError as e:
+            last_error = e
+            status = e.response.status_code
+            if status < 500 and status != 429:
+                raise
+            logger.warning(
+                "[poll_approval] 读 sheet HTTP %s,第 %d 次重试 range=%s",
+                status, attempt, range_str,
+            )
+        except httpx.RequestError as e:
+            last_error = e
+            logger.warning(
+                "[poll_approval] 读 sheet 网络异常,第 %d 次重试 range=%s error=%s",
+                attempt, range_str, e,
+            )
+    else:
+        assert last_error is not None
+        raise last_error
+    assert resp is not None
     data = resp.json()
     if data.get("code") != 0:
         raise RuntimeError(f"读 sheet 失败 range={range_str} data={data}")

examples/auto_put_ad_mini/tools/material_recall.py

@@ -24,8 +24,9 @@ queryText 选择优先级(选第一个非空且非占位符 "-"):
 
 import logging
 import os
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
-from typing import List, Optional
+from typing import Iterable, List, Optional
 
 import httpx
 
@@ -80,6 +81,12 @@ SUBSTANCE_TO_VIDEO_CONFIG = {
     "KEYPOINT_SUBSTANCE": "VIDEO_KEYPOINT",
     "PURPOSE_SUBSTANCE": "VIDEO_PURPOSE",
 }
+ODPS_FEATURE_TO_CONFIG = {
+    ("解构选题", ""): "VIDEO_TOPIC",
+    ("实质", "灵感点"): "INSPIRATION_SUBSTANCE",
+    ("实质", "关键点"): "KEYPOINT_SUBSTANCE",
+    ("实质", "目的点"): "PURPOSE_SUBSTANCE",
+}
 
 # fallback(动态接口失败时用)
 MATERIAL_EFFECTIVE_CONFIG_CODES = ["VIDEO_TOPIC"] + list(POINT_TYPE_TO_SUBSTANCE.values())
@@ -100,6 +107,37 @@ EXCLUDED_COVER_URL_PATTERNS = (
 )
 
 
+@dataclass
+class RecallQuery:
+    """One ODPS-derived material recall query."""
+
+    query_text: str
+    config_code: str
+    element_dimension: str
+    point_type: str
+    standard_element: str
+    contribution_score: float = 0.0
+    dt: str = ""
+
+    @property
+    def strategy_name(self) -> str:
+        if self.element_dimension == "解构选题":
+            return "解构选题"
+        return f"{self.element_dimension}-{self.point_type}"
+
+    def to_hit(self) -> dict:
+        return {
+            "strategy": self.strategy_name,
+            "element_dimension": self.element_dimension,
+            "point_type": self.point_type,
+            "standard_element": self.standard_element,
+            "query_text": self.query_text,
+            "config_code": self.config_code,
+            "contribution_score": self.contribution_score,
+            "dt": self.dt,
+        }
+
+
 @dataclass
 class Material:
     """召回的创意素材。
@@ -119,6 +157,13 @@ class Material:
     roi: Optional[float] = None               # ROI(收入/成本)
     impressions: Optional[int] = None         # 累计曝光
     quality_score: Optional[float] = None     # 服务端 qualityScore
+    recall_strategy: str = ""
+    recall_query_text: str = ""
+    recall_config_code: str = ""
+    recall_element_dimension: str = ""
+    recall_point_type: str = ""
+    recall_standard_element: str = ""
+    recall_hit_queries: List[dict] = field(default_factory=list)
     # 原始 item dict(以备后用)
     raw: dict = field(default_factory=dict, repr=False)
 
@@ -295,21 +340,42 @@ def _call_batch_by_text(
     return payload.get("items") or []
 
 
-def _items_to_materials(items: List[dict], min_impressions: int, min_ctr: float) -> tuple:
-    """把 batchByText items 过滤 + 转 Material(2026-06-10 复用 helper)。
+def _as_float(value, default: float = 0.0) -> float:
+    try:
+        if value is None:
+            return default
+        return float(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def _as_int(value, default: int = 0) -> int:
+    try:
+        if value is None:
+            return default
+        return int(float(value))
+    except (TypeError, ValueError):
+        return default
+
 
-    过滤条件(全部满足才保留):
+def _items_to_materials(items: List[dict], sim_threshold: float) -> tuple:
+    """把召回 items 过滤 + 转 Material。
+
+    过滤条件:
     - modality=MATERIAL(防御性)
     - cover URL 不在黑名单
-    - impressions > min_impressions
-    - ctr >= min_ctr(2026-06-10 加,5% 太低)
+    - score >= sim_threshold
 
-    返回 (materials, blacklist_n, low_imp_n, low_ctr_n)。
+    CTR / impressions 只作为审批展示和兜底排序参考,不再作为硬筛。
+    返回 (materials, stats)。
     """
     out: List[Material] = []
-    excluded_blacklist = 0
-    excluded_low_imp = 0
-    excluded_low_ctr = 0
+    stats = {
+        "blacklist": 0,
+        "low_score": 0,
+        "low_imp": 0,
+        "low_ctr": 0,
+    }
     for it in items:
         if it.get("modality") != "MATERIAL":
             continue
@@ -318,119 +384,273 @@ def _items_to_materials(items: List[dict], min_impressions: int, min_ctr: float)
             continue
         cover = it.get("cover") or ""
         if any(p in cover for p in EXCLUDED_COVER_URL_PATTERNS):
-            excluded_blacklist += 1
+            stats["blacklist"] += 1
+            continue
+        score = _as_float(it.get("score"))
+        if score < sim_threshold:
+            stats["low_score"] += 1
             continue
         md = it.get("materialDetail") or {}
         q = md.get("quality") or {}
-        imp = q.get("impressions") or 0
-        if imp <= min_impressions:
-            excluded_low_imp += 1
-            continue
-        ctr = q.get("ctr") or 0.0
-        if ctr < min_ctr:
-            excluded_low_ctr += 1
-            continue
         out.append(Material(
             material_id=str(mid),
-            score=float(it.get("score") or 0.0),
+            score=score,
             title=it.get("title") or "",
             cover=cover,
             video_url=it.get("videoUrl") or "",
-            cost=q.get("cost"),
-            ctr=q.get("ctr"),
-            cvr=q.get("cvr"),
-            roi=q.get("roi"),
-            impressions=imp,
-            quality_score=q.get("qualityScore"),
+            cost=_as_float(q.get("cost")) if q.get("cost") is not None else None,
+            ctr=_as_float(q.get("ctr")) if q.get("ctr") is not None else None,
+            cvr=_as_float(q.get("cvr")) if q.get("cvr") is not None else None,
+            roi=_as_float(q.get("roi")) if q.get("roi") is not None else None,
+            impressions=_as_int(q.get("impressions")) if q.get("impressions") is not None else None,
+            quality_score=_as_float(q.get("qualityScore")) if q.get("qualityScore") is not None else None,
             raw=it,
         ))
-    return out, excluded_blacklist, excluded_low_imp, excluded_low_ctr
+    return out, stats
+
+
+def _sort_materials_by_policy(materials: List[Material]) -> List[Material]:
+    """生产排序策略:先相关性准入,再按历史消耗倒序。"""
+    return sorted(
+        materials,
+        key=lambda m: (
+            m.cost is not None,
+            m.cost or 0,
+            m.roi or 0,
+            m.impressions or 0,
+            m.ctr or 0,
+            m.quality_score or 0,
+            m.score or 0,
+        ),
+        reverse=True,
+    )
+
+
+def _material_rank(material: Material) -> tuple:
+    return (
+        material.cost is not None,
+        material.cost or 0,
+        material.roi or 0,
+        material.impressions or 0,
+        material.ctr or 0,
+        material.quality_score or 0,
+        material.score or 0,
+    )
+
+
+def _feature_attr(feature, name: str, default=""):
+    if isinstance(feature, dict):
+        return feature.get(name, default)
+    return getattr(feature, name, default)
+
+
+def _build_recall_queries_from_features(
+    element_features: Iterable,
+    query_limit: int,
+) -> List[RecallQuery]:
+    """Build recall queries from ODPS features.
+
+    Supported dimensions:
+    - 解构选题 -> VIDEO_TOPIC
+    - 实质 + 灵感点/关键点/目的点 -> *_SUBSTANCE
+    """
+    queries: List[RecallQuery] = []
+    seen = set()
+    raw_features = list(element_features or [])
+    sorted_features = sorted(
+        raw_features,
+        key=lambda f: (
+            0 if str(_feature_attr(f, "element_dimension") or "") == "解构选题" else 1,
+            -float(_feature_attr(f, "contribution_score", 0) or 0),
+        ),
+    )
+    for feature in sorted_features:
+        element_dimension = str(_feature_attr(feature, "element_dimension") or "").strip()
+        point_type = str(_feature_attr(feature, "point_type") or "").strip()
+        standard_element = str(_feature_attr(feature, "standard_element") or "").strip()
+        if standard_element in PLACEHOLDER_VALUES:
+            continue
+        config_code = ODPS_FEATURE_TO_CONFIG.get((element_dimension, point_type))
+        if not config_code and element_dimension == "解构选题":
+            config_code = ODPS_FEATURE_TO_CONFIG.get(("解构选题", ""))
+        if not config_code:
+            continue
+        key = (standard_element, config_code)
+        if key in seen:
+            continue
+        seen.add(key)
+        queries.append(RecallQuery(
+            query_text=standard_element,
+            config_code=config_code,
+            element_dimension=element_dimension,
+            point_type=point_type,
+            standard_element=standard_element,
+            contribution_score=float(_feature_attr(feature, "contribution_score", 0) or 0),
+            dt=str(_feature_attr(feature, "dt") or ""),
+        ))
+        if len(queries) >= query_limit:
+            break
+    return queries
+
+
+def _call_batch_for_recall_query(
+    query: RecallQuery,
+    *,
+    display_k: int,
+    days: int,
+    sim_threshold: float,
+    alpha: float,
+    w_ctr: float,
+    w_cvr: float,
+    w_roi: float,
+    w_open_rate: float,
+    w_fission_rate: float,
+    deconstruct_boost: float,
+    source_labels: List[str],
+) -> tuple[RecallQuery, List[dict]]:
+    items = _call_batch_by_text(
+        query_text=query.query_text,
+        config_codes=[query.config_code],
+        display_k=display_k,
+        days=days,
+        sim_threshold=sim_threshold,
+        alpha=alpha,
+        w_ctr=w_ctr,
+        w_cvr=w_cvr,
+        w_roi=w_roi,
+        w_open_rate=w_open_rate,
+        w_fission_rate=w_fission_rate,
+        deconstruct_boost=deconstruct_boost,
+        source_labels=source_labels,
+        modalities=["MATERIAL"],
+    )
+    return query, items
+
+
+def _merge_materials_by_policy(query_materials: List[tuple[RecallQuery, List[Material]]]) -> List[Material]:
+    by_mid: dict[str, Material] = {}
+    for query, materials in query_materials:
+        hit = query.to_hit()
+        for material in materials:
+            material.recall_hit_queries = [hit]
+            material.recall_strategy = query.strategy_name
+            material.recall_query_text = query.query_text
+            material.recall_config_code = query.config_code
+            material.recall_element_dimension = query.element_dimension
+            material.recall_point_type = query.point_type
+            material.recall_standard_element = query.standard_element
+            existing = by_mid.get(material.material_id)
+            if existing is None:
+                by_mid[material.material_id] = material
+                continue
+            merged_hits = existing.recall_hit_queries + [
+                h for h in material.recall_hit_queries
+                if h not in existing.recall_hit_queries
+            ]
+            if _material_rank(material) > _material_rank(existing):
+                material.recall_hit_queries = merged_hits
+                by_mid[material.material_id] = material
+            else:
+                existing.recall_hit_queries = merged_hits
+    return _sort_materials_by_policy(list(by_mid.values()))
 
 
 def recall_materials_for_video(
     landing: LandingVideo,
     final_top_n: int = DEFAULT_FINAL_TOP_N,
     source_labels: Optional[List[str]] = None,
+    element_features: Optional[Iterable] = None,
 ) -> List[Material]:
-    """素材召回(2026-06-10 用户最终:2 维度 strategy + fallback,batchByText 单 configCode)。
+    """素材召回:用 ODPS 多维特征并行召回并合并排序。
 
     流程:
-      1. _pick_query_strategies_for_batch(landing) 返回 1~2 个策略候选
-         维度 1: (standard_element, *_SUBSTANCE 按 point_type) — 优先
-         维度 2: (demand_content_topic, VIDEO_TOPIC)
-      2. 对每个策略调一次 batchByText(只传 1 个 configCode)
-         首次返回 ≥ 1 条 impressions>阈值 的 material → break,直接返回
-      3. 全部策略都失败 → 返回空 → 上层 prepare 换 landing
-
-    阈值不变保证质量(用户 2026-06-10 确认):依赖"多次尝试"提升命中率。
+      1. 从 ODPS features 生成 query:解构选题 + 实质三点。
+      2. 多 query 并行调用 batchByText。
+      3. 汇总、material_id 去重、score>=阈值、按 cost 倒序。
+
+    当前硬筛只保留相似度阈值;曝光/CTR 进入审批表但不拦截。
     """
     from config import (
         RECALL_ALPHA, RECALL_DAYS, RECALL_DECONSTRUCT_BOOST,
-        RECALL_DISPLAY_K, RECALL_MIN_CTR, RECALL_MIN_IMPRESSIONS,
+        RECALL_DISPLAY_K, RECALL_PARALLEL_MAX_WORKERS, RECALL_QUERY_LIMIT_PER_VIDEO,
         RECALL_SIM_THRESHOLD, RECALL_SOURCE_LABELS,
         RECALL_W_CTR, RECALL_W_CVR, RECALL_W_FISSION_RATE,
         RECALL_W_OPEN_RATE, RECALL_W_ROI,
     )
 
-    strategies = _pick_query_strategies_for_batch(landing)
-    if not strategies:
+    if element_features is None:
+        element_features = landing.raw.get("element_features") or []
+    queries = _build_recall_queries_from_features(
+        element_features,
+        query_limit=max(1, RECALL_QUERY_LIMIT_PER_VIDEO),
+    )
+    if not queries:
         logger.warning(
-            "[material_recall] landing video_id=%d 无可用策略(2 维度都缺),返回空",
+            "[material_recall] landing video_id=%d 无 ODPS 可用召回特征,返回空",
             landing.video_id,
         )
         return []
 
     logger.info(
-        "[material_recall] landing video_id=%d 走 %d 个策略",
-        landing.video_id, len(strategies),
+        "[material_recall] landing video_id=%d 走 %d 个 ODPS 策略:%s",
+        landing.video_id, len(queries),
+        "; ".join(f"{q.strategy_name}:{q.query_text}->{q.config_code}" for q in queries),
     )
 
-    for query_text, config_code, name in strategies:
-        logger.info(
-            "[material_recall]   策略=%s q=%r configCode=%s",
-            name, query_text[:30], config_code,
-        )
-        try:
-            items = _call_batch_by_text(
-                query_text=query_text,
-                config_codes=[config_code],          # ← 只传当前策略的 1 个
+    query_materials: List[tuple[RecallQuery, List[Material]]] = []
+    labels = source_labels or RECALL_SOURCE_LABELS
+    max_workers = max(1, min(RECALL_PARALLEL_MAX_WORKERS, len(queries)))
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = [
+            executor.submit(
+                _call_batch_for_recall_query,
+                query,
                 display_k=RECALL_DISPLAY_K,
                 days=RECALL_DAYS,
                 sim_threshold=RECALL_SIM_THRESHOLD,
                 alpha=RECALL_ALPHA,
-                w_ctr=RECALL_W_CTR, w_cvr=RECALL_W_CVR, w_roi=RECALL_W_ROI,
-                w_open_rate=RECALL_W_OPEN_RATE, w_fission_rate=RECALL_W_FISSION_RATE,
+                w_ctr=RECALL_W_CTR,
+                w_cvr=RECALL_W_CVR,
+                w_roi=RECALL_W_ROI,
+                w_open_rate=RECALL_W_OPEN_RATE,
+                w_fission_rate=RECALL_W_FISSION_RATE,
                 deconstruct_boost=RECALL_DECONSTRUCT_BOOST,
-                source_labels=source_labels or RECALL_SOURCE_LABELS,
-                modalities=["MATERIAL"],
+                source_labels=labels,
             )
-        except Exception as e:
-            logger.error(
-                "[material_recall] 策略 %s 失败,试下一个:%s", name, e,
+            for query in queries
+        ]
+        for future in as_completed(futures):
+            try:
+                query, items = future.result()
+            except Exception as e:
+                logger.error("[material_recall] 并行 batchByText 失败:%s", e)
+                continue
+            mats, stats = _items_to_materials(items, RECALL_SIM_THRESHOLD)
+            query_materials.append((query, mats))
+            logger.info(
+                "[material_recall]   策略=%s q=%r configCode=%s 返回 %d 条,⊘ 黑名单 %d,⊘ score<%.2f %d → 保留 %d",
+                query.strategy_name, query.query_text[:30], query.config_code,
+                len(items), stats["blacklist"], RECALL_SIM_THRESHOLD,
+                stats["low_score"], len(mats),
             )
-            continue
 
-        mats, n_bl, n_low_imp, n_low_ctr = _items_to_materials(
-            items, RECALL_MIN_IMPRESSIONS, RECALL_MIN_CTR,
-        )
-        logger.info(
-            "[material_recall]     服务端返回 %d 条,⊘ 黑名单 %d,⊘ imp<=%d %d,⊘ ctr<%.2f %d → 保留 %d",
-            len(items), n_bl,
-            RECALL_MIN_IMPRESSIONS, n_low_imp,
-            RECALL_MIN_CTR, n_low_ctr,
-            len(mats),
-        )
-        if mats:
-            return mats[:final_top_n]
+    merged = _merge_materials_by_policy(query_materials)
+    logger.info(
+        "[material_recall] landing video_id=%d 多维召回合并后保留 %d 条(cost desc)",
+        landing.video_id, len(merged),
+    )
+    if merged:
+        return merged[:final_top_n]
 
+    for query in queries:
         # batchByText 当前可能对 MATERIAL 返回 0;降级到历史 matchByText 路径。
-        for fallback_cc in _fallback_config_codes_for_strategy(config_code, landing):
+        for fallback_cc in _fallback_config_codes_for_strategy(query.config_code, landing):
             try:
                 fallback_items = _call_match_by_text(
-                    query_text=query_text,
+                    query_text=query.query_text,
                     config_code=fallback_cc,
                     material_top_n=RECALL_DISPLAY_K,
-                    source_labels=source_labels or RECALL_SOURCE_LABELS,
+                    source_labels=labels,
                 )
             except Exception as e:
                 logger.error(
@@ -439,34 +659,28 @@ def recall_materials_for_video(
                 )
                 continue
 
-            fmats, fn_bl, fn_low_imp, fn_low_ctr = _items_to_materials(
-                fallback_items, RECALL_MIN_IMPRESSIONS, RECALL_MIN_CTR,
+            fmats, fstats = _items_to_materials(fallback_items, RECALL_SIM_THRESHOLD)
+            fquery = RecallQuery(
+                query_text=query.query_text,
+                config_code=fallback_cc,
+                element_dimension=query.element_dimension,
+                point_type=query.point_type,
+                standard_element=query.standard_element,
+                contribution_score=query.contribution_score,
+                dt=query.dt,
             )
-            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]
+            fmats = _merge_materials_by_policy([(fquery, fmats)])
             logger.info(
-                "[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,
+                "[material_recall]     fallback=%s 返回 %d 条,⊘ 黑名单 %d,⊘ score<%.2f %d → 保留 %d(cost desc)",
+                fallback_cc, len(fallback_items), fstats["blacklist"],
+                RECALL_SIM_THRESHOLD, fstats["low_score"],
                 len(fmats),
             )
             if fmats:
-                fmats.sort(
-                    key=lambda m: (
-                        m.ctr or 0,
-                        m.impressions or 0,
-                        m.quality_score or 0,
-                        m.score or 0,
-                    ),
-                    reverse=True,
-                )
                 return fmats[:final_top_n]
-        # else 继续下一个策略
 
     logger.info(
-        "[material_recall] landing video_id=%d 所有 %d 策略全失败,返回空(上层换 landing)",
-        landing.video_id, len(strategies),
+        "[material_recall] landing video_id=%d 所有 %d ODPS 策略全失败,返回空(上层换 landing)",
+        landing.video_id, len(queries),
     )
     return []

examples/auto_put_ad_mini/tools/video_feature_query.py

@@ -1,7 +1,7 @@
-"""ODPS video feature lookup for hot landing-video fallback.
+"""ODPS video feature lookup for landing-video material recall.
 
-Hot videos from videoContentList may not include pointType / standardElement.
-This module enriches them from loghubods.dwd_video_element_contribution_analysis.
+videoContentList only chooses landing videos. Material recall features are read
+from loghubods.dwd_video_element_contribution_analysis by video id.
 """
 
 from __future__ import annotations
@@ -19,6 +19,7 @@ _MINI_DIR = Path(__file__).resolve().parent.parent
 @dataclass(frozen=True)
 class VideoElementFeature:
     video_id: int
+    element_dimension: str
     point_type: str
     standard_element: str
     contribution_score: float
@@ -31,7 +32,7 @@ CREATE TABLE IF NOT EXISTS video_element_feature_cache (
     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 '标准化元素',
+    standard_element VARCHAR(1024) 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 '是否为无特征负缓存',
@@ -39,7 +40,7 @@ CREATE TABLE IF NOT EXISTS video_element_feature_cache (
     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),
+    UNIQUE KEY uk_dt_video_point_element (dt, video_id, element_dimension, point_type, standard_element(191)),
     KEY idx_video_dt (video_id, dt),
     KEY idx_dt (dt),
     KEY idx_updated_at (updated_at)
@@ -116,6 +117,33 @@ def _ensure_cache_table() -> None:
     try:
         with conn.cursor() as cur:
             cur.execute(CREATE_CACHE_TABLE_SQL)
+            cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'element_dimension'")
+            if not cur.fetchone():
+                cur.execute(
+                    """
+ALTER TABLE video_element_feature_cache
+ADD COLUMN element_dimension VARCHAR(50) NOT NULL DEFAULT '实质' COMMENT '元素维度'
+AFTER standard_element
+"""
+                )
+            cur.execute("SHOW INDEX FROM video_element_feature_cache WHERE Key_name = 'uk_dt_video_point_element'")
+            index_rows = cur.fetchall()
+            index_cols = [str(row.get("Column_name") or "") for row in index_rows]
+            index_ok = (
+                index_cols == ["dt", "video_id", "element_dimension", "point_type", "standard_element"]
+                and str((index_rows[-1] or {}).get("Sub_part") or "") == "191"
+            )
+            if index_rows and not index_ok:
+                cur.execute("ALTER TABLE video_element_feature_cache DROP INDEX uk_dt_video_point_element")
+            cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'standard_element'")
+            standard_col = cur.fetchone()
+            if standard_col and "varchar(1024)" not in str(standard_col.get("Type") or "").lower():
+                cur.execute(
+                    """
+ALTER TABLE video_element_feature_cache
+MODIFY COLUMN standard_element VARCHAR(1024) NOT NULL COMMENT '标准化元素或解构选题文本'
+"""
+                )
             cur.execute("SHOW COLUMNS FROM video_element_feature_cache LIKE 'is_miss'")
             if not cur.fetchone():
                 cur.execute(
@@ -123,6 +151,14 @@ def _ensure_cache_table() -> None:
 ALTER TABLE video_element_feature_cache
 ADD COLUMN is_miss BOOLEAN NOT NULL DEFAULT FALSE COMMENT '是否为无特征负缓存'
 AFTER contribution_score
+"""
+                )
+            if not index_ok:
+                cur.execute(
+                    """
+ALTER TABLE video_element_feature_cache
+ADD UNIQUE KEY uk_dt_video_point_element
+    (dt, video_id, element_dimension, point_type, standard_element(191))
 """
                 )
     finally:
@@ -143,7 +179,7 @@ def _read_cache(video_ids: list[int], dt: str) -> tuple[dict[int, list[VideoElem
                 placeholders = ",".join(["%s"] * len(batch))
                 cur.execute(
                     f"""
-SELECT video_id, dt, point_type, standard_element, contribution_score, is_miss
+SELECT video_id, dt, element_dimension, point_type, standard_element, contribution_score, is_miss
 FROM video_element_feature_cache
 WHERE dt = %s
 AND video_id IN ({placeholders})
@@ -158,6 +194,7 @@ ORDER BY video_id, contribution_score DESC
                         continue
                     out.setdefault(video_id, []).append(VideoElementFeature(
                         video_id=video_id,
+                        element_dimension=str(row["element_dimension"] or ""),
                         point_type=str(row["point_type"] or ""),
                         standard_element=str(row["standard_element"] or ""),
                         contribution_score=float(row["contribution_score"] or 0.0),
@@ -179,7 +216,7 @@ def _write_cache(
             feature.dt,
             feature.point_type,
             feature.standard_element,
-            "实质",
+            feature.element_dimension,
             feature.contribution_score,
             False,
         )
@@ -220,12 +257,13 @@ 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.
+    """Fetch recall feature rows per video id from latest ODPS partition.
 
     One video can have multiple usable element rows. Keep all rows where:
-    - 元素维度 = 实质
+    - 元素维度 = 解构选题
+    - 元素维度 = 实质 and 点类型 in 灵感点/关键点/目的点
     - 贡献分 >= 0.8
-    - 点类型 and 标准化元素 are non-empty
+    - 标准化元素 is non-empty
     """
     ids: list[int] = []
     seen: set[int] = set()
@@ -263,42 +301,69 @@ def fetch_video_element_features(
         vid_list = ",".join(_quote_sql_string(str(v)) for v in batch)
         sql = f"""
 SELECT  vid
+        ,`元素维度` AS element_dimension
         ,`点类型` AS point_type
         ,`标准化元素` AS standard_element
+        ,`解构选题` AS deconstruct_topic
         ,`贡献分` AS contribution_score
         ,dt
 FROM    loghubods.dwd_video_element_contribution_analysis
 WHERE   dt = {_quote_sql_string(dt)}
 AND     vid IN ({vid_list})
-AND     `元素维度` = '实质'
+AND     (
+            (`元素维度` = '实质' AND `点类型` IN ('灵感点', '关键点', '目的点'))
+            OR (`解构选题` IS NOT NULL 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"],
+            ["vid", "element_dimension", "point_type", "standard_element", "deconstruct_topic", "contribution_score", "dt"],
         )
+        seen_features: set[tuple[int, str, str, str]] = set()
         for row in rows:
             try:
                 video_id = int(row.get("vid"))
             except (TypeError, ValueError):
                 continue
+            element_dimension = str(row.get("element_dimension") or "").strip()
             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 ""),
-            ))
+            contribution_score = float(row.get("contribution_score") or 0.0)
+            dt_value = str(row.get("dt") or "")
+
+            if (
+                element_dimension == "实质"
+                and point_type in {"灵感点", "关键点", "目的点"}
+                and standard_element
+            ):
+                key = (video_id, element_dimension, point_type, standard_element)
+                if key not in seen_features:
+                    seen_features.add(key)
+                    fetched.setdefault(video_id, []).append(VideoElementFeature(
+                        video_id=video_id,
+                        element_dimension=element_dimension,
+                        point_type=point_type,
+                        standard_element=standard_element,
+                        contribution_score=contribution_score,
+                        dt=dt_value,
+                    ))
+
+            topic = str(row.get("deconstruct_topic") or "").strip()
+            if topic:
+                key = (video_id, "解构选题", "", topic)
+                if key not in seen_features:
+                    seen_features.add(key)
+                    fetched.setdefault(video_id, []).append(VideoElementFeature(
+                        video_id=video_id,
+                        element_dimension="解构选题",
+                        point_type="",
+                        standard_element=topic,
+                        contribution_score=contribution_score,
+                        dt=dt_value,
+                    ))
 
     fetched_ids = set(fetched.keys())
     miss_ids = [video_id for video_id in missing_ids if video_id not in fetched_ids]

examples/auto_put_ad_mini/tools/video_recall.py

@@ -19,6 +19,7 @@
 
 import logging
 import os
+import json
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import List, Optional
@@ -48,6 +49,30 @@ PIAOQUANTV_HOT_FALLBACK_ENABLED = os.getenv(
 PIAOQUANTV_HOT_FALLBACK_SOURCE = os.getenv("PIAOQUANTV_HOT_FALLBACK_SOURCE", "hot")
 
 
+def _load_video_crowd_package_map() -> dict[str, str]:
+    raw = os.getenv("VIDEO_RECALL_CROWD_PACKAGE_MAP", "").strip()
+    if raw:
+        try:
+            data = json.loads(raw)
+            if isinstance(data, dict):
+                return {
+                    str(k).strip(): str(v).strip()
+                    for k, v in data.items()
+                    if str(k).strip() and str(v).strip()
+                }
+        except json.JSONDecodeError:
+            logger.warning("[video_recall] VIDEO_RECALL_CROWD_PACKAGE_MAP 不是合法 JSON,使用默认映射")
+    return {"cell*year*商业": "wx*商业"}
+
+
+VIDEO_RECALL_CROWD_PACKAGE_MAP = _load_video_crowd_package_map()
+
+
+def map_crowd_package_for_video_recall(crowd_package: str) -> str:
+    """只映射内容服务 videoContentList 的 crowdPackage,不影响腾讯投放人群包。"""
+    return VIDEO_RECALL_CROWD_PACKAGE_MAP.get(crowd_package, crowd_package)
+
+
 def _normalize_category(raw) -> str:
     if raw is None:
         return ""
@@ -74,6 +99,7 @@ def _enrich_hot_video_elements(videos: List["LandingVideo"]) -> None:
             continue
         feature_payload = [
             {
+                "element_dimension": f.element_dimension,
                 "point_type": f.point_type,
                 "standard_element": f.standard_element,
                 "contribution_score": f.contribution_score,
@@ -253,9 +279,10 @@ def fetch_landing_videos_for_account(
     主池不足时用同一个 crowdPackage + source=hot 补充热门视频。
     """
     crowd_package = get_account_crowd_package(account_id)
+    video_crowd_package = map_crowd_package_for_video_recall(crowd_package)
     selected_source = PIAOQUANTV_VIDEO_SOURCE if source is None else source
     primary = fetch_landing_videos(
-        crowd_package=crowd_package,
+        crowd_package=video_crowd_package,
         page_size=page_size,
         source=selected_source,
     )
@@ -270,11 +297,11 @@ def fetch_landing_videos_for_account(
 
     missing = page_size - len(primary)
     logger.info(
-        "[video_recall] primary 不足,用 hot 兜底: account=%d crowd=%r primary=%d need=%d",
-        account_id, crowd_package, len(primary), missing,
+        "[video_recall] primary 不足,用 hot 兜底: account=%d crowd=%r video_crowd=%r primary=%d need=%d",
+        account_id, crowd_package, video_crowd_package, len(primary), missing,
     )
     fallback = fetch_landing_videos(
-        crowd_package=crowd_package,
+        crowd_package=video_crowd_package,
         page_size=missing,
         source=PIAOQUANTV_HOT_FALLBACK_SOURCE,
     )
@@ -286,7 +313,7 @@ def fetch_landing_videos_for_account(
         merged.append(v)
         seen.add(v.video_id)
     logger.info(
-        "[video_recall] primary + hot 合并后 %d 条(account=%d crowd=%r)",
-        len(merged), account_id, crowd_package,
+        "[video_recall] primary + hot 合并后 %d 条(account=%d crowd=%r video_crowd=%r)",
+        len(merged), account_id, crowd_package, video_crowd_package,
     )
     return merged