WCX/scripts/import_site.py

#!/usr/bin/env python3

"""
Import WCX site metadata into the local SQLite movie index.

The script reads the JSON snapshot produced by the WCX site scraper and
synchronizes it with the movie table in the local SQLite database.

Import rules:
- A movie whose ID does not exist in the database is inserted.
- An existing movie is updated only when the site provides an update date
  and that date is newer than the value currently stored in the database.
- If the database has no update date but the site does, the movie is updated.
- Movies without a newer site update date are left unchanged.

When an existing movie is updated, all fields owned by the site are replaced:
name, duration, web URL, thumbnail URL, publication date, and update date.
Manually maintained metadata fields are not modified.

Durations from the site are accepted as mm:ss or hh:mm:ss and stored as
seconds in the duration_seconds column.

Input:
    /storage/disk1/WCX/import/wcx_site_index.json

Database:
    /storage/disk1/WCX/database/wcx.db

The script prints a summary showing the number of inserted, updated, and
unchanged records.
"""


import json
import sqlite3
from pathlib import Path
from typing import Any


JSON_FILE = Path("/storage/disk1/WCX/import/wcx_site_index.json")
DATABASE_FILE = Path("/storage/disk1/WCX/database/wcx.db")


def parse_duration(value: str | None) -> int | None:
    """Convert mm:ss or hh:mm:ss to seconds."""
    if not value:
        return None

    parts = value.split(":")

    try:
        numbers = [int(part) for part in parts]
    except ValueError as error:
        raise ValueError(f"Ogiltig speltid: {value!r}") from error

    if len(numbers) == 2:
        minutes, seconds = numbers
        hours = 0
    elif len(numbers) == 3:
        hours, minutes, seconds = numbers
    else:
        raise ValueError(f"Ogiltigt duration-format: {value!r}")

    if hours < 0 or minutes < 0 or seconds < 0:
        raise ValueError(f"Speltiden får inte vara negativ: {value!r}")

    if seconds >= 60:
        raise ValueError(f"Ogiltigt sekundvärde i speltid: {value!r}")

    if len(numbers) == 3 and minutes >= 60:
        raise ValueError(f"Ogiltigt minutvärde i speltid: {value!r}")

    return hours * 3600 + minutes * 60 + seconds


def get_site_updated(movie: dict[str, Any]) -> str | None:
    """
    Read the site's update date.

    Accept both 'update' and 'updated', although the site's normal field
    is expected to be named 'update'.
    """
    return movie.get("update") or movie.get("updated")


def validate_movie(movie: Any, index: int) -> dict[str, Any]:
    if not isinstance(movie, dict):
        raise ValueError(f"Post {index} är inte ett JSON-objekt.")

    for required_field in ("id", "titel"):
        value = movie.get(required_field)

        if not isinstance(value, str) or not value.strip():
            raise ValueError(
                f"Post {index} saknar ett giltigt fält: {required_field}"
            )

    return movie


def insert_movie(
    connection: sqlite3.Connection,
    movie: dict[str, Any],
    site_updated: str | None,
) -> None:
    connection.execute(
        """
        INSERT INTO movie (
            id,
            name,
            duration_seconds,
            web_url,
            thumbnail,
            published,
            updated
        )
        VALUES (?, ?, ?, ?, ?, ?, ?)
        """,
        (
            movie["id"],
            movie["titel"],
            parse_duration(movie.get("duration")),
            movie.get("details"),
            movie.get("thumb"),
            movie.get("published"),
            site_updated,
        ),
    )


def update_movie(
    connection: sqlite3.Connection,
    movie: dict[str, Any],
    site_updated: str,
) -> None:
    connection.execute(
        """
        UPDATE movie
        SET
            name = ?,
            duration_seconds = ?,
            web_url = ?,
            thumbnail = ?,
            published = ?,
            updated = ?,
            modified_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (
            movie["titel"],
            parse_duration(movie.get("duration")),
            movie.get("details"),
            movie.get("thumb"),
            movie.get("published"),
            site_updated,
            movie["id"],
        ),
    )


def main() -> None:
    if not JSON_FILE.is_file():
        raise FileNotFoundError(f"JSON-filen finns inte: {JSON_FILE}")

    if not DATABASE_FILE.is_file():
        raise FileNotFoundError(f"Databasen finns inte: {DATABASE_FILE}")

    with JSON_FILE.open("r", encoding="utf-8") as file:
        movies = json.load(file)

    if not isinstance(movies, list):
        raise ValueError("JSON-filen måste innehålla en array.")

    inserted = 0
    updated = 0
    unchanged = 0

    with sqlite3.connect(DATABASE_FILE) as connection:
        for index, raw_movie in enumerate(movies, start=1):
            movie = validate_movie(raw_movie, index)
            movie_id = movie["id"]
            site_updated = get_site_updated(movie)

            existing = connection.execute(
                """
                SELECT updated
                FROM movie
                WHERE id = ?
                """,
                (movie_id,),
            ).fetchone()

            if existing is None:
                insert_movie(connection, movie, site_updated)
                inserted += 1
                continue

            database_updated = existing[0]

            should_update = (
                site_updated is not None
                and (
                    database_updated is None
                    or site_updated > database_updated
                )
            )

            if should_update:
                update_movie(connection, movie, site_updated)
                updated += 1
            else:
                unchanged += 1

    print(f"Poster i JSON: {len(movies)}")
    print(f"Tillagda:       {inserted}")
    print(f"Uppdaterade:    {updated}")
    print(f"Oförändrade:    {unchanged}")


if __name__ == "__main__":
    main()