diff --git a/tests/test_subtitle/test_rounded_renderer.py b/tests/test_subtitle/test_rounded_renderer.py new file mode 100644 index 00000000..39677cc7 --- /dev/null +++ b/tests/test_subtitle/test_rounded_renderer.py @@ -0,0 +1,113 @@ +"""Tests for rounded subtitle video rendering.""" + +import subprocess +from pathlib import Path + +from PIL import Image + +from videocaptioner.core.asr.asr_data import ASRData, ASRDataSeg +from videocaptioner.core.entities import SubtitleLayoutEnum +from videocaptioner.core.subtitle import rounded_renderer + + +class FixedTemporaryDirectory: + """TemporaryDirectory replacement that leaves files available for assertions.""" + + def __init__(self, path: Path): + self.path = path + + def __enter__(self): + self.path.mkdir(parents=True, exist_ok=True) + return str(self.path) + + def __exit__(self, exc_type, exc, tb): + return False + + +def make_asr_data(count: int) -> ASRData: + segments = [ + ASRDataSeg( + text=f"source {index}", + translated_text=f"target {index}", + start_time=index * 1000, + end_time=index * 1000 + 500, + ) + for index in range(count) + ] + return ASRData(segments) + + +def stub_subtitle_image(*args, **kwargs): + return Image.new("RGBA", (8, 8), (0, 0, 0, 0)) + + +def test_rounded_video_intermediate_batches_use_bounded_encoding( + monkeypatch, tmp_path: Path +) -> None: + commands: list[list[str]] = [] + + def fake_run(cmd, **kwargs): + commands.append(cmd) + Path(cmd[-1]).write_bytes(b"video") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(rounded_renderer, "_get_video_info", lambda path: (320, 180, 3.0)) + monkeypatch.setattr(rounded_renderer, "render_subtitle_image", stub_subtitle_image) + monkeypatch.setattr( + rounded_renderer.tempfile, + "TemporaryDirectory", + lambda *args, **kwargs: FixedTemporaryDirectory(tmp_path / "rounded-temp"), + ) + monkeypatch.setattr(rounded_renderer.subprocess, "run", fake_run) + + output_path = tmp_path / "output.mp4" + rounded_renderer.render_rounded_video( + video_path=str(tmp_path / "input.mp4"), + asr_data=make_asr_data(51), + output_path=str(output_path), + layout=SubtitleLayoutEnum.ONLY_TRANSLATE, + crf=28, + preset="medium", + ) + + assert len(commands) == 2 + first_cmd = commands[0] + assert first_cmd[first_cmd.index("-crf") + 1] == "18" + assert first_cmd[first_cmd.index("-preset") + 1] == "ultrafast" + assert "-hide_banner" in first_cmd + assert first_cmd[first_cmd.index("-loglevel") + 1] == "error" + + final_cmd = commands[-1] + assert final_cmd[final_cmd.index("-crf") + 1] == "28" + assert final_cmd[final_cmd.index("-preset") + 1] == "medium" + + +def test_rounded_video_cleans_processed_temp_files(monkeypatch, tmp_path: Path) -> None: + temp_path = tmp_path / "rounded-temp" + + def fake_run(cmd, **kwargs): + Path(cmd[-1]).write_bytes(b"video") + return subprocess.CompletedProcess(cmd, 0, "", "") + + monkeypatch.setattr(rounded_renderer, "_get_video_info", lambda path: (320, 180, 3.0)) + monkeypatch.setattr(rounded_renderer, "render_subtitle_image", stub_subtitle_image) + monkeypatch.setattr( + rounded_renderer.tempfile, + "TemporaryDirectory", + lambda *args, **kwargs: FixedTemporaryDirectory(temp_path), + ) + monkeypatch.setattr(rounded_renderer.subprocess, "run", fake_run) + + output_path = tmp_path / "output.mp4" + rounded_renderer.render_rounded_video( + video_path=str(tmp_path / "input.mp4"), + asr_data=make_asr_data(101), + output_path=str(output_path), + layout=SubtitleLayoutEnum.ONLY_TRANSLATE, + crf=28, + preset="medium", + ) + + assert output_path.exists() + assert not list(temp_path.glob("batch_*.mp4")) + assert not list(temp_path.glob("subtitle_*.png")) diff --git a/videocaptioner/core/subtitle/rounded_renderer.py b/videocaptioner/core/subtitle/rounded_renderer.py index b9fda706..addaf038 100644 --- a/videocaptioner/core/subtitle/rounded_renderer.py +++ b/videocaptioner/core/subtitle/rounded_renderer.py @@ -23,6 +23,13 @@ logger = setup_logger("subtitle.rounded") +def _tail_text(text: str, limit: int = 6000) -> str: + """Keep FFmpeg errors useful without dumping minutes of progress output.""" + if len(text) <= limit: + return text + return f"...(truncated, showing last {limit} chars)\n{text[-limit:]}" + + def _get_video_info(video_path: str) -> Tuple[int, int, float]: """获取视频分辨率和时长""" result = subprocess.run( @@ -375,7 +382,11 @@ def render_rounded_video( logger.debug("Overlaying subtitle batches onto video") BATCH_SIZE = 50 current_video = video_path + current_intermediate: Optional[Path] = None total_batches = (len(subtitle_frames) + BATCH_SIZE - 1) // BATCH_SIZE + # Keep intermediate files bounded; the final batch still uses requested quality. + intermediate_crf = "18" + intermediate_preset = "ultrafast" for batch_idx in range(total_batches): start_idx = batch_idx * BATCH_SIZE @@ -403,12 +414,16 @@ def render_rounded_video( batch_output = ( output_path if is_last_batch else temp_path / f"batch_{batch_idx:03d}.mp4" ) + previous_intermediate = current_intermediate logger.debug(f"Processing batch {batch_idx + 1}/{total_batches}（{len(batch_frames)}个字幕）") # 构建 ffmpeg Command # -t 参数强制保持原视频时长，防止因 overlay ended而截断视频 cmd = [ "ffmpeg", + "-hide_banner", + "-loglevel", + "error", "-y", *input_args, "-filter_complex", @@ -422,9 +437,9 @@ def render_rounded_video( "-c:v", "libx264", "-preset", - "ultrafast" if not is_last_batch else preset, + intermediate_preset if not is_last_batch else preset, "-crf", - "0" if not is_last_batch else str(crf), + intermediate_crf if not is_last_batch else str(crf), "-pix_fmt", "yuv420p", "-c:a", @@ -448,9 +463,15 @@ def render_rounded_video( ) if result.returncode != 0: - logger.error(f"批次 {batch_idx + 1} 失败: {result.stderr}") + logger.error(f"批次 {batch_idx + 1} 失败: {_tail_text(result.stderr)}") raise RuntimeError(f"Subtitle processing failed（批次 {batch_idx + 1}）") + if previous_intermediate: + previous_intermediate.unlink(missing_ok=True) + + for _, _, png_path in batch_frames: + png_path.unlink(missing_ok=True) + # 更新进度 (30-100%) if progress_callback: progress = 30 + int((batch_idx + 1) / total_batches * 70) @@ -458,5 +479,6 @@ def render_rounded_video( # 更新当前视频 current_video = str(batch_output) + current_intermediate = None if is_last_batch else Path(batch_output) logger.debug("Video synthesis complete")