kim/clean-broken-tgvideos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Separate stage1 scan from stage2 verification

Browse Source
This commit is contained in:
kim
2026-03-30 15:53:53 +08:00
parent 7e74de3e08
commit cec0920f0c
9 changed files with 420 additions and 108 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/tgvideo_check/cli.py
+131 -64
View File
@@ -153,14 +153,11 @@ def scan(
db_path: Path = typer.Option(Path("scan_state.db"), "--db"),
resume: bool = typer.Option(True, "--resume/--no-resume"),
stage1_workers: int | None = typer.Option(None, "--stage1-workers"),
stage2_workers: int | None = typer.Option(None, "--stage2-workers"),
queue_limit: int = typer.Option(64, "--queue-limit"),
sample_duration: int | None = typer.Option(None, "--sample-duration"),
full_decode_confirm: bool = typer.Option(False, "--full-decode-confirm"),
export_jsonl_path: Path | None = typer.Option(None, "--export-jsonl"),
export_csv_path: Path | None = typer.Option(None, "--export-csv"),
) -> None:
"""Scan directories for suspicious video files."""
"""Run the first-pass ffprobe scan and leave stage2 pending."""
config = resolve_config(
ScanConfig(
@@ -171,6 +168,53 @@ def scan(
db_path=db_path,
resume=resume,
stage1_workers=stage1_workers,
queue_limit=queue_limit,
export_jsonl=export_jsonl_path,
export_csv=export_csv_path,
)
)
rows = _execute_stage1_scan(config)
typer.echo(f"Scanned {len(rows)} file(s). {_format_summary_text(rows)}".strip())
@app.command("verify-stage2")
def verify_stage2(
db_path: Path = typer.Option(Path("scan_state.db"), "--db"),
run_id: int | None = typer.Option(None, "--run-id"),
profile: str | None = typer.Option(
None,
"--profile",
help="Override the stored profile when resolving stage2 defaults.",
),
stage2_workers: int | None = typer.Option(None, "--stage2-workers"),
queue_limit: int = typer.Option(64, "--queue-limit"),
sample_duration: int | None = typer.Option(None, "--sample-duration"),
full_decode_confirm: bool = typer.Option(False, "--full-decode-confirm"),
export_jsonl_path: Path | None = typer.Option(None, "--export-jsonl"),
export_csv_path: Path | None = typer.Option(None, "--export-csv"),
) -> None:
"""Run the second-pass ffmpeg verification for pending_stage2 files."""
store = ScanStore(db_path)
store.initialize()
selected_run_id = run_id or store.get_latest_run_id()
if selected_run_id is None:
raise typer.BadParameter(f"No scan runs found in database: {db_path}")
scan_run = store.get_run(selected_run_id)
if scan_run is None:
raise typer.BadParameter(f"Scan run does not exist: {selected_run_id}")
typer.echo(
f"Stage2 source: run_id={selected_run_id} profile={scan_run.profile}"
)
resolved_profile = profile or scan_run.profile
config = resolve_config(
ScanConfig(
roots=(),
profile=resolved_profile,
db_path=db_path,
stage2_workers=stage2_workers,
queue_limit=queue_limit,
sample_duration=sample_duration,
@@ -180,12 +224,14 @@ def scan(
)
)
rows = _execute_scan(config)
summary = Counter(row.final_status for row in rows)
summary_text = " ".join(
f"{status}={summary[status]}" for status in sorted(summary.keys())
rows = _execute_stage2_verification(
store,
run_id=selected_run_id,
config=config,
)
typer.echo(
f"Verified stage2 for run {selected_run_id}. {_format_summary_text(rows)}".strip()
)
typer.echo(f"Scanned {len(rows)} file(s). {summary_text}".strip())
@app.command()
@@ -265,12 +311,11 @@ def main() -> None:
app()
def _execute_scan(config: ScanConfig) -> list[ReportRow]:
def _execute_stage1_scan(config: ScanConfig) -> list[ReportRow]:
store = ScanStore(config.db_path)
store.initialize()
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:
@@ -287,6 +332,7 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
run_id,
candidates,
reuse_previous_results=config.resume,
reuse_stage2_results=False,
)
progress.discovery_completed(stored_files)
@@ -368,62 +414,9 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
reason_code=classification.reason_code,
)
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=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:
for stored_file, (stage2_status, reason_code, stage2_result) in iter_stage_batch_results(
batch,
lambda item: _run_stage2_for_file(
item.path,
payload=stage1_payloads.get(item.file_id)
or _parse_probe_payload(store.get_stage1_stdout_text(item.file_id) or "")
or _reload_probe_payload(item.path),
sample_duration=config.sample_duration or 3,
full_decode_confirm=config.full_decode_confirm,
),
heartbeat_seconds=PROGRESS_HEARTBEAT_SECONDS,
on_heartbeat=(
(lambda unfinished: stage2_progress.heartbeat(unfinished))
if stage2_progress is not None
else None
),
):
store.record_stage2_result(
stored_file.file_id,
mode=stage2_result["mode"],
sample_points_json=stage2_result["sample_points_json"],
exit_code=stage2_result["exit_code"],
stderr_text=stage2_result["stderr_text"],
reason_code=reason_code,
elapsed_ms=stage2_result["elapsed_ms"],
)
store.record_stage2_status(
stored_file.file_id,
stage2_status=stage2_status,
final_status=stage2_status,
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:
export_jsonl(rows, config.export_jsonl)
@@ -437,6 +430,73 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
raise
def _execute_stage2_verification(
store: ScanStore,
*,
run_id: int,
config: ScanConfig,
) -> list[ReportRow]:
reasons: dict[int, str] = {}
progress = ScanProgressReporter()
stored_files = store.list_files(run_id)
stage2_files = select_pending_stage2_files(stored_files)
stage2_batches = build_stage_plan(
stage1_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:
for stored_file, (stage2_status, reason_code, stage2_result) in iter_stage_batch_results(
batch,
lambda item: _run_stage2_for_file(
item.path,
payload=_parse_probe_payload(store.get_stage1_stdout_text(item.file_id) or "")
or _reload_probe_payload(item.path),
sample_duration=config.sample_duration or 3,
full_decode_confirm=config.full_decode_confirm,
),
heartbeat_seconds=PROGRESS_HEARTBEAT_SECONDS,
on_heartbeat=(
(lambda unfinished: stage2_progress.heartbeat(unfinished))
if stage2_progress is not None
else None
),
):
store.record_stage2_result(
stored_file.file_id,
mode=stage2_result["mode"],
sample_points_json=stage2_result["sample_points_json"],
exit_code=stage2_result["exit_code"],
stderr_text=stage2_result["stderr_text"],
reason_code=reason_code,
elapsed_ms=stage2_result["elapsed_ms"],
)
store.record_stage2_status(
stored_file.file_id,
stage2_status=stage2_status,
final_status=stage2_status,
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:
export_jsonl(rows, config.export_jsonl)
if config.export_csv is not None:
export_csv_summary(rows, config.export_csv)
return rows
def _build_report_rows(
stored_files: list,
reasons: dict[int, str],
@@ -456,6 +516,13 @@ def _build_report_rows(
return rows
def _format_summary_text(rows: list[ReportRow]) -> str:
summary = Counter(row.final_status for row in rows)
return " ".join(
f"{status}={summary[status]}" for status in sorted(summary.keys())
)
def _run_stage2_for_file(
path: Path,
*,
src/tgvideo_check/reporting.py
+1
View File
@@ -18,6 +18,7 @@ class ReportRow:
FINAL_STATUS_ZH: dict[str, str] = {
"confirmed_broken": "明确损坏",
"suspicious": "可疑",
"pending_stage2": "待第二轮复核",
"passed_stage1": "第一轮通过",
"passed_stage2": "第二轮通过",
"skipped": "已跳过",
src/tgvideo_check/store.py
+46 -6
View File
@@ -124,11 +124,17 @@ class ScanStore:
candidates: list[FileCandidate],
*,
reuse_previous_results: bool = True,
reuse_stage2_results: bool = True,
) -> list[StoredFile]:
with self._connect() as connection:
for candidate in candidates:
previous = (
self._find_previous_match(connection, run_id, candidate)
self._find_previous_match(
connection,
run_id,
candidate,
allow_stage2_match=reuse_stage2_results,
)
if reuse_previous_results
else None
)
@@ -192,6 +198,7 @@ class ScanStore:
connection,
previous_file_id=previous["file_id"],
current_file_id=current_file_id,
copy_stage2_results=reuse_stage2_results,
)
rows = connection.execute(
@@ -218,6 +225,26 @@ class ScanStore:
return self._rows_to_stored_files(rows)
def get_run(self, run_id: int) -> ScanRun | None:
with self._connect() as connection:
row = connection.execute(
"""
SELECT id, profile, root_paths_json, status
FROM scan_runs
WHERE id = ?
""",
(run_id,),
).fetchone()
if row is None:
return None
return ScanRun(
run_id=int(row["id"]),
profile=row["profile"],
root_paths_json=row["root_paths_json"],
status=row["status"],
)
def record_stage1_status(
self,
file_id: int,
@@ -400,17 +427,27 @@ class ScanStore:
connection: sqlite3.Connection,
run_id: int,
candidate: FileCandidate,
*,
allow_stage2_match: bool,
) -> sqlite3.Row | None:
return connection.execute(
"""
query = """
SELECT id AS file_id, run_id, stage1_status, stage2_status, final_status, reason_code
FROM files
WHERE path = ? AND size = ? AND mtime_ns = ? AND run_id <> ?
"""
params: tuple[object, ...] = (
str(candidate.path),
candidate.size,
candidate.mtime_ns,
run_id,
)
if not allow_stage2_match:
query += " AND stage2_status IS NULL"
query += """
ORDER BY run_id DESC
LIMIT 1
""",
(str(candidate.path), candidate.size, candidate.mtime_ns, run_id),
).fetchone()
"""
return connection.execute(query, params).fetchone()
def _copy_previous_results(
self,
@@ -418,6 +455,7 @@ class ScanStore:
*,
previous_file_id: int,
current_file_id: int,
copy_stage2_results: bool,
) -> None:
connection.execute(
"""
@@ -433,6 +471,8 @@ class ScanStore:
""",
(current_file_id, previous_file_id),
)
if not copy_stage2_results:
return
connection.execute(
"""
INSERT INTO stage2_results(