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

Improve progress reporting and video format coverage

Browse Source
This commit is contained in:
kim
2026-03-30 15:24:46 +08:00
parent 6125a9389c
commit 7e74de3e08
8 changed files with 244 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+14 -2
View File
@@ -13,12 +13,12 @@
### 当前能力
- 递归扫描一个或多个目录中的常见视频扩展名
- 递归扫描一个或多个目录中的常见视频扩展名，默认覆盖 `mp4/mov/m4v/mkv/webm/avi/flv/f4v/ts/m2ts/mts/mpg/mpeg/mpe/mpv/m2v/wmv/asf/3gp/3g2/ogv/ogm/vob/qt/rm/rmvb/mxf/divx`
- 支持从 `txt` 文件批量导入目录列表
- 使用 SQLite 落盘扫描状态，并基于 `path + size + mtime` 复用旧结果
- stage1 使用 `ffprobe` 做容器级快速校验
- stage2 只对可疑文件使用 `ffmpeg` 做定向采样或完整确认
- 扫描过程中输出 discovery、stage1、stage2 的阶段进度
- 扫描过程中输出 discovery、stage1、stage2 的阶段进度；长耗时阶段会持续输出 heartbeat
- 输出 JSONL / CSV 报告
- 支持按扫描结果做 `confirmed_broken` 文件删除预览和真实删除
@@ -124,10 +124,22 @@ tgvideo-check scan "D:\extra-folder" --paths-file .\dirs.txt --db .\scan_state.d
- `--paths-file`：从 txt 文件读取目录列表，一行一个目录
- `--export-jsonl` / `--export-csv`：导出结果报告
- `--stage1-workers` / `--stage2-workers`：覆盖 profile 默认 worker 数
- `--extensions`：覆盖默认视频扩展名列表
- `--queue-limit`：限制单阶段批次大小，避免把待处理文件一次性堆进内存
- `--full-decode-confirm`：对 stage2 强制使用完整解码确认
- `--resume / --no-resume`：控制是否复用旧结果
默认扩展名列表：
```text
mp4,mov,m4v,mkv,webm,avi,flv,f4v,ts,m2ts,mts,mpg,mpeg,mpe,mpv,m2v,wmv,asf,3gp,3g2,ogv,ogm,vob,qt,rm,rmvb,mxf,divx
```
说明：
- `tg-hdd` / `tg-fast` 仍保持 Telegram 常见容器的保守高风险复核范围，避免因为格式表扩大导致 stage2 backlog 暴涨
- 如果你希望对全部已支持格式都走更激进的二次校验，使用 `--profile deep-verify`
### `--paths-file` 规则
- 按 `UTF-8` / `UTF-8 BOM` 读取
src/tgvideo_check/cli.py
+33 -3
View File
@@ -8,10 +8,15 @@ import orjson
import typer
from tgvideo_check.classifiers import classify_stage1_result
from tgvideo_check.config import ScanConfig, normalize_extension
from tgvideo_check.config import (
DEFAULT_VIDEO_EXTENSIONS,
ScanConfig,
normalize_extension,
)
from tgvideo_check.discovery import discover_files
from tgvideo_check.ffmpeg_runner import generate_sample_offsets, run_ffmpeg_verification
from tgvideo_check.ffprobe_runner import run_ffprobe
from tgvideo_check.models import StoredFile
from tgvideo_check.profiles import resolve_config
from tgvideo_check.reporting import ReportRow, export_csv_summary, export_jsonl
from tgvideo_check.scheduler import (
@@ -29,6 +34,9 @@ app = typer.Typer(
)
DISCOVERY_PROGRESS_EVERY = 100
PROGRESS_HEARTBEAT_SECONDS = 15.0
PROGRESS_COUNT_MAX_INTERVAL = 25
DEFAULT_VIDEO_EXTENSIONS_CSV = ",".join(DEFAULT_VIDEO_EXTENSIONS)
@dataclass
@@ -54,9 +62,19 @@ class _StageProgress:
def _should_report(self, completed: int) -> bool:
if completed == self.total or completed == 1:
return True
report_interval = max(1, self.total // 20)
report_interval = min(PROGRESS_COUNT_MAX_INTERVAL, max(1, self.total // 20))
return completed - self.last_reported >= report_interval
def heartbeat(self, unfinished: tuple[StoredFile, ...]) -> None:
completed = sum(self.counts.values())
percent = (completed * 100) // self.total if self.total else 100
remaining = self.total - completed
summary_text = _format_progress_counts(self.label, self.counts) or "no_result_yet"
typer.echo(
f"{self.label} heartbeat: {completed}/{self.total} ({percent}%) "
f"remaining={remaining} batch_unfinished={len(unfinished)} {summary_text}".rstrip()
)
class ScanProgressReporter:
def discovery_started(self, roots: tuple[Path, ...]) -> None:
@@ -128,7 +146,7 @@ def scan(
profile: str = typer.Option("tg-hdd", "--profile"),
recursive: bool = typer.Option(True, "--recursive/--no-recursive"),
extensions: str = typer.Option(
"mp4,mov,m4v,mkv,webm,avi",
DEFAULT_VIDEO_EXTENSIONS_CSV,
"--extensions",
help="Comma-separated video extensions.",
),
@@ -289,6 +307,12 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
for stored_file, probe_result in iter_stage_batch_results(
batch,
lambda item: run_ffprobe(item.path),
heartbeat_seconds=PROGRESS_HEARTBEAT_SECONDS,
on_heartbeat=(
(lambda unfinished: stage1_progress.heartbeat(unfinished))
if stage1_progress is not None
else None
),
):
store.record_stage1_result(
stored_file.file_id,
@@ -374,6 +398,12 @@ def _execute_scan(config: ScanConfig) -> list[ReportRow]:
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,
src/tgvideo_check/config.py
+33 -1
View File
@@ -4,6 +4,38 @@ from dataclasses import dataclass, replace
from pathlib import Path
DEFAULT_VIDEO_EXTENSIONS: tuple[str, ...] = (
"mp4",
"mov",
"m4v",
"mkv",
"webm",
"avi",
"flv",
"f4v",
"ts",
"m2ts",
"mts",
"mpg",
"mpeg",
"mpe",
"mpv",
"m2v",
"wmv",
"asf",
"3gp",
"3g2",
"ogv",
"ogm",
"vob",
"qt",
"rm",
"rmvb",
"mxf",
"divx",
)
def normalize_extension(value: str) -> str:
return value.lower().lstrip(".")
@@ -13,7 +45,7 @@ class ScanConfig:
roots: tuple[Path, ...]
profile: str = "tg-hdd"
recursive: bool = True
extensions: tuple[str, ...] = ("mp4", "mov", "m4v", "mkv", "webm", "avi")
extensions: tuple[str, ...] = DEFAULT_VIDEO_EXTENSIONS
db_path: Path = Path("scan_state.db")
resume: bool = True
stage1_workers: int | None = None
src/tgvideo_check/profiles.py
+7 -4
View File
@@ -2,7 +2,10 @@ from __future__ import annotations
from dataclasses import dataclass, replace
from tgvideo_check.config import ScanConfig
from tgvideo_check.config import DEFAULT_VIDEO_EXTENSIONS, ScanConfig
TG_HIGH_RISK_EXTENSIONS: tuple[str, ...] = ("mp4", "mov", "m4v", "mkv", "webm")
@dataclass(frozen=True)
@@ -20,21 +23,21 @@ PROFILES: dict[str, Profile] = {
stage1_workers=1,
stage2_workers=1,
sample_duration=3,
high_risk_extensions=("mp4", "mov", "m4v", "mkv", "webm"),
high_risk_extensions=TG_HIGH_RISK_EXTENSIONS,
),
"tg-fast": Profile(
name="tg-fast",
stage1_workers=2,
stage2_workers=1,
sample_duration=3,
high_risk_extensions=("mp4", "mov", "m4v", "mkv", "webm"),
high_risk_extensions=TG_HIGH_RISK_EXTENSIONS,
),
"deep-verify": Profile(
name="deep-verify",
stage1_workers=1,
stage2_workers=1,
sample_duration=5,
high_risk_extensions=("mp4", "mov", "m4v", "mkv", "webm", "avi"),
high_risk_extensions=DEFAULT_VIDEO_EXTENSIONS,
),
}
src/tgvideo_check/scheduler.py
+36 -5
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from collections.abc import Callable, Iterator
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, as_completed, wait
from typing import TypeVar
from tgvideo_check.models import StoredFile
@@ -84,20 +84,51 @@ def execute_stage_batch(
def iter_stage_batch_results(
batch: StageBatch,
worker: Callable[[StoredFile], T],
*,
heartbeat_seconds: float | None = None,
on_heartbeat: Callable[[tuple[StoredFile, ...]], None] | None = None,
) -> Iterator[tuple[StoredFile, T]]:
if len(batch.files) <= 1 or batch.worker_count <= 1:
use_heartbeat = (
on_heartbeat is not None
and heartbeat_seconds is not None
and heartbeat_seconds > 0
)
if not use_heartbeat and (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))
max_workers = min(max(1, 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()
if not use_heartbeat:
for future in as_completed(futures):
yield futures[future], future.result()
return
pending = set(futures)
while pending:
done, pending = wait(
pending,
timeout=heartbeat_seconds,
return_when=FIRST_COMPLETED,
)
if not done:
pending_items = tuple(
sorted(
(futures[future] for future in pending),
key=lambda item: item.discovery_order,
)
)
on_heartbeat(pending_items)
continue
for future in done:
yield futures[future], future.result()
def _chunk_by_limit(
tests/test_cli.py
+48
View File
@@ -1,3 +1,5 @@
import time
from typer.testing import CliRunner
from tgvideo_check.cli import app
@@ -167,6 +169,52 @@ def test_scan_command_reports_discovery_and_stage_progress(tmp_path, monkeypatch
assert "Stage2 progress: 1/1 (100%) passed_stage2=1" in result.stdout
def test_scan_command_reports_stage2_heartbeat_while_waiting(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
for index in range(60):
(root / f"risky-{index:03d}.mp4").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):
time.sleep(0.02)
return FfmpegExecutionResult(
status="completed",
command=("ffmpeg", str(path)),
exit_code=0,
stdout_text="frame=10",
stderr_text="",
elapsed_ms=20,
reason_code=None,
)
monkeypatch.setattr("tgvideo_check.cli.PROGRESS_HEARTBEAT_SECONDS", 0.01)
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 "Stage2 heartbeat:" 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_profiles.py
+41 -2
View File
@@ -1,4 +1,4 @@
from tgvideo_check.config import ScanConfig
from tgvideo_check.config import DEFAULT_VIDEO_EXTENSIONS, ScanConfig
from tgvideo_check.profiles import get_profile, resolve_config
@@ -12,6 +12,46 @@ def test_tg_hdd_profile_uses_conservative_defaults() -> None:
assert profile.high_risk_extensions == ("mp4", "mov", "m4v", "mkv", "webm")
def test_default_extensions_cover_broad_common_video_formats() -> None:
assert DEFAULT_VIDEO_EXTENSIONS == (
"mp4",
"mov",
"m4v",
"mkv",
"webm",
"avi",
"flv",
"f4v",
"ts",
"m2ts",
"mts",
"mpg",
"mpeg",
"mpe",
"mpv",
"m2v",
"wmv",
"asf",
"3gp",
"3g2",
"ogv",
"ogm",
"vob",
"qt",
"rm",
"rmvb",
"mxf",
"divx",
)
assert ScanConfig(roots=()).extensions == DEFAULT_VIDEO_EXTENSIONS
def test_deep_verify_profile_treats_all_supported_formats_as_high_risk() -> None:
profile = get_profile("deep-verify")
assert profile.high_risk_extensions == DEFAULT_VIDEO_EXTENSIONS
def test_resolve_config_applies_profile_overrides() -> None:
config = resolve_config(
ScanConfig(
@@ -26,4 +66,3 @@ def test_resolve_config_applies_profile_overrides() -> None:
assert config.stage1_workers == 3
assert config.stage2_workers == 1
assert config.extensions == ("mp4", "mkv")
tests/test_scheduler_hdd.py
+32
View File
@@ -169,3 +169,35 @@ def test_iter_stage_batch_results_yields_items_as_they_finish() -> None:
results = list(iter_stage_batch_results(batch, worker))
assert [(item.file_id, output) for item, output in results] == [(2, 20), (1, 10)]
def test_iter_stage_batch_results_can_emit_heartbeat_while_waiting() -> None:
batch = build_stage_plan(
stage1_files=[
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
],
stage2_files=[],
stage1_workers=1,
stage2_workers=1,
queue_limit=64,
)[0]
heartbeats: list[tuple[int, ...]] = []
def worker(item: StoredFile) -> int:
time.sleep(0.03)
return item.file_id
results = list(
iter_stage_batch_results(
batch,
worker,
heartbeat_seconds=0.01,
on_heartbeat=lambda items: heartbeats.append(
tuple(entry.file_id for entry in items)
),
)
)
assert results == [(batch.files[0], 1)]
assert heartbeats
assert heartbeats[0] == (1,)