Improve scan progress reporting

README.md

@@ -18,6 +18,7 @@
 - 使用 SQLite 落盘扫描状态，并基于 `path + size + mtime` 复用旧结果
 - stage1 使用 `ffprobe` 做容器级快速校验
 - stage2 只对可疑文件使用 `ffmpeg` 做定向采样或完整确认
+- 扫描过程中输出 discovery、stage1、stage2 的阶段进度
 - 输出 JSONL / CSV 报告
 - 支持按扫描结果做 `confirmed_broken` 文件删除预览和真实删除
 
@@ -95,7 +96,7 @@ uv run tgvideo-check --help
 .\.venv\Scripts\tgvideo-check.exe scan "D:\small-test" --db ".\small-test.db" --export-jsonl ".\small-test.jsonl" --export-csv ".\small-test.csv"
 ```
 
-如果能正常看到扫描摘要，并生成 `db/jsonl/csv` 三个文件，说明目标机器环境基本没问题。
+如果能正常看到 discovery/stage1/stage2 进度以及最终扫描摘要，并生成 `db/jsonl/csv` 三个文件，说明目标机器环境基本没问题。
 
 ### 基本扫描用法
 

src/tgvideo_check/cli.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections import Counter
+from dataclasses import dataclass, field
 from pathlib import Path
 
 import orjson
@@ -15,7 +16,7 @@ from tgvideo_check.profiles import resolve_config
 from tgvideo_check.reporting import ReportRow, export_csv_summary, export_jsonl
 from tgvideo_check.scheduler import (
     build_stage_plan,
-    execute_stage_batch,
+    iter_stage_batch_results,
     select_pending_stage1_files,
     select_pending_stage2_files,
 )
@@ -27,6 +28,82 @@ app = typer.Typer(
     help="Batch detector for truncated or damaged Telegram-exported videos.",
 )
 
+DISCOVERY_PROGRESS_EVERY = 100
+
+
+@dataclass
+class _StageProgress:
+    label: str
+    total: int
+    counts: Counter[str] = field(default_factory=Counter)
+    last_reported: int = 0
+
+    def record(self, status: str) -> None:
+        self.counts[status] += 1
+        completed = sum(self.counts.values())
+        if not self._should_report(completed):
+            return
+
+        percent = (completed * 100) // self.total if self.total else 100
+        status_text = _format_progress_counts(self.label, self.counts)
+        typer.echo(
+            f"{self.label} progress: {completed}/{self.total} ({percent}%) {status_text}".rstrip()
+        )
+        self.last_reported = completed
+
+    def _should_report(self, completed: int) -> bool:
+        if completed == self.total or completed == 1:
+            return True
+        report_interval = max(1, self.total // 20)
+        return completed - self.last_reported >= report_interval
+
+
+class ScanProgressReporter:
+    def discovery_started(self, roots: tuple[Path, ...]) -> None:
+        typer.echo(f"Discovery: scanning {len(roots)} root(s)")
+
+    def discovery_found(self, discovered_count: int) -> None:
+        if discovered_count == 1 or discovered_count % DISCOVERY_PROGRESS_EVERY == 0:
+            typer.echo(f"Discovery progress: found={discovered_count}")
+
+    def discovery_completed(self, stored_files: list) -> None:
+        pending_stage1 = sum(
+            1 for item in stored_files if item.stage1_status == "pending_stage1"
+        )
+        pending_stage2 = sum(
+            1
+            for item in stored_files
+            if item.stage1_status == "pending_stage2" and item.stage2_status is None
+        )
+        completed = sum(1 for item in stored_files if item.final_status is not None)
+        reused = sum(1 for item in stored_files if item.resumed_from_run_id is not None)
+        typer.echo(
+            "Discovery summary: "
+            f"found={len(stored_files)} reused={reused} "
+            f"pending_stage1={pending_stage1} pending_stage2={pending_stage2} "
+            f"completed={completed}"
+        )
+
+    def stage_started(self, *, stage_name: str, total: int, workers: int) -> _StageProgress | None:
+        label = stage_name.capitalize()
+        if total == 0:
+            typer.echo(f"{label}: no pending files")
+            return None
+        typer.echo(f"{label}: total={total} workers={workers}")
+        return _StageProgress(label=label, total=total)
+
+
+def _format_progress_counts(label: str, counts: Counter[str]) -> str:
+    if label == "Stage1":
+        ordered_keys = ("pending_stage2", "confirmed_broken", "passed_stage1", "command_error")
+    else:
+        ordered_keys = ("passed_stage2", "confirmed_broken", "command_error")
+    return " ".join(
+        f"{key}={counts[key]}"
+        for key in ordered_keys
+        if counts[key]
+    )
+
 
 @app.callback()
 def root() -> None:
@@ -176,32 +253,43 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
     run_id = store.create_run(profile=config.profile, root_paths=config.roots)
     reasons: dict[int, str] = {}
     stage1_payloads: dict[int, dict[str, object]] = {}
+    progress = ScanProgressReporter()
 
     try:
+        progress.discovery_started(config.roots)
         candidates = discover_files(
             config.roots,
             extensions=config.extensions,
             recursive=config.recursive,
+            on_discovered=lambda candidate: progress.discovery_found(
+                candidate.discovery_order + 1
+            ),
         )
         stored_files = store.upsert_files(
             run_id,
             candidates,
             reuse_previous_results=config.resume,
         )
+        progress.discovery_completed(stored_files)
 
+        stage1_files = select_pending_stage1_files(stored_files)
         stage1_batches = build_stage_plan(
-            stage1_files=select_pending_stage1_files(stored_files),
+            stage1_files=stage1_files,
             stage2_files=[],
             stage1_workers=config.stage1_workers or 1,
             stage2_workers=config.stage2_workers or 1,
             queue_limit=config.queue_limit,
         )
+        stage1_progress = progress.stage_started(
+            stage_name="stage1",
+            total=len(stage1_files),
+            workers=config.stage1_workers or 1,
+        )
         for batch in stage1_batches:
-            probe_results = execute_stage_batch(
+            for stored_file, probe_result in iter_stage_batch_results(
                 batch,
                 lambda item: run_ffprobe(item.path),
-            )
+            ):
-            for stored_file, probe_result in zip(batch.files, probe_results):
                 store.record_stage1_result(
                     stored_file.file_id,
                     exit_code=probe_result.exit_code,
@@ -211,24 +299,30 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
                 )
 
                 if probe_result.status != "completed":
+                    stage1_status = "command_error"
                     store.record_stage1_status(
                         stored_file.file_id,
-                        stage1_status="command_error",
+                        stage1_status=stage1_status,
                         final_status="command_error",
                         reason_code=probe_result.status,
                     )
                     reasons[stored_file.file_id] = probe_result.status
+                    if stage1_progress is not None:
+                        stage1_progress.record(stage1_status)
                     continue
 
                 payload = _parse_probe_payload(probe_result.stdout_text)
                 if probe_result.stdout_text and payload is None:
+                    stage1_status = "command_error"
                     store.record_stage1_status(
                         stored_file.file_id,
-                        stage1_status="command_error",
+                        stage1_status=stage1_status,
                         final_status="command_error",
                         reason_code="invalid_probe_output",
                     )
                     reasons[stored_file.file_id] = "invalid_probe_output"
+                    if stage1_progress is not None:
+                        stage1_progress.record(stage1_status)
                     continue
 
                 classification = classify_stage1_result(
@@ -252,17 +346,25 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
                 reasons[stored_file.file_id] = classification.reason_code
                 if payload is not None:
                     stage1_payloads[stored_file.file_id] = payload
+                if stage1_progress is not None:
+                    stage1_progress.record(classification.status)
 
         stored_files = store.list_files(run_id)
+        stage2_files = select_pending_stage2_files(stored_files)
         stage2_batches = build_stage_plan(
             stage1_files=[],
-            stage2_files=select_pending_stage2_files(stored_files),
+            stage2_files=stage2_files,
             stage1_workers=config.stage1_workers or 1,
             stage2_workers=config.stage2_workers or 1,
             queue_limit=config.queue_limit,
         )
+        stage2_progress = progress.stage_started(
+            stage_name="stage2",
+            total=len(stage2_files),
+            workers=config.stage2_workers or 1,
+        )
         for batch in stage2_batches:
-            stage2_outputs = execute_stage_batch(
+            for stored_file, (stage2_status, reason_code, stage2_result) in iter_stage_batch_results(
                 batch,
                 lambda item: _run_stage2_for_file(
                     item.path,
@@ -272,9 +374,6 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
                     sample_duration=config.sample_duration or 3,
                     full_decode_confirm=config.full_decode_confirm,
                 ),
-            )
-            for stored_file, (stage2_status, reason_code, stage2_result) in zip(
-                batch.files, stage2_outputs
             ):
                 store.record_stage2_result(
                     stored_file.file_id,
@@ -292,6 +391,8 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
                     reason_code=reason_code,
                 )
                 reasons[stored_file.file_id] = reason_code
+                if stage2_progress is not None:
+                    stage2_progress.record(stage2_status)
 
         rows = _build_report_rows(store.list_files(run_id), reasons)
         if config.export_jsonl is not None:

src/tgvideo_check/discovery.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from pathlib import Path
 
 from tgvideo_check.config import normalize_extension
@@ -11,6 +12,7 @@ def discover_files(
     *,
     extensions: tuple[str, ...],
     recursive: bool = True,
+    on_discovered: Callable[[FileCandidate], None] | None = None,
 ) -> list[FileCandidate]:
     normalized_extensions = {normalize_extension(item) for item in extensions}
     discovered: list[FileCandidate] = []
@@ -21,6 +23,7 @@ def discover_files(
             recursive=recursive,
             extensions=normalized_extensions,
             discovered=discovered,
+            on_discovered=on_discovered,
         )
 
     return discovered
@@ -32,6 +35,7 @@ def _discover_root(
     recursive: bool,
     extensions: set[str],
     discovered: list[FileCandidate],
+    on_discovered: Callable[[FileCandidate], None] | None,
 ) -> None:
     entries: list[tuple[Path, bool]] = []
     try:
@@ -61,6 +65,7 @@ def _discover_root(
                     recursive=recursive,
                     extensions=extensions,
                     discovered=discovered,
+                    on_discovered=on_discovered,
                 )
             continue
 
@@ -73,12 +78,13 @@ def _discover_root(
         except OSError:
             continue
 
-        discovered.append(
+        candidate = FileCandidate(
-            FileCandidate(
             path=entry,
             size=stat.st_size,
             mtime_ns=stat.st_mtime_ns,
             extension=extension,
             discovery_order=len(discovered),
         )
-        )
+        discovered.append(candidate)
+        if on_discovered is not None:
+            on_discovered(candidate)

src/tgvideo_check/scheduler.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+from collections.abc import Callable, Iterator
 from dataclasses import dataclass
-from concurrent.futures import ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor, as_completed
-from typing import Callable, TypeVar
+from typing import TypeVar
 
 from tgvideo_check.models import StoredFile
 
@@ -80,6 +81,25 @@ def execute_stage_batch(
         return tuple(executor.map(worker, batch.files))
 
 
+def iter_stage_batch_results(
+    batch: StageBatch,
+    worker: Callable[[StoredFile], T],
+) -> Iterator[tuple[StoredFile, T]]:
+    if len(batch.files) <= 1 or batch.worker_count <= 1:
+        for item in batch.files:
+            yield item, worker(item)
+        return
+
+    max_workers = min(batch.worker_count, len(batch.files))
+    with ThreadPoolExecutor(max_workers=max_workers) as executor:
+        futures = {
+            executor.submit(worker, item): item
+            for item in batch.files
+        }
+        for future in as_completed(futures):
+            yield futures[future], future.result()
+
+
 def _chunk_by_limit(
     files: list[StoredFile], queue_limit: int
 ) -> tuple[tuple[StoredFile, ...], ...]:

tests/test_cli.py

@@ -117,6 +117,56 @@ def test_scan_command_runs_stage2_for_pending_files(tmp_path, monkeypatch) -> No
     assert "passed_stage2=1" in result.stdout
 
 
+def test_scan_command_reports_discovery_and_stage_progress(tmp_path, monkeypatch) -> None:
+    root = tmp_path / "videos"
+    root.mkdir()
+    risky = root / "risky.mp4"
+    risky.write_bytes(b"risky")
+
+    def fake_run_ffprobe(path, **kwargs):
+        return ProbeExecutionResult(
+            status="completed",
+            command=("ffprobe", str(path)),
+            exit_code=0,
+            stdout_text='{"format":{"format_name":"mov,mp4,m4a,3gp,3g2,mj2","duration":"20.0"},"streams":[{"codec_type":"video"}]}',
+            stderr_text="",
+            elapsed_ms=1,
+        )
+
+    def fake_run_ffmpeg(path, **kwargs):
+        return FfmpegExecutionResult(
+            status="completed",
+            command=("ffmpeg", str(path)),
+            exit_code=0,
+            stdout_text="frame=10",
+            stderr_text="",
+            elapsed_ms=2,
+            reason_code=None,
+        )
+
+    monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
+    monkeypatch.setattr("tgvideo_check.cli.run_ffmpeg_verification", fake_run_ffmpeg)
+
+    result = runner.invoke(
+        app,
+        [
+            "scan",
+            str(root),
+            "--db",
+            str(tmp_path / "scan_state.db"),
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert "Discovery: scanning 1 root(s)" in result.stdout
+    assert (
+        "Discovery summary: found=1 reused=0 pending_stage1=1 pending_stage2=0 completed=0"
+        in result.stdout
+    )
+    assert "Stage1 progress: 1/1 (100%) pending_stage2=1" in result.stdout
+    assert "Stage2 progress: 1/1 (100%) passed_stage2=1" in result.stdout
+
+
 def test_scan_resume_preserves_reason_code_in_exported_report(tmp_path, monkeypatch) -> None:
     root = tmp_path / "videos"
     root.mkdir()

tests/test_scheduler_hdd.py

@@ -6,6 +6,7 @@ from tgvideo_check.models import StoredFile
 from tgvideo_check.scheduler import (
     build_stage_plan,
     execute_stage_batch,
+    iter_stage_batch_results,
     select_pending_stage1_files,
     select_pending_stage2_files,
 )
@@ -144,3 +145,27 @@ def test_execute_stage_batch_uses_multiple_threads_when_workers_allow() -> None:
     execute_stage_batch(batch, worker)
 
     assert len(thread_ids) >= 2
+
+
+def test_iter_stage_batch_results_yields_items_as_they_finish() -> None:
+    batch = build_stage_plan(
+        stage1_files=[
+            make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
+            make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
+        ],
+        stage2_files=[],
+        stage1_workers=2,
+        stage2_workers=1,
+        queue_limit=64,
+    )[0]
+
+    def worker(item: StoredFile) -> int:
+        if item.file_id == 1:
+            time.sleep(0.05)
+        else:
+            time.sleep(0.01)
+        return item.file_id * 10
+
+    results = list(iter_stage_batch_results(batch, worker))
+
+    assert [(item.file_id, output) for item, output in results] == [(2, 20), (1, 10)]