Add inspect-stage1 debug command

README.md

@@ -148,6 +148,7 @@ tgvideo-check scan "D:\extra-folder" --paths-file .\dirs.txt --db .\scan_state.d
 
 - `scan`：只做第一轮快速筛查
 - `verify-stage2`：只做第二轮详细复核
+- `inspect-stage1`：排查第一轮崩溃前最后写入的 `ffprobe` 原始输出
 
 默认扩展名列表：
 
@@ -212,6 +213,27 @@ tgvideo-check verify-stage2 --db ".\small-test.db" --export-jsonl ".\small-test-
 | `skipped` | 已跳过 | 预留状态 |
 | `command_error` | 命令异常 | `ffprobe` / `ffmpeg` 执行失败，不应直接视为媒体损坏 |
 
+### Stage1 崩溃排查
+
+如果 `scan` 在第一轮中途抛异常，例如卡在某个文件的 `ffprobe` JSON 解析阶段，不需要手动打开 SQLite 客户端。直接用内建命令查看最近一次扫描里，已经写入 `stage1_results` 但仍停在 `pending_stage1` 的记录：
+
+```powershell
+tgvideo-check inspect-stage1 --db "C:\Users\Administrator\Documents\clean-tgvideo-data\scan_state.db"
+```
+
+常用变体：
+
+- `--run-id 12`：指定要检查的扫描批次
+- `--limit 20`：多看几条最近记录
+- `--all`：如果默认模式没有命中，就直接查看最近保存的 Stage1 原始结果，不再只限 `pending_stage1`
+
+输出重点：
+
+- `path=`：最值得怀疑的文件路径
+- `stdout_len=`：`ffprobe` 标准输出长度；很短通常表示只吐出了半截 JSON
+- `stdout_head=`：标准输出开头预览，便于快速看出是不是残缺 JSON
+- `stderr_head=`：标准错误预览，常能直接看到 `moov atom not found`、`error reading header` 等真实媒体错误
+
 ### 原因码说明
 
 导出的 `reason_code` / `reason_code_zh` 含义如下：

src/tgvideo_check/cli.py

@@ -234,6 +234,58 @@ def verify_stage2(
     )
 
 
+@app.command("inspect-stage1")
+def inspect_stage1(
+    db_path: Path = typer.Option(Path("scan_state.db"), "--db"),
+    run_id: int | None = typer.Option(None, "--run-id"),
+    limit: int = typer.Option(
+        10,
+        "--limit",
+        min=1,
+        help="Maximum number of saved stage1 probe rows to print.",
+    ),
+    all_rows: bool = typer.Option(
+        False,
+        "--all",
+        help="Show recent stage1 rows even if stage1_status is no longer pending_stage1.",
+    ),
+) -> None:
+    """Inspect saved stage1 probe rows to locate the file that likely crashed scan."""
+
+    store = ScanStore(db_path)
+    store.initialize()
+    selected_run_id = run_id or store.get_latest_run_id(prefer_completed=False)
+    if selected_run_id is None:
+        raise typer.BadParameter(f"No scan runs found in database: {db_path}")
+
+    rows = store.list_stage1_debug_rows(
+        selected_run_id,
+        limit=limit,
+        only_pending=not all_rows,
+    )
+    mode = "all" if all_rows else "pending_only"
+    typer.echo(
+        f"Stage1 inspect: run_id={selected_run_id} mode={mode} matched={len(rows)} limit={limit}"
+    )
+    if not rows:
+        if not all_rows:
+            typer.echo("Hint: re-run with --all to inspect the latest saved stage1 rows.")
+        return
+
+    for index, row in enumerate(rows, start=1):
+        typer.echo(
+            f"[{index}] order={row.discovery_order} "
+            f"stage1_status={row.stage1_status} "
+            f"final_status={row.final_status or '-'} "
+            f"reason_code={row.reason_code or '-'} "
+            f"exit_code={row.exit_code} "
+            f"stdout_len={row.stdout_len}"
+        )
+        typer.echo(f"path={row.path}")
+        typer.echo(f"stdout_head={row.stdout_head or '<empty>'}")
+        typer.echo(f"stderr_head={row.stderr_head or '<empty>'}")
+
+
 @app.command()
 def prune(
     db_path: Path = typer.Option(Path("scan_state.db"), "--db"),

src/tgvideo_check/models.py

@@ -27,3 +27,18 @@ class StoredFile:
     final_status: str | None
     resumed_from_run_id: int | None
     reason_code: str | None = None
+
+
+@dataclass(frozen=True)
+class Stage1DebugRow:
+    file_id: int
+    run_id: int
+    path: Path
+    discovery_order: int
+    stage1_status: str
+    final_status: str | None
+    reason_code: str | None
+    exit_code: int | None
+    stdout_len: int
+    stdout_head: str
+    stderr_head: str

src/tgvideo_check/store.py

@@ -6,7 +6,7 @@ from pathlib import Path
 
 import orjson
 
-from tgvideo_check.models import FileCandidate, StoredFile
+from tgvideo_check.models import FileCandidate, Stage1DebugRow, StoredFile
 
 
 @dataclass(frozen=True)
@@ -398,6 +398,59 @@ class ScanStore:
 
         return self._rows_to_stored_files(rows)
 
+    def list_stage1_debug_rows(
+        self,
+        run_id: int,
+        *,
+        limit: int = 10,
+        only_pending: bool = True,
+    ) -> list[Stage1DebugRow]:
+        query = """
+            SELECT
+                f.id,
+                f.run_id,
+                f.path,
+                f.discovery_order,
+                f.stage1_status,
+                f.final_status,
+                f.reason_code,
+                r.exit_code,
+                r.stdout_text,
+                r.stderr_text
+            FROM files AS f
+            JOIN stage1_results AS r ON r.file_id = f.id
+            WHERE f.run_id = ?
+        """
+        params: list[object] = [run_id]
+        if only_pending:
+            query += " AND f.stage1_status = ?"
+            params.append("pending_stage1")
+        query += """
+            ORDER BY f.discovery_order DESC
+            LIMIT ?
+        """
+        params.append(limit)
+
+        with self._connect() as connection:
+            rows = connection.execute(query, tuple(params)).fetchall()
+
+        return [
+            Stage1DebugRow(
+                file_id=row["id"],
+                run_id=row["run_id"],
+                path=Path(row["path"]),
+                discovery_order=row["discovery_order"],
+                stage1_status=row["stage1_status"],
+                final_status=row["final_status"],
+                reason_code=row["reason_code"],
+                exit_code=row["exit_code"],
+                stdout_len=len(row["stdout_text"] or ""),
+                stdout_head=_preview_text(row["stdout_text"]),
+                stderr_head=_preview_text(row["stderr_text"]),
+            )
+            for row in rows
+        ]
+
     def _rows_to_stored_files(self, rows: list[sqlite3.Row]) -> list[StoredFile]:
         return [
             StoredFile(
@@ -514,3 +567,12 @@ class ScanStore:
         connection.execute(
             f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_definition}"
         )
+
+
+def _preview_text(value: str | None, *, max_length: int = 160) -> str:
+    if not value:
+        return ""
+    normalized = value.replace("\r", "\\r").replace("\n", "\\n")
+    if len(normalized) <= max_length:
+        return normalized
+    return normalized[: max_length - 3] + "..."

tests/test_cli.py

@@ -1,21 +1,107 @@
 import time
+from pathlib import Path
 
 from typer.testing import CliRunner
 
 from tgvideo_check.cli import app
 from tgvideo_check.ffmpeg_runner import FfmpegExecutionResult
 from tgvideo_check.ffprobe_runner import ProbeExecutionResult
+from tgvideo_check.models import FileCandidate
+from tgvideo_check.store import ScanStore
 
 
 runner = CliRunner()
 
 
+def _make_candidate(path: Path, *, order: int, size: int = 10, mtime_ns: int = 100) -> FileCandidate:
+    path.write_bytes(b"x" * size)
+    return FileCandidate(
+        path=path,
+        size=size,
+        mtime_ns=mtime_ns,
+        extension=path.suffix.lower().lstrip("."),
+        discovery_order=order,
+    )
+
+
 def test_help_shows_scan_command() -> None:
     result = runner.invoke(app, ["--help"])
 
     assert result.exit_code == 0
     assert "scan" in result.stdout
     assert "verify-stage2" in result.stdout
+    assert "inspect-stage1" in result.stdout
+
+
+def test_inspect_stage1_command_reports_pending_stage1_probe_rows(tmp_path: Path) -> None:
+    store = ScanStore(tmp_path / "scan_state.db")
+    store.initialize()
+    run_id = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
+
+    candidate = _make_candidate(tmp_path / "broken.mp4", order=0)
+    [stored] = store.upsert_files(run_id, [candidate])
+    store.record_stage1_result(
+        stored.file_id,
+        exit_code=1,
+        stdout_text='{\n',
+        stderr_text="moov atom not found",
+        elapsed_ms=12,
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "inspect-stage1",
+            "--db",
+            str(store.db_path),
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert f"Stage1 inspect: run_id={run_id} mode=pending_only matched=1" in result.stdout
+    assert "path=" in result.stdout
+    assert "broken.mp4" in result.stdout
+    assert "stdout_len=2" in result.stdout
+    assert r"stdout_head={\n" in result.stdout
+    assert "stderr_head=moov atom not found" in result.stdout
+
+
+def test_inspect_stage1_command_can_show_recent_rows_with_all_flag(tmp_path: Path) -> None:
+    store = ScanStore(tmp_path / "scan_state.db")
+    store.initialize()
+    run_id = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
+
+    candidate = _make_candidate(tmp_path / "passed.avi", order=0)
+    [stored] = store.upsert_files(run_id, [candidate])
+    store.record_stage1_result(
+        stored.file_id,
+        exit_code=0,
+        stdout_text='{"format":{"format_name":"avi"}}',
+        stderr_text="soft warning",
+        elapsed_ms=8,
+    )
+    store.record_stage1_status(
+        stored.file_id,
+        stage1_status="passed_stage1",
+        final_status="passed_stage1",
+        reason_code="passed_stage1",
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "inspect-stage1",
+            "--db",
+            str(store.db_path),
+            "--all",
+        ],
+    )
+
+    assert result.exit_code == 0
+    assert f"Stage1 inspect: run_id={run_id} mode=all matched=1" in result.stdout
+    assert "stage1_status=passed_stage1" in result.stdout
+    assert "reason_code=passed_stage1" in result.stdout
+    assert "stderr_head=soft warning" in result.stdout
 
 
 def test_scan_command_processes_stage1_results_and_exports_reports(