93 lines
2.9 KiB
Python
93 lines
2.9 KiB
Python
import urllib.request
|
|
|
|
import imdb
|
|
import jsonfile
|
|
from jsonfile import PROJECT_ROOT
|
|
|
|
DETAILS_FILE = "mediaDetails.json"
|
|
POSTERS_DIR = PROJECT_ROOT / "static" / "covers" / "media"
|
|
POSTERS_PUBLIC_PATH = "/covers/media"
|
|
MISSING_VALUE = "N/A"
|
|
REQUEST_HEADERS = {"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0"}
|
|
|
|
|
|
def load_details() -> dict[str, dict]:
|
|
return jsonfile.load_data(DETAILS_FILE, {})
|
|
|
|
|
|
def save_details(details: dict[str, dict]) -> None:
|
|
jsonfile.save_data(DETAILS_FILE, dict(sorted(details.items())))
|
|
|
|
|
|
def present(value: str | None) -> str | None:
|
|
return value if value and value != MISSING_VALUE else None
|
|
|
|
|
|
def poster_path(identifier: str):
|
|
return POSTERS_DIR / f"{identifier}.jpg"
|
|
|
|
|
|
def download_poster(identifier: str, source_url: str | None) -> str | None:
|
|
if not source_url:
|
|
return None
|
|
request = urllib.request.Request(source_url, headers=REQUEST_HEADERS)
|
|
with urllib.request.urlopen(request, timeout=15) as response:
|
|
image = response.read()
|
|
POSTERS_DIR.mkdir(parents=True, exist_ok=True)
|
|
poster_path(identifier).write_bytes(image)
|
|
return f"{POSTERS_PUBLIC_PATH}/{identifier}.jpg"
|
|
|
|
|
|
def has_local_poster(identifier: str, stored: dict) -> bool:
|
|
poster = stored.get("poster")
|
|
return poster is None or (poster.startswith(POSTERS_PUBLIC_PATH) and poster_path(identifier).exists())
|
|
|
|
|
|
def has_details(identifier: str) -> bool:
|
|
stored = load_details().get(identifier)
|
|
return bool(stored) and has_local_poster(identifier, stored)
|
|
|
|
|
|
def fetch_details(identifier: str) -> dict:
|
|
result = imdb.lookup(identifier)
|
|
if not result:
|
|
raise ValueError(f"Could not look up {identifier} on OMDb. Check OMDB_API_KEY in .env")
|
|
|
|
poster_source = present(result.get("Poster"))
|
|
try:
|
|
poster = download_poster(identifier, poster_source)
|
|
except OSError:
|
|
poster = poster_source
|
|
|
|
stored = {
|
|
"title": present(result.get("Title")),
|
|
"poster": poster,
|
|
"posterSource": poster_source,
|
|
"year": present(result.get("Year")),
|
|
"genre": present(result.get("Genre")),
|
|
"runtime": present(result.get("Runtime")),
|
|
"plot": present(result.get("Plot")),
|
|
}
|
|
details = load_details()
|
|
details[identifier] = stored
|
|
save_details(details)
|
|
return stored
|
|
|
|
|
|
def cached_details(identifier: str, fresh: bool = False) -> dict:
|
|
if not fresh and has_details(identifier):
|
|
return load_details()[identifier]
|
|
return fetch_details(identifier)
|
|
|
|
|
|
def remove_details(identifier: str) -> None:
|
|
details = load_details()
|
|
if details.pop(identifier, None) is not None:
|
|
save_details(details)
|
|
if poster_path(identifier).exists():
|
|
poster_path(identifier).unlink()
|
|
|
|
|
|
def missing_identifiers(identifiers: list[str]) -> list[str]:
|
|
return [identifier for identifier in identifiers if not has_details(identifier)]
|