fix: reduce rounded subtitle temp file usage

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

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,15 +463,22 @@ 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)
                 progress_callback(progress, f"合成视频 {batch_idx + 1}/{total_batches}")
 
             # 更新当前视频
             current_video = str(batch_output)
+            current_intermediate = None if is_last_batch else Path(batch_output)
 
         logger.debug("Video synthesis complete")