Move duration diagnostic to collection project

2026-07-19 18:40:18 +02:00
parent f910e4fa31
commit a34f9cf08d
1 changed files with 0 additions and 270 deletions
--- a/scripts/test_duration_matching.py
+++ b/scripts/test_duration_matching.py
@ -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 "
            "<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)