Save OMDb details and posters locally

tools/content/media.py

@@ -1,5 +1,6 @@
 import imdb
 import jsonfile
+import media_cache
 
 ABOUT_FILE = "about.json"
 MEDIA_KINDS = {
@@ -52,11 +53,13 @@ def find_index(items: list[dict], text: str) -> int | None:
     return None
 
 
-def look_up_name(identifier: str) -> str:
-    details = imdb.lookup(identifier)
-    if not details:
+def details_for(identifier: str, name: str | None) -> dict:
+    try:
+        return media_cache.cached_details(identifier)
+    except (OSError, ValueError):
+        if name:
+            return {}
         raise ValueError("Could not look up the title on OMDb. Enter a name, or set OMDB_API_KEY in .env")
-    return details.get("Title") or identifier
 
 
 def with_optional(item: dict, key: str, value) -> dict:
@@ -83,7 +86,8 @@ def add_item(kind: str, text: str, rating: int | None, name: str | None = None,
     data, items = load_items(kind)
     if find_index(items, identifier) is not None:
         raise ValueError(f"{identifier} is already listed")
-    item = build_item(name or look_up_name(identifier), imdb.title_url(identifier), rating, status, collection)
+    details = details_for(identifier, name)
+    item = build_item(name or details.get("title") or identifier, imdb.title_url(identifier), rating, status, collection)
     items.append(item)
     jsonfile.save(ABOUT_FILE, data)
     return item
@@ -106,6 +110,9 @@ def remove_item(kind: str, text: str) -> dict:
         raise ValueError(f"{text} is not listed")
     removed = items.pop(index)
     jsonfile.save(ABOUT_FILE, data)
+    identifier = imdb.imdb_id(removed.get("url", ""))
+    if identifier:
+        media_cache.remove_details(identifier)
     return removed
 
 
@@ -121,3 +128,20 @@ def set_status(kind: str, text: str, status: str) -> dict:
 
 def set_collection(kind: str, text: str, collection: str | None) -> dict:
     return update_item(kind, text, lambda item: with_optional(item, "collection", collection))
+
+
+def item_identifiers(kind: str) -> list[str]:
+    _, items = load_items(kind)
+    return [identifier for identifier in (imdb.imdb_id(item.get("url", "")) for item in items) if identifier]
+
+
+def missing_details(kind: str) -> list[str]:
+    return media_cache.missing_identifiers(item_identifiers(kind))
+
+
+def refresh_details(identifier: str) -> dict:
+    return media_cache.cached_details(identifier, fresh=True)
+
+
+def stored_details(identifier: str) -> dict | None:
+    return media_cache.load_details().get(identifier)

tools/content/media_cache.py

@@ -0,0 +1,92 @@
+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)]

tools/content/tui/media_pane.py

@@ -1,13 +1,19 @@
+import time
+
 from rich.text import Text
+from textual import work
 from textual.app import ComposeResult
 from textual.binding import Binding
 from textual.widgets import DataTable
 
 import imdb
 import media
+import media_cache
+from tui.cache_indicator import cache_state, cache_text
 from tui.content_pane import ContentPane
 from tui.dialogs import ConfirmDialog, FormDialog, FormField
 
+REQUEST_DELAY_SECONDS = 0.5
 STATUS_STYLES = {"watching": "blue", "watched": "green", "planning": "magenta"}
 
 
@@ -22,6 +28,8 @@ class MediaPane(ContentPane):
         Binding("s", "status", "Status"),
         Binding("g", "collection", "Group"),
         Binding("d", "delete", "Delete"),
+        Binding("u", "refresh", "Refresh"),
+        Binding("C", "cache_all", "Cache all"),
         Binding("r", "reload", "Reload"),
     ]
 
@@ -42,20 +50,27 @@ class MediaPane(ContentPane):
         table.add_column("Rating", key="rating")
         table.add_column("Title", key="title")
         table.add_column("Group", key="collection")
+        table.add_column("Year", key="year")
+        table.add_column("Cache", key="cache")
         table.add_column("IMDb", key="imdb")
         self.reload()
 
     def reload(self, focus_key: str | None = None):
         table = self.table()
         table.clear()
+        stored = media_cache.load_details()
         for item in media.list_items(self.kind):
             identifier = imdb.imdb_id(item.get("url", "")) or item.get("name", "")
             status = media.status_of(item)
+            details = stored.get(identifier)
+            state = cache_state(bool(details), bool(details) and media_cache.has_local_poster(identifier, details))
             table.add_row(
                 Text(status, style=STATUS_STYLES[status]),
                 rating_text(item),
                 item.get("name", ""),
                 item.get("collection", ""),
+                (details or {}).get("year") or "-",
+                cache_text(state),
                 identifier,
                 key=identifier,
             )
@@ -148,3 +163,38 @@ class MediaPane(ContentPane):
                 self.run_now(lambda: media.remove_item(self.kind, identifier), f"Removed {name}")
 
         self.app.push_screen(ConfirmDialog(f"Remove {name}?"), confirmed)
+
+    def action_refresh(self):
+        item = self.selected_item()
+        if not item:
+            return
+        identifier = imdb.imdb_id(item["url"])
+
+        def refreshed(details: dict):
+            self.reload(identifier)
+            self.notify(f"Saved details and poster for {item['name']}")
+
+        self.run_in_background(f"Looking up {item['name']} on OMDb...", lambda: media.refresh_details(identifier), refreshed)
+
+    def action_cache_all(self):
+        missing = media.missing_details(self.kind)
+        if not missing:
+            self.notify(f"Every {self.label} already has saved details and a poster")
+            return
+        self.notify(f"Saving details for {len(missing)} titles...")
+        self.cache_missing_details(missing)
+
+    @work(thread=True, exclusive=True, group="media-cache")
+    def cache_missing_details(self, missing: list[str]):
+        saved = 0
+        for index, identifier in enumerate(missing):
+            if index:
+                time.sleep(REQUEST_DELAY_SECONDS)
+            try:
+                media.refresh_details(identifier)
+            except (OSError, ValueError) as error:
+                self.app.call_from_thread(self.notify, f"{identifier}: {error}", title="Skipped", severity="warning")
+                continue
+            saved += 1
+        self.app.call_from_thread(lambda: self.reload(self.selected_key()))
+        self.app.call_from_thread(self.notify, f"Saved {saved} of {len(missing)} titles")