initial commit

scripts/import_site.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+
+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()
+