From 1b651be90ec9640702a119c715b89e9540e3ce9b Mon Sep 17 00:00:00 2001 From: Urban Date: Tue, 21 Jul 2026 19:50:40 +0200 Subject: [PATCH] Add JSON report for WCX detection --- README.md | 14 +++ scripts/detect_wcx_title.py | 185 +++++++++++++++++++++++---- tests/test_detect_wcx_title.py | 224 ++++++++++++++++++++++++++++++++- 3 files changed, 398 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index d00fdd2..4861ff0 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/scripts/detect_wcx_title.py b/scripts/detect_wcx_title.py index 490c03c..904d415 100755 --- a/scripts/detect_wcx_title.py +++ b/scripts/detect_wcx_title.py @@ -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,20 +808,41 @@ 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 - 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( diff --git a/tests/test_detect_wcx_title.py b/tests/test_detect_wcx_title.py index fdb3c8a..761da81 100644 --- a/tests/test_detect_wcx_title.py +++ b/tests/test_detect_wcx_title.py @@ -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()