diff --git a/app/services/SDE/short_drama_explanation.py b/app/services/SDE/short_drama_explanation.py index 4fc2478..6910324 100644 --- a/app/services/SDE/short_drama_explanation.py +++ b/app/services/SDE/short_drama_explanation.py @@ -11,7 +11,7 @@ import os import json import requests -from typing import Dict, Any, Optional +from typing import Dict, Any, Optional, Tuple from loguru import logger from app.config import config from app.utils.utils import get_uuid, storage_dir @@ -363,7 +363,179 @@ class SubtitleAnalyzer: logger.error(f"保存分析结果时发生错误: {str(e)}") return "" - def generate_narration_script(self, short_name: str, plot_analysis: str, subtitle_content: str = "", temperature: float = 0.7) -> Dict[str, Any]: + def _render_prompt(self, name: str, parameters: Dict[str, Any]) -> Tuple[str, Optional[str]]: + prompt = PromptManager.get_prompt( + category="short_drama_narration", + name=name, + parameters=parameters, + ) + prompt_object = PromptManager.get_prompt_object( + category="short_drama_narration", + name=name, + ) + return prompt, prompt_object.get_system_prompt() + + def _generate_json_text( + self, + prompt: str, + system_prompt: Optional[str], + temperature: float, + ) -> Dict[str, Any]: + if self.is_native_gemini: + return self._generate_narration_with_native_gemini(prompt, temperature, system_prompt, json_output=True) + return self._generate_narration_with_openai_compatible(prompt, temperature, system_prompt, json_output=True) + + def _generate_plain_text( + self, + prompt: str, + system_prompt: Optional[str], + temperature: float, + ) -> Dict[str, Any]: + if self.is_native_gemini: + result = self._generate_narration_with_native_gemini(prompt, temperature, system_prompt, json_output=False) + else: + result = self._generate_narration_with_openai_compatible(prompt, temperature, system_prompt, json_output=False) + if result.get("status") == "success": + result["narration_copy"] = str(result.get("narration_script", "")).strip() + return result + + def generate_narration_copy( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str = "", + temperature: float = 0.7, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> Dict[str, Any]: + """生成供用户审核修改的解说正文。""" + try: + prompt, system_prompt = self._render_prompt( + "narration_copy", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "narration_language": narration_language, + }, + ) + return self._generate_plain_text(prompt, system_prompt, temperature) + except Exception as e: + logger.error(f"解说文案正文生成过程中发生错误: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } + + def match_narration_copy_to_script( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str, + narration_copy: str, + temperature: float = 0.3, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + original_sound_ratio: int = 30, + ) -> Dict[str, Any]: + """将用户审核后的解说正文匹配到字幕时间戳。""" + try: + prompt, system_prompt = self._render_prompt( + "script_matching", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "narration_copy": narration_copy, + "narration_language": narration_language, + "original_sound_ratio": int(original_sound_ratio), + }, + ) + return self._generate_json_text(prompt, system_prompt, min(float(temperature), 0.3)) + except Exception as e: + logger.error(f"解说文案画面匹配过程中发生错误: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } + + def plan_narration_segments( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str = "", + temperature: float = 0.3, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> Dict[str, Any]: + """规划短剧解说片段，只输出片段来源和意图。""" + try: + prompt, system_prompt = self._render_prompt( + "segment_planning", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "narration_language": narration_language, + }, + ) + return self._generate_json_text(prompt, system_prompt, min(float(temperature), 0.3)) + except Exception as e: + logger.error(f"片段规划过程中发生错误: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } + + def repair_narration_script( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str, + invalid_script: str, + validation_errors: str, + temperature: float = 0.3, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> Dict[str, Any]: + """根据确定性校验错误修复解说脚本。""" + try: + prompt, system_prompt = self._render_prompt( + "script_repair", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "invalid_script": invalid_script, + "validation_errors": validation_errors, + "narration_language": narration_language, + }, + ) + return self._generate_json_text(prompt, system_prompt, min(float(temperature), 0.3)) + except Exception as e: + logger.error(f"解说文案修复过程中发生错误: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } + + def generate_narration_script( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str = "", + temperature: float = 0.7, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> Dict[str, Any]: """ 根据剧情分析生成解说文案 @@ -372,28 +544,36 @@ class SubtitleAnalyzer: plot_analysis: 剧情分析内容 subtitle_content: 原始字幕内容，用于提供准确的时间戳信息 temperature: 生成温度，控制创造性，默认0.7 + narration_language: 解说台词目标语言 Returns: Dict[str, Any]: 包含生成结果的字典 """ try: - # 使用新的提示词管理系统构建提示词 - prompt = PromptManager.get_prompt( - category="short_drama_narration", - name="script_generation", - parameters={ + segment_plan_result = self.plan_narration_segments( + short_name=short_name, + plot_analysis=plot_analysis, + subtitle_content=subtitle_content, + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, + ) + if segment_plan_result["status"] != "success": + return segment_plan_result + + prompt, system_prompt = self._render_prompt( + "script_generation", + { "drama_name": short_name, + "drama_genre": drama_genre, "plot_analysis": plot_analysis, - "subtitle_content": subtitle_content - } + "subtitle_content": subtitle_content, + "segment_plan": segment_plan_result["narration_script"], + "narration_language": narration_language, + }, ) - if self.is_native_gemini: - # 使用原生Gemini API格式 - return self._generate_narration_with_native_gemini(prompt, temperature) - else: - # 使用OpenAI兼容格式 - return self._generate_narration_with_openai_compatible(prompt, temperature) + return self._generate_json_text(prompt, system_prompt, temperature) except Exception as e: logger.error(f"解说文案生成过程中发生错误: {str(e)}") @@ -403,16 +583,35 @@ class SubtitleAnalyzer: "temperature": self.temperature } - def _generate_narration_with_native_gemini(self, prompt: str, temperature: float) -> Dict[str, Any]: + def _generate_narration_with_native_gemini( + self, + prompt: str, + temperature: float, + system_prompt: Optional[str] = None, + json_output: bool = True, + ) -> Dict[str, Any]: """使用原生Gemini API生成解说文案""" try: # 构建原生Gemini API请求数据 # 为了确保JSON输出，在提示词中添加更强的约束 - enhanced_prompt = f"{prompt}\n\n请确保输出严格的JSON格式，不要包含任何其他文字或标记。" + enhanced_prompt = ( + f"{prompt}\n\n请确保输出严格的JSON格式，不要包含任何其他文字或标记。" + if json_output + else prompt + ) payload = { "systemInstruction": { - "parts": [{"text": "你是一位专业的短视频解说脚本撰写专家。你必须严格按照JSON格式输出，不能包含任何其他文字、说明或代码块标记。"}] + "parts": [ + { + "text": system_prompt + or ( + "你必须严格按照JSON格式输出，不能包含任何其他文字、说明或代码块标记。" + if json_output + else "你是一位专业的短剧解说文案助手。" + ) + } + ] }, "contents": [{ "parts": [{"text": enhanced_prompt}] @@ -423,7 +622,6 @@ class SubtitleAnalyzer: "topP": 0.95, "maxOutputTokens": 64000, "candidateCount": 1, - "stopSequences": ["```", "注意", "说明"] }, "safetySettings": [ { @@ -444,6 +642,8 @@ class SubtitleAnalyzer: } ] } + if json_output: + payload["generationConfig"]["stopSequences"] = ["```", "注意", "说明"] # 构建请求URL url = f"{self.base_url}/models/{self.model}:generateContent" @@ -523,21 +723,27 @@ class SubtitleAnalyzer: "temperature": temperature } - def _generate_narration_with_openai_compatible(self, prompt: str, temperature: float) -> Dict[str, Any]: + def _generate_narration_with_openai_compatible( + self, + prompt: str, + temperature: float, + system_prompt: Optional[str] = None, + json_output: bool = True, + ) -> Dict[str, Any]: """使用OpenAI兼容API生成解说文案""" try: # 构建OpenAI格式的请求数据 payload = { "model": self.model, "messages": [ - {"role": "system", "content": "你是一位专业的短视频解说脚本撰写专家。"}, + {"role": "system", "content": system_prompt or ("你必须严格按照JSON格式输出。" if json_output else "你是一位专业的短剧解说文案助手。")}, {"role": "user", "content": prompt} ], "temperature": temperature } # 对特定模型添加响应格式设置 - if self.model not in ["deepseek-reasoner"]: + if json_output and self.model not in ["deepseek-reasoner"]: payload["response_format"] = {"type": "json_object"} # 构建请求地址 @@ -691,7 +897,9 @@ def generate_narration_script( temperature: float = 1.0, save_result: bool = False, output_path: Optional[str] = None, - provider: Optional[str] = None + provider: Optional[str] = None, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", ) -> Dict[str, Any]: """ 根据剧情分析生成解说文案的便捷函数 @@ -707,6 +915,7 @@ def generate_narration_script( save_result: 是否保存结果到文件 output_path: 输出文件路径 provider: 提供商类型 + narration_language: 解说台词目标语言 Returns: Dict[str, Any]: 包含生成结果的字典 @@ -721,7 +930,14 @@ def generate_narration_script( ) # 生成解说文案 - result = analyzer.generate_narration_script(short_name, plot_analysis, subtitle_content or "", temperature) + result = analyzer.generate_narration_script( + short_name, + plot_analysis, + subtitle_content or "", + temperature, + narration_language, + drama_genre, + ) # 保存结果 if save_result and result["status"] == "success": @@ -730,6 +946,107 @@ def generate_narration_script( return result +def generate_narration_copy( + short_name: str = None, + plot_analysis: str = None, + subtitle_content: str = None, + api_key: Optional[str] = None, + model: Optional[str] = None, + base_url: Optional[str] = None, + temperature: float = 0.7, + provider: Optional[str] = None, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", +) -> Dict[str, Any]: + """生成可供用户审核修改的解说正文。""" + analyzer = SubtitleAnalyzer( + temperature=temperature, + api_key=api_key, + model=model, + base_url=base_url, + provider=provider, + ) + + return analyzer.generate_narration_copy( + short_name=short_name, + plot_analysis=plot_analysis or "", + subtitle_content=subtitle_content or "", + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, + ) + + +def match_narration_copy_to_script( + short_name: str = None, + plot_analysis: str = None, + subtitle_content: str = None, + narration_copy: str = None, + api_key: Optional[str] = None, + model: Optional[str] = None, + base_url: Optional[str] = None, + temperature: float = 0.3, + provider: Optional[str] = None, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + original_sound_ratio: int = 30, +) -> Dict[str, Any]: + """将用户审核后的解说正文匹配到字幕时间戳。""" + analyzer = SubtitleAnalyzer( + temperature=temperature, + api_key=api_key, + model=model, + base_url=base_url, + provider=provider, + ) + + return analyzer.match_narration_copy_to_script( + short_name=short_name, + plot_analysis=plot_analysis or "", + subtitle_content=subtitle_content or "", + narration_copy=narration_copy or "", + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, + original_sound_ratio=original_sound_ratio, + ) + + +def repair_narration_script( + short_name: str = None, + plot_analysis: str = None, + subtitle_content: str = None, + invalid_script: str = None, + validation_errors: str = None, + api_key: Optional[str] = None, + model: Optional[str] = None, + base_url: Optional[str] = None, + temperature: float = 0.3, + provider: Optional[str] = None, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", +) -> Dict[str, Any]: + """根据校验错误修复解说文案的便捷函数。""" + analyzer = SubtitleAnalyzer( + temperature=temperature, + api_key=api_key, + model=model, + base_url=base_url, + provider=provider, + ) + + return analyzer.repair_narration_script( + short_name=short_name, + plot_analysis=plot_analysis or "", + subtitle_content=subtitle_content or "", + invalid_script=invalid_script or "", + validation_errors=validation_errors or "", + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, + ) + + if __name__ == '__main__': text_api_key = "skxxxx" text_model = "gemini-2.0-flash" diff --git a/app/services/llm/base.py b/app/services/llm/base.py index 737ceb9..e1a899a 100644 --- a/app/services/llm/base.py +++ b/app/services/llm/base.py @@ -178,6 +178,27 @@ class TextModelProvider(BaseLLMProvider): 生成的文本内容 """ pass + + async def generate_text_stream(self, + prompt: str, + system_prompt: Optional[str] = None, + temperature: float = 1.0, + max_tokens: Optional[int] = None, + response_format: Optional[str] = None, + on_chunk=None, + **kwargs) -> str: + """生成文本内容并尽可能回调流式片段；默认退化为一次性输出。""" + result = await self.generate_text( + prompt=prompt, + system_prompt=system_prompt, + temperature=temperature, + max_tokens=max_tokens, + response_format=response_format, + **kwargs, + ) + if on_chunk: + on_chunk({"type": "content", "text": result}) + return result def _build_messages(self, prompt: str, system_prompt: Optional[str] = None) -> List[Dict[str, str]]: """构建消息列表""" diff --git a/app/services/llm/migration_adapter.py b/app/services/llm/migration_adapter.py index 7bd5142..96b165f 100644 --- a/app/services/llm/migration_adapter.py +++ b/app/services/llm/migration_adapter.py @@ -225,6 +225,229 @@ class SubtitleAnalyzerAdapter: output = output.strip() return output + + def _render_prompt(self, name: str, parameters: Dict[str, Any]) -> tuple[str, Optional[str]]: + prompt = PromptManager.get_prompt( + category="short_drama_narration", + name=name, + parameters=parameters, + ) + prompt_object = PromptManager.get_prompt_object( + category="short_drama_narration", + name=name, + ) + return prompt, prompt_object.get_system_prompt() + + def _generate_json_text( + self, + prompt: str, + system_prompt: Optional[str], + temperature: float, + stream_callback=None, + ) -> str: + generate_func = ( + UnifiedLLMService.generate_text_stream + if stream_callback + else UnifiedLLMService.generate_text + ) + kwargs = { + "prompt": prompt, + "system_prompt": system_prompt, + "provider": self.provider, + "temperature": temperature, + "response_format": "json", + "api_key": self.api_key, + "api_base": self.base_url, + } + if stream_callback: + kwargs["on_chunk"] = stream_callback + result = self._run_async_safely(generate_func, **kwargs) + return self._clean_json_output(result) + + def _generate_plain_text(self, prompt: str, system_prompt: Optional[str], temperature: float) -> str: + result = self._run_async_safely( + UnifiedLLMService.generate_text, + prompt=prompt, + system_prompt=system_prompt, + provider=self.provider, + temperature=temperature, + api_key=self.api_key, + api_base=self.base_url, + ) + return str(result or "").strip() + + def generate_narration_copy( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str = "", + temperature: float = 0.7, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> Dict[str, Any]: + """Generate editable narration copy before timeline matching.""" + try: + prompt, system_prompt = self._render_prompt( + "narration_copy", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "narration_language": narration_language, + }, + ) + narration_copy = self._generate_plain_text(prompt, system_prompt, temperature) + return { + "status": "success", + "narration_copy": narration_copy, + "model": self.model, + "temperature": temperature, + } + except Exception as e: + logger.error(f"解说文案正文生成失败: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } + + def match_narration_copy_to_script( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str, + narration_copy: str, + temperature: float = 0.3, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + original_sound_ratio: int = 30, + stream_callback=None, + ) -> Dict[str, Any]: + """Match reviewed narration copy to source footage and return JSON script.""" + try: + prompt, system_prompt = self._render_prompt( + "script_matching", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "narration_copy": narration_copy, + "narration_language": narration_language, + "original_sound_ratio": int(original_sound_ratio), + }, + ) + narration_script = self._generate_json_text( + prompt, + system_prompt, + min(float(temperature), 0.3), + stream_callback=stream_callback, + ) + return { + "status": "success", + "narration_script": narration_script, + "model": self.model, + "temperature": temperature, + } + except Exception as e: + logger.error(f"解说文案画面匹配失败: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } + + def plan_narration_segments( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str = "", + temperature: float = 0.3, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> str: + """Plan source segments before generating final copy.""" + prompt, system_prompt = self._render_prompt( + "segment_planning", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "narration_language": narration_language, + }, + ) + return self._generate_json_text(prompt, system_prompt, min(float(temperature), 0.3)) + + def generate_narration_script_from_plan( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str, + segment_plan: str, + temperature: float = 0.7, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> str: + prompt, system_prompt = self._render_prompt( + "script_generation", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "segment_plan": segment_plan, + "narration_language": narration_language, + }, + ) + return self._generate_json_text(prompt, system_prompt, temperature) + + def repair_narration_script( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str, + invalid_script: str, + validation_errors: str, + temperature: float = 0.3, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + stream_callback=None, + ) -> Dict[str, Any]: + """Repair a generated script once after deterministic validation fails.""" + try: + prompt, system_prompt = self._render_prompt( + "script_repair", + { + "drama_name": short_name, + "drama_genre": drama_genre, + "plot_analysis": plot_analysis, + "subtitle_content": subtitle_content, + "invalid_script": invalid_script, + "validation_errors": validation_errors, + "narration_language": narration_language, + }, + ) + repaired_script = self._generate_json_text( + prompt, + system_prompt, + min(float(temperature), 0.3), + stream_callback=stream_callback, + ) + return { + "status": "success", + "narration_script": repaired_script, + "model": self.model, + "temperature": temperature, + } + except Exception as e: + logger.error(f"解说文案修复失败: {str(e)}") + return { + "status": "error", + "message": str(e), + "temperature": temperature, + } def analyze_subtitle(self, subtitle_content: str) -> Dict[str, Any]: """ @@ -262,7 +485,15 @@ class SubtitleAnalyzerAdapter: "temperature": 1.0 } - def generate_narration_script(self, short_name: str, plot_analysis: str, subtitle_content: str = "", temperature: float = 0.7) -> Dict[str, Any]: + def generate_narration_script( + self, + short_name: str, + plot_analysis: str, + subtitle_content: str = "", + temperature: float = 0.7, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", + ) -> Dict[str, Any]: """ 生成解说文案 - 兼容原有接口 @@ -271,36 +502,30 @@ class SubtitleAnalyzerAdapter: plot_analysis: 剧情分析内容 subtitle_content: 原始字幕内容，用于提供准确的时间戳信息 temperature: 生成温度 + narration_language: 解说台词目标语言 Returns: 生成结果字典 """ try: - # 使用新的提示词管理系统构建提示词 - prompt = PromptManager.get_prompt( - category="short_drama_narration", - name="script_generation", - parameters={ - "drama_name": short_name, - "plot_analysis": plot_analysis, - "subtitle_content": subtitle_content - } - ) - - # 使用统一服务生成文案 - result = self._run_async_safely( - UnifiedLLMService.generate_text, - prompt=prompt, - system_prompt="你是一位专业的短视频解说脚本撰写专家。", - provider=self.provider, + segment_plan = self.plan_narration_segments( + short_name=short_name, + plot_analysis=plot_analysis, + subtitle_content=subtitle_content, temperature=temperature, - response_format="json", - api_key=self.api_key, - api_base=self.base_url + narration_language=narration_language, + drama_genre=drama_genre, + ) + + cleaned_result = self.generate_narration_script_from_plan( + short_name=short_name, + plot_analysis=plot_analysis, + subtitle_content=subtitle_content, + segment_plan=segment_plan, + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, ) - - # 清理JSON输出 - cleaned_result = self._clean_json_output(result) # 新的提示词系统返回的是包含items数组的JSON格式 # 为了保持向后兼容，我们需要直接返回这个JSON字符串 diff --git a/app/services/llm/openai_compatible_provider.py b/app/services/llm/openai_compatible_provider.py index 9a2b183..d469955 100644 --- a/app/services/llm/openai_compatible_provider.py +++ b/app/services/llm/openai_compatible_provider.py @@ -233,25 +233,17 @@ class OpenAICompatibleVisionProvider(_OpenAICompatibleBase, VisionModelProvider) class OpenAICompatibleTextProvider(_OpenAICompatibleBase, TextModelProvider): """OpenAI 兼容文本模型提供商。""" - async def generate_text( + def _build_text_completion_kwargs( self, - prompt: str, - system_prompt: Optional[str] = None, - temperature: float = 1.0, - max_tokens: Optional[int] = None, - response_format: Optional[str] = None, - **kwargs, - ) -> str: - messages = self._build_messages(prompt, system_prompt) + messages: List[Dict[str, str]], + temperature: float, + max_tokens: Optional[int], + response_format: Optional[str], + kwargs: Dict[str, Any], + ) -> Dict[str, Any]: model_name = _normalize_model_name(self.model_name) - - client = self._build_client( - api_key_override=kwargs.get("api_key"), - base_url_override=kwargs.get("api_base"), - timeout_override=config.app.get("llm_text_timeout", 180), - ) - - temperature_override = kwargs.pop("temperature", None) + generation_kwargs = dict(kwargs) + temperature_override = generation_kwargs.pop("temperature", None) if temperature_override is None and temperature != 1.0: temperature_override = temperature @@ -263,12 +255,63 @@ class OpenAICompatibleTextProvider(_OpenAICompatibleBase, TextModelProvider): self._build_chat_completion_options( "text", temperature=temperature_override, - max_tokens=kwargs.pop("max_tokens", max_tokens), - **kwargs, + max_tokens=generation_kwargs.pop("max_tokens", max_tokens), + **generation_kwargs, ) ) if response_format == "json": completion_kwargs["response_format"] = {"type": "json_object"} + return completion_kwargs + + @staticmethod + def _emit_stream_chunk(on_chunk, chunk_type: str, text: str): + if not on_chunk or not text: + return + try: + on_chunk({"type": chunk_type, "text": text}) + except Exception as exc: + logger.debug(f"流式回调更新失败: {exc}") + + @staticmethod + def _extract_reasoning_delta(delta: Any) -> str: + if delta is None: + return "" + if hasattr(delta, "reasoning_content"): + value = getattr(delta, "reasoning_content") + if value: + return str(value) + if hasattr(delta, "model_dump"): + data = delta.model_dump(exclude_none=True) + for key in ("reasoning_content", "reasoning", "thinking"): + value = data.get(key) + if value: + return str(value) + return "" + + async def generate_text( + self, + prompt: str, + system_prompt: Optional[str] = None, + temperature: float = 1.0, + max_tokens: Optional[int] = None, + response_format: Optional[str] = None, + **kwargs, + ) -> str: + messages = self._build_messages(prompt, system_prompt) + + client = self._build_client( + api_key_override=kwargs.get("api_key"), + base_url_override=kwargs.get("api_base"), + timeout_override=config.app.get("llm_text_timeout", 180), + ) + + completion_kwargs = self._build_text_completion_kwargs( + messages, + temperature, + max_tokens, + response_format, + kwargs, + ) try: response = await client.chat.completions.create(**completion_kwargs) @@ -306,5 +349,81 @@ class OpenAICompatibleTextProvider(_OpenAICompatibleBase, TextModelProvider): logger.error(f"OpenAI 兼容接口调用失败: {exc}") raise APICallError(f"调用失败: {exc}") + async def generate_text_stream( + self, + prompt: str, + system_prompt: Optional[str] = None, + temperature: float = 1.0, + max_tokens: Optional[int] = None, + response_format: Optional[str] = None, + on_chunk=None, + **kwargs, + ) -> str: + messages = self._build_messages(prompt, system_prompt) + client = self._build_client( + api_key_override=kwargs.get("api_key"), + base_url_override=kwargs.get("api_base"), + timeout_override=config.app.get("llm_text_timeout", 180), + ) + completion_kwargs = self._build_text_completion_kwargs( + messages, + temperature, + max_tokens, + response_format, + kwargs, + ) + completion_kwargs["stream"] = True + + async def collect_stream() -> str: + content_parts: List[str] = [] + stream = await client.chat.completions.create(**completion_kwargs) + async for chunk in stream: + if not getattr(chunk, "choices", None): + continue + delta = chunk.choices[0].delta + reasoning_delta = self._extract_reasoning_delta(delta) + if reasoning_delta: + self._emit_stream_chunk(on_chunk, "reasoning", reasoning_delta) + + content_delta = getattr(delta, "content", None) if delta is not None else None + if content_delta: + content_parts.append(content_delta) + self._emit_stream_chunk(on_chunk, "content", content_delta) + + result = "".join(content_parts).strip() + if result: + self._emit_stream_chunk(on_chunk, "done", "") + return result + raise APICallError("OpenAI 兼容接口返回空响应") + + try: + return await collect_stream() + + except OpenAIBadRequestError as exc: + error_msg = str(exc) + if response_format == "json" and _is_response_format_error(error_msg): + logger.warning("目标网关不支持流式 response_format，回退为提示词约束 JSON 输出") + completion_kwargs.pop("response_format", None) + messages[-1]["content"] += "\n\n请确保输出严格的JSON格式，不要包含任何其他文字或标记。" + result = await collect_stream() + return _clean_json_output(result) + + if _is_content_filter_error(error_msg): + raise ContentFilterError(f"内容被安全过滤器阻止: {error_msg}") + raise APICallError(f"请求错误: {error_msg}") + + except OpenAIAuthError as exc: + logger.error(f"OpenAI 兼容接口认证失败: {exc}") + raise AuthenticationError(str(exc)) + except OpenAIRateLimitError as exc: + logger.error(f"OpenAI 兼容接口速率限制: {exc}") + raise RateLimitError(str(exc)) + except OpenAIAPIError as exc: + logger.error(f"OpenAI 兼容接口 API 错误: {exc}") + raise APICallError(f"API 错误: {exc}") + except Exception as exc: + logger.error(f"OpenAI 兼容接口流式调用失败: {exc}") + raise APICallError(f"流式调用失败: {exc}") + async def _make_api_call(self, payload: Dict[str, Any]) -> Dict[str, Any]: return payload diff --git a/app/services/llm/test_subtitle_adapter_pipeline_unittest.py b/app/services/llm/test_subtitle_adapter_pipeline_unittest.py new file mode 100644 index 0000000..2245031 --- /dev/null +++ b/app/services/llm/test_subtitle_adapter_pipeline_unittest.py @@ -0,0 +1,177 @@ +import json +import unittest +from unittest import mock + +from app.services.llm.migration_adapter import SubtitleAnalyzerAdapter +from app.services.llm.unified_service import UnifiedLLMService + + +class SubtitleAnalyzerAdapterPipelineTests(unittest.TestCase): + def test_generate_narration_copy_uses_plain_text_prompt_with_selected_type(self): + adapter = SubtitleAnalyzerAdapter( + api_key="sk-test", + model="test-model", + base_url="https://example.test/v1", + provider="openai", + ) + + with mock.patch.object(adapter, "_run_async_safely", return_value="她被家人逼到绝路，反击从这一刻开始。") as call: + result = adapter.generate_narration_copy( + short_name="测试短剧", + plot_analysis="女主被家人误会后反击。", + subtitle_content="# 视频 1: 1.mp4\n00:00:01,000 --> 00:00:04,000\n女主被误会。", + temperature=0.7, + narration_language="简体中文（中国）", + drama_genre="家庭伦理", + ) + + self.assertEqual("success", result["status"]) + self.assertIn("反击", result["narration_copy"]) + self.assertIn("家庭伦理", call.call_args.kwargs["prompt"]) + self.assertNotIn("response_format", call.call_args.kwargs) + + def test_match_narration_copy_to_script_uses_json_prompt_with_selected_type(self): + adapter = SubtitleAnalyzerAdapter( + api_key="sk-test", + model="test-model", + base_url="https://example.test/v1", + provider="openai", + ) + matched = json.dumps( + { + "items": [ + { + "_id": 1, + "video_id": 1, + "video_name": "1.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被家人误会", + "narration": "她被家人逼到绝路，反击从这一刻开始。", + "OST": 0, + } + ] + }, + ensure_ascii=False, + ) + + with mock.patch.object(adapter, "_run_async_safely", return_value=matched) as call: + result = adapter.match_narration_copy_to_script( + short_name="测试短剧", + plot_analysis="女主被家人误会后反击。", + subtitle_content="# 视频 1: 1.mp4\n00:00:01,000 --> 00:00:04,000\n女主被误会。", + narration_copy="她被家人逼到绝路，反击从这一刻开始。", + temperature=0.7, + narration_language="简体中文（中国）", + drama_genre="家庭伦理", + original_sound_ratio=60, + ) + + self.assertEqual("success", result["status"]) + self.assertEqual(1, json.loads(result["narration_script"])["items"][0]["_id"]) + self.assertIn("家庭伦理", call.call_args.kwargs["prompt"]) + self.assertIn("60%", call.call_args.kwargs["prompt"]) + self.assertEqual("json", call.call_args.kwargs["response_format"]) + + def test_match_narration_copy_to_script_uses_streaming_when_callback_exists(self): + adapter = SubtitleAnalyzerAdapter( + api_key="sk-test", + model="test-model", + base_url="https://example.test/v1", + provider="openai", + ) + matched = json.dumps({"items": []}, ensure_ascii=False) + + with mock.patch.object(adapter, "_run_async_safely", return_value=matched) as call: + result = adapter.match_narration_copy_to_script( + short_name="测试短剧", + plot_analysis="女主被家人误会后反击。", + subtitle_content="# 视频 1: 1.mp4", + narration_copy="她被家人逼到绝路，反击从这一刻开始。", + stream_callback=lambda _event: None, + ) + + self.assertEqual("success", result["status"]) + self.assertIs(UnifiedLLMService.generate_text_stream, call.call_args.args[0]) + self.assertIn("on_chunk", call.call_args.kwargs) + + def test_generate_narration_script_plans_segments_before_copywriting(self): + adapter = SubtitleAnalyzerAdapter( + api_key="sk-test", + model="test-model", + base_url="https://example.test/v1", + provider="openai", + ) + responses = iter( + [ + json.dumps( + { + "segments": [ + { + "_id": 1, + "video_id": 1, + "video_name": "1.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "OST": 0, + "intent": "开场钩子", + } + ] + }, + ensure_ascii=False, + ), + json.dumps( + { + "items": [ + { + "_id": 1, + "video_id": 1, + "video_name": "1.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "她被所有人误会，真正的反击却刚刚开始。", + "OST": 0, + } + ] + }, + ensure_ascii=False, + ), + ] + ) + + with mock.patch.object(adapter, "_run_async_safely", side_effect=lambda *_args, **_kwargs: next(responses)) as call: + result = adapter.generate_narration_script( + short_name="测试短剧", + plot_analysis="女主被误会后反击。", + subtitle_content="# 视频 1: 1.mp4\n00:00:01,000 --> 00:00:04,000\n女主被误会。", + temperature=0.7, + narration_language="简体中文（中国）", + ) + + self.assertEqual("success", result["status"]) + self.assertEqual(2, call.call_count) + self.assertEqual(1, json.loads(result["narration_script"])["items"][0]["_id"]) + + def test_repair_narration_script_returns_repaired_json(self): + adapter = SubtitleAnalyzerAdapter( + api_key="sk-test", + model="test-model", + base_url="https://example.test/v1", + provider="openai", + ) + repaired = json.dumps({"items": []}, ensure_ascii=False) + + with mock.patch.object(adapter, "_run_async_safely", return_value=repaired): + result = adapter.repair_narration_script( + short_name="测试短剧", + plot_analysis="", + subtitle_content="# 视频 1: 1.mp4", + invalid_script="{bad}", + validation_errors="时间戳错误", + narration_language="简体中文（中国）", + ) + + self.assertEqual("success", result["status"]) + self.assertEqual(repaired, result["narration_script"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/app/services/llm/unified_service.py b/app/services/llm/unified_service.py index 63cc48f..071e8da 100644 --- a/app/services/llm/unified_service.py +++ b/app/services/llm/unified_service.py @@ -108,6 +108,37 @@ class UnifiedLLMService: except Exception as e: logger.error(f"文本生成失败: {str(e)}") raise LLMServiceError(f"文本生成失败: {str(e)}") + + @staticmethod + async def generate_text_stream(prompt: str, + system_prompt: Optional[str] = None, + provider: Optional[str] = None, + temperature: float = 1.0, + max_tokens: Optional[int] = None, + response_format: Optional[str] = None, + on_chunk=None, + **kwargs) -> str: + """ + 流式生成文本内容；不支持流式的 provider 会退化为一次性返回。 + """ + try: + text_provider = LLMServiceManager.get_text_provider(provider) + result = await text_provider.generate_text_stream( + prompt=prompt, + system_prompt=system_prompt, + temperature=temperature, + max_tokens=max_tokens, + response_format=response_format, + on_chunk=on_chunk, + **kwargs + ) + + logger.info(f"流式文本生成完成，生成内容长度: {len(result)} 字符") + return result + + except Exception as e: + logger.error(f"流式文本生成失败: {str(e)}") + raise LLMServiceError(f"流式文本生成失败: {str(e)}") @staticmethod async def generate_narration_script(prompt: str, diff --git a/app/services/prompts/short_drama_narration/__init__.py b/app/services/prompts/short_drama_narration/__init__.py index dfa0171..3cee4de 100644 --- a/app/services/prompts/short_drama_narration/__init__.py +++ b/app/services/prompts/short_drama_narration/__init__.py @@ -10,7 +10,11 @@ """ from .plot_analysis import PlotAnalysisPrompt +from .narration_copy import NarrationCopyPrompt +from .segment_planning import SegmentPlanningPrompt from .script_generation import ScriptGenerationPrompt +from .script_matching import ScriptMatchingPrompt +from .script_repair import ScriptRepairPrompt from ..manager import PromptManager @@ -20,14 +24,34 @@ def register_prompts(): # 注册剧情分析提示词 plot_analysis_prompt = PlotAnalysisPrompt() PromptManager.register_prompt(plot_analysis_prompt, is_default=True) + + # 注册可审核解说文案提示词 + narration_copy_prompt = NarrationCopyPrompt() + PromptManager.register_prompt(narration_copy_prompt, is_default=True) + + # 注册片段规划提示词 + segment_planning_prompt = SegmentPlanningPrompt() + PromptManager.register_prompt(segment_planning_prompt, is_default=True) # 注册解说脚本生成提示词 script_generation_prompt = ScriptGenerationPrompt() PromptManager.register_prompt(script_generation_prompt, is_default=True) + # 注册文案画面匹配提示词 + script_matching_prompt = ScriptMatchingPrompt() + PromptManager.register_prompt(script_matching_prompt, is_default=True) + + # 注册解说脚本修复提示词 + script_repair_prompt = ScriptRepairPrompt() + PromptManager.register_prompt(script_repair_prompt, is_default=True) + __all__ = [ "PlotAnalysisPrompt", + "NarrationCopyPrompt", + "SegmentPlanningPrompt", "ScriptGenerationPrompt", + "ScriptMatchingPrompt", + "ScriptRepairPrompt", "register_prompts" ] diff --git a/app/services/prompts/short_drama_narration/narration_copy.py b/app/services/prompts/short_drama_narration/narration_copy.py new file mode 100644 index 0000000..362e0fd --- /dev/null +++ b/app/services/prompts/short_drama_narration/narration_copy.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- + +""" +@Project: 短剧解说-解说文案 +@File : narration_copy.py +@Description: 生成可供用户审核修改的短剧解说正文 +""" + +from ..base import ParameterizedPrompt, PromptMetadata, ModelType, OutputFormat + + +class NarrationCopyPrompt(ParameterizedPrompt): + """短剧解说正文生成提示词""" + + def __init__(self): + metadata = PromptMetadata( + name="narration_copy", + category="short_drama_narration", + version="v1.0", + description="基于剧情理解和字幕生成可审核修改的短剧解说正文，不绑定时间戳", + model_type=ModelType.TEXT, + output_format=OutputFormat.TEXT, + tags=["短剧", "解说文案", "爆款开头", "叙事连续性", "用户审核"], + parameters=["drama_name", "drama_genre", "plot_analysis", "subtitle_content", "narration_language"], + ) + super().__init__(metadata, required_parameters=["drama_name", "plot_analysis", "subtitle_content"]) + + self._system_prompt = ( + "你是一位短剧解说文案创作者。你只输出可供用户审核修改的解说正文，" + "不要输出JSON、时间戳、编号、标题、解释或Markdown。" + ) + + def get_template(self) -> str: + return """# 短剧解说正文创作任务 + +## 目标 +为短剧《${drama_name}》创作一份可直接给用户审核修改的解说文案正文。此阶段不做画面匹配，不输出时间戳。 + +## 剧情理解材料 + +${plot_analysis} + + +## 原始字幕 + +${subtitle_content} + + +## 输出语言 + +${narration_language} + + +## 用户选择的短剧类型 + +${drama_genre} + + +## 类型写作规则 +必须按用户选择的短剧类型调整表达重点，不要自行改判类型： +- 霸总/甜宠：突出误会、身份差、暧昧拉扯、守护感和情绪反差。 +- 逆袭/复仇：突出羞辱、反击、打脸、身份揭露和爽点升级。 +- 家庭伦理：突出亲情撕扯、秘密、委屈、选择和道德冲突。 +- 古装/权谋：突出身份、局势、算计、立场和反转。 +- 悬疑/犯罪：突出线索、危机、动机和未揭开的疑问。 +- 都市情感：突出关系裂痕、现实压力、误会和情绪拉扯。 +- 年代/乡村：突出家庭处境、人情压力、生活困境和命运转折。 +- 自定义类型：严格服从用户填写的类型方向。 + +## 开头钩子公式 +开头必须使用“高能反转 + 情绪冲突 + 悬念钩子”： +1. 强身份或强处境：兵王、单亲妈妈、被赶出家门的女人、被全家看不起的人等。 +2. 致命反差：刚立功就被迫退役、刚回家就发现钱被输光、刚结婚就遇到孩子/婆婆阻挠。 +3. 后续悬念：真正的噩梦才开始、他要讨回的不是钱、这段关系真正难的不是相爱。 + +## 写作规则 +1. 必须使用 ${narration_language}。 +2. 严格基于剧情理解和字幕事实，不编造核心情节、身份、结局。 +3. 先写完整故事线，再写金句；不要只堆爆点。 +4. 每句话只表达一个信息点，适合后续按句匹配画面。 +5. 句子尽量短，单句优先 15-35 字；信息复杂时拆成多句。 +6. 每 2-3 句要有明确因果承接，让观众知道为什么从上一幕来到下一幕。 +7. 总长度控制在 300-650 字；短素材取下限，长素材取上限。 +8. 不要使用编号、项目符号、章节标题或括号说明。 + +## 输出要求 +只输出解说正文。不要输出 JSON、时间戳、代码块或任何解释。""" diff --git a/app/services/prompts/short_drama_narration/script_generation.py b/app/services/prompts/short_drama_narration/script_generation.py index 234fc98..955f0c1 100644 --- a/app/services/prompts/short_drama_narration/script_generation.py +++ b/app/services/prompts/short_drama_narration/script_generation.py @@ -19,234 +19,112 @@ class ScriptGenerationPrompt(ParameterizedPrompt): metadata = PromptMetadata( name="script_generation", category="short_drama_narration", - version="v2.0", - description="基于短剧解说创作核心要素，生成高质量解说脚本，包含黄金开场、爽点放大、个性吐槽等专业技巧", + version="v2.1", + description="基于已规划片段生成高质量短剧解说脚本，重点补足剧情承接、因果解释和观众理解路径", model_type=ModelType.TEXT, output_format=OutputFormat.JSON, tags=["短剧", "解说脚本", "文案生成", "原声片段", "黄金开场", "爽点放大", "个性吐槽", "悬念预埋"], - parameters=["drama_name", "plot_analysis", "subtitle_content"] + parameters=[ + "drama_name", + "drama_genre", + "plot_analysis", + "subtitle_content", + "segment_plan", + "narration_language", + ] ) - super().__init__(metadata, required_parameters=["drama_name", "plot_analysis"]) + super().__init__(metadata, required_parameters=["drama_name", "plot_analysis", "segment_plan"]) - self._system_prompt = "你是一位顶级的短剧解说up主，精通短视频创作的所有核心技巧。你必须严格按照JSON格式输出，绝不能包含任何其他文字、说明或代码块标记。" + self._system_prompt = ( + "你是一位短剧解说文案写手。你必须严格按照JSON格式输出，" + "只能补充picture和narration，不能改动上游片段规划中的_id、video_id、video_name、timestamp和OST。" + ) def get_template(self) -> str: - return """# 短剧解说脚本创作任务 + return """# 短剧解说脚本文案生成任务 ## 任务目标 -我是一位专业的短剧解说up主，需要为短剧《${drama_name}》创作一份高质量的解说脚本。目标是让观众在短时间内了解剧情精华，并产生强烈的继续观看欲望。 +为短剧《${drama_name}》生成最终可剪辑解说脚本。片段已经由上游规划完成，你只能补充 picture 和 narration，不能改变片段来源和时间戳。 -## 素材信息 +## 输入材料 ### 剧情概述 ${plot_analysis} +### 已规划片段（必须逐项照抄结构字段） + +${segment_plan} + + ### 原始字幕（含视频编号和精确时间戳） ${subtitle_content} +### 解说台词语言 + +${narration_language} + + +### 用户选择的短剧类型 + +${drama_genre} + + 字幕可能来自多个视频文件。每个字幕分段标题会以“视频 1: 文件名”“视频 2: 文件名”等形式标识来源。 生成脚本时必须把每个片段绑定到对应视频来源，时间戳表示该视频文件内部的局部时间，不是把多个视频拼接后的全局时间。 +所有 OST=0 的 narration 字段必须使用上方指定的解说台词语言输出；不要因为原始字幕是其他语言就切回字幕原语言。 +OST=1 的原声片段 narration 字段必须继续使用“播放原片+序号”格式，不要翻译这个固定标记。 -## 短剧解说创作核心要素 +## 绝对绑定规则 +1. 输出 items 数量、顺序和 _id 必须与 segment_plan 完全一致。 +2. 每个 item 的 _id、video_id、video_name、timestamp、OST 必须逐字复制 segment_plan，不得新增、删除、合并、拆分或改动。 +3. 你只能补充 picture 和 narration 两个字段。 +4. OST=1 的 narration 必须写成“播放原片+_id”，例如 _id 为 5 时写“播放原片5”。 +5. OST=0 的 narration 必须使用 ${narration_language}，并严格基于剧情和字幕，不虚构字幕外的具体事件。 -### 1. 黄金开场（3秒法则） -**开头3秒内必须制造强烈钩子，激发"想知道后续发展"的强烈好奇心** -- **悬念设置**：直接抛出最核心的冲突或疑问 - * 示例："身为一个名声恶臭的政客，他知道自己早晚会被暗杀" - * 技巧：直接定性角色身份和处境，制造紧张感 -- **冲突展示**：展现最激烈的对立关系 - * 示例："而这一天，就在他刚露头的时候..." - * 技巧：用时间节点强调关键时刻的到来 -- **情感共鸣**：触及观众内心的普遍情感 -- **反转预告**：暗示即将发生的惊人转折 - * 技巧：使用"没想到"、"原来"、"竟然"等词汇预告反转 +## 叙事连续性要求 +- 你必须把每个 OST=0 当成“观众理解剧情的桥”，不能只概括当前画面。 +- 每个 OST=0 narration 要尽量回答：上一段发生了什么、为什么会发展到这一段、这一段带来什么新矛盾。 +- 跨 video_id 或跨时间大跳跃时，OST=0 必须明确补出承接句，例如“可这段婚姻真正难的不是相爱，而是两个孩子和婆婆都还没接纳她”。 +- 原声片段前后的 OST=0 要解释原声的重要性，避免观众只看到对白片段合集。 +- 如果 segment_plan 中有 story_role、intent、transition 字段，必须利用它们组织 narration，但不要把这些字段输出到最终 JSON。 +- 结尾 OST=0 要留下后续阻力或悬念；如果结尾是 OST=1，则前一个 OST=0 必须提前点出这段原声会把矛盾推向哪里。 -### 2. 主线提炼（去繁就简） -**快节奏解说，速度超越原剧，专注核心主线** -- 舍弃次要情节和配角，只保留推动主线的关键人物 -- 突出核心矛盾冲突，每个片段都要推进主要故事线 -- 快速跳过铺垫，直击剧情要害 -- 确保每个解说片段都有明确的剧情推进作用 -- **转折技巧**：大量使用"而这时"、"就在这时"、"没多久"等时间转折词 +## 开头钩子要求 +- 第一段必须是 OST=0 解说钩子，不能直接播放原片。 +- 开头用“高能反转 + 情绪冲突 + 悬念钩子”：强身份/强处境 + 致命反差 + 后续悬念。 +- 写法示例方向：一个刚立功的兵王，下一秒却被迫脱下军装；他回家的第一天，家里的钱和尊严都被赌桌吞了。 +- 示例只用于理解公式，必须基于当前字幕事实原创，不要夸大到字幕没有的情节。 -### 3. 爽点放大（情绪引爆） -**精准识别剧中"爽点"并用富有感染力的语言放大** -- **主角逆袭**：突出弱者变强、反败为胜的瞬间 -- **反派被打脸**：强调恶人得到报应的痛快感 -- **智商在线**：赞美角色的机智和策略 - * 示例："豺狼已经提前数日跟踪这名清洁工，并在他身上放了窃听器" - * 技巧：展现角色的深谋远虑和专业能力 -- **情感爆发**：放大感人、愤怒、震撼等强烈情绪 -- 使用激昂语气和富有感染力的词汇调动观众情绪 +## 解说密度与画面节奏 +- OST=0 文案必须能被当前 timestamp 的画面承载，按“解说字数 / 5 = 所需视频秒数”估算。 +- 如果画面只有 6 秒，就不要写 80 字；应压缩到约 30 字，或依赖 segment_plan 选择更长画面。 +- 优先短句，单句只表达一个信息点；不要把人物介绍、前因、反转和悬念全塞进一个短画面。 +- 长信息要拆成多段，每段只承担一个叙事功能，让画面节奏跟上解说。 -### 4. 个性吐槽（增加趣味） -**以观众视角进行犀利点评，体现解说员独特人设** -- 避免单纯复述剧情，要有自己的观点和态度 -- **"上帝视角"分析技巧**： - * 揭示角色内心："他莫名地笑了一下" - * 分析动机："豺狼的这几步都是事先算好的" - * 预判后果："这又会有何代价呢" -- 适当吐槽剧情的套路或角色的愚蠢行为 -- 用幽默、犀利的语言增加观看趣味 -- 站在观众立场，说出观众想说的话 -- **心理活动描述**：深入角色内心，增强代入感 +## 用户选择类型文案规则 +短剧类型由用户手动选择为 ${drama_genre}，不得自行改判。必须按对应方向写： +- 霸总/甜宠：突出误会、身份差、暧昧拉扯、守护感和情绪反差。 +- 逆袭/复仇：突出羞辱、反击、打脸、身份揭露和爽点升级。 +- 家庭伦理：突出亲情撕扯、秘密、委屈、选择和道德冲突。 +- 古装/权谋：突出身份、局势、算计、立场和反转。 +- 悬疑/犯罪：突出线索、危机、动机和未揭开的疑问。 +- 都市情感：突出关系裂痕、现实压力、误会和情绪拉扯。 +- 年代/乡村：突出家庭处境、人情压力、生活困境和命运转折。 +- 自定义类型：严格服从用户填写的类型方向。 -### 5. 悬念预埋（引导互动） -**在关键节点和结尾处"卖关子"，激发互动欲望** -- 在剧情高潮前停止，留下"接下来会发生什么"的疑问 -- **悬念设置技巧**： - * 问题抛出："那么，UDC究竟是谁呢？" - * 反转预告："而从这句话开始，所有的专业、体面和虚伪的平静都将分崩瓦解" - * 时间悬念："几分钟后..."、"不久之后..." -- 提出引导性问题："你们觉得他会怎么做？" -- 预告后续精彩："更劲爆的还在后面" -- 为后续内容预热，激发评论、点赞、关注 +## 文案质量要求 +- 开场片段要有强钩子，直接点出冲突、悬念或情绪爆点。 +- 每段解说优先 25-90 字，具体长度必须服从画面时长；短画面宁可少说，不要密集灌信息。 +- 可以使用“没想到”“可下一秒”“而这时”“真正的问题来了”等短剧转折语，但不要堆砌。 +- picture 要描述画面和人物状态，便于后期识别素材。 +- 少用孤立信息句，多用承接句；不要让观众感觉剧情突然跳场。 +- 不要解释规则，不要输出 Markdown，不要输出代码块。 -### 6. 卡点配合（视听协调） -**考虑文案与画面、音乐的完美结合** -- 在情感高潮处预设BGM卡点 -- 解说节奏要配合画面节奏 -- 重要台词处保留原声，解说适时停顿 -- 追求文案+画面+音乐的协同效应 - -## 专业解说语言技巧 - -### 1. 氛围营造技巧 -**通过环境和细节描述增强画面感和代入感** -- **环境描述**："在这个距离，枪声都无法传到那边" -- **细节刻画**："他的床头有酒，身边的纸碟堆满烟头" -- **氛围渲染**："黑暗树林里有一间仓房" -- **情绪描述**："孤独又无助的豺狼，竟在这时露出了反常的一面" - -### 2. 情感词汇运用 -**使用富有感染力的词汇调动观众情绪** -- **紧张感**："名声恶臭"、"早晚会被暗杀"、"动用军警资源" -- **神秘感**："尘封的传奇"、"高度机密"、"暗藏玄机" -- **震撼感**："空前绝后的一枪"、"天衣无缝"、"神不知鬼不觉" -- **悲伤感**："目光非常悲伤"、"注定永远无法哀悼" - -### 3. 节奏控制技巧 -**通过语言节奏控制观众注意力** -- **快节奏推进**：使用短句，密集信息 -- **慢节奏渲染**：使用长句，详细描述 -- **停顿技巧**：在关键信息前适当停顿 -- **重复强调**：重要信息适当重复 - -## 严格技术要求 - -### 时间戳管理（绝对不能违反） -- **时间戳绝对不能重叠**，确保剪辑后无重复画面 -- **同一个 video_id 内的时间段必须连续且不交叉**，严格按该视频内时间顺序排列 -- **跨视频可以切换 video_id**，但每个时间戳都必须来自对应视频字幕分段 -- **每个时间戳都必须在对应视频的原始字幕中找到对应范围** -- 可以拆分原时间片段，但必须保持时间连续性 -- 时间戳的格式必须与原始字幕中的格式完全一致 - -### 多视频来源规范（多集/多文件必须遵守） -- **video_id**：必须填写，取字幕分段标题里的视频编号，例如“视频 3”就填 3 -- **video_name**：必须填写对应的视频文件名，例如“3_20260607002212.mp4” -- **timestamp**：只填写对应 video_id 内部的时间范围，不要换算成多个视频拼接后的累计时间 -- 如果剧情跨多个视频推进，脚本可以按故事顺序在不同 video_id 之间切换，但不得把视频 2 的时间戳写到 video_id=1 - -### 时长控制（1/3原则） -- **解说视频总长度 = 原视频长度的 1/3** -- 精确控制节奏和密度，既不能过短也不能过长 -- 合理分配解说和原声的时间比例 - -### 剧情连贯性 -- **保持故事逻辑完整**，确保情节发展自然流畅 -- **严格按照时间顺序**，禁止跳跃式叙述 -- **符合因果逻辑**：先发生A，再发生B，A导致B - -## 原声片段使用规范 - -### 原声片段格式要求 -原声片段必须严格按照以下JSON格式： -```json -{ - "_id": 序号, - "video_id": 视频编号, - "video_name": "视频文件名", - "timestamp": "开始时间-结束时间", - "picture": "画面内容描述", - "narration": "播放原片+序号", - "OST": 1 -} -``` - -### 原声片段插入策略 - -#### 1. 关键情绪爆发点 -**在角色强烈情绪表达时必须保留原声，增强观众代入感** -- **愤怒爆发**：角色愤怒咆哮、情绪失控的瞬间 - * 参考："Come on, you bastard. Reaching."（愤怒对峙） -- **感动落泪**：角色感动哭泣、情感宣泄的时刻 -- **震惊反应**：角色震惊、不敢置信的表情和台词 - * 参考："Are you sure about that?"（质疑震惊） -- **绝望崩溃**：角色绝望、崩溃的情感表达 - * 参考："Charles you're scaring me, what's wrong"（恐惧绝望） -- **狂欢庆祝**：角色兴奋、狂欢的情绪高潮 - -#### 2. 重要对白时刻 -**保留推动剧情发展的关键台词和对话** -- **身份揭露**：揭示角色真实身份的重要台词 -- **真相大白**：揭晓谜底、真相的关键对话 -- **情感告白**：爱情告白、情感表达的重要台词 - * 参考："i'm really not good"（情感表达） -- **威胁警告**：反派威胁、警告的重要对白 - * 参考："You do not want to make enemies of these people"（威胁警告） -- **决定宣布**：角色做出重要决定的宣告 - -#### 3. 爽点瞬间 -**在"爽点"时刻保留原声增强痛快感** -- **主角逆袭**：弱者反击、逆转局面的台词 -- **反派被打脸**：恶人得到报应、被揭穿的瞬间 -- **智商碾压**：主角展现智慧、碾压对手的台词 - * 参考："That is a fucking work of art guys"（技能展示） -- **正义伸张**：正义得到伸张、恶有恶报的时刻 -- **实力展现**：主角展现真实实力、震撼全场 - -#### 4. 悬念节点 -**在制造悬念或揭晓答案的关键时刻保留原声** -- **悬念制造**：制造悬念、留下疑问的台词 -- **答案揭晓**：揭晓答案、解开谜团的对话 -- **转折预告**：暗示即将发生转折的重要台词 -- **危机降临**：危机来临、紧张时刻的对白 - -#### 5. 经典台词时刻 -**保留具有强烈感染力和记忆点的经典台词** -- **哲理感悟**：角色的人生感悟和哲理思考 -- **幽默调侃**：轻松幽默的对话增加趣味性 -- **专业术语**：体现角色专业性的术语和对话 - * 参考："The scanner will pick up the metal components"（专业解释） -- **情感共鸣**：能引起观众共鸣的经典表达 - -### 原声片段技术规范 - -#### 格式规范 -- **OST字段**：设置为1表示保留原声（解说片段设置为0） -- **narration格式**：严格使用"播放原片+序号"（如"播放原片26"） -- **picture字段**：详细描述画面内容，便于后期剪辑参考 -- **时间戳精度**：必须与字幕中的重要对白时间精确匹配 - -#### 比例控制 -- **原声与解说比例**：7:3（原声70%，解说30%） -- **分布均匀**：原声片段要在整个视频中均匀分布 -- **长度适中**：单个原声片段时长控制在3-8秒 -- **衔接自然**：原声片段与解说片段之间衔接自然流畅 - -#### 选择原则 -- **情感优先**：优先选择情感强烈的台词和对话 -- **剧情关键**：必须是推动剧情发展的重要内容 -- **观众共鸣**：选择能引起观众共鸣的经典台词 -- **视听效果**：考虑台词的声音效果和表演张力 -- **代入感强**：选择能让观众产生强烈代入感的对话 - -## 输出格式要求 +## 输出格式 请严格按照以下JSON格式输出，绝不添加任何其他文字、说明或代码块标记： @@ -282,45 +160,4 @@ ${subtitle_content} ] } -## 质量标准 - -### 解说文案要求： -- **字数控制**：每段解说文案80-150字 -- **语言风格**：生动有趣，富有感染力，符合短视频观众喜好 - * 参考风格："身为一个名声恶臭的政客，他知道自己早晚会被暗杀" - * 直接定性，制造紧张感和代入感 -- **情感调动**：能够有效调动观众情绪，产生代入感 - * 使用"而这时"、"没想到"、"原来"等转折词增强戏剧性 -- **节奏把控**：快节奏但不失条理，紧凑但不混乱 - * 短句推进剧情，长句渲染氛围 - -### 技术规范： -- **解说与原片比例**：3:7（解说30%，原片70%） -- **原声片段标识**：OST=1表示原声，OST=0表示解说 -- **原声格式规范**：narration字段必须使用"播放原片+序号"格式 -- **关键情绪点**：必须保留原片原声，增强观众代入感 -- **视频来源**：每个片段必须包含 video_id 和 video_name，用于定位多个上传视频中的源文件 -- **时间戳精度**：精确到毫秒级别，确保与字幕完美匹配 -- **逻辑连贯性**：严格遵循剧情发展顺序 - -### 创作原则： -1. **只输出JSON内容**，不要任何说明性文字 -2. **严格基于提供的剧情和字幕**，不虚构内容 -3. **突出核心冲突**，舍弃无关细节 -4. **强化观众体验**，始终考虑观看感受 -5. **保持专业水准**，体现解说up主的专业素养 -6. **融入经典解说技巧**： - - 大量使用"上帝视角"分析 - - 适时插入心理活动描述 - - 运用悬念设置和反转技巧 - - 保持强烈的画面感和代入感 - -### 参考解说风格示例： -- **开场悬念**："身为一个名声恶臭的政客，他知道自己早晚会被暗杀" -- **转折技巧**："而这一天，就在他刚露头的时候..." -- **上帝视角**："豺狼已经提前数日跟踪这名清洁工" -- **情感渲染**："孤独又无助的豺狼，竟在这时露出了反常的一面" -- **悬念设置**："那么，UDC究竟是谁呢？" -- **反转预告**："而从这句话开始，所有的专业、体面和虚伪的平静都将分崩瓦解" - 现在请基于以上要求，为短剧《${drama_name}》创作解说脚本：""" diff --git a/app/services/prompts/short_drama_narration/script_matching.py b/app/services/prompts/short_drama_narration/script_matching.py new file mode 100644 index 0000000..61052e5 --- /dev/null +++ b/app/services/prompts/short_drama_narration/script_matching.py @@ -0,0 +1,131 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- + +""" +@Project: 短剧解说-文案画面匹配 +@File : script_matching.py +@Description: 将用户审核后的解说文案匹配到字幕时间戳并生成最终剪辑脚本 +""" + +from ..base import ParameterizedPrompt, PromptMetadata, ModelType, OutputFormat + + +class ScriptMatchingPrompt(ParameterizedPrompt): + """短剧解说文案画面匹配提示词""" + + def __init__(self): + metadata = PromptMetadata( + name="script_matching", + category="short_drama_narration", + version="v1.0", + description="将审核后的解说文案按叙事节奏拆分，并匹配到字幕时间戳生成最终剪辑JSON", + model_type=ModelType.TEXT, + output_format=OutputFormat.JSON, + tags=["短剧", "画面匹配", "剪辑脚本", "时间戳", "用户文案"], + parameters=[ + "drama_name", + "drama_genre", + "plot_analysis", + "subtitle_content", + "narration_copy", + "narration_language", + "original_sound_ratio", + ], + ) + super().__init__( + metadata, + required_parameters=["drama_name", "subtitle_content", "narration_copy"], + ) + + self._system_prompt = ( + "你是一位懂叙事节奏的短剧剪辑师。你必须严格输出JSON，" + "核心任务是把用户审核后的解说文案逐句匹配到最合适的原视频字幕时间戳。" + ) + + def get_template(self) -> str: + return """# 短剧解说文案画面匹配任务 + +## 目标 +用户已经审核并修改了解说文案。请根据这份文案和原始字幕，生成最终可剪辑 JSON 脚本。 + +## 剧名 +${drama_name} + +## 剧情理解材料 + +${plot_analysis} + + +## 用户审核后的解说文案 + +${narration_copy} + + +## 原始字幕（含视频编号和局部时间戳） + +${subtitle_content} + + +## 输出语言 + +${narration_language} + + +## 用户选择的短剧类型 + +${drama_genre} + + +## 用户选择的原片占比 + +${original_sound_ratio}% + + +## 匹配流程 +1. 先按句号、问号、感叹号、省略号切分解说文案，得到候选解说句。 +2. 逗号只在明显分割两个动作、场景、观点或描述对象时切分；不要切出没有独立意义的碎片。 +3. 不要求每个候选句都单独输出为 OST=0；可以合并、压缩相邻候选句作为剧情桥段，但不能改变用户文案的核心意思。 +4. 为每个解说片段寻找最匹配的原始字幕画面，优先选择能表达该句核心含义的画面。 +5. 使用公式估算所需画面时长：所需秒数 = 解说字数 / 5。匹配画面时长尽量接近，误差优先控制在 ±0.5 秒。 +6. 如果一句解说太长，必须拆成多个 OST=0 片段，分别匹配不同或连续画面。 +7. timestamp 必须使用对应 video_id 内部局部时间戳，不得换算为多个视频拼接后的累计时间。 +8. 同一 video_id 内时间段不得交叉或重叠。 +9. 第一段必须是 OST=0 解说钩子，不能直接播放原片。 +10. OST=1 原声片段的总时长占比要尽量接近用户选择的 ${original_sound_ratio}%。这里按最终 items 的 timestamp 总时长估算，不按片段数量估算。 +11. 不要自行判断或改写短剧类型；画面匹配和 picture 描述要服务用户选择的 ${drama_genre} 叙事重点。 + +## 原片占比规则 +- ${original_sound_ratio}% = 0% 时，不要输出 OST=1，全部使用解说承接。 +- ${original_sound_ratio}% 在 10%-30% 时，只保留关键对白、反转、情绪爆发或爽点原声。 +- ${original_sound_ratio}% 在 40%-60% 时，解说负责串联因果，原片负责承载关键场面和对白。 +- ${original_sound_ratio}% 在 70%-90% 时，以原片对白和表演为主，解说只做开场钩子、转场桥和必要补充。 +- 如果原片占比与“第一段必须 OST=0”冲突，优先保证第一段是 OST=0，然后在后续片段提高 OST=1 时长占比。 +- 选择高原片占比时，可以把用户文案合并成更少的 OST=0 桥段，不要为了逐句使用文案而压低原片占比。 + +## 字段规则 +- _id：从 1 开始连续递增。 +- video_id：来自字幕分段标题，例如“视频 2”就填 2。 +- video_name：对应视频文件名，必须从字幕分段标题提取。 +- timestamp：格式为 "HH:MM:SS,mmm-HH:MM:SS,mmm"。 +- picture：描述匹配画面中人物、动作、情绪和场景。 +- narration：OST=0 时填写用户文案片段；OST=1 时填写“播放原片+_id”。 +- OST：解说片段填 0，原声片段填 1。 + +## 输出格式 +只输出严格 JSON： + +{ + "items": [ + { + "_id": 1, + "video_id": 1, + "video_name": "1.mp4", + "timestamp": "00:00:01,000-00:00:06,000", + "picture": "主角站在门口，震惊地看着屋内混乱的场面", + "narration": "一个刚立功的兵王，回家的第一天就发现家里四百万被亲爹输光。", + "OST": 0 + } + ] +} + +现在请基于用户审核后的解说文案生成最终剪辑脚本。""" diff --git a/app/services/prompts/short_drama_narration/script_repair.py b/app/services/prompts/short_drama_narration/script_repair.py new file mode 100644 index 0000000..0784c6c --- /dev/null +++ b/app/services/prompts/short_drama_narration/script_repair.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- + +""" +@Project: 短剧解说-脚本修复 +@File : script_repair.py +@Description: 短剧解说脚本校验失败后的JSON修复提示词 +""" + +from ..base import ParameterizedPrompt, PromptMetadata, ModelType, OutputFormat + + +class ScriptRepairPrompt(ParameterizedPrompt): + """短剧解说脚本修复提示词""" + + def __init__(self): + metadata = PromptMetadata( + name="script_repair", + category="short_drama_narration", + version="v1.0", + description="根据确定性校验错误修复短剧解说脚本JSON，优先修正时间戳、视频来源和格式问题", + model_type=ModelType.TEXT, + output_format=OutputFormat.JSON, + tags=["短剧", "解说脚本", "JSON修复", "时间戳校验", "多视频"], + parameters=[ + "drama_name", + "drama_genre", + "plot_analysis", + "subtitle_content", + "invalid_script", + "validation_errors", + "narration_language", + ], + ) + super().__init__( + metadata, + required_parameters=["drama_name", "subtitle_content", "invalid_script", "validation_errors"], + ) + + self._system_prompt = ( + "你是一位短剧解说脚本JSON修复器。你只能根据校验错误修复JSON，" + "必须输出严格JSON，不能输出解释、Markdown或代码块。" + ) + + def get_template(self) -> str: + return """# 短剧解说脚本修复任务 + +## 修复目标 +下面的短剧《${drama_name}》解说脚本未通过剪辑校验。请只根据校验错误和字幕内容修复它，输出一个完整可剪辑的 JSON。 + +## 剧情理解材料 + +${plot_analysis} + + +## 校验错误 + +${validation_errors} + + +## 当前无效脚本 + +${invalid_script} + + +## 可用字幕窗口 + +${subtitle_content} + + +## 解说台词目标语言 + +${narration_language} + + +## 用户选择的短剧类型 + +${drama_genre} + + +## 修复规则 +1. 只输出 JSON，不要任何解释、标题、Markdown 或代码块。 +2. 输出根对象必须是 {"items": [...]}。 +3. 每个 item 必须包含 _id、video_id、video_name、timestamp、picture、narration、OST。 +4. video_id、video_name 和 timestamp 必须来自对应字幕窗口；不得把不同视频的同名时间戳混用。 +5. 同一 video_id 内片段不得交叉或重叠。 +6. OST=1 的 narration 必须是“播放原片+序号”；OST=0 的 narration 必须使用 ${narration_language}。 +7. 禁止连续 3 个或更多 OST=1；必须插入或改写 OST=0 解说片段承接剧情。 +8. 跨 video_id 切换前后不能都是 OST=1；必须至少有一个 OST=0 片段解释场景和剧情为什么切换。 +9. OST=0 narration 要补足因果承接，不要只概括当前画面。 +10. 第一段必须是 OST=0 解说钩子，按“高能反转 + 情绪冲突 + 悬念钩子”写，不要直接播放原片。 +11. OST=0 文案必须匹配画面时长，按“解说字数 / 5 = 所需视频秒数”估算；过密时要缩短文案、延长时间戳或拆成多个片段。 +12. 不要自行改判短剧类型；如需改写 narration，必须按用户选择的 ${drama_genre} 保持表达重点。 +13. 尽量保留原脚本中没有错误的片段；无法修复的片段可以删除，但剩余片段必须重新按 1 开始编号。 + +请输出修复后的完整 JSON。""" diff --git a/app/services/prompts/short_drama_narration/segment_planning.py b/app/services/prompts/short_drama_narration/segment_planning.py new file mode 100644 index 0000000..da574bf --- /dev/null +++ b/app/services/prompts/short_drama_narration/segment_planning.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- + +""" +@Project: 短剧解说-片段规划 +@File : segment_planning.py +@Description: 短剧解说脚本片段规划提示词 +""" + +from ..base import ParameterizedPrompt, PromptMetadata, ModelType, OutputFormat + + +class SegmentPlanningPrompt(ParameterizedPrompt): + """短剧解说片段规划提示词""" + + def __init__(self): + metadata = PromptMetadata( + name="segment_planning", + category="short_drama_narration", + version="v1.1", + description="基于剧情理解和原始字幕规划可剪辑片段，优先保证叙事连续性、跨视频承接和原声解说节奏", + model_type=ModelType.TEXT, + output_format=OutputFormat.JSON, + tags=["短剧", "解说脚本", "片段规划", "时间戳", "多视频", "原声"], + parameters=["drama_name", "drama_genre", "plot_analysis", "subtitle_content", "narration_language"], + ) + super().__init__(metadata, required_parameters=["drama_name", "plot_analysis", "subtitle_content"]) + + self._system_prompt = ( + "你是一位短剧解说剪辑规划师。你的任务是从字幕中选择可剪辑片段，" + "必须严格输出JSON，不能写解说文案，不能输出Markdown或额外说明。" + ) + + def get_template(self) -> str: + return """# 短剧解说片段规划任务 + +## 目标 +为短剧《${drama_name}》规划一组可直接剪辑的视频片段。你只负责选片段和标注用途，不写最终解说台词。 + +## 剧情理解材料 + +${plot_analysis} + + +## 原始字幕（含视频编号和局部时间戳） + +${subtitle_content} + + +## 解说台词目标语言 + +${narration_language} + + +## 用户选择的短剧类型 + +${drama_genre} + + +## 叙事规划目标 +你不是在挑精彩片段合集，而是在规划一条观众能顺着看懂的短剧解说故事线。必须先想清楚“人物困境 -> 冲突触发 -> 关系变化 -> 新阻力 -> 悬念”的因果链，再选片段。 + +## 爆款开头钩子规则 +第一段必须是 OST=0 解说开场，不要直接播放原片。开头参考“高能反转 + 情绪冲突 + 悬念钩子”的公式： +- 先给人物一个强身份或强处境：兵王、单亲妈妈、被赶出家门的女人、被全家看不起的赘婿。 +- 再给一个反差冲突：刚立功就被迫退役、刚回家就发现钱被输光、刚结婚就遇到孩子/婆婆阻挠。 +- 最后抛出悬念：真正的噩梦才开始、他要讨回的不是钱、这场婚姻真正难的不是相爱。 +- 不要照抄示例，要基于字幕事实改写成当前剧情自己的钩子。 + +## 规划规则 +1. 只能使用原始字幕中真实存在的视频编号、视频文件名和时间范围。 +2. timestamp 必须是对应 video_id 内部的局部时间戳，禁止换算成多个视频拼接后的累计时间。 +3. 同一个 video_id 内的片段不得交叉或重叠；尽量按故事顺序排列。 +4. 每个片段必须推动主线、制造情绪点、承接原声或保留关键对白。 +5. OST=1 表示保留原声，适合关键对白、情绪爆发、身份揭露、反转和爽点；OST=0 表示后续需要配解说。 +6. 原声片段单段优先控制在 3-8 秒；解说片段可以更长，但必须能从字幕范围中定位。 +7. 短剧类型由用户手动选择为 ${drama_genre}，不得自行改判；选片段时优先服务该类型的主要看点。 +8. 禁止连续 3 个或更多 OST=1；每 1-2 个原声片段后必须安排 OST=0 解说片段承接剧情。 +9. 跨 video_id 切换前后必须至少有一个 OST=0 片段作为剧情桥段，解释为什么从上一场转到下一场。 +10. 每个 OST=0 片段必须承担明确叙事功能：开场钩子、人物介绍、因果过渡、冲突升级、关系转折、阻力解释、结尾悬念。 +11. 不要跳过关键因果：例如从求婚直接跳到孩子/婆婆阻挠，中间必须用 OST=0 解释“婚姻真正的难题变成家庭接纳”。 +12. 结尾优先选择能留下后续阻力或新矛盾的片段，不要只停在原声对白堆叠上。 +13. 解说画面必须给足时长：按“解说字数 / 5 = 所需视频秒数”预估，短画面不要承载长解说。 +14. OST=0 片段如果需要讲清多层信息，应选择更长的连续画面，或拆成多个 OST=0 片段分别承接。 + +## 输出格式 +只输出严格 JSON： + +{ + "segments": [ + { + "_id": 1, + "video_id": 1, + "video_name": "1.mp4", + "timestamp": "00:00:01,000-00:00:05,500", + "OST": 0, + "story_role": "开场钩子", + "intent": "女主被羞辱，制造逆袭期待", + "transition": "从灾后恢复现场切入女主处境，引出她为什么敢和领导硬刚" + } + ] +} + +现在请规划短剧《${drama_name}》的解说片段。""" diff --git a/app/services/short_drama_narration_validation.py b/app/services/short_drama_narration_validation.py new file mode 100644 index 0000000..eb899cc --- /dev/null +++ b/app/services/short_drama_narration_validation.py @@ -0,0 +1,435 @@ +#!/usr/bin/env python +# -*- coding: UTF-8 -*- + +"""Validation helpers for short drama narration scripts.""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + + +TIMESTAMP_RE = re.compile(r"^\d{2}:\d{2}:\d{2},\d{3}$") +SCRIPT_RANGE_RE = re.compile( + r"^(?P\d{2}:\d{2}:\d{2}[,.]\d{3})-(?P\d{2}:\d{2}:\d{2}[,.]\d{3})$" +) +SRT_RANGE_RE = re.compile( + r"(?P\d{2}:\d{2}:\d{2}[,.]\d{3})\s*-->\s*" + r"(?P\d{2}:\d{2}:\d{2}[,.]\d{3})" +) +VIDEO_HEADER_RE = re.compile(r"^#\s*视频\s*(?P\d+)(?:\s*[:：]\s*(?P.+?))?\s*$") +NARRATION_CHARS_PER_SECOND = 5.0 +NARRATION_DURATION_TOLERANCE_SECONDS = 0.5 + + +@dataclass(frozen=True) +class SubtitleCue: + video_id: int + video_name: str + start_ms: int + end_ms: int + text: str + timestamp: str + + +@dataclass(frozen=True) +class ScriptValidationResult: + valid: bool + errors: List[str] + items: List[Dict[str, Any]] + + +class NarrationScriptValidationError(ValueError): + """Raised when a narration script cannot be made safe for clipping.""" + + +def timestamp_to_ms(timestamp: str) -> int: + value = str(timestamp or "").strip().replace(".", ",") + if not TIMESTAMP_RE.match(value): + raise ValueError(f"时间戳格式错误: {timestamp}") + + hh, mm, rest = value.split(":") + ss, ms = rest.split(",") + return ((int(hh) * 60 + int(mm)) * 60 + int(ss)) * 1000 + int(ms) + + +def ms_to_timestamp(ms: int) -> str: + if ms < 0: + raise ValueError("毫秒时间不能为负数") + + hours, remainder = divmod(ms, 60 * 60 * 1000) + minutes, remainder = divmod(remainder, 60 * 1000) + seconds, millis = divmod(remainder, 1000) + return f"{hours:02d}:{minutes:02d}:{seconds:02d},{millis:03d}" + + +def parse_script_timestamp_range(timestamp_range: str) -> Tuple[int, int, str]: + value = str(timestamp_range or "").strip().replace(".", ",") + match = SCRIPT_RANGE_RE.match(value) + if not match: + raise ValueError("时间戳格式应为 'HH:MM:SS,mmm-HH:MM:SS,mmm'") + + start = timestamp_to_ms(match.group("start")) + end = timestamp_to_ms(match.group("end")) + return start, end, f"{ms_to_timestamp(start)}-{ms_to_timestamp(end)}" + + +def _normalize_paths(paths: Optional[Iterable[str]]) -> List[str]: + if isinstance(paths, str): + paths = [paths] + if not paths: + return [] + + normalized = [] + for path in paths: + if not isinstance(path, str): + continue + path = path.strip() + if path: + normalized.append(path) + return normalized + + +def _default_video_name(video_id: int, video_paths: Sequence[str]) -> str: + if 1 <= video_id <= len(video_paths): + return os.path.basename(video_paths[video_id - 1]) + return "" + + +def _split_subtitle_sections( + subtitle_content: str, + video_paths: Sequence[str], +) -> List[Tuple[int, str, str]]: + sections: List[Tuple[int, str, str]] = [] + current_video_id = 1 + current_video_name = _default_video_name(1, video_paths) + current_lines: List[str] = [] + saw_header = False + + for line in str(subtitle_content or "").splitlines(): + header_match = VIDEO_HEADER_RE.match(line.strip()) + if header_match: + if current_lines or saw_header: + sections.append((current_video_id, current_video_name, "\n".join(current_lines))) + current_lines = [] + + saw_header = True + current_video_id = int(header_match.group("video_id")) + header_video_name = str(header_match.group("video_name") or "").strip() + current_video_name = header_video_name or _default_video_name(current_video_id, video_paths) + continue + + current_lines.append(line) + + if current_lines or not sections: + sections.append((current_video_id, current_video_name, "\n".join(current_lines))) + + return sections + + +def _extract_cues_from_section(video_id: int, video_name: str, section_text: str) -> List[SubtitleCue]: + lines = str(section_text or "").splitlines() + cues: List[SubtitleCue] = [] + index = 0 + + while index < len(lines): + match = SRT_RANGE_RE.search(lines[index]) + if not match: + index += 1 + continue + + start_ms = timestamp_to_ms(match.group("start")) + end_ms = timestamp_to_ms(match.group("end")) + timestamp = f"{ms_to_timestamp(start_ms)}-{ms_to_timestamp(end_ms)}" + index += 1 + + text_lines: List[str] = [] + while index < len(lines) and lines[index].strip(): + text_lines.append(lines[index].strip()) + index += 1 + + cues.append( + SubtitleCue( + video_id=video_id, + video_name=video_name, + start_ms=start_ms, + end_ms=end_ms, + text=" ".join(text_lines).strip(), + timestamp=timestamp, + ) + ) + index += 1 + + return cues + + +def build_subtitle_index(subtitle_content: str, video_paths: Optional[Iterable[str]] = None) -> List[SubtitleCue]: + """Build a per-video subtitle index from combined SRT text.""" + normalized_video_paths = _normalize_paths(video_paths) + cues: List[SubtitleCue] = [] + + for video_id, video_name, section_text in _split_subtitle_sections(subtitle_content, normalized_video_paths): + cues.extend(_extract_cues_from_section(video_id, video_name, section_text)) + + return cues + + +def _coerce_positive_int(value: Any) -> Optional[int]: + try: + number = int(value) + except (TypeError, ValueError): + return None + return number if number > 0 else None + + +def _video_id_by_name(video_name: Any, video_paths: Sequence[str]) -> Optional[int]: + normalized_name = os.path.basename(str(video_name or "").strip()) + if not normalized_name: + return None + + for index, path in enumerate(video_paths, start=1): + if os.path.basename(path) == normalized_name: + return index + return None + + +def normalize_script_video_sources( + items: Sequence[Dict[str, Any]], + video_paths: Optional[Iterable[str]] = None, +) -> List[Dict[str, Any]]: + """Normalize video_name from a valid source without inventing video_id.""" + normalized_video_paths = _normalize_paths(video_paths) + normalized_items: List[Dict[str, Any]] = [] + + for raw_item in items: + item = dict(raw_item) + video_id = _coerce_positive_int(item.get("video_id") or item.get("video_index")) + matched_video_id = _video_id_by_name(item.get("video_name") or item.get("source_video"), normalized_video_paths) + if matched_video_id is not None: + video_id = matched_video_id + + if video_id is not None: + item["video_id"] = video_id + if 1 <= video_id <= len(normalized_video_paths): + item["video_name"] = os.path.basename(normalized_video_paths[video_id - 1]) + + normalized_items.append(item) + + return normalized_items + + +def _cues_for_video(cues: Sequence[SubtitleCue], video_id: int) -> List[SubtitleCue]: + return [cue for cue in cues if cue.video_id == video_id] + + +def _range_overlaps_subtitle(cues: Sequence[SubtitleCue], start_ms: int, end_ms: int) -> bool: + return any(start_ms < cue.end_ms and end_ms > cue.start_ms for cue in cues) + + +def _range_within_subtitle_bounds(cues: Sequence[SubtitleCue], start_ms: int, end_ms: int) -> bool: + if not cues: + return False + return min(cue.start_ms for cue in cues) <= start_ms and end_ms <= max(cue.end_ms for cue in cues) + + +def _item_ost(item: Dict[str, Any]) -> Optional[int]: + try: + return int(item.get("OST")) + except (TypeError, ValueError): + return None + + +def _item_video_id(item: Dict[str, Any]) -> Optional[int]: + return _coerce_positive_int(item.get("video_id")) + + +def count_narration_chars(text: str) -> int: + """Count visible narration characters for rough TTS/video-duration matching.""" + return len(re.sub(r"\s+", "", str(text or ""))) + + +def max_narration_chars_for_duration(start_ms: int, end_ms: int) -> int: + duration_seconds = max(0.0, (end_ms - start_ms) / 1000) + return max(8, int((duration_seconds + NARRATION_DURATION_TOLERANCE_SECONDS) * NARRATION_CHARS_PER_SECOND)) + + +def _validate_story_continuity(items: Sequence[Dict[str, Any]]) -> List[str]: + """Validate structural continuity rules that affect viewer comprehension.""" + errors: List[str] = [] + consecutive_ost = 0 + previous_item: Optional[Dict[str, Any]] = None + + for index, item in enumerate(items): + if not isinstance(item, dict): + consecutive_ost = 0 + previous_item = None + continue + + item_id = item.get("_id", index + 1) + ost = _item_ost(item) + if index == 0 and ost != 0: + errors.append(f"片段 {item_id} 必须是 OST=0 解说开场钩子，不能直接播放原片") + + if ost == 1: + consecutive_ost += 1 + if consecutive_ost > 2: + errors.append(f"片段 {item_id} 连续原声过多，必须插入 OST=0 解说承接剧情") + else: + consecutive_ost = 0 + + if previous_item is not None: + previous_video_id = _item_video_id(previous_item) + current_video_id = _item_video_id(item) + if ( + previous_video_id is not None + and current_video_id is not None + and previous_video_id != current_video_id + and _item_ost(previous_item) == 1 + and ost == 1 + ): + errors.append( + f"片段 {previous_item.get('_id')} 到片段 {item_id} 跨视频切换缺少 OST=0 解说桥段" + ) + + previous_item = item + + return errors + + +def validate_narration_script_items( + items: Any, + subtitle_index: Sequence[SubtitleCue], + video_paths: Optional[Iterable[str]] = None, +) -> ScriptValidationResult: + """Validate final narration items against subtitle/video source constraints.""" + errors: List[str] = [] + if not isinstance(items, list) or not items: + return ScriptValidationResult(False, ["解说脚本 items 必须是非空数组"], []) + + normalized_video_paths = _normalize_paths(video_paths) + normalized_items = normalize_script_video_sources(items, normalized_video_paths) + available_video_ids = {cue.video_id for cue in subtitle_index} + if normalized_video_paths: + available_video_ids.update(range(1, len(normalized_video_paths) + 1)) + + ranges_by_video: Dict[int, List[Tuple[int, int, int]]] = {} + seen_ids = set() + required_fields = ["_id", "video_id", "video_name", "timestamp", "picture", "narration", "OST"] + + for index, item in enumerate(normalized_items): + if not isinstance(item, dict): + errors.append(f"第 {index + 1} 个片段必须是对象") + continue + + item_id = item.get("_id", index + 1) + coerced_item_id = _coerce_positive_int(item_id) + if coerced_item_id is None: + errors.append(f"第 {index + 1} 个片段缺少有效 _id") + coerced_item_id = index + 1 + elif coerced_item_id in seen_ids: + errors.append(f"片段 _id={coerced_item_id} 重复") + seen_ids.add(coerced_item_id) + + for field in required_fields: + if field not in item: + errors.append(f"片段 {item_id} 缺少字段 {field}") + + video_id = _coerce_positive_int(item.get("video_id")) + if video_id is None: + errors.append(f"片段 {item_id} 缺少有效 video_id") + continue + + if available_video_ids and video_id not in available_video_ids: + errors.append(f"片段 {item_id} 的 video_id={video_id} 不在已选视频范围内") + + expected_video_name = _default_video_name(video_id, normalized_video_paths) + if expected_video_name and os.path.basename(str(item.get("video_name") or "")) != expected_video_name: + errors.append(f"片段 {item_id} 的 video_name 必须是 {expected_video_name}") + + try: + start_ms, end_ms, normalized_timestamp = parse_script_timestamp_range(item.get("timestamp", "")) + item["timestamp"] = normalized_timestamp + except ValueError as exc: + errors.append(f"片段 {item_id}: {exc}") + continue + + if start_ms >= end_ms: + errors.append(f"片段 {item_id} 的开始时间必须早于结束时间") + continue + + video_cues = _cues_for_video(subtitle_index, video_id) + if not _range_within_subtitle_bounds(video_cues, start_ms, end_ms): + errors.append(f"片段 {item_id} 的时间戳不在视频 {video_id} 的字幕范围内") + elif not _range_overlaps_subtitle(video_cues, start_ms, end_ms): + errors.append(f"片段 {item_id} 的时间戳没有命中视频 {video_id} 的字幕内容") + + for text_field in ["picture", "narration"]: + if not isinstance(item.get(text_field), str) or not item[text_field].strip(): + errors.append(f"片段 {item_id} 的 {text_field} 不能为空") + + ost = _item_ost(item) + if item.get("OST") not in [0, 1, 2]: + errors.append(f"片段 {item_id} 的 OST 必须是 0、1 或 2") + if ost == 1 and not str(item.get("narration", "")).startswith("播放原片"): + errors.append(f"片段 {item_id} 是原声片段，narration 必须使用“播放原片+序号”") + if ost == 0: + narration_chars = count_narration_chars(item.get("narration", "")) + max_chars = max_narration_chars_for_duration(start_ms, end_ms) + if narration_chars > max_chars: + duration_seconds = (end_ms - start_ms) / 1000 + errors.append( + f"片段 {item_id} 解说过密：{narration_chars} 字需要约 {narration_chars / NARRATION_CHARS_PER_SECOND:.1f} 秒，" + f"但画面只有 {duration_seconds:.1f} 秒，建议不超过 {max_chars} 字或延长画面" + ) + + ranges_by_video.setdefault(video_id, []).append((start_ms, end_ms, coerced_item_id)) + + for video_id, ranges in ranges_by_video.items(): + sorted_ranges = sorted(ranges, key=lambda item: (item[0], item[1], item[2])) + previous_start, previous_end, previous_id = sorted_ranges[0] + for start_ms, end_ms, item_id in sorted_ranges[1:]: + if start_ms < previous_end: + errors.append(f"视频 {video_id} 的片段 {item_id} 与片段 {previous_id} 时间戳重叠") + if end_ms > previous_end: + previous_start, previous_end, previous_id = start_ms, end_ms, item_id + + errors.extend(_validate_story_continuity(normalized_items)) + + return ScriptValidationResult(not errors, errors, normalized_items) + + +def require_valid_narration_script_items( + items: Any, + subtitle_index: Sequence[SubtitleCue], + video_paths: Optional[Iterable[str]] = None, +) -> List[Dict[str, Any]]: + result = validate_narration_script_items(items, subtitle_index, video_paths) + if not result.valid: + raise NarrationScriptValidationError("\n".join(result.errors)) + return result.items + + +def summarize_subtitle_window( + subtitle_index: Sequence[SubtitleCue], + max_cues_per_video: int = 80, +) -> str: + """Return compact subtitle context for a repair prompt.""" + lines: List[str] = [] + by_video: Dict[int, List[SubtitleCue]] = {} + for cue in subtitle_index: + by_video.setdefault(cue.video_id, []).append(cue) + + for video_id in sorted(by_video): + cues = by_video[video_id][:max_cues_per_video] + video_name = cues[0].video_name if cues else "" + header = f"# 视频 {video_id}: {video_name}" if video_name else f"# 视频 {video_id}" + lines.append(header) + for cue in cues: + text = cue.text.replace("\n", " ").strip() + lines.append(f"{cue.timestamp} {text}") + if len(by_video[video_id]) > max_cues_per_video: + lines.append(f"... 已省略 {len(by_video[video_id]) - max_cues_per_video} 条字幕") + + return "\n".join(lines) diff --git a/app/services/test_short_drama_narration_validation_unittest.py b/app/services/test_short_drama_narration_validation_unittest.py new file mode 100644 index 0000000..8b2cebe --- /dev/null +++ b/app/services/test_short_drama_narration_validation_unittest.py @@ -0,0 +1,290 @@ +import unittest + +from app.services.short_drama_narration_validation import ( + build_subtitle_index, + normalize_script_video_sources, + validate_narration_script_items, +) + + +SUBTITLE_CONTENT = """# 视频 1: first.mp4 +字幕文件: first.srt +1 +00:00:01,000 --> 00:00:04,000 +女主被众人误会。 + +2 +00:00:04,000 --> 00:00:08,000 +男主冷眼看着她。 + +# 视频 2: second.mp4 +字幕文件: second.srt +1 +00:00:02,000 --> 00:00:05,000 +女主终于拿出证据。 + +2 +00:00:05,000 --> 00:00:09,000 +众人震惊，反派慌了。 +""" + + +class ShortDramaNarrationValidationTests(unittest.TestCase): + def setUp(self): + self.video_paths = ["/tmp/first.mp4", "/tmp/second.mp4"] + self.subtitle_index = build_subtitle_index(SUBTITLE_CONTENT, self.video_paths) + + def test_build_subtitle_index_preserves_multi_video_sources(self): + self.assertEqual(4, len(self.subtitle_index)) + self.assertEqual({1, 2}, {cue.video_id for cue in self.subtitle_index}) + self.assertEqual("first.mp4", self.subtitle_index[0].video_name) + self.assertEqual("second.mp4", self.subtitle_index[2].video_name) + self.assertEqual("00:00:02,000-00:00:05,000", self.subtitle_index[2].timestamp) + + def test_valid_script_passes_and_normalizes_video_name(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "wrong-name.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "她被当众误会。", + "OST": 0, + }, + { + "_id": 2, + "video_name": "second.mp4", + "timestamp": "00:00:02,000-00:00:05,000", + "picture": "女主拿出证据", + "narration": "播放原片2", + "OST": 1, + }, + ] + + normalized = normalize_script_video_sources(items, self.video_paths) + result = validate_narration_script_items(normalized, self.subtitle_index, self.video_paths) + + self.assertTrue(result.valid, result.errors) + self.assertEqual(2, result.items[1]["video_id"]) + self.assertEqual("second.mp4", result.items[1]["video_name"]) + + def test_invalid_timestamp_and_overlap_fail(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:01,000-00:00:05,000", + "picture": "画面", + "narration": "解说", + "OST": 0, + }, + { + "_id": 2, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:04,500-00:00:08,000", + "picture": "画面", + "narration": "解说", + "OST": 0, + }, + { + "_id": 3, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "bad", + "picture": "画面", + "narration": "解说", + "OST": 0, + }, + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("重叠" in error for error in result.errors)) + self.assertTrue(any("时间戳格式" in error for error in result.errors)) + + def test_invalid_video_id_does_not_default_to_first_video(self): + items = [ + { + "_id": 1, + "video_id": 99, + "video_name": "missing.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "画面", + "narration": "解说", + "OST": 0, + } + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("video_id=99" in error for error in result.errors)) + + def test_out_of_range_timestamp_fails(self): + items = [ + { + "_id": 1, + "video_id": 2, + "video_name": "second.mp4", + "timestamp": "00:00:20,000-00:00:25,000", + "picture": "画面", + "narration": "解说", + "OST": 0, + } + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("不在视频 2 的字幕范围内" in error for error in result.errors)) + + def test_three_consecutive_original_audio_segments_fail(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "她被当众误会。", + "OST": 0, + }, + { + "_id": 2, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:04,000-00:00:05,000", + "picture": "男主看着她", + "narration": "播放原片2", + "OST": 1, + }, + { + "_id": 3, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:05,000-00:00:06,000", + "picture": "男主看着她", + "narration": "播放原片3", + "OST": 1, + }, + { + "_id": 4, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:06,000-00:00:08,000", + "picture": "男主继续观察", + "narration": "播放原片4", + "OST": 1, + }, + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("连续原声过多" in error for error in result.errors)) + + def test_cross_video_original_audio_requires_narration_bridge(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "她被当众误会。", + "OST": 0, + }, + { + "_id": 2, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:04,000-00:00:08,000", + "picture": "男主看着她", + "narration": "播放原片2", + "OST": 1, + }, + { + "_id": 3, + "video_id": 2, + "video_name": "second.mp4", + "timestamp": "00:00:02,000-00:00:05,000", + "picture": "女主拿出证据", + "narration": "播放原片3", + "OST": 1, + }, + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("跨视频切换缺少 OST=0 解说桥段" in error for error in result.errors)) + + def test_cross_video_switch_with_narration_bridge_passes(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "她被当众误会。", + "OST": 0, + }, + { + "_id": 2, + "video_id": 2, + "video_name": "second.mp4", + "timestamp": "00:00:02,000-00:00:05,000", + "picture": "女主拿出证据", + "narration": "播放原片2", + "OST": 1, + }, + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertTrue(result.valid, result.errors) + + def test_first_segment_must_be_narration_hook(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "播放原片1", + "OST": 1, + } + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("解说开场钩子" in error for error in result.errors)) + + def test_dense_narration_fails_when_video_duration_is_too_short(self): + items = [ + { + "_id": 1, + "video_id": 1, + "video_name": "first.mp4", + "timestamp": "00:00:01,000-00:00:04,000", + "picture": "女主被误会", + "narration": "她明明什么都没做却被所有人推到风口浪尖只能独自承受委屈", + "OST": 0, + } + ] + + result = validate_narration_script_items(items, self.subtitle_index, self.video_paths) + + self.assertFalse(result.valid) + self.assertTrue(any("解说过密" in error for error in result.errors)) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui/components/script_settings.py b/webui/components/script_settings.py index d8b296e..d73eba0 100644 --- a/webui/components/script_settings.py +++ b/webui/components/script_settings.py @@ -14,12 +14,59 @@ from app.services.subtitle_text import decode_subtitle_bytes, read_subtitle_text from app.utils import utils, check_script from webui.tools.generate_script_docu import generate_script_docu from webui.tools.generate_script_short import generate_script_short -from webui.tools.generate_short_summary import analyze_short_drama_plot, generate_script_short_sunmmary +from webui.tools.generate_short_summary import ( + analyze_short_drama_plot, + generate_script_short_sunmmary, + generate_short_drama_narration_copy, +) SCRIPT_TABLE_BASE_COLUMNS = ["_id", "video_id", "video_name", "timestamp", "picture", "narration", "OST"] VIDEO_UPLOAD_TYPES = ["mp4", "mov", "avi", "flv", "mkv", "mpeg4"] VIDEO_GLOB_PATTERNS = [f"*.{suffix}" for suffix in VIDEO_UPLOAD_TYPES] +SHORT_DRAMA_NARRATION_LANGUAGE_OPTIONS = [ + ("zh-CN", "简体中文（中国）"), + ("en-US", "英语（美国）"), + ("ja-JP", "日语（日本）"), + ("ko-KR", "韩语（韩国）"), + ("fr-FR", "法语（法国）"), + ("de-DE", "德语（德国）"), + ("es-ES", "西班牙语（西班牙）"), + ("pt-BR", "葡萄牙语（巴西）"), + ("ru-RU", "俄语（俄罗斯）"), + ("custom", "自定义"), +] +SHORT_DRAMA_NARRATION_LANGUAGE_VALUES = { + "zh-CN": "简体中文（中国）", + "en-US": "英语（美国）", + "ja-JP": "日语（日本）", + "ko-KR": "韩语（韩国）", + "fr-FR": "法语（法国）", + "de-DE": "德语（德国）", + "es-ES": "西班牙语（西班牙）", + "pt-BR": "葡萄牙语（巴西）", + "ru-RU": "俄语（俄罗斯）", +} +SHORT_DRAMA_TYPE_OPTIONS = [ + ("counterattack", "逆袭/复仇"), + ("ceo_romance", "霸总/甜宠"), + ("family", "家庭伦理"), + ("costume", "古装/权谋"), + ("suspense", "悬疑/犯罪"), + ("urban_emotion", "都市情感"), + ("period_rural", "年代/乡村"), + ("custom", "自定义"), +] +SHORT_DRAMA_TYPE_VALUES = { + "counterattack": "逆袭/复仇", + "ceo_romance": "霸总/甜宠", + "family": "家庭伦理", + "costume": "古装/权谋", + "suspense": "悬疑/犯罪", + "urban_emotion": "都市情感", + "period_rural": "年代/乡村", +} +SHORT_DRAMA_ORIGINAL_SOUND_RATIO_OPTIONS = list(range(0, 100, 10)) def _normalize_video_paths(paths): @@ -154,6 +201,22 @@ def _short_drama_plot_analysis_signature(subtitle_paths, video_theme, web_search ) +def _resolve_short_drama_narration_language(): + selected_language = st.session_state.get('short_drama_narration_language_option', 'zh-CN') + custom_language = str(st.session_state.get('short_drama_custom_narration_language', '') or '').strip() + if selected_language == "custom" and custom_language: + return custom_language + return SHORT_DRAMA_NARRATION_LANGUAGE_VALUES.get(selected_language, "简体中文（中国）") + + +def _resolve_short_drama_type(): + selected_type = st.session_state.get('short_drama_type_option', 'counterattack') + custom_type = str(st.session_state.get('short_drama_custom_type', '') or '').strip() + if selected_type == "custom" and custom_type: + return custom_type + return SHORT_DRAMA_TYPE_VALUES.get(selected_type, "逆袭/复仇") + + def render_script_panel(tr): """渲染脚本配置面板""" with st.container(border=True): @@ -1211,41 +1274,136 @@ def render_script_buttons(tr, params): elif script_path == "short": button_name = tr("Generate Short Video Script") elif script_path == "summary": - button_name = tr("生成短剧解说脚本") + button_name = tr("生成剪辑脚本") elif script_path.endswith("json"): button_name = tr("Load Video Script") else: button_name = tr("Please Select Script File") - if st.button(button_name, key="script_action", disabled=not script_path): - if script_path == "auto": - # 执行纪录片视频脚本生成（视频无字幕无配音） - generate_script_docu(params, tr) - elif script_path == "short": - # 执行 短剧混剪 脚本生成 - custom_clips = st.session_state.get('custom_clips') - generate_script_short(tr, params, custom_clips) - elif script_path == "summary": - # 执行 短剧解说 脚本生成 - subtitle_paths = _selected_subtitle_paths() - subtitle_path = subtitle_paths[0] if subtitle_paths else None - video_theme = st.session_state.get('video_theme') - temperature = st.session_state.get('temperature') - web_search_enabled = bool(st.session_state.get('short_drama_web_search_enabled', False)) - current_signature = _short_drama_plot_analysis_signature( - subtitle_paths, - video_theme, - web_search_enabled, - _selected_video_paths(), + if script_path == "summary": + config_cols = st.columns([1.15, 1.15, 0.9, 1.15, 1.15], vertical_alignment="bottom") + with config_cols[0]: + st.selectbox( + tr("短剧类型"), + options=[code for code, _ in SHORT_DRAMA_TYPE_OPTIONS], + format_func=lambda code: tr(dict(SHORT_DRAMA_TYPE_OPTIONS).get(code, code)), + key="short_drama_type_option", ) - plot_analysis = "" - if st.session_state.get('short_drama_plot_analysis_signature') == current_signature: - plot_analysis = st.session_state.get('short_drama_plot_analysis', '') - elif ( - not web_search_enabled - and st.session_state.get('short_drama_plot_analysis_subtitle_path') == subtitle_path - ): - plot_analysis = st.session_state.get('short_drama_plot_analysis', '') + with config_cols[1]: + custom_type_disabled = ( + st.session_state.get('short_drama_type_option', 'counterattack') != "custom" + ) + st.text_input( + tr("自定义短剧类型"), + key="short_drama_custom_type", + placeholder=tr("例如：豪门虐恋"), + disabled=custom_type_disabled, + ) + with config_cols[2]: + st.selectbox( + tr("原片占比"), + options=SHORT_DRAMA_ORIGINAL_SOUND_RATIO_OPTIONS, + format_func=lambda ratio: f"{ratio}%", + index=SHORT_DRAMA_ORIGINAL_SOUND_RATIO_OPTIONS.index(30), + key="short_drama_original_sound_ratio", + ) + with config_cols[3]: + st.selectbox( + tr("解说语言"), + options=[code for code, _ in SHORT_DRAMA_NARRATION_LANGUAGE_OPTIONS], + format_func=lambda code: tr(dict(SHORT_DRAMA_NARRATION_LANGUAGE_OPTIONS).get(code, code)), + key="short_drama_narration_language_option", + ) + with config_cols[4]: + custom_language_disabled = ( + st.session_state.get('short_drama_narration_language_option', 'zh-CN') != "custom" + ) + st.text_input( + tr("自定义解说语言"), + key="short_drama_custom_narration_language", + placeholder=tr("例如：意大利语（意大利）"), + disabled=custom_language_disabled, + ) + + action_cols = st.columns([1, 1], vertical_alignment="bottom") + with action_cols[0]: + narration_copy_clicked = st.button( + tr("生成解说文案"), + key="short_drama_narration_copy_action", + disabled=not script_path, + use_container_width=True, + ) + with action_cols[1]: + action_clicked = st.button( + button_name, + key="script_action", + disabled=not script_path, + use_container_width=True, + ) + else: + narration_copy_clicked = False + action_clicked = st.button(button_name, key="script_action", disabled=not script_path) + + if script_path == "summary" and (narration_copy_clicked or action_clicked): + narration_language = _resolve_short_drama_narration_language() + drama_genre = _resolve_short_drama_type() + original_sound_ratio = int(st.session_state.get('short_drama_original_sound_ratio', 30)) + if ( + st.session_state.get('short_drama_type_option') == "custom" + and not str(st.session_state.get('short_drama_custom_type', '') or '').strip() + ): + st.error(tr("请输入自定义短剧类型")) + st.stop() + if ( + st.session_state.get('short_drama_narration_language_option') == "custom" + and not str(st.session_state.get('short_drama_custom_narration_language', '') or '').strip() + ): + st.error(tr("请输入自定义解说语言")) + st.stop() + + subtitle_paths = _selected_subtitle_paths() + subtitle_path = subtitle_paths[0] if subtitle_paths else None + video_theme = st.session_state.get('video_theme') + temperature = st.session_state.get('temperature') + web_search_enabled = bool(st.session_state.get('short_drama_web_search_enabled', False)) + current_signature = _short_drama_plot_analysis_signature( + subtitle_paths, + video_theme, + web_search_enabled, + _selected_video_paths(), + ) + plot_analysis = "" + if st.session_state.get('short_drama_plot_analysis_signature') == current_signature: + plot_analysis = st.session_state.get('short_drama_plot_analysis', '') + elif ( + not web_search_enabled + and st.session_state.get('short_drama_plot_analysis_subtitle_path') == subtitle_path + ): + plot_analysis = st.session_state.get('short_drama_plot_analysis', '') + + if narration_copy_clicked: + with st.spinner(tr("Generating narration copy...")): + copy_result = generate_short_drama_narration_copy( + subtitle_paths, + video_theme, + temperature, + tr, + plot_analysis=plot_analysis, + subtitle_content=st.session_state.get('subtitle_content', ''), + enable_web_search=web_search_enabled, + video_paths=_selected_video_paths(), + narration_language=narration_language, + drama_genre=drama_genre, + ) + if copy_result: + st.session_state['short_drama_narration_copy'] = copy_result["narration_copy"] + if not plot_analysis: + st.session_state['short_drama_plot_analysis'] = copy_result["plot_analysis"] + st.session_state['short_drama_plot_analysis_subtitle_path'] = subtitle_path + st.session_state['short_drama_plot_analysis_signature'] = current_signature + st.success(tr("Narration copy generated successfully")) + + if action_clicked: generate_script_short_sunmmary( params, subtitle_paths, @@ -1256,7 +1414,28 @@ def render_script_buttons(tr, params): subtitle_content=st.session_state.get('subtitle_content', ''), enable_web_search=web_search_enabled, video_paths=_selected_video_paths(), + narration_language=narration_language, + narration_copy=st.session_state.get('short_drama_narration_copy', ''), + drama_genre=drama_genre, + original_sound_ratio=original_sound_ratio, ) + + if script_path == "summary": + st.text_area( + tr("短剧解说文案"), + key="short_drama_narration_copy", + height=220, + help=tr("Narration Copy Help"), + ) + + if action_clicked and script_path != "summary": + if script_path == "auto": + # 执行纪录片视频脚本生成（视频无字幕无配音） + generate_script_docu(params, tr) + elif script_path == "short": + # 执行 短剧混剪 脚本生成 + custom_clips = st.session_state.get('custom_clips') + generate_script_short(tr, params, custom_clips) else: load_script(tr, script_path) diff --git a/webui/i18n/en.json b/webui/i18n/en.json index 284d9a6..e4de584 100644 --- a/webui/i18n/en.json +++ b/webui/i18n/en.json @@ -266,6 +266,37 @@ "字幕文件内容似乎为空，请检查文件": "The subtitle file appears to be empty. Please check the file.", "字幕上传成功": "Subtitle uploaded successfully", "短剧名称": "Short Drama Name", + "解说语言": "Narration Language", + "自定义解说语言": "Custom Narration Language", + "例如：意大利语（意大利）": "For example: Italian (Italy)", + "请输入自定义解说语言": "Please enter a custom narration language", + "简体中文（中国）": "Simplified Chinese (China)", + "英语（美国）": "English (United States)", + "日语（日本）": "Japanese (Japan)", + "韩语（韩国）": "Korean (South Korea)", + "法语（法国）": "French (France)", + "德语（德国）": "German (Germany)", + "西班牙语（西班牙）": "Spanish (Spain)", + "葡萄牙语（巴西）": "Portuguese (Brazil)", + "俄语（俄罗斯）": "Russian (Russia)", + "自定义": "Custom", + "短剧类型": "Short Drama Type", + "自定义短剧类型": "Custom Short Drama Type", + "原片占比": "Original Footage Ratio", + "例如：豪门虐恋": "For example: billionaire angst romance", + "请输入自定义短剧类型": "Please enter a custom short drama type", + "逆袭/复仇": "Counterattack / Revenge", + "霸总/甜宠": "CEO Romance / Sweet Romance", + "家庭伦理": "Family Ethics", + "古装/权谋": "Costume / Power Struggle", + "悬疑/犯罪": "Suspense / Crime", + "都市情感": "Urban Romance", + "年代/乡村": "Period / Rural", + "生成解说文案": "Generate Narration Copy", + "生成剪辑脚本": "Generate Editing Script", + "短剧解说文案": "Short Drama Narration Copy", + "Narration Copy Help": "Generate the narration copy first, review or rewrite it here, then generate the editing script to match footage and timestamps.", + "Narration copy generated successfully": "Narration copy generated. Please review and edit it.", "生成短剧解说脚本": "Generate Short Drama Narration Script", "请输入视频脚本": "Please enter the video script", "TTS engine does not support precise subtitles": "⚠️ {engine} does not support precise subtitle generation", @@ -587,11 +618,22 @@ "Preparing script generation": "Preparing script generation", "Script generation failed check logs": "Script generation failed. Please check the logs.", "Parsing subtitles...": "Parsing subtitles...", + "Analyzing subtitles with model...": "Waiting for the model to analyze subtitles...", "Subtitle file does not exist": "Subtitle file does not exist", "Subtitle file is empty or unreadable": "Subtitle file is empty or unreadable", "Generating narration copy...": "Generating narration copy...", + "Generated narration copy is empty": "The generated narration copy is empty", + "Please generate and review narration copy first": "Please generate and review the narration copy first", + "Matching narration copy to footage...": "Matching narration copy to footage and timestamps...", + "Waiting for model stream...": "Waiting for model stream...", + "Streaming unavailable fallback waiting...": "Streaming is unavailable for this request. Waiting for the full response...", + "LLM stream window title": "Model reasoning / output stream", + "Model reasoning stream": "[Model reasoning]", + "Model output preview": "[Model output preview]", + "Repairing narration script...": "Repairing narration script...", "Generated narration JSON parse failed": "The generated narration format is invalid and could not be parsed as JSON", "Generated narration missing items field": "The generated narration is missing the required 'items' field", + "Generated narration validation failed": "The generated narration script failed validation", "Preparing output...": "Preparing output..." } } diff --git a/webui/i18n/zh.json b/webui/i18n/zh.json index 76872eb..e400adb 100644 --- a/webui/i18n/zh.json +++ b/webui/i18n/zh.json @@ -561,6 +561,37 @@ "字幕文件内容似乎为空，请检查文件": "字幕文件内容似乎为空，请检查文件", "字幕上传成功": "字幕上传成功", "短剧名称": "短剧名称", + "解说语言": "解说语言", + "自定义解说语言": "自定义解说语言", + "例如：意大利语（意大利）": "例如：意大利语（意大利）", + "请输入自定义解说语言": "请输入自定义解说语言", + "简体中文（中国）": "简体中文（中国）", + "英语（美国）": "英语（美国）", + "日语（日本）": "日语（日本）", + "韩语（韩国）": "韩语（韩国）", + "法语（法国）": "法语（法国）", + "德语（德国）": "德语（德国）", + "西班牙语（西班牙）": "西班牙语（西班牙）", + "葡萄牙语（巴西）": "葡萄牙语（巴西）", + "俄语（俄罗斯）": "俄语（俄罗斯）", + "自定义": "自定义", + "短剧类型": "短剧类型", + "自定义短剧类型": "自定义短剧类型", + "原片占比": "原片占比", + "例如：豪门虐恋": "例如：豪门虐恋", + "请输入自定义短剧类型": "请输入自定义短剧类型", + "逆袭/复仇": "逆袭/复仇", + "霸总/甜宠": "霸总/甜宠", + "家庭伦理": "家庭伦理", + "古装/权谋": "古装/权谋", + "悬疑/犯罪": "悬疑/犯罪", + "都市情感": "都市情感", + "年代/乡村": "年代/乡村", + "生成解说文案": "生成解说文案", + "生成剪辑脚本": "生成剪辑脚本", + "短剧解说文案": "短剧解说文案", + "Narration Copy Help": "先点击生成解说文案；审核、删改或重写这段文案后，再点击生成剪辑脚本匹配画面和时间戳。", + "Narration copy generated successfully": "解说文案已生成，可先审核修改", "生成短剧解说脚本": "生成短剧解说脚本", "请输入视频脚本": "请输入视频脚本", "自定义片段": "自定义片段", @@ -583,11 +614,22 @@ "Preparing script generation": "开始准备生成脚本", "Script generation failed check logs": "生成脚本失败，请检查日志", "Parsing subtitles...": "正在解析字幕...", + "Analyzing subtitles with model...": "正在等待模型分析字幕...", "Subtitle file does not exist": "字幕文件不存在", "Subtitle file is empty or unreadable": "字幕文件内容为空或无法读取", "Generating narration copy...": "正在生成文案...", + "Generated narration copy is empty": "生成的解说文案为空", + "Please generate and review narration copy first": "请先生成并审核解说文案", + "Matching narration copy to footage...": "正在根据解说文案匹配画面和时间戳...", + "Waiting for model stream...": "正在等待模型流式输出...", + "Streaming unavailable fallback waiting...": "当前接口未返回流式内容，正在等待完整响应...", + "LLM stream window title": "模型思考 / 输出流", + "Model reasoning stream": "【模型思考】", + "Model output preview": "【模型输出预览】", + "Repairing narration script...": "正在修复解说脚本...", "Generated narration JSON parse failed": "生成的解说文案格式错误，无法解析为 JSON", "Generated narration missing items field": "生成的解说文案缺少必要的 'items' 字段", + "Generated narration validation failed": "生成的解说脚本校验失败", "Preparing output...": "整理输出..." } } diff --git a/webui/tools/generate_short_summary.py b/webui/tools/generate_short_summary.py index d06431c..ab1e71b 100644 --- a/webui/tools/generate_short_summary.py +++ b/webui/tools/generate_short_summary.py @@ -11,12 +11,20 @@ import os import json import time import traceback +import html import streamlit as st from loguru import logger from app.config import config -from app.services.SDE.short_drama_explanation import analyze_subtitle, generate_narration_script +from app.services.SDE.short_drama_explanation import ( + analyze_subtitle, + generate_narration_copy as generate_narration_copy_legacy, + match_narration_copy_to_script as match_narration_copy_to_script_legacy, +) from app.services.subtitle_text import read_subtitle_text +from app.services.short_drama_narration_validation import ( + normalize_script_video_sources, +) from app.services.tavily_search import TavilySearchError, format_search_context, search_short_drama # 导入新的LLM服务模块 - 确保提供商被注册 import app.services.llm # 这会触发提供商注册 @@ -24,6 +32,9 @@ from app.services.llm.migration_adapter import SubtitleAnalyzerAdapter import re +PUBLIC_SCRIPT_FIELDS = ["_id", "video_id", "video_name", "timestamp", "picture", "narration", "OST"] + + def _normalize_paths(paths): if isinstance(paths, str): paths = [paths] @@ -63,53 +74,23 @@ def _build_combined_subtitle_content(subtitle_paths, video_paths=None): return "\n\n".join(sections) -def _coerce_video_id(value): - try: - video_id = int(value) - except (TypeError, ValueError): - return None - return video_id if video_id > 0 else None - - -def _match_video_id_by_name(video_name, video_paths): - video_name = str(video_name or "").strip() - if not video_name: - return None - - for index, video_path in enumerate(video_paths, start=1): - if os.path.basename(video_path) == os.path.basename(video_name): - return index - return None - - def _normalize_narration_items_video_sources(items, video_paths): - video_paths = _normalize_paths(video_paths) - if not video_paths: - return items + return normalize_script_video_sources(items, _normalize_paths(video_paths)) - normalized_items = [] - for item in items: - if not isinstance(item, dict): - normalized_items.append(item) - continue - item_copy = item.copy() - video_id = _coerce_video_id(item_copy.get("video_id") or item_copy.get("video_index")) - matched_video_id = _match_video_id_by_name( - item_copy.get("video_name") or item_copy.get("source_video"), - video_paths, - ) - if matched_video_id: - video_id = matched_video_id - if video_id is None or video_id > len(video_paths): - logger.warning(f"片段 {item_copy.get('_id')} 未提供有效 video_id，默认使用视频 1") - video_id = 1 +def _strip_planner_only_fields(items): + return [ + {field: item[field] for field in PUBLIC_SCRIPT_FIELDS if field in item} + for item in items + if isinstance(item, dict) + ] - item_copy["video_id"] = video_id - item_copy["video_name"] = os.path.basename(video_paths[video_id - 1]) - normalized_items.append(item_copy) - return normalized_items +def _format_progress_status(progress, message: str = "", tr=lambda key: key): + message = str(message or "").strip() + if message: + return message + return f"{tr('Progress')}: {progress}%" def parse_and_fix_json(json_string): @@ -203,25 +184,9 @@ def parse_and_fix_json(json_string): logger.debug(f"综合修复失败: {e}") pass - # 如果所有方法都失败，尝试创建一个基本的结构 + # 如果所有方法都失败，直接返回 None，避免生成不可剪辑的默认假脚本 logger.error(f"所有JSON解析方法都失败，原始内容: {json_string[:200]}...") - - # 尝试从文本中提取关键信息创建基本结构 - try: - # 这是一个简单的回退方案 - return { - "items": [ - { - "_id": 1, - "timestamp": "00:00:00,000-00:00:10,000", - "picture": "解析失败，使用默认内容", - "narration": json_string[:100] + "..." if len(json_string) > 100 else json_string, - "OST": 0 - } - ] - } - except Exception: - return None + return None def _get_tavily_api_key() -> str: @@ -350,6 +315,100 @@ def analyze_short_drama_plot( return analysis_result["analysis"] +def generate_short_drama_narration_copy( + subtitle_path, + video_theme, + temperature, + tr=lambda key: key, + plot_analysis=None, + subtitle_content=None, + enable_web_search: bool = False, + video_paths=None, + narration_language: str = "简体中文（中国）", + drama_genre: str = "逆袭/复仇", +): + """生成可由用户审核修改的短剧解说正文，不绑定时间戳。""" + subtitle_paths = _normalize_paths(subtitle_path) + if not subtitle_paths: + st.error(tr("Please generate or upload subtitles first")) + return None + missing_subtitle_paths = [path for path in subtitle_paths if not os.path.exists(path)] + if missing_subtitle_paths: + st.error(tr("Subtitle file does not exist")) + return None + + selected_video_paths = _normalize_paths(video_paths) + subtitle_content = str(subtitle_content or "").strip() or _build_combined_subtitle_content( + subtitle_paths, + selected_video_paths, + ) + if not subtitle_content: + st.error(tr("Subtitle file is empty or unreadable")) + return None + + analysis_text = str(plot_analysis or "").strip() + if not analysis_text: + analysis_text = analyze_short_drama_plot( + subtitle_paths, + temperature, + tr, + subtitle_content=subtitle_content, + short_name=video_theme, + enable_web_search=enable_web_search, + video_paths=selected_video_paths, + ) + if not analysis_text: + return None + + text_provider = config.app.get('text_llm_provider', 'gemini').lower() + text_api_key = config.app.get(f'text_{text_provider}_api_key') + text_model = config.app.get(f'text_{text_provider}_model_name') + text_base_url = config.app.get(f'text_{text_provider}_base_url') + + try: + logger.info("使用新的LLM服务架构生成可审核解说文案") + analyzer = SubtitleAnalyzerAdapter(text_api_key, text_model, text_base_url, text_provider) + narration_result = analyzer.generate_narration_copy( + short_name=video_theme, + plot_analysis=analysis_text, + subtitle_content=subtitle_content, + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, + ) + except Exception as e: + logger.warning(f"使用新LLM服务生成文案失败，回退到旧实现: {str(e)}") + narration_result = generate_narration_copy_legacy( + short_name=video_theme, + plot_analysis=analysis_text, + subtitle_content=subtitle_content, + api_key=text_api_key, + model=text_model, + base_url=text_base_url, + temperature=temperature, + provider=text_provider, + narration_language=narration_language, + drama_genre=drama_genre, + ) + + if narration_result.get("status") != "success": + logger.error(f"解说文案正文生成失败: {narration_result.get('message')}") + st.error(tr("Script generation failed check logs")) + return None + + narration_copy = str(narration_result.get("narration_copy", "")).strip() + if not narration_copy: + logger.error("模型返回空解说文案正文") + st.error(tr("Generated narration copy is empty")) + return None + + return { + "narration_copy": narration_copy, + "plot_analysis": analysis_text, + "subtitle_content": subtitle_content, + } + + def generate_script_short_sunmmary( params, subtitle_path, @@ -360,21 +419,79 @@ def generate_script_short_sunmmary( subtitle_content=None, enable_web_search: bool = False, video_paths=None, + narration_language: str = "简体中文（中国）", + narration_copy: str = "", + drama_genre: str = "逆袭/复仇", + original_sound_ratio: int = 30, ): """ 生成 短剧解说 视频脚本 要求: 提供高质量短剧字幕 适合场景: 短剧 """ - progress_bar = st.progress(0) + progress_bar = st.empty() status_text = st.empty() + stream_text = st.empty() + stream_state = { + "reasoning": "", + "content": "", + "last_update": 0.0, + } def update_progress(progress: float, message: str = ""): progress_bar.progress(progress) + status_text.text(_format_progress_status(progress, message, tr)) + + def update_waiting(message: str = ""): + progress_bar.empty() if message: - status_text.text(f"{progress}% - {message}") + status_text.text(message) else: - status_text.text(f"{tr('Progress')}: {progress}%") + status_text.empty() + + def update_stream_window(event): + event = event or {} + chunk_type = str(event.get("type") or "content") + chunk_text = str(event.get("text") or "") + if chunk_type == "done" or not chunk_text: + return + + bucket = "reasoning" if chunk_type == "reasoning" else "content" + stream_state[bucket] += chunk_text + + now = time.time() + if now - stream_state["last_update"] < 0.12: + return + stream_state["last_update"] = now + + blocks = [] + if stream_state["reasoning"].strip(): + blocks.append( + f"{tr('Model reasoning stream')}\n" + f"{stream_state['reasoning'][-900:]}" + ) + if stream_state["content"].strip(): + blocks.append( + f"{tr('Model output preview')}\n" + f"{stream_state['content'][-900:]}" + ) + + preview = "\n\n".join(blocks)[-1800:] + escaped_preview = html.escape(preview) + stream_text.markdown( + f""" +
+
+ {html.escape(tr('LLM stream window title'))} +
+ 
{escaped_preview}
+
+ """, + unsafe_allow_html=True, + ) try: with st.spinner(tr("Generating script...")): @@ -414,9 +531,14 @@ def generate_script_short_sunmmary( st.error(tr("Subtitle file is empty or unreadable")) return + narration_copy = str(narration_copy or "").strip() + if not narration_copy: + st.error(tr("Please generate and review narration copy first")) + return + analyzer = SubtitleAnalyzerAdapter(text_api_key, text_model, text_base_url, text_provider) if plot_analysis and str(plot_analysis).strip(): - logger.info("使用用户编辑后的剧情理解结果生成解说文案") + logger.info("使用用户编辑后的剧情理解结果匹配剪辑脚本") analysis_result = { "status": "success", "analysis": str(plot_analysis).strip(), @@ -424,7 +546,7 @@ def generate_script_short_sunmmary( else: plot_analysis_input = subtitle_content if enable_web_search: - update_progress(40, tr("Searching short drama with Tavily...")) + update_waiting(tr("Searching short drama with Tavily...")) plot_analysis_input = _build_plot_analysis_input( subtitle_content, short_name=video_theme, @@ -436,11 +558,13 @@ def generate_script_short_sunmmary( try: # 优先使用新的LLM服务架构 logger.info("使用新的LLM服务架构进行字幕分析") + update_waiting(tr("Analyzing subtitles with model...")) analysis_result = analyzer.analyze_subtitle(plot_analysis_input) except Exception as e: logger.warning(f"使用新LLM服务失败，回退到旧实现: {str(e)}") # 回退到旧的实现 + update_waiting(tr("Analyzing subtitles with model...")) analysis_result = analyze_subtitle( subtitle_content=plot_analysis_input, api_key=text_api_key, @@ -451,42 +575,50 @@ def generate_script_short_sunmmary( provider=text_provider ) """ - 3. 根据剧情生成解说文案 + 3. 根据用户审核后的文案匹配画面与时间戳 """ if analysis_result["status"] == "success": logger.info("字幕分析成功！") - update_progress(60, tr("Generating narration copy...")) + update_waiting() - # 根据剧情生成解说文案 - 使用新的LLM服务架构 try: - # 优先使用新的LLM服务架构 - logger.info("使用新的LLM服务架构生成解说文案") - narration_result = analyzer.generate_narration_script( + logger.info("使用新的LLM服务架构将审核文案匹配到字幕画面") + update_waiting(tr("Matching narration copy to footage...")) + stream_text.info(tr("Waiting for model stream...")) + narration_result = analyzer.match_narration_copy_to_script( short_name=video_theme, plot_analysis=analysis_result["analysis"], - subtitle_content=subtitle_content, # 传递原始字幕内容 - temperature=temperature + subtitle_content=subtitle_content, + narration_copy=narration_copy, + temperature=temperature, + narration_language=narration_language, + drama_genre=drama_genre, + original_sound_ratio=original_sound_ratio, + stream_callback=update_stream_window, ) except Exception as e: - logger.warning(f"使用新LLM服务失败，回退到旧实现: {str(e)}") - # 回退到旧的实现 - narration_result = generate_narration_script( + logger.warning(f"使用新LLM服务匹配画面失败，回退到旧实现: {str(e)}") + stream_text.info(tr("Streaming unavailable fallback waiting...")) + narration_result = match_narration_copy_to_script_legacy( short_name=video_theme, plot_analysis=analysis_result["analysis"], - subtitle_content=subtitle_content, # 传递原始字幕内容 + subtitle_content=subtitle_content, + narration_copy=narration_copy, api_key=text_api_key, model=text_model, base_url=text_base_url, - save_result=True, temperature=temperature, - provider=text_provider + provider=text_provider, + narration_language=narration_language, + drama_genre=drama_genre, + original_sound_ratio=original_sound_ratio, ) if narration_result["status"] == "success": - logger.info("\n解说文案生成成功！") + logger.info("\n剪辑脚本匹配成功！") logger.info(narration_result["narration_script"]) else: - logger.info(f"\n解说文案生成失败: {narration_result['message']}") + logger.info(f"\n剪辑脚本匹配失败: {narration_result['message']}") st.error(tr("Script generation failed check logs")) st.stop() else: @@ -519,6 +651,7 @@ def generate_script_short_sunmmary( narration_dict['items'], selected_video_paths, ) + narration_items = _strip_planner_only_fields(narration_items) script = json.dumps(narration_items, ensure_ascii=False, indent=2) if script is None: @@ -543,3 +676,4 @@ def generate_script_short_sunmmary( time.sleep(2) progress_bar.empty() status_text.empty() + stream_text.empty() diff --git a/webui/tools/test_generate_short_summary_unittest.py b/webui/tools/test_generate_short_summary_unittest.py new file mode 100644 index 0000000..30122d6 --- /dev/null +++ b/webui/tools/test_generate_short_summary_unittest.py @@ -0,0 +1,27 @@ +import unittest + +from webui.tools.generate_short_summary import _format_progress_status, parse_and_fix_json + + +class GenerateShortSummaryJsonTests(unittest.TestCase): + def test_progress_message_does_not_prefix_fake_percentage(self): + status = _format_progress_status(60, "正在生成文案...") + + self.assertEqual("正在生成文案...", status) + self.assertNotIn("60%", status) + + def test_invalid_json_does_not_create_default_fake_script(self): + self.assertIsNone(parse_and_fix_json("not a json response")) + + def test_json_code_block_is_parsed(self): + parsed = parse_and_fix_json( + """```json +{"items": [{"_id": 1, "timestamp": "00:00:01,000-00:00:02,000"}]} +```""" + ) + + self.assertEqual(1, parsed["items"][0]["_id"]) + + +if __name__ == "__main__": + unittest.main()