From d5f6e9499d22b446ceeba376b10592833b2c7ba3 Mon Sep 17 00:00:00 2001 From: xiaoxia-saas-bot Date: Sun, 4 Oct 2026 16:41:05 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat(viral=5Fvideo):=20#2040=20=E6=8E=A5?= =?UTF-8?q?=E5=85=A5Prompt=E6=A8=A1=E6=9D=BF=E7=B3=BB=E7=BB=9F+=E5=88=A0?= =?UTF-8?q?=E9=99=A4=E5=86=97=E4=BD=99=E7=A1=AC=E7=BC=96=E7=A0=81prompt?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/worker/worker_app/tasks/viral_video.py | 588 +++++++++----------- packages/application/viral_video/prompts.py | 46 +- 2 files changed, 313 insertions(+), 321 deletions(-) diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index 4d410c0b9..95b798695 100644 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -47,10 +47,8 @@ from packages.shared import get_shared_settings logger = logging.getLogger(__name__) - # ── WS 进度推送 ────────────────────────────────────────────────────────── - def _emit_progress( job_id: str, stage: str, @@ -76,22 +74,18 @@ def _emit_progress( except Exception as e: logger.warning("[爆款视频] WS 进度推送失败: %s", e) - # ── 仓储辅助 ──────────────────────────────────────────────────────────── - def _get_repo_and_job(job_id: str): session = SessionLocal() repo = SQLAlchemyViralVideoJobRepository(session) job = repo.get(job_id) return session, repo, job - def _save_job(repo, job, session): repo.update(job) session.commit() - def _start_trust_chain_preheat(job_id: str, portrait_descriptions: list[str]) -> None: """#2172/#2174 后台启动信任链预热（Seedream t2i 文生图人像），不阻塞调用方。 @@ -157,7 +151,6 @@ def _start_trust_chain_preheat(job_id: str, portrait_descriptions: list[str]) -> t = threading.Thread(target=_run_preheat, name=f"tc-preheat-{job_id[:8]}", daemon=True) t.start() - def _set_stage(job, repo, session, stage: str, message: str, persist: bool = True) -> None: """更新细粒度阶段并持久化到 DB，同时通过 Redis 推送进度事件。 @@ -173,7 +166,6 @@ def _set_stage(job, repo, session, stage: str, message: str, persist: bool = Tru except Exception as e: # 阶段持久化失败不阻塞主流程 logger.warning("[爆款视频] 阶段持久化失败 stage=%s err=%s", stage, e) - # ── worker 心跳（僵尸任务检测） ───────────────────────────────────────── # 心跳间隔（秒）；超过此时间未更新 heartbeat_at 视为 worker 异常 @@ -183,7 +175,6 @@ _STALE_RUNNING_TIMEOUT_SEC = 10 * 60 # 10 分钟 # 心跳过期窗口：heartbeat_at 距 now 超过此时长视为失效 _HEARTBEAT_EXPIRE_SEC = 2 * 60 # 2 分钟 - def _heartbeat_once(job_id: str) -> None: """在独立 session 中更新一次 heartbeat_at（不捕获主流程事务状态）。""" ssn = None @@ -208,7 +199,6 @@ def _heartbeat_once(job_id: str) -> None: except Exception: pass - def _start_heartbeat_thread(job_id: str) -> tuple[threading.Event, threading.Thread]: """启动后台心跳线程，每 _HEARTBEAT_INTERVAL_SEC 秒更新一次 heartbeat_at。 返回 (stop_event, thread)；任务结束时调用 stop_event.set() 停止心跳。 @@ -225,7 +215,6 @@ def _start_heartbeat_thread(job_id: str) -> tuple[threading.Event, threading.Thr t.start() return stop, t - def _recover_stale_jobs() -> int: """启动/定时扫描：把僵尸任务（running 超时且心跳停止）标记为 failed。 返回本次回收的任务数。可由 celery beat 周期性调用，也可在任务启动前顺带扫一次。 @@ -263,7 +252,6 @@ def _recover_stale_jobs() -> int: except Exception: pass - # ── 默认结构 ───────────────────────────────────────────────────────────── _DEFAULT_HARD_CONSTRAINTS = [ @@ -294,7 +282,6 @@ _DEFAULT_NEGATIVE_PROMPTS = [ "低分辨率", ] - def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict: return { "overview": {"theme": "好物推荐", "total_duration": duration, "aspect_ratio": ratio}, @@ -308,38 +295,8 @@ def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict: "title": "", } - # ── 流水线各步骤 ──────────────────────────────────────────────────────── - -_IMAGE_ANALYSIS_SYSTEM_PROMPT = """你是电商商品视觉分析师，从商品图片中提取关键商品信息。严格规则： -1. 只说图片里真实可见的内容，看不清/没有的填「无法判断」，不要瞎猜。 -2. 输出必须是严格 JSON（不要 Markdown 代码块，不要额外解释文字）。 -3. 字段说明： -{ - "name": "商品全名（品牌+产品名+规格，如『大公鸡头管家 多功能油污净 625ml』，从包装 OCR 读出）", - "brand": "品牌名（从 Logo/包装文字读出，看不清填『无法判断』）", - "category": "商品品类（如『家用清洁/油污清洁剂』『日化/洗衣液』；非产品图填『非产品图』）", - "appearance": "外观特征（50-100字：瓶身形状、颜色、瓶盖、标签颜色、尺寸感）", - "packaging": "包装细节（50-100字：标签分区、图案元素、瓶盖/泵头样式、塑封状态）", - "text_on_package": ["包装上清晰可见的文字列表（品牌、产品名、卖点、规格等，看不清的不列）"], - "key_features": [ - "3-5 条图片中能看到的外观/视觉特征（如『红色瓶盖白色瓶身』『鸡头图案 Logo』等）" - ], - "scene": "图片场景（如白底棚拍/浴室实拍/桌面静物/手持实拍等）", - "portrait_prompt": "如果图片中有清晰人物面部，用60-100字中文描述该人物外貌（性别、年龄段、发型/发色、肤色、脸型、五官特征、当前穿着、表情姿态），用于AI生图参考；没有人物或看不清面部填「无人像」", - "summary": "100-180字中文导购描述，连贯自然段落，像电商详情页介绍，前端直接展示，必须提到品牌/品名/核心外观特征，不能写『无法判断』" -}""" - -_IMAGE_ANALYSIS_USER_PROMPT = """请分析这张商品图片，输出严格 JSON。重点： -1. name/brand/text_on_package 从图片包装 OCR 读取，不编造； -2. appearance/packaging 各写 50-100 字，要具体； -3. portrait_prompt：有人物时详细描述外貌（性别/年龄/发型/肤色/穿着/表情）用于AI人像生成参考，无人像填「无人像」； -4. summary 必须是 100-180 字连贯中文段落，说清商品是什么、长什么样、适合谁用，不要写「无法判断」； -5. 非产品图时 category 填「非产品图」，name 填实际看到的内容； -6. 看不清的字段填「无法判断」。""" - - def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict: d = { "name": "未识别", @@ -357,7 +314,6 @@ def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict: d.update(extra) return d - def _is_vision_result_usable(result: dict) -> bool: """判断 VLM 返回是否有效：name/summary 不能为未识别/无法判断/空，summary 要够长。""" if not isinstance(result, dict): @@ -376,7 +332,6 @@ def _is_vision_result_usable(result: dict) -> bool: return False return True - def _analyze_single_image( idx: int, img_url: str, @@ -385,88 +340,97 @@ def _analyze_single_image( *, pro_fallback_model: str | None = None, ) -> dict: - """单张图片 VLM 分析（线程池并行调用）。 - - lite 失败/结果不可用 时自动用 pro 模型降级重试 1 次。 - - 失败/None/不可用最终返回含默认字段的 dict（不会让用户看到「未识别·无法判断」裸结果）。 + """单张图片 VLM 分析（#2040：改为从 prompt_loader 读模板 + XML 解析）。 + + lite 失败/不可用时用 pro 降级重试 1 次。失败/None 最终返回含默认字段的 dict。 """ - from packages.shared.ai_service import call_vision + try: + from packages.shared.ai_service import call_vision + from packages.application.viral_video.prompt_loader import ( + get_template, render_system_prompt, render_user_prompt, + ) + from packages.application.viral_video import xml_parser as xp + except ImportError as e: + logger.warning("[爆款视频] prompt 模板/解析模块不可用: %s", e) + return _vision_fallback(idx, f"fallback_import_error:{e}") if not img_url or not isinstance(img_url, str): return _vision_fallback(idx, "invalid_url") + template = get_template("image_analysis") + system = render_system_prompt(template) + user = render_user_prompt( + template, image_count=1, industry="通用", + image_urls=f"第1张：{img_url}", + ) + def _call(model: str, tmo: int): try: return call_vision( - image_url=img_url, - prompt=_IMAGE_ANALYSIS_USER_PROMPT, - model=model, - max_tokens=800, - temperature=0.1, - timeout=tmo, - system_prompt=_IMAGE_ANALYSIS_SYSTEM_PROMPT, + image_url=img_url, prompt=user, model=model, + max_tokens=2048, temperature=0.3, timeout=tmo, + system_prompt=system, ) except Exception as e: logger.warning("[爆款视频] 图片 #%d call_vision(%s) 异常 err=%s", idx, model, e) return None + def _xml_to_product(nodes: list, raw_text: str) -> dict: + product_nodes = [n for n in nodes if n["tag"] == "product"] + scene = xp.text_of(raw_text, "scene") or "通用" + mood = xp.text_of(raw_text, "mood") or "" + for p in product_nodes: + a = p["attrs"] + text_on_pkg = a.get("text_on_package", "") + text_list = [x.strip() for x in re.split(r"[,，;；]", text_on_pkg) if x.strip()] if text_on_pkg else [] + features = a.get("features", "") + feat_list = [x.strip() for x in re.split(r"[,，;；]", features) if x.strip()] if features else [] + name = a.get("name", "") or "未识别" + brand = a.get("brand", "") or "无法判断" + category = a.get("category", "") or "无法判断" + appearance = a.get("appearance", "") or "无法判断" + packaging = a.get("packaging", "") or "无法判断" + summary = a.get("summary", "") or f"{brand} {name}" + return { + "name": name, "brand": brand, "category": category, + "appearance": appearance, "packaging": packaging, + "text_on_package": text_list, + "key_features": feat_list or [features] if features else ["无法判断"], + "scene": scene, "mood": mood, + "portrait_prompt": a.get("portrait_prompt", "无人像"), + "summary": summary, "_source": "xml", + } + return _vision_fallback(idx, "no_product_tag") + def _normalize(raw, source: str) -> dict: if raw is None: return _vision_fallback(idx, f"{source}_none") - if isinstance(raw, str): - logger.warning("[爆款视频] 图片 #%d VLM(%s) 返回非 JSON: %s", idx, source, raw[:200]) - return _vision_fallback(idx, f"{source}_text", {"_raw": raw[:500]}) - if not isinstance(raw, dict): + if not isinstance(raw, str): return _vision_fallback(idx, f"{source}_badtype") - raw.setdefault("_source", source) - raw.setdefault("name", "未识别") - raw.setdefault("brand", "无法判断") - raw.setdefault("category", "无法判断") - raw.setdefault("appearance", "无法判断") - raw.setdefault("packaging", "无法判断") - raw.setdefault("text_on_package", []) - raw.setdefault("key_features", []) - raw.setdefault("scene", "通用") - raw.setdefault("portrait_prompt", "无人像") - raw.setdefault("summary", "") - if not isinstance(raw.get("text_on_package"), list): - raw["text_on_package"] = [] - if not isinstance(raw.get("key_features"), list): - raw["key_features"] = [] - return raw + nodes = xp.parse_tags(raw) + if not nodes: + logger.warning("[爆款视频] 图片 #%d XML 解析失败 source=%s", idx, source) + return _vision_fallback(idx, f"{source}_xml_fail", {"_raw": raw[:500]}) + product = _xml_to_product(nodes, raw) + product.setdefault("_source", source) + product["raw"] = raw[:500] + return product - # 第一次：传入模型（通常是 lite） first_raw = _call(vision_model, timeout) tag1 = vision_model.split("/")[-1] if "/" in vision_model else vision_model first_result = _normalize(first_raw, tag1) if _is_vision_result_usable(first_result): return first_result - logger.warning( - "[爆款视频] 图片 #%d VLM(%s) 结果不可用 name=%r summary_len=%d，尝试 pro 降级", - idx, - vision_model, - first_result.get("name"), - len(first_result.get("summary") or ""), - ) - - # 第二次：pro 降级重试 if pro_fallback_model and pro_fallback_model != vision_model: pro_raw = _call(pro_fallback_model, 60) # #2180: pro VLM 实测也需25-38s，原25s太短，提到60s pro_result = _normalize(pro_raw, "pro_fallback") if _is_vision_result_usable(pro_result): pro_result["_fallback_used"] = True return pro_result - logger.warning( - "[爆款视频] 图片 #%d pro 降级仍不可用 name=%r summary_len=%d", - idx, - pro_result.get("name"), - len(pro_result.get("summary") or ""), - ) return pro_result - return first_result - def _step_image_analysis(job: ViralVideoJob) -> dict: """步骤 1: 图片 VLM 分析 — 识别产品特征（v1.6 优化：并行 + lite 模型提速）。""" try: @@ -522,7 +486,6 @@ def _step_image_analysis(job: ViralVideoJob) -> dict: return {"products": results} - def _step_video_analysis(job: ViralVideoJob) -> dict | None: """步骤 1.5: 参考视频风格分析（可选）。""" if not job.reference_video_url: @@ -546,11 +509,14 @@ def _step_video_analysis(job: ViralVideoJob) -> dict | None: logger.error("[爆款视频] 视频风格分析失败: %s", e) return {"error": str(e), "source": "failed"} - def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: - """步骤 2: 用户文案意图解析。""" + """步骤 2: 用户文案意图解析（#2040：改为模板 + XML 解析）。""" try: from packages.shared.ai_service import call_llm + from packages.application.viral_video.prompt_loader import ( + get_template, render_system_prompt, render_user_prompt, + ) + from packages.application.viral_video import xml_parser as xp except ImportError: return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""} @@ -572,25 +538,25 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: feat_str = ", ".join([str(x) for x in feats + extras]) products_summary += f"- {p.get('name', '产品')}: {feat_str}\n" - prompt = f"""你是一个营销编导。请分析以下信息，理解用户的营销意图并给出短视频主题建议： + template = get_template("intent_parsing") + system = render_system_prompt(template) + user = render_user_prompt( + template, + user_copy_text=job.user_copy_text or "（未提供，全由 AI 创作）", + industry=job.industry or "未指定", + image_analysis=products_summary or "- （无图片分析结果）", + ) -用户原始文案：{job.user_copy_text or "（未提供，全由 AI 创作）"} -行业：{job.industry or "未指定"} -目标客户：{job.target_customer or "未指定"} -营销目的：{job.marketing_purpose or "未指定"} -视频时长：{job.duration}秒 -产品信息： -{products_summary or "- （无图片分析结果）"} + def _parse(raw: str) -> dict: + summary = xp.text_of(raw, "intent_summary") + msgs = [n["text"] for n in xp.find_all(raw, "message") if n["text"]] + tone = xp.text_of(raw, "emotion_tone") or "亲切自然" + title = xp.text_of(raw, "suggested_title") or xp.text_of(raw, "title") + return {"intent": summary or "推广产品", "key_messages": msgs or ["产品亮点"], "tone": tone, "suggested_title": title} -请返回严格 JSON（不要 Markdown，不要解释）： -{{ - "intent": "核心营销意图（一句话）", - "key_messages": ["要传达的3-5个关键信息"], - "tone": "文案调性（如亲切/专业/高端/活力/治愈/搞笑）", - "target_emotion": "希望触发的用户情感", - "call_to_action": "行动号召短句（口语化，5-10字）", - "suggested_title": "视频主题标题（5-15字）" -}}""" + def _fallback(raw: str) -> dict: + t = (job.user_copy_text or "").strip() + return {"intent": t[:30] or "推广产品", "key_messages": [t[:80]] if t else ["产品亮点"], "tone": "专业", "suggested_title": ""} _s = get_shared_settings() _fast = _s.doubao_fast_model @@ -598,22 +564,17 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: for _m, _lbl in [(_fast, "fast"), (_pro, "pro-fallback")]: try: logger.info("[爆款视频] 意图解析 model=%s label=%s", _m, _lbl) - result = call_llm(prompt, temperature=0.4, max_tokens=800, model=_m, timeout=45) - return ( - result - if isinstance(result, dict) - else {"intent": str(result)[:200], "key_messages": [], "tone": "专业", "suggested_title": ""} - ) + raw = call_llm([{"role": "system", "content": system}, {"role": "user", "content": user}], + temperature=0.4, max_tokens=1024, model=_m, timeout=45) # #2180: 意图解析 LLM 实测需更长响应，原25s太紧 + if not raw: + continue + parsed = _parse(raw) + if parsed["intent"] or parsed["key_messages"]: + return parsed except Exception as e: logger.warning("[爆款视频] 意图解析失败 label=%s err=%s", _lbl, e) - continue - return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""} + return _fallback("") - -# ── 编导分镜脚本生成（核心，v1.6 新 prompt） ────────────────────────────── - - -# 人设 IP 类型 → 文案/出镜风格指导（前端下拉 10 个 IP 类型） _PERSONA_STYLE_GUIDE = { "通用个人IP": "亲切自然、像朋友分享好物，第一人称口语化，不端着", "老板型IP": "沉稳大气、有行业格局感，适度使用『我做了XX年』『我一直坚持』等老板视角，语气自信不夸张", @@ -627,7 +588,6 @@ _PERSONA_STYLE_GUIDE = { "测评种草型": "真实测评感、讲使用体验和优缺点对比，带『亲测』『我用了XX天』『实测下来』真实感词汇", } - def _persona_style_hint(persona_id: str) -> str: """根据 persona_id 查文案风格指导；未命中/空值返回通用提示。""" pid = (persona_id or "").strip() @@ -638,74 +598,6 @@ def _persona_style_hint(persona_id: str) -> str: return f"【人设风格：{pid}】按该人设的口吻、话术习惯组织口播和出镜动作" return "【人设风格：未指定】亲切自然、像朋友分享好物" - -_SCRIPT_GENERATION_PROMPT = """你是资深短视频导演，为 Seedance 2.5（单次生成最多{duration}秒）写编导分镜脚本。脚本将整体作为 prompt 一次性传给视频模型，必须让模型在连贯镜头流中清楚每段时间拍什么、画面如何、人物说什么。 - -## 产品 -{products_summary} - -## 营销参数 -- 主题/意图：{intent} -- 关键信息：{key_messages} -- 调性：{tone} -- 目标客户：{target_customer} -- 用户原始卖点（必须融入口播）：{user_copy} -- 时长：{duration}秒 / 画幅：{ratio} / 产品图：{n_images}张（第1张通常是主图/首帧） -- 风格参考：{style_hint} -- 爆款结构（必须严格遵循节奏/段落顺序）：{viral_structure_block} -- 人设/出镜口吻（必须贯穿全部对白和动作描写）：{persona_hint} - -## 输出格式（必须输出严格 JSON，不要 Markdown，不要解释，字段一个都不能少） - -```json -{{ - "overview": {{ - "theme": "视频主题（一句话概括）", - "total_duration": {duration}, - "aspect_ratio": "{ratio}" - }}, - "scene_and_lighting": "整体场景描述+光线设定（100-200字，要具体：在哪拍、什么光线、什么色调、什么氛围）", - "shots": [ - {{ - "time_range": "0-3秒", - "shot_type_angle_movement": "景别+角度+运镜（例：近景俯拍45度，缓慢推镜；中景平视，固定镜头；特写平视，快速拉镜）", - "scene_and_dialogue": "画面场景描述 + 人物口播台词（对白要自然口语化，像朋友聊天，不要硬广推销腔）", - "action_details": "人物动作、表情、物品操作细节（手怎么动、表情变化、产品怎么展示）", - "audio_bgm": "环境音+BGM提示（例：轻快流行BGM，环境嘈杂咖啡店背景音）", - "transition": "硬切/淡入淡出/叠化（最后一镜写『结束』即可）", - "reference_image_index": 0 - }} - // ... 按时间顺序列出所有镜头，总时长累计 = {duration} 秒 - ], - "hard_constraints": [ - "无字幕、无水印、无任何自动生成文字、无logo", - "同一人物全程五官、发型、服装、身材保持一致，不得换脸变形", - "口播语音在总时长内自然念完，语速自然，口型与语音严格同步", - "画面流畅无闪烁、无多余肢体、无扭曲变形、无穿模", - "色彩自然、曝光正确、电影级质感、高清细节" - ], - "negative_prompts": [ - "字幕","自动字幕","水印","logo","图标","错误文字","乱码文字", - "男女声错配","中途换声","五官崩坏","脸部变形","多余手指", - "肢体扭曲","闪烁","画面抖动","模糊","低分辨率" - ], - "voiceover_script": "完整口播稿（把 shots 里所有对白自然拼接成一段，口语化，不加旁白标注、不加镜头标注、不加'主播：'之类前缀，就是纯念出来的文本，长度适配{duration}秒，约{approx_chars}字）" -}} -``` - -## 关键要求 -1. 每镜写清景别/角度/运镜（特写/近景/中景+平视/俯拍+推/拉/固定）。 -2. 画面具体：主体（性别/年龄/穿着）、场景、动作、光线、镜头运动要可落地。 -3. 对白自然口语化，像朋友分享好物；拒绝"家人们""宝子们""太好用了"等浮夸/硬广腔。 -4. reference_image_index 填 0-based 索引（产品特写用索引0主图），人像/场景可 null。 -5. shots time_range 累计={duration}秒，单镜2-8秒。 -6. hard_constraints/negative_prompts 保留默认项可追加，不要删减。 -7. voiceover_script 为纯口播文本（无标记/括号/前缀），{duration}秒约{approx_chars}字。 -8. 严格按上方「爆款结构」的节奏/段落顺序编排（钩子/痛点/反转/案例/行动号召与结构对齐）。 -9. 输出前自检：口播对白禁止错别字和语病，**严禁使用"很近"，正确用词是"最近"**（指"最近一段时间/最近在用"，绝不能写成"很近"）；其他同音字、形近字错误一律修正。 -10. 必须使用产品信息中真实的品牌、品名和外观特征，不要编造与产品无关的内容。""" - - def _build_products_summary(image_analysis: dict) -> str: """把 VLM 返回的商品分析结果拼给文案/分镜生成 prompt 用。 优先用 summary（自然段落）；没有时用结构化字段兜底拼一段。""" @@ -774,7 +666,6 @@ def _build_products_summary(image_analysis: dict) -> str: lines.append("- " + "，".join(parts)) return "\n".join(lines) - def _safe_json_loads(raw: str | dict | list | None): if raw is None: return None @@ -801,7 +692,6 @@ def _safe_json_loads(raw: str | dict | list | None): pass return None - def _replace_henjin_everywhere(obj: Any) -> Any: """递归遍历 copy_result 里所有字符串值，把'很近'替换成'最近'。 覆盖 overview.theme、scene_and_lighting、voiceover_script、 @@ -817,7 +707,6 @@ def _replace_henjin_everywhere(obj: Any) -> Any: return {k: _replace_henjin_everywhere(v) for k, v in obj.items()} return obj - def _fallback_script(job: ViralVideoJob) -> dict: """脚本生成失败时的兜底脚本（极简但可用）。""" dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) @@ -842,7 +731,6 @@ def _fallback_script(job: ViralVideoJob) -> dict: base["title"] = "好物分享" return base - def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict: """把 LLM 返回的脚本规范化、补默认、校验结构。""" dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) @@ -939,11 +827,68 @@ def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict: base["title"] = base["overview"]["theme"] return base +def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None: + """把 LLM 返回的 XML 分镜规范化为旧 copy_result 结构（供 Seedance 使用）。""" + from packages.application.viral_video import xml_parser as xp + dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) + ratio = getattr(job, "video_ratio", None) or "9:16" + 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 "好物分享" + est = xp.attr_int(xp.text_of(raw, "estimated_duration"), 0) + if est: + base["overview"]["total_duration"] = est + sl = xp.text_of(raw, "scene_and_lighting") + if sl: + base["scene_and_lighting"] = sl + clips = xp.find_all(raw, "clip") + shots: list[dict] = [] + voice_parts: list[str] = [] + for i, c in enumerate(clips): + a = c["attrs"] + body = c.get("text", "") or "" + ref_idx_raw = a.get("reference_image_index", "") + if ref_idx_raw in (None, "", "null", "None"): + body_ref = xp.text_of(body, "reference_image_index") if body else "" + ref_idx = xp.attr_int(body_ref, 0) if body_ref else None + else: + ref_idx = xp.attr_int(ref_idx_raw, 0) + shot = { + "time_range": a.get("time_range") or f"{i * 3}-{(i + 1) * 3}秒", + "shot_type_angle_movement": (xp.text_of(body, "shot_type_angle_movement") if body else "") or a.get("shot_type_angle_movement", "") or "中景平视，固定镜头", + "scene_and_dialogue": (xp.text_of(body, "scene_and_dialogue") if body else "") or "", + "action_details": (xp.text_of(body, "action_details") if body else "") or "", + "audio_bgm": (xp.text_of(body, "audio_bgm") if body else "") or a.get("bgm_note", "") or "轻快BGM", + "transition": (xp.text_of(body, "transition") if body else "") or a.get("transition", "") or ("硬切" if i < len(clips) - 1 else "结束"), + "reference_image_index": ref_idx, + } + voice = (xp.text_of(body, "voice_text") if body else "") + if voice: + voice_parts.append(voice) + if not shot["scene_and_dialogue"]: + shot["scene_and_dialogue"] = voice + shots.append(shot) + if not shots: + return None + base["shots"] = shots + joined = xp.text_of(raw, "voiceover_script") or "。".join(voice_parts) + base["voiceover_script"] = joined + base["final_copy"] = joined + base["suggested_copy"] = joined + base["title"] = base["overview"]["theme"] + return base def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: dict) -> dict: - """步骤 3: 编导分镜脚本生成（v1.6 核心，输出 copy_result 结构）。""" + """步骤 3: 编导分镜脚本生成（#2040：模板 + XML 解析；输出 copy_result 结构）。""" try: from packages.shared.ai_service import call_llm + from packages.application.viral_video.prompt_loader import ( + get_template, render_system_prompt, render_user_prompt, + ) + from packages.application.viral_video.prompts import ( + FUSION_INSTRUCTIONS, GLOBAL_CONSTRAINTS, NEGATIVE_RULES, + ) except ImportError: return _fallback_script(job) @@ -957,20 +902,15 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) ratio = getattr(job, "video_ratio", None) or "9:16" - approx_chars = max(20, dur * 4) - intent_str = "" - key_msgs = "" - tone = "" + intent_str = "推广产品" + key_msgs = "产品亮点" + tone = "亲切自然" if isinstance(intent, dict): - intent_str = intent.get("intent") or "推广产品" - key_msgs = "、".join(intent.get("key_messages") or []) - tone = intent.get("tone") or "亲切自然" - else: - intent_str = "推广产品" - tone = "亲切自然" + intent_str = intent.get("intent") or intent_str + key_msgs = "、".join(intent.get("key_messages") or []) or key_msgs + tone = intent.get("tone") or tone - # 爆款结构：用户在 STEP1/STEP2 选的中文结构名，必须严格注入 prompt 指导 AI 编排 _vs = (job.viral_structure or "").strip() if _vs: viral_structure_block = f"【{_vs}】—— 请严格按照这个爆款结构的节奏/段落顺序编排镜头、台词和情绪节点（开场钩子、痛点、反转、案例、行动号召等按结构走），不要打乱顺序" @@ -978,59 +918,56 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di viral_structure_block = "未指定（自由编排，但仍需有钩子开头+产品展示+行动号召的基本节奏）" persona_hint = _persona_style_hint(getattr(job, "persona_id", "")) - prompt = _SCRIPT_GENERATION_PROMPT.format( - products_summary=products_summary, - intent=intent_str, - key_messages=key_msgs or "产品亮点", - tone=tone, - target_customer=job.target_customer or "通用人群", - user_copy=job.user_copy_text or "（未提供，自由创作）", - duration=dur, - ratio=ratio, - n_images=len(job.images or []), - style_hint=style_hint, - approx_chars=approx_chars, - viral_structure_block=viral_structure_block, - persona_hint=persona_hint, + fusion_level = getattr(job, "fusion_level", "ai_polish") or "ai_polish" + fusion_instruction = FUSION_INSTRUCTIONS.get(fusion_level, FUSION_INSTRUCTIONS["ai_polish"]) + + # 使用 storyboard 模板，注入融合指令/硬约束/反套路词 + template = get_template("storyboard") + system_tpl = template.system_prompt + system_tpl = system_tpl.replace("{fusion_instruction}", fusion_instruction) + system_tpl = system_tpl.replace("{global_constraints}", GLOBAL_CONSTRAINTS) + system_tpl = system_tpl.replace("{negative_rules}", NEGATIVE_RULES) + + fusion_brief = ( + f"意图：{intent_str}\n关键信息：{key_msgs}\n调性：{tone}\n" + f"用户原文：{job.user_copy_text or '（未提供）'}\n创作模式：{fusion_level}" ) + user = render_user_prompt( + template, + duration=dur, image_count=len(job.images or []), + fusion_result=fusion_brief, + image_analysis=products_summary, + ) + + def _try_gen(model: str, temp: float, max_tok: int, label: str, tmo: int = 25): + logger.info("[爆款视频] 编导脚本生成 model=%s label=%s timeout=%d", model, label, tmo) + raw = call_llm([{"role": "system", "content": system_tpl}, {"role": "user", "content": user}], + temperature=temp, max_tokens=max_tok, model=model, timeout=tmo) + if not raw: + return None + normalized = _script_from_xml(raw, job) + if normalized is None: + # 兼容：万一 LLM 仍输出 JSON，走旧规范化 + parsed_json = _safe_json_loads(raw) + if isinstance(parsed_json, dict): + normalized = _validate_and_normalize_script(parsed_json, job) + else: + return None + voiceover = (normalized or {}).get("voiceover_script") or "" + shots_cnt = len((normalized or {}).get("shots") or []) + _before_dump = json.dumps(normalized, ensure_ascii=False) + if "很近" in _before_dump: + normalized = _replace_henjin_everywhere(normalized) + voiceover = (normalized or {}).get("voiceover_script") or "" + fallback_marker = "我最近在用的好物" in voiceover + has_typo_henjin = "很近" in json.dumps(normalized, ensure_ascii=False) + is_fallback = fallback_marker or shots_cnt < 1 or len(voiceover) < 20 or has_typo_henjin + logger.info("[爆款视频] 编导脚本结果 label=%s voiceover_len=%d shots=%d fallback=%s", label, len(voiceover), shots_cnt, is_fallback) + return None if is_fallback else normalized _s = get_shared_settings() _fast = _s.doubao_fast_model _pro = getattr(_s, "doubao_model", None) or _fast - - def _try_gen(model: str, temp: float, max_tok: int, label: str, tmo: int = 25): - logger.info("[爆款视频] 编导脚本生成 model=%s label=%s timeout=%d", model, label, tmo) - r = call_llm(prompt, temperature=temp, max_tokens=max_tok, model=model, timeout=tmo) - if r is None: - logger.warning("[爆款视频] 编导脚本返回None label=%s", label) - return None - parsed = _safe_json_loads(r) - normalized = _validate_and_normalize_script(parsed, job) - voiceover = (normalized or {}).get("voiceover_script") or "" - voiceover_len = len(voiceover) - shots_cnt = len((normalized or {}).get("shots") or []) - # 判定是否"退化到兜底质量"：口播过短（<20字）或镜头数<1；正常的短口播（如15s视频~40字）不视为兜底 - # v1.6.1 P1修复：递归替换 copy_result 里所有字符串字段的"很近"→"最近"（覆盖 overview/scene_and_lighting/voiceover/shots.* 全部字段） - _before_dump = json.dumps(normalized, ensure_ascii=False) - if "很近" in _before_dump: - logger.warning("[爆款视频] 编导脚本含错别字'很近'，递归替换为'最近' label=%s", label) - normalized = _replace_henjin_everywhere(normalized) - voiceover = (normalized or {}).get("voiceover_script") or "" - fallback_marker = "我最近在用的好物" in voiceover # _fallback_script 的特征串 - has_typo_henjin = "很近" in json.dumps(normalized, ensure_ascii=False) # 递归检查仍有"很近"视为不合格 - is_fallback = fallback_marker or shots_cnt < 1 or voiceover_len < 20 or has_typo_henjin - logger.info( - "[爆款视频] 编导脚本结果 label=%s voiceover_len=%d shots=%d fallback=%s raw_type=%s", - label, - voiceover_len, - shots_cnt, - is_fallback, - type(r).__name__, - ) - if is_fallback: - return None # 触发重试 - return normalized - try: # 第一次：快模型 25s normalized = _try_gen(_fast, 0.8, 2500, "fast-first", tmo=45) @@ -1040,7 +977,6 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di normalized = _try_gen(_fast, 0.6, 3200, "fast-retry", tmo=45) if normalized is not None: return normalized - # 第三次：用主力模型兜底，给 40s if _pro and _pro != _fast: normalized = _try_gen(_pro, 0.7, 3500, "pro-fallback", tmo=60) if normalized is not None: @@ -1051,41 +987,81 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di logger.warning("[爆款视频] 编导脚本生成异常: %s，使用兜底脚本", e, exc_info=True) return _fallback_script(job) - def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: - """步骤 4: 合规审核（简化版：基于脚本的 voiceover_script+shots 文本）。""" - dimensions = ["广告法合规", "平台规范", "内容真实性", "版权安全", "价值观", "风格一致性"] + """步骤 4: 合规审核（#2040：使用 Reviewer + review 模板，6 维度 + 自动重写 1 次）。 + + 返回结构与旧版兼容：{passed, score, details, issues, rewritten_copy?} + """ try: - from packages.shared.ai_service import call_llm - except ImportError: - return {"passed": True, "score": 90, "details": {d: "通过" for d in dimensions}} + from packages.application.viral_video.reviewer import Reviewer + from packages.application.viral_video.schemas import ( + FusionResult, IntentResult, CoreMessage, PersonalBrand, ScriptSegment, + ) + except ImportError as e: + logger.warning("[爆款视频] reviewer 模块不可用，跳过审核: %s", e) + return {"passed": True, "score": 80, "details": {}, "issues": []} - voiceover = (copy_result or {}).get("voiceover_script", "") - shots_preview = json.dumps((copy_result or {}).get("shots", [])[:3], ensure_ascii=False) - prompt = f"""请对以下短视频编导脚本进行合规审核，检查6个维度：{", ".join(dimensions)} + voiceover = (copy_result or {}).get("voiceover_script", "") or "" + title = (copy_result or {}).get("title") or (copy_result or {}).get("overview", {}).get("theme", "") -口播文案：{voiceover} -前3个镜头：{shots_preview} -行业：{job.industry} - -请以JSON格式返回： -- passed: bool（是否全部通过） -- score: int（0-100分） -- details: 各维度评分和说明 -- issues: 需要修改的问题列表（如有）""" - _s = get_shared_settings() - _fast = _s.doubao_fast_model - _pro = _s.doubao_model - for _m, _lbl in [(_fast, "fast"), (_pro, "pro-fallback")]: - try: - logger.info("[爆款视频] 合规审核 model=%s label=%s", _m, _lbl) - result = call_llm(prompt, temperature=0.1, max_tokens=500, model=_m, timeout=30) - return result if isinstance(result, dict) else {"passed": True, "score": 80, "details": {}} - except Exception as e: - logger.warning("[爆款视频] 合规审核失败 label=%s err=%s", _lbl, e) - continue - return {"passed": True, "score": 75, "details": {d: "默认通过" for d in dimensions}} + intent_data = job.intent_result or {} + core_msgs = [CoreMessage(text=str(m), must_keep=True, confidence=0.9) for m in (intent_data.get("key_messages") or [])] + brands: list[PersonalBrand] = [] + brand_text = intent_data.get("brand_text") or intent_data.get("suggested_title") or "" + if brand_text: + brands.append(PersonalBrand(text=str(brand_text), category="brand")) + intent_obj = IntentResult( + intent_summary=intent_data.get("intent", "") or "推广产品", + core_messages=core_msgs, personal_brands=brands, + ) + fusion_obj = FusionResult( + title=title or "", + hook=(voiceover[:30] if voiceover else ""), + cta="", + script_segments=[], + word_count=len(voiceover), + estimated_duration=int(getattr(job, "duration", 15) or 15), + ) + for shot in (copy_result or {}).get("shots", []) or []: + if isinstance(shot, dict) and shot.get("scene_and_dialogue"): + fusion_obj.script_segments.append(ScriptSegment(text=shot["scene_and_dialogue"])) + fusion_level = getattr(job, "fusion_level", "ai_polish") or "ai_polish" + try: + reviewer = Reviewer() + review_res = reviewer.review(fusion_obj, intent_obj, fusion_level) + new_copy = copy_result + rewritten_voice = None + if not review_res.passed and review_res.rewrite_suggestions: + try: + rewritten = reviewer.rewrite(fusion_obj, review_res, intent_obj, fusion_level) + if rewritten and (rewritten.title or rewritten.script_segments): + new_voice = rewritten.script_segments[0].text if rewritten.script_segments else (rewritten.hook or voiceover) + new_copy = dict(copy_result) + new_copy["voiceover_script"] = new_voice + new_copy["final_copy"] = new_voice + new_copy["suggested_copy"] = new_voice + if rewritten.title: + new_copy.setdefault("overview", {})["theme"] = rewritten.title + new_copy["title"] = rewritten.title + rewritten_voice = new_voice + review_res = reviewer.review(rewritten, intent_obj, fusion_level) + except Exception as e: + logger.warning("[爆款视频] 自动重写失败: %s", e) + result = { + "passed": review_res.passed, + "score": 90 if review_res.passed else 60, + "details": {i.dimension: i.text for i in review_res.issues}, + "issues": [{"dimension": i.dimension, "severity": i.severity, "location": i.location, "text": i.text} for i in review_res.issues], + } + if rewritten_voice is not None: + result["rewritten_copy"] = new_copy + job.copy_result = new_copy + job.generated_copy_text = rewritten_voice + return result + except Exception as e: + logger.warning("[爆款视频] 审核异常，跳过: %s", e, exc_info=True) + return {"passed": True, "score": 75, "details": {}, "issues": []} def _step_tts(job: ViralVideoJob, voiceover_script: str): """步骤 5: CosyVoice 整段配音 → 返回本地 MP3 Path；失败返回 None。""" @@ -1125,7 +1101,6 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str): logger.warning("[爆款视频] TTS 配音失败: %s", e, exc_info=True) return None - def _upload_tts_to_oss(job: ViralVideoJob, tts_path) -> str | None: """把 TTS 本地 mp3 上传到 OSS，返回公网 URL（供 Seedance 做 reference_audios 口型驱动用）。""" if tts_path is None: @@ -1145,7 +1120,6 @@ def _upload_tts_to_oss(job: ViralVideoJob, tts_path) -> str | None: logger.warning("[爆款视频] TTS 上传 OSS 失败: %s", e, exc_info=True) return None - def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str: """把编导脚本拼成 Seedance 长 prompt。""" if not isinstance(copy_result, dict) or not copy_result: @@ -1196,7 +1170,6 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str: lines.append("，".join([str(x) for x in np if x])) return "\n".join(lines) - def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | None) -> tuple[str, dict | None]: """步骤 6: v1.6 单次 Seedance 生成（不再分段/拼接）。 @@ -1310,7 +1283,6 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non logger.info("[爆款视频] 单次生成完成: path=%s size=%d usage=%s", video_path, size, usage) return str(video_path), (usage if isinstance(usage, dict) else None) - def _step_upload(job: ViralVideoJob, video_path: str) -> str: """步骤 7: OSS 上传。""" from video_processing.oss_helpers import upload_to_oss @@ -1323,7 +1295,6 @@ def _step_upload(job: ViralVideoJob, video_path: str) -> str: raise RuntimeError(f"OSS 上传失败: storage_key={storage_key}") return video_url - def _wait_oss_ready(url: str, timeout_sec: int = 10) -> bool: """轮询 OSS 公网 URL，直到 HEAD 返回 200 或超时。 用于缓解 OSS 上传后 1-5s 公网 eventual consistency 导致的 NoSuchKey。 @@ -1344,10 +1315,8 @@ def _wait_oss_ready(url: str, timeout_sec: int = 10) -> bool: logger.warning("[爆款视频] OSS 成片在 %ds 内未就绪 last_status=%s url=%s", timeout_sec, last_status, url[:120]) return False - # ── 主编排器 ──────────────────────────────────────────────────────────── - @shared_task( bind=True, max_retries=2, @@ -1432,7 +1401,6 @@ def run_viral_video_pipeline(self: Task, job_id: str) -> dict: if session: session.close() - @shared_task(bind=True, max_retries=2, name="worker.resume_viral_video_pipeline") def resume_viral_video_pipeline(self: Task, job_id: str) -> dict: """旧 confirm-intent 路径兼容：从 WAIT_USER_CONFIRM 跑完整个渲染。""" @@ -1454,7 +1422,6 @@ def resume_viral_video_pipeline(self: Task, job_id: str) -> dict: if session: session.close() - @shared_task(bind=True, max_retries=1, name="worker.run_video_style_analysis") def run_video_style_analysis(self: Task, job_id: str) -> dict: """独立的视频风格分析任务。""" @@ -1478,10 +1445,8 @@ def run_video_style_analysis(self: Task, job_id: str) -> dict: if session: session.close() - # ── 失败处理 ──────────────────────────────────────────────────────────── - def _mark_failed_and_notify(job_id: str, session, repo, job, err_msg: str, stage: str = "") -> None: """标记任务失败并通知。若传入的 session 已失效（因前面异常导致 rollback 状态）， 会自动 fallback 到新建 SessionLocal 重新标记，确保状态一定落库。""" @@ -1522,10 +1487,8 @@ def _mark_failed_and_notify(job_id: str, session, repo, job, err_msg: str, stage event_type="viral_video:failed", ) - # ── v1.5/v1.6 三步分步流水线 Celery 任务 ───────────────────────────────── - @shared_task( bind=True, max_retries=1, @@ -1593,7 +1556,6 @@ def run_viral_video_analyze(self: Task, job_id: str) -> dict: if session: session.close() - @shared_task( bind=True, max_retries=1, @@ -1696,7 +1658,6 @@ def run_viral_video_generate_copy(self: Task, job_id: str) -> dict: if session: session.close() - def _quick_compliance_blacklist_check(copy_result: dict) -> None: """阶段2快速黑名单检查：不调用 LLM，只扫描高风险关键词；命中则在 voiceover 中就地替换。 @@ -1738,7 +1699,6 @@ def _quick_compliance_blacklist_check(copy_result: dict) -> None: for bk, bv in BLACKLIST.items(): copy_result[k] = copy_result[k].replace(bk, bv) - def _try_refund_viral_video(job: ViralVideoJob) -> None: """爆款视频生成失败：若已预扣积分则全额退款。""" try: @@ -1768,7 +1728,6 @@ def _try_refund_viral_video(job: ViralVideoJob) -> None: except Exception: logger.exception("[爆款视频] 失败退款异常 job_id=%s", job.id) - def _settle_viral_video(job: ViralVideoJob, usage: dict | None) -> None: """爆款视频生成成功：按实际 usage 结算，多退少补，写 credits_cost。""" try: @@ -1839,7 +1798,6 @@ def _settle_viral_video(job: ViralVideoJob, usage: dict | None) -> None: job.credits_cost = float(getattr(job, "credits_prepaid", 0) or 0) job.credits_prepaid = 0.0 - def _run_render_pipeline(job_id: str, session, repo, job) -> dict: """v1.6.1 阶段3：出片前合规审核（LLM 深度）→ TTS → Seedance → Upload → Completed。 @@ -1863,9 +1821,14 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict: review_result = _step_review(job, copy_result) if not review_result.get("passed", True): _emit_progress(job_id, ViralVideoStage.REVIEW, 67.0, "审核未通过，正在自动重写...") - intent = job.intent_result or _step_intent_parsing(job, image_analysis) - copy_result = _step_script_generation(job, intent, image_analysis) - _step_review(job, copy_result) # 二次审核，不通过也继续出片（避免反复循环） + # #2040: Reviewer 已在 _step_review 内完成 1 次自动重写 + rewritten = review_result.get("rewritten_copy") + if isinstance(rewritten, dict) and rewritten: + copy_result = rewritten + else: + intent = job.intent_result or _step_intent_parsing(job, image_analysis) + copy_result = _step_script_generation(job, intent, image_analysis) + _step_review(job, copy_result) job.copy_result = copy_result job.generated_copy_text = copy_result.get("voiceover_script", "") or "" _save_job(repo, job, session) @@ -1921,7 +1884,6 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict: logger.info("[爆款视频] 任务完成: job_id=%s video_url=%s", job_id, video_url) return {"ok": True, "job_id": job_id, "video_url": video_url} - @shared_task(bind=True, max_retries=2, name="worker.run_viral_video_render") def run_viral_video_render(self: Task, job_id: str) -> dict: """v1.6 阶段3：TTS + 单次 Seedance 生成 + 上传。""" diff --git a/packages/application/viral_video/prompts.py b/packages/application/viral_video/prompts.py index 2eab68e7e..520743788 100644 --- a/packages/application/viral_video/prompts.py +++ b/packages/application/viral_video/prompts.py @@ -132,7 +132,10 @@ _FUSION_SYSTEM = """你负责为短视频生成营销文案。请按思维链分 开头3秒钩子，5到15字。 每个要点用一个 标签，属性 elaboration 是展开说明、image_index 是对应第几张图（从0开始），标签内容写要点。 口语化的行动号召。 - 每段配音用一个 标签，属性 duration_sec 是秒数、image_index 是对应图片，标签内容写配音文案。 + 每段配音用一个 标签，属性 duration_sec 是秒数、image_index 是对应图片，标签内容写配音文案（纯口播文本，不加旁白标注、不加镜头标注、不加"主播："之类前缀）。 + 把所有 segment 的配音文案按顺序自然拼接成一段完整的纯口播文本（无标记、无括号、无前缀），长度要适配 {duration} 秒，约 {approx_chars} 字。 + 视频主题（一句话概括）。 + 整体场景描述+光线设定（100-200字，要具体：在哪拍、什么光线、什么色调、什么氛围）。 配音总字数，只写数字。 预计时长秒数，只写数字。 @@ -161,25 +164,34 @@ _FUSION_EXAMPLE = """厨房重油污，别再用洗洁精硬擦了后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净 39块钱625ml，厨房重油污的可以试一瓶 +这油污我真的忍很久了，用洗洁精擦半天都没用。后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净。39块钱625ml，厨房重油污的可以试一瓶。 +厨房油污清洁好物分享 +简洁明亮的厨房台面场景，自然光从窗户洒入，色调温暖柔和，突出产品白色瓶身与去油污对比效果。 58 13""" # ── 模板4：编导级分镜（LLM）──────────────────────────────────────────── -_STORYBOARD_SYSTEM = f"""你是短视频编导，负责把文案拆成可拍摄的分镜。 +_STORYBOARD_SYSTEM = f"""你是短视频编导，负责把文案拆成可拍摄的分镜，为 Seedance 2.5 视频模型写编导分镜脚本。脚本将整体作为 prompt 一次性传给视频模型，必须让模型在连贯镜头流中清楚每段时间拍什么、画面如何、人物说什么。 工作方式： 1. 按文案的 script_segments 顺序分配镜头。 -2. 每个镜头确定画面、运镜、时长、配音和字幕。 +2. 每个镜头确定景别/角度/运镜、画面场景与对白、人物动作细节、音效/BGM、转场。 3. 检查所有镜头时长加起来接近目标时长，误差不超过2秒。 4. image_index 必须在已上传图片范围内，第一张主图必须用在第一个镜头。 {GLOBAL_CONSTRAINTS} 请严格按下面的标签格式输出，不要解释，不要用代码块： - 下面每个镜头用一个 标签，属性 image_index 是图片序号（从0开始）、transition 取 fade、cut、zoom_in、slide_left、dissolve、wipe 之一、zoom 取 in、out 或 null、duration_sec 是该镜头秒数、bgm_note 是该段BGM情绪。每个 里面包含： - 该镜头配音文本； - 字幕文本，可与配音一致或更精简； - 用一个空标签，属性 start、end 写“x,y”坐标、ease 写缓动方式；不需要运镜时坐标相同。""" + 下面每个镜头用一个 标签，属性 image_index 是图片序号（从0开始）、transition 取 fade/cut/zoom_in/slide_left/dissolve/wipe 之一、zoom 取 in/out/null、duration_sec 是该镜头秒数、bgm_note 是该段BGM情绪。每个 里面包含： + 该镜头配音文本（纯口播文本，不加旁白标注）； + 字幕文本，可与配音一致或更精简； + 景别+角度+运镜（例：近景俯拍45度，缓慢推镜；中景平视，固定镜头；特写平视，快速拉镜）； + 画面场景描述 + 人物口播台词（对白要自然口语化，像朋友聊天，不要硬广推销腔）； + 人物动作、表情、物品操作细节（手怎么动、表情变化、产品怎么展示）； + 环境音+BGM提示（例：轻快流行BGM，环境嘈杂咖啡店背景音）； + 硬切/淡入淡出/叠化（最后一镜写『结束』即可）； + 参考图片索引（0-based，对应第几张产品图，无则空）； + 用一个空标签，属性 start、end 写"x,y"坐标、ease 写缓动方式；不需要运镜时坐标相同。""" _STORYBOARD_USER = """目标时长：{duration}秒 上传图片数量：{image_count}张（第1张是主图/封面） @@ -194,17 +206,35 @@ _STORYBOARD_EXAMPLE = """ 这油污我真的忍很久了 这油污忍很久了 +近景俯拍45度，缓慢推镜 +厨房台面，主妇皱眉看着灶台油污。对白：这油污我真的忍很久了 +右手拿着脏抹布，无奈摇头 +轻快日常BGM，带一点烦躁感 +硬切 +0 后来换了大公鸡头油污净，喷上等几分钟，一擦就干净 喷上等几分钟，一擦就干净 +特写平视，固定镜头 +手部特写，喷油污净在油污处。对白：后来换了这个大公鸡头油污净，喷上等几分钟，一擦就干净 +左手拿产品瓶身，右手按压喷头，等待片刻后用抹布轻擦 +轻快转折BGM，带清爽感 +淡入淡出 +0 39块钱625ml，厨房重油污的可以试一瓶 39元625ml，可以试一瓶 - +中景平视，缓慢拉镜 +产品正面展示，明亮背景。对白：39块钱625ml，厨房重油污的可以试一瓶 +产品置于画面中央，轻微转动展示瓶身 +温暖收尾BGM +结束 +0 + """ From 252ea71d5033801eac069a65021e52b02d721ee1 Mon Sep 17 00:00:00 2001 From: Xiaoxia Agent Date: Sun, 4 Oct 2026 17:37:49 +0800 Subject: [PATCH 2/5] =?UTF-8?q?test:=20=E6=96=B0=E5=A2=9E=20prompt=20?= =?UTF-8?q?=E6=A8=A1=E6=9D=BF=E6=8E=A5=E7=BA=BF=E9=9B=86=E6=88=90=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=20+=20=E4=BF=AE=E5=A4=8D=E6=97=A7=E6=B5=8B=E8=AF=95?= =?UTF-8?q?=E9=80=82=E9=85=8D=20XML=20=E8=BE=93=E5=87=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 test_viral_video_wiring.py：9 个集成测试覆盖四步接线全流程 - image_analysis / intent_parsing 走 loader + XML 解析 - storyboard 三档 fusion_level 注入 system_prompt - review 走 Reviewer 且 pass/rewrite 两条路径正确 - 端到端验证每步都调用 prompt_loader - 修复 test_script_generation_returns_copy_result：mock 改为返回 XML 字符串 - 修复 test_review_pass_v16：mock Reviewer.review 返回真实 ReviewResult - viral_video.py 增加 import re，_xml_to_product 的 text_on_package 同时从子标签读取（兜底） - prompts.py _STORYBOARD_SYSTEM 改为普通字符串 + {fusion_instruction}/{global_constraints}/{negative_rules} 占位符，支持运行时注入 全量 viral_video 单测 150/150 通过，无回归。 Closes #2040 --- apps/worker/worker_app/tasks/viral_video.py | 4 + packages/application/viral_video/prompts.py | 8 +- tests/unit/test_viral_video.py | 58 ++-- tests/unit/test_viral_video_wiring.py | 288 ++++++++++++++++++++ 4 files changed, 326 insertions(+), 32 deletions(-) create mode 100644 tests/unit/test_viral_video_wiring.py diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index 95b798695..13b7e91ba 100644 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -22,6 +22,7 @@ from __future__ import annotations import json import logging import os +import re import tempfile import threading import time @@ -382,6 +383,9 @@ def _analyze_single_image( for p in product_nodes: a = p["attrs"] text_on_pkg = a.get("text_on_package", "") + p_body = p.get("text", "") or "" + if not text_on_pkg and p_body: + text_on_pkg = xp.text_of(p_body, "text_on_package") or "" text_list = [x.strip() for x in re.split(r"[,，;；]", text_on_pkg) if x.strip()] if text_on_pkg else [] features = a.get("features", "") feat_list = [x.strip() for x in re.split(r"[,，;；]", features) if x.strip()] if features else [] diff --git a/packages/application/viral_video/prompts.py b/packages/application/viral_video/prompts.py index 520743788..51a19151d 100644 --- a/packages/application/viral_video/prompts.py +++ b/packages/application/viral_video/prompts.py @@ -171,7 +171,7 @@ _FUSION_EXAMPLE = """厨房重油污，别再用洗洁精硬擦了13""" # ── 模板4：编导级分镜（LLM）──────────────────────────────────────────── -_STORYBOARD_SYSTEM = f"""你是短视频编导，负责把文案拆成可拍摄的分镜，为 Seedance 2.5 视频模型写编导分镜脚本。脚本将整体作为 prompt 一次性传给视频模型，必须让模型在连贯镜头流中清楚每段时间拍什么、画面如何、人物说什么。 +_STORYBOARD_SYSTEM = """你是短视频编导，负责把文案拆成可拍摄的分镜，为 Seedance 2.5 视频模型写编导分镜脚本。脚本将整体作为 prompt 一次性传给视频模型，必须让模型在连贯镜头流中清楚每段时间拍什么、画面如何、人物说什么。 工作方式： 1. 按文案的 script_segments 顺序分配镜头。 @@ -179,7 +179,11 @@ _STORYBOARD_SYSTEM = f"""你是短视频编导，负责把文案拆成可拍摄 3. 检查所有镜头时长加起来接近目标时长，误差不超过2秒。 4. image_index 必须在已上传图片范围内，第一张主图必须用在第一个镜头。 -{GLOBAL_CONSTRAINTS} +{fusion_instruction} + +{global_constraints} + +{negative_rules} 请严格按下面的标签格式输出，不要解释，不要用代码块： 下面每个镜头用一个 标签，属性 image_index 是图片序号（从0开始）、transition 取 fade/cut/zoom_in/slide_left/dissolve/wipe 之一、zoom 取 in/out/null、duration_sec 是该镜头秒数、bgm_note 是该段BGM情绪。每个 里面包含： diff --git a/tests/unit/test_viral_video.py b/tests/unit/test_viral_video.py index a7d363cf7..5e915d568 100755 --- a/tests/unit/test_viral_video.py +++ b/tests/unit/test_viral_video.py @@ -403,33 +403,30 @@ class TestViralVideoPipeline: """v1.6: _step_script_generation 返回 dict 形式的 CopyResult，含 voiceover_script + shots。""" from apps.worker.worker_app.tasks.viral_video import _step_script_generation - mock_llm.return_value = { - "overview": {"theme": "口红推荐", "total_duration": 15, "aspect_ratio": "9:16"}, - "scene_and_lighting": "明亮化妆台，柔和自然光", - "shots": [ - { - "time_range": "0-5秒", - "shot_type_angle_movement": "近景平视，缓慢推镜", - "scene_and_dialogue": "女主微笑展示口红：大家好，今天分享一款口红", - "action_details": "手持口红特写", - "audio_bgm": "轻快流行BGM", - "transition": "硬切", - "reference_image_index": 0, - }, - { - "time_range": "5-15秒", - "shot_type_angle_movement": "特写，固定镜头", - "scene_and_dialogue": "涂抹口红：颜色特别好看很显白", - "action_details": "嘴唇涂抹特写", - "audio_bgm": "轻快BGM继续", - "transition": "结束", - "reference_image_index": 1, - }, - ], - "hard_constraints": ["无字幕无水印"], - "negative_prompts": ["字幕", "水印"], - "voiceover_script": "大家好，今天分享一款口红，颜色特别好看很显白。", - } + mock_llm.return_value = """ + +大家好，今天分享一款口红 +大家好，今天分享一款口红 +近景平视，缓慢推镜 +女主微笑展示口红：大家好，今天分享一款口红 +手持口红特写 +轻快流行BGM +硬切 +0 + + + +颜色特别好看很显白 +颜色特别好看很显白 +特写，固定镜头 +涂抹口红：颜色特别好看很显白 +嘴唇涂抹特写 +轻快BGM继续 +结束 +1 + + +""" result = _step_script_generation( mock_job, {"intent": "推广口红", "key_messages": [], "tone": "亲切"}, {"products": []} ) @@ -452,12 +449,13 @@ class TestViralVideoPipeline: assert result["voiceover_script"] assert len(result["shots"]) >= 1 - @patch("packages.shared.ai_service.call_llm") - def test_review_pass_v16(self, mock_llm, mock_job): + @patch("packages.application.viral_video.reviewer.Reviewer.review") + def test_review_pass_v16(self, mock_review, mock_job): """v1.6 _step_review 接收 copy_result dict。""" from apps.worker.worker_app.tasks.viral_video import _step_review + from packages.application.viral_video.reviewer import ReviewResult, ReviewIssue - mock_llm.return_value = {"passed": True, "score": 90, "details": {}} + mock_review.return_value = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[]) cr = {"voiceover_script": "大家好", "shots": []} result = _step_review(mock_job, cr) assert result["passed"] is True diff --git a/tests/unit/test_viral_video_wiring.py b/tests/unit/test_viral_video_wiring.py new file mode 100644 index 000000000..e72a8822a --- /dev/null +++ b/tests/unit/test_viral_video_wiring.py @@ -0,0 +1,288 @@ +"""#2040 接线集成测试：验证运行中的 viral_video 任务使用 prompt_loader 从 DB 读取模板。 + +mock LLM/Vision 调用，验证： +1. image_analysis 走 loader 模板 + XML 解析 +2. intent_parsing 走 loader 模板 + XML 解析 +3. script_generation 走 storyboard 模板 + XML 解析，输出兼容 Seedance 的 copy_result +4. review 走 Reviewer（review 模板）带自动重写 +5. 三档融合（ai_full / ai_polish / user_primary）注入不同 FUSION_INSTRUCTIONS +""" + +from __future__ import annotations + +import sys +from pathlib import Path as _Path + +_WORKER_ROOT = _Path(__file__).resolve().parents[2] / "apps" / "worker" +if str(_WORKER_ROOT) not in sys.path: + sys.path.insert(0, str(_WORKER_ROOT)) + +from unittest.mock import MagicMock, patch + +import pytest + +from packages.domain.viral_video import ViralVideoJob + + +@pytest.fixture +def job(): + j = ViralVideoJob( + user_id="u1", + images=["https://img/1.jpg", "https://img/2.jpg"], + industry="美妆", + duration=15, + user_copy_text="这款口红真的太绝了，显白又持久，姐妹们冲！", + fusion_level="ai_polish", + ) + return j + + +# ── Mock LLM/Vision 返回的 XML 文本 ───────────────────────────────── + +IMAGE_XML = """ + +室内桌面拍摄，柔和自然光 +清新温暖 + + 品牌X,211 + + +""".strip() + +INTENT_XML = """ + +推广显白持久口红 + + 显白 + 持久 + + + + +亲切自然 +显白持久口红推荐 + +""".strip() + +STORYBOARD_XML = """ + + + 这款口红真的太绝了 + 显白又持久 + 近景俯拍45度，缓慢推镜 + 厨房台面，主妇展示口红。对白：这款口红真的太绝了 + 右手持口红展示膏体 + 轻快BGM + 硬切 + 0 + + + +""".strip() + + +@pytest.fixture(autouse=True) +def invalidate_loader_cache(): + from packages.application.viral_video import prompt_loader as pl + pl.invalidate() + yield + pl.invalidate() + + +# ── 1) 图片分析走模板 ─────────────────────────────────────────────── + + +class TestImageAnalysisWiring: + def test_uses_loader_template_and_xml_parse(self, job): + from apps.worker.worker_app.tasks import viral_video as vv + + with patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML) as mock_v: + result = vv._analyze_single_image(0, "https://img/1.jpg", "vlm-lite", 15) + + mock_v.assert_called_once() + # 验证调用时传入了 system_prompt（说明走了 loader 渲染的模板） + call_kwargs = mock_v.call_args.kwargs + assert "system_prompt" in call_kwargs and call_kwargs["system_prompt"] + # 结果包含从 XML 解析出的产品信息 + assert result["name"] == "lipstick" + assert result["brand"] == "品牌X" + assert "显白" in result["key_features"] + assert result["text_on_package"] == ["品牌X", "211"] + + +# ── 2) 意图解析走模板 ─────────────────────────────────────────────── + + +class TestIntentParsingWiring: + def test_uses_loader_and_parses_xml(self, job): + from apps.worker.worker_app.tasks import viral_video as vv + + img_result = {"products": [{"name": "lipstick", "brand": "品牌X", + "key_features": ["显白", "持久"]}]} + with patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML) as mock_llm: + result = vv._step_intent_parsing(job, img_result) + + mock_llm.assert_called_once() + assert result["intent"] == "推广显白持久口红" + assert "显白" in result["key_messages"] + assert result["suggested_title"] == "显白持久口红推荐" + + +# ── 3) 脚本生成：storyboard 模板 + XML 解析 + fusion_level 注入 ──── + + +class TestScriptGenerationWiring: + @pytest.mark.parametrize("level", ["ai_full", "ai_polish", "user_primary"]) + def test_fusion_level_injected(self, job, level): + """三档融合水平被注入到 storyboard 模板的 system_prompt""" + from apps.worker.worker_app.tasks import viral_video as vv + from packages.application.viral_video.prompts import FUSION_INSTRUCTIONS + + job.fusion_level = level + intent = {"intent": "推广", "key_messages": ["显白"], "tone": "亲切"} + + captured_system = {} + + def fake_call_llm(messages, **kw): + captured_system["final"] = messages[0]["content"] + return STORYBOARD_XML + + with patch("packages.shared.ai_service.call_llm", side_effect=fake_call_llm): + result = vv._step_script_generation(job, intent, {}) + + # fusion_level 对应的指令文本被注入到 system prompt 中 + assert FUSION_INSTRUCTIONS[level] in captured_system["final"], \ + f"fusion_level {level} 指令未注入 system_prompt" + # 输出保持 Seedance 兼容结构 + assert "overview" in result + assert "shots" in result + assert len(result["shots"]) >= 1 + assert result["shots"][0]["shot_type_angle_movement"] + assert result["voiceover_script"] + + def test_fallback_when_xml_and_json_unparseable(self, job): + """XML 解析失败且无法解析为 JSON 时，回退到兜底脚本""" + from apps.worker.worker_app.tasks import viral_video as vv + + job.fusion_level = "ai_polish" + intent = {"intent": "推广", "key_messages": [], "tone": "亲切"} + with patch("packages.shared.ai_service.call_llm", return_value="not xml not json"): + result = vv._step_script_generation(job, intent, {}) + assert isinstance(result, dict) + assert "voiceover_script" in result + assert "shots" in result + + +# ── 4) Review 使用 Reviewer + 自动重写 ───────────────────────────── + + +class TestReviewWiring: + def test_pass_path(self, job): + from apps.worker.worker_app.tasks import viral_video as vv + from packages.application.viral_video.reviewer import Reviewer, ReviewResult + + copy_result = { + "title": "口红推荐", + "overview": {"theme": "口红推荐"}, + "voiceover_script": "这款口红显白又持久", + "shots": [{"scene_and_dialogue": "展示口红"}], + } + job.intent_result = {"key_messages": ["显白", "持久"], "intent": "推广"} + + pass_result = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[]) + with patch.object(Reviewer, "review", return_value=pass_result): + out = vv._step_review(job, copy_result) + assert out["passed"] is True + + def test_rewrite_path(self, job): + """审核不通过时触发自动重写，并更新 job.copy_result""" + from apps.worker.worker_app.tasks import viral_video as vv + from packages.application.viral_video.reviewer import Reviewer, ReviewResult + from packages.application.viral_video.schemas import FusionResult, ScriptSegment, ReviewIssue + + copy_result = { + "title": "原标题", + "overview": {"theme": "原标题"}, + "voiceover_script": "这款口红绝了", + "shots": [{"scene_and_dialogue": "展示"}], + } + job.intent_result = {"key_messages": ["显白"], "intent": "推广"} + + fail_result = ReviewResult( + passed=False, score=50, + issues=[ReviewIssue(dimension="违规词", severity="high", location="开头", text="绝了")], + rewrite_suggestions=["去掉夸大词"], + ) + rewritten = FusionResult( + title="新标题", hook="修改后钩子", + script_segments=[ScriptSegment(text="修改后口播正文")], cta="行动号召", + word_count=10, estimated_duration=10, + ) + pass_after = ReviewResult(passed=True, score=88, issues=[], rewrite_suggestions=[]) + + with patch.object(Reviewer, "review", side_effect=[fail_result, pass_after]), \ + patch.object(Reviewer, "rewrite", return_value=rewritten): + out = vv._step_review(job, copy_result) + + assert out["passed"] is True + assert "rewritten_copy" in out + assert job.generated_copy_text == "修改后口播正文" + + +# ── 5) 端到端：每个 step 调用 loader 对应 prompt_type ────────────── + + +class TestEndToEndLoaderUsed: + def test_each_step_calls_loader(self, job): + from apps.worker.worker_app.tasks import viral_video as vv + from packages.application.viral_video import prompt_loader as pl + + called_types = [] + real_get = pl.get_template + + def spy_get(prompt_type, **kwargs): + called_types.append(prompt_type) + return real_get(prompt_type, **kwargs) + + with patch.object(pl, "get_template", side_effect=spy_get), \ + patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML), \ + patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML): + # 1) image + img_res = vv._analyze_single_image(0, "https://img/1.jpg", "vlm", 15) + # 2) intent + intent_res = vv._step_intent_parsing(job, {"products": [img_res]}) + + # 前两步分别调用了 image_analysis 和 intent_parsing + assert "image_analysis" in called_types + assert "intent_parsing" in called_types + + # script 和 review 单独验证（需要不同的 LLM 返回） + called_types_2 = [] + def spy_get_2(prompt_type, **kwargs): + called_types_2.append(prompt_type) + return real_get(prompt_type, **kwargs) + + with patch.object(pl, "get_template", side_effect=spy_get_2), \ + patch("packages.shared.ai_service.call_llm", return_value=STORYBOARD_XML): + copy_res = vv._step_script_generation(job, intent_res, {"products": [img_res]}) + assert "storyboard" in called_types_2 + + called_types_3 = [] + def spy_get_3(prompt_type, **kwargs): + called_types_3.append(prompt_type) + return real_get(prompt_type, **kwargs) + + from packages.application.viral_video.reviewer import Reviewer, ReviewResult + pass_result = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[]) + job.intent_result = intent_res + job.copy_result = copy_res + with patch.object(pl, "get_template", side_effect=spy_get_3), \ + patch.object(Reviewer, "review", return_value=pass_result) as mock_review: + review_res = vv._step_review(job, copy_res) + # review 步骤内部直接调用 Reviewer.review，该方法被 mock，因此 get_template 不会被调用； + # 此处验证 Reviewer.review 被调用即可说明 review 步骤走通了。 + assert mock_review.called, "_step_review 未调用 Reviewer.review" + assert isinstance(review_res, dict) and "passed" in review_res From 50489b05d68dc74150e1d1c1e0a9002dcee6aeb8 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Sun, 4 Oct 2026 09:43:40 +0000 Subject: [PATCH 3/5] style: auto-format with black + isort + ruff + prettier [skip ci-format-check] --- tests/unit/test_viral_video.py | 2 +- tests/unit/test_viral_video_wiring.py | 50 +++++++++++++++++---------- 2 files changed, 33 insertions(+), 19 deletions(-) diff --git a/tests/unit/test_viral_video.py b/tests/unit/test_viral_video.py index 5e915d568..7d5ba2845 100755 --- a/tests/unit/test_viral_video.py +++ b/tests/unit/test_viral_video.py @@ -453,7 +453,7 @@ class TestViralVideoPipeline: def test_review_pass_v16(self, mock_review, mock_job): """v1.6 _step_review 接收 copy_result dict。""" from apps.worker.worker_app.tasks.viral_video import _step_review - from packages.application.viral_video.reviewer import ReviewResult, ReviewIssue + from packages.application.viral_video.reviewer import ReviewIssue, ReviewResult mock_review.return_value = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[]) cr = {"voiceover_script": "大家好", "shots": []} diff --git a/tests/unit/test_viral_video_wiring.py b/tests/unit/test_viral_video_wiring.py index e72a8822a..fe5ada00e 100644 --- a/tests/unit/test_viral_video_wiring.py +++ b/tests/unit/test_viral_video_wiring.py @@ -87,6 +87,7 @@ STORYBOARD_XML = """ @pytest.fixture(autouse=True) def invalidate_loader_cache(): from packages.application.viral_video import prompt_loader as pl + pl.invalidate() yield pl.invalidate() @@ -120,8 +121,7 @@ class TestIntentParsingWiring: def test_uses_loader_and_parses_xml(self, job): from apps.worker.worker_app.tasks import viral_video as vv - img_result = {"products": [{"name": "lipstick", "brand": "品牌X", - "key_features": ["显白", "持久"]}]} + img_result = {"products": [{"name": "lipstick", "brand": "品牌X", "key_features": ["显白", "持久"]}]} with patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML) as mock_llm: result = vv._step_intent_parsing(job, img_result) @@ -154,8 +154,7 @@ class TestScriptGenerationWiring: result = vv._step_script_generation(job, intent, {}) # fusion_level 对应的指令文本被注入到 system prompt 中 - assert FUSION_INSTRUCTIONS[level] in captured_system["final"], \ - f"fusion_level {level} 指令未注入 system_prompt" + assert FUSION_INSTRUCTIONS[level] in captured_system["final"], f"fusion_level {level} 指令未注入 system_prompt" # 输出保持 Seedance 兼容结构 assert "overview" in result assert "shots" in result @@ -201,7 +200,7 @@ class TestReviewWiring: """审核不通过时触发自动重写，并更新 job.copy_result""" from apps.worker.worker_app.tasks import viral_video as vv from packages.application.viral_video.reviewer import Reviewer, ReviewResult - from packages.application.viral_video.schemas import FusionResult, ScriptSegment, ReviewIssue + from packages.application.viral_video.schemas import FusionResult, ReviewIssue, ScriptSegment copy_result = { "title": "原标题", @@ -212,19 +211,25 @@ class TestReviewWiring: job.intent_result = {"key_messages": ["显白"], "intent": "推广"} fail_result = ReviewResult( - passed=False, score=50, + passed=False, + score=50, issues=[ReviewIssue(dimension="违规词", severity="high", location="开头", text="绝了")], rewrite_suggestions=["去掉夸大词"], ) rewritten = FusionResult( - title="新标题", hook="修改后钩子", - script_segments=[ScriptSegment(text="修改后口播正文")], cta="行动号召", - word_count=10, estimated_duration=10, + title="新标题", + hook="修改后钩子", + script_segments=[ScriptSegment(text="修改后口播正文")], + cta="行动号召", + word_count=10, + estimated_duration=10, ) pass_after = ReviewResult(passed=True, score=88, issues=[], rewrite_suggestions=[]) - with patch.object(Reviewer, "review", side_effect=[fail_result, pass_after]), \ - patch.object(Reviewer, "rewrite", return_value=rewritten): + with ( + patch.object(Reviewer, "review", side_effect=[fail_result, pass_after]), + patch.object(Reviewer, "rewrite", return_value=rewritten), + ): out = vv._step_review(job, copy_result) assert out["passed"] is True @@ -247,9 +252,11 @@ class TestEndToEndLoaderUsed: called_types.append(prompt_type) return real_get(prompt_type, **kwargs) - with patch.object(pl, "get_template", side_effect=spy_get), \ - patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML), \ - patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML): + with ( + patch.object(pl, "get_template", side_effect=spy_get), + patch("packages.shared.ai_service.call_vision", return_value=IMAGE_XML), + patch("packages.shared.ai_service.call_llm", return_value=INTENT_XML), + ): # 1) image img_res = vv._analyze_single_image(0, "https://img/1.jpg", "vlm", 15) # 2) intent @@ -261,26 +268,33 @@ class TestEndToEndLoaderUsed: # script 和 review 单独验证（需要不同的 LLM 返回） called_types_2 = [] + def spy_get_2(prompt_type, **kwargs): called_types_2.append(prompt_type) return real_get(prompt_type, **kwargs) - with patch.object(pl, "get_template", side_effect=spy_get_2), \ - patch("packages.shared.ai_service.call_llm", return_value=STORYBOARD_XML): + with ( + patch.object(pl, "get_template", side_effect=spy_get_2), + patch("packages.shared.ai_service.call_llm", return_value=STORYBOARD_XML), + ): copy_res = vv._step_script_generation(job, intent_res, {"products": [img_res]}) assert "storyboard" in called_types_2 called_types_3 = [] + def spy_get_3(prompt_type, **kwargs): called_types_3.append(prompt_type) return real_get(prompt_type, **kwargs) from packages.application.viral_video.reviewer import Reviewer, ReviewResult + pass_result = ReviewResult(passed=True, score=90, issues=[], rewrite_suggestions=[]) job.intent_result = intent_res job.copy_result = copy_res - with patch.object(pl, "get_template", side_effect=spy_get_3), \ - patch.object(Reviewer, "review", return_value=pass_result) as mock_review: + with ( + patch.object(pl, "get_template", side_effect=spy_get_3), + patch.object(Reviewer, "review", return_value=pass_result) as mock_review, + ): review_res = vv._step_review(job, copy_res) # review 步骤内部直接调用 Reviewer.review，该方法被 mock，因此 get_template 不会被调用； # 此处验证 Reviewer.review 被调用即可说明 review 步骤走通了。 From 631dd643c06691104ebc62bc89ebc05586d7c4a3 Mon Sep 17 00:00:00 2001 From: Xiaoxia Agent Date: Sun, 4 Oct 2026 18:25:04 +0800 Subject: [PATCH 4/5] =?UTF-8?q?fix(viral=5Fvideo):=20#2040=20=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=20style=20check=20F401/F821=20=E5=AF=BC=E5=85=A5?= =?UTF-8?q?=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _step_image_analysis/_step_intent_parsing 的局部导入补 render_system_prompt（F821 未定义名） - _step_script_generation 的冗余 render_system_prompt 导入删除（F401 未使用） black/isort/ruff 全绿，viral_video 单测 150/150 通过。 --- apps/worker/worker_app/tasks/viral_video.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index 13b7e91ba..4e7c572b7 100644 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -888,7 +888,8 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di try: from packages.shared.ai_service import call_llm from packages.application.viral_video.prompt_loader import ( - get_template, render_system_prompt, render_user_prompt, + get_template, + render_user_prompt, ) from packages.application.viral_video.prompts import ( FUSION_INSTRUCTIONS, GLOBAL_CONSTRAINTS, NEGATIVE_RULES, From 233d0272a9a83b082f976dbfbef37c5d95482025 Mon Sep 17 00:00:00 2001 From: xiaoxia-saas-bot Date: Sun, 4 Oct 2026 19:24:00 +0800 Subject: [PATCH 5/5] =?UTF-8?q?fix(viral=5Fvideo):=20#2040=20rebase=20?= =?UTF-8?q?=E5=90=8E=E6=B8=85=E7=90=86F841=E6=AD=BB=E4=BB=A3=E7=A0=81+?= =?UTF-8?q?=E4=BF=9D=E7=95=99#2180=20timeout=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 删除 script_generation 中旧prompt遗留的未使用变量(style_hint/ratio/viral_structure_block/persona_hint) - 冲突解决保留 #2180 全部修复: VLM/LLM timeout 15/25→45, pro 25→60, max_retries 2→1 - black/isort/ruff 全绿；viral 相关71测试全过 --- apps/worker/worker_app/tasks/viral_video.py | 193 +++++++++++++++----- 1 file changed, 145 insertions(+), 48 deletions(-) diff --git a/apps/worker/worker_app/tasks/viral_video.py b/apps/worker/worker_app/tasks/viral_video.py index 4e7c572b7..54dbc16ae 100644 --- a/apps/worker/worker_app/tasks/viral_video.py +++ b/apps/worker/worker_app/tasks/viral_video.py @@ -50,6 +50,7 @@ logger = logging.getLogger(__name__) # ── WS 进度推送 ────────────────────────────────────────────────────────── + def _emit_progress( job_id: str, stage: str, @@ -75,18 +76,22 @@ def _emit_progress( except Exception as e: logger.warning("[爆款视频] WS 进度推送失败: %s", e) + # ── 仓储辅助 ──────────────────────────────────────────────────────────── + def _get_repo_and_job(job_id: str): session = SessionLocal() repo = SQLAlchemyViralVideoJobRepository(session) job = repo.get(job_id) return session, repo, job + def _save_job(repo, job, session): repo.update(job) session.commit() + def _start_trust_chain_preheat(job_id: str, portrait_descriptions: list[str]) -> None: """#2172/#2174 后台启动信任链预热（Seedream t2i 文生图人像），不阻塞调用方。 @@ -152,6 +157,7 @@ def _start_trust_chain_preheat(job_id: str, portrait_descriptions: list[str]) -> t = threading.Thread(target=_run_preheat, name=f"tc-preheat-{job_id[:8]}", daemon=True) t.start() + def _set_stage(job, repo, session, stage: str, message: str, persist: bool = True) -> None: """更新细粒度阶段并持久化到 DB，同时通过 Redis 推送进度事件。 @@ -167,6 +173,7 @@ def _set_stage(job, repo, session, stage: str, message: str, persist: bool = Tru except Exception as e: # 阶段持久化失败不阻塞主流程 logger.warning("[爆款视频] 阶段持久化失败 stage=%s err=%s", stage, e) + # ── worker 心跳（僵尸任务检测） ───────────────────────────────────────── # 心跳间隔（秒）；超过此时间未更新 heartbeat_at 视为 worker 异常 @@ -176,6 +183,7 @@ _STALE_RUNNING_TIMEOUT_SEC = 10 * 60 # 10 分钟 # 心跳过期窗口：heartbeat_at 距 now 超过此时长视为失效 _HEARTBEAT_EXPIRE_SEC = 2 * 60 # 2 分钟 + def _heartbeat_once(job_id: str) -> None: """在独立 session 中更新一次 heartbeat_at（不捕获主流程事务状态）。""" ssn = None @@ -200,6 +208,7 @@ def _heartbeat_once(job_id: str) -> None: except Exception: pass + def _start_heartbeat_thread(job_id: str) -> tuple[threading.Event, threading.Thread]: """启动后台心跳线程，每 _HEARTBEAT_INTERVAL_SEC 秒更新一次 heartbeat_at。 返回 (stop_event, thread)；任务结束时调用 stop_event.set() 停止心跳。 @@ -216,6 +225,7 @@ def _start_heartbeat_thread(job_id: str) -> tuple[threading.Event, threading.Thr t.start() return stop, t + def _recover_stale_jobs() -> int: """启动/定时扫描：把僵尸任务（running 超时且心跳停止）标记为 failed。 返回本次回收的任务数。可由 celery beat 周期性调用，也可在任务启动前顺带扫一次。 @@ -253,6 +263,7 @@ def _recover_stale_jobs() -> int: except Exception: pass + # ── 默认结构 ───────────────────────────────────────────────────────────── _DEFAULT_HARD_CONSTRAINTS = [ @@ -283,6 +294,7 @@ _DEFAULT_NEGATIVE_PROMPTS = [ "低分辨率", ] + def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict: return { "overview": {"theme": "好物推荐", "total_duration": duration, "aspect_ratio": ratio}, @@ -296,8 +308,10 @@ def _empty_copy_result(duration: int = 15, ratio: str = "9:16") -> dict: "title": "", } + # ── 流水线各步骤 ──────────────────────────────────────────────────────── + def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict: d = { "name": "未识别", @@ -315,6 +329,7 @@ def _vision_fallback(idx: int, reason: str, extra: dict | None = None) -> dict: d.update(extra) return d + def _is_vision_result_usable(result: dict) -> bool: """判断 VLM 返回是否有效：name/summary 不能为未识别/无法判断/空，summary 要够长。""" if not isinstance(result, dict): @@ -333,6 +348,7 @@ def _is_vision_result_usable(result: dict) -> bool: return False return True + def _analyze_single_image( idx: int, img_url: str, @@ -346,11 +362,13 @@ def _analyze_single_image( lite 失败/不可用时用 pro 降级重试 1 次。失败/None 最终返回含默认字段的 dict。 """ try: - from packages.shared.ai_service import call_vision - from packages.application.viral_video.prompt_loader import ( - get_template, render_system_prompt, render_user_prompt, - ) from packages.application.viral_video import xml_parser as xp + from packages.application.viral_video.prompt_loader import ( + get_template, + render_system_prompt, + render_user_prompt, + ) + from packages.shared.ai_service import call_vision except ImportError as e: logger.warning("[爆款视频] prompt 模板/解析模块不可用: %s", e) return _vision_fallback(idx, f"fallback_import_error:{e}") @@ -361,15 +379,21 @@ def _analyze_single_image( template = get_template("image_analysis") system = render_system_prompt(template) user = render_user_prompt( - template, image_count=1, industry="通用", + template, + image_count=1, + industry="通用", image_urls=f"第1张：{img_url}", ) def _call(model: str, tmo: int): try: return call_vision( - image_url=img_url, prompt=user, model=model, - max_tokens=2048, temperature=0.3, timeout=tmo, + image_url=img_url, + prompt=user, + model=model, + max_tokens=2048, + temperature=0.3, + timeout=tmo, system_prompt=system, ) except Exception as e: @@ -396,13 +420,18 @@ def _analyze_single_image( packaging = a.get("packaging", "") or "无法判断" summary = a.get("summary", "") or f"{brand} {name}" return { - "name": name, "brand": brand, "category": category, - "appearance": appearance, "packaging": packaging, + "name": name, + "brand": brand, + "category": category, + "appearance": appearance, + "packaging": packaging, "text_on_package": text_list, "key_features": feat_list or [features] if features else ["无法判断"], - "scene": scene, "mood": mood, + "scene": scene, + "mood": mood, "portrait_prompt": a.get("portrait_prompt", "无人像"), - "summary": summary, "_source": "xml", + "summary": summary, + "_source": "xml", } return _vision_fallback(idx, "no_product_tag") @@ -435,6 +464,7 @@ def _analyze_single_image( return pro_result return first_result + def _step_image_analysis(job: ViralVideoJob) -> dict: """步骤 1: 图片 VLM 分析 — 识别产品特征（v1.6 优化：并行 + lite 模型提速）。""" try: @@ -490,6 +520,7 @@ def _step_image_analysis(job: ViralVideoJob) -> dict: return {"products": results} + def _step_video_analysis(job: ViralVideoJob) -> dict | None: """步骤 1.5: 参考视频风格分析（可选）。""" if not job.reference_video_url: @@ -513,14 +544,17 @@ def _step_video_analysis(job: ViralVideoJob) -> dict | None: logger.error("[爆款视频] 视频风格分析失败: %s", e) return {"error": str(e), "source": "failed"} + def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: """步骤 2: 用户文案意图解析（#2040：改为模板 + XML 解析）。""" try: - from packages.shared.ai_service import call_llm - from packages.application.viral_video.prompt_loader import ( - get_template, render_system_prompt, render_user_prompt, - ) from packages.application.viral_video import xml_parser as xp + from packages.application.viral_video.prompt_loader import ( + get_template, + render_system_prompt, + render_user_prompt, + ) + from packages.shared.ai_service import call_llm except ImportError: return {"intent": "推广产品", "key_messages": ["产品亮点"], "tone": "专业", "suggested_title": ""} @@ -556,11 +590,21 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: msgs = [n["text"] for n in xp.find_all(raw, "message") if n["text"]] tone = xp.text_of(raw, "emotion_tone") or "亲切自然" title = xp.text_of(raw, "suggested_title") or xp.text_of(raw, "title") - return {"intent": summary or "推广产品", "key_messages": msgs or ["产品亮点"], "tone": tone, "suggested_title": title} + return { + "intent": summary or "推广产品", + "key_messages": msgs or ["产品亮点"], + "tone": tone, + "suggested_title": title, + } def _fallback(raw: str) -> dict: t = (job.user_copy_text or "").strip() - return {"intent": t[:30] or "推广产品", "key_messages": [t[:80]] if t else ["产品亮点"], "tone": "专业", "suggested_title": ""} + return { + "intent": t[:30] or "推广产品", + "key_messages": [t[:80]] if t else ["产品亮点"], + "tone": "专业", + "suggested_title": "", + } _s = get_shared_settings() _fast = _s.doubao_fast_model @@ -568,8 +612,13 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: for _m, _lbl in [(_fast, "fast"), (_pro, "pro-fallback")]: try: logger.info("[爆款视频] 意图解析 model=%s label=%s", _m, _lbl) - raw = call_llm([{"role": "system", "content": system}, {"role": "user", "content": user}], - temperature=0.4, max_tokens=1024, model=_m, timeout=45) # #2180: 意图解析 LLM 实测需更长响应，原25s太紧 + raw = call_llm( + [{"role": "system", "content": system}, {"role": "user", "content": user}], + temperature=0.4, + max_tokens=1024, + model=_m, + timeout=45, + ) # #2180: 意图解析 LLM 实测需更长响应，原25s太紧 if not raw: continue parsed = _parse(raw) @@ -579,6 +628,7 @@ def _step_intent_parsing(job: ViralVideoJob, image_analysis: dict) -> dict: logger.warning("[爆款视频] 意图解析失败 label=%s err=%s", _lbl, e) return _fallback("") + _PERSONA_STYLE_GUIDE = { "通用个人IP": "亲切自然、像朋友分享好物，第一人称口语化，不端着", "老板型IP": "沉稳大气、有行业格局感，适度使用『我做了XX年』『我一直坚持』等老板视角，语气自信不夸张", @@ -592,6 +642,7 @@ _PERSONA_STYLE_GUIDE = { "测评种草型": "真实测评感、讲使用体验和优缺点对比，带『亲测』『我用了XX天』『实测下来』真实感词汇", } + def _persona_style_hint(persona_id: str) -> str: """根据 persona_id 查文案风格指导；未命中/空值返回通用提示。""" pid = (persona_id or "").strip() @@ -602,6 +653,7 @@ def _persona_style_hint(persona_id: str) -> str: return f"【人设风格：{pid}】按该人设的口吻、话术习惯组织口播和出镜动作" return "【人设风格：未指定】亲切自然、像朋友分享好物" + def _build_products_summary(image_analysis: dict) -> str: """把 VLM 返回的商品分析结果拼给文案/分镜生成 prompt 用。 优先用 summary（自然段落）；没有时用结构化字段兜底拼一段。""" @@ -670,6 +722,7 @@ def _build_products_summary(image_analysis: dict) -> str: lines.append("- " + "，".join(parts)) return "\n".join(lines) + def _safe_json_loads(raw: str | dict | list | None): if raw is None: return None @@ -696,6 +749,7 @@ def _safe_json_loads(raw: str | dict | list | None): pass return None + def _replace_henjin_everywhere(obj: Any) -> Any: """递归遍历 copy_result 里所有字符串值，把'很近'替换成'最近'。 覆盖 overview.theme、scene_and_lighting、voiceover_script、 @@ -711,6 +765,7 @@ def _replace_henjin_everywhere(obj: Any) -> Any: return {k: _replace_henjin_everywhere(v) for k, v in obj.items()} return obj + def _fallback_script(job: ViralVideoJob) -> dict: """脚本生成失败时的兜底脚本（极简但可用）。""" dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) @@ -735,6 +790,7 @@ def _fallback_script(job: ViralVideoJob) -> dict: base["title"] = "好物分享" return base + def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict: """把 LLM 返回的脚本规范化、补默认、校验结构。""" dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) @@ -831,9 +887,11 @@ def _validate_and_normalize_script(raw, job: ViralVideoJob) -> dict: base["title"] = base["overview"]["theme"] return base + def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None: """把 LLM 返回的 XML 分镜规范化为旧 copy_result 结构（供 Seedance 使用）。""" from packages.application.viral_video import xml_parser as xp + dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) ratio = getattr(job, "video_ratio", None) or "9:16" base = _empty_copy_result(dur, ratio) @@ -860,14 +918,18 @@ def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None: ref_idx = xp.attr_int(ref_idx_raw, 0) shot = { "time_range": a.get("time_range") or f"{i * 3}-{(i + 1) * 3}秒", - "shot_type_angle_movement": (xp.text_of(body, "shot_type_angle_movement") if body else "") or a.get("shot_type_angle_movement", "") or "中景平视，固定镜头", + "shot_type_angle_movement": (xp.text_of(body, "shot_type_angle_movement") if body else "") + or a.get("shot_type_angle_movement", "") + or "中景平视，固定镜头", "scene_and_dialogue": (xp.text_of(body, "scene_and_dialogue") if body else "") or "", "action_details": (xp.text_of(body, "action_details") if body else "") or "", "audio_bgm": (xp.text_of(body, "audio_bgm") if body else "") or a.get("bgm_note", "") or "轻快BGM", - "transition": (xp.text_of(body, "transition") if body else "") or a.get("transition", "") or ("硬切" if i < len(clips) - 1 else "结束"), + "transition": (xp.text_of(body, "transition") if body else "") + or a.get("transition", "") + or ("硬切" if i < len(clips) - 1 else "结束"), "reference_image_index": ref_idx, } - voice = (xp.text_of(body, "voice_text") if body else "") + voice = xp.text_of(body, "voice_text") if body else "" if voice: voice_parts.append(voice) if not shot["scene_and_dialogue"]: @@ -883,30 +945,25 @@ def _script_from_xml(raw: str, job: ViralVideoJob) -> dict | None: base["title"] = base["overview"]["theme"] return base + def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: dict) -> dict: """步骤 3: 编导分镜脚本生成（#2040：模板 + XML 解析；输出 copy_result 结构）。""" try: - from packages.shared.ai_service import call_llm from packages.application.viral_video.prompt_loader import ( get_template, render_user_prompt, ) from packages.application.viral_video.prompts import ( - FUSION_INSTRUCTIONS, GLOBAL_CONSTRAINTS, NEGATIVE_RULES, + FUSION_INSTRUCTIONS, + GLOBAL_CONSTRAINTS, + NEGATIVE_RULES, ) + from packages.shared.ai_service import call_llm except ImportError: return _fallback_script(job) products_summary = _build_products_summary(image_analysis) - style_hint = "无" - if isinstance(job.style_guide, dict): - style_hint = ( - f"节奏{job.style_guide.get('cut_speed', '')}、转场{job.style_guide.get('transition', '')}、" - f"色调{job.style_guide.get('color_grade', '')}、能量{job.style_guide.get('energy', '')}" - ) - dur = max(5, min(30, int(getattr(job, "duration", 15) or 15))) - ratio = getattr(job, "video_ratio", None) or "9:16" intent_str = "推广产品" key_msgs = "产品亮点" @@ -916,13 +973,6 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di key_msgs = "、".join(intent.get("key_messages") or []) or key_msgs tone = intent.get("tone") or tone - _vs = (job.viral_structure or "").strip() - if _vs: - viral_structure_block = f"【{_vs}】—— 请严格按照这个爆款结构的节奏/段落顺序编排镜头、台词和情绪节点（开场钩子、痛点、反转、案例、行动号召等按结构走），不要打乱顺序" - else: - viral_structure_block = "未指定（自由编排，但仍需有钩子开头+产品展示+行动号召的基本节奏）" - - persona_hint = _persona_style_hint(getattr(job, "persona_id", "")) fusion_level = getattr(job, "fusion_level", "ai_polish") or "ai_polish" fusion_instruction = FUSION_INSTRUCTIONS.get(fusion_level, FUSION_INSTRUCTIONS["ai_polish"]) @@ -939,15 +989,21 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di ) user = render_user_prompt( template, - duration=dur, image_count=len(job.images or []), + duration=dur, + image_count=len(job.images or []), fusion_result=fusion_brief, image_analysis=products_summary, ) def _try_gen(model: str, temp: float, max_tok: int, label: str, tmo: int = 25): logger.info("[爆款视频] 编导脚本生成 model=%s label=%s timeout=%d", model, label, tmo) - raw = call_llm([{"role": "system", "content": system_tpl}, {"role": "user", "content": user}], - temperature=temp, max_tokens=max_tok, model=model, timeout=tmo) + raw = call_llm( + [{"role": "system", "content": system_tpl}, {"role": "user", "content": user}], + temperature=temp, + max_tokens=max_tok, + model=model, + timeout=tmo, + ) if not raw: return None normalized = _script_from_xml(raw, job) @@ -967,7 +1023,13 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di fallback_marker = "我最近在用的好物" in voiceover has_typo_henjin = "很近" in json.dumps(normalized, ensure_ascii=False) is_fallback = fallback_marker or shots_cnt < 1 or len(voiceover) < 20 or has_typo_henjin - logger.info("[爆款视频] 编导脚本结果 label=%s voiceover_len=%d shots=%d fallback=%s", label, len(voiceover), shots_cnt, is_fallback) + logger.info( + "[爆款视频] 编导脚本结果 label=%s voiceover_len=%d shots=%d fallback=%s", + label, + len(voiceover), + shots_cnt, + is_fallback, + ) return None if is_fallback else normalized _s = get_shared_settings() @@ -992,6 +1054,7 @@ def _step_script_generation(job: ViralVideoJob, intent: dict, image_analysis: di logger.warning("[爆款视频] 编导脚本生成异常: %s，使用兜底脚本", e, exc_info=True) return _fallback_script(job) + def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: """步骤 4: 合规审核（#2040：使用 Reviewer + review 模板，6 维度 + 自动重写 1 次）。 @@ -1000,7 +1063,11 @@ def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: try: from packages.application.viral_video.reviewer import Reviewer from packages.application.viral_video.schemas import ( - FusionResult, IntentResult, CoreMessage, PersonalBrand, ScriptSegment, + CoreMessage, + FusionResult, + IntentResult, + PersonalBrand, + ScriptSegment, ) except ImportError as e: logger.warning("[爆款视频] reviewer 模块不可用，跳过审核: %s", e) @@ -1010,14 +1077,17 @@ def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: title = (copy_result or {}).get("title") or (copy_result or {}).get("overview", {}).get("theme", "") intent_data = job.intent_result or {} - core_msgs = [CoreMessage(text=str(m), must_keep=True, confidence=0.9) for m in (intent_data.get("key_messages") or [])] + core_msgs = [ + CoreMessage(text=str(m), must_keep=True, confidence=0.9) for m in (intent_data.get("key_messages") or []) + ] brands: list[PersonalBrand] = [] brand_text = intent_data.get("brand_text") or intent_data.get("suggested_title") or "" if brand_text: brands.append(PersonalBrand(text=str(brand_text), category="brand")) intent_obj = IntentResult( intent_summary=intent_data.get("intent", "") or "推广产品", - core_messages=core_msgs, personal_brands=brands, + core_messages=core_msgs, + personal_brands=brands, ) fusion_obj = FusionResult( title=title or "", @@ -1041,7 +1111,11 @@ def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: try: rewritten = reviewer.rewrite(fusion_obj, review_res, intent_obj, fusion_level) if rewritten and (rewritten.title or rewritten.script_segments): - new_voice = rewritten.script_segments[0].text if rewritten.script_segments else (rewritten.hook or voiceover) + new_voice = ( + rewritten.script_segments[0].text + if rewritten.script_segments + else (rewritten.hook or voiceover) + ) new_copy = dict(copy_result) new_copy["voiceover_script"] = new_voice new_copy["final_copy"] = new_voice @@ -1057,7 +1131,10 @@ def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: "passed": review_res.passed, "score": 90 if review_res.passed else 60, "details": {i.dimension: i.text for i in review_res.issues}, - "issues": [{"dimension": i.dimension, "severity": i.severity, "location": i.location, "text": i.text} for i in review_res.issues], + "issues": [ + {"dimension": i.dimension, "severity": i.severity, "location": i.location, "text": i.text} + for i in review_res.issues + ], } if rewritten_voice is not None: result["rewritten_copy"] = new_copy @@ -1068,6 +1145,7 @@ def _step_review(job: ViralVideoJob, copy_result: dict) -> dict: logger.warning("[爆款视频] 审核异常，跳过: %s", e, exc_info=True) return {"passed": True, "score": 75, "details": {}, "issues": []} + def _step_tts(job: ViralVideoJob, voiceover_script: str): """步骤 5: CosyVoice 整段配音 → 返回本地 MP3 Path；失败返回 None。""" try: @@ -1106,6 +1184,7 @@ def _step_tts(job: ViralVideoJob, voiceover_script: str): logger.warning("[爆款视频] TTS 配音失败: %s", e, exc_info=True) return None + def _upload_tts_to_oss(job: ViralVideoJob, tts_path) -> str | None: """把 TTS 本地 mp3 上传到 OSS，返回公网 URL（供 Seedance 做 reference_audios 口型驱动用）。""" if tts_path is None: @@ -1125,6 +1204,7 @@ def _upload_tts_to_oss(job: ViralVideoJob, tts_path) -> str | None: logger.warning("[爆款视频] TTS 上传 OSS 失败: %s", e, exc_info=True) return None + def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str: """把编导脚本拼成 Seedance 长 prompt。""" if not isinstance(copy_result, dict) or not copy_result: @@ -1175,6 +1255,7 @@ def _assemble_seedance_prompt(copy_result: dict, job: ViralVideoJob) -> str: lines.append("，".join([str(x) for x in np if x])) return "\n".join(lines) + def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | None) -> tuple[str, dict | None]: """步骤 6: v1.6 单次 Seedance 生成（不再分段/拼接）。 @@ -1288,6 +1369,7 @@ def _step_render(job: ViralVideoJob, copy_result: dict, tts_audio_url: str | Non logger.info("[爆款视频] 单次生成完成: path=%s size=%d usage=%s", video_path, size, usage) return str(video_path), (usage if isinstance(usage, dict) else None) + def _step_upload(job: ViralVideoJob, video_path: str) -> str: """步骤 7: OSS 上传。""" from video_processing.oss_helpers import upload_to_oss @@ -1300,6 +1382,7 @@ def _step_upload(job: ViralVideoJob, video_path: str) -> str: raise RuntimeError(f"OSS 上传失败: storage_key={storage_key}") return video_url + def _wait_oss_ready(url: str, timeout_sec: int = 10) -> bool: """轮询 OSS 公网 URL，直到 HEAD 返回 200 或超时。 用于缓解 OSS 上传后 1-5s 公网 eventual consistency 导致的 NoSuchKey。 @@ -1320,8 +1403,10 @@ def _wait_oss_ready(url: str, timeout_sec: int = 10) -> bool: logger.warning("[爆款视频] OSS 成片在 %ds 内未就绪 last_status=%s url=%s", timeout_sec, last_status, url[:120]) return False + # ── 主编排器 ──────────────────────────────────────────────────────────── + @shared_task( bind=True, max_retries=2, @@ -1406,6 +1491,7 @@ def run_viral_video_pipeline(self: Task, job_id: str) -> dict: if session: session.close() + @shared_task(bind=True, max_retries=2, name="worker.resume_viral_video_pipeline") def resume_viral_video_pipeline(self: Task, job_id: str) -> dict: """旧 confirm-intent 路径兼容：从 WAIT_USER_CONFIRM 跑完整个渲染。""" @@ -1427,6 +1513,7 @@ def resume_viral_video_pipeline(self: Task, job_id: str) -> dict: if session: session.close() + @shared_task(bind=True, max_retries=1, name="worker.run_video_style_analysis") def run_video_style_analysis(self: Task, job_id: str) -> dict: """独立的视频风格分析任务。""" @@ -1450,8 +1537,10 @@ def run_video_style_analysis(self: Task, job_id: str) -> dict: if session: session.close() + # ── 失败处理 ──────────────────────────────────────────────────────────── + def _mark_failed_and_notify(job_id: str, session, repo, job, err_msg: str, stage: str = "") -> None: """标记任务失败并通知。若传入的 session 已失效（因前面异常导致 rollback 状态）， 会自动 fallback 到新建 SessionLocal 重新标记，确保状态一定落库。""" @@ -1492,8 +1581,10 @@ def _mark_failed_and_notify(job_id: str, session, repo, job, err_msg: str, stage event_type="viral_video:failed", ) + # ── v1.5/v1.6 三步分步流水线 Celery 任务 ───────────────────────────────── + @shared_task( bind=True, max_retries=1, @@ -1561,6 +1652,7 @@ def run_viral_video_analyze(self: Task, job_id: str) -> dict: if session: session.close() + @shared_task( bind=True, max_retries=1, @@ -1663,6 +1755,7 @@ def run_viral_video_generate_copy(self: Task, job_id: str) -> dict: if session: session.close() + def _quick_compliance_blacklist_check(copy_result: dict) -> None: """阶段2快速黑名单检查：不调用 LLM，只扫描高风险关键词；命中则在 voiceover 中就地替换。 @@ -1704,6 +1797,7 @@ def _quick_compliance_blacklist_check(copy_result: dict) -> None: for bk, bv in BLACKLIST.items(): copy_result[k] = copy_result[k].replace(bk, bv) + def _try_refund_viral_video(job: ViralVideoJob) -> None: """爆款视频生成失败：若已预扣积分则全额退款。""" try: @@ -1733,6 +1827,7 @@ def _try_refund_viral_video(job: ViralVideoJob) -> None: except Exception: logger.exception("[爆款视频] 失败退款异常 job_id=%s", job.id) + def _settle_viral_video(job: ViralVideoJob, usage: dict | None) -> None: """爆款视频生成成功：按实际 usage 结算，多退少补，写 credits_cost。""" try: @@ -1803,6 +1898,7 @@ def _settle_viral_video(job: ViralVideoJob, usage: dict | None) -> None: job.credits_cost = float(getattr(job, "credits_prepaid", 0) or 0) job.credits_prepaid = 0.0 + def _run_render_pipeline(job_id: str, session, repo, job) -> dict: """v1.6.1 阶段3：出片前合规审核（LLM 深度）→ TTS → Seedance → Upload → Completed。 @@ -1889,6 +1985,7 @@ def _run_render_pipeline(job_id: str, session, repo, job) -> dict: logger.info("[爆款视频] 任务完成: job_id=%s video_url=%s", job_id, video_url) return {"ok": True, "job_id": job_id, "video_url": video_url} + @shared_task(bind=True, max_retries=2, name="worker.run_viral_video_render") def run_viral_video_render(self: Task, job_id: str) -> dict: """v1.6 阶段3：TTS + 单次 Seedance 生成 + 上传。"""