feat: 增加扫描进度提示

src/video_metadata_analyzer/cli.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import argparse
+import sys
 from datetime import datetime
 from pathlib import Path
 from typing import Sequence
@@ -36,6 +37,21 @@ def _format_probe_error(error: Exception) -> str:
     return message or error.__class__.__name__
 
 
+def _build_progress_bar(current: int, total: int, *, width: int = 24) -> str:
+    if total <= 0:
+        return "[" + ("-" * width) + "]"
+    completed = min(width, round(width * current / total))
+    remaining = width - completed
+    return "[" + ("#" * completed) + ("-" * remaining) + "]"
+
+
+def _emit_scan_progress(current: int, total: int, path: Path) -> None:
+    progress_bar = _build_progress_bar(current, total)
+    message = f"\r扫描进度 {progress_bar} {current}/{total} 当前文件: {path.name}"
+    sys.stderr.write(message)
+    sys.stderr.flush()
+
+
 def main(
     argv: Sequence[str] | None = None,
     *,
@@ -56,10 +72,14 @@ def main(
         ffmpeg_bin=args.ffmpeg_bin,
         runner=command_runner,
     )
+    print("正在递归扫描输入路径，统计待分析的视频文件...", file=sys.stderr)
     video_files = discover_video_files(input_paths)
+    print(f"已发现 {len(video_files)} 个视频文件，开始提取元数据...", file=sys.stderr)
     records: list[VideoMetadata] = []
     failures: list[ProbeFailure] = []
-    for path in video_files:
+    total_files = len(video_files)
+    for index, path in enumerate(video_files, start=1):
+        _emit_scan_progress(index, total_files, path)
         try:
             records.append(
                 probe_video_metadata(
@@ -76,6 +96,9 @@ def main(
                     error_message=_format_probe_error(error),
                 )
             )
+    if total_files:
+        sys.stderr.write("\n")
+        sys.stderr.flush()
 
     summary = build_summary(records, failures=failures)
     paths = write_report_bundle(

tests/test_cli.py

@@ -114,3 +114,55 @@ def test_main_continues_when_single_file_probe_fails(tmp_path: Path) -> None:
             "error_message": "moov atom not found",
         }
     ]
+
+
+def test_main_reports_scan_progress_to_stderr(tmp_path: Path, capsys) -> None:
+    input_dir = tmp_path / "videos"
+    input_dir.mkdir()
+    first_sample = input_dir / "01-first.mp4"
+    second_sample = input_dir / "02-second.mp4"
+    first_sample.write_bytes(b"x" * 2048)
+    second_sample.write_bytes(b"y" * 1024)
+    output_dir = tmp_path / "output"
+
+    def fake_runner(command: list[str]) -> str:
+        if command[:2] == ["ffprobe", "-version"]:
+            return "ffprobe version 8.0.1\nbuilt with ...\n"
+        if command[:2] == ["ffmpeg", "-version"]:
+            return "ffmpeg version 8.0.1\nbuilt with ...\n"
+        if command[0] == "ffprobe" and command[-1] in {str(first_sample), str(second_sample)}:
+            target = Path(command[-1])
+            return json.dumps(
+                {
+                    "format": {
+                        "filename": str(target),
+                        "format_name": "mov,mp4,m4a,3gp,3g2,mj2",
+                        "duration": "8.000000",
+                        "bit_rate": "1024000",
+                        "size": str(target.stat().st_size),
+                    },
+                    "streams": [
+                        {
+                            "codec_type": "video",
+                            "codec_name": "h264",
+                            "width": 1280,
+                            "height": 720,
+                            "avg_frame_rate": "25/1",
+                        }
+                    ],
+                }
+            )
+        raise AssertionError(f"unexpected command: {command}")
+
+    exit_code = main(
+        [str(input_dir), "--output-dir", str(output_dir)],
+        command_runner=fake_runner,
+    )
+
+    captured = capsys.readouterr()
+
+    assert exit_code == 0
+    assert "正在递归扫描输入路径" in captured.err
+    assert "已发现 2 个视频文件" in captured.err
+    assert "2/2" in captured.err
+    assert "02-second.mp4" in captured.err