Improve creative run recovery

AGENTS.md

@@ -49,7 +49,8 @@
 - 素材通过相似度筛选后,默认按历史消耗 `cost` 倒序选择。
 - 落地页视频去重按素材来源拆池。历史素材链路同一 `crowd_package + landing_video_id` 默认最多 1 次。
 - AI 生成素材链路和历史素材链路互不占用 landing 去重名额;AI 链路同一 `crowd_package + landing_video_id` 默认最多 1 次。
-- 落地页视频还需要做跨轮近期排重:默认按同人群包下最近 7 天进入 pending 或已提交腾讯的 `landing_video_id` 统计使用次数。
+- 落地页视频跨轮近期排重只能基于已提交腾讯、已执行或已有明确投放结果的记录。
+- 未提交腾讯的 `prepared` / `pending` 记录不能进入跨轮排重;它们只能用于当前运行内去重,或作为中断后可复用的恢复缓存。
 - 同一广告内同一 `landing_video_id` 仍保留限频保护,默认最多 1 次。
 - 素材需要跨账户/跨天排重,默认按同人群包下近期使用过的 `material_id` 排除。
 - 当前运行内的审批候选需要做轻量展示去重:同一 `crowd_package + material_id` 只进入本轮审批表一次;该去重只存在内存,不写入历史排重库。

examples/auto_put_ad_mini/CREATIVE_CREATION_TODO.md

@@ -118,6 +118,15 @@
 - 风险点:飞书 server 拉图是否被 yishihui CDN 拦截待验证;如被拦,需跟 yishihui 运维协调 CDN 白名单(加 feishu.cn 或放行空 Referer)
 - 触发条件:挂创意量上百条/天时再升级 — 当前 ≤ 5 条/天,运营 5 次点击代价 < 工程协调成本
 
+### P3-7. AI 生成素材画面主体多样性优化
+- 现象:近期 AI 生成素材里背景经常是中老年人物,且几乎每张都有人物主体。
+- 初步判断:当前 prompt 多次强调目标用户为中老年、生活场景、家庭/邻里/父母子女关系,图片模型容易把"目标用户"误理解为"画面必须出现老年人"。
+- 优化方向:
+  - 区分"目标用户"和"画面主体":目标用户是 45-75 岁用户,但画面主体可以是物品、清单、场景、事件现场、道具特写、讲解场面、自然/工业过程等。
+  - 不强制人物:只有 pattern 或视频主题确实需要人物情绪/关系时才出现人物。
+  - 不强制老人形象:即使有人物,也可以是中年人、家庭成员、工作人员、邻里、讲解者或背影/手部等,不必总是正脸老人。
+  - 后续评估 prompt 是否过度绑定"真实生活化/家庭场景",避免所有素材都长成同一类家庭老人封面。
+
 ## 已知腾讯错误码 — 排查表
 
 | code | 含义 | 修复方向 |

examples/auto_put_ad_mini/PRODUCTION_AUTOMATION.md

@@ -214,7 +214,8 @@ AI 图默认上传到:
 - 历史素材召回链路:同一 `crowd_package + landing_video_id` 默认最多使用 1 次。
 - AI 生成素材链路:同一 `crowd_package + landing_video_id` 默认最多使用 1 次。
 - 两条链路互不占用名额。历史素材已用过的 landing,不阻止 AI 生成素材链路继续使用;AI 使用记录也不阻止历史素材链路按自己的规则判断。
-- 跨轮近期排重默认读取 `creative_material_usage` 和 `creative_creation_task`,按同人群包最近 7 天进入 pending 或已提交腾讯的 `landing_video_id` 统计使用次数。
+- 跨轮近期排重默认读取 `creative_material_usage` 和 `creative_creation_task`,只按同人群包最近 7 天已提交腾讯、已执行或已有明确投放结果的 `landing_video_id` 统计使用次数。
+- 未提交腾讯的 `prepared` 记录不参与跨轮排重;端到端中断重跑时会优先作为可恢复缓存复用,避免重复选视频和重复生成 AI 图。
 - 同一广告内同一 `landing_video_id` 仍保留限频保护,默认最多 1 次。
 
 可通过环境变量调整同广告落地页视频本轮限频:

examples/auto_put_ad_mini/db/connection.py

@@ -59,6 +59,10 @@ def get_connection():
             "  DB_PORT=3306  # 可选，默认3306"
         )
 
+    connect_timeout = int(os.getenv("DB_CONNECT_TIMEOUT", "10"))
+    read_timeout = int(os.getenv("DB_READ_TIMEOUT", "60"))
+    write_timeout = int(os.getenv("DB_WRITE_TIMEOUT", "60"))
+
     try:
         conn = pymysql.connect(
             host=host,
@@ -69,6 +73,9 @@ def get_connection():
             charset="utf8mb4",
             cursorclass=pymysql.cursors.DictCursor,  # 返回字典格式
             autocommit=True,  # 自动提交（简化事务管理）
+            connect_timeout=connect_timeout,
+            read_timeout=read_timeout,
+            write_timeout=write_timeout,
         )
         logger.debug(f"数据库连接成功: {user}@{host}:{port}/{database}")
         return conn

examples/auto_put_ad_mini/execute_creation_once.py

@@ -66,6 +66,7 @@ from tools.creative_creation import (  # noqa: E402
 )
 from tools.creative_material_usage import (  # noqa: E402
     load_recent_landing_usage_counts,
+    load_recoverable_prepared_records,
     record_prepared_material_usage,
 )
 from tools.account_material_strategy import (  # noqa: E402
@@ -460,6 +461,63 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
     # landing_video 按素材来源拆池。历史素材和 AI 生成素材互不占用,但池内都严格去重。
     landing_usage_counts_by_crowd: dict[str, dict[str, dict[int, int]]] = {}
 
+    def accept_pending_record(
+        rec: dict,
+        *,
+        crowd_package: str,
+        display_excluded_material_ids: set[str],
+        crowd_landing_counts: dict[str, dict[int, int]],
+        landing_counts_for_ad: dict[int, int],
+        recovered: bool = False,
+    ) -> None:
+        pending_records.append(rec)
+        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%s",
+                crowd_package, material_id,
+                len(display_excluded_material_ids),
+                " recovered" if recovered else "",
+            )
+        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
+            )
+            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
+            )
+            source_counts = crowd_landing_counts.setdefault(actual_material_source, {})
+            source_counts[landing_video_id] = source_counts.get(landing_video_id, 0) + 1
+            logger.info(
+                "[phase1] 同人群包 landing 使用登记 crowd=%r material_source=%s landing=%d count=%d max=%d%s",
+                crowd_package, actual_material_source, landing_video_id,
+                source_counts[landing_video_id],
+                MAX_SAME_LANDING_PER_AD_IN_RUN,
+                " recovered" if recovered else "",
+            )
+            logger.info(
+                "[phase1] 本轮同广告 landing 计数 adgroup=%s landing=%d count=%d limit=%d%s",
+                rec.get("adgroup_id"), landing_video_id,
+                landing_counts_for_ad[landing_video_id],
+                MAX_SAME_LANDING_PER_AD_IN_RUN,
+                " recovered" if recovered else "",
+            )
+        if not recovered:
+            try:
+                record_prepared_material_usage(rec)
+            except Exception as e:
+                logger.warning(
+                    "[phase1] material usage 记录失败 account=%s adgroup=%s material=%s:%s",
+                    rec.get("account_id"), rec.get("adgroup_id"),
+                    rec.get("_material_id"), e,
+                )
+
     for account_id in creation_accounts:
         logger.info("=" * 60)
         logger.info("[phase1] 账户 %d 处理开始", account_id)
@@ -547,8 +605,30 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
                 "[phase1]   adgroup=%d(have=%d need=%d)",
                 adgroup_id, already_have, to_add,
             )
+            recovered_records = load_recoverable_prepared_records(
+                account_id=account_id,
+                adgroup_id=adgroup_id,
+                crowd_package=crowd_package,
+                material_source=requested_material_source,
+                limit=to_add,
+            )
+            for rec in recovered_records:
+                accept_pending_record(
+                    rec,
+                    crowd_package=crowd_package,
+                    display_excluded_material_ids=display_excluded_material_ids,
+                    crowd_landing_counts=crowd_landing_counts,
+                    landing_counts_for_ad=landing_counts_for_ad,
+                    recovered=True,
+                )
+            if recovered_records:
+                prepared_for_ad += len(recovered_records)
+                logger.info(
+                    "[phase1]   adgroup=%d 恢复未提交 prepared records=%d remaining=%d",
+                    adgroup_id, len(recovered_records), max(0, to_add - prepared_for_ad),
+                )
 
-            for _ in range(to_add):
+            for _ in range(max(0, to_add - prepared_for_ad)):
                 landing_excluded_for_ad = {
                     vid
                     for vid, count in landing_counts_for_ad.items()
@@ -578,52 +658,13 @@ def phase1_prepare(target_creatives: int = TARGET_CREATIVES_PER_AD) -> list[dict
 
                 if rec:
                     prepared_for_ad += 1
-                    pending_records.append(rec)
-                    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
-                        )
-                        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
-                        )
-                        source_counts = crowd_landing_counts.setdefault(
-                            actual_material_source, {},
-                        )
-                        source_counts[landing_video_id] = (
-                            source_counts.get(landing_video_id, 0) + 1
-                        )
-                        logger.info(
-                            "[phase1] 同人群包 landing 使用登记 crowd=%r material_source=%s landing=%d count=%d max=%d",
-                            crowd_package, actual_material_source, landing_video_id,
-                            source_counts[landing_video_id],
-                            MAX_SAME_LANDING_PER_AD_IN_RUN,
-                        )
-                        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,
-                        )
+                    accept_pending_record(
+                        rec,
+                        crowd_package=crowd_package,
+                        display_excluded_material_ids=display_excluded_material_ids,
+                        crowd_landing_counts=crowd_landing_counts,
+                        landing_counts_for_ad=landing_counts_for_ad,
+                    )
                 else:
                     failed_prepare_for_ad += 1
                     # 2026-06-10 用户要求:单条 prepare 失败 → continue 不 break

examples/auto_put_ad_mini/tools/ai_generated_material.py

@@ -963,7 +963,7 @@ def load_available_generated_assets(
                 WHERE account_id=%s
                   AND adgroup_id=%s
                   AND landing_video_id=%s
-                  AND status='generated'
+                  AND status IN ('generated', 'prepared')
                   AND ai_review_status='pass'
                   AND oss_url IS NOT NULL
                   AND oss_url <> ''

examples/auto_put_ad_mini/tools/creative_material_usage.py

@@ -19,6 +19,9 @@ CREATIVE_MATERIAL_DEDUPE_LOOKBACK_DAYS = int(
 CREATIVE_LANDING_DEDUPE_LOOKBACK_DAYS = int(
     os.getenv("CREATIVE_LANDING_DEDUPE_LOOKBACK_DAYS", "7")
 )
+CREATION_RECOVERY_LOOKBACK_HOURS = int(
+    os.getenv("CREATION_RECOVERY_LOOKBACK_HOURS", "24")
+)
 
 CREATE_USAGE_TABLE_SQL = """
 CREATE TABLE IF NOT EXISTS creative_material_usage (
@@ -130,6 +133,7 @@ def load_recent_used_landing_video_ids(
                     WHERE crowd_package=%s
                       AND landing_video_id IS NOT NULL
                       AND landing_video_id > 0
+                      AND status IN ('posted_ok', 'submitted')
                       AND created_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
 
                     UNION
@@ -201,6 +205,7 @@ def load_recent_landing_usage_counts(
                 WHERE crowd_package=%s
                   AND landing_video_id IS NOT NULL
                   AND landing_video_id > 0
+                  AND status IN ('posted_ok', 'submitted')
                   AND created_at >= DATE_SUB(NOW(), INTERVAL %s DAY)
                 """,
                 (crowd_package, int(lookback_days)),
@@ -241,6 +246,72 @@ def load_recent_landing_usage_counts(
     return counts
 
 
+def load_recoverable_prepared_records(
+    *,
+    account_id: int,
+    adgroup_id: int,
+    crowd_package: str,
+    material_source: str,
+    limit: int,
+    lookback_hours: int = CREATION_RECOVERY_LOOKBACK_HOURS,
+) -> list[dict]:
+    """Load unsubmitted prepared records after an interrupted Phase 1 run."""
+    if limit <= 0 or not crowd_package:
+        return []
+    ensure_usage_table()
+    from db.connection import get_connection
+
+    conn = get_connection()
+    try:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT raw_record
+                FROM creative_material_usage
+                WHERE account_id=%s
+                  AND adgroup_id <=> %s
+                  AND crowd_package=%s
+                  AND status='prepared'
+                  AND dynamic_creative_id IS NULL
+                  AND raw_record IS NOT NULL
+                  AND raw_record <> ''
+                  AND created_at >= DATE_SUB(NOW(), INTERVAL %s HOUR)
+                ORDER BY id ASC
+                LIMIT %s
+                """,
+                (
+                    int(account_id),
+                    int(adgroup_id) if adgroup_id else None,
+                    crowd_package,
+                    int(lookback_hours),
+                    max(int(limit) * 3, int(limit)),
+                ),
+            )
+            rows = cur.fetchall() or []
+    finally:
+        conn.close()
+
+    out: list[dict] = []
+    seen_materials: set[str] = set()
+    for row in rows:
+        try:
+            record = json.loads(str(row.get("raw_record") or ""))
+        except Exception:
+            continue
+        material_id = str(record.get("_material_id") or "").strip()
+        if not material_id or material_id in seen_materials:
+            continue
+        if _material_source_from_record(material_id, row.get("raw_record")) != material_source:
+            continue
+        if not record.get("_request_body"):
+            continue
+        seen_materials.add(material_id)
+        out.append(record)
+        if len(out) >= limit:
+            break
+    return out
+
+
 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()