urban/WCX-collection
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add JSON report for WCX detection

Browse Source
This commit is contained in:
Urban
2026-07-21 19:50:40 +02:00
parent a60ef87ddc
commit 1b651be90e
3 changed files with 398 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
README.md
View File

@ -47,6 +47,20 @@ scripts/detect_wcx_title.py \
--ending mp4,avi
```
Write a machine-readable report without changing the terminal output:
```bash
scripts/detect_wcx_title.py \
/storage/disk1/X \
--batch \
--ending mp4 \
--json-report reports/wcx-detection.json
```
The JSON report contains every selected file with its classification, score,
best timestamp, or error details, followed by the same aggregate summary as
the terminal output.
Batch output contains one line per file with its classification, score, best
timestamp, and full file path. A summary after the run reports the number
processed and the totals for `wcx`, `uncertain`, `not_wcx`, and errors.

177
scripts/detect_wcx_title.py
View File

@ -15,10 +15,13 @@ Example:
"""
import argparse
import json
import os
import shlex
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Callable, Sequence
@ -90,6 +93,12 @@ def parse_arguments() -> argparse.Namespace:
),
)
parser.add_argument(
"--json-report",
type=Path,
help="Write a machine-readable JSON report after a batch run.",
)
return parser.parse_args()
@ -484,38 +493,141 @@ def analyze_video_for_batch(
def process_batch(
video_files: Sequence[Path],
analyzer: Callable[[Path], tuple[float, int, str]],
) -> int:
counts = {
) -> tuple[list[dict[str, object]], dict[str, int]]:
summary = {
"processed": 0,
"wcx": 0,
"uncertain": 0,
"not_wcx": 0,
"errors": 0,
}
files = []
for video_file in video_files:
counts["processed"] += 1
absolute_path = str(video_file.resolve())
summary["processed"] += 1
try:
timestamp, score, classification = analyzer(video_file)
except (OSError, RuntimeError, ValueError) as error:
counts["errors"] += 1
summary["errors"] += 1
files.append(
{
"path": absolute_path,
"classification": "error",
"score": None,
"best_timestamp": None,
"error": str(error),
}
)
print(f"error - - {video_file}: {error}")
continue
counts[classification] += 1
summary[classification] += 1
files.append(
{
"path": absolute_path,
"classification": classification,
"score": score,
"best_timestamp": timestamp,
}
)
print(
f"{classification:<10} {score}/8 "
f"{timestamp:>4.1f} {video_file}"
)
print()
print(f"processed: {counts['processed']}")
print(f"wcx: {counts['wcx']}")
print(f"uncertain: {counts['uncertain']}")
print(f"not_wcx: {counts['not_wcx']}")
print(f"errors: {counts['errors']}")
print(f"processed: {summary['processed']}")
print(f"wcx: {summary['wcx']}")
print(f"uncertain: {summary['uncertain']}")
print(f"not_wcx: {summary['not_wcx']}")
print(f"errors: {summary['errors']}")
return 1 if counts["errors"] else 0
return files, summary
def build_json_report(
source_directory: Path,
recursive: bool,
endings: set[str],
summary: dict[str, int],
files: list[dict[str, object]],
) -> dict[str, object]:
return {
"schema_version": 1,
"source_directory": str(source_directory.resolve()),
"recursive": recursive,
"endings": sorted(endings),
"summary": dict(summary),
"files": list(files),
}
def write_json_report(
report_path: Path,
report: dict[str, object],
) -> None:
if report_path.exists():
raise FileExistsError(
f"JSON report already exists: {report_path}"
)
report_path.parent.mkdir(parents=True, exist_ok=True)
temporary_path = None
try:
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=report_path.parent,
prefix=f".{report_path.name}.",
suffix=".tmp",
delete=False,
) as temporary_file:
temporary_path = Path(temporary_file.name)
json.dump(
report,
temporary_file,
indent=2,
ensure_ascii=False,
)
temporary_file.write("\n")
temporary_file.flush()
os.fsync(temporary_file.fileno())
if report_path.exists():
raise FileExistsError(
f"JSON report already exists: {report_path}"
)
temporary_path.replace(report_path)
temporary_path = None
finally:
if temporary_path is not None:
temporary_path.unlink(missing_ok=True)
def write_optional_json_report(
report_path: Path | None,
report: dict[str, object],
writer: Callable[
[Path, dict[str, object]],
None,
] = write_json_report,
) -> int:
if report_path is None:
return 0
try:
writer(report_path, report)
except (OSError, TypeError, ValueError) as error:
print(
f"Error: Could not write JSON report {report_path}: {error}",
file=sys.stderr,
)
return 1
return 0
def diagnose_layout(
@ -663,6 +775,16 @@ def main() -> int:
)
return 2
if (
args.json_report is not None
and args.json_report.exists()
):
print(
f"Error: JSON report already exists: {args.json_report}",
file=sys.stderr,
)
return 2
if not args.input_path.is_dir():
print(
f"Error: Batch directory does not exist: {args.input_path}",
@ -686,21 +808,42 @@ def main() -> int:
f"No matching video files found in: "
f"{args.input_path.resolve()}"
)
return process_batch((), lambda _path: (0.0, 0, "not_wcx"))
analyzer = lambda _path: (0.0, 0, "not_wcx")
else:
ffmpeg = shutil.which("ffmpeg")
if ffmpeg is None:
print("Error: ffmpeg was not found in PATH.", file=sys.stderr)
return 1
return process_batch(
video_files,
lambda video_file: analyze_video_for_batch(
analyzer = lambda video_file: analyze_video_for_batch(
ffmpeg,
video_file,
),
)
files, summary = process_batch(
video_files,
analyzer,
)
report = build_json_report(
source_directory=args.input_path,
recursive=args.recursive,
endings=endings,
summary=summary,
files=files,
)
report_exit_code = write_optional_json_report(
args.json_report,
report,
)
return 1 if summary["errors"] or report_exit_code else 0
if args.json_report is not None:
print(
"Error: --json-report requires --batch.",
file=sys.stderr,
)
return 2
if args.recursive:
print(
"Error: --recursive requires --batch.",

224
tests/test_detect_wcx_title.py
View File

@ -1,9 +1,12 @@
import contextlib
import importlib.util
import io
import json
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
SCRIPT_PATH = (
@ -413,12 +416,13 @@ class BatchModeTests(unittest.TestCase):
output = io.StringIO()
with contextlib.redirect_stdout(output):
exit_code = detect_wcx_title.process_batch(
files, summary = detect_wcx_title.process_batch(
video_files,
lambda path: results[path],
)
self.assertEqual(exit_code, 0)
self.assertEqual(len(files), 3)
self.assertEqual(summary["errors"], 0)
self.assertIn("processed: 3", output.getvalue())
self.assertIn("wcx: 1", output.getvalue())
self.assertIn("uncertain: 1", output.getvalue())
@ -438,16 +442,228 @@ class BatchModeTests(unittest.TestCase):
output = io.StringIO()
with contextlib.redirect_stdout(output):
exit_code = detect_wcx_title.process_batch(
files, summary = detect_wcx_title.process_batch(
video_files,
analyze,
)
self.assertEqual(exit_code, 1)
self.assertEqual(summary["errors"], 1)
self.assertEqual(files[1]["classification"], "error")
self.assertIn("error", output.getvalue())
self.assertIn("processed: 2", output.getvalue())
self.assertIn("errors: 1", output.getvalue())
class JsonReportTests(unittest.TestCase):
def test_report_contains_results_summary_and_sorted_endings(self) -> None:
with tempfile.TemporaryDirectory() as temporary_directory:
source_directory = Path(temporary_directory)
video_files = (
source_directory / "a.mp4",
source_directory / "b.avi",
source_directory / "c.mp4",
)
results = {
video_files[0]: (8.0, 8, "wcx"),
video_files[1]: (10.0, 4, "not_wcx"),
video_files[2]: (12.0, 7, "uncertain"),
}
with contextlib.redirect_stdout(io.StringIO()):
files, summary = detect_wcx_title.process_batch(
video_files,
lambda path: results[path],
)
report = detect_wcx_title.build_json_report(
source_directory=source_directory,
recursive=False,
endings={"mp4", "avi"},
summary=summary,
files=files,
)
report_path = (
source_directory
/ "new-parent"
/ "report.json"
)
detect_wcx_title.write_json_report(
report_path,
report,
)
self.assertTrue(report_path.parent.is_dir())
self.assertTrue(
report_path.read_bytes().endswith(b"\n")
)
with report_path.open(encoding="utf-8") as report_file:
loaded = json.load(report_file)
self.assertEqual(loaded["endings"], ["avi", "mp4"])
self.assertEqual(
loaded["source_directory"],
str(source_directory.resolve()),
)
self.assertEqual(
[item["classification"] for item in loaded["files"]],
["wcx", "not_wcx", "uncertain"],
)
self.assertTrue(
all(
Path(item["path"]).is_absolute()
for item in loaded["files"]
)
)
self.assertEqual(
loaded["summary"],
{
"processed": 3,
"wcx": 1,
"uncertain": 1,
"not_wcx": 1,
"errors": 0,
},
)
def test_report_contains_error_result(self) -> None:
video_file = Path("/videos/broken.mp4")
with contextlib.redirect_stdout(io.StringIO()):
files, summary = detect_wcx_title.process_batch(
(video_file,),
lambda _path: (_ for _ in ()).throw(
RuntimeError("mock ffmpeg failure")
),
)
report = detect_wcx_title.build_json_report(
source_directory=Path("/videos"),
recursive=False,
endings={"mp4"},
summary=summary,
files=files,
)
self.assertEqual(
report["files"],
[
{
"path": str(video_file.resolve()),
"classification": "error",
"score": None,
"best_timestamp": None,
"error": "mock ffmpeg failure",
}
],
)
self.assertEqual(report["summary"]["errors"], 1)
def test_empty_batch_report(self) -> None:
with tempfile.TemporaryDirectory() as temporary_directory:
with contextlib.redirect_stdout(io.StringIO()):
files, summary = detect_wcx_title.process_batch(
(),
lambda _path: (8.0, 8, "wcx"),
)
report = detect_wcx_title.build_json_report(
source_directory=Path(temporary_directory),
recursive=True,
endings={"mp4"},
summary=summary,
files=files,
)
report_path = (
Path(temporary_directory)
/ "reports"
/ "empty.json"
)
detect_wcx_title.write_json_report(report_path, report)
with report_path.open(encoding="utf-8") as report_file:
loaded = json.load(report_file)
self.assertEqual(loaded["files"], [])
self.assertEqual(
loaded["summary"],
{
"processed": 0,
"wcx": 0,
"uncertain": 0,
"not_wcx": 0,
"errors": 0,
},
)
def test_existing_report_is_rejected(self) -> None:
with tempfile.TemporaryDirectory() as temporary_directory:
report_path = Path(temporary_directory) / "report.json"
report_path.write_text("existing", encoding="utf-8")
with self.assertRaises(FileExistsError):
detect_wcx_title.write_json_report(
report_path,
{"schema_version": 1},
)
self.assertEqual(
report_path.read_text(encoding="utf-8"),
"existing",
)
def test_report_write_failure_returns_one(self) -> None:
def failing_writer(
_path: Path,
_report: dict[str, object],
) -> None:
raise OSError("mock write failure")
with contextlib.redirect_stderr(io.StringIO()):
exit_code = detect_wcx_title.write_optional_json_report(
Path("/reports/report.json"),
{"schema_version": 1},
writer=failing_writer,
)
self.assertEqual(exit_code, 1)
def test_json_report_without_batch_is_rejected(self) -> None:
arguments = [
"detect_wcx_title.py",
"/videos/movie.mp4",
"--json-report",
"/reports/report.json",
]
with patch.object(sys, "argv", arguments):
with contextlib.redirect_stderr(io.StringIO()):
exit_code = detect_wcx_title.main()
self.assertEqual(exit_code, 2)
def test_existing_report_stops_before_batch_processing(self) -> None:
with tempfile.TemporaryDirectory() as temporary_directory:
directory = Path(temporary_directory)
report_path = directory / "report.json"
report_path.write_text("existing", encoding="utf-8")
arguments = [
"detect_wcx_title.py",
str(directory),
"--batch",
"--json-report",
str(report_path),
]
with patch.object(sys, "argv", arguments):
with patch.object(
detect_wcx_title,
"process_batch",
) as process_batch:
with contextlib.redirect_stderr(io.StringIO()):
exit_code = detect_wcx_title.main()
self.assertEqual(exit_code, 2)
process_batch.assert_not_called()
if __name__ == "__main__":
unittest.main()