diff --git a/scripts/match_filenames.py b/scripts/match_filenames.py index 65ff7c7..d1b0b25 100755 --- a/scripts/match_filenames.py +++ b/scripts/match_filenames.py @@ -42,6 +42,8 @@ Examples: """ import argparse +import json +import os import shutil import sqlite3 import subprocess @@ -192,6 +194,15 @@ def parse_arguments() -> argparse.Namespace: help="Show candidates, scores, durations and classification details.", ) + parser.add_argument( + "--detection-report", + type=Path, + help=( + "JSON detection report; only files classified as wcx " + "will be matched." + ), + ) + return parser.parse_args() @@ -277,6 +288,116 @@ def parse_video_extensions( return extensions +def normalize_absolute_path(path: Path) -> Path: + return Path(os.path.abspath(path)) + + +def load_detection_report( + report_file: Path, +) -> dict[Path, str]: + if not report_file.is_file(): + raise FileNotFoundError( + f"Detection report does not exist: {report_file}" + ) + + try: + with report_file.open(encoding="utf-8") as file: + report = json.load(file) + except json.JSONDecodeError as error: + raise ValueError( + f"Detection report is not valid JSON: {report_file}: {error}" + ) from error + + if not isinstance(report, dict): + raise ValueError( + "Detection report top level must be a JSON object." + ) + + schema_version = report.get("schema_version") + if type(schema_version) is not int or schema_version != 1: + raise ValueError( + "Detection report schema_version must be 1." + ) + + entries = report.get("files") + if not isinstance(entries, list): + raise ValueError( + "Detection report files must be a list." + ) + + allowed_classifications = { + "wcx", + "uncertain", + "not_wcx", + "error", + } + classifications_by_path = {} + + for index, entry in enumerate(entries): + if not isinstance(entry, dict): + raise ValueError( + f"Detection report files[{index}] must be an object." + ) + + if "path" not in entry: + raise ValueError( + f"Detection report files[{index}] is missing path." + ) + + path_value = entry["path"] + if not isinstance(path_value, str): + raise ValueError( + f"Detection report files[{index}].path must be a string." + ) + + if "classification" not in entry: + raise ValueError( + f"Detection report files[{index}] is missing classification." + ) + + classification = entry["classification"] + if not isinstance(classification, str): + raise ValueError( + f"Detection report files[{index}].classification " + f"must be a string." + ) + + if classification not in allowed_classifications: + raise ValueError( + f"Detection report files[{index}] has unknown " + f"classification: {classification}" + ) + + normalized_path = normalize_absolute_path( + Path(path_value) + ) + if normalized_path in classifications_by_path: + raise ValueError( + f"Detection report contains duplicate path: " + f"{normalized_path}" + ) + + classifications_by_path[normalized_path] = classification + + return classifications_by_path + + +def select_wcx_files( + inventoried_files: list[Path], + report_entries: dict[Path, str], +) -> list[Path]: + wcx_paths = { + path + for path, classification in report_entries.items() + if classification == "wcx" + } + return [ + path + for path in inventoried_files + if normalize_absolute_path(path) in wcx_paths + ] + + def load_movies(database_file: Path) -> list[Movie]: if not database_file.is_file(): raise FileNotFoundError( @@ -1267,25 +1388,73 @@ def main() -> None: "--limit must be greater than zero." ) - if shutil.which("ffprobe") is None: - raise FileNotFoundError( - "ffprobe was not found in PATH." + if args.detection_report is None: + if shutil.which("ffprobe") is None: + raise FileNotFoundError( + "ffprobe was not found in PATH." + ) + + extensions = parse_video_extensions( + args.ending ) - extensions = parse_video_extensions( - args.ending - ) + movies = load_movies(args.database) - movies = load_movies(args.database) + video_files = find_video_files( + directory=args.directory, + recursive=args.recursive, + extensions=extensions, + ) - video_files = find_video_files( - directory=args.directory, - recursive=args.recursive, - extensions=extensions, - ) + if args.limit is not None: + video_files = video_files[:args.limit] + else: + extensions = parse_video_extensions( + args.ending + ) + inventoried_files = find_video_files( + directory=args.directory, + recursive=args.recursive, + extensions=extensions, + ) - if args.limit is not None: - video_files = video_files[:args.limit] + if args.limit is not None: + inventoried_files = inventoried_files[:args.limit] + + report_entries = load_detection_report( + args.detection_report + ) + video_files = select_wcx_files( + inventoried_files, + report_entries, + ) + report_wcx_files = sum( + classification == "wcx" + for classification in report_entries.values() + ) + + print( + f"detection_report: " + f"{normalize_absolute_path(args.detection_report)}" + ) + print(f"inventoried_files: {len(inventoried_files)}") + print(f"report_wcx_files: {report_wcx_files}") + print(f"selected_files: {len(video_files)}") + print() + + if not video_files: + print( + "No inventoried files were classified as wcx; " + "nothing to match." + ) + return + + if shutil.which("ffprobe") is None: + raise FileNotFoundError( + "ffprobe was not found in PATH." + ) + + movies = load_movies(args.database) print( f"{'Filename':<55} " diff --git a/tests/test_match_filenames.py b/tests/test_match_filenames.py new file mode 100644 index 0000000..1b4c103 --- /dev/null +++ b/tests/test_match_filenames.py @@ -0,0 +1,340 @@ +import contextlib +import importlib.util +import io +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + + +SCRIPT_PATH = ( + Path(__file__).resolve().parents[1] + / "scripts" + / "match_filenames.py" +) +SPEC = importlib.util.spec_from_file_location( + "match_filenames", + SCRIPT_PATH, +) +if SPEC is None or SPEC.loader is None: + raise RuntimeError(f"Could not load script: {SCRIPT_PATH}") + +match_filenames = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = match_filenames +SPEC.loader.exec_module(match_filenames) + + +class DetectionReportTests(unittest.TestCase): + def write_report( + self, + directory: Path, + report: object, + ) -> Path: + report_file = directory / "report.json" + report_file.write_text( + json.dumps(report), + encoding="utf-8", + ) + return report_file + + def valid_report( + self, + files: list[dict[str, object]] | None = None, + ) -> dict[str, object]: + return { + "schema_version": 1, + "files": files if files is not None else [], + "future_field": "allowed", + } + + def assert_report_error( + self, + report: object, + message: str, + ) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + report_file = self.write_report( + Path(temporary_directory), + report, + ) + with self.assertRaisesRegex(ValueError, message): + match_filenames.load_detection_report(report_file) + + def test_valid_schema_version_one(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + video_file = directory / "movie.mp4" + report_file = self.write_report( + directory, + self.valid_report( + [ + { + "path": str(video_file), + "classification": "wcx", + "extra": 123, + } + ] + ), + ) + + entries = match_filenames.load_detection_report( + report_file + ) + + self.assertEqual( + entries, + { + match_filenames.normalize_absolute_path( + video_file + ): "wcx" + }, + ) + + def test_only_wcx_classification_is_selected(self) -> None: + inventoried = [ + Path("/videos/wcx.mp4"), + Path("/videos/uncertain.mp4"), + Path("/videos/not-wcx.mp4"), + Path("/videos/error.mp4"), + ] + report_entries = { + inventoried[0]: "wcx", + inventoried[1]: "uncertain", + inventoried[2]: "not_wcx", + inventoried[3]: "error", + } + + self.assertEqual( + match_filenames.select_wcx_files( + inventoried, + report_entries, + ), + [inventoried[0]], + ) + + def test_selection_is_intersection_with_inventory(self) -> None: + included = Path("/videos/included.mp4") + not_wcx = Path("/videos/not-wcx.mp4") + outside = Path("/elsewhere/outside.mp4") + + selected = match_filenames.select_wcx_files( + [included, not_wcx], + { + included: "wcx", + not_wcx: "not_wcx", + outside: "wcx", + }, + ) + + self.assertEqual(selected, [included]) + + def test_relative_and_absolute_paths_normalize_consistently(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + previous_directory = Path.cwd() + os.chdir(temporary_directory) + try: + relative_path = Path("movie.mp4") + absolute_path = Path(temporary_directory) / "movie.mp4" + + selected = match_filenames.select_wcx_files( + [relative_path], + {absolute_path: "wcx"}, + ) + finally: + os.chdir(previous_directory) + + self.assertEqual(selected, [relative_path]) + + def test_missing_report_file(self) -> None: + with self.assertRaisesRegex( + FileNotFoundError, + "does not exist", + ): + match_filenames.load_detection_report( + Path("/missing/detection-report.json") + ) + + def test_invalid_json(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + report_file = ( + Path(temporary_directory) + / "report.json" + ) + report_file.write_text("{invalid", encoding="utf-8") + + with self.assertRaisesRegex( + ValueError, + "not valid JSON", + ): + match_filenames.load_detection_report(report_file) + + def test_top_level_must_be_object(self) -> None: + self.assert_report_error([], "top level") + + def test_wrong_schema_version(self) -> None: + self.assert_report_error( + {"schema_version": 2, "files": []}, + "schema_version", + ) + + def test_files_must_be_list(self) -> None: + self.assert_report_error( + {"schema_version": 1, "files": {}}, + "files must be a list", + ) + + def test_missing_path(self) -> None: + self.assert_report_error( + self.valid_report( + [{"classification": "wcx"}] + ), + "missing path", + ) + + def test_path_must_be_string(self) -> None: + self.assert_report_error( + self.valid_report( + [{"path": 123, "classification": "wcx"}] + ), + "path must be a string", + ) + + def test_missing_classification(self) -> None: + self.assert_report_error( + self.valid_report( + [{"path": "/videos/movie.mp4"}] + ), + "missing classification", + ) + + def test_classification_must_be_string(self) -> None: + self.assert_report_error( + self.valid_report( + [ + { + "path": "/videos/movie.mp4", + "classification": 1, + } + ] + ), + "classification must be a string", + ) + + def test_unknown_classification(self) -> None: + self.assert_report_error( + self.valid_report( + [ + { + "path": "/videos/movie.mp4", + "classification": "maybe", + } + ] + ), + "unknown classification", + ) + + def test_duplicate_normalized_path(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + path = directory / "movie.mp4" + self.assert_report_error( + self.valid_report( + [ + { + "path": str(path), + "classification": "wcx", + }, + { + "path": str( + directory + / "." + / "movie.mp4" + ), + "classification": "not_wcx", + }, + ] + ), + "duplicate path", + ) + + def test_empty_report_selects_nothing(self) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + report_file = self.write_report( + directory, + self.valid_report(), + ) + entries = match_filenames.load_detection_report( + report_file + ) + + self.assertEqual(entries, {}) + self.assertEqual( + match_filenames.select_wcx_files( + [Path("/videos/movie.mp4")], + entries, + ), + [], + ) + + def test_without_detection_report_argument_is_unchanged(self) -> None: + arguments = [ + "match_filenames.py", + "/videos", + "--database", + "/database/wcx.db", + ] + with patch.object(sys, "argv", arguments): + args = match_filenames.parse_arguments() + + self.assertIsNone(args.detection_report) + self.assertEqual(args.directory, Path("/videos")) + + def test_empty_selection_does_not_open_database_or_run_ffprobe( + self, + ) -> None: + with tempfile.TemporaryDirectory() as temporary_directory: + directory = Path(temporary_directory) + video_file = directory / "movie.mp4" + video_file.touch() + report_file = self.write_report( + directory, + self.valid_report( + [ + { + "path": str(video_file), + "classification": "not_wcx", + } + ] + ), + ) + arguments = [ + "match_filenames.py", + str(directory), + "--database", + "/missing/wcx.db", + "--detection-report", + str(report_file), + ] + + with patch.object(sys, "argv", arguments): + with patch.object( + match_filenames, + "load_movies", + ) as load_movies: + with patch.object( + match_filenames, + "read_file_duration", + ) as read_file_duration: + with contextlib.redirect_stdout(io.StringIO()): + match_filenames.main() + + load_movies.assert_not_called() + read_file_duration.assert_not_called() + + +if __name__ == "__main__": + unittest.main()