Filter filename matching by WCX detection report

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,6 +1388,7 @@ def main() -> None:
             "--limit must be greater than zero."
         )
 
+    if args.detection_report is None:
         if shutil.which("ffprobe") is None:
             raise FileNotFoundError(
                 "ffprobe was not found in PATH."
@@ -1286,6 +1408,53 @@ def main() -> None:
 
         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:
+            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} "

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()