feat(short-drama): 完整实现短剧解说剪辑全流程并新增LLM流式生成支持

- 新增短剧解说全流程四类提示词模板:解说文案生成、片段规划、文案画面匹配、脚本修复
- 重构原有脚本生成提示词至v2.1,改为基于上游规划片段生成合规解说脚本
- 为LLM基础服务层新增流式文本生成接口,完善OpenAI兼容提供商的流式实现,支持流式回调与推理内容提取
- 重构OpenAI兼容文本提供商的生成逻辑,提取公共参数构建方法
- 新增多语言国际化文案,覆盖解说语言、短剧类型、原片占比等配置项与交互提示
- 新增多套单元测试,覆盖脚本校验、适配器流程、工具函数等模块
- 封装SubtitleAnalyzerAdapter,统一短剧解说脚本生成的整套业务接口
- 新增前端交互所需的解说文案审核相关提示文案
This commit is contained in:
viccy 2026-06-07 17:10:48 +08:00
parent 342fc15f3b
commit e6e39d2dcd
19 changed files with 2737 additions and 418 deletions

View File

@ -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"

View File

@ -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]]:
"""构建消息列表"""

View File

@ -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字符串

View File

@ -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

View File

@ -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()

View File

@ -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,

View File

@ -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"
]

View File

@ -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>
${plot_analysis}
</plot>
## 原始字幕
<subtitles>
${subtitle_content}
</subtitles>
## 输出语言
<narration_language>
${narration_language}
</narration_language>
## 用户选择的短剧类型
<drama_genre>
${drama_genre}
</drama_genre>
## 类型写作规则
必须按用户选择的短剧类型调整表达重点不要自行改判类型
- 霸总/甜宠突出误会身份差暧昧拉扯守护感和情绪反差
- 逆袭/复仇突出羞辱反击打脸身份揭露和爽点升级
- 家庭伦理突出亲情撕扯秘密委屈选择和道德冲突
- 古装/权谋突出身份局势算计立场和反转
- 悬疑/犯罪突出线索危机动机和未揭开的疑问
- 都市情感突出关系裂痕现实压力误会和情绪拉扯
- 年代/乡村突出家庭处境人情压力生活困境和命运转折
- 自定义类型严格服从用户填写的类型方向
## 开头钩子公式
开头必须使用高能反转 + 情绪冲突 + 悬念钩子
1. 强身份或强处境兵王单亲妈妈被赶出家门的女人被全家看不起的人等
2. 致命反差刚立功就被迫退役刚回家就发现钱被输光刚结婚就遇到孩子/婆婆阻挠
3. 后续悬念真正的噩梦才开始他要讨回的不是钱这段关系真正难的不是相爱
## 写作规则
1. 必须使用 ${narration_language}
2. 严格基于剧情理解和字幕事实不编造核心情节身份结局
3. 先写完整故事线再写金句不要只堆爆点
4. 每句话只表达一个信息点适合后续按句匹配画面
5. 句子尽量短单句优先 15-35 信息复杂时拆成多句
6. 2-3 句要有明确因果承接让观众知道为什么从上一幕来到下一幕
7. 总长度控制在 300-650 短素材取下限长素材取上限
8. 不要使用编号项目符号章节标题或括号说明
## 输出要求
只输出解说正文不要输出 JSON时间戳代码块或任何解释"""

View File

@ -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>
${plot_analysis}
</plot>
### 已规划片段(必须逐项照抄结构字段)
<segment_plan>
${segment_plan}
</segment_plan>
### 原始字幕(含视频编号和精确时间戳)
<subtitles>
${subtitle_content}
</subtitles>
### 解说台词语言
<narration_language>
${narration_language}
</narration_language>
### 用户选择的短剧类型
<drama_genre>
${drama_genre}
</drama_genre>
字幕可能来自多个视频文件每个字幕分段标题会以视频 1: 文件名视频 2: 文件名等形式标识来源
生成脚本时必须把每个片段绑定到对应视频来源时间戳表示该视频文件内部的局部时间不是把多个视频拼接后的全局时间
所有 OST=0 narration 字段必须使用上方指定的解说台词语言输出不要因为原始字幕是其他语言就切回字幕原语言
OST=1 的原声片段 narration 字段必须继续使用播放原片+序号格式不要翻译这个固定标记
## 短剧解说创作核心要素
## 绝对绑定规则
1. 输出 items 数量顺序和 _id 必须与 segment_plan 完全一致
2. 每个 item _idvideo_idvideo_nametimestampOST 必须逐字复制 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_roleintenttransition 字段必须利用它们组织 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再发生BA导致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}创作解说脚本"""

View File

@ -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>
${plot_analysis}
</plot>
## 用户审核后的解说文案
<narration_copy>
${narration_copy}
</narration_copy>
## 原始字幕(含视频编号和局部时间戳)
<subtitles>
${subtitle_content}
</subtitles>
## 输出语言
<narration_language>
${narration_language}
</narration_language>
## 用户选择的短剧类型
<drama_genre>
${drama_genre}
</drama_genre>
## 用户选择的原片占比
<original_sound_ratio>
${original_sound_ratio}%
</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描述匹配画面中人物动作情绪和场景
- narrationOST=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
}
]
}
现在请基于用户审核后的解说文案生成最终剪辑脚本"""

View File

@ -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>
${plot_analysis}
</plot>
## 校验错误
<validation_errors>
${validation_errors}
</validation_errors>
## 当前无效脚本
<invalid_script>
${invalid_script}
</invalid_script>
## 可用字幕窗口
<subtitles>
${subtitle_content}
</subtitles>
## 解说台词目标语言
<narration_language>
${narration_language}
</narration_language>
## 用户选择的短剧类型
<drama_genre>
${drama_genre}
</drama_genre>
## 修复规则
1. 只输出 JSON不要任何解释标题Markdown 或代码块
2. 输出根对象必须是 {"items": [...]}
3. 每个 item 必须包含 _idvideo_idvideo_nametimestamppicturenarrationOST
4. video_idvideo_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"""

View File

@ -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>
${plot_analysis}
</plot>
## 原始字幕(含视频编号和局部时间戳)
<subtitles>
${subtitle_content}
</subtitles>
## 解说台词目标语言
<narration_language>
${narration_language}
</narration_language>
## 用户选择的短剧类型
<drama_genre>
${drama_genre}
</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}的解说片段"""

View File

@ -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<start>\d{2}:\d{2}:\d{2}[,.]\d{3})-(?P<end>\d{2}:\d{2}:\d{2}[,.]\d{3})$"
)
SRT_RANGE_RE = re.compile(
r"(?P<start>\d{2}:\d{2}:\d{2}[,.]\d{3})\s*-->\s*"
r"(?P<end>\d{2}:\d{2}:\d{2}[,.]\d{3})"
)
VIDEO_HEADER_RE = re.compile(r"^#\s*视频\s*(?P<video_id>\d+)(?:\s*[:]\s*(?P<video_name>.+?))?\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)

View File

@ -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()

View File

@ -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)

View File

@ -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..."
}
}

View File

@ -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...": "整理输出..."
}
}

View File

@ -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"""
<div style="height:150px; overflow:hidden; border:1px solid #e5e7eb;
border-radius:8px; padding:10px 12px; background:#f8fafc;
color:#334155;">
<div style="font-size:12px; font-weight:600; color:#64748b; margin-bottom:6px;">
{html.escape(tr('LLM stream window title'))}
</div>
<pre style="white-space:pre-wrap; margin:0; font-size:12px; line-height:1.45;
font-family:ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace;">{escaped_preview}</pre>
</div>
""",
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()

View File

@ -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()