feat(auto-put): add external material recall workflow

AGENTS.md

@@ -77,8 +77,12 @@
 - 当前素材硬筛只看相似度:`score >= 0.8`。
 - 曝光、CTR、ROI 只作为审批展示和兜底排序参考,不作为硬筛。
 - 素材通过相似度筛选后,默认按历史消耗 `cost` 倒序选择。
+- 飞书「素材来源」支持 `history`、`ai_generated`、`external_recall`;外部素材链路必须独立执行,不能改变历史召回和原 AI 生成链路的行为。
+- 外部素材使用独立 `sourceLabels` 召回,相似度准入后按 `loghubods.cooperate_top_cards_daily_v1.card_cover_id` 关联配置窗口内 `访问uv` 之和倒序,默认窗口为 90 天,同 UV 时按相似度倒序。
+- 外部素材通过参考图编辑只清除已有播放按钮或播放器控件,不主动改写原图主题、标题、主体和版式;派生图必须经过 AI 预审和飞书人工审批,审批通过后才上传腾讯并创建创意。
 - 落地页视频去重按素材来源拆池。历史素材链路同一 `crowd_package + landing_video_id` 默认最多 1 次。
 - AI 生成素材链路和历史素材链路互不占用 landing 去重名额;AI 链路同一 `crowd_package + landing_video_id` 默认最多 1 次。
+- 外部素材链路拥有独立 landing 去重池,同一 `crowd_package + landing_video_id` 默认最多 1 次。
 - 落地页视频跨轮近期排重只能基于已提交腾讯、已执行或已有明确投放结果的记录。
 - 未提交腾讯的 `prepared` / `pending` 记录不能进入跨轮排重;它们只能用于当前运行内去重,或作为中断后可复用的恢复缓存。
 - 同一广告内同一 `landing_video_id` 仍保留限频保护,默认最多 1 次。

examples/auto_put_ad_mini/.env.example

@@ -84,6 +84,14 @@ FEISHU_OPERATOR_CHAT_ID=oc_88e0a1970a7de02eb5ac225a8b0cedea
 # AI 图片生成模型。当前默认使用 OpenRouter Images API 可用的 pro 图片模型。
 # OPENROUTER_IMAGE_MODEL=google/gemini-3-pro-image
 
+# 外部素材召回账户专用。飞书「素材来源」填“外部素材”或“外部素材召回”时生效。
+EXTERNAL_RECALL_SOURCE_LABEL=外部合作
+EXTERNAL_RECALL_CANDIDATE_LIMIT=300
+EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING=3
+EXTERNAL_RECALL_UV_WINDOW_DAYS=90
+# 默认复用 OPENROUTER_IMAGE_MODEL；可单独指定支持 input_references 的图片模型。
+# EXTERNAL_IMAGE_MODEL=google/gemini-3.1-flash-image
+
 # ========================================
 # 数据库配置（MySQL）
 # ========================================

examples/auto_put_ad_mini/DEPLOYMENT.md

@@ -187,6 +187,11 @@ curl -X POST http://localhost:8080/trigger | jq .
 | `ODPS_ACCESS_ID` | ODPS 访问ID | - | 是 |
 | `ODPS_ACCESS_SECRET` | ODPS 访问密钥 | - | 是 |
 | `OPEN_ROUTER_API_KEY` | OpenRouter API Key | - | 是 |
+| `EXTERNAL_RECALL_SOURCE_LABEL` | 外部素材召回源标签 | `外部合作` | 否 |
+| `EXTERNAL_RECALL_CANDIDATE_LIMIT` | 外部素材进入 UV 排序的候选上限 | `300` | 否 |
+| `EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING` | 单个承接视频最多尝试清理的外部素材数 | `3` | 否 |
+| `EXTERNAL_RECALL_UV_WINDOW_DAYS` | 外部素材访问 UV 汇总窗口天数 | `90` | 否 |
+| `EXTERNAL_IMAGE_MODEL` | 外部素材参考图编辑模型 | 复用 `OPENROUTER_IMAGE_MODEL` | 否 |
 
 ### 白名单配置
 

examples/auto_put_ad_mini/db/schema.sql

@@ -54,7 +54,7 @@ CREATE TABLE IF NOT EXISTS ad_creation_account_config (
     delivery_version VARCHAR(100) NOT NULL COMMENT '投放版本',
     brand_key VARCHAR(100) DEFAULT NULL COMMENT '品牌资产模板key',
     feedback_key VARCHAR(100) DEFAULT NULL COMMENT '监测链接模板key',
-    material_source VARCHAR(50) NOT NULL DEFAULT 'history' COMMENT '创意素材来源:history/ai_generated',
+    material_source VARCHAR(50) NOT NULL DEFAULT 'history' COMMENT '创意素材来源:history/ai_generated/external_recall',
     ai_fallback_to_history BOOLEAN NOT NULL DEFAULT FALSE COMMENT 'AI生成失败时是否回退历史素材',
     audience_name VARCHAR(200) NOT NULL COMMENT '人工配置的人群包名称',
     audience_pack_id BIGINT DEFAULT NULL COMMENT '系统解析/验证后的人群包ID;NULL表示泛人群',
@@ -335,7 +335,47 @@ CREATE TABLE IF NOT EXISTS ai_generated_material (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='AI生成创意图片素材';
 
 -- =====================================================
--- 14. 高消耗素材表现快照任务
+-- 14. 外部召回素材AI清理派生图
+-- =====================================================
+CREATE TABLE IF NOT EXISTS external_recalled_material (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
+    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
+    adgroup_id BIGINT NOT NULL COMMENT '广告ID',
+    crowd_package VARCHAR(200) NOT NULL COMMENT '投放人群包名称',
+    landing_video_id BIGINT NOT NULL COMMENT '承接视频ID',
+    source_material_id VARCHAR(255) NOT NULL COMMENT '外部召回materialId/card_cover_id',
+    source_image_url VARCHAR(1000) NOT NULL COMMENT '外部素材原图URL',
+    source_title VARCHAR(1000) DEFAULT NULL COMMENT '外部素材标题',
+    similarity_score DOUBLE NOT NULL DEFAULT 0 COMMENT '向量召回相似度',
+    visit_uv_30d BIGINT NOT NULL DEFAULT 0 COMMENT '配置窗口内访问UV之和（列名兼容历史）',
+    uv_window_start VARCHAR(8) DEFAULT NULL COMMENT 'UV窗口开始分区',
+    uv_window_end VARCHAR(8) DEFAULT NULL COMMENT 'UV窗口结束分区',
+    recall_json MEDIUMTEXT DEFAULT NULL COMMENT '召回命中与原始素材JSON',
+    edit_prompt MEDIUMTEXT NOT NULL COMMENT '去播放按钮图生图prompt',
+    model VARCHAR(200) NOT NULL COMMENT '图生图模型',
+    oss_object_key VARCHAR(500) DEFAULT NULL COMMENT '派生图OSS key',
+    oss_url VARCHAR(1000) DEFAULT NULL COMMENT '派生图公网URL',
+    status VARCHAR(50) NOT NULL DEFAULT 'generated' COMMENT '生成、审批和腾讯提交状态',
+    approval_status VARCHAR(50) DEFAULT NULL COMMENT '人工审批状态',
+    tencent_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
+    dynamic_creative_id BIGINT DEFAULT NULL COMMENT '腾讯动态创意ID',
+    ai_review_status VARCHAR(50) DEFAULT NULL COMMENT 'AI审核状态',
+    ai_review_score INT DEFAULT NULL COMMENT 'AI审核评分',
+    ai_review_reason TEXT DEFAULT NULL COMMENT 'AI审核原因',
+    ai_review_json MEDIUMTEXT DEFAULT NULL COMMENT 'AI审核JSON',
+    error TEXT DEFAULT NULL COMMENT '处理错误',
+    raw_response MEDIUMTEXT DEFAULT NULL COMMENT '图生图模型响应摘要',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+    UNIQUE KEY uk_account_ad_landing_source
+        (account_id, adgroup_id, landing_video_id, source_material_id),
+    KEY idx_account_ad_status (account_id, adgroup_id, status),
+    KEY idx_source_created (source_material_id, created_at),
+    KEY idx_status_created (status, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='外部召回素材AI清理派生图';
+
+-- =====================================================
+-- 15. 高消耗素材表现快照任务
 -- =====================================================
 CREATE TABLE IF NOT EXISTS material_performance_snapshot_run (
     id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',

examples/auto_put_ad_mini/execute_creation_apply.py

@@ -40,6 +40,7 @@ sys.path.insert(0, str(_HERE))
 from tools.creative_creation import post_creative_with_prepared_body  # noqa: E402
 from tools.ad_api import images_add  # noqa: E402
 from tools.ai_generated_material import update_generated_material_status  # noqa: E402
+from tools.external_recalled_material import update_external_material_status  # noqa: E402
 from tools.creative_review import (  # noqa: E402
     mark_creation_submit_failed,
     record_creation_submission,
@@ -49,6 +50,11 @@ from tools.creative_material_usage import update_material_usage_status  # noqa:
 logger = logging.getLogger("execute_creation_apply")
 
 
+def _update_deferred_material_status(record: dict, status: str, **kwargs) -> None:
+    update_generated_material_status(record, status, **kwargs)
+    update_external_material_status(record, status, **kwargs)
+
+
 def _fmt_quality_summary(r: dict) -> str:
     """Task 29:把 record 里的素材质量字段(prepare 阶段写入)格式化成一行。
 
@@ -191,7 +197,7 @@ def apply_pending_records(records: list[dict]) -> dict:
         if action != "approve":
             try:
                 update_material_usage_status(rec, action)
-                update_generated_material_status(rec, action)
+                _update_deferred_material_status(rec, action)
             except Exception as e:
                 logger.warning(
                     "[apply] 更新素材 usage 状态失败 action=%s material=%s: %s",
@@ -207,7 +213,7 @@ def apply_pending_records(records: list[dict]) -> dict:
             posted_failed += 1
             try:
                 update_material_usage_status(rec, "post_failed", error=rec["error"])
-                update_generated_material_status(rec, "post_failed", error=rec["error"])
+                _update_deferred_material_status(rec, "post_failed", error=rec["error"])
             except Exception as e:
                 logger.warning("[apply] 更新素材 usage 状态失败: %s", e)
             out_records.append(rec)
@@ -221,7 +227,7 @@ def apply_pending_records(records: list[dict]) -> dict:
                 image_comp = body.get("creative_components", {}).get("image") or []
                 if image_comp:
                     image_comp[0].setdefault("value", {})["image_id"] = material_image_id
-                update_generated_material_status(
+                _update_deferred_material_status(
                     rec,
                     "approve",
                     tencent_image_id=material_image_id,
@@ -232,7 +238,7 @@ def apply_pending_records(records: list[dict]) -> dict:
                 mark_creation_submit_failed(rec, rec["error"])
                 try:
                     update_material_usage_status(rec, "post_failed", error=rec["error"])
-                    update_generated_material_status(rec, "post_failed", error=rec["error"])
+                    _update_deferred_material_status(rec, "post_failed", error=rec["error"])
                 except Exception as update_e:
                     logger.warning("[apply] 更新素材 usage 状态失败: %s", update_e)
                 out_records.append(rec)
@@ -248,7 +254,7 @@ def apply_pending_records(records: list[dict]) -> dict:
             posted_ok += 1
             try:
                 update_material_usage_status(rec, "posted_ok", dynamic_creative_id=cid)
-                update_generated_material_status(
+                _update_deferred_material_status(
                     rec,
                     "posted_ok",
                     dynamic_creative_id=cid,
@@ -266,7 +272,7 @@ def apply_pending_records(records: list[dict]) -> dict:
             mark_creation_submit_failed(rec, "post_failed")
             try:
                 update_material_usage_status(rec, "post_failed", error=rec["error"])
-                update_generated_material_status(rec, "post_failed", error=rec["error"])
+                _update_deferred_material_status(rec, "post_failed", error=rec["error"])
             except Exception as e:
                 logger.warning("[apply] 更新素材 usage 状态失败: %s", e)
         out_records.append(rec)

examples/auto_put_ad_mini/execute_creation_once.py

@@ -77,6 +77,7 @@ from tools.creative_material_usage import (  # noqa: E402
 )
 from tools.account_material_strategy import (  # noqa: E402
     MATERIAL_SOURCE_AI_GENERATED,
+    MATERIAL_SOURCE_EXTERNAL_RECALL,
     load_account_material_strategy,
 )
 from tools.video_recall import get_account_crowd_package  # noqa: E402
@@ -492,12 +493,20 @@ def phase1_prepare(
             )
             return False
         landing_video_id = rec.get("landing_video_id")
-        actual_material_source = (
-            MATERIAL_SOURCE_AI_GENERATED
-            if str(rec.get("material_source") or "") == MATERIAL_SOURCE_AI_GENERATED
-            or str(rec.get("_material_id") or "").startswith("ai:")
-            else MATERIAL_SOURCE_HISTORY
-        )
+        record_material_source = str(rec.get("material_source") or "")
+        record_material_id = str(rec.get("_material_id") or "")
+        if (
+            record_material_source == MATERIAL_SOURCE_EXTERNAL_RECALL
+            or record_material_id.startswith("external:")
+        ):
+            actual_material_source = MATERIAL_SOURCE_EXTERNAL_RECALL
+        elif (
+            record_material_source == MATERIAL_SOURCE_AI_GENERATED
+            or record_material_id.startswith("ai:")
+        ):
+            actual_material_source = MATERIAL_SOURCE_AI_GENERATED
+        else:
+            actual_material_source = MATERIAL_SOURCE_HISTORY
         if landing_video_id is not None:
             landing_video_id = int(landing_video_id)
             if landing_counts_for_ad.get(landing_video_id, 0) >= MAX_SAME_LANDING_PER_AD_IN_RUN:
@@ -602,16 +611,19 @@ def phase1_prepare(
                 landing_usage_counts_by_crowd[crowd_package] = {
                     MATERIAL_SOURCE_HISTORY: {},
                     MATERIAL_SOURCE_AI_GENERATED: {},
+                    MATERIAL_SOURCE_EXTERNAL_RECALL: {},
                 }
             logger.info(
-                "[phase1] crowd=%r 近期 landing 使用 history=%d ai_generated=%d",
+                "[phase1] crowd=%r 近期 landing 使用 history=%d ai_generated=%d external_recall=%d",
                 crowd_package,
                 len(landing_usage_counts_by_crowd[crowd_package].get(MATERIAL_SOURCE_HISTORY, {})),
                 len(landing_usage_counts_by_crowd[crowd_package].get(MATERIAL_SOURCE_AI_GENERATED, {})),
+                len(landing_usage_counts_by_crowd[crowd_package].get(MATERIAL_SOURCE_EXTERNAL_RECALL, {})),
             )
         crowd_landing_counts = landing_usage_counts_by_crowd[crowd_package]
         crowd_landing_counts.setdefault(MATERIAL_SOURCE_HISTORY, {})
         crowd_landing_counts.setdefault(MATERIAL_SOURCE_AI_GENERATED, {})
+        crowd_landing_counts.setdefault(MATERIAL_SOURCE_EXTERNAL_RECALL, {})
 
         try:
             material_strategy = load_account_material_strategy(account_id)
@@ -622,8 +634,8 @@ def phase1_prepare(
             )
             material_strategy = None
         requested_material_source = (
-            MATERIAL_SOURCE_AI_GENERATED
-            if material_strategy is not None and material_strategy.use_ai_generated
+            material_strategy.material_source
+            if material_strategy is not None
             else MATERIAL_SOURCE_HISTORY
         )
         landing_max_uses = MAX_SAME_LANDING_PER_AD_IN_RUN

examples/auto_put_ad_mini/prompts/external_material_cleanup.md

@@ -0,0 +1,20 @@
+编辑输入的中文广告素材图，只做必要的局部清理。
+
+目标：
+- 如果画面中存在播放三角形、圆形播放图标、播放器控件或伪装成可点击播放入口的覆盖元素，完整移除，并自然修复被遮挡背景。
+- 如果不存在上述元素，保持原图内容，不主动改版。
+
+必须保持：
+- 主体人物、物品、场景、构图、色彩和视觉风格。
+- 原有中文标题的文字内容、字形可读性、位置和层级。
+- 原图的广告主题及信息表达。
+- 16:9 横版构图。
+
+禁止：
+- 不得新增、删除、改写或纠正任何标题和文案。
+- 不得新增人物、物品、按钮、图标、logo、水印、二维码或装饰文字。
+- 不得改变人物身份、性别、年龄、表情和主要动作。
+- 不得把中国人物或场景改成外国人物或外国场景。
+- 不得重做整张海报，不得更换背景，不得改变主题。
+
+输出一张清晰、完整、无播放按钮覆盖的 16:9 中文广告素材图。

examples/auto_put_ad_mini/sync_feishu_account_config.py

@@ -12,7 +12,7 @@
   - 「否」会禁用已有的 ad_creation_account_config,不删除历史配置。
   - 「初始出价」支持固定值 0.35 或范围 0.28-0.31。
   - 「预算（单广告）」为空时使用模板默认预算;不限制/不限/- 按腾讯不限预算 0 传递;数字按元转换为分。
-  - 「素材来源」为空默认历史素材;填 AI生成素材 时走 AI 图片生成链路。
+  - 「素材来源」为空默认历史素材;支持历史素材、AI生成素材、外部合作素材。
   - 「生成失败是否回退历史素材」为空/否默认不回退。
   - 「出价方式」为空默认稳定成本;填「最大转化量」时使用「最大转化量出价」作为控制成本。
   - 「投放版位」为空使用投放模板默认;填 AIM+ 时使用智能版位。

examples/auto_put_ad_mini/test_ai_generated_materials.py

@@ -22,6 +22,7 @@ class AiGeneratedMaterialsTest(unittest.TestCase):
         self.assertEqual(normalize_material_source(""), "history")
         self.assertEqual(normalize_material_source("历史素材"), "history")
         self.assertEqual(normalize_material_source("AI生成素材"), "ai_generated")
+        self.assertEqual(normalize_material_source("外部素材"), "external_recall")
 
     def test_parse_bool_flag_accepts_feishu_yes_no(self):
         self.assertTrue(parse_bool_flag("是"))

examples/auto_put_ad_mini/test_im_approval_creation.py

@@ -62,7 +62,7 @@ class CreativeApprovalSheetTest(unittest.TestCase):
             self.assertIn("素材预览", headers)
             self.assertIn("素材链接", headers)
             self.assertEqual("决策", headers[-1])
-            self.assertEqual("AB", DECISION_COL_LETTER)
+            self.assertEqual("AD", DECISION_COL_LETTER)
 
             material_link_col = headers.index("素材链接") + 1
             self.assertEqual(

examples/auto_put_ad_mini/tools/account_material_strategy.py

@@ -1,7 +1,7 @@
 """Account-level creative material source strategy.
 
-The default remains historical material recall. Accounts explicitly configured
-as AI generated use generated image assets in module B.
+The default remains historical recall. AI generation and external recall are
+enabled only for accounts explicitly configured for those module B branches.
 """
 
 from __future__ import annotations
@@ -13,6 +13,7 @@ logger = logging.getLogger(__name__)
 
 MATERIAL_SOURCE_HISTORY = "history"
 MATERIAL_SOURCE_AI_GENERATED = "ai_generated"
+MATERIAL_SOURCE_EXTERNAL_RECALL = "external_recall"
 
 _AI_SOURCE_VALUES = {
     "ai",
@@ -30,6 +31,14 @@ _HISTORY_SOURCE_VALUES = {
     "历史投放素材",
     "历史已投素材",
 }
+_EXTERNAL_SOURCE_VALUES = {
+    "external",
+    "external_recall",
+    "外部",
+    "外部素材",
+    "外部素材召回",
+    "外部合作素材",
+}
 _TRUE_VALUES = {"1", "true", "yes", "y", "是", "允许", "回退"}
 
 
@@ -43,6 +52,10 @@ class AccountMaterialStrategy:
     def use_ai_generated(self) -> bool:
         return self.material_source == MATERIAL_SOURCE_AI_GENERATED
 
+    @property
+    def use_external_recall(self) -> bool:
+        return self.material_source == MATERIAL_SOURCE_EXTERNAL_RECALL
+
 
 def normalize_material_source(raw: object) -> str:
     text = str(raw or "").strip().lower()
@@ -50,6 +63,8 @@ def normalize_material_source(raw: object) -> str:
         return MATERIAL_SOURCE_AI_GENERATED
     if text in _HISTORY_SOURCE_VALUES:
         return MATERIAL_SOURCE_HISTORY
+    if text in _EXTERNAL_SOURCE_VALUES:
+        return MATERIAL_SOURCE_EXTERNAL_RECALL
     raise ValueError(f"未知素材来源:{raw}")
 
 
@@ -79,7 +94,7 @@ def ensure_account_material_strategy_columns() -> None:
                     """
                     ALTER TABLE ad_creation_account_config
                     ADD COLUMN material_source VARCHAR(50) NOT NULL DEFAULT 'history'
-                    COMMENT '创意素材来源:history/ai_generated'
+                    COMMENT '创意素材来源:history/ai_generated/external_recall'
                     AFTER feedback_key
                     """
                 )

examples/auto_put_ad_mini/tools/ai_generated_material.py

@@ -721,7 +721,12 @@ def _extract_image_ref_from_openrouter(data: dict) -> str:
     raise RuntimeError("OpenRouter 响应未包含可识别图片URL/base64")
 
 
-def generate_image_bytes(prompt: str, model: str = OPENROUTER_IMAGE_MODEL) -> tuple[bytes, str, dict]:
+def generate_image_bytes(
+    prompt: str,
+    model: str = OPENROUTER_IMAGE_MODEL,
+    *,
+    input_image_url: str = "",
+) -> tuple[bytes, str, dict]:
     """Generate one image through OpenRouter Images API.
 
     The Images API lets us pass aspect_ratio directly. We still normalize the
@@ -735,6 +740,13 @@ def generate_image_bytes(prompt: str, model: str = OPENROUTER_IMAGE_MODEL) -> tu
         "resolution": AI_IMAGE_RESOLUTION,
         "n": 1,
     }
+    if input_image_url:
+        body["input_references"] = [
+            {
+                "type": "image_url",
+                "image_url": {"url": input_image_url},
+            }
+        ]
     headers = {
         "authorization": f"Bearer {_openrouter_api_key()}",
         "content-type": "application/json",

examples/auto_put_ad_mini/tools/creative_creation.py

@@ -41,6 +41,10 @@ from tools.ai_generated_material import (
     get_or_generate_assets_for_landing,
     update_generated_material_status,
 )
+from tools.external_recalled_material import (
+    prepare_external_material_for_landing,
+    update_external_material_status,
+)
 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 (
@@ -711,6 +715,7 @@ def prepare_one_creative_for_ad(
     chosen_landing_source = ""
     chosen_material_source = material_strategy.material_source
     chosen_ai_generated_material_id = None
+    chosen_external_recalled_material_id = None
 
     source_stats_by_label: dict[str, dict] = {}
 
@@ -722,6 +727,9 @@ def prepare_one_creative_for_ad(
             "ai_attempts": 0,
             "ai_failed": 0,
             "ai_empty": 0,
+            "external_attempts": 0,
+            "external_failed": 0,
+            "external_empty": 0,
             "history_recall_empty": 0,
             "all_excluded": 0,
             "selected": 0,
@@ -749,7 +757,29 @@ def prepare_one_creative_for_ad(
         source_stats = source_stats_by_label[source_label]
         materials = []
         candidate_material_source = material_strategy.material_source
-        if material_strategy.use_ai_generated:
+        if material_strategy.use_external_recall:
+            source_stats["external_attempts"] += 1
+            try:
+                processed = prepare_external_material_for_landing(
+                    account_id=account_id,
+                    adgroup_id=adgroup_id,
+                    crowd_package=crowd_package,
+                    landing=v,
+                    element_features=element_features,
+                    excluded_material_ids=effective_excluded_material_ids,
+                )
+                materials = [processed] if processed is not None else []
+            except Exception as e:
+                source_stats["external_failed"] += 1
+                logger.exception(
+                    "[prepare_one_creative]   landing=%d 外部素材处理失败:%s",
+                    v.video_id, e,
+                )
+            if not materials:
+                source_stats["external_empty"] += 1
+                failed_landing_ids.add(v.video_id)
+                continue
+        elif material_strategy.use_ai_generated:
             source_stats["ai_attempts"] += 1
             try:
                 assets = get_or_generate_assets_for_landing(
@@ -787,8 +817,14 @@ def prepare_one_creative_for_ad(
             continue
 
         if (not materials) and (
-            not material_strategy.use_ai_generated
-            or material_strategy.ai_fallback_to_history
+            (
+                not material_strategy.use_ai_generated
+                and not material_strategy.use_external_recall
+            )
+            or (
+                material_strategy.use_ai_generated
+                and material_strategy.ai_fallback_to_history
+            )
         ):
             if material_strategy.use_ai_generated:
                 logger.info(
@@ -817,6 +853,10 @@ def prepare_one_creative_for_ad(
             chosen_material_source = candidate_material_source
             if chosen_material.material_id.startswith("ai:"):
                 chosen_ai_generated_material_id = chosen_material.raw.get("ai_generated_material_id")
+            if chosen_material.material_id.startswith("external:"):
+                chosen_external_recalled_material_id = chosen_material.raw.get(
+                    "external_recalled_material_id"
+                )
             source_stats["selected"] += 1
             logger.info(
                 "[prepare_one_creative]   选中 landing=%d source=%s category=%r material_source=%s material=%s recall=%s/%s/%s cost=%s roi=%s ctr=%s imp=%s score=%s policy=%s",
@@ -831,7 +871,13 @@ def prepare_one_creative_for_ad(
                 chosen_material.ctr,
                 chosen_material.impressions,
                 chosen_material.score,
-                "ai_generated" if chosen_material.material_id.startswith("ai:") else "score>=0.8,cost_desc",
+                (
+                    "ai_generated"
+                    if chosen_material.material_id.startswith("ai:")
+                    else "score>=0.8,visit_uv_window_desc,ai_cleanup"
+                    if chosen_material.material_id.startswith("external:")
+                    else "score>=0.8,cost_desc"
+                ),
             )
             break
         if materials:
@@ -873,8 +919,12 @@ def prepare_one_creative_for_ad(
         return None
 
     is_ai_generated_material = chosen_material.material_id.startswith("ai:")
-    if is_ai_generated_material:
-        # AI 图先走飞书人工审批,审批通过后在 Phase 3 上传腾讯图片。
+    is_external_recalled_material = chosen_material.material_id.startswith("external:")
+    is_deferred_generated_material = (
+        is_ai_generated_material or is_external_recalled_material
+    )
+    if is_deferred_generated_material:
+        # AI生成图与外部素材派生图先走人工审批,Phase 3 才上传腾讯。
         material_image_id = ""
     else:
         # 历史素材保持原有行为:Phase 1 上传腾讯图片(MD5 幂等)。
@@ -928,11 +978,15 @@ def prepare_one_creative_for_ad(
         "material_cost": chosen_material.cost,
         "material_roi": chosen_material.roi,
         "material_impressions": chosen_material.impressions,
+        "material_visit_uv_30d": chosen_material.visit_uv_30d,
         "material_quality_score": chosen_material.quality_score,
         "material_score": chosen_material.score,
         "material_selection_policy": (
             "ai_generated_manual_review"
-            if is_ai_generated_material else "score>=0.8,cost_desc"
+            if is_ai_generated_material
+            else "score>=0.8,visit_uv_window_desc,ai_cleanup_manual_review"
+            if is_external_recalled_material
+            else "score>=0.8,cost_desc"
         ),
         "material_recall_strategy": chosen_material.recall_strategy,
         "material_recall_query_text": chosen_material.recall_query_text,
@@ -951,8 +1005,17 @@ def prepare_one_creative_for_ad(
         # === 追溯元数据(写 summary JSON)===
         "_material_id": chosen_material.material_id,
         "_material_image_id": material_image_id,
-        "_pending_image_url": image_url if is_ai_generated_material else "",
+        "_pending_image_url": image_url if is_deferred_generated_material else "",
         "_ai_generated_material_id": chosen_ai_generated_material_id,
+        "_external_recalled_material_id": chosen_external_recalled_material_id,
+        "_external_source_material_id": (
+            chosen_material.raw.get("source_material_id")
+            if is_external_recalled_material else ""
+        ),
+        "_external_source_image_url": (
+            chosen_material.raw.get("source_image_url")
+            if is_external_recalled_material else ""
+        ),
         "_plan_id": plan.plan_id,
         "_experiment_id": chosen_landing.experiment_id,
         "_brand_image_id": brand["brand_image_id"],
@@ -973,6 +1036,14 @@ def prepare_one_creative_for_ad(
                 "[prepare_one_creative] AI素材状态更新失败 id=%s:%s",
                 chosen_ai_generated_material_id, e,
             )
+    if chosen_external_recalled_material_id:
+        try:
+            update_external_material_status(rec, "prepared")
+        except Exception as e:
+            logger.warning(
+                "[prepare_one_creative] 外部素材状态更新失败 id=%s:%s",
+                chosen_external_recalled_material_id, e,
+            )
     return rec
 
 

examples/auto_put_ad_mini/tools/creative_material_usage.py

@@ -163,6 +163,8 @@ def load_recent_used_landing_video_ids(
 
 def _material_source_from_record(material_id: object, raw_record: object) -> str:
     material_id_text = str(material_id or "").strip()
+    if material_id_text.startswith("external:"):
+        return "external_recall"
     if material_id_text.startswith("ai:"):
         return "ai_generated"
     if raw_record:
@@ -170,6 +172,8 @@ def _material_source_from_record(material_id: object, raw_record: object) -> str
             data = json.loads(str(raw_record))
             source = str(data.get("material_source") or "").strip()
             raw_material_id = str(data.get("_material_id") or "").strip()
+            if source == "external_recall" or raw_material_id.startswith("external:"):
+                return "external_recall"
             if source == "ai_generated" or raw_material_id.startswith("ai:"):
                 return "ai_generated"
         except Exception:
@@ -189,6 +193,7 @@ def load_recent_landing_usage_counts(
     counts: dict[str, dict[int, int]] = {
         "history": {},
         "ai_generated": {},
+        "external_recall": {},
     }
     if not crowd_package:
         return counts

examples/auto_put_ad_mini/tools/external_recalled_material.py

@@ -0,0 +1,655 @@
+"""External recalled material processing for module B.
+
+This path is enabled only for accounts configured with
+``material_source=external_recall``. It keeps historical recall and the
+existing text-to-image flow unchanged.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any, Iterable
+
+from tools.ai_generated_material import (
+    OPENROUTER_IMAGE_MODEL,
+    build_ai_image_object_key,
+    generate_image_bytes,
+    upload_image_to_oss,
+)
+from tools.ai_material_review import MaterialReviewResult, review_generated_material
+from tools.material_recall import Material, recall_materials_for_video
+from tools.video_recall import LandingVideo
+
+logger = logging.getLogger(__name__)
+
+EXTERNAL_RECALL_SOURCE_LABEL = os.getenv(
+    "EXTERNAL_RECALL_SOURCE_LABEL", "外部合作"
+).strip()
+EXTERNAL_RECALL_CANDIDATE_LIMIT = int(
+    os.getenv("EXTERNAL_RECALL_CANDIDATE_LIMIT", "300")
+)
+EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING = int(
+    os.getenv("EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING", "3")
+)
+EXTERNAL_RECALL_UV_WINDOW_DAYS = int(
+    os.getenv("EXTERNAL_RECALL_UV_WINDOW_DAYS", "90")
+)
+EXTERNAL_IMAGE_MODEL = os.getenv(
+    "EXTERNAL_IMAGE_MODEL", OPENROUTER_IMAGE_MODEL
+).strip()
+EXTERNAL_CLEANUP_PROMPT_PATH = (
+    Path(__file__).resolve().parents[1] / "prompts" / "external_material_cleanup.md"
+)
+
+
+CREATE_EXTERNAL_MATERIAL_TABLE_SQL = """
+CREATE TABLE IF NOT EXISTS external_recalled_material (
+    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '自增主键',
+    account_id BIGINT NOT NULL COMMENT '腾讯广告账户ID',
+    adgroup_id BIGINT NOT NULL COMMENT '广告ID',
+    crowd_package VARCHAR(200) NOT NULL COMMENT '投放人群包名称',
+    landing_video_id BIGINT NOT NULL COMMENT '承接视频ID',
+    source_material_id VARCHAR(255) NOT NULL COMMENT '外部召回materialId/card_cover_id',
+    source_image_url VARCHAR(1000) NOT NULL COMMENT '外部素材原图URL',
+    source_title VARCHAR(1000) DEFAULT NULL COMMENT '外部素材标题',
+    similarity_score DOUBLE NOT NULL DEFAULT 0 COMMENT '向量召回相似度',
+    visit_uv_30d BIGINT NOT NULL DEFAULT 0 COMMENT '配置窗口内访问UV之和(兼容历史字段名)',
+    uv_window_start VARCHAR(8) DEFAULT NULL COMMENT 'UV窗口开始分区',
+    uv_window_end VARCHAR(8) DEFAULT NULL COMMENT 'UV窗口结束分区',
+    recall_json MEDIUMTEXT DEFAULT NULL COMMENT '召回命中与原始素材JSON',
+    edit_prompt MEDIUMTEXT NOT NULL COMMENT '去播放按钮图生图prompt',
+    model VARCHAR(200) NOT NULL COMMENT '图生图模型',
+    oss_object_key VARCHAR(500) DEFAULT NULL COMMENT '派生图OSS key',
+    oss_url VARCHAR(1000) DEFAULT NULL COMMENT '派生图公网URL',
+    status VARCHAR(50) NOT NULL DEFAULT 'generated' COMMENT 'generated/prepared/approved/rejected/hold/skip/posted_ok/post_failed/error',
+    approval_status VARCHAR(50) DEFAULT NULL COMMENT '人工审批状态',
+    tencent_image_id VARCHAR(100) DEFAULT NULL COMMENT '腾讯图片ID',
+    dynamic_creative_id BIGINT DEFAULT NULL COMMENT '腾讯动态创意ID',
+    ai_review_status VARCHAR(50) DEFAULT NULL COMMENT 'AI审核状态',
+    ai_review_score INT DEFAULT NULL COMMENT 'AI审核评分',
+    ai_review_reason TEXT DEFAULT NULL COMMENT 'AI审核原因',
+    ai_review_json MEDIUMTEXT DEFAULT NULL COMMENT 'AI审核JSON',
+    error TEXT DEFAULT NULL COMMENT '处理错误',
+    raw_response MEDIUMTEXT DEFAULT NULL COMMENT '图生图模型响应摘要',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+    UNIQUE KEY uk_account_ad_landing_source
+        (account_id, adgroup_id, landing_video_id, source_material_id),
+    KEY idx_account_ad_status (account_id, adgroup_id, status),
+    KEY idx_source_created (source_material_id, created_at),
+    KEY idx_status_created (status, created_at)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='外部召回素材AI清理派生图'
+"""
+
+
+@dataclass(frozen=True)
+class ExternalMaterialAsset:
+    id: int
+    source_material_id: str
+    source_image_url: str
+    source_title: str
+    similarity_score: float
+    visit_uv_30d: int
+    uv_window_start: str
+    uv_window_end: str
+    oss_url: str
+    recall_json: dict[str, Any]
+
+    def to_material(self) -> Material:
+        recall = self.recall_json
+        return Material(
+            material_id=f"external:{self.source_material_id}",
+            score=self.similarity_score,
+            title=self.source_title,
+            cover=self.oss_url,
+            cost=None,
+            ctr=None,
+            cvr=None,
+            roi=None,
+            impressions=None,
+            quality_score=None,
+            visit_uv_30d=self.visit_uv_30d,
+            recall_strategy=str(recall.get("recall_strategy") or ""),
+            recall_query_text=str(recall.get("recall_query_text") or ""),
+            recall_config_code=str(recall.get("recall_config_code") or ""),
+            recall_element_dimension=str(
+                recall.get("recall_element_dimension") or ""
+            ),
+            recall_point_type=str(recall.get("recall_point_type") or ""),
+            recall_standard_element=str(
+                recall.get("recall_standard_element") or ""
+            ),
+            recall_hit_queries=list(recall.get("recall_hit_queries") or []),
+            raw={
+                "external_recalled_material_id": self.id,
+                "source_material_id": self.source_material_id,
+                "source_image_url": self.source_image_url,
+                "visit_uv_30d": self.visit_uv_30d,
+                "uv_window_start": self.uv_window_start,
+                "uv_window_end": self.uv_window_end,
+                "oss_url": self.oss_url,
+            },
+        )
+
+
+def ensure_external_material_table() -> None:
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(CREATE_EXTERNAL_MATERIAL_TABLE_SQL)
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _source_image_url(material: Material) -> str:
+    if material.cover:
+        return material.cover
+    images = (material.raw or {}).get("imageList") or []
+    return str(images[0]) if images else ""
+
+
+def _quote_sql(value: str) -> str:
+    return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"
+
+
+def _get_odps_client():
+    from tools.odps_module import get_odps_client
+
+    client = get_odps_client(project="loghubods")
+    if client is None:
+        raise RuntimeError("无法初始化ODPS客户端")
+    return client
+
+
+def _latest_partition_dt(client) -> str:
+    table = client._odps.get_table("cooperate_top_cards_daily_v1")
+    partition = table.get_max_partition()
+    if partition is None:
+        raise RuntimeError("cooperate_top_cards_daily_v1没有可用分区")
+    name = getattr(partition, "name", "") or str(partition)
+    for part in name.split(","):
+        if part.strip().startswith("dt="):
+            return part.split("=", 1)[1].strip(" '\"")
+    raise RuntimeError(f"无法解析cooperate_top_cards_daily_v1最新分区:{name}")
+
+
+def _query_visit_uv_30d(
+    material_ids: Iterable[str],
+) -> tuple[dict[str, int], str, str]:
+    ids = sorted({str(value).strip() for value in material_ids if str(value).strip()})
+    if not ids:
+        return {}, "", ""
+    client = _get_odps_client()
+    end_dt = _latest_partition_dt(client)
+    end_date = datetime.strptime(end_dt, "%Y%m%d").date()
+    start_dt = (end_date - timedelta(days=EXTERNAL_RECALL_UV_WINDOW_DAYS - 1)).strftime(
+        "%Y%m%d"
+    )
+    totals: dict[str, int] = {}
+    for offset in range(0, len(ids), 500):
+        batch = ids[offset:offset + 500]
+        quoted = ",".join(_quote_sql(value) for value in batch)
+        sql = f"""
+SELECT card_cover_id, SUM(`访问uv`) AS visit_uv_30d
+FROM loghubods.cooperate_top_cards_daily_v1
+WHERE dt BETWEEN '{start_dt}' AND '{end_dt}'
+  AND card_cover_id IN ({quoted})
+GROUP BY card_cover_id
+"""
+        instance = client._odps.execute_sql(sql)
+        instance.wait_for_success()
+        with instance.open_reader(tunnel=False) as reader:
+            for row in reader:
+                totals[str(row[0])] = int(row[1] or 0)
+    return totals, start_dt, end_dt
+
+
+def rank_external_materials_by_visit_uv(
+    materials: list[Material],
+) -> list[Material]:
+    """Attach configured-window visit UV and sort by UV, then similarity."""
+    if not materials:
+        return []
+    totals, start_dt, end_dt = _query_visit_uv_30d(
+        material.material_id for material in materials
+    )
+    for material in materials:
+        material.visit_uv_30d = totals.get(material.material_id, 0)
+        material.raw["visit_uv_30d"] = material.visit_uv_30d
+        material.raw["uv_window_start"] = start_dt
+        material.raw["uv_window_end"] = end_dt
+    ranked = sorted(
+        materials,
+        key=lambda material: (material.visit_uv_30d or 0, material.score or 0),
+        reverse=True,
+    )
+    logger.info(
+        "[external_material] UV排序 candidates=%d days=%d window=%s-%s top=%s",
+        len(ranked), EXTERNAL_RECALL_UV_WINDOW_DAYS, start_dt, end_dt,
+        [
+            (material.material_id, material.visit_uv_30d, round(material.score, 4))
+            for material in ranked[:5]
+        ],
+    )
+    return ranked
+
+
+def recall_external_materials_for_video(
+    landing: LandingVideo,
+    *,
+    element_features: Iterable,
+) -> list[Material]:
+    materials = recall_materials_for_video(
+        landing,
+        final_top_n=None,
+        source_labels=[EXTERNAL_RECALL_SOURCE_LABEL],
+        element_features=element_features,
+        apply_cover_blacklist=False,
+    )
+    ranked = rank_external_materials_by_visit_uv(materials)
+    return ranked[:max(1, EXTERNAL_RECALL_CANDIDATE_LIMIT)]
+
+
+def _load_cleanup_prompt() -> str:
+    text = EXTERNAL_CLEANUP_PROMPT_PATH.read_text(encoding="utf-8").strip()
+    if not text:
+        raise RuntimeError(f"外部素材清理prompt为空:{EXTERNAL_CLEANUP_PROMPT_PATH}")
+    return text
+
+
+def _recall_json(material: Material) -> dict[str, Any]:
+    return {
+        "recall_strategy": material.recall_strategy,
+        "recall_query_text": material.recall_query_text,
+        "recall_config_code": material.recall_config_code,
+        "recall_element_dimension": material.recall_element_dimension,
+        "recall_point_type": material.recall_point_type,
+        "recall_standard_element": material.recall_standard_element,
+        "recall_hit_queries": material.recall_hit_queries,
+        "source_raw": material.raw,
+    }
+
+
+def _asset_from_row(row: dict[str, Any]) -> ExternalMaterialAsset:
+    try:
+        recall = json.loads(row.get("recall_json") or "{}")
+    except Exception:
+        recall = {}
+    return ExternalMaterialAsset(
+        id=int(row["id"]),
+        source_material_id=str(row["source_material_id"]),
+        source_image_url=str(row.get("source_image_url") or ""),
+        source_title=str(row.get("source_title") or ""),
+        similarity_score=float(row.get("similarity_score") or 0),
+        visit_uv_30d=int(row.get("visit_uv_30d") or 0),
+        uv_window_start=str(row.get("uv_window_start") or ""),
+        uv_window_end=str(row.get("uv_window_end") or ""),
+        oss_url=str(row.get("oss_url") or ""),
+        recall_json=recall,
+    )
+
+
+def _load_existing_row(
+    account_id: int,
+    adgroup_id: int,
+    landing_video_id: int,
+    source_material_id: str,
+) -> dict[str, Any] | None:
+    ensure_external_material_table()
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+SELECT *
+FROM external_recalled_material
+WHERE account_id=%s AND adgroup_id=%s AND landing_video_id=%s
+  AND source_material_id=%s
+LIMIT 1
+""",
+                (account_id, adgroup_id, landing_video_id, source_material_id),
+            )
+            return cur.fetchone()
+    finally:
+        conn.close()
+
+
+def _insert_generated_asset(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    crowd_package: str,
+    landing: LandingVideo,
+    material: Material,
+    source_image_url: str,
+    prompt: str,
+    object_key: str,
+    oss_url: str,
+    raw_response: dict[str, Any],
+) -> int:
+    ensure_external_material_table()
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+INSERT INTO external_recalled_material
+    (account_id, adgroup_id, crowd_package, landing_video_id,
+     source_material_id, source_image_url, source_title, similarity_score,
+     visit_uv_30d, uv_window_start, uv_window_end, recall_json,
+     edit_prompt, model, oss_object_key, oss_url, raw_response)
+VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
+""",
+                (
+                    account_id,
+                    adgroup_id,
+                    crowd_package,
+                    landing.video_id,
+                    material.material_id,
+                    source_image_url,
+                    material.title,
+                    material.score,
+                    int(material.visit_uv_30d or 0),
+                    str(material.raw.get("uv_window_start") or ""),
+                    str(material.raw.get("uv_window_end") or ""),
+                    json.dumps(_recall_json(material), ensure_ascii=False, default=str),
+                    prompt,
+                    EXTERNAL_IMAGE_MODEL,
+                    object_key,
+                    oss_url,
+                    json.dumps(raw_response, ensure_ascii=False, default=str)[:16000000],
+                ),
+            )
+            asset_id = int(cur.lastrowid)
+        conn.commit()
+        return asset_id
+    finally:
+        conn.close()
+
+
+def _update_review(asset_id: int, review) -> None:
+    from db.connection import get_connection
+
+    status = "generated" if review.status == "pass" else review.status
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+UPDATE external_recalled_material
+SET status=%s, ai_review_status=%s, ai_review_score=%s,
+    ai_review_reason=%s, ai_review_json=%s, error=%s,
+    updated_at=CURRENT_TIMESTAMP
+WHERE id=%s
+""",
+                (
+                    status,
+                    review.status,
+                    review.score,
+                    review.reason,
+                    json.dumps(review.raw, ensure_ascii=False, default=str),
+                    review.reason if review.status == "error" else None,
+                    asset_id,
+                ),
+            )
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def _review_asset(
+    *,
+    asset_id: int,
+    oss_url: str,
+    prompt: str,
+    landing: LandingVideo,
+    material: Material,
+):
+    try:
+        review = review_generated_material(
+            image_url=oss_url,
+            prompt_type="external_cleanup",
+            prompt_text=prompt,
+            feature_hits=[
+                {
+                    "landing_video_id": landing.video_id,
+                    "landing_title": landing.title,
+                    "source_material_id": material.material_id,
+                    "source_title": material.title,
+                    "visit_uv_30d": int(material.visit_uv_30d or 0),
+                    "recall_hit_queries": material.recall_hit_queries,
+                }
+            ],
+        )
+    except Exception as exc:
+        logger.exception(
+            "[external_material] AI审核异常 asset=%d source=%s:%s",
+            asset_id, material.material_id, exc,
+        )
+        review = MaterialReviewResult(
+            status="error",
+            score=0,
+            reason=f"AI审核异常:{exc}",
+            risk_tags=["review_error"],
+            ocr_text="",
+            raw={"error": str(exc)},
+        )
+    _update_review(asset_id, review)
+    return review
+
+
+def get_or_create_external_asset(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    crowd_package: str,
+    landing: LandingVideo,
+    material: Material,
+) -> Material | None:
+    existing = _load_existing_row(
+        account_id,
+        adgroup_id,
+        landing.video_id,
+        material.material_id,
+    )
+    if existing:
+        if (
+            existing.get("oss_url")
+            and existing.get("status") in {"generated", "error"}
+            and existing.get("ai_review_status") in {None, "", "error"}
+        ):
+            review = _review_asset(
+                asset_id=int(existing["id"]),
+                oss_url=str(existing["oss_url"]),
+                prompt=str(existing.get("edit_prompt") or _load_cleanup_prompt()),
+                landing=landing,
+                material=material,
+            )
+            existing = _load_existing_row(
+                account_id,
+                adgroup_id,
+                landing.video_id,
+                material.material_id,
+            )
+            if review.passed and existing:
+                return _asset_from_row(existing).to_material()
+        if (
+            existing.get("ai_review_status") == "pass"
+            and existing.get("status") in {"generated", "prepared"}
+            and existing.get("oss_url")
+        ):
+            logger.info(
+                "[external_material] 复用派生图 account=%d adgroup=%d landing=%d source=%s id=%s",
+                account_id, adgroup_id, landing.video_id, material.material_id,
+                existing["id"],
+            )
+            return _asset_from_row(existing).to_material()
+        logger.info(
+            "[external_material] 已有不可用派生记录 source=%s status=%s review=%s",
+            material.material_id, existing.get("status"),
+            existing.get("ai_review_status"),
+        )
+        return None
+
+    source_image_url = _source_image_url(material)
+    if not source_image_url:
+        logger.warning(
+            "[external_material] source=%s 没有可编辑图片URL", material.material_id
+        )
+        return None
+
+    prompt = _load_cleanup_prompt()
+    image_bytes, content_type, raw_response = generate_image_bytes(
+        prompt,
+        model=EXTERNAL_IMAGE_MODEL,
+        input_image_url=source_image_url,
+    )
+    source_hash = hashlib.sha256(material.material_id.encode("utf-8")).hexdigest()[:12]
+    object_key = build_ai_image_object_key(
+        account_id=account_id,
+        landing_video_id=landing.video_id,
+        prompt_type=f"external_cleanup_{source_hash}",
+        extension="jpg",
+    )
+    oss_url = upload_image_to_oss(image_bytes, content_type, object_key)
+    asset_id = _insert_generated_asset(
+        account_id=account_id,
+        adgroup_id=adgroup_id,
+        crowd_package=crowd_package,
+        landing=landing,
+        material=material,
+        source_image_url=source_image_url,
+        prompt=prompt,
+        object_key=object_key,
+        oss_url=oss_url,
+        raw_response=raw_response,
+    )
+    review = _review_asset(
+        asset_id=asset_id,
+        oss_url=oss_url,
+        prompt=prompt,
+        landing=landing,
+        material=material,
+    )
+    logger.info(
+        "[external_material] 派生图完成 account=%d adgroup=%d landing=%d source=%s "
+        "asset=%d uv_window=%d review=%s score=%d url=%s",
+        account_id, adgroup_id, landing.video_id, material.material_id,
+        asset_id, int(material.visit_uv_30d or 0), review.status, review.score, oss_url,
+    )
+    if not review.passed:
+        return None
+    row = _load_existing_row(
+        account_id,
+        adgroup_id,
+        landing.video_id,
+        material.material_id,
+    )
+    return _asset_from_row(row).to_material() if row else None
+
+
+def prepare_external_material_for_landing(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    crowd_package: str,
+    landing: LandingVideo,
+    element_features: Iterable,
+    excluded_material_ids: set[str],
+) -> Material | None:
+    ranked = recall_external_materials_for_video(
+        landing,
+        element_features=element_features,
+    )
+    attempted = 0
+    for material in ranked:
+        external_id = f"external:{material.material_id}"
+        if external_id in excluded_material_ids:
+            continue
+        attempted += 1
+        try:
+            processed = get_or_create_external_asset(
+                account_id=account_id,
+                adgroup_id=adgroup_id,
+                crowd_package=crowd_package,
+                landing=landing,
+                material=material,
+            )
+        except Exception as exc:
+            logger.exception(
+                "[external_material] 图生图失败 landing=%d source=%s:%s",
+                landing.video_id, material.material_id, exc,
+            )
+            processed = None
+        if processed is not None:
+            return processed
+        if attempted >= max(1, EXTERNAL_RECALL_EDIT_LIMIT_PER_LANDING):
+            break
+    logger.info(
+        "[external_material] landing=%d 无可用派生素材 recalled=%d attempted=%d",
+        landing.video_id, len(ranked), attempted,
+    )
+    return None
+
+
+def update_external_material_status(
+    record: dict,
+    status: str,
+    *,
+    dynamic_creative_id: int | str | None = None,
+    tencent_image_id: str = "",
+    error: str = "",
+) -> None:
+    asset_id = record.get("_external_recalled_material_id")
+    if not asset_id:
+        return
+    ensure_external_material_table()
+    from db.connection import get_connection
+
+    normalized_status = {
+        "approve": "approved",
+        "reject": "rejected",
+    }.get(status, status)
+    approval_status = {
+        "approve": "approved",
+        "reject": "rejected",
+        "hold": "hold",
+        "skip": "skip",
+    }.get(status, normalized_status)
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+UPDATE external_recalled_material
+SET status=%s, approval_status=%s,
+    dynamic_creative_id=COALESCE(%s, dynamic_creative_id),
+    tencent_image_id=COALESCE(NULLIF(%s, ''), tencent_image_id),
+    error=COALESCE(NULLIF(%s, ''), error), updated_at=CURRENT_TIMESTAMP
+WHERE id=%s
+""",
+                (
+                    normalized_status,
+                    approval_status,
+                    int(dynamic_creative_id) if dynamic_creative_id else None,
+                    tencent_image_id,
+                    error[:2000] if error else "",
+                    int(asset_id),
+                ),
+            )
+        conn.commit()
+    finally:
+        conn.close()

examples/auto_put_ad_mini/tools/im_approval_creation.py

@@ -2,13 +2,13 @@
 
 数据流:
   Phase 1 准备 pending records (list[dict])
-    → generate_approval_xlsx          生成 20 列 xlsx(hyperlink/下拉/颜色/冻结)
+    → generate_approval_xlsx          生成 30 列 xlsx(hyperlink/下拉/颜色/冻结)
     → send_approval_to_feishu         上传转飞书 sheet + 发链接到群
     → poll_approval_actions           轮询读 sheet "决策"列,拿 approve/reject/hold
     → 调用方按 row_idx 匹配回 records 写决策
 
 设计决策(2026-06-09 用户确认):
-  - 20 列:A(灰)5 + B(紫)3 + C(橙)11 + D(绿)1,header 颜色分组
+  - 30 列:A(灰)5 + B(紫)3 + C(橙)21 + D(绿)1,header 颜色分组
   - 素材预览用 HYPERLINK 公式(=IMAGE() 不被飞书支持,write-images API 留作 P3-5)
   - 落地视频列也用 HYPERLINK(点 video_id 跳转 video_url)
   - 决策列下拉:approve / reject / hold
@@ -41,16 +41,20 @@ from config import (
 logger = logging.getLogger(__name__)
 
 FEISHU_BASE_URL = "https://open.feishu.cn/open-apis"
+EXTERNAL_RECALL_UV_WINDOW_DAYS = int(
+    os.getenv("EXTERNAL_RECALL_UV_WINDOW_DAYS", "90")
+)
 
-# 28 列(中文)— 素材排序改为 score 准入 + cost 倒序,报表同步展示召回依据。
+# 30 列(中文)— 外部素材额外展示原图与配置窗口访问UV。
 HEADERS = [
     # A 浅灰 5 列
     "日期", "账户ID", "人群包", "广告ID", "广告名称",
     # B 浅紫 3 列
     "出价(元)", "投放版位", "年龄定向",
-    # C 浅橙 19 列(落地视频 + 风险审核 + 素材来源 + 素材质量)
+    # C 浅橙 21 列(落地视频 + 风险审核 + 素材来源 + 素材质量)
     "落地视频", "落地视频标题", "风险等级", "风险标签", "风险原因",
-    "素材来源", "素材预览", "素材链接", "创意文案",
+    "素材来源", "素材预览", "素材链接", "原始素材",
+    f"{EXTERNAL_RECALL_UV_WINDOW_DAYS}天访问UV", "创意文案",
     "成本(元)", "ROI", "CTR", "曝光数", "相似度",
     "召回维度", "召回点类型", "召回元素", "命中维度明细",
     "创意名(归因)",
@@ -61,22 +65,22 @@ HEADERS = [
 GROUP_COLORS = [
     ((1, 5), "FFD9D9D9"),    # A 浅灰
     ((6, 8), "FFD9C8E8"),    # B 浅紫
-    ((9, 27), "FFFCD8B4"),   # C 浅橙
-    ((28, 28), "FFC6E0B4"),  # D 浅绿
+    ((9, 29), "FFFCD8B4"),   # C 浅橙
+    ((30, 30), "FFC6E0B4"),  # D 浅绿
 ]
 
 COL_WIDTHS = {
     "A": 12, "B": 14, "C": 16, "D": 16, "E": 28,
     "F": 10, "G": 28, "H": 12,
     "I": 14, "J": 26, "K": 10, "L": 24, "M": 32,
-    "N": 14, "O": 18, "P": 24, "Q": 30,
-    "R": 12, "S": 10, "T": 10, "U": 10, "V": 10,
-    "W": 14, "X": 14, "Y": 18, "Z": 42,
-    "AA": 38,                     # 创意名(归因)
-    "AB": 14,                     # 决策
+    "N": 14, "O": 18, "P": 24, "Q": 18, "R": 14, "S": 30,
+    "T": 12, "U": 10, "V": 10, "W": 10, "X": 10,
+    "Y": 14, "Z": 14, "AA": 18, "AB": 42,
+    "AC": 38,                     # 创意名(归因)
+    "AD": 14,                     # 决策
 }
 
-DECISION_COL_LETTER = "AB"
+DECISION_COL_LETTER = "AD"
 VALID_ACTIONS = ("approve", "reject", "hold")
 EMBED_MATERIAL_PREVIEW_IMAGES = os.getenv(
     "CREATION_APPROVAL_EMBED_IMAGES", "1"
@@ -133,6 +137,14 @@ def _format_record_to_row(rec: dict) -> list:
         # 素材预览:HYPERLINK(cover_url, "查看素材")— 见模块顶部说明
         f'=HYPERLINK("{rec["material_cover_url"]}","查看素材")',
         f'=HYPERLINK("{rec["material_cover_url"]}","打开素材")',
+        (
+            f'=HYPERLINK("{rec["_external_source_image_url"]}","打开原图")'
+            if rec.get("_external_source_image_url") else ""
+        ),
+        (
+            str(int(rec["material_visit_uv_30d"]))
+            if rec.get("material_visit_uv_30d") is not None else ""
+        ),
         # 创意文案(2026-06-09 加列):description 换行展示
         descriptions_str,
         f"{cost:.2f}" if cost is not None else "",
@@ -320,7 +332,8 @@ async def send_approval_to_feishu(
         preamble = (
             "【创意搭建审批】下方在线表格列出本次准备挂的创意。\n"
             "系统已在创建落地计划前过滤高风险视频,表格中的【风险等级/风险标签/风险原因】为通过项的审核摘要。\n"
-            "素材按相似度≥0.8准入,再按历史消耗倒序选择;ROI/CTR/曝光仅供审批参考。\n"
+            "历史和外部召回素材按相似度≥0.8准入;历史素材按消耗倒序,"
+            f"外部素材按最近{EXTERNAL_RECALL_UV_WINDOW_DAYS}天访问UV倒序并经AI清理。\n"
             "请在最右侧【决策】列下拉框选择 approve / reject / hold。\n"
             "我们会在 2 小时内或所有行决策完成后,执行 approve 项,跳过 reject / hold。\n"
             "提示:点击【素材预览】或【落地视频】可在浏览器打开查看(若 403 请右键新标签或复制 URL)。"

examples/auto_put_ad_mini/tools/material_recall.py

@@ -157,6 +157,7 @@ class Material:
     roi: Optional[float] = None               # ROI(收入/成本)
     impressions: Optional[int] = None         # 累计曝光
     quality_score: Optional[float] = None     # 服务端 qualityScore
+    visit_uv_30d: Optional[int] = None        # 外部素材配置窗口内访问UV之和（列名兼容历史）
     recall_strategy: str = ""
     recall_query_text: str = ""
     recall_config_code: str = ""
@@ -358,7 +359,12 @@ def _as_int(value, default: int = 0) -> int:
         return default
 
 
-def _items_to_materials(items: List[dict], sim_threshold: float) -> tuple:
+def _items_to_materials(
+    items: List[dict],
+    sim_threshold: float,
+    *,
+    apply_cover_blacklist: bool = True,
+) -> tuple:
     """把召回 items 过滤 + 转 Material。
 
     过滤条件:
@@ -383,7 +389,9 @@ def _items_to_materials(items: List[dict], sim_threshold: float) -> tuple:
         if not mid:
             continue
         cover = it.get("cover") or ""
-        if any(p in cover for p in EXCLUDED_COVER_URL_PATTERNS):
+        if apply_cover_blacklist and any(
+            pattern in cover for pattern in EXCLUDED_COVER_URL_PATTERNS
+        ):
             stats["blacklist"] += 1
             continue
         score = _as_float(it.get("score"))
@@ -557,9 +565,10 @@ def _merge_materials_by_policy(query_materials: List[tuple[RecallQuery, List[Mat
 
 def recall_materials_for_video(
     landing: LandingVideo,
-    final_top_n: int = DEFAULT_FINAL_TOP_N,
+    final_top_n: Optional[int] = DEFAULT_FINAL_TOP_N,
     source_labels: Optional[List[str]] = None,
     element_features: Optional[Iterable] = None,
+    apply_cover_blacklist: bool = True,
 ) -> List[Material]:
     """素材召回:用 ODPS 多维特征并行召回并合并排序。
 
@@ -625,7 +634,11 @@ def recall_materials_for_video(
             except Exception as e:
                 logger.error("[material_recall] 并行 batchByText 失败:%s", e)
                 continue
-            mats, stats = _items_to_materials(items, RECALL_SIM_THRESHOLD)
+            mats, stats = _items_to_materials(
+                items,
+                RECALL_SIM_THRESHOLD,
+                apply_cover_blacklist=apply_cover_blacklist,
+            )
             query_materials.append((query, mats))
             logger.info(
                 "[material_recall]   策略=%s q=%r configCode=%s 返回 %d 条,⊘ 黑名单 %d,⊘ score<%.2f %d → 保留 %d",
@@ -640,7 +653,7 @@ def recall_materials_for_video(
         landing.video_id, len(merged),
     )
     if merged:
-        return merged[:final_top_n]
+        return merged if final_top_n is None else merged[:final_top_n]
 
     for query in queries:
         # batchByText 当前可能对 MATERIAL 返回 0;降级到历史 matchByText 路径。
@@ -659,7 +672,11 @@ def recall_materials_for_video(
                 )
                 continue
 
-            fmats, fstats = _items_to_materials(fallback_items, RECALL_SIM_THRESHOLD)
+            fmats, fstats = _items_to_materials(
+                fallback_items,
+                RECALL_SIM_THRESHOLD,
+                apply_cover_blacklist=apply_cover_blacklist,
+            )
             fquery = RecallQuery(
                 query_text=query.query_text,
                 config_code=fallback_cc,
@@ -677,7 +694,7 @@ def recall_materials_for_video(
                 len(fmats),
             )
             if fmats:
-                return fmats[:final_top_n]
+                return fmats if final_top_n is None else fmats[:final_top_n]
 
     logger.info(
         "[material_recall] landing video_id=%d 所有 %d ODPS 策略全失败,返回空(上层换 landing)",