WCX/scripts/check_ocr.py

#!/usr/bin/env python3

"""
Run and validate OCR processing for one movie in the SQLite index.

The script accepts a WCX movie ID, retrieves the movie name and thumbnail URL
from the database, and then coordinates the complete OCR workflow:

    1. Run ocr.sh for the thumbnail URL.
    2. Pass the raw OCR text to parse_ocr.py.
    3. Compare the OCR name with the movie name stored in the database.
    4. Store validated metadata and OCR processing information.

When processing succeeds and the names match, the script updates:

    nationality
    shoot_location
    shoot_date
    ocr_raw_text
    ocr_status = completed
    ocr_processed_at
    modified_at

If OCR execution fails, the movie is marked as failed.

If the OCR text cannot be parsed, required fields are missing, or the OCR name
does not match the database name, the raw OCR text is retained and the movie
is marked as manual_review. In these cases, the parsed metadata fields are not
updated.

Usage:
    check_ocr.py <movie-id>

Example:
    check_ocr.py susana-melo_6707
"""


import sqlite3
import subprocess
import sys
from pathlib import Path


DATABASE_FILE = Path("/storage/disk1/WCX/database/wcx.db")
OCR_SCRIPT = Path("/storage/disk1/WCX/scripts/ocr.sh")
PARSER_SCRIPT = Path("/storage/disk1/WCX/scripts/parse_ocr.py")


def parse_key_value_output(output: str) -> dict[str, str]:
    result: dict[str, str] = {}

    for line in output.splitlines():
        if "=" not in line:
            continue

        key, value = line.split("=", 1)
        result[key.strip()] = value.strip()

    return result


def normalize_name(value: str) -> str:
    return " ".join(value.casefold().split())


def set_ocr_failed(
    connection: sqlite3.Connection,
    movie_id: str,
    error_message: str,
    raw_text: str | None = None,
) -> None:
    connection.execute(
        """
        UPDATE movie
        SET
            ocr_status = 'failed',
            ocr_raw_text = ?,
            ocr_error = ?,
            ocr_processed_at = CURRENT_TIMESTAMP,
            modified_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (
            raw_text,
            error_message,
            movie_id,
        ),
    )


def set_manual_review(
    connection: sqlite3.Connection,
    movie_id: str,
    raw_text: str,
    error_message: str,
) -> None:
    connection.execute(
        """
        UPDATE movie
        SET
            ocr_status = 'manual_review',
            ocr_raw_text = ?,
            ocr_error = ?,
            ocr_processed_at = CURRENT_TIMESTAMP,
            modified_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (
            raw_text,
            error_message,
            movie_id,
        ),
    )


def set_ocr_completed(
    connection: sqlite3.Connection,
    movie_id: str,
    raw_text: str,
    nationality: str,
    shoot_location: str,
    shoot_date: str,
) -> None:
    connection.execute(
        """
        UPDATE movie
        SET
            nationality = ?,
            shoot_location = ?,
            shoot_date = ?,
            ocr_status = 'completed',
            ocr_raw_text = ?,
            ocr_error = NULL,
            ocr_processed_at = CURRENT_TIMESTAMP,
            modified_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (
            nationality,
            shoot_location,
            shoot_date,
            raw_text,
            movie_id,
        ),
    )


def main() -> None:
    if len(sys.argv) != 2:
        print(f"Användning: {sys.argv[0]} <movie-id>", file=sys.stderr)
        sys.exit(1)

    movie_id = sys.argv[1]

    with sqlite3.connect(DATABASE_FILE) as connection:
        row = connection.execute(
            """
            SELECT name, thumbnail
            FROM movie
            WHERE id = ?
            """,
            (movie_id,),
        ).fetchone()

        if row is None:
            print(
                f"Filmen finns inte i databasen: {movie_id}",
                file=sys.stderr,
            )
            sys.exit(1)

        database_name, thumbnail = row

        if not thumbnail:
            error_message = "Filmen saknar thumbnail."

            set_ocr_failed(
                connection,
                movie_id,
                error_message,
            )

            print(error_message, file=sys.stderr)
            sys.exit(1)

        ocr_result = subprocess.run(
            [str(OCR_SCRIPT), thumbnail],
            capture_output=True,
            text=True,
        )

        if ocr_result.returncode != 0:
            error_message = (
                ocr_result.stderr.strip()
                or "OCR-körningen misslyckades."
            )

            set_ocr_failed(
                connection,
                movie_id,
                error_message,
                ocr_result.stdout.strip() or None,
            )

            print("OCR-körningen misslyckades:", file=sys.stderr)
            print(error_message, file=sys.stderr)
            sys.exit(1)

        raw_text = ocr_result.stdout.strip()

        parser_result = subprocess.run(
            [str(PARSER_SCRIPT)],
            input=raw_text,
            capture_output=True,
            text=True,
        )

        if parser_result.returncode != 0:
            error_message = (
                parser_result.stderr.strip()
                or "Tolkningen av OCR-resultatet misslyckades."
            )

            set_manual_review(
                connection,
                movie_id,
                raw_text,
                error_message,
            )

            print(
                "Tolkningen av OCR-resultatet misslyckades:",
                file=sys.stderr,
            )
            print(error_message, file=sys.stderr)
            sys.exit(1)

        parsed = parse_key_value_output(parser_result.stdout)

        required_fields = (
            "ocr_name",
            "nationality",
            "shoot_location",
            "shoot_date",
        )

        missing_fields = [
            field
            for field in required_fields
            if not parsed.get(field)
        ]

        if missing_fields:
            error_message = (
                "Parsern saknar fält: "
                + ", ".join(missing_fields)
            )

            set_manual_review(
                connection,
                movie_id,
                raw_text,
                error_message,
            )

            print(error_message, file=sys.stderr)
            sys.exit(1)

        ocr_name = parsed["ocr_name"]
        names_match = (
            normalize_name(database_name)
            == normalize_name(ocr_name)
        )

        if not names_match:
            error_message = (
                "OCR-namnet matchar inte databasnamnet: "
                f"{ocr_name!r} != {database_name!r}"
            )

            set_manual_review(
                connection,
                movie_id,
                raw_text,
                error_message,
            )

            print(error_message, file=sys.stderr)
            sys.exit(1)

        set_ocr_completed(
            connection=connection,
            movie_id=movie_id,
            raw_text=raw_text,
            nationality=parsed["nationality"],
            shoot_location=parsed["shoot_location"],
            shoot_date=parsed["shoot_date"],
        )

    print(f"id={movie_id}")
    print(f"database_name={database_name}")
    print(f"ocr_name={ocr_name}")
    print("name_match=yes")
    print(f"nationality={parsed['nationality']}")
    print(f"shoot_location={parsed['shoot_location']}")
    print(f"shoot_date={parsed['shoot_date']}")
    print("ocr_status=completed")


if __name__ == "__main__":
    main()