fix(viral-video): 修复「重新生成文案」409 冲突问题 (#2222)

- generate-copy 接口允许 COPY_GENERATED/COMPLETED 状态重新触发文案生成
- resume_from_image_analyzed 支持重入：重新生成时清空旧 copy_result/intent_result/storyboard/generated_copy_text/result_video_url，重置 completed_at/error_msg/phase_message
- 重新生成 retry_count 自增，与 FAILED 重试一致
- 首次进入（IMAGE_ANALYZED/PENDING）不清空已有字段
- 仍拒绝 WAIT_USER_CONFIRM 等中间状态
- 新增单测覆盖重入场景

alembic/versions/100_fix_capability_model_bindings.py

@@ -1,107 +0,0 @@
-# -*- coding: utf-8 -*-
-"""100: 修正已有 capability 的模型绑定.
-
-幂等：仅当 primary_model_id 当前绑定到旧模型 (doubao-seed-1-6) 时才更新，
-避免覆盖用户在后台的自定义配置。
-
-- 更新 5 个 LLM capability (intent_parsing, copy_fusion, storyboard, copy_review, asset_classify)
-  的 primary_model_id 从 doubao-seed-1-6 改为 doubao-seed-2-1-pro-260915
-- 更新 image_analysis 的 primary/lite/fallback 模型绑定
-"""
-
-import sqlalchemy as sa
-
-from alembic import op
-
-revision = "100_fix_capability_model_bindings"
-down_revision = "099_ai_model_router_seed"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-
-    # Check tables exist
-    table_check = conn.execute(sa.text("SELECT to_regclass('public.ai_models')")).scalar()
-    if not table_check:
-        return
-
-    config_table_check = conn.execute(sa.text("SELECT to_regclass('public.ai_capability_configs')")).scalar()
-    if not config_table_check:
-        return
-
-    # Look up model IDs by model_key (not hardcoded UUIDs)
-    pro_model_row = conn.execute(
-        sa.text(
-            "SELECT id FROM ai_models WHERE model_key = 'doubao-seed-2-1-pro-260915' AND deleted_at IS NULL LIMIT 1"
-        )
-    ).first()
-    if not pro_model_row:
-        return
-    pro_model_id = pro_model_row[0]
-
-    old_model_row = conn.execute(
-        sa.text("SELECT id FROM ai_models WHERE model_key = 'doubao-seed-1-6-250615' LIMIT 1")
-    ).first()
-    old_model_id = old_model_row[0] if old_model_row else None
-
-    llm_capabilities = [
-        "intent_parsing",
-        "copy_fusion",
-        "storyboard",
-        "copy_review",
-        "asset_classify",
-    ]
-
-    for cap_key in llm_capabilities:
-        if old_model_id:
-            conn.execute(
-                sa.text(
-                    "UPDATE ai_capability_configs SET primary_model_id = :new_id, updated_at = NOW() "
-                    "WHERE capability_key = :cap_key AND primary_model_id = :old_id"
-                ),
-                {"new_id": pro_model_id, "old_id": old_model_id, "cap_key": cap_key},
-            )
-
-    # Update image_analysis
-    qwen38_row = conn.execute(
-        sa.text("SELECT id FROM ai_models WHERE model_key = 'qwen3.8-flash' AND deleted_at IS NULL LIMIT 1")
-    ).first()
-    qwen37_row = conn.execute(
-        sa.text("SELECT id FROM ai_models WHERE model_key = 'qwen3.7-plus' AND deleted_at IS NULL LIMIT 1")
-    ).first()
-
-    if qwen38_row and qwen37_row:
-        qwen38_id = qwen38_row[0]
-        qwen37_id = qwen37_row[0]
-
-        current_ia = conn.execute(
-            sa.text(
-                "SELECT primary_model_id, lite_model_id, fallback_model_id "
-                "FROM ai_capability_configs WHERE capability_key = 'image_analysis'"
-            )
-        ).first()
-
-        if current_ia:
-            current_primary, current_lite, current_fallback = current_ia
-            updates = {}
-            if current_primary != qwen38_id:
-                updates["primary_model_id"] = qwen38_id
-            if current_lite != qwen38_id:
-                updates["lite_model_id"] = qwen38_id
-            if current_fallback != qwen37_id:
-                updates["fallback_model_id"] = qwen37_id
-
-            if updates:
-                set_clause = ", ".join([f"{k} = :{k}" for k in updates.keys()])
-                set_clause += ", updated_at = NOW()"
-                updates["cap_key"] = "image_analysis"
-                conn.execute(
-                    sa.text(f"UPDATE ai_capability_configs SET {set_clause} WHERE capability_key = :cap_key"),
-                    updates,
-                )
-
-
-def downgrade() -> None:
-    pass

alembic/versions/101_qwen_vl_plus_and_max_tokens.py

@@ -1,153 +0,0 @@
-# -*- coding: utf-8 -*-
-"""101: 补齐 qwen-vl-plus 视觉模型并修正 image_analysis 绑定与 max_tokens.
-
-背景：
-- qwen-vl-plus 做图片识别时返回 JSON 约 500-600 tokens，旧硬编码
-  max_tokens=350 导致 JSON 被截断、解析失败返回"未识别"。
-- 代码侧已移除硬编码，改由 capability 的 DB 配置决定 max_tokens。
-
-幂等：
-- qwen-vl-plus 已存在则不插入；
-- 仅当 image_analysis 当前 primary_model 不是 qwen-vl-plus 时才更新绑定，
-  避免覆盖后台手动配置。
-"""
-
-import sqlalchemy as sa
-
-from alembic import op
-
-revision = "101_qwen_vl_plus_and_max_tokens"
-down_revision = "100_fix_capability_model_bindings"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-
-    models_table = conn.execute(sa.text("SELECT to_regclass('public.ai_models')")).scalar()
-    if not models_table:
-        return
-
-    caps_table = conn.execute(sa.text("SELECT to_regclass('public.ai_capability_configs')")).scalar()
-    if not caps_table:
-        return
-
-    # ── c. 补全其他 capability 的 max_tokens 默认值（幂等）──────────────────
-    # 放在 image_analysis 特定逻辑之前，确保任何分支 return 都不会跳过本段。
-    # 仅在当前值为 NULL 或过小 (<100) 时更新，不覆盖已有合理配置。
-    # embedding / tts / voice_clone 不走 chat 接口，无需设置。
-    default_max_tokens = {
-        "intent_parsing": 500,
-        "copy_fusion": 2500,
-        "storyboard": 4000,
-        "copy_review": 1000,
-        "asset_classify": 500,
-        "image_generation": 500,
-        "video_generation": 500,
-    }
-    for cap_key, mt in default_max_tokens.items():
-        conn.execute(
-            sa.text(
-                "UPDATE ai_capability_configs "
-                "SET max_tokens = :mt, updated_at = now() "
-                "WHERE capability_key = :key "
-                "AND (max_tokens IS NULL OR max_tokens < 100)"
-            ),
-            {"mt": mt, "key": cap_key},
-        )
-
-    # ── a. 确保 qwen-vl-plus 模型存在 ────────────────────────────────────────
-    conn.execute(sa.text("""
-            INSERT INTO ai_models (id, name, provider, model_key, api_key, api_base,
-                                   description, status, is_default, usage_today,
-                                   created_at, updated_at)
-            SELECT gen_random_uuid()::text,
-                   '通义千问VL Plus',
-                   'dashscope',
-                   'qwen-vl-plus',
-                   COALESCE(
-                       (SELECT api_key FROM ai_models
-                        WHERE provider = 'dashscope' AND deleted_at IS NULL
-                          AND api_key IS NOT NULL AND api_key != ''
-                        LIMIT 1),
-                       ''
-                   ),
-                   'https://dashscope.aliyuncs.com/compatible-mode/v1',
-                   '阿里云视觉理解模型（图片识别/分析）',
-                   'active', false, 0, now(), now()
-            WHERE NOT EXISTS (
-                SELECT 1 FROM ai_models
-                WHERE model_key = 'qwen-vl-plus' AND deleted_at IS NULL
-            )
-            """))
-
-    qwen_vl_row = conn.execute(
-        sa.text(
-            "SELECT id FROM ai_models WHERE model_key = 'qwen-vl-plus' "
-            "AND deleted_at IS NULL AND status = 'active' LIMIT 1"
-        )
-    ).first()
-    if not qwen_vl_row:
-        return
-    qwen_vl_id = qwen_vl_row[0]
-
-    qwen37_row = conn.execute(
-        sa.text(
-            "SELECT id FROM ai_models WHERE model_key = 'qwen3.7-plus' "
-            "AND deleted_at IS NULL AND status = 'active' LIMIT 1"
-        )
-    ).first()
-    qwen37_id = qwen37_row[0] if qwen37_row else None
-
-    # ── b. 仅当当前 primary 不是 qwen-vl-plus 时修正绑定与 max_tokens ───────
-    current = conn.execute(
-        sa.text(
-            "SELECT primary_model_id, lite_model_id, fallback_model_id, max_tokens "
-            "FROM ai_capability_configs WHERE capability_key = 'image_analysis'"
-        )
-    ).first()
-
-    if current is None:
-        # capability 不存在则创建
-        conn.execute(
-            sa.text("""
-                INSERT INTO ai_capability_configs
-                    (id, capability_key, capability_name, primary_model_id,
-                     lite_model_id, fallback_model_id, timeout_seconds,
-                     max_retries, max_tokens, concurrency, extra_params,
-                     is_enabled, created_at, updated_at)
-                VALUES (gen_random_uuid()::text, 'image_analysis', '图片分析',
-                        :primary, :primary, :fallback, 30, 1, 1000, 2,
-                        '{}'::jsonb, true, now(), now())
-                """),
-            {"primary": qwen_vl_id, "fallback": qwen37_id},
-        )
-        return
-
-    current_primary = current[0]
-    if current_primary == qwen_vl_id:
-        # 已经绑定 qwen-vl-plus：视为后台/数据迁移已处理，不覆盖任何配置
-        return
-
-    set_parts = [
-        "primary_model_id = :vl_id",
-        "lite_model_id = :vl_id",
-        "max_tokens = 1000",
-        "updated_at = now()",
-    ]
-    params: dict = {"vl_id": qwen_vl_id}
-    if qwen37_id is not None:
-        set_parts.insert(2, "fallback_model_id = :qwen37_id")
-        params["qwen37_id"] = qwen37_id
-
-    conn.execute(
-        sa.text(
-            "UPDATE ai_capability_configs SET " + ", ".join(set_parts) + " WHERE capability_key = 'image_analysis'"
-        ),
-        params,
-    )
-
-
-def downgrade() -> None:
-    pass

alembic/versions/102_image_analysis_max_tokens_1500.py

@@ -1,36 +0,0 @@
-# -*- coding: utf-8 -*-
-"""102: image_analysis max_tokens 1200 -> 1500.
-
-v6 prompt 更长、字段更多，旧 max_tokens 容易截断 JSON。
-仅在 image_analysis 当前 max_tokens < 1500 时更新（幂等，不覆盖后台已调到 >=1500 的配置）。
-"""
-
-import sqlalchemy as sa
-
-from alembic import op
-
-revision = "102_image_analysis_max_tokens_1500"
-down_revision = "101_qwen_vl_plus_and_max_tokens"
-branch_labels = None
-depends_on = None
-
-
-def upgrade() -> None:
-    conn = op.get_bind()
-
-    caps_table = conn.execute(sa.text("SELECT to_regclass('public.ai_capability_configs')")).scalar()
-    if not caps_table:
-        return
-
-    conn.execute(
-        sa.text(
-            "UPDATE ai_capability_configs "
-            "SET max_tokens = 1500, updated_at = now() "
-            "WHERE capability_key = 'image_analysis' "
-            "AND (max_tokens IS NULL OR max_tokens < 1500)"
-        )
-    )
-
-
-def downgrade() -> None:
-    pass

alembic/versions/103_v7_prompt_and_tokens_3000.py

@@ -1,194 +0,0 @@
-# -*- coding: utf-8 -*-
-"""image_analysis v7 prompt + max_tokens 3000 + max_retries 3
-
-Revision ID: 103_v7_prompt_and_tokens_3000
-Revises: 102_image_analysis_max_tokens_1500
-Create Date: 2026-10-07
-
-变更：
-1. 插入v7精简prompt（~1KB，v6 ~4.5KB，删除few-shot/冗长规则，减少输出token占用），设为active
-2. v6停用（is_active=False），保留历史
-3. image_analysis capability: max_tokens 1500→3000，max_retries 1→3
-
-ai_capability_configs 由应用 create_all 创建，全新 alembic-only 库可能不存在，
-故第3步做 to_regclass 守卫（同 102）。
-"""
-
-from sqlalchemy import text
-
-from alembic import op
-
-revision = "103_v7_prompt_and_tokens_3000"
-down_revision = "102_image_analysis_max_tokens_1500"
-branch_labels = None
-depends_on = None
-
-V7_SYSTEM = """# 角色
-你是一位专业的图片分析师，擅长准确识别图片中的场景、人物、物体、文字、氛围。
-
-# 任务
-对用户上传的图片逐张分析，描述你看到的内容，输出JSON格式。
-
-## 技能
-
-### 技能1：判断图片类型
-判断图片属于哪种类型，type字段填对应的英文值：
-- 商品图（product）：单个或多个商品、产品包装
-- 门店场景图（store）：店铺内部、门头招牌、货架陈列
-- 人物图（person）：人物形象、穿搭造型、肖像照片
-- 风景图（scene）：风景、动物、美食、街景
-- 其他（other）：以上都不是
-
-### 技能2：描述通用信息
-不管什么图都要描述：
-- type：图片类型，填product/store/person/scene/other其中一个
-- scene：一句话描述场景，例如"理疗养生店内部，摆着多张理疗床和产品货架"
-- mood：整体氛围，2-4个词，例如"整洁专业"、"热闹温馨"
-- colors：主要颜色，最多5个，写具体颜色名（亮红色/米白色/深蓝色，不写笼统的红色蓝色）
-- visible_text：图片里看到的文字，说明什么字、在什么位置，最多5条；没看到就空数组
-- lighting：光线情况，例如"明亮柔光"、"自然光"、"室内暖黄灯"
-- composition：怎么拍的，例如"居中特写"、"中景平视"、"俯拍"
-- has_person：有没有人，true或false
-
-### 技能3：描述门店场景
-如果是门店场景图（type="store"），还要描述：
-- store_type：什么类型的店，例如"养生馆"、"便利店"、"餐饮店"、"母婴店"
-- brand_signage：招牌上写了什么字、有什么品牌标识
-- visual_elements：看到哪些显眼的东西（招牌样式、灯光、货架、商品陈列、海报、收银台等），最多8个
-- product_categories：看到哪些品类的商品，例如"饮料零食"、"养生产品"
-- promotion_elements：有没有促销活动（打折海报、满减吊旗等），没有就空数组
-- atmosphere：店内什么氛围，例如"亲民生活化"、"老字号专业感"
-- cleanliness：店内干净程度，例如"干净整洁"、"货架整齐"
-- 看到顾客或店员要描述他们在做什么，has_person填true
-
-### 技能4：描述商品
-如果是商品图（type="product"），逐个商品描述：
-- product_name：商品名称，尽量具体，例如"OMO奥妙除菌除螨洗衣液"；看不出来填null
-- brand：什么牌子，看不出来填null
-- category：类目，从以下选一个：服饰鞋包/美妆/数码/食品/家居清洁/母婴/配饰/其他
-- package_type：什么包装，例如"瓶装"、"盒装"、"罐装"、"袋装"、"多瓶装"
-- package_color：包装主要颜色，写具体色（亮红色不写红色）
-- body_shape：瓶身或包装形状，例如"圆润胖瓶"、"竖款带把手瓶身"
-- label_design：标签设计，例如"红色标签印白色品牌logo"
-- key_text_on_package：包装上最显眼的文字（品牌名、功能词、卖点词），最多5个
-- product_features：包装特征，3-6个短语，包含颜色、瓶盖、形状、标签图案
-- key_selling_points：核心卖点，1-3个短语
-
-### 技能5：描述人物
-如果是人物图（type="person"），描述：
-- person_count：几个人
-- gender：性别（男/女/无法判断）
-- age_range：年龄段（儿童/青少年/青年/中年/老年/无法判断）
-- outfit_style：穿搭风格，例如"休闲日常"、"通勤商务"、"街头潮流"
-- upper_wear：上装（颜色+款式+材质），穿裙装不填
-- lower_wear：下装（颜色+款式+版型），穿裙装不填
-- dress_wear：裙装描述，穿上下装不填
-- outerwear：外套
-- shoes：鞋子
-- bag：包袋，没有填null
-- accessories：配饰（眼镜/帽子/项链/耳环/手表/手链/围巾/腰带等），没有填空数组
-- hairstyle：发型
-- makeup：妆容，男生或看不出填null
-- expression：表情，例如"微笑看镜头"、"冷酷无表情"
-- pose：姿势动作，例如"身直立正对镜头"、"单手撩发"
-- body_type：身材，例如"纤细苗条"、"高挑身材"、"丰满匀称"
-- portrait_prompt：80-150字详细描述人物形象（后面用来AI生成肖像图），要写清年龄段、穿搭完整细节、发型发色、妆容、表情、姿势、场景、光线、风格感觉，语言要有画面感
-
-### 技能6：描述风景
-如果是风景图（type="scene"），描述：
-- scene_type：什么场景，例如"自然风景"、"城市街景"、"动物"、"美食"
-- main_subject：画面主体是什么
-- key_elements：关键元素，最多8个
-- environment_objects：周围环境物体，最多8个
-- atmosphere：整体氛围，例如"秋日慵懒氛围感"、"清新自然氧气感"
-- 有人物就描述人物特征
-
-## 限制
-- 只输出JSON，不要任何解释文字，不要markdown代码块包裹，不要写"好的""以下是分析结果"这种废话
-- 颜色写具体色调（亮红色/米白色/深蓝色/翠绿色），不写笼统词汇
-- 瓶身、包装、招牌上的文字尽量识别出来（品牌名、功能词、卖点词）
-- 多个商品、多个人物分开描述，不要合并
-- 看不出来、不确定的字段填null或空数组，布尔值填true/false，绝对不要瞎编
-- 确保JSON格式合法，所有大括号、中括号、引号正确闭合
-- 数组字段控制数量：colors最多5个，visible_text最多5条，visual_elements最多8个，accessories最多10个"""
-V7_USER = "请分析这张图片，按系统消息的JSON结构输出。"
-
-
-def _capability_table_exists(bind) -> bool:
-    return bool(bind.execute(text("SELECT to_regclass('public.ai_capability_configs')")).scalar())
-
-
-def upgrade() -> None:
-    bind = op.get_bind()
-    # 1. 停用旧的active image_analysis prompt（含v6）
-    bind.execute(
-        text(
-            "UPDATE viral_video_prompt_templates SET is_active = FALSE "
-            "WHERE prompt_type = 'image_analysis' AND is_active = TRUE"
-        )
-    )
-    # 2. 幂等插入v7（存在则更新并重新激活）
-    existing = bind.execute(
-        text("SELECT id FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = 7")
-    ).fetchone()
-    if existing:
-        bind.execute(
-            text(
-                "UPDATE viral_video_prompt_templates SET is_active = TRUE, "
-                "system_prompt = :sys, user_prompt_template = :usr, "
-                "name = 'v7 精简结构化分析', updated_at = NOW() "
-                "WHERE prompt_type = 'image_analysis' AND version = 7"
-            ),
-            {"sys": V7_SYSTEM, "usr": V7_USER},
-        )
-    else:
-        bind.execute(
-            text(
-                "INSERT INTO viral_video_prompt_templates "
-                "(prompt_type, version, name, system_prompt, user_prompt_template, "
-                "is_active, created_at, updated_at) "
-                "VALUES ('image_analysis', 7, 'v7 精简结构化分析', "
-                ":sys, :usr, TRUE, NOW(), NOW())"
-            ),
-            {"sys": V7_SYSTEM, "usr": V7_USER},
-        )
-    # 3. capability max_tokens=3000、max_retries=3（表不存在则跳过）
-    if _capability_table_exists(bind):
-        bind.execute(
-            text(
-                "UPDATE ai_capability_configs SET max_tokens = 3000, "
-                "updated_at = NOW() "
-                "WHERE capability_key = 'image_analysis' AND "
-                "(max_tokens IS NULL OR max_tokens < 3000)"
-            )
-        )
-        bind.execute(
-            text(
-                "UPDATE ai_capability_configs SET max_retries = 3, updated_at = NOW() "
-                "WHERE capability_key = 'image_analysis' AND "
-                "(max_retries IS NULL OR max_retries < 3)"
-            )
-        )
-
-
-def downgrade() -> None:
-    bind = op.get_bind()
-    # 删除v7
-    bind.execute(
-        text("DELETE FROM viral_video_prompt_templates " "WHERE prompt_type = 'image_analysis' AND version = 7")
-    )
-    # 恢复v6为active
-    bind.execute(
-        text(
-            "UPDATE viral_video_prompt_templates SET is_active = TRUE "
-            "WHERE prompt_type = 'image_analysis' AND version = 6"
-        )
-    )
-    # tokens/retries回退
-    if _capability_table_exists(bind):
-        bind.execute(
-            text(
-                "UPDATE ai_capability_configs SET max_tokens = 1500, max_retries = 1, "
-                "updated_at = NOW() WHERE capability_key = 'image_analysis'"
-            )
-        )

apps/web/src/pages/viral-video/ViralVideoPage.css

@@ -1050,13 +1050,10 @@
   flex-direction: column;
   align-items: center;
   justify-content: center;
-  height: 360px;
   padding: 28px 16px;
   gap: 10px;
   background: #fff;
   border: 1px solid #e5e7eb;
-  border-radius: 10px;
-  margin-top: 8px;
 }
 .vv-copy-loading .vv-spinner {
   width: 28px;
@@ -1085,38 +1082,18 @@
 
 /* ── Storyboard (linear doc style) ── */
 .vv-storyboard {
-  display: flex;
-  flex-direction: column;
-  height: 360px;
-  padding: 10px 12px;
-  background: #fff;
-  border: 1px solid #e5e7eb;
-  border-radius: 10px;
-  margin-top: 8px;
-  overflow: hidden;
+  padding: 6px 2px;
+  background: transparent;
+  border: none;
 }
 
 .vv-sb-doc {
-  flex: 1 1 auto;
   display: flex;
   flex-direction: column;
   gap: 3px;
   color: #1f2937;
   font-size: 13px;
   line-height: 1.55;
-  overflow-y: auto;
-  padding-right: 4px;
-  margin-right: -4px;
-}
-.vv-sb-doc::-webkit-scrollbar {
-  width: 6px;
-}
-.vv-sb-doc::-webkit-scrollbar-thumb {
-  background: #d8c4ff;
-  border-radius: 3px;
-}
-.vv-sb-doc::-webkit-scrollbar-track {
-  background: transparent;
 }
 .vv-sb-h {
   margin: 6px 0 2px;
@@ -1406,14 +1383,12 @@
 /* 口播稿 —— 复用 vv-sb-field 样式，无额外需求 */
 
 .vv-sb-actions {
-  flex-shrink: 0;
   display: flex;
   align-items: center;
   gap: 8px;
   margin-top: 8px;
   padding-top: 8px;
   border-top: 1px solid #e5e7eb;
-  background: #fff;
 }
 .vv-sb-actions .vv-btn-ghost {
   padding: 6px 14px;

apps/worker/worker_app/tasks/viral_video.py

@@ -264,8 +264,8 @@ def _recover_stale_jobs() -> int:
             "SET status='failed', error_msg='任务执行超时，请重试', updated_at=:now "
             "WHERE status='running' "
             "  AND started_at IS NOT NULL AND started_at < :cutoff_start "
-            "  AND updated_at < :cutoff_beat "
-            "  AND (heartbeat_at IS NULL OR heartbeat_at < :cutoff_beat)"
+            "  AND (heartbeat_at IS NULL OR heartbeat_at < :cutoff_beat) "
+            "  AND (heartbeat_at IS NOT NULL OR updated_at < :cutoff_beat)"
         )
         result = ssn.execute(sql, {"now": now, "cutoff_start": cutoff_start, "cutoff_beat": cutoff_beat})
         ssn.commit()
@@ -316,9 +316,9 @@ _DEFAULT_NEGATIVE_PROMPTS = [
 ]
 
 
-def _empty_copy_result(duration: int = 15, ratio: str = "9:16", theme: str = "") -> dict:
+def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict:
     return {
-        "overview": {"theme": theme or "好物推荐", "total_duration": duration, "aspect_ratio": ratio},
+        "overview": {"theme": "好物推荐", "total_duration": duration, "aspect_ratio": ratio},
         "scene_and_lighting": "简洁明亮的室内场景，柔和自然光，产品主体清晰",
         "shots": [],
         "hard_constraints": list(_DEFAULT_HARD_CONSTRAINTS),
@@ -471,15 +471,11 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
 
     template = get_template("intent_parsing")
     system = render_system_prompt(template)
-    marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
-    image_category_hint = _determine_theme(image_analysis, marketing_purpose)
     user = render_user_prompt(
         template,
         user_copy_text=job.user_copy_text or "（未提供，全由 AI 创作）",
         industry=job.industry or "未指定",
         image_analysis=products_summary or "- （无图片分析结果）",
-        marketing_purpose=marketing_purpose,
-        image_category_hint=image_category_hint,
     )
 
     def _parse(raw: str) -> dict:
@@ -508,25 +504,16 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict:
 
     _client_fast = ai_router.get_llm_client("intent_parsing", variant="primary")
     _client_pro = ai_router.get_llm_client("intent_parsing", variant="lite")
-    _intent_deadline = time.time() + 60
-    _seen_models: set[str] = set()
     for _client, _lbl in [(_client_fast, "fast"), (_client_pro, "pro-fallback")]:
         if not _client or not _client.is_available:
             continue
-        if _client.model in _seen_models:
-            logger.info("[爆款视频] 意图解析跳过重复模型 %s label=%s", _client.model, _lbl)
-            continue
-        _seen_models.add(_client.model)
-        if time.time() > _intent_deadline:
-            logger.warning("[爆款视频] 意图解析超过60s总预算，跳过 label=%s", _lbl)
-            break
         try:
             logger.info("[爆款视频] 意图解析 model=%s label=%s", _client.model, _lbl)
             raw = _client.chat_completion(
                 [{"role": "system", "content": system}, {"role": "user", "content": user}],
                 temperature=0.4,
                 max_tokens=1024,
-                timeout=20,
+                timeout=60,
             )
             if not raw:
                 continue
@@ -563,70 +550,6 @@ def _persona_style_hint(persona_id: str) -> str:
     return "【人设风格：未指定】亲切自然、像朋友分享好物"
 
 
-def _determine_theme(image_analysis: dict | None, marketing_purpose: str = "") -> str:
-    """根据图片分析结果和营销目的，智能推断默认主题。
-
-    门店类→门店探店/到店体验；商品图→好物分享/产品种草；
-    人物图→穿搭/人物故事；场景图→场景氛围/空间体验。
-    """
-    products = (image_analysis or {}).get("products", []) or []
-    type_counts: dict[str, int] = {}
-    for p in products:
-        if not isinstance(p, dict):
-            continue
-        cat = (p.get("category") or "").strip()
-        if any(
-            k in cat
-            for k in (
-                "门店",
-                "店铺",
-                "餐饮",
-                "美容",
-                "美发",
-                "养生",
-                "健身",
-                "酒店",
-                "咖啡",
-                "奶茶",
-                "餐厅",
-                "颈肩",
-                "调理",
-            )
-        ):
-            type_counts["store"] = type_counts.get("store", 0) + 1
-        elif any(k in cat for k in ("人物", "穿搭", "人像", "服装")):
-            type_counts["person"] = type_counts.get("person", 0) + 1
-        elif any(k in cat for k in ("场景", "空间", "环境", "非产品")):
-            type_counts["scene"] = type_counts.get("scene", 0) + 1
-        elif cat and cat not in ("无法判断", "非产品图", ""):
-            type_counts["product"] = type_counts.get("product", 0) + 1
-        src = p.get("_source") or ""
-        if "store" in src:
-            type_counts["store"] = type_counts.get("store", 0) + 1
-        elif "person" in src:
-            type_counts["person"] = type_counts.get("person", 0) + 1
-
-    dominant = max(type_counts, key=type_counts.get) if type_counts else "product"
-    mp = (marketing_purpose or "").strip()
-
-    if any(k in mp for k in ("获客", "引流", "到店")):
-        if dominant == "store":
-            return "门店探店·到店体验"
-        return "门店探店·到店体验"
-    if any(k in mp for k in ("品牌", "宣传")):
-        return "品牌故事·门店体验" if dominant == "store" else "品牌故事·产品展示"
-    if any(k in mp for k in ("种草", "推荐")):
-        return "穿搭分享·人物种草" if dominant == "person" else "好物分享·产品种草"
-
-    theme_map = {
-        "store": "门店探店·到店体验",
-        "person": "穿搭分享·人物故事",
-        "scene": "空间体验·场景氛围",
-        "product": "好物分享·产品种草",
-    }
-    return theme_map.get(dominant, "好物分享·产品种草")
-
-
 def _build_products_summary(image_analysis: dict) -> str:
     """把 VLM 返回的商品分析结果拼给文案/分镜生成 prompt 用。
     优先用 summary（自然段落）；没有时用结构化字段兜底拼一段。"""
@@ -743,36 +666,14 @@ def _fallback_script(job: ViralVideoJob) -> dict:
     """脚本生成失败时的兜底脚本（极简但可用）。"""
     dur = max(5, min(30, int(getattr(job, "duration", 15) or 15)))
     ratio = getattr(job, "video_ratio", None) or "9:16"
-    _ia = getattr(job, "image_analysis", None) or {}
-    _mp = getattr(job, "marketing_purpose", "") or ""
-    default_theme = _determine_theme(_ia, _mp)
-    base = _empty_copy_result(dur, ratio, theme=default_theme)
-    _voiceover_map = {
-        "store": "带你探店！今天来到这家店，环境真的超棒，服务也很到位，推荐大家来体验一下。",
-        "person": "哈喽，今天给大家分享我的日常穿搭，简单舒适又好看，你们觉得怎么样？",
-        "scene": "带大家感受一下这个空间，氛围感拉满，真的很适合打卡体验。",
-        "product": "你好，给大家分享一款我最近在用的好物，真的很不错，推荐你们也试试。",
-    }
-    _products = (_ia or {}).get("products", []) or []
-    _dominant = "product"
-    for p in _products:
-        if not isinstance(p, dict):
-            continue
-        src = p.get("_source") or ""
-        cat = p.get("category") or ""
-        if "store" in src or any(k in cat for k in ("门店", "店铺", "餐饮", "美容", "颈肩", "调理")):
-            _dominant = "store"
-            break
-        elif "person" in src or any(k in cat for k in ("人物", "穿搭", "人像")):
-            _dominant = "person"
-            break
-    voiceover = job.user_copy_text or _voiceover_map.get(_dominant, _voiceover_map["product"])
+    base = _empty_copy_result(dur, ratio)
+    voiceover = job.user_copy_text or "你好，给大家分享一款我最近在用的好物，真的很不错，推荐你们也试试。"
     shots = [
         {
             "time_range": f"0-{dur}秒",
             "shot_type_angle_movement": "中景平视，缓慢推镜",
-            "scene_and_dialogue": voiceover,
-            "action_details": "自然展示，表情亲切，动作流畅",
+            "scene_and_dialogue": "明亮室内，人物自然出镜，微笑着看向镜头。" + voiceover,
+            "action_details": "人物手持产品自然展示，表情亲切，动作流畅",
             "audio_bgm": "轻快流行BGM",
             "transition": "结束",
             "reference_image_index": 0 if job.images else None,
@@ -782,7 +683,7 @@ def _fallback_script(job: ViralVideoJob) -> dict:
     base["voiceover_script"] = voiceover
     base["final_copy"] = voiceover
     base["suggested_copy"] = voiceover
-    base["title"] = default_theme
+    base["title"] = "好物分享"
     return base
 
 
@@ -800,18 +701,12 @@ def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict:
     ov = raw.get("overview")
     if isinstance(ov, dict):
         base["overview"] = {
-            "theme": str(
-                ov.get("theme")
-                or _determine_theme(getattr(job, "image_analysis", None), getattr(job, "marketing_purpose", ""))
-            ),
+            "theme": str(ov.get("theme") or "好物分享"),
             "total_duration": int(ov.get("total_duration") or dur),
             "aspect_ratio": str(ov.get("aspect_ratio") or ratio),
         }
     else:
-        base["overview"]["theme"] = str(
-            raw.get("title")
-            or _determine_theme(getattr(job, "image_analysis", None), getattr(job, "marketing_purpose", ""))
-        )
+        base["overview"]["theme"] = str(raw.get("title") or "好物分享")
 
     base["scene_and_lighting"] = str(raw.get("scene_and_lighting") or base["scene_and_lighting"])
 
@@ -898,11 +793,7 @@ def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None:
     base = _empty_copy_result(dur, ratio)
     if not raw:
         return None
-    base["overview"]["theme"] = (
-        xp.text_of(raw, "overview_theme")
-        or xp.text_of(raw, "title")
-        or _determine_theme(getattr(job, "image_analysis", None), getattr(job, "marketing_purpose", ""))
-    )
+    base["overview"]["theme"] = xp.text_of(raw, "overview_theme") or xp.text_of(raw, "title") or "好物分享"
     est = xp.attr_int(xp.text_of(raw, "estimated_duration"), 0)
     if est:
         base["overview"]["total_duration"] = est
@@ -992,11 +883,8 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
     system_tpl = system_tpl.replace("{global_constraints}", GLOBAL_CONSTRAINTS)
     system_tpl = system_tpl.replace("{negative_rules}", NEGATIVE_RULES)
 
-    marketing_purpose = getattr(job, "marketing_purpose", "") or "未指定"
-    image_category_hint = _determine_theme(image_analysis, marketing_purpose)
     fusion_brief = (
         f"意图：{intent_str}\n关键信息：{key_msgs}\n调性：{tone}\n"
-        f"营销目的：{marketing_purpose}\n建议主题方向：{image_category_hint}\n"
         f"用户原文：{job.user_copy_text or '（未提供）'}\n创作模式：{fusion_level}"
     )
     user = render_user_prompt(
@@ -1048,32 +936,22 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di
     # #2220: 直接用 ai_router 获取 client，不再手动提取 model_key
     _client_fast = ai_router.get_llm_client("storyboard", variant="primary")
     _client_pro = ai_router.get_llm_client("storyboard", variant="lite")
-    _script_fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "90"))
-    _script_pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "60"))
-    _script_deadline = time.time() + 180
+    _script_fast_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_FAST_TIMEOUT", "150"))
+    _script_pro_tmo = int(os.environ.get("VIRAL_VIDEO_SCRIPT_PRO_TIMEOUT", "150"))
     try:
-        # #2233: fast_timeout=90s, pro_timeout=60s，总deadline 180s
+        # #2217: doubao-seed-2-1-pro生成长编导脚本高峰期>90s，上调到150s，支持ENV覆盖
         normalized = _try_gen(_client_fast, 0.8, 2500, "fast-first", tmo=_script_fast_tmo)
         if normalized is not None:
             return normalized
-        if time.time() > _script_deadline:
-            logger.warning("[爆款视频] 编导脚本超过180s总预算，使用兜底脚本")
-            return _fallback_script(job)
         normalized = _try_gen(_client_fast, 0.6, 3200, "fast-retry", tmo=_script_fast_tmo)
         if normalized is not None:
             return normalized
-        # 第三次：用 lite/pro 模型兜底，跳过与 primary 相同的模型
+        # 第三次：用 lite/pro 模型兜底
         if _client_pro and _client_pro.is_available:
-            if _client_pro.model != _client_fast.model:
-                if time.time() <= _script_deadline:
-                    normalized = _try_gen(_client_pro, 0.7, 3500, "pro-fallback", tmo=_script_pro_tmo)
-                    if normalized is not None:
-                        return normalized
-                else:
-                    logger.warning("[爆款视频] 编导脚本超过180s总预算，跳过pro-fallback")
-            else:
-                logger.info("[爆款视频] pro-fallback模型与primary相同(%s)，跳过重复调用", _client_pro.model)
-        logger.warning("[爆款视频] 编导脚本均未生成合格结果，使用兜底脚本")
+            normalized = _try_gen(_client_pro, 0.7, 3500, "pro-fallback", tmo=_script_pro_tmo)
+            if normalized is not None:
+                return normalized
+        logger.warning("[爆款视频] 编导脚本三次都未生成合格结果，使用兜底脚本")
         return _fallback_script(job)
     except Exception as e:
         logger.warning("[爆款视频] 编导脚本生成异常: %s，使用兜底脚本", e, exc_info=True)

apps/worker/worker_app/tasks/vision/assembler.py

@@ -170,7 +170,7 @@ def _is_v4_schema(fj: dict) -> bool:
     """判断是v4嵌套schema还是旧扁平schema"""
     return (
         isinstance(fj.get("products"), list)
-        or fj.get("type") in ("product", "store", "person", "scene", "other")
+        or fj.get("type") in ("product", "store", "person", "other")
         or isinstance(fj.get("people"), dict)
     )
 
@@ -381,55 +381,9 @@ def assemble_result(idx: int, fast_json: dict | None, ocr_texts: list[str]) -> d
     ocr_texts = ocr_texts or []
 
     if _is_v4_schema(fj):
-        result = _assemble_v4(idx, fj, ocr_texts)
+        return _assemble_v4(idx, fj, ocr_texts)
     else:
-        result = _assemble_old(idx, fj, ocr_texts)
-    return _apply_partial_fallback(result, fj)
-
-
-def _apply_partial_fallback(result: dict[str, Any], fj: dict) -> dict[str, Any]:
-    """partial（截断修复）产物的字段兜底：用已有碎片填充空字段，
-    避免"无法判断"直接透传给下游。非partial产物原样返回。"""
-    if not fj.get("_partial"):
-        return result
-    desc = str(fj.get("description") or "").strip()
-    # 收集所有顶层标量碎片作为兜底素材
-    fragments: list[str] = []
-    for k in ("main_subject", "store_type", "scene_type", "description"):
-        v = fj.get(k)
-        if isinstance(v, str) and v.strip() and v != "无法判断":
-            fragments.append(v.strip())
-    for arr_k in ("environment_objects", "key_elements", "visual_elements"):
-        arr = fj.get(arr_k) or []
-        if isinstance(arr, list):
-            for item in arr[:3]:
-                if isinstance(item, str) and item.strip():
-                    fragments.append(item.strip())
-                elif isinstance(item, dict):
-                    tv = item.get("text") or item.get("name")
-                    if tv:
-                        fragments.append(str(tv))
-    frag_text = "；".join(fragments[:3])
-
-    if result.get("name") in ("未识别", "", None) and (desc or frag_text):
-        result["name"] = (desc or fragments[0])[:30]
-    if str(result.get("appearance", "")).startswith("无法判断"):
-        if desc:
-            result["appearance"] = desc[:200]
-        elif frag_text:
-            result["appearance"] = frag_text[:200]
-    if result.get("key_features") in (["无法判断"], []) and (desc or fragments):
-        kf = []
-        if desc:
-            kf.append(desc[:30])
-        for f in fragments[:3]:
-            if f not in kf:
-                kf.append(f[:40])
-        result["key_features"] = kf[:8]
-    if result.get("summary") in ("未识别", "", None) and (desc or fragments):
-        result["summary"] = (desc or fragments[0])[:40]
-    result["_partial"] = True
-    return result
+        return _assemble_old(idx, fj, ocr_texts)
 
 
 def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
@@ -464,41 +418,8 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
 
     has_person = fj.get("has_person", False)
 
-    # 兼容老schema：无type字段或type不在已知枚举时，按has_person/products兜底
-    _KNOWN_V4_TYPES = ("product", "store", "person", "scene", "other")
-    if vtype not in _KNOWN_V4_TYPES:
-        if has_person:
-            vtype = "person"
-        elif products:
-            vtype = "product"
-        else:
-            vtype = "other"
-
-    # 人物信息提取辅助（scene/store分支有人物时追加到key_features）
-    def _extract_person_features(source: dict) -> list[str]:
-        parts = []
-        for pk in ("outfit_style", "upper_wear", "lower_wear", "dress_wear", "outerwear", "pose", "expression"):
-            pv = source.get(pk)
-            if pv and str(pv) not in ("null", None, "无法判断") and len(str(pv)) <= 40:
-                parts.append(f"人物:{pv}")
-        for pk2 in ("shoes", "bag", "hairstyle"):
-            pv2 = source.get(pk2)
-            if pv2 and str(pv2) not in ("null", None) and len(str(pv2)) <= 40:
-                parts.append(f"人物:{pv2}")
-        return parts
-
-    # partial截断保护：声明了product但products数组没来得及输出时，
-    # 按已返回的碎片字段改路由，避免直接掉到other丢信息
-    if fj.get("_partial") and vtype == "product" and not products:
-        if any(fj.get(k) for k in ("signage_details", "store_layout", "brand_signage", "store_type")):
-            vtype = "store"
-        elif any(fj.get(k) for k in ("key_elements", "main_subject", "scene_type", "spatial_layout")):
-            vtype = "scene"
-        else:
-            vtype = "other"
-
-    # ── 人物类 ──
-    if vtype == "person":
+    # 人物类
+    if vtype == "person" or has_person:
         # 取第一个人物信息（v5 schema人物信息在顶层）
         person_info = fj
         # 兼容people嵌套
@@ -558,12 +479,6 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
             v = person_info.get(k)
             if v and v not in ("null", None):
                 kf.append(str(v))
-        env_obj = person_info.get("environment_objects") or []
-        if not isinstance(env_obj, list):
-            env_obj = [env_obj]
-        for o in env_obj:
-            if o and str(o) not in kf and len(str(o)) <= 40:
-                kf.append(str(o))
         if text_on_package:
             kf.append(f"文字:{'/'.join(text_on_package[:3])}")
         kf = kf[:8] or ["无法判断"]
@@ -624,15 +539,9 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
             for s in sell[:2]:
                 if s and len(str(s)) <= 30 and str(s) not in kf:
                     kf.append(f"卖点:{s}")
-        env_obj = fj.get("environment_objects") or []
-        if not isinstance(env_obj, list):
-            env_obj = [env_obj]
-        for o in env_obj:
-            if o and str(o) not in kf and len(str(o)) <= 40:
-                kf.append(str(o))
         if text_on_package:
             kf.append(f"文字: {'/'.join(text_on_package[:3])}")
-        kf = kf[:8] or ["无法判断"]
+        kf = kf[:6] or ["无法判断"]
         portrait_prompt = _build_product_prompt_from_v4(main, fj)
         if brand != "无法判断" and brand not in name:
             summary = f"{brand} {name}"
@@ -654,222 +563,57 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
             "_source": "v2_fast_json_v4",
         }
 
-    # 门店类
+    # 门店类或其他
     if vtype == "store":
         store_type = fj.get("store_type") or "店铺"
-        # brand 多级兜底：brand_signage → visible_text招牌文字 → text_on_package短词
-        brand_raw = fj.get("brand_signage")
-        if not brand_raw or brand_raw in ("无法判断", "", None):
-            brand = None
-            # 从visible_text找招牌文字（通常是位置含招牌/门头/背景的短词）
-            for vt in visible_text:
-                vt_str = vt.get("text") if isinstance(vt, dict) else str(vt)
-                if not vt_str or len(vt_str) < 2 or len(vt_str) > 12:
-                    continue
-                loc = (vt.get("location") or "") if isinstance(vt, dict) else ""
-                if any(k in loc for k in ("招牌", "门头", "背景", "招牌墙")):
-                    brand = vt_str
-                    break
-            # 从text_on_package找2-8字的短词（非描述性）
-            if not brand:
-                _desc_words = {"干净", "整洁", "温馨", "专业", "明亮", "舒适", "宽敞", "现代", "传统", "时尚"}
-                for t in text_on_package:
-                    if 2 <= len(t) <= 8 and t not in _desc_words and not any(c in t for c in "的了是在我"):
-                        brand = t
-                        break
-            if not brand:
-                brand = "无法判断"
-        else:
-            brand = brand_raw
-        # name兜底：store_type为空时用brand
-        name = store_type if store_type != "店铺" else (brand if brand != "无法判断" else store_type)
+        name = store_type
+        brand = fj.get("brand_signage") or "无法判断"
         category = "门店场景"
-        # appearance: store_layout + furnishings + 陈设色调
+        visual = fj.get("visual_elements") or []
+        if isinstance(visual, str):
+            visual = [visual]
+        atmosphere = fj.get("atmosphere") or mood
         appearance_parts = []
         if fj.get("store_layout"):
             appearance_parts.append(str(fj["store_layout"]))
-        furnishings = fj.get("furnishings") or []
-        if isinstance(furnishings, dict):
-            _furn_vals = []
-            for fk in ("materials", "furniture", "shelving", "seating"):
-                fv = furnishings.get(fk)
-                if isinstance(fv, list):
-                    _furn_vals.extend(str(x) for x in fv if x)
-                elif isinstance(fv, str) and fv:
-                    _furn_vals.append(fv)
-            if _furn_vals:
-                appearance_parts.append("陈设：" + "、".join(_furn_vals[:4]))
-        elif isinstance(furnishings, list) and furnishings:
-            appearance_parts.append("陈设：" + "、".join(str(f) for f in furnishings[:4] if f))
+        if visual:
+            appearance_parts.append("、".join(str(v) for v in visual[:3]))
         if fj.get("cleanliness"):
             appearance_parts.append(str(fj["cleanliness"]))
         appearance = "；".join(appearance_parts) if appearance_parts else "门店环境"
-        # key_features: 招牌细节+海报+外部物品+视觉元素+陈列商品+环境物件
         kf = []
-        for arr_key in (
-            "signage_details",
-            "signage_posters",
-            "exterior_items",
-            "visual_elements",
-            "products_on_display",
-            "environment_objects",
-        ):
-            arr = fj.get(arr_key) or []
-            if not isinstance(arr, list):
-                arr = [arr]
-            for item in arr:
-                if item and str(item) not in kf and len(str(item)) <= 40:
-                    kf.append(str(item))
+        if isinstance(visual, list):
+            kf.extend(str(v) for v in visual if v and len(str(v)) <= 30)
         prods_vis = fj.get("product_categories_visible") or []
         if isinstance(prods_vis, list):
-            for c in prods_vis[:3]:
-                if c and str(c) not in kf:
-                    kf.append(str(c))
+            kf.extend(str(c) for c in prods_vis[:3] if c)
         promo = fj.get("promotion_elements") or []
         if isinstance(promo, list) and promo:
             kf.append("促销活动:" + "、".join(str(p) for p in promo[:2]))
-        if has_person:
-            for pf in _extract_person_features(fj):
-                if pf not in kf:
-                    kf.append(pf)
         if text_on_package:
-            kf.append("文字:" + "/".join(text_on_package[:3]))
-        kf = kf[:8] or ["门店场景"]
-        atmosphere = fj.get("atmosphere") or fj.get("mood") or mood
-        # portrait_prompt: 品牌+store_type+主色+atmosphere+scene+核心视觉元素
-        pieces = []
-        if brand != "无法判断":
-            pieces.append(brand)
-        pieces.append(store_type)
-        if color_names:
-            pieces.append("、".join(color_names[:3]) + "配色")
-        if atmosphere:
-            pieces.append(atmosphere)
-        if scene and scene != "通用":
-            pieces.append(scene)
-        visual = fj.get("visual_elements") or []
-        if isinstance(visual, list) and visual:
-            pieces.append("、".join(str(v) for v in visual[:3] if v))
-        pieces.append("门店实拍")
-        portrait_prompt = "，".join(p for p in pieces if p)
-        if len(portrait_prompt) > 200:
-            portrait_prompt = portrait_prompt[:200].rstrip("，")
-        summary = f"{brand} {store_type}" if brand != "无法判断" else f"{store_type}场景"
+            kf.append(f"文字: {'/'.join(text_on_package[:3])}")
+        kf = kf[:6] or ["门店场景"]
+        portrait_prompt = f"{brand if brand!='无法判断' else ''}{store_type}，{atmosphere}，{scene}场景，{('、'.join(color_names[:3])+'配色，') if color_names else ''}产品陈列丰富，门店实拍"
+        portrait_prompt = portrait_prompt.strip("，")
+        summary = f"{store_type}场景"
         return {
             "name": name[:30],
             "brand": str(brand)[:30],
             "category": category,
-            "appearance": appearance[:300],
+            "appearance": appearance[:200],
             "packaging": "门店场景无包装",
             "text_on_package": text_on_package,
             "key_features": kf,
             "scene": scene,
-            "mood": atmosphere,
+            "mood": atmosphere or mood,
             "portrait_prompt": portrait_prompt[:200],
             "summary": summary[:40],
-            "has_person": bool(fj.get("has_person", False)),
-            "_source": "v2_fast_json_v6_store",
+            "has_person": False,
+            "_source": "v2_fast_json_v4",
         }
 
-    # 场景类（纯场景图，无产品/人物）
-    if vtype == "scene":
-        main_subject = fj.get("main_subject") or ""
-        scene_type = fj.get("scene_type") or "场景图"
-        name = main_subject or scene_type
-        brand = "无法判断"
-        category = scene_type or "场景图"
-        # appearance: 空间布局 + 关键元素
-        app_parts = []
-        if fj.get("spatial_layout"):
-            app_parts.append(str(fj["spatial_layout"]))
-        key_elements = fj.get("key_elements") or []
-        if not isinstance(key_elements, list):
-            key_elements = [key_elements]
-        if key_elements:
-            app_parts.append("关键元素：" + "、".join(str(k) for k in key_elements[:4] if k))
-        if fj.get("lighting"):
-            app_parts.append("光线：" + str(fj["lighting"]))
-        appearance = "；".join(app_parts) if app_parts else "场景环境"
-        # key_features: key_elements + environment_objects（去重，最多8）
-        kf = []
-        for k in key_elements:
-            if k and str(k) not in kf and len(str(k)) <= 40:
-                kf.append(str(k))
-        env_obj = fj.get("environment_objects") or []
-        if not isinstance(env_obj, list):
-            env_obj = [env_obj]
-        for o in env_obj:
-            if o and str(o) not in kf and len(str(o)) <= 40:
-                kf.append(str(o))
-        if has_person:
-            for pf in _extract_person_features(fj):
-                if pf not in kf:
-                    kf.append(pf)
-        if text_on_package:
-            kf.append("文字:" + "/".join(text_on_package[:3]))
-        kf = kf[:8] or ["场景元素"]
-        scene_name = fj.get("scene") or scene_type
-        scene_mood = fj.get("mood") or mood
-        packaging = "场景无包装"
-        # portrait_prompt: main_subject + key_elements + lighting + atmosphere + composition + style
-        pieces = []
-        if main_subject:
-            pieces.append(main_subject)
-        if key_elements:
-            pieces.append("、".join(str(k) for k in key_elements[:3] if k))
-        if fj.get("lighting"):
-            pieces.append(str(fj["lighting"]) + "光线")
-        if scene_mood:
-            pieces.append(scene_mood + "氛围")
-        if fj.get("composition"):
-            pieces.append(str(fj["composition"]))
-        if fj.get("style"):
-            pieces.append(str(fj["style"]))
-        pieces.append("场景实拍，摄影级画质")
-        portrait_prompt = "，".join(p for p in pieces if p)
-        if len(portrait_prompt) > 200:
-            portrait_prompt = portrait_prompt[:200].rstrip("，")
-        summary = main_subject or scene_type
-        return {
-            "name": str(name)[:30],
-            "brand": brand,
-            "category": str(category)[:20],
-            "appearance": appearance[:300],
-            "packaging": packaging,
-            "text_on_package": text_on_package,
-            "key_features": kf,
-            "scene": scene_name,
-            "mood": scene_mood,
-            "portrait_prompt": portrait_prompt[:200],
-            "summary": str(summary)[:40],
-            "has_person": bool(fj.get("has_person", False)),
-            "_source": "v2_fast_json_v6_scene",
-        }
-
-    # other 兜底：当 description 为空时，用 environment_objects/scene 拼基本描述
-    desc = fj.get("description") or ""
-    if not desc.strip():
-        env_obj = fj.get("environment_objects") or []
-        if not isinstance(env_obj, list):
-            env_obj = [env_obj]
-        env_names = [str(e) for e in env_obj if e and len(str(e)) <= 25][:5]
-        scene_text = fj.get("scene") or ""
-        if env_names:
-            desc = f"{scene_text}场景中" + "、".join(env_names) if scene_text else "、".join(env_names)
-        elif scene_text and scene_text != "通用":
-            desc = f"{scene_text}场景"
-        else:
-            desc = "未识别"
-    kf_other = []
-    if desc != "未识别":
-        kf_other.append(desc[:30])
-    env_obj = fj.get("environment_objects") or []
-    if not isinstance(env_obj, list):
-        env_obj = [env_obj]
-    for o in env_obj:
-        if o and str(o) not in kf_other and len(str(o)) <= 40:
-            kf_other.append(str(o))
-    kf_other = kf_other[:8] or ["无法判断"]
+    # other 兜底
+    desc = fj.get("description") or "未识别"
     return {
         "name": desc[:30],
         "brand": "无法判断",
@@ -877,7 +621,7 @@ def _assemble_v4(idx: int, fj: dict, ocr_texts: list[str]) -> dict[str, Any]:
         "appearance": desc[:200],
         "packaging": "无法判断",
         "text_on_package": text_on_package,
-        "key_features": kf_other,
+        "key_features": [desc[:30]] if desc != "未识别" else ["无法判断"],
         "scene": scene,
         "mood": mood,
         "portrait_prompt": f"{scene}，{mood}氛围，{desc}"[:200],

apps/worker/worker_app/tasks/vision/fast_path.py

@@ -23,10 +23,10 @@ logger = logging.getLogger(__name__)
 
 # 超时（可通过环境变量覆盖）
 _IMG_WORKERS = int(os.environ.get("VISION_V2_IMG_WORKERS", "8"))
-_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "20"))
-_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "20"))
+_FAST_TIMEOUT = float(os.environ.get("VISION_V2_FAST_TIMEOUT", "15"))
+_FAST_JSON_TIMEOUT = float(os.environ.get("VISION_V2_FAST_JSON_TIMEOUT", "15"))
 _OCR_TIMEOUT = float(os.environ.get("VISION_V2_OCR_TIMEOUT", "6"))
-_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "45"))
+_PRO_TIMEOUT = float(os.environ.get("VISION_V2_PRO_TIMEOUT", "30"))
 
 _FALLBACK_RESULT = {
     "name": "未识别",
@@ -58,29 +58,27 @@ def analyze_image_v2(idx: int, img_url: str) -> dict[str, Any]:
 
     fj_result: dict[str, Any] | None = None
     ocr_result: list[str] = []
-    fast_elapsed = 0.0
-    pool = ThreadPoolExecutor(max_workers=2)
-    f_fj = pool.submit(vlm_fast_json.call_fast_json, img_url, timeout=_FAST_JSON_TIMEOUT)
-    f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=_OCR_TIMEOUT)
-    try:
-        for fut in as_completed([f_fj, f_ocr], timeout=_FAST_TIMEOUT):
-            try:
-                res = fut.result(timeout=1)
-            except Exception as e:
-                logger.warning("[vision.v2] 图片 #%d 子任务异常: %s", idx, e)
-                continue
-            if fut is f_fj and isinstance(res, dict):
-                fj_result = res
-            elif fut is f_ocr and isinstance(res, list):
-                ocr_result = res
-    except TimeoutError:
-        for f in (f_fj, f_ocr):
-            if not f.done():
-                f.cancel()
-        logger.warning("[vision.v2] 图片 #%d fast路径超时(%.0fs)，走pro兜底", idx, _FAST_TIMEOUT)
-    finally:
-        fast_elapsed = time.time() - t0
-        pool.shutdown(wait=False)  # 不等待未完成的线程，避免计时膨胀
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        f_fj = pool.submit(vlm_fast_json.call_fast_json, img_url, timeout=_FAST_JSON_TIMEOUT)
+        f_ocr = pool.submit(ocr_volc.call_ocr, img_url, timeout=_OCR_TIMEOUT)
+        try:
+            for fut in as_completed([f_fj, f_ocr], timeout=_FAST_TIMEOUT):
+                try:
+                    res = fut.result(timeout=1)
+                except Exception as e:
+                    logger.warning("[vision.v2] 图片 #%d 子任务异常: %s", idx, e)
+                    continue
+                if fut is f_fj and isinstance(res, dict):
+                    fj_result = res
+                elif fut is f_ocr and isinstance(res, list):
+                    ocr_result = res
+        except TimeoutError:
+            for f in (f_fj, f_ocr):
+                if not f.done():
+                    f.cancel()
+            logger.warning("[vision.v2] 图片 #%d fast路径超时(%.0fs)，走pro兜底", idx, _FAST_TIMEOUT)
+
+    fast_elapsed = time.time() - t0
 
     if fj_result:
         assembled = assembler.assemble_result(idx, fj_result, ocr_result)

apps/worker/worker_app/tasks/vision/json_utils.py

@@ -1,138 +0,0 @@
-# -*- coding: utf-8 -*-
-"""VLM 返回文本的稳健 JSON 提取工具。
-
-背景：复杂门店图 VLM 输出经常被 max_tokens 截断（finish_reason=length），
-json.loads 失败后整个结果被丢弃，导致"未识别"。本工具提供：
-1. markdown 代码块剥离（含只开不闭的截断场景）
-2. 最外层 { } 切片
-3. 非法控制字符清理
-4. 直接 json.loads
-5. 截断 JSON 括号/引号栈补全修复
-6. 尾部逐字符截断重试（去除最后一个不完整 token 后修复）
-
-成功返回 dict；截断修复产物带 _partial=True 标记；彻底失败返回 None。
-"""
-
-from __future__ import annotations
-
-import json
-import logging
-import re
-
-logger = logging.getLogger(__name__)
-
-_CODE_FENCE_RE = re.compile(r"^```(?:json)?\s*\n?(.*?)\n?```\s*$", re.DOTALL)
-
-
-def _strip_code_fence(s: str) -> str:
-    s = s.strip()
-    m = _CODE_FENCE_RE.match(s)
-    if m:
-        return m.group(1).strip()
-    # 兼容开头 ```json 但结尾无 ```（截断场景）
-    if s.startswith("```"):
-        lines = s.split("\n")
-        if lines and lines[0].startswith("```"):
-            lines = lines[1:]
-        s = "\n".join(lines).strip()
-    return s
-
-
-def _repair_truncated_json(text: str) -> str:
-    """尝试补全被截断的JSON：维护 bracket/quote 栈，在末尾补闭合符。"""
-    stack: list[str] = []
-    in_string = False
-    escape = False
-    for ch in text:
-        if escape:
-            escape = False
-            continue
-        if ch == "\\" and in_string:
-            escape = True
-            continue
-        if ch == '"':
-            in_string = not in_string
-            continue
-        if in_string:
-            continue
-        if ch in "{[":
-            stack.append(ch)
-        elif ch == "}":
-            if stack and stack[-1] == "{":
-                stack.pop()
-        elif ch == "]":
-            if stack and stack[-1] == "[":
-                stack.pop()
-    repair = ""
-    if in_string:
-        repair += '"'
-    for opener in reversed(stack):
-        repair += "}" if opener == "{" else "]"
-    if repair:
-        logger.info(
-            "[json_utils] 截断JSON修复: 补全%d个闭合符 in_string=%s",
-            len(repair),
-            in_string,
-        )
-    return text + repair
-
-
-def _clean_invalid_chars(text: str) -> str:
-    """清理JSON中非法的控制字符（tab/newline 之外的 0x00-0x1f 段）。"""
-    return re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text)
-
-
-def extract_json_object(text: str) -> dict | None:
-    """从VLM返回文本中稳健提取JSON对象。
-
-    返回 dict 或 None。成功的 dict 可能带 _partial=True 标记，
-    表示原始文本被截断、经括号补全后得到的产物。
-    """
-    if not text or not isinstance(text, str):
-        return None
-    # 1. 剥离 markdown
-    text = _strip_code_fence(text)
-    # 2. 找最外层 { }
-    lpos = text.find("{")
-    if lpos < 0:
-        return None
-    rpos = text.rfind("}")
-    if rpos > lpos:
-        text = text[lpos : rpos + 1]
-    else:
-        # 截断场景：无任何闭合 }，取到末尾交给修复器
-        text = text[lpos:]
-    # 3. 清理非法控制字符
-    text = _clean_invalid_chars(text)
-    # 4. 直接 loads
-    try:
-        obj = json.loads(text)
-        return obj if isinstance(obj, dict) else None
-    except json.JSONDecodeError:
-        pass
-    # 5. 尝试截断修复
-    repaired = _repair_truncated_json(text)
-    try:
-        obj = json.loads(repaired)
-        if isinstance(obj, dict):
-            obj["_partial"] = True
-            return obj
-    except json.JSONDecodeError:
-        pass
-    # 6. 尾部逐字符截断重试（去除最后一个不完整 token）
-    for _ in range(50):
-        last_comma = repaired.rfind(",")
-        last_brace = max(repaired.rfind("}"), repaired.rfind("]"))
-        cut = max(last_comma, last_brace)
-        if cut < 10:
-            break
-        repaired = repaired[: cut + 1]
-        repaired = _repair_truncated_json(repaired)
-        try:
-            obj = json.loads(repaired)
-            if isinstance(obj, dict):
-                obj["_partial"] = True
-                return obj
-        except json.JSONDecodeError:
-            continue
-    return None

apps/worker/worker_app/tasks/vision/vlm_fallback.py

@@ -1,18 +1,18 @@
 # -*- coding: utf-8 -*-
-"""V2 兜底路径：image_analysis（默认 qwen-vl-plus 视觉模型，fallback qwen3.7-plus / DashScope）单图调用。
+"""V2 兜底路径：image_analysis lite/fallback（默认 qwen3.7-plus / DashScope）单图调用。
 
 fast_json 超时/返回非 JSON/识别为空时，本路径单次调用兜底。
 设计要点：
 - 通过 ai_router.get_vision_client() 获取 DoubaoClient 实例，不再自己拼 httpx 请求
 - enable_thinking=False + response_format=json_object
 - system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
-- max_tokens 不传，使用 client 中 capability 的 DB 配置（避免硬编码截断 JSON）
-- timeout=30s
+- timeout=25s
 - 返回 dict 统一走 assembler.assemble_result 组装，与 fast 路径输出格式完全一致
 """
 
 from __future__ import annotations
 
+import json
 import logging
 import time
 from typing import Any
@@ -21,7 +21,8 @@ from . import _prompt, assembler
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_TIMEOUT = 45
+_DEFAULT_TIMEOUT = 30
+_DEFAULT_MAX_TOKENS = 800
 
 
 def call_pro_vlm(
@@ -29,15 +30,13 @@ def call_pro_vlm(
     idx: int,
     *,
     timeout: int = _DEFAULT_TIMEOUT,
-    max_tokens: int | None = None,
 ) -> dict[str, Any] | None:
-    """max_tokens 默认 None：不显式传参，使用 client 内 capability 的 DB 配置。"""
     t0 = time.time()
 
     try:
         from packages.shared.ai_router import ai_router
 
-        client = ai_router.get_vision_client("image_analysis", variant="fallback")
+        client = ai_router.get_vision_client("image_analysis", variant="primary")
         if not client or not client.is_available:
             logger.warning("[vision.v2] pro vision client 不可用，跳过")
             return None
@@ -59,51 +58,36 @@ def call_pro_vlm(
     ]
 
     try:
-        call_kwargs: dict[str, Any] = {
-            "messages": messages,
-            "images": None,  # 图片已在 messages 中
-            "temperature": 0.3,
-            "timeout": timeout,
-            "enable_thinking": False,
-            "response_format": {"type": "json_object"},
-        }
-        # pro fallback：显式4000 tokens给复杂门店图留足空间
-        call_kwargs["max_tokens"] = max_tokens if max_tokens is not None else 4000
-
-        from .json_utils import extract_json_object
-
-        raw = None
-        obj = None
-        for _outer in range(2):
-            kw = dict(call_kwargs)
-            if _outer == 1:
-                kw.pop("response_format", None)
-                msgs2 = [dict(messages[0]), dict(messages[1])]
-                cont = [dict(c) for c in list(msgs2[1]["content"])]
-                cont[-1] = {"type": "text", "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。"}
-                msgs2[1] = {"role": "user", "content": cont}
-                kw["messages"] = msgs2
-            raw = client.vision_completion(**kw)
-            if not raw:
-                logger.warning("[vision.v2] pro 返回空 outer=%s", _outer)
-                continue
-            obj = extract_json_object(raw)
-            if obj is not None:
-                break
-            logger.warning("[vision.v2] pro 非JSON(100字) outer=%s: %s", _outer, raw[:100])
-
+        raw = client.vision_completion(
+            messages=messages,
+            images=None,  # 图片已在 messages 中
+            temperature=0.3,
+            max_tokens=_DEFAULT_MAX_TOKENS,
+            timeout=timeout,
+            enable_thinking=False,
+            response_format={"type": "json_object"},
+        )
         elapsed = time.time() - t0
-        if obj is None:
-            logger.warning("[vision.v2] pro 两次均未得到JSON elapsed=%.1fs", elapsed)
+        if not raw:
+            logger.warning("[vision.v2] pro 返回空 elapsed=%.1fs", elapsed)
             return None
-        if obj.get("_partial"):
-            logger.warning("[vision.v2] pro 返回截断JSON(partial) elapsed=%.1fs", elapsed)
+
         logger.info(
-            "[vision.v2] pro 完成 model=%s elapsed=%.1fs type=%s",
+            "[vision.v2] pro 完成 model=%s elapsed=%.1fs",
             client.model,
             elapsed,
-            obj.get("type"),
         )
+        s = _strip_code_fence(raw)
+        lpos, rr = s.find("{"), s.rfind("}")
+        if lpos >= 0 and rr > lpos:
+            s = s[lpos : rr + 1]
+        try:
+            obj = json.loads(s)
+        except json.JSONDecodeError:
+            logger.warning("[vision.v2] pro JSON 解析失败 head=%s", raw[:200])
+            return None
+        if not isinstance(obj, dict):
+            return None
 
         # 通过assembler统一组装，兼容v4嵌套schema和旧扁平schema
         result = assembler.assemble_result(idx, obj, [])
@@ -114,3 +98,15 @@ def call_pro_vlm(
         elapsed = time.time() - t0
         logger.warning("[vision.v2] pro 异常 elapsed=%.1fs err=%s", elapsed, e, exc_info=True)
         return None
+
+
+def _strip_code_fence(s: str) -> str:
+    s = s.strip()
+    if s.startswith("```"):
+        lines = s.split("\n")
+        if lines and lines[0].startswith("```"):
+            lines = lines[1:]
+        if lines and lines[-1].strip().startswith("```"):
+            lines = lines[:-1]
+        s = "\n".join(lines).strip()
+    return s

apps/worker/worker_app/tasks/vision/vlm_fast_json.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""V2 快速路径：image_analysis capability（默认 qwen-vl-plus 视觉模型 / DashScope）强约束 JSON-only 调用。
+"""V2 快速路径：image_analysis capability（默认 qwen3.8-flash / DashScope）强约束 JSON-only 调用。
 
 目标：替代"人体属性/商品检测/图像标签"三个火山不存在的专用云端 API。
 设计要点：
@@ -7,13 +7,13 @@
 - enable_thinking=False 关闭推理链（reasoning 是延迟主因）
 - response_format=json_object 强约束JSON输出
 - system prompt 优先读后台 viral_video_prompt_templates 配置，DB不可用时fallback到硬编码JSON schema
-- max_tokens 不传，使用 client 中 capability 的 DB 配置（避免硬编码截断 JSON）
-- temperature=0.1（稳定输出 JSON）
-- timeout=15s（失败由外层走 pro 兜底）
+- max_tokens=350、temperature=0.1（稳定输出 JSON）
+- timeout=12s（失败由外层走 pro 兜底）
 """
 
 from __future__ import annotations
 
+import json
 import logging
 import time
 from typing import Any
@@ -22,20 +22,29 @@ from . import _prompt
 
 logger = logging.getLogger(__name__)
 
-_DEFAULT_TIMEOUT = 20
+_DEFAULT_TIMEOUT = 15
+_DEFAULT_MAX_TOKENS = 350
+
+
+def _strip_code_fence(s: str) -> str:
+    s = s.strip()
+    if s.startswith("```"):
+        lines = s.split("\n")
+        if lines and lines[0].startswith("```"):
+            lines = lines[1:]
+        if lines and lines[-1].strip().startswith("```"):
+            lines = lines[:-1]
+        s = "\n".join(lines).strip()
+    return s
 
 
 def call_fast_json(
     img_url: str,
     *,
     timeout: int = _DEFAULT_TIMEOUT,
-    max_tokens: int | None = None,
+    max_tokens: int = _DEFAULT_MAX_TOKENS,
 ) -> dict[str, Any] | None:
-    """调用 vision client 返回结构化 dict；失败/非 JSON 返回 None。
-
-    max_tokens 默认 None：不显式传参，使用 client 内 capability 的 DB 配置；
-    显式传入时作为覆盖。
-    """
+    """调用 vision client 返回结构化 dict；失败/非 JSON 返回 None。"""
     t0 = time.time()
 
     try:
@@ -63,58 +72,43 @@ def call_fast_json(
     ]
 
     try:
-        call_kwargs: dict[str, Any] = {
-            "messages": messages,
-            "images": None,  # 图片已在 messages 中
-            "temperature": 0.1,
-            "timeout": timeout,
-            "enable_thinking": False,
-            "response_format": {"type": "json_object"},
-        }
-        if max_tokens is not None:
-            call_kwargs["max_tokens"] = max_tokens
-
-        # 双重防护：第1次正常调用；第2次去掉json_object强约束（部分模型在该约束下
-        # 反而幻觉），并加严格指令。解析全部走 json_utils，截断partial产物可用。
-        from .json_utils import extract_json_object
-
-        raw = None
-        obj = None
-        for _outer in range(2):
-            kw = dict(call_kwargs)
-            if _outer == 1:
-                kw.pop("response_format", None)
-                msgs2 = [dict(messages[0]), dict(messages[1])]
-                cont = list(msgs2[1]["content"])
-                cont = [dict(c) for c in cont]
-                cont[-1] = {"type": "text", "text": user_prompt + "\n严格只输出JSON对象，不要解释或markdown。"}
-                msgs2[1] = {"role": "user", "content": cont}
-                kw["messages"] = msgs2
-            raw = client.vision_completion(**kw)
-            if not raw:
-                logger.warning("[vision.v2] fast_json 返回空 outer=%s", _outer)
-                continue
-            obj = extract_json_object(raw)
-            if obj is not None:
-                break
-            logger.warning(
-                "[vision.v2] fast_json 非JSON(100字) outer=%s: %s",
-                _outer,
-                raw[:100],
-            )
-
+        raw = client.vision_completion(
+            messages=messages,
+            images=None,  # 图片已在 messages 中
+            temperature=0.1,
+            max_tokens=max_tokens,
+            timeout=timeout,
+            enable_thinking=False,
+            response_format={"type": "json_object"},
+        )
         elapsed = time.time() - t0
-        if obj is None:
-            logger.warning("[vision.v2] fast_json 两次均未得到JSON elapsed=%.1fs", elapsed)
+        if not raw:
+            logger.warning("[vision.v2] fast_json 返回空 elapsed=%.1fs", elapsed)
             return None
-        if obj.get("_partial"):
-            logger.warning("[vision.v2] fast_json 返回截断JSON(partial) elapsed=%.1fs", elapsed)
+
         logger.info(
-            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs has_person=%s type=%s",
+            "[vision.v2] fast_json 完成 model=%s elapsed=%.1fs",
             client.model,
             elapsed,
+        )
+        text = _strip_code_fence(raw)
+        lpos, r = text.find("{"), text.rfind("}")
+        if lpos >= 0 and r > lpos:
+            text = text[lpos : r + 1]
+        try:
+            obj = json.loads(text)
+        except json.JSONDecodeError:
+            logger.warning("[vision.v2] fast_json JSON 解析失败 elapsed=%.1fs head=%s", elapsed, raw[:200])
+            return None
+        if not isinstance(obj, dict):
+            logger.warning("[vision.v2] fast_json 非 dict: %s", type(obj))
+            return None
+        logger.info(
+            "[vision.v2] fast_json 完成 elapsed=%.1fs has_person=%s has_product=%s category=%s",
+            elapsed,
             obj.get("has_person"),
-            obj.get("type"),
+            obj.get("has_product"),
+            obj.get("category"),
         )
         return obj
     except Exception as e:

packages/application/cosyvoice_service.py

@@ -369,7 +369,7 @@ class CosyVoiceService:
         self._api_key = api_key or _router_key or settings.cosyvoice_api_key
         self._base_url = base_url or _router_url or settings.cosyvoice_base_url
         self._model = model or _router_model or settings.cosyvoice_model
-        self._clone_model = clone_model or getattr(settings, "cosyvoice_clone_model", "")
+        self._clone_model = clone_model or getattr(settings, "cosyvoice_clone_model", "voice-enrollment")
         self._audio_url_signer = audio_url_signer
 
         # base_url 规范化：去掉末尾的路径残留（兼容旧版配置）

packages/application/viral_video/prompts.py

@@ -110,12 +110,10 @@ _INTENT_SYSTEM = f"""你负责理解用户的营销意图。用户给的文案
 
 _INTENT_USER = """用户原始文案：{user_copy_text}
 所属行业：{industry}
-营销目的：{marketing_purpose}
 图片分析结果（供参考）：
 {image_analysis}
-图片类型推断：{image_category_hint}
 
-请理解用户意图，按标签格式输出。注意：theme和emotion_tone应与图片类型和营销目的匹配——门店类图片偏向"门店探店/到店体验"，商品图偏向"好物分享/产品种草"，人物图偏向"穿搭/人物故事"。"""
+请理解用户意图，按标签格式输出。"""
 
 _INTENT_EXAMPLE = """<intent_summary>一款厨房去油污神器，喷一喷油污就掉</intent_summary>
 <core_messages>
@@ -178,7 +176,7 @@ _FUSION_EXAMPLE = """<title>厨房重油污，别再用洗洁精硬擦了</title
 <segment duration_sec="4" image_index="0">39块钱625ml，厨房重油污的可以试一瓶</segment>
 </script_segments>
 <voiceover_script>这油污我真的忍很久了，用洗洁精擦半天都没用。后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净。39块钱625ml，厨房重油污的可以试一瓶。</voiceover_script>
-<overview_theme>厨房好物分享·产品种草</overview_theme>
+<overview_theme>厨房油污清洁好物分享</overview_theme>
 <scene_and_lighting>简洁明亮的厨房台面场景，自然光从窗户洒入，色调温暖柔和，突出产品白色瓶身与去油污对比效果。</scene_and_lighting>
 <word_count>58</word_count>
 <estimated_duration>13</estimated_duration>"""
@@ -217,8 +215,6 @@ _STORYBOARD_USER = """目标时长：{duration}秒
 图片分析结果：
 {image_analysis}
 
-重要：overview_theme 必须与图片实际内容和营销目的匹配。门店/餐饮/服务类图片用"门店探店·到店体验"；商品图用"好物分享·产品种草"；人物图用"穿搭分享·人物故事"；场景图用"空间体验·场景氛围"。不要对所有图片都使用"好物分享"。
-
 请按标签格式输出分镜。"""
 
 _STORYBOARD_EXAMPLE = """<clips>

packages/application/viral_video/reviewer.py

@@ -101,7 +101,6 @@ class Reviewer:
             ],
             temperature=0.2,
             max_tokens=1024,
-            timeout=25,
         )
         if not raw:
             return None
@@ -248,7 +247,6 @@ class Reviewer:
             ],
             temperature=0.5,
             max_tokens=2048,
-            timeout=25,
         )
         if not raw:
             return self._rule_fix(fusion, review)

packages/config/base.py

@@ -96,7 +96,7 @@ class SharedSettings(BaseSettings):
     doubao_fast_model: str = ""
     doubao_base_url: str = ""
     doubao_timeout: int = 45
-    doubao_max_retries: int = 3
+    doubao_max_retries: int = 1
     doubao_vision_model: str = ""
     doubao_vision_lite_model: str = ""
     doubao_vision_use_lite: bool = True

packages/domain/viral_video.py

@@ -225,11 +225,8 @@ class ViralVideoJob:
             self.error_msg = ""
             self.completed_at = None
             self.heartbeat_at = None
-        _now = datetime.now(timezone.utc)
-        self.started_at = _now
-        self.heartbeat_at = _now
         self.status = ViralVideoStatus.RUNNING
-        self.updated_at = _now
+        self.updated_at = datetime.now(timezone.utc)
 
     def resume_from_copy_generated(self, edited_copy: str | None = None) -> None:
         """阶段2->阶段3：用户确认/编辑口播文案，开始跑 TTS+单次Seedance渲染。"""
@@ -238,20 +235,14 @@ class ViralVideoJob:
         if edited_copy and isinstance(self.copy_result, dict):
             self.copy_result = {**self.copy_result, "voiceover_script": edited_copy}
             self.generated_copy_text = edited_copy
-        _now = datetime.now(timezone.utc)
-        self.started_at = _now
-        self.heartbeat_at = _now
         self.status = ViralVideoStatus.RUNNING
-        self.updated_at = _now
+        self.updated_at = datetime.now(timezone.utc)
 
     def resume_from_confirm(self) -> None:
         if self.status != ViralVideoStatus.WAIT_USER_CONFIRM:
             raise ValueError(f"Cannot resume from {self.status}")
-        _now = datetime.now(timezone.utc)
-        self.started_at = _now
-        self.heartbeat_at = _now
         self.status = ViralVideoStatus.RUNNING
-        self.updated_at = _now
+        self.updated_at = datetime.now(timezone.utc)
 
     def mark_completed(self, video_url: str) -> None:
         self.status = ViralVideoStatus.COMPLETED

packages/shared/ai_client.py

@@ -183,22 +183,16 @@ class DoubaoClient:
         provider: str = "volcengine",
     ) -> None:
         settings = get_shared_settings()
-        self.provider: str = provider
-        if provider == "dashscope":
-            self.api_key: str = api_key or getattr(settings, "dashscope_api_key", "")
-            self.model: str = model or getattr(settings, "dashscope_model", "")
-            self.base_url: str = (base_url or getattr(settings, "dashscope_base_url", "")).rstrip("/")
-        else:  # volcengine (default)
-            self.api_key = api_key or settings.doubao_api_key
-            self.model = model or settings.doubao_model
-            self.base_url = (base_url or settings.doubao_base_url).rstrip("/")
+        self.api_key: str = api_key or settings.doubao_api_key
+        self.model: str = model or settings.doubao_model
+        self.base_url: str = (base_url or settings.doubao_base_url).rstrip("/")
         self.timeout: int = timeout or settings.doubao_timeout
         self.max_retries: int = max_retries or settings.doubao_max_retries
         self.max_tokens: int | None = max_tokens
         self.temperature: float | None = temperature
         self.extra_params: dict = extra_params or {}
+        self.provider: str = provider
         self.vision_model: str = settings.doubao_vision_model
-        self.last_finish_reason: str = ""
         self.vision_lite_model: str = settings.doubao_vision_lite_model
         self.fast_model: str = settings.doubao_fast_model
         self.embedding_model: str = settings.doubao_embedding_model
@@ -211,13 +205,6 @@ class DoubaoClient:
         # 最近一次图片生成的详细错误，供上层读取
         self.last_image_error: dict = {}
 
-    def _resolve_timeout(self, timeout) -> "httpx.Timeout":
-        """将整数超时转为 httpx.Timeout，区分 connect/read/write/pool，避免 read 卡到 TCP 120s 默认值."""
-        if isinstance(timeout, httpx.Timeout):
-            return timeout
-        t = int(timeout) if timeout else 60
-        return httpx.Timeout(connect=10, read=max(t, 10), write=10, pool=5)
-
     def embed_text(self, text: str, timeout: int | None = None) -> list[float] | None:
         """调用豆包文本 Embedding API，返回浮点向量；失败返回 None。"""
         if not self.is_available or not text or not text.strip():
@@ -268,7 +255,7 @@ class DoubaoClient:
         self,
         messages: list[dict[str, str]],
         temperature: float = 0.7,
-        max_tokens: int | None = None,
+        max_tokens: int = 1024,
         model: str | None = None,
         timeout: int | None = None,
         **kwargs,
@@ -278,7 +265,7 @@ class DoubaoClient:
         Args:
             messages: 对话消息列表，[{"role": "user"/"system"/"assistant", "content": "..."}]
             temperature: 采样温度，0-2，默认0.7
-            max_tokens: 最大生成token数，默认 None（使用实例 self.max_tokens DB 配置，兜底 1024）
+            max_tokens: 最大生成token数，默认1024
 
         Returns:
             模型返回的文本内容，失败返回 None
@@ -291,12 +278,11 @@ class DoubaoClient:
             "Authorization": f"Bearer {self.api_key}",
             "Content-Type": "application/json",
         }
-        effective_max_tokens = max_tokens if max_tokens is not None else (self.max_tokens or 1024)
         payload: dict[str, Any] = {
             "model": model or self.model,
             "messages": messages,
             "temperature": temperature,
-            "max_tokens": effective_max_tokens,
+            "max_tokens": max_tokens,
         }
         # 合并实例级额外参数和调用方传入的额外参数
         if self.extra_params:
@@ -308,7 +294,7 @@ class DoubaoClient:
         _t0 = time.time()
         for attempt in range(self.max_retries + 1):
             try:
-                _req_timeout = self._resolve_timeout(timeout if timeout is not None else self.timeout)
+                _req_timeout = timeout if timeout is not None else self.timeout
                 response = httpx.post(
                     url,
                     headers=headers,
@@ -317,24 +303,7 @@ class DoubaoClient:
                 )
                 response.raise_for_status()
                 data = response.json()
-                finish_reason = (data.get("choices") or [{}])[0].get("finish_reason", "")
-                if finish_reason == "length" and attempt < self.max_retries:
-                    # 输出被 max_tokens 截断：2.0x 扩容后重试（计入 max_retries，不额外增加）
-                    old_max = int(payload["max_tokens"])
-                    new_max = int(old_max * 2)
-                    payload["max_tokens"] = new_max
-                    wait = 0.5 * (2**attempt)
-                    logger.warning(
-                        "输出被max_tokens截断(%d)，扩容到%d后重试 (第%d/%d次)",
-                        old_max,
-                        new_max,
-                        attempt + 1,
-                        self.max_retries + 1,
-                    )
-                    time.sleep(wait)
-                    continue
                 content = data["choices"][0]["message"]["content"]
-                self.last_finish_reason = finish_reason
                 _elapsed = time.time() - _t0
                 logger.info(
                     "[doubao] chat_completion 完成 model=%s tokens_in=%d tokens_out=%d elapsed=%.1fs attempt=%d timeout=%d",
@@ -367,7 +336,7 @@ class DoubaoClient:
         self,
         messages: list[dict],
         images: list[str] | None = None,
-        max_tokens: int | None = None,
+        max_tokens: int = 2048,
         temperature: float = 0.3,
         timeout: int | None = None,
         model: str | None = None,
@@ -376,12 +345,12 @@ class DoubaoClient:
         """调用豆包视觉理解 API（OpenAI 兼容多模态格式）.
 
         将 images 附加到最后一条 user message 的 content 中，
-        使用构造函数传入的 self.model（DB capability 绑定的视觉模型，默认 qwen-vl-plus）。
+        使用 vision_model（默认 doubao-1-5-vision-pro-250328）。
 
         Args:
             messages: 对话消息列表。最后一条 user message 会被注入图片内容。
             images: 图片列表，支持 base64 data URI 或 HTTP(S) URL。
-            max_tokens: 最大生成 token 数，默认 None（使用实例 self.max_tokens DB 配置，兜底 2048）。
+            max_tokens: 最大生成 token 数，默认 2048。
             temperature: 采样温度，默认 0.3（视觉任务偏低更稳定）。
             timeout: 单次请求超时秒数，不传则使用默认 self.timeout。
 
@@ -421,19 +390,18 @@ class DoubaoClient:
             "Authorization": f"Bearer {self.api_key}",
             "Content-Type": "application/json",
         }
-        effective_max_tokens = max_tokens if max_tokens is not None else (self.max_tokens or 2048)
         payload: dict[str, Any] = {
-            "model": model or self.model,
+            "model": model or self.vision_model,
             "messages": vision_messages,
             "temperature": temperature,
-            "max_tokens": effective_max_tokens,
+            "max_tokens": max_tokens,
         }
         if self.extra_params:
             payload.update(self.extra_params)
         if kwargs:
             payload.update(kwargs)
 
-        req_timeout = self._resolve_timeout(timeout or self.timeout)
+        req_timeout = timeout or self.timeout
         last_error: Optional[Exception] = None
         _t0 = time.time()
         for attempt in range(self.max_retries + 1):
@@ -446,24 +414,7 @@ class DoubaoClient:
                 )
                 response.raise_for_status()
                 data = response.json()
-                finish_reason = (data.get("choices") or [{}])[0].get("finish_reason", "")
-                if finish_reason == "length" and attempt < self.max_retries:
-                    # 视觉输出被 max_tokens 截断：2.0x 扩容后重试（计入 max_retries）
-                    old_max = int(payload["max_tokens"])
-                    new_max = int(old_max * 2)
-                    payload["max_tokens"] = new_max
-                    wait = 0.5 * (2**attempt)
-                    logger.warning(
-                        "视觉输出被max_tokens截断(%d)，扩容到%d后重试 (第%d/%d次)",
-                        old_max,
-                        new_max,
-                        attempt + 1,
-                        self.max_retries + 1,
-                    )
-                    time.sleep(wait)
-                    continue
                 content = data["choices"][0]["message"]["content"]
-                self.last_finish_reason = finish_reason
                 _elapsed = time.time() - _t0
                 logger.info(
                     "[doubao] vision_completion 完成 model=%s tokens_in=%d tokens_out=%d elapsed=%.1fs attempt=%d",

packages/shared/ai_router.py

@@ -62,15 +62,7 @@ class CapabilityConfig:
 class TTSClient:
     """TTS 客户端（简单配置持有者，实际调用由 CosyVoiceService 完成）"""
 
-    def __init__(
-        self,
-        provider: str,
-        api_key: str,
-        base_url: str,
-        model: str,
-        timeout: int = 60,
-        extra_params: dict | None = None,
-    ):
+    def __init__(self, provider: str, api_key: str, base_url: str, model: str, timeout: int = 60, extra_params: dict | None = None):
         self.provider = provider
         self.api_key = api_key
         self.base_url = base_url
@@ -86,15 +78,7 @@ class TTSClient:
 class ImageGenClient:
     """图片生成客户端（简单配置持有者）"""
 
-    def __init__(
-        self,
-        provider: str,
-        api_key: str,
-        base_url: str,
-        model: str,
-        timeout: int = 60,
-        extra_params: dict | None = None,
-    ):
+    def __init__(self, provider: str, api_key: str, base_url: str, model: str, timeout: int = 60, extra_params: dict | None = None):
         self.provider = provider
         self.api_key = api_key
         self.base_url = base_url
@@ -110,15 +94,7 @@ class ImageGenClient:
 class VideoGenClient:
     """视频生成客户端（简单配置持有者）"""
 
-    def __init__(
-        self,
-        provider: str,
-        api_key: str,
-        base_url: str,
-        model: str,
-        timeout: int = 600,
-        extra_params: dict | None = None,
-    ):
+    def __init__(self, provider: str, api_key: str, base_url: str, model: str, timeout: int = 600, extra_params: dict | None = None):
         self.provider = provider
         self.api_key = api_key
         self.base_url = base_url
@@ -277,27 +253,13 @@ class AIRouter:
             return config
 
     def _get_model_or_fallback(self, cap: CapabilityConfig, variant: str = "primary") -> ModelConfig | None:
-        """按 variant 选择模型，不存在则降级。
-
-        - primary: primary → fallback
-        - lite: lite → primary
-        - fallback: fallback → primary（修复点：此前 fallback variant 被忽略，错误地使用了 primary 模型）
-        """
-        if variant == "fallback":
-            if cap.fallback_model:
-                return cap.fallback_model
-            if cap.primary_model:
-                return cap.primary_model
-        elif variant == "lite":
-            if cap.lite_model:
-                return cap.lite_model
-            if cap.primary_model:
-                return cap.primary_model
-        else:  # primary
-            if cap.primary_model:
-                return cap.primary_model
-            if cap.fallback_model:
-                return cap.fallback_model
+        """按 variant 选择模型，不存在则 fallback"""
+        if variant == "lite" and cap.lite_model:
+            return cap.lite_model
+        if cap.primary_model:
+            return cap.primary_model
+        if cap.fallback_model:
+            return cap.fallback_model
         return None
 
     # ── 构建客户端 ─────────────────────────────────────────────────────────
@@ -467,8 +429,8 @@ class AIRouter:
         api_key = getattr(settings, "cosyvoice_api_key", "")
         if not api_key:
             return None
-        base_url = getattr(settings, "cosyvoice_base_url", "")
-        model = getattr(settings, "cosyvoice_model", "")
+        base_url = getattr(settings, "cosyvoice_base_url", "https://dashscope.aliyuncs.com/api/v1")
+        model = getattr(settings, "cosyvoice_model", "cosyvoice-v3-flash")
 
         return TTSClient(provider="dashscope", api_key=api_key, base_url=base_url, model=model)
 
@@ -477,8 +439,8 @@ class AIRouter:
         api_key = getattr(settings, "doubao_api_key", "")
         if not api_key:
             return None
-        base_url = getattr(settings, "doubao_base_url", "")
-        model = getattr(settings, "doubao_image_model", "")
+        base_url = getattr(settings, "doubao_base_url", "https://ark.cn-beijing.volces.com/api/v3")
+        model = getattr(settings, "doubao_image_model", "doubao-seedream-5-0-flash-260915")
 
         return ImageGenClient(
             provider="volcengine",
@@ -493,8 +455,8 @@ class AIRouter:
         api_key = getattr(settings, "doubao_api_key", "")
         if not api_key:
             return None
-        base_url = getattr(settings, "doubao_base_url", "")
-        model = getattr(settings, "doubao_video_model", "")
+        base_url = getattr(settings, "doubao_base_url", "https://ark.cn-beijing.volces.com/api/v3")
+        model = getattr(settings, "doubao_video_model", "doubao-seedance-2-5-260628")
 
         return VideoGenClient(
             provider="volcengine",

tests/unit/test_api_settings.py

@@ -81,7 +81,7 @@ class TestSharedSettingsDefaults:
         s = SharedSettings()
         assert s.doubao_model == ""  # 零硬编码：默认值已清空
         assert s.doubao_timeout == 45  # #2180 默认提到45s
-        assert s.doubao_max_retries == 3
+        assert s.doubao_max_retries == 1
 
 
 class TestAPISettingsDefaults:

tests/unit/test_config_base.py

@@ -111,7 +111,7 @@ class TestSharedSettingsDefaults:
         """豆包默认配置"""
         s = self._make_settings()
         assert s.doubao_timeout == 45  # #2180 默认提到45s
-        assert s.doubao_max_retries == 3
+        assert s.doubao_max_retries == 1
         assert s.doubao_base_url == ""  # 零硬编码：默认值已清空
 
     def test_default_empty_api_keys(self):