148 lines
4.9 KiB
Python
148 lines
4.9 KiB
Python
import imdb
|
|
import jsonfile
|
|
import media_cache
|
|
|
|
ABOUT_FILE = "about.json"
|
|
MEDIA_KINDS = {
|
|
"show": {"key": "shows", "label": "TV show", "omdb_type": "series"},
|
|
"movie": {"key": "movies", "label": "movie", "omdb_type": "movie"},
|
|
}
|
|
STATUSES = ["watched", "watching", "planning"]
|
|
STATUS_ORDER = {"watching": 0, "watched": 1, "planning": 2}
|
|
|
|
|
|
def check_rating(rating: int | None) -> int | None:
|
|
if rating is not None and not 0 <= rating <= 10:
|
|
raise ValueError("Rating must be between 0 and 10")
|
|
return rating
|
|
|
|
|
|
def check_status(status: str) -> str:
|
|
if status not in STATUSES:
|
|
raise ValueError(f"Status must be one of {', '.join(STATUSES)}")
|
|
return status
|
|
|
|
|
|
def load_items(kind: str) -> tuple[dict, list[dict]]:
|
|
data = jsonfile.load(ABOUT_FILE)
|
|
return data, data.setdefault(MEDIA_KINDS[kind]["key"], [])
|
|
|
|
|
|
def status_of(item: dict) -> str:
|
|
return item.get("status", "watched")
|
|
|
|
|
|
def list_items(kind: str) -> list[dict]:
|
|
_, items = load_items(kind)
|
|
rating = lambda item: item["rating"] if item.get("rating") is not None else -1
|
|
return sorted(items, key=lambda item: (STATUS_ORDER[status_of(item)], -rating(item)))
|
|
|
|
|
|
def collection_names(kind: str) -> list[str]:
|
|
_, items = load_items(kind)
|
|
return sorted({item["collection"] for item in items if item.get("collection")})
|
|
|
|
|
|
def find_index(items: list[dict], text: str) -> int | None:
|
|
identifier = imdb.imdb_id(text)
|
|
for index, item in enumerate(items):
|
|
if identifier and imdb.imdb_id(item.get("url", "")) == identifier:
|
|
return index
|
|
if item.get("name", "").lower() == text.lower():
|
|
return index
|
|
return None
|
|
|
|
|
|
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")
|
|
|
|
|
|
def with_optional(item: dict, key: str, value) -> dict:
|
|
updated = {name: field for name, field in item.items() if name != key}
|
|
if value is not None and value != "":
|
|
updated[key] = value
|
|
return updated
|
|
|
|
|
|
def build_item(name: str, url: str, rating: int | None, status: str, collection: str | None) -> dict:
|
|
item = {"name": name, "url": url}
|
|
item = with_optional(item, "rating", rating)
|
|
if status != "watched":
|
|
item["status"] = status
|
|
return with_optional(item, "collection", collection)
|
|
|
|
|
|
def add_item(kind: str, text: str, rating: int | None, name: str | None = None, status: str = "watched", collection: str | None = None) -> dict:
|
|
identifier = imdb.imdb_id(text)
|
|
if not identifier:
|
|
raise ValueError(f"{text} is not an IMDb link or ID")
|
|
check_rating(rating)
|
|
check_status(status)
|
|
data, items = load_items(kind)
|
|
if find_index(items, identifier) is not None:
|
|
raise ValueError(f"{identifier} is already listed")
|
|
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
|
|
|
|
|
|
def update_item(kind: str, text: str, change) -> dict:
|
|
data, items = load_items(kind)
|
|
index = find_index(items, text)
|
|
if index is None:
|
|
raise ValueError(f"{text} is not listed")
|
|
items[index] = change(items[index])
|
|
jsonfile.save(ABOUT_FILE, data)
|
|
return items[index]
|
|
|
|
|
|
def remove_item(kind: str, text: str) -> dict:
|
|
data, items = load_items(kind)
|
|
index = find_index(items, text)
|
|
if index is None:
|
|
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
|
|
|
|
|
|
def rate_item(kind: str, text: str, rating: int | None) -> dict:
|
|
check_rating(rating)
|
|
return update_item(kind, text, lambda item: with_optional(item, "rating", rating))
|
|
|
|
|
|
def set_status(kind: str, text: str, status: str) -> dict:
|
|
check_status(status)
|
|
return update_item(kind, text, lambda item: with_optional(item, "status", None if status == "watched" else status))
|
|
|
|
|
|
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)
|