Add FFmpeg-based video integrity check
This commit is contained in:
+1
-1
@@ -1,2 +1,2 @@
|
||||
"""Auto-generated — do not edit manually."""
|
||||
__version__ = "1.9.0"
|
||||
__version__ = "1.10.0"
|
||||
|
||||
@@ -6,7 +6,8 @@ from .file import File
|
||||
from .tag_manager import TagManager
|
||||
from .pool_index import PoolIndex
|
||||
from .utils import list_files
|
||||
from typing import Iterable
|
||||
from .integrity import check_video_integrity, ffmpeg_available, IntegrityResult
|
||||
from typing import Callable, Iterable
|
||||
from src.core.config import (
|
||||
load_global_config, save_global_config, DEFAULT_TAG_SCHEMA
|
||||
)
|
||||
@@ -226,6 +227,49 @@ class FileManager:
|
||||
"disk_count": len(disk_keys),
|
||||
}
|
||||
|
||||
def scan_video_integrity(
|
||||
self,
|
||||
files: list[File],
|
||||
deep: bool = True,
|
||||
on_progress: Callable[[int, int, File], None] | None = None,
|
||||
should_cancel: Callable[[], bool] | None = None,
|
||||
) -> dict:
|
||||
"""Check a list of movies for corrupted/unreadable video data.
|
||||
|
||||
Runs :func:`check_video_integrity` on each file (deep = full ffmpeg
|
||||
decode, otherwise a quick ffprobe probe). ``on_progress(done, total,
|
||||
file)`` is called after each file and ``should_cancel()`` is polled
|
||||
before each — so a GUI can show progress and stop between files.
|
||||
|
||||
Returns a dict with:
|
||||
ok — False if ffmpeg/ffprobe isn't available (with 'error')
|
||||
problems — IntegrityResult list for files that failed (bad or
|
||||
unable-to-check), in scan order
|
||||
checked — how many files were actually processed
|
||||
cancelled — True if the scan stopped early
|
||||
"""
|
||||
if not ffmpeg_available(deep):
|
||||
tool = "ffmpeg" if deep else "ffprobe"
|
||||
return {"ok": False, "error": f"{tool} není k dispozici (nainstaluj FFmpeg).",
|
||||
"problems": [], "checked": 0, "cancelled": False}
|
||||
|
||||
problems: list[IntegrityResult] = []
|
||||
total = len(files)
|
||||
checked = 0
|
||||
cancelled = False
|
||||
for i, f in enumerate(files, 1):
|
||||
if should_cancel is not None and should_cancel():
|
||||
cancelled = True
|
||||
break
|
||||
result = check_video_integrity(f.file_path, deep=deep)
|
||||
checked += 1
|
||||
if not result.ok:
|
||||
problems.append(result)
|
||||
if on_progress is not None:
|
||||
on_progress(i, total, f)
|
||||
return {"ok": True, "problems": problems, "checked": checked,
|
||||
"cancelled": cancelled}
|
||||
|
||||
def statistics(self) -> dict:
|
||||
"""Aggregate stats over the loaded pool movies for the stats view.
|
||||
|
||||
|
||||
@@ -0,0 +1,114 @@
|
||||
"""Video integrity checking via FFmpeg.
|
||||
|
||||
Detects corrupted or unreadable video files. Two depths:
|
||||
|
||||
- **deep** (default): a full ``ffmpeg`` decode to the null muxer
|
||||
(``-f null``). It actually decodes every frame, so it catches mid-file data
|
||||
corruption (broken GOPs, truncated streams, bad packets). Thorough but slow —
|
||||
it reads and decodes the whole file.
|
||||
- **quick**: an ``ffprobe`` container/stream probe. Fast but shallow — it catches
|
||||
an unreadable container, a truncated header or a missing video stream, but not
|
||||
corruption buried inside otherwise-parseable data.
|
||||
|
||||
Both tools ship with FFmpeg; ``ffmpeg_available`` reports whether they are on the
|
||||
PATH so callers can degrade gracefully.
|
||||
"""
|
||||
import shutil
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
FFMPEG = "ffmpeg"
|
||||
FFPROBE = "ffprobe"
|
||||
|
||||
# Cap a single stderr excerpt so a chatty decoder can't flood the report.
|
||||
_MAX_ERROR_LEN = 300
|
||||
|
||||
|
||||
@dataclass
|
||||
class IntegrityResult:
|
||||
"""Outcome of checking one file.
|
||||
|
||||
``ok`` — the file decoded/probed cleanly.
|
||||
``checked`` — the check actually ran; False means it couldn't (missing tool,
|
||||
missing file), which is *not* the same as "corrupted".
|
||||
``error`` — a short reason when not ``ok``.
|
||||
"""
|
||||
|
||||
path: Path
|
||||
ok: bool
|
||||
error: str = ""
|
||||
checked: bool = True
|
||||
|
||||
|
||||
def ffmpeg_available(deep: bool = True) -> bool:
|
||||
"""Is the required FFmpeg tool (ffmpeg for deep, ffprobe for quick) on PATH?"""
|
||||
return shutil.which(FFMPEG if deep else FFPROBE) is not None
|
||||
|
||||
|
||||
def check_video_integrity(
|
||||
path: Path, deep: bool = True, timeout: Optional[int] = None
|
||||
) -> IntegrityResult:
|
||||
"""Check a single video file for corruption.
|
||||
|
||||
Args:
|
||||
path: the video file to check.
|
||||
deep: True → full ffmpeg decode (catches data corruption, slow);
|
||||
False → ffprobe probe (catches unreadable/truncated files, fast).
|
||||
timeout: per-file timeout in seconds (None = no limit).
|
||||
|
||||
Returns:
|
||||
An :class:`IntegrityResult`. ``checked=False`` when the check could not
|
||||
be run at all (tool or file missing) — distinct from a real failure.
|
||||
"""
|
||||
path = Path(path)
|
||||
if not path.is_file():
|
||||
return IntegrityResult(path, ok=False, error="Soubor neexistuje", checked=False)
|
||||
|
||||
tool = FFMPEG if deep else FFPROBE
|
||||
if shutil.which(tool) is None:
|
||||
return IntegrityResult(
|
||||
path, ok=False, error=f"{tool} není k dispozici", checked=False
|
||||
)
|
||||
|
||||
if deep:
|
||||
cmd = [FFMPEG, "-v", "error", "-xerror", "-i", str(path), "-f", "null", "-"]
|
||||
else:
|
||||
cmd = [
|
||||
FFPROBE, "-v", "error", "-select_streams", "v:0",
|
||||
"-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path),
|
||||
]
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
cmd, capture_output=True, text=True, errors="replace", timeout=timeout
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return IntegrityResult(path, ok=False, error="Kontrola vypršela (timeout)")
|
||||
except OSError as exc:
|
||||
return IntegrityResult(path, ok=False, error=str(exc), checked=False)
|
||||
|
||||
stderr = (result.stderr or "").strip()
|
||||
if deep:
|
||||
ok = result.returncode == 0 and not stderr
|
||||
else:
|
||||
# Quick probe: a healthy file returns rc 0 and a "video" codec_type row.
|
||||
ok = result.returncode == 0 and "video" in (result.stdout or "")
|
||||
if not ok and not stderr:
|
||||
stderr = "Žádný video stream / nečitelný kontejner"
|
||||
|
||||
if ok:
|
||||
return IntegrityResult(path, ok=True)
|
||||
return IntegrityResult(
|
||||
path, ok=False, error=_first_error_line(stderr) or "Neznámá chyba dekódování"
|
||||
)
|
||||
|
||||
|
||||
def _first_error_line(stderr: str) -> str:
|
||||
"""First non-empty stderr line, truncated — a compact reason for the report."""
|
||||
for line in stderr.splitlines():
|
||||
line = line.strip()
|
||||
if line:
|
||||
return line[:_MAX_ERROR_LEN]
|
||||
return ""
|
||||
+85
-1
@@ -20,7 +20,7 @@ from PySide6.QtWidgets import (
|
||||
QApplication, QMainWindow, QWidget, QSplitter, QTreeWidget, QTreeWidgetItem,
|
||||
QTableWidget, QTableWidgetItem, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit,
|
||||
QPushButton, QFileDialog, QMessageBox, QInputDialog, QDialog, QDialogButtonBox,
|
||||
QHeaderView, QMenu, QAbstractItemView, QCheckBox, QComboBox,
|
||||
QHeaderView, QMenu, QAbstractItemView, QCheckBox, QComboBox, QProgressDialog,
|
||||
)
|
||||
|
||||
from src.core.file_manager import FileManager
|
||||
@@ -518,6 +518,7 @@ class QtApp(QMainWindow):
|
||||
|
||||
tests_menu = bar.addMenu("&Testy")
|
||||
self._add_action(tests_menu, "Kontrola konzistence dat…", self.check_consistency)
|
||||
self._add_action(tests_menu, "Kontrola integrity videa…", self.check_video_integrity)
|
||||
|
||||
def _add_action(self, menu: QMenu, text: str, slot, shortcut: str | None = None) -> QAction:
|
||||
action = QAction(text, self)
|
||||
@@ -1130,6 +1131,89 @@ class QtApp(QMainWindow):
|
||||
box.setDetailedText("\n".join(detail))
|
||||
box.exec()
|
||||
|
||||
def check_video_integrity(self) -> None:
|
||||
"""Test: scan movies for corrupted / unreadable video data (FFmpeg)."""
|
||||
if not self.filehandler.movies_dir:
|
||||
QMessageBox.information(self, "Testy", "Nejprve nastavte pool.")
|
||||
return
|
||||
|
||||
# Scope: selected movies, or the whole pool when nothing is selected.
|
||||
files = self._selected_movies() or self.filehandler.filelist
|
||||
if not files:
|
||||
QMessageBox.information(self, "Kontrola integrity", "Pool je prázdný.")
|
||||
return
|
||||
scope = "vybrané filmy" if self._selected_movies() else "celý pool"
|
||||
|
||||
# Depth: deep = full decode (thorough, slow), quick = ffprobe (fast).
|
||||
box = QMessageBox(self)
|
||||
box.setIcon(QMessageBox.Question)
|
||||
box.setWindowTitle("Kontrola integrity videa")
|
||||
box.setText(
|
||||
f"Zkontrolovat {scope} ({len(files)}).\n\n"
|
||||
"Důkladná: plné dekódování přes ffmpeg — odhalí i poškozená data "
|
||||
"uvnitř souboru, ale čte celý soubor (pomalé).\n"
|
||||
"Rychlá: ffprobe — odhalí nečitelné/uříznuté soubory (rychlé)."
|
||||
)
|
||||
deep_btn = box.addButton("Důkladná", QMessageBox.AcceptRole)
|
||||
quick_btn = box.addButton("Rychlá", QMessageBox.AcceptRole)
|
||||
box.addButton("Zrušit", QMessageBox.RejectRole)
|
||||
box.exec()
|
||||
clicked = box.clickedButton()
|
||||
if clicked not in (deep_btn, quick_btn):
|
||||
return
|
||||
deep = clicked is deep_btn
|
||||
|
||||
progress = QProgressDialog(
|
||||
"Kontroluji integritu videa…", "Zrušit", 0, len(files), self)
|
||||
progress.setWindowTitle("Kontrola integrity videa")
|
||||
progress.setWindowModality(Qt.WindowModal)
|
||||
progress.setMinimumDuration(0)
|
||||
|
||||
def on_progress(done: int, total: int, f: File) -> None:
|
||||
progress.setValue(done)
|
||||
progress.setLabelText(f"[{done}/{total}] {f.filename}")
|
||||
QApplication.processEvents()
|
||||
|
||||
result = self.filehandler.scan_video_integrity(
|
||||
files, deep=deep, on_progress=on_progress,
|
||||
should_cancel=progress.wasCanceled)
|
||||
progress.setValue(len(files))
|
||||
|
||||
if not result["ok"]:
|
||||
QMessageBox.warning(self, "Kontrola integrity", result["error"])
|
||||
return
|
||||
|
||||
problems = result["problems"]
|
||||
checked = result["checked"]
|
||||
suffix = " (zrušeno)" if result["cancelled"] else ""
|
||||
if not problems:
|
||||
QMessageBox.information(
|
||||
self, "Kontrola integrity videa",
|
||||
f"✅ Bez problémů — zkontrolováno {checked} souborů{suffix}.")
|
||||
return
|
||||
|
||||
bad = [r for r in problems if r.checked]
|
||||
unchecked = [r for r in problems if not r.checked]
|
||||
detail: list[str] = []
|
||||
if bad:
|
||||
detail.append(f"■ Poškozené / vadné ({len(bad)}):")
|
||||
detail += [f" • {r.path.name} — {r.error}" for r in bad]
|
||||
detail.append("")
|
||||
if unchecked:
|
||||
detail.append(f"■ Nešlo zkontrolovat ({len(unchecked)}):")
|
||||
detail += [f" • {r.path.name} — {r.error}" for r in unchecked]
|
||||
|
||||
msg = QMessageBox(self)
|
||||
msg.setIcon(QMessageBox.Warning)
|
||||
msg.setWindowTitle("Kontrola integrity — nalezeny problémy")
|
||||
msg.setText(
|
||||
f"Zkontrolováno: {checked}{suffix}\n"
|
||||
f"Poškozené / vadné: {len(bad)}\n"
|
||||
f"Nešlo zkontrolovat: {len(unchecked)}\n\n"
|
||||
"Podrobnosti zobrazíš přes „Show Details…\".")
|
||||
msg.setDetailedText("\n".join(detail))
|
||||
msg.exec()
|
||||
|
||||
def closeEvent(self, event) -> None: # noqa: N802 — Qt override
|
||||
self.filehandler.global_config["window_geometry"] = f"{self.width()}x{self.height()}"
|
||||
from src.core.config import save_global_config
|
||||
|
||||
Reference in New Issue
Block a user