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

Fix resume DB growth and add compaction command

Browse Source
This commit is contained in:
kim
2026-03-30 19:55:29 +08:00
parent 834577dafc
commit 6a599ae5d4
5 changed files with 586 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+30
View File
@@ -16,6 +16,7 @@
- 递归扫描一个或多个目录中的常见视频扩展名，默认覆盖 `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` 复用旧结果
- 当同一组 `roots + profile` 的扫描上次未完成时，`scan --resume` 会直接续跑同一个 run，不再把 discovery 结果整轮复制一遍
- stage1 使用 `ffprobe` 做容器级快速校验
- `scan` 默认只执行 stage1，把需要详细复核的文件标记为 `pending_stage2`
- `verify-stage2` 只对 `pending_stage2` 文件使用 `ffmpeg` 做定向采样或完整确认
@@ -143,6 +144,7 @@ tgvideo-check scan "D:\extra-folder" --paths-file .\dirs.txt --db .\scan_state.d
- `--queue-limit`：限制单阶段批次大小，避免把待处理文件一次性堆进内存
- `verify-stage2 --full-decode-confirm`：对第二轮强制使用完整解码确认
- `scan --resume / --no-resume`：控制是否复用旧的第一轮结果
- `compact-db`：清理同一组 `roots + profile` 下的重复历史 run，并可选执行 `VACUUM`
命令边界：
@@ -234,6 +236,34 @@ tgvideo-check inspect-stage1 --db "C:\Users\Administrator\Documents\clean-tgvide
- `stdout_head=`：标准输出开头预览，便于快速看出是不是残缺 JSON
- `stderr_head=`：标准错误预览，常能直接看到 `moov atom not found`、`error reading header` 等真实媒体错误
### 压缩重复扫描数据库
如果你之前在同一批目录上反复 `scan`，导致同一组 `roots + profile` 下面堆积了多份历史 run，可以保留一个目标 run，把其余重复 run 删掉：
先预览：
```powershell
tgvideo-check compact-db --db .\scan_state.db
```
真实执行：
```powershell
tgvideo-check compact-db --db .\scan_state.db --no-dry-run
```
如果要明确保留某个 run：
```powershell
tgvideo-check compact-db --db .\scan_state.db --run-id 12 --no-dry-run
```
说明：
- 默认保留最新 run；如果你在做中断恢复，通常就保留当前正在续跑的那个 run
- 这个命令只清理同一组 `roots + profile` 下的重复历史 run，不会顺手删除别的扫描任务
- 默认会在真实删除后执行 `VACUUM`，把 SQLite 文件物理收缩下来
### 原因码说明
导出的 `reason_code` / `reason_code_zh` 含义如下：
src/tgvideo_check/cli.py
+60
View File
@@ -359,6 +359,54 @@ def prune(
typer.echo(f"Reasons: {reason_text}")
@app.command("compact-db")
def compact_db(
db_path: Path = typer.Option(Path("scan_state.db"), "--db"),
run_id: int | None = typer.Option(
None,
"--run-id",
help="Keep this run and remove duplicate runs with the same roots/profile.",
),
dry_run: bool = typer.Option(
True,
"--dry-run/--no-dry-run",
help="Preview duplicate run cleanup without deleting rows.",
),
vacuum: bool = typer.Option(
True,
"--vacuum/--no-vacuum",
help="Run SQLite VACUUM after deleting duplicate runs.",
),
) -> None:
"""Compact duplicate scan runs for the same roots/profile lineage."""
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}")
try:
summary = store.compact_duplicate_runs(
selected_run_id,
dry_run=dry_run,
vacuum=vacuum,
)
except ValueError as exc:
raise typer.BadParameter(str(exc)) from exc
typer.echo(
"Compact-db: "
f"keep_run_id={summary.keep_run_id} "
f"deleted_runs={summary.deleted_run_count} "
f"deleted_files={summary.deleted_file_count} "
f"deleted_stage1_results={summary.deleted_stage1_count} "
f"deleted_stage2_results={summary.deleted_stage2_count} "
f"dry_run={'yes' if summary.dry_run else 'no'} "
f"vacuumed={'yes' if summary.vacuumed else 'no'}"
)
def main() -> None:
app()
@@ -366,11 +414,23 @@ def main() -> None:
def _execute_stage1_scan(config: ScanConfig) -> list[ReportRow]:
store = ScanStore(config.db_path)
store.initialize()
resumed_existing_run = False
run_id = (
store.find_resumable_run_id(profile=config.profile, root_paths=config.roots)
if config.resume
else None
)
if run_id is None:
run_id = store.create_run(profile=config.profile, root_paths=config.roots)
else:
resumed_existing_run = True
store.update_run_status(run_id, status="running")
reasons: dict[int, str] = {}
progress = ScanProgressReporter()
try:
if resumed_existing_run:
typer.echo(f"Resume: reusing unfinished run_id={run_id}")
progress.discovery_started(config.roots)
candidates = discover_files(
config.roots,
src/tgvideo_check/store.py
+246 -2
View File
@@ -17,6 +17,21 @@ class ScanRun:
status: str
@dataclass(frozen=True)
class RunCompactionSummary:
keep_run_id: int
duplicate_run_ids: tuple[int, ...]
deleted_file_count: int
deleted_stage1_count: int
deleted_stage2_count: int
dry_run: bool
vacuumed: bool
@property
def deleted_run_count(self) -> int:
return len(self.duplicate_run_ids)
class ScanStore:
def __init__(self, db_path: Path) -> None:
self.db_path = db_path
@@ -79,7 +94,7 @@ class ScanStore:
self._ensure_column(connection, "files", "reason_code", "TEXT")
def create_run(self, *, profile: str, root_paths: tuple[Path, ...]) -> int:
root_paths_json = orjson.dumps([str(path) for path in root_paths]).decode("utf-8")
root_paths_json = self._serialize_root_paths(root_paths)
with self._connect() as connection:
cursor = connection.execute(
"""
@@ -90,6 +105,29 @@ class ScanStore:
)
return int(cursor.lastrowid)
def find_resumable_run_id(
self,
*,
profile: str,
root_paths: tuple[Path, ...],
) -> int | None:
root_paths_json = self._serialize_root_paths(root_paths)
with self._connect() as connection:
row = connection.execute(
"""
SELECT id
FROM scan_runs
WHERE profile = ? AND root_paths_json = ? AND status <> 'completed'
ORDER BY id DESC
LIMIT 1
""",
(profile, root_paths_json),
).fetchone()
if row is None:
return None
return int(row["id"])
def get_latest_run_id(self, *, prefer_completed: bool = True) -> int | None:
with self._connect() as connection:
if prefer_completed:
@@ -128,6 +166,20 @@ class ScanStore:
) -> list[StoredFile]:
with self._connect() as connection:
for candidate in candidates:
current = self._find_current_file(connection, run_id, candidate.path)
current_matches = (
current is not None
and self._row_matches_candidate(current, candidate)
)
previous = None
if current_matches:
stage1_status = current["stage1_status"]
stage2_status = current["stage2_status"]
final_status = current["final_status"]
resumed_from_run_id = current["resumed_from_run_id"]
reason_code = current["reason_code"]
else:
previous = (
self._find_previous_match(
connection,
@@ -185,7 +237,11 @@ class ScanStore:
reason_code,
),
)
current_file_id = connection.execute(
current_file_id = (
int(current["file_id"])
if current is not None
else int(
connection.execute(
"""
SELECT id
FROM files
@@ -193,6 +249,12 @@ class ScanStore:
""",
(run_id, str(candidate.path)),
).fetchone()["id"]
)
)
if current is not None and not current_matches:
self._clear_file_results(connection, current_file_id)
if previous is not None:
self._copy_previous_results(
connection,
@@ -451,6 +513,128 @@ class ScanStore:
for row in rows
]
def compact_duplicate_runs(
self,
keep_run_id: int,
*,
dry_run: bool = True,
vacuum: bool = True,
) -> RunCompactionSummary:
keep_run = self.get_run(keep_run_id)
if keep_run is None:
raise ValueError(f"Scan run does not exist: {keep_run_id}")
duplicate_run_ids: tuple[int, ...]
deleted_file_count = 0
deleted_stage1_count = 0
deleted_stage2_count = 0
with self._connect() as connection:
duplicate_rows = connection.execute(
"""
SELECT id
FROM scan_runs
WHERE profile = ? AND root_paths_json = ? AND id <> ?
ORDER BY id ASC
""",
(keep_run.profile, keep_run.root_paths_json, keep_run_id),
).fetchall()
duplicate_run_ids = tuple(int(row["id"]) for row in duplicate_rows)
if duplicate_run_ids:
placeholders = ",".join("?" for _ in duplicate_run_ids)
deleted_file_count = int(
connection.execute(
f"""
SELECT COUNT(*)
FROM files
WHERE run_id IN ({placeholders})
""",
duplicate_run_ids,
).fetchone()[0]
)
deleted_stage1_count = int(
connection.execute(
f"""
SELECT COUNT(*)
FROM stage1_results
WHERE file_id IN (
SELECT id
FROM files
WHERE run_id IN ({placeholders})
)
""",
duplicate_run_ids,
).fetchone()[0]
)
deleted_stage2_count = int(
connection.execute(
f"""
SELECT COUNT(*)
FROM stage2_results
WHERE file_id IN (
SELECT id
FROM files
WHERE run_id IN ({placeholders})
)
""",
duplicate_run_ids,
).fetchone()[0]
)
if not dry_run:
connection.execute(
f"""
DELETE FROM stage1_results
WHERE file_id IN (
SELECT id
FROM files
WHERE run_id IN ({placeholders})
)
""",
duplicate_run_ids,
)
connection.execute(
f"""
DELETE FROM stage2_results
WHERE file_id IN (
SELECT id
FROM files
WHERE run_id IN ({placeholders})
)
""",
duplicate_run_ids,
)
connection.execute(
f"""
DELETE FROM files
WHERE run_id IN ({placeholders})
""",
duplicate_run_ids,
)
connection.execute(
f"""
DELETE FROM scan_runs
WHERE id IN ({placeholders})
""",
duplicate_run_ids,
)
vacuumed = bool(duplicate_run_ids) and not dry_run and vacuum
if vacuumed:
with self._connect() as connection:
connection.execute("VACUUM")
return RunCompactionSummary(
keep_run_id=keep_run_id,
duplicate_run_ids=duplicate_run_ids,
deleted_file_count=deleted_file_count,
deleted_stage1_count=deleted_stage1_count,
deleted_stage2_count=deleted_stage2_count,
dry_run=dry_run,
vacuumed=vacuumed,
)
def _rows_to_stored_files(self, rows: list[sqlite3.Row]) -> list[StoredFile]:
return [
StoredFile(
@@ -475,6 +659,46 @@ class ScanStore:
connection.row_factory = sqlite3.Row
return connection
def _serialize_root_paths(self, root_paths: tuple[Path, ...]) -> str:
return orjson.dumps([str(path) for path in root_paths]).decode("utf-8")
def _find_current_file(
self,
connection: sqlite3.Connection,
run_id: int,
path: Path,
) -> sqlite3.Row | None:
return connection.execute(
"""
SELECT
id AS file_id,
run_id,
path,
size,
mtime_ns,
extension,
stage1_status,
stage2_status,
final_status,
resumed_from_run_id,
reason_code
FROM files
WHERE run_id = ? AND path = ?
""",
(run_id, str(path)),
).fetchone()
def _row_matches_candidate(
self,
row: sqlite3.Row,
candidate: FileCandidate,
) -> bool:
return (
row["size"] == candidate.size
and row["mtime_ns"] == candidate.mtime_ns
and row["extension"] == candidate.extension
)
def _find_previous_match(
self,
connection: sqlite3.Connection,
@@ -502,6 +726,26 @@ class ScanStore:
"""
return connection.execute(query, params).fetchone()
def _clear_file_results(
self,
connection: sqlite3.Connection,
file_id: int,
) -> None:
connection.execute(
"""
DELETE FROM stage1_results
WHERE file_id = ?
""",
(file_id,),
)
connection.execute(
"""
DELETE FROM stage2_results
WHERE file_id = ?
""",
(file_id,),
)
def _copy_previous_results(
self,
connection: sqlite3.Connection,
tests/test_cli.py
+131
View File
@@ -1,3 +1,4 @@
import sqlite3
import time
from pathlib import Path
@@ -490,6 +491,66 @@ def test_scan_resume_preserves_reason_code_in_exported_report(tmp_path, monkeypa
assert '"reason_code":"ffprobe_exit_nonzero"' in second_jsonl.read_text(encoding="utf-8")
def test_scan_resume_reuses_incomplete_run_without_duplication(tmp_path: Path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
video = root / "video.avi"
video.write_bytes(b"video")
db_path = tmp_path / "scan_state.db"
store = ScanStore(db_path)
store.initialize()
run_id = store.create_run(profile="tg-hdd", root_paths=(root,))
candidate = FileCandidate(
path=video,
size=video.stat().st_size,
mtime_ns=video.stat().st_mtime_ns,
extension="avi",
discovery_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","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
store.record_stage1_status(
stored.file_id,
stage1_status="passed_stage1",
final_status="passed_stage1",
reason_code="passed_stage1",
)
def fail_ffprobe(*args, **kwargs):
raise AssertionError("resume within same run should not call ffprobe")
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fail_ffprobe)
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert result.exit_code == 0
assert "passed_stage1=1" in result.stdout
with sqlite3.connect(db_path) as connection:
run_count = connection.execute("SELECT COUNT(*) FROM scan_runs").fetchone()[0]
file_count = connection.execute("SELECT COUNT(*) FROM files").fetchone()[0]
stage1_count = connection.execute("SELECT COUNT(*) FROM stage1_results").fetchone()[0]
assert run_count == 1
assert file_count == 1
assert stage1_count == 1
def test_scan_resume_does_not_reuse_previous_stage2_results(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
@@ -581,6 +642,76 @@ def test_scan_resume_does_not_reuse_previous_stage2_results(tmp_path, monkeypatc
assert probe_calls == 1
def test_compact_db_command_keeps_selected_run_and_removes_duplicate_runs(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
video = root / "video.mp4"
video.write_bytes(b"video")
db_path = tmp_path / "scan_state.db"
store = ScanStore(db_path)
store.initialize()
candidate = FileCandidate(
path=video,
size=video.stat().st_size,
mtime_ns=video.stat().st_mtime_ns,
extension="mp4",
discovery_order=0,
)
first_run = store.create_run(profile="tg-hdd", root_paths=(root,))
[first_stored] = store.upsert_files(first_run, [candidate])
store.record_stage1_result(
first_stored.file_id,
exit_code=0,
stdout_text='{"format":{"format_name":"mp4","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
store.record_stage1_status(
first_stored.file_id,
stage1_status="passed_stage1",
final_status="passed_stage1",
reason_code="passed_stage1",
)
store.update_run_status(first_run, status="failed")
second_run = store.create_run(profile="tg-hdd", root_paths=(root,))
[second_stored] = store.upsert_files(second_run, [candidate])
store.update_run_status(second_run, status="running")
before_size = db_path.stat().st_size
result = runner.invoke(
app,
[
"compact-db",
"--db",
str(db_path),
"--run-id",
str(second_run),
"--no-dry-run",
],
)
assert result.exit_code == 0
assert f"keep_run_id={second_run}" in result.stdout
assert "deleted_runs=1" in result.stdout
with sqlite3.connect(db_path) as connection:
run_count = connection.execute("SELECT COUNT(*) FROM scan_runs").fetchone()[0]
file_count = connection.execute("SELECT COUNT(*) FROM files").fetchone()[0]
stage1_count = connection.execute("SELECT COUNT(*) FROM stage1_results").fetchone()[0]
remaining_run_id = connection.execute("SELECT id FROM scan_runs").fetchone()[0]
assert second_stored.file_id != first_stored.file_id
assert run_count == 1
assert file_count == 1
assert stage1_count == 1
assert remaining_run_id == second_run
assert db_path.stat().st_size <= before_size
def test_scan_no_resume_forces_probe_execution(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
tests/test_resume_logic.py
+97
View File
@@ -132,3 +132,100 @@ def test_upsert_files_does_not_reuse_changed_file(tmp_path: Path) -> None:
assert fresh.stage1_status == "pending_stage1"
assert fresh.final_status is None
assert fresh.resumed_from_run_id is None
def test_upsert_files_same_run_preserves_existing_status_without_duplication(
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,))
file_path = tmp_path / "clip.mp4"
candidate = make_candidate(file_path, order=0, size=12, mtime_ns=999)
[stored] = store.upsert_files(run_id, [candidate])
store.record_stage1_result(
stored.file_id,
exit_code=0,
stdout_text='{"format":{"format_name":"mp4","duration":"2.0"}}',
stderr_text="",
elapsed_ms=10,
)
store.record_stage1_status(
stored.file_id,
stage1_status="passed_stage1",
final_status="passed_stage1",
reason_code="passed_stage1",
)
[resumed] = store.upsert_files(run_id, [candidate])
assert resumed.file_id == stored.file_id
assert resumed.stage1_status == "passed_stage1"
assert resumed.final_status == "passed_stage1"
assert resumed.resumed_from_run_id is None
with sqlite3.connect(store.db_path) as connection:
file_count = connection.execute(
"SELECT COUNT(*) FROM files WHERE run_id = ?",
(run_id,),
).fetchone()[0]
stage1_count = connection.execute(
"SELECT COUNT(*) FROM stage1_results WHERE file_id = ?",
(stored.file_id,),
).fetchone()[0]
assert file_count == 1
assert stage1_count == 1
def test_upsert_files_same_run_changed_file_clears_stale_stage_results(
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,))
file_path = tmp_path / "clip.mp4"
original = make_candidate(file_path, order=0, size=12, mtime_ns=999)
[stored] = store.upsert_files(run_id, [original])
store.record_stage1_result(
stored.file_id,
exit_code=0,
stdout_text='{"format":{"format_name":"mp4","duration":"2.0"}}',
stderr_text="",
elapsed_ms=10,
)
store.record_stage1_status(
stored.file_id,
stage1_status="passed_stage1",
final_status="passed_stage1",
reason_code="passed_stage1",
)
changed = FileCandidate(
path=file_path,
size=13,
mtime_ns=1000,
extension="mp4",
discovery_order=0,
)
[refreshed] = store.upsert_files(run_id, [changed])
assert refreshed.file_id == stored.file_id
assert refreshed.stage1_status == "pending_stage1"
assert refreshed.final_status is None
assert refreshed.reason_code is None
with sqlite3.connect(store.db_path) as connection:
stage1_count = connection.execute(
"SELECT COUNT(*) FROM stage1_results WHERE file_id = ?",
(stored.file_id,),
).fetchone()[0]
stage2_count = connection.execute(
"SELECT COUNT(*) FROM stage2_results WHERE file_id = ?",
(stored.file_id,),
).fetchone()[0]
assert stage1_count == 0
assert stage2_count == 0