567 lines
13 KiB
Python
Executable File
567 lines
13 KiB
Python
Executable File
#!/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:
|
|
- The complete current movie row is copied to movie_history.
|
|
- change_summary describes which site-owned fields changed.
|
|
- The site-owned fields in movie are then replaced.
|
|
- The history insert and movie update occur in the same transaction.
|
|
|
|
History is created only by this site import. OCR processing, CSV migration,
|
|
and manual database changes do not trigger history creation.
|
|
|
|
Fields owned by the site:
|
|
- name
|
|
- duration_seconds
|
|
- web_url
|
|
- thumbnail
|
|
- published
|
|
- updated
|
|
|
|
Manually maintained metadata fields are not modified by the site import.
|
|
|
|
Durations from the site are accepted as mm:ss or hh:mm:ss and stored as
|
|
seconds in the duration_seconds column.
|
|
|
|
Default input:
|
|
/storage/disk1/WCX/import/wcx_site_index.json
|
|
|
|
Default database:
|
|
/storage/disk1/WCX/database/wcx.db
|
|
|
|
Usage:
|
|
import_site.py
|
|
|
|
import_site.py \
|
|
--input /path/to/wcx_site_index.json \
|
|
--database /path/to/wcx.db
|
|
|
|
The script prints a summary showing the number of inserted, updated,
|
|
historized, and unchanged records.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_JSON_FILE = Path(
|
|
"/storage/disk1/WCX/import/wcx_site_index.json"
|
|
)
|
|
|
|
DEFAULT_DATABASE_FILE = Path(
|
|
"/storage/disk1/WCX/database/wcx.db"
|
|
)
|
|
|
|
SITE_FIELDS = (
|
|
"name",
|
|
"duration_seconds",
|
|
"web_url",
|
|
"thumbnail",
|
|
"published",
|
|
"updated",
|
|
)
|
|
|
|
MOVIE_HISTORY_FIELDS = (
|
|
"movie_id",
|
|
"name",
|
|
"aka",
|
|
"nationality",
|
|
"age",
|
|
"shoot_location",
|
|
"shoot_date",
|
|
"duration_seconds",
|
|
"description",
|
|
"rating",
|
|
"web_url",
|
|
"thumbnail",
|
|
"published",
|
|
"updated",
|
|
"created_at",
|
|
"modified_at",
|
|
"ocr_status",
|
|
"ocr_raw_text",
|
|
"ocr_error",
|
|
"ocr_processed_at",
|
|
)
|
|
|
|
|
|
def parse_arguments() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Import WCX site metadata into SQLite."
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--input",
|
|
type=Path,
|
|
default=DEFAULT_JSON_FILE,
|
|
help=f"Input JSON file; default: {DEFAULT_JSON_FILE}",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--database",
|
|
type=Path,
|
|
default=DEFAULT_DATABASE_FILE,
|
|
help=f"SQLite database; default: {DEFAULT_DATABASE_FILE}",
|
|
)
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
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: "
|
|
f"{required_field}"
|
|
)
|
|
|
|
return movie
|
|
|
|
|
|
def build_site_values(
|
|
movie: dict[str, Any],
|
|
site_updated: str | None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Convert one JSON movie to the site-owned values stored in SQLite.
|
|
"""
|
|
return {
|
|
"name": movie["titel"],
|
|
"duration_seconds": parse_duration(
|
|
movie.get("duration")
|
|
),
|
|
"web_url": movie.get("details"),
|
|
"thumbnail": movie.get("thumb"),
|
|
"published": movie.get("published"),
|
|
"updated": site_updated,
|
|
}
|
|
|
|
|
|
def insert_movie(
|
|
connection: sqlite3.Connection,
|
|
movie_id: str,
|
|
site_values: dict[str, Any],
|
|
) -> None:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO movie (
|
|
id,
|
|
name,
|
|
duration_seconds,
|
|
web_url,
|
|
thumbnail,
|
|
published,
|
|
updated
|
|
)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
movie_id,
|
|
site_values["name"],
|
|
site_values["duration_seconds"],
|
|
site_values["web_url"],
|
|
site_values["thumbnail"],
|
|
site_values["published"],
|
|
site_values["updated"],
|
|
),
|
|
)
|
|
|
|
|
|
def format_value(
|
|
field_name: str,
|
|
value: Any,
|
|
) -> str:
|
|
"""
|
|
Format a database value for the human-readable change summary.
|
|
"""
|
|
if value is None:
|
|
return "NULL"
|
|
|
|
if field_name == "duration_seconds":
|
|
total_seconds = int(value)
|
|
hours, remainder = divmod(total_seconds, 3600)
|
|
minutes, seconds = divmod(remainder, 60)
|
|
|
|
if hours:
|
|
return f"{hours}:{minutes:02d}:{seconds:02d}"
|
|
|
|
return f"{minutes}:{seconds:02d}"
|
|
|
|
return str(value)
|
|
|
|
|
|
def build_change_summary(
|
|
existing: sqlite3.Row,
|
|
site_values: dict[str, Any],
|
|
) -> str:
|
|
"""
|
|
Describe changes to fields owned by the site.
|
|
"""
|
|
changes: list[str] = []
|
|
|
|
for field_name in SITE_FIELDS:
|
|
old_value = existing[field_name]
|
|
new_value = site_values[field_name]
|
|
|
|
if old_value == new_value:
|
|
continue
|
|
|
|
old_text = format_value(
|
|
field_name,
|
|
old_value,
|
|
)
|
|
new_text = format_value(
|
|
field_name,
|
|
new_value,
|
|
)
|
|
|
|
summary = (
|
|
f"{field_name}: {old_text} -> {new_text}"
|
|
)
|
|
|
|
if (
|
|
field_name == "duration_seconds"
|
|
and old_value is not None
|
|
and new_value is not None
|
|
):
|
|
difference = int(new_value) - int(old_value)
|
|
sign = "+" if difference >= 0 else "-"
|
|
absolute_difference = abs(difference)
|
|
|
|
difference_text = format_value(
|
|
field_name,
|
|
absolute_difference,
|
|
)
|
|
|
|
summary += f" ({sign}{difference_text})"
|
|
|
|
changes.append(summary)
|
|
|
|
if not changes:
|
|
return "Site update date was newer, but no site fields changed."
|
|
|
|
return "; ".join(changes)
|
|
|
|
|
|
def archive_movie(
|
|
connection: sqlite3.Connection,
|
|
existing: sqlite3.Row,
|
|
change_summary: str,
|
|
) -> None:
|
|
"""
|
|
Copy the complete current movie row to movie_history.
|
|
|
|
This function is called only from the site import immediately before
|
|
the current movie row is updated.
|
|
"""
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO movie_history (
|
|
movie_id,
|
|
name,
|
|
aka,
|
|
nationality,
|
|
age,
|
|
shoot_location,
|
|
shoot_date,
|
|
duration_seconds,
|
|
description,
|
|
rating,
|
|
web_url,
|
|
thumbnail,
|
|
published,
|
|
updated,
|
|
created_at,
|
|
modified_at,
|
|
ocr_status,
|
|
ocr_raw_text,
|
|
ocr_error,
|
|
ocr_processed_at,
|
|
change_source,
|
|
change_summary
|
|
)
|
|
VALUES (
|
|
?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
|
|
?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
|
|
'site_import',
|
|
?
|
|
)
|
|
""",
|
|
(
|
|
existing["id"],
|
|
existing["name"],
|
|
existing["aka"],
|
|
existing["nationality"],
|
|
existing["age"],
|
|
existing["shoot_location"],
|
|
existing["shoot_date"],
|
|
existing["duration_seconds"],
|
|
existing["description"],
|
|
existing["rating"],
|
|
existing["web_url"],
|
|
existing["thumbnail"],
|
|
existing["published"],
|
|
existing["updated"],
|
|
existing["created_at"],
|
|
existing["modified_at"],
|
|
existing["ocr_status"],
|
|
existing["ocr_raw_text"],
|
|
existing["ocr_error"],
|
|
existing["ocr_processed_at"],
|
|
change_summary,
|
|
),
|
|
)
|
|
|
|
|
|
def update_movie(
|
|
connection: sqlite3.Connection,
|
|
movie_id: str,
|
|
site_values: dict[str, Any],
|
|
) -> None:
|
|
connection.execute(
|
|
"""
|
|
UPDATE movie
|
|
SET
|
|
name = ?,
|
|
duration_seconds = ?,
|
|
web_url = ?,
|
|
thumbnail = ?,
|
|
published = ?,
|
|
updated = ?,
|
|
modified_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(
|
|
site_values["name"],
|
|
site_values["duration_seconds"],
|
|
site_values["web_url"],
|
|
site_values["thumbnail"],
|
|
site_values["published"],
|
|
site_values["updated"],
|
|
movie_id,
|
|
),
|
|
)
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_arguments()
|
|
|
|
json_file = args.input
|
|
database_file = args.database
|
|
|
|
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
|
|
historized = 0
|
|
unchanged = 0
|
|
|
|
with sqlite3.connect(database_file) as connection:
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA foreign_keys = ON")
|
|
|
|
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)
|
|
|
|
site_values = build_site_values(
|
|
movie,
|
|
site_updated,
|
|
)
|
|
|
|
existing = connection.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
name,
|
|
aka,
|
|
nationality,
|
|
age,
|
|
shoot_location,
|
|
shoot_date,
|
|
duration_seconds,
|
|
description,
|
|
rating,
|
|
web_url,
|
|
thumbnail,
|
|
published,
|
|
updated,
|
|
created_at,
|
|
modified_at,
|
|
ocr_status,
|
|
ocr_raw_text,
|
|
ocr_error,
|
|
ocr_processed_at
|
|
FROM movie
|
|
WHERE id = ?
|
|
""",
|
|
(movie_id,),
|
|
).fetchone()
|
|
|
|
if existing is None:
|
|
insert_movie(
|
|
connection,
|
|
movie_id,
|
|
site_values,
|
|
)
|
|
|
|
inserted += 1
|
|
continue
|
|
|
|
database_updated = existing["updated"]
|
|
|
|
should_update = (
|
|
site_updated is not None
|
|
and (
|
|
database_updated is None
|
|
or site_updated > database_updated
|
|
)
|
|
)
|
|
|
|
if not should_update:
|
|
unchanged += 1
|
|
continue
|
|
|
|
change_summary = build_change_summary(
|
|
existing,
|
|
site_values,
|
|
)
|
|
|
|
archive_movie(
|
|
connection,
|
|
existing,
|
|
change_summary,
|
|
)
|
|
|
|
update_movie(
|
|
connection,
|
|
movie_id,
|
|
site_values,
|
|
)
|
|
|
|
historized += 1
|
|
updated += 1
|
|
|
|
print(f"Poster i JSON: {len(movies)}")
|
|
print(f"Tillagda: {inserted}")
|
|
print(f"Uppdaterade: {updated}")
|
|
print(f"Historiserade: {historized}")
|
|
print(f"Oförändrade: {unchanged}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except (
|
|
FileNotFoundError,
|
|
ValueError,
|
|
json.JSONDecodeError,
|
|
sqlite3.Error,
|
|
) as error:
|
|
raise SystemExit(f"Fel: {error}") from error
|
|
|