diff --git a/scripts/diagnose_duration_match.py b/scripts/diagnose_duration_match.py new file mode 100755 index 0000000..1fa8b73 --- /dev/null +++ b/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 \ + --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) +