Add duration matching diagnostic

scripts/diagnose_duration_match.py

@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+
+"""
+Manually diagnose a file duration against current and historical movie
+durations.
+
+This script does not modify the database or any files.
+
+Usage:
+    diagnose_duration_match.py <video-file> <movie-id> \
+        --database /path/to/wcx.db
+"""
+
+import argparse
+import sqlite3
+import subprocess
+import sys
+from pathlib import Path
+
+
+def parse_arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Diagnose a video duration against current and historical "
+            "movie durations."
+        )
+    )
+
+    parser.add_argument(
+        "video_file",
+        type=Path,
+        help="Video file to diagnose.",
+    )
+
+    parser.add_argument(
+        "movie_id",
+        help="Stable movie.id from the reference database.",
+    )
+
+    parser.add_argument(
+        "--database",
+        type=Path,
+        required=True,
+        help="SQLite reference database.",
+    )
+
+    return parser.parse_args()
+
+
+def read_file_duration(file_path: Path) -> int:
+    result = subprocess.run(
+        [
+            "ffprobe",
+            "-v",
+            "error",
+            "-show_entries",
+            "format=duration",
+            "-of",
+            "default=noprint_wrappers=1:nokey=1",
+            str(file_path),
+        ],
+        capture_output=True,
+        text=True,
+    )
+
+    if result.returncode != 0:
+        raise RuntimeError(
+            result.stderr.strip()
+            or f"ffprobe failed for {file_path}"
+        )
+
+    try:
+        return round(float(result.stdout.strip()))
+    except ValueError as error:
+        raise RuntimeError(
+            f"Invalid ffprobe duration: {result.stdout!r}"
+        ) from error
+
+
+def format_duration(seconds: int | None) -> str:
+    if seconds is None:
+        return "unknown"
+
+    hours, remainder = divmod(seconds, 3600)
+    minutes, seconds = divmod(remainder, 60)
+
+    if hours:
+        return f"{hours}:{minutes:02d}:{seconds:02d}"
+
+    return f"{minutes}:{seconds:02d}"
+
+
+def load_movie_durations(
+    connection: sqlite3.Connection,
+    movie_id: str,
+) -> tuple[str, int | None, list[int]]:
+    current = connection.execute(
+        """
+        SELECT name, duration_seconds
+        FROM movie
+        WHERE id = ?
+        """,
+        (movie_id,),
+    ).fetchone()
+
+    if current is None:
+        raise ValueError(
+            f"Movie not found: {movie_id}"
+        )
+
+    movie_name = str(current[0])
+    current_duration = current[1]
+
+    historical_rows = connection.execute(
+        """
+        SELECT DISTINCT duration_seconds
+        FROM movie_history
+        WHERE movie_id = ?
+          AND duration_seconds IS NOT NULL
+        ORDER BY duration_seconds
+        """,
+        (movie_id,),
+    ).fetchall()
+
+    historical_durations = [
+        int(row[0])
+        for row in historical_rows
+        if row[0] != current_duration
+    ]
+
+    return (
+        movie_name,
+        current_duration,
+        historical_durations,
+    )
+
+
+def classify_duration(
+    file_duration: int,
+    current_duration: int | None,
+    historical_durations: list[int],
+) -> tuple[str, int | None, int | None]:
+    """
+    Return:
+        classification
+        closest matching duration
+        absolute difference in seconds
+    """
+    candidates: list[tuple[str, int]] = []
+
+    if current_duration is not None:
+        candidates.append(
+            ("current_version_match", current_duration)
+        )
+
+    for duration in historical_durations:
+        candidates.append(
+            ("historical_version_match", duration)
+        )
+
+    if not candidates:
+        return "unknown_version", None, None
+
+    classification, closest_duration = min(
+        candidates,
+        key=lambda item: abs(file_duration - item[1]),
+    )
+
+    difference = abs(
+        file_duration - closest_duration
+    )
+
+    relative_difference = (
+        difference / closest_duration
+        if closest_duration > 0
+        else 1.0
+    )
+
+    close_match = (
+        difference <= 180
+        or relative_difference <= 0.05
+    )
+
+    if close_match:
+        return (
+            classification,
+            closest_duration,
+            difference,
+        )
+
+    if file_duration < closest_duration:
+        return (
+            "possible_truncated_version",
+            closest_duration,
+            difference,
+        )
+
+    if file_duration > closest_duration:
+        return (
+            "possible_extended_version",
+            closest_duration,
+            difference,
+        )
+
+    return (
+        "duration_mismatch",
+        closest_duration,
+        difference,
+    )
+
+
+def main() -> None:
+    args = parse_arguments()
+
+    file_path = args.video_file
+    movie_id = args.movie_id
+    database_file = args.database
+
+    if not file_path.is_file():
+        raise FileNotFoundError(
+            f"Video file not found: {file_path}"
+        )
+
+    if not database_file.is_file():
+        raise FileNotFoundError(
+            f"Database file not found: {database_file}"
+        )
+
+    file_duration = read_file_duration(
+        file_path
+    )
+
+    database_uri = f"{database_file.resolve().as_uri()}?mode=ro"
+
+    with sqlite3.connect(database_uri, uri=True) as connection:
+        (
+            movie_name,
+            current_duration,
+            historical_durations,
+        ) = load_movie_durations(
+            connection,
+            movie_id,
+        )
+
+    (
+        classification,
+        matched_duration,
+        difference,
+    ) = classify_duration(
+        file_duration=file_duration,
+        current_duration=current_duration,
+        historical_durations=historical_durations,
+    )
+
+    print(f"Movie ID:             {movie_id}")
+    print(f"Movie name:           {movie_name}")
+    print(
+        f"File duration:        "
+        f"{format_duration(file_duration)} "
+        f"({file_duration} seconds)"
+    )
+    print(
+        f"Current duration:     "
+        f"{format_duration(current_duration)}"
+    )
+
+    if historical_durations:
+        print("Historical durations:")
+
+        for duration in historical_durations:
+            print(
+                f"  {format_duration(duration)} "
+                f"({duration} seconds)"
+            )
+    else:
+        print("Historical durations: none")
+
+    print(
+        f"Closest duration:     "
+        f"{format_duration(matched_duration)}"
+    )
+    print(
+        f"Difference:           "
+        f"{format_duration(difference)}"
+    )
+    print(
+        f"Classification:       "
+        f"{classification}"
+    )
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (
+        FileNotFoundError,
+        RuntimeError,
+        ValueError,
+        sqlite3.Error,
+    ) as error:
+        print(f"Error: {error}", file=sys.stderr)
+        sys.exit(1)
+