matching script working

scripts/schema.sql

@@ -61,3 +61,24 @@ CREATE INDEX IF NOT EXISTS idx_movie_history_movie_id
 
 CREATE INDEX IF NOT EXISTS idx_movie_history_archived_at
     ON movie_history(archived_at);
+
+CREATE TABLE IF NOT EXISTS movie_name_alias (
+    alias_id INTEGER PRIMARY KEY AUTOINCREMENT,
+
+    movie_id TEXT NOT NULL,
+    alias TEXT NOT NULL,
+    normalized_alias TEXT NOT NULL,
+    source TEXT NOT NULL DEFAULT 'manual',
+
+    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    FOREIGN KEY (movie_id) REFERENCES movie(id),
+
+    UNIQUE (movie_id, normalized_alias)
+);
+
+CREATE INDEX IF NOT EXISTS idx_movie_name_alias_movie_id
+    ON movie_name_alias(movie_id);
+
+CREATE INDEX IF NOT EXISTS idx_movie_name_alias_normalized
+    ON movie_name_alias(normalized_alias);

scripts/test_duration_matching.py

@@ -0,0 +1,270 @@
+#!/usr/bin/env python3
+
+"""
+Small local test for matching a file duration against current and historical
+movie durations.
+
+This script does not modify the database or any files.
+"""
+
+import sqlite3
+import subprocess
+import sys
+from pathlib import Path
+
+
+DATABASE_FILE = Path(
+    "/storage/disk1/WCX/database/wcx.db"
+)
+
+
+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_name: str,
+) -> tuple[str, int | None, list[int]]:
+    current = connection.execute(
+        """
+        SELECT id, duration_seconds
+        FROM movie
+        WHERE name = ?
+        """,
+        (movie_name,),
+    ).fetchone()
+
+    if current is None:
+        raise ValueError(
+            f"Movie not found: {movie_name}"
+        )
+
+    movie_id = 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_id,
+        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:
+    if len(sys.argv) != 3:
+        raise SystemExit(
+            "Usage: test_duration_matching.py "
+            "<video-file> <movie-name>"
+        )
+
+    file_path = Path(sys.argv[1])
+    movie_name = sys.argv[2]
+
+    if not file_path.is_file():
+        raise FileNotFoundError(
+            f"Video file not found: {file_path}"
+        )
+
+    file_duration = read_file_duration(
+        file_path
+    )
+
+    with sqlite3.connect(DATABASE_FILE) as connection:
+        (
+            movie_id,
+            current_duration,
+            historical_durations,
+        ) = load_movie_durations(
+            connection,
+            movie_name,
+        )
+
+    (
+        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)
+