From ca4f2bf59455df5faf03f3b5f1d635a41f79a9c7 Mon Sep 17 00:00:00 2001 From: viccy Date: Sun, 7 Jun 2026 19:05:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=E8=84=9A=E6=9C=AC?= =?UTF-8?q?=E8=87=AA=E5=8A=A8=E5=AD=97=E5=B9=95=E7=94=9F=E6=88=90=E5=8A=9F?= =?UTF-8?q?=E8=83=BD=E5=B9=B6=E4=BC=98=E5=8C=96=E7=BD=91=E9=A1=B5=E8=A7=86?= =?UTF-8?q?=E9=A2=91=E5=B1=95=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 添加script_subtitle服务，支持基于脚本内容自动生成标准SRT字幕文件 修改任务处理流程，优先使用新的脚本字幕生成逻辑，失败时回退至原TTS字幕合并方案 优化最终视频自动转录逻辑，已生成脚本字幕时跳过重复的自动转录步骤 改进网页端弹窗视频的展示样式，根据宽高比调整预览宽度并添加黑色背景 新增完整的单元测试覆盖字幕生成相关功能 --- app/services/script_subtitle.py | 213 ++++++++++++++++++ app/services/task.py | 31 ++- app/services/test_script_subtitle_unittest.py | 94 ++++++++ webui.py | 15 +- 4 files changed, 348 insertions(+), 5 deletions(-) create mode 100644 app/services/script_subtitle.py create mode 100644 app/services/test_script_subtitle_unittest.py diff --git a/app/services/script_subtitle.py b/app/services/script_subtitle.py new file mode 100644 index 0000000..2259580 --- /dev/null +++ b/app/services/script_subtitle.py @@ -0,0 +1,213 @@ +import os +import re +import unicodedata +from typing import Iterable, List, Optional, Sequence, Tuple + +from loguru import logger + +from app.utils import utils + + +DEFAULT_SUBTITLE_OST_TYPES = (0, 2) +DEFAULT_MAX_CHARS_PER_SUBTITLE = 12 +SENTENCE_PART_RE = re.compile(r"[^。！？!?；;，,、\n]+[。！？!?；;，,、]?") + + +def _normalize_text(text: str) -> str: + return re.sub(r"\s+", " ", str(text or "")).strip() + + +def _remove_punctuation(text: str) -> str: + return "".join( + char for char in str(text or "") + if not unicodedata.category(char).startswith("P") + ) + + +def clean_subtitle_text(text: str) -> str: + """Normalize subtitle text for burn-in display.""" + return _normalize_text(_remove_punctuation(text)) + + +def split_narration(text: str, max_chars: int = DEFAULT_MAX_CHARS_PER_SUBTITLE) -> List[str]: + """Split narration into readable subtitle chunks.""" + text = _normalize_text(text) + if not text: + return [] + + max_chars = max(1, int(max_chars or DEFAULT_MAX_CHARS_PER_SUBTITLE)) + parts = [match.group(0).strip() for match in SENTENCE_PART_RE.finditer(text)] + if not parts: + parts = [text] + + chunks = [] + current = "" + + def flush_long_part(part: str) -> str: + while len(part) > max_chars: + chunks.append(part[:max_chars].strip()) + part = part[max_chars:].strip() + return part + + for part in parts: + if not part: + continue + + if len(part) > max_chars: + if current: + chunks.append(current.strip()) + current = "" + current = flush_long_part(part) + continue + + candidate = f"{current}{part}" if current else part + if len(candidate) <= max_chars: + current = candidate + else: + if current: + chunks.append(current.strip()) + current = part + + if current: + chunks.append(current.strip()) + + return [cleaned for chunk in chunks if (cleaned := clean_subtitle_text(chunk))] + + +def parse_srt_like_time(time_text: str) -> float: + time_text = str(time_text or "").strip().replace(",", ".") + parts = time_text.split(":") + if len(parts) != 3: + raise ValueError(f"不支持的时间格式: {time_text}") + + hours = int(parts[0]) + minutes = int(parts[1]) + seconds = float(parts[2]) + return hours * 3600 + minutes * 60 + seconds + + +def parse_time_range(time_range: str) -> Tuple[float, float]: + if not time_range or "-" not in str(time_range): + raise ValueError(f"不支持的时间范围: {time_range}") + + start_text, end_text = str(time_range).split("-", 1) + start = parse_srt_like_time(start_text) + end = parse_srt_like_time(end_text) + if end <= start: + raise ValueError(f"结束时间必须晚于开始时间: {time_range}") + + return start, end + + +def format_srt_time(seconds: float) -> str: + milliseconds_total = max(0, int(round(float(seconds) * 1000))) + milliseconds = milliseconds_total % 1000 + total_seconds = milliseconds_total // 1000 + hours = total_seconds // 3600 + minutes = (total_seconds % 3600) // 60 + secs = total_seconds % 60 + return f"{hours:02d}:{minutes:02d}:{secs:02d},{milliseconds:03d}" + + +def _safe_ost_value(value) -> Optional[int]: + try: + return int(value) + except (TypeError, ValueError): + return None + + +def _resolve_item_time_range(item: dict, current_time: float) -> Tuple[Optional[Tuple[float, float]], float]: + edited_time_range = item.get("editedTimeRange") + if edited_time_range: + try: + start, end = parse_time_range(edited_time_range) + return (start, end), end + except ValueError as e: + logger.warning(f"解析 editedTimeRange 失败，将尝试使用 duration: {e}") + + duration = float(item.get("duration", 0.0) or 0.0) + if duration <= 0: + return None, current_time + + start = current_time + end = current_time + duration + return (start, end), end + + +def _build_srt_blocks( + list_script: Sequence[dict], + include_ost: Iterable[int], + max_chars: int, +) -> List[str]: + include_ost_set = {int(item) for item in include_ost} + blocks = [] + subtitle_index = 1 + current_time = 0.0 + + for item in list_script: + time_range, current_time = _resolve_item_time_range(item, current_time) + if not time_range: + continue + + ost = _safe_ost_value(item.get("OST")) + if ost not in include_ost_set: + continue + + chunks = split_narration(item.get("narration", ""), max_chars=max_chars) + if not chunks: + continue + + start, end = time_range + segment_duration = end - start + if segment_duration <= 0: + continue + + chunk_duration = segment_duration / len(chunks) + for chunk_index, chunk in enumerate(chunks): + chunk_start = start + chunk_duration * chunk_index + chunk_end = end if chunk_index == len(chunks) - 1 else start + chunk_duration * (chunk_index + 1) + blocks.append( + "\n".join( + [ + str(subtitle_index), + f"{format_srt_time(chunk_start)} --> {format_srt_time(chunk_end)}", + chunk, + ] + ) + ) + subtitle_index += 1 + + return blocks + + +def create_script_subtitle_file( + task_id: str, + list_script: Sequence[dict], + output_file: Optional[str] = None, + include_ost: Optional[Iterable[int]] = None, + max_chars: int = DEFAULT_MAX_CHARS_PER_SUBTITLE, +) -> str: + """Create a full SRT file from script narration and edited timeline ranges.""" + if not list_script: + return "" + + if include_ost is None: + include_ost = DEFAULT_SUBTITLE_OST_TYPES + + blocks = _build_srt_blocks(list_script, include_ost=include_ost, max_chars=max_chars) + if not blocks: + logger.warning("程序化字幕未生成内容") + return "" + + if output_file is None: + output_file = os.path.join(utils.task_dir(task_id), "script_subtitles.srt") + + output_dir = os.path.dirname(output_file) + if output_dir: + os.makedirs(output_dir, exist_ok=True) + with open(output_file, "w", encoding="utf-8") as f: + f.write("\n\n".join(blocks)) + f.write("\n") + + logger.info(f"程序化字幕生成成功: {output_file}, 共 {len(blocks)} 条") + return output_file diff --git a/app/services/task.py b/app/services/task.py index 356b7a6..f5f60ce 100644 --- a/app/services/task.py +++ b/app/services/task.py @@ -10,7 +10,16 @@ from app.config import config from app.config.audio_config import AudioConfig, get_recommended_volumes_for_content from app.models import const from app.models.schema import VideoClipParams -from app.services import (voice, audio_merger, subtitle_merger, clip_video, merger_video, update_script, generate_video) +from app.services import ( + voice, + audio_merger, + subtitle_merger, + clip_video, + merger_video, + update_script, + generate_video, + script_subtitle, +) from app.services import state as sm from app.utils import utils @@ -561,8 +570,20 @@ def start_subclip_unified(task_id: str, params: VideoClipParams): ) logger.info(f"音频文件合并成功->{merged_audio_path}") - # 合并字幕文件 - merged_subtitle_path = subtitle_merger.merge_subtitle_files(new_script_list) + # 优先基于脚本文案和成片时间线生成字幕，失败时回退到TTS字幕合并 + merged_subtitle_path = "" + if getattr(params, "subtitle_enabled", True): + try: + merged_subtitle_path = script_subtitle.create_script_subtitle_file( + task_id=task_id, + list_script=new_script_list, + ) + except Exception as e: + logger.warning(f"程序化字幕生成失败，将尝试合并TTS字幕: {e}") + + if not merged_subtitle_path and getattr(params, "subtitle_enabled", True): + merged_subtitle_path = subtitle_merger.merge_subtitle_files(new_script_list) + if merged_subtitle_path: logger.info(f"字幕文件合并成功->{merged_subtitle_path}") else: @@ -630,7 +651,9 @@ def start_subclip_unified(task_id: str, params: VideoClipParams): 6. 合并字幕/BGM/配音/视频 """ output_video_path = path.join(utils.task_dir(task_id), f"combined.mp4") - auto_transcription_enabled = _is_auto_transcription_enabled(params) + auto_transcription_enabled = _is_auto_transcription_enabled(params) and not bool(merged_subtitle_path) + if _is_auto_transcription_enabled(params) and merged_subtitle_path: + logger.info("已生成字幕文件，跳过最终视频自动转录") merge_output_video_path = ( path.join(utils.task_dir(task_id), "combined_without_auto_subtitles.mp4") if auto_transcription_enabled diff --git a/app/services/test_script_subtitle_unittest.py b/app/services/test_script_subtitle_unittest.py new file mode 100644 index 0000000..a4eed37 --- /dev/null +++ b/app/services/test_script_subtitle_unittest.py @@ -0,0 +1,94 @@ +import tempfile +import unittest +from pathlib import Path + +from app.services import script_subtitle + + +class ScriptSubtitleTests(unittest.TestCase): + def test_split_narration_prefers_punctuation_boundaries(self): + chunks = script_subtitle.split_narration( + "她终于意识到，这场婚姻不是爱情，而是一场交易。", + max_chars=12, + ) + + self.assertEqual( + ["她终于意识到", "这场婚姻不是爱情", "而是一场交易"], + chunks, + ) + + def test_time_range_parsing_supports_milliseconds(self): + start, end = script_subtitle.parse_time_range("00:00:01,500-00:00:03,250") + + self.assertAlmostEqual(1.5, start) + self.assertAlmostEqual(3.25, end) + + def test_create_script_subtitle_file_skips_original_audio_segments(self): + list_script = [ + { + "_id": 1, + "OST": 0, + "narration": "第一句解说。第二句解说。", + "editedTimeRange": "00:00:00-00:00:04", + "duration": 4, + }, + { + "_id": 2, + "OST": 1, + "narration": "这句是原声，不应该默认生成。", + "editedTimeRange": "00:00:04-00:00:08", + "duration": 4, + }, + { + "_id": 3, + "OST": 2, + "narration": "混合片段也保留解说字幕。", + "editedTimeRange": "00:00:08-00:00:12", + "duration": 4, + }, + ] + + with tempfile.TemporaryDirectory() as temp_dir: + output_file = Path(temp_dir) / "script_subtitles.srt" + result = script_subtitle.create_script_subtitle_file( + task_id="test", + list_script=list_script, + output_file=str(output_file), + max_chars=16, + ) + + self.assertEqual(str(output_file), result) + content = output_file.read_text(encoding="utf-8") + + self.assertIn("00:00:00,000 -->", content) + self.assertIn("第一句解说", content) + self.assertIn("混合片段也保留解说字幕", content) + self.assertNotIn("这句是原声", content) + self.assertNotIn("。", content) + self.assertNotIn("，", content) + + def test_create_script_subtitle_file_uses_duration_when_edited_range_missing(self): + list_script = [ + { + "_id": 1, + "OST": 0, + "narration": "没有 editedTimeRange 时使用 duration。", + "duration": 3, + } + ] + + with tempfile.TemporaryDirectory() as temp_dir: + output_file = Path(temp_dir) / "script_subtitles.srt" + script_subtitle.create_script_subtitle_file( + task_id="test", + list_script=list_script, + output_file=str(output_file), + ) + content = output_file.read_text(encoding="utf-8") + + self.assertIn("00:00:00,000 -->", content) + self.assertIn("--> 00:00:03,000", content) + + +if __name__ == "__main__": + unittest.main() diff --git a/webui.py b/webui.py index 3eac2eb..34d1204 100644 --- a/webui.py +++ b/webui.py @@ -181,6 +181,11 @@ def render_generate_button(): div[data-testid="stDialog"] div[data-testid="stProgress"] { margin-bottom: 0.75rem; } + div[data-testid="stDialog"] video { + max-height: 62vh; + object-fit: contain; + background: #000; + } """, unsafe_allow_html=True, @@ -246,8 +251,16 @@ def render_generate_button(): video_files = task.get("videos", []) try: if video_files: + aspect = getattr(params, "video_aspect", "") + aspect = getattr(aspect, "value", aspect) + preview_width = 320 if aspect in { + VideoAspect.portrait.value, + VideoAspect.portrait_2.value, + } else 600 for url in video_files: - st.video(url) + _, preview_col, _ = st.columns([1, 2, 1]) + with preview_col: + st.video(url, width=preview_width) except Exception as e: logger.error(f"播放视频失败: {e}")