diff --git a/scripts/test_duration_matching.py b/scripts/test_duration_matching.py deleted file mode 100755 index 450145a..0000000 --- a/scripts/test_duration_matching.py +++ /dev/null @@ -1,270 +0,0 @@ -#!/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 " - " " - ) - - 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) -