Add SQLite library with API, library desktop apps and MobyGames scraper

This commit is contained in:
SirBlob committed 2026-11-02 23:26:31 -05:00
1 parent 6674428346
commit a16444451a
69 files changed
+2988 -1026

No files matched your search

-39
View File
@@ -1,14 +1,9 @@
import json
import re
import urllib.request
import goodreads
from jsonfile import PROJECT_ROOT
SCRAPE_CACHE_DIR = PROJECT_ROOT / "tools" / "content" / ".cache" / "goodreads"
COVERS_DIR = PROJECT_ROOT / "static" / "covers" / "books"
COVERS_PUBLIC_PATH = "/covers/books"
IMAGE_EXTENSION_PATTERN = re.compile(r"\.(jpe?g|png|webp|gif)(?:$|\?)", re.I)
def scrape_cache_path(identifier: str):
@@ -39,37 +34,3 @@ def scrape_book(url: str, fresh: bool = False) -> dict:
if scraped.get("title"):
store_scrape(scraped)
return scraped
def is_local_cover(cover: str | None) -> bool:
return bool(cover) and cover.startswith(COVERS_PUBLIC_PATH)
def existing_cover(identifier: str):
return next(iter(sorted(COVERS_DIR.glob(f"{identifier}.*"))), None) if COVERS_DIR.exists() else None
def cache_cover(identifier: str, source_url: str | None, fresh: bool = False) -> str | None:
existing = existing_cover(identifier)
if existing and not fresh:
return f"{COVERS_PUBLIC_PATH}/{existing.name}"
if not source_url or is_local_cover(source_url):
return f"{COVERS_PUBLIC_PATH}/{existing.name}" if existing else None
match = IMAGE_EXTENSION_PATTERN.search(source_url)
extension = match.group(1).lower().replace("jpeg", "jpg") if match else "jpg"
request = urllib.request.Request(source_url, headers=goodreads.REQUEST_HEADERS)
with urllib.request.urlopen(request, timeout=15) as response:
image = response.read()
COVERS_DIR.mkdir(parents=True, exist_ok=True)
if existing:
existing.unlink()
(COVERS_DIR / f"{identifier}.{extension}").write_bytes(image)
return f"{COVERS_PUBLIC_PATH}/{identifier}.{extension}"
def remove_cover(identifier: str) -> None:
existing = existing_cover(identifier)
if existing:
existing.unlink()
+39 -104
View File
@@ -1,12 +1,7 @@
import datetime
import book_cache
import goodreads
import jsonfile
import library_api
BOOKS_FILE = "books.json"
BOOK_INFO_FILE = "bookDetails.json"
PERSONAL_KEYS = ["goodreadsId", "title", "url", "shelf", "rating", "added", "series"]
SHELVES = ["read", "reading", "want-to-read"]
DESCRIPTION_MAX_LENGTH = 600
GENRE_LIMIT = 5
@@ -25,55 +20,36 @@ def check_shelf(shelf: str) -> str:
def load_books() -> list[dict]:
if not jsonfile.content_path(BOOKS_FILE).exists():
return []
info = jsonfile.load_data(BOOK_INFO_FILE, {})
return [{**info.get(book["goodreadsId"], {}), **book} for book in jsonfile.load(BOOKS_FILE)]
return library_api.fetch_library()["books"]
def personal_fields(book: dict) -> dict:
return {key: book[key] for key in PERSONAL_KEYS if key in book}
def info_fields(book: dict) -> dict:
return {key: value for key, value in book.items() if key not in PERSONAL_KEYS}
def save_books(books: list[dict]) -> None:
jsonfile.save(BOOKS_FILE, [personal_fields(book) for book in books])
jsonfile.save_data(BOOK_INFO_FILE, {book["goodreadsId"]: info_fields(book) for book in books})
def find_index(books: list[dict], text: str) -> int | None:
def find_book(books: list[dict], text: str) -> dict | None:
identifier = goodreads.book_id(text) or (text if text.isdigit() else None)
for index, book in enumerate(books):
for book in books:
if identifier and book.get("goodreadsId") == identifier:
return index
return book
if book.get("title", "").lower() == text.lower():
return index
return book
return None
def listed_identifier(text: str) -> str:
book = find_book(load_books(), text)
if not book:
raise ValueError(f"{text} is not listed")
return book["goodreadsId"]
def short_description(text: str | None) -> str | None:
if not text or len(text) <= DESCRIPTION_MAX_LENGTH:
return text
return text[:DESCRIPTION_MAX_LENGTH].rsplit(" ", 1)[0] + "..."
def local_cover(scraped: dict) -> str | None:
try:
return book_cache.cache_cover(scraped["goodreadsId"], scraped["cover"]) or scraped["cover"]
except OSError:
return scraped["cover"]
def stored_fields(scraped: dict) -> dict:
def scraped_fields(scraped: dict) -> dict:
return {
"goodreadsId": scraped["goodreadsId"],
"url": scraped["url"],
"title": scraped["title"],
"authors": scraped["authors"],
"cover": local_cover(scraped),
"coverSource": scraped["cover"],
"description": short_description(scraped["description"]),
"pages": scraped["pages"],
@@ -84,14 +60,8 @@ def stored_fields(scraped: dict) -> dict:
}
def update_book(text: str, change) -> dict:
books = load_books()
index = find_index(books, text)
if index is None:
raise ValueError(f"{text} is not listed")
books[index] = change(books[index])
save_books(books)
return books[index]
def update_book(text: str, fields: dict) -> dict:
return library_api.save_book(listed_identifier(text), fields)
def series_tag(series: dict, entry: dict) -> dict:
@@ -104,92 +74,57 @@ def series_tag(series: dict, entry: dict) -> dict:
def add_series_book(series: dict, entry: dict, shelf: str) -> dict:
check_shelf(shelf)
tag = series_tag(series, entry)
books = load_books()
index = find_index(books, entry["goodreadsId"])
if index is not None:
books[index] = {**books[index], "shelf": shelf, "series": tag}
save_books(books)
return books[index]
book = add_book(entry["url"], shelf)
return update_book(book["goodreadsId"], lambda stored: {**stored, "series": tag})
if find_book(load_books(), entry["goodreadsId"]):
return library_api.save_book(entry["goodreadsId"], {"shelf": shelf, "series": tag})
return add_book(entry["url"], shelf, series=tag)
def library_shelves() -> dict[str, str]:
return {book["goodreadsId"]: book.get("shelf", "read") for book in load_books()}
def add_book(url: str, shelf: str = "read", rating: int | None = None) -> dict:
def add_book(url: str, shelf: str = "read", rating: int | None = None, series: dict | None = None) -> dict:
check_shelf(shelf)
if rating is not None:
check_rating(rating)
identifier = goodreads.book_id(url)
if not identifier:
raise ValueError(f"{url} is not a Goodreads book link")
if find_index(load_books(), identifier) is not None:
if find_book(load_books(), identifier):
raise ValueError(f"{identifier} is already listed")
scraped = book_cache.scrape_book(url)
if not scraped["title"]:
raise ValueError("Could not read the book from Goodreads")
book = stored_fields(scraped)
book["shelf"] = shelf
if rating is not None:
book["rating"] = rating
book["added"] = datetime.date.today().isoformat()
books = load_books()
books.append(book)
save_books(books)
return book
fields = {**scraped_fields(scraped), "shelf": shelf, "rating": rating}
if series:
fields["series"] = series
return library_api.save_book(identifier, fields)
def remove_book(text: str) -> dict:
books = load_books()
index = find_index(books, text)
if index is None:
raise ValueError(f"{text} is not listed")
removed = books.pop(index)
save_books(books)
book_cache.remove_cover(removed["goodreadsId"])
return removed
def remove_book(text: str) -> None:
library_api.delete_book(listed_identifier(text))
def rate_book(text: str, rating: int | None) -> dict:
if rating is not None:
check_rating(rating)
def change(book: dict) -> dict:
updated = {key: value for key, value in book.items() if key != "rating"}
if rating is not None:
updated["rating"] = rating
return updated
return update_book(text, change)
return update_book(text, {"rating": rating})
def move_book(text: str, shelf: str) -> dict:
check_shelf(shelf)
return update_book(text, lambda book: {**book, "shelf": shelf})
return update_book(text, {"shelf": shelf})
def set_dates(text: str, started: str | None, finished: str | None) -> dict:
return update_book(text, {"started": started, "finished": finished})
def refresh_book(text: str) -> dict:
books = load_books()
index = find_index(books, text)
if index is None:
book = find_book(load_books(), text)
if not book:
raise ValueError(f"{text} is not listed")
scraped = book_cache.scrape_book(books[index]["url"], fresh=True)
return update_book(text, lambda book: {**book, **stored_fields(scraped)})
scraped = book_cache.scrape_book(book["url"], fresh=True)
return library_api.save_book(book["goodreadsId"], scraped_fields(scraped))
def books_missing_local_covers() -> list[dict]:
return [book for book in load_books() if not book_cache.is_local_cover(book.get("cover"))]
def cache_book_cover(identifier: str) -> dict:
def change(book: dict) -> dict:
source = book.get("coverSource") or book.get("cover")
cover = book_cache.cache_cover(identifier, source)
return {**book, "cover": cover or book.get("cover"), "coverSource": source}
return update_book(identifier, change)
def store_book_cover(book: dict) -> dict:
return library_api.save_book(book["goodreadsId"], {"coverSource": book["coverSource"]})
+10
View File
@@ -0,0 +1,10 @@
import datetime
def check_date(text: str | None) -> str | None:
if not text:
return None
try:
return datetime.date.fromisoformat(text).isoformat()
except ValueError:
raise ValueError(f"{text} is not a date like 2026-01-31") from None
+17
View File
@@ -0,0 +1,17 @@
import os
from jsonfile import PROJECT_ROOT
ENV_FILE = PROJECT_ROOT / ".env"
def env_value(name: str, default: str | None = None) -> str | None:
if os.environ.get(name):
return os.environ[name]
if not ENV_FILE.exists():
return default
for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
key, _, value = line.partition("=")
if key.strip() == name and value.strip():
return value.strip().strip("\"'")
return default
+2 -47
View File
@@ -4,51 +4,25 @@ import html
import json
import re
import sys
import urllib.request
from pathlib import Path
from html_page import PageResult, download_page, meta_tags, page_title, plain_text
BOOK_URL_PATTERN = re.compile(r"goodreads\.com/(?:\w+/)?book/show/(\d+)")
JSON_LD_PATTERN = re.compile(r"<script[^>]*type=[\"']application/ld\+json[\"'][^>]*>(.*?)</script>", re.S | re.I)
NEXT_DATA_PATTERN = re.compile(r"<script[^>]*id=[\"']__NEXT_DATA__[\"'][^>]*>(.*?)</script>", re.S | re.I)
META_TAG_PATTERN = re.compile(r"<meta\s[^>]*>", re.S | re.I)
ATTRIBUTE_PATTERN = re.compile(r"([\w:-]+)=\"([^\"]*)\"|([\w:-]+)='([^']*)'")
PAGE_TITLE_PATTERN = re.compile(r"<title[^>]*>(.*?)</title>", re.S | re.I)
TAG_PATTERN = re.compile(r"<[^>]+>")
BLOCKED_PAGE_HINTS = ["captcha", "awswaf", "robot check", "access denied", "are you a human", "unusual traffic"]
REQUEST_HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0",
"Accept": "text/html,application/xhtml+xml",
"Accept-Language": "en-US,en;q=0.9",
}
class BlockedError(ValueError):
pass
class PageResult:
def __init__(self, status: int, final_url: str, text: str):
self.status = status
self.final_url = final_url
self.text = text
def book_id(url: str) -> str | None:
match = BOOK_URL_PATTERN.search(url)
return match.group(1) if match else None
def download_page(url: str) -> PageResult:
request = urllib.request.Request(url, headers=REQUEST_HEADERS)
with urllib.request.urlopen(request, timeout=15) as response:
return PageResult(response.status, response.geturl(), response.read().decode("utf-8", errors="replace"))
def page_title(page: str) -> str:
match = PAGE_TITLE_PATTERN.search(page)
return html.unescape(match.group(1)).strip() if match else ""
def blocked_reason(page: str) -> str | None:
lowered = page.lower()
hint = next((hint for hint in BLOCKED_PAGE_HINTS if hint in lowered), None)
@@ -71,25 +45,6 @@ def json_ld_book(page: str) -> dict:
return {}
def meta_tags(page: str) -> dict:
tags = {}
for tag in META_TAG_PATTERN.findall(page):
attributes = {}
for double_name, double_value, single_name, single_value in ATTRIBUTE_PATTERN.findall(tag):
attributes[(double_name or single_name).lower()] = double_value or single_value
name = attributes.get("property") or attributes.get("name")
if name and "content" in attributes:
tags[name] = html.unescape(attributes["content"])
return tags
def plain_text(text: str | None) -> str | None:
if not text:
return None
with_breaks = re.sub(r"<br\s*/?>", "\n", text)
return html.unescape(TAG_PATTERN.sub("", with_breaks)).strip() or None
def apollo_state(page: str) -> dict:
match = NEXT_DATA_PATTERN.search(page)
if not match:
+51
View File
@@ -0,0 +1,51 @@
import html
import re
import urllib.request
META_TAG_PATTERN = re.compile(r"<meta\s[^>]*>", re.S | re.I)
ATTRIBUTE_PATTERN = re.compile(r"([\w:-]+)=\"([^\"]*)\"|([\w:-]+)='([^']*)'")
PAGE_TITLE_PATTERN = re.compile(r"<title[^>]*>(.*?)</title>", re.S | re.I)
TAG_PATTERN = re.compile(r"<[^>]+>")
REQUEST_HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:140.0) Gecko/20100101 Firefox/140.0",
"Accept": "text/html,application/xhtml+xml",
"Accept-Language": "en-US,en;q=0.9",
}
class PageResult:
def __init__(self, status: int, final_url: str, text: str):
self.status = status
self.final_url = final_url
self.text = text
def download_page(url: str) -> PageResult:
request = urllib.request.Request(url, headers=REQUEST_HEADERS)
with urllib.request.urlopen(request, timeout=15) as response:
return PageResult(response.status, response.geturl(), response.read().decode("utf-8", errors="replace"))
def page_title(page: str) -> str:
match = PAGE_TITLE_PATTERN.search(page)
return html.unescape(match.group(1)).strip() if match else ""
def meta_tags(page: str) -> dict:
tags = {}
for tag in META_TAG_PATTERN.findall(page):
attributes = {}
for double_name, double_value, single_name, single_value in ATTRIBUTE_PATTERN.findall(tag):
attributes[(double_name or single_name).lower()] = double_value or single_value
name = attributes.get("property") or attributes.get("name")
if name and "content" in attributes:
tags[name] = html.unescape(attributes["content"])
return tags
def plain_text(text: str | None) -> str | None:
if not text:
return None
with_breaks = re.sub(r"<br\s*/?>", "\n", text)
return html.unescape(TAG_PATTERN.sub("", with_breaks)).strip() or None
+21 -16
View File
@@ -1,17 +1,16 @@
import json
import os
import re
import urllib.error
import urllib.parse
import urllib.request
from jsonfile import PROJECT_ROOT
from envfile import env_value
IMDB_ID_PATTERN = re.compile(r"tt\d{7,9}")
ENV_FILE = PROJECT_ROOT / ".env"
OMDB_URL = "https://www.omdbapi.com/"
OMDB_DEMO_URL = "http://www.omdbapi.com/demo.aspx/"
RATE_LIMITED_STATUS = 401
MISSING_VALUE = "N/A"
def imdb_id(text: str) -> str | None:
@@ -23,25 +22,13 @@ def title_url(identifier: str) -> str:
return f"https://www.imdb.com/title/{identifier}/"
def omdb_api_key() -> str | None:
if os.environ.get("OMDB_API_KEY"):
return os.environ["OMDB_API_KEY"]
if not ENV_FILE.exists():
return None
for line in ENV_FILE.read_text(encoding="utf-8").splitlines():
key, _, value = line.partition("=")
if key.strip() == "OMDB_API_KEY" and value.strip():
return value.strip().strip("\"'")
return None
def request_json(url: str, parameters: dict) -> dict:
with urllib.request.urlopen(f"{url}?{urllib.parse.urlencode(parameters)}", timeout=10) as response:
return json.load(response)
def lookup(identifier: str) -> dict | None:
api_key = omdb_api_key()
api_key = env_value("OMDB_API_KEY")
if not api_key:
return None
try:
@@ -54,3 +41,21 @@ def lookup(identifier: str) -> dict | None:
except (OSError, ValueError):
return None
return result if result.get("Response") == "True" else None
def present(value: str | None) -> str | None:
return value if value and value != MISSING_VALUE else None
def details(identifier: str) -> dict:
result = lookup(identifier)
if not result:
raise ValueError(f"Could not look up {identifier} on OMDb. Check OMDB_API_KEY in .env")
return {
"name": present(result.get("Title")),
"year": present(result.get("Year")),
"genre": present(result.get("Genre")),
"runtime": present(result.get("Runtime")),
"plot": present(result.get("Plot")),
"posterSource": present(result.get("Poster")),
}
+79
View File
@@ -0,0 +1,79 @@
import json
import urllib.error
import urllib.parse
import urllib.request
from envfile import env_value
DEFAULT_API_URL = "http://localhost:5000"
IMAGE_PATH_PREFIX = "/api/library/images/"
REQUEST_TIMEOUT_SECONDS = 60
class LibraryError(ValueError):
pass
def api_url() -> str:
return (env_value("LIBRARY_API_URL") or DEFAULT_API_URL).rstrip("/")
def api_key() -> str:
key = env_value("LIBRARY_API_KEY")
if not key:
raise LibraryError("Set LIBRARY_API_KEY in .env to change the library")
return key
def error_message(error: urllib.error.HTTPError) -> str:
try:
return json.loads(error.read().decode("utf-8")).get("message") or error.reason
except ValueError:
return str(error.reason)
def send(method: str, path: str, body: dict | None = None, signed: bool = True):
headers = {"Accept": "application/json"}
data = None
if signed:
headers["Authorization"] = f"Bearer {api_key()}"
if body is not None:
headers["Content-Type"] = "application/json"
data = json.dumps(body).encode("utf-8")
request = urllib.request.Request(f"{api_url()}{path}", data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
text = response.read().decode("utf-8")
except urllib.error.HTTPError as error:
raise LibraryError(f"{api_url()} said {error.code}: {error_message(error)}") from error
except urllib.error.URLError as error:
raise LibraryError(f"Could not reach {api_url()}: {error.reason}") from error
return json.loads(text) if text else None
def quoted(identifier: str) -> str:
return urllib.parse.quote(identifier, safe="")
def fetch_library() -> dict:
return send("GET", "/api/library", signed=False)
def save_media(imdb_id: str, fields: dict) -> dict:
return send("PUT", f"/api/library/media/{quoted(imdb_id)}", fields)
def delete_media(imdb_id: str) -> None:
send("DELETE", f"/api/library/media/{quoted(imdb_id)}")
def save_book(goodreads_id: str, fields: dict) -> dict:
return send("PUT", f"/api/library/books/{quoted(goodreads_id)}", fields)
def delete_book(goodreads_id: str) -> None:
send("DELETE", f"/api/library/books/{quoted(goodreads_id)}")
def is_stored_image(path: str | None) -> bool:
return bool(path) and path.startswith(IMAGE_PATH_PREFIX)
+50 -76
View File
@@ -1,8 +1,6 @@
import imdb
import jsonfile
import media_cache
import library_api
ABOUT_FILE = "about.json"
MEDIA_KINDS = {
"show": {"key": "shows", "label": "TV show", "omdb_type": "series"},
"movie": {"key": "movies", "label": "movie", "omdb_type": "movie"},
@@ -23,125 +21,101 @@ def check_status(status: str) -> str:
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 with_details(item: dict, details: dict) -> dict:
return {**item, "imdbId": imdb.imdb_id(item["url"]), "details": details.get(item["url"]) or {}}
def list_items(kind: str) -> list[dict]:
_, items = load_items(kind)
library = library_api.fetch_library()
items = [with_details(item, library["mediaDetails"]) for item in library[MEDIA_KINDS[kind]["key"]]]
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:
def find_item(items: list[dict], text: str) -> dict | None:
identifier = imdb.imdb_id(text)
for index, item in enumerate(items):
if identifier and imdb.imdb_id(item.get("url", "")) == identifier:
return index
for item in items:
if identifier and item["imdbId"] == identifier:
return item
if item.get("name", "").lower() == text.lower():
return index
return item
return None
def details_for(identifier: str, name: str | None) -> dict:
def listed_identifier(kind: str, text: str) -> str:
item = find_item(list_items(kind), text)
if not item:
raise ValueError(f"{text} is not listed")
return item["imdbId"]
def lookup_details(identifier: str, name: str | None) -> dict:
try:
return media_cache.cached_details(identifier)
except (OSError, ValueError):
return imdb.details(identifier)
except 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:
if find_item(list_items(kind), identifier):
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
details = lookup_details(identifier, name)
fields = {
**details,
"kind": kind,
"name": name or details.get("name") or identifier,
"status": status,
"rating": rating,
"collection": collection or None,
}
return library_api.save_media(identifier, fields)["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 update_item(kind: str, text: str, fields: dict) -> dict:
return library_api.save_media(listed_identifier(kind, text), fields)["item"]
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 remove_item(kind: str, text: str) -> None:
library_api.delete_media(listed_identifier(kind, text))
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))
return update_item(kind, text, {"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))
return update_item(kind, text, {"status": status})
def set_collection(kind: str, text: str, collection: str | None) -> dict:
return update_item(kind, text, lambda item: with_optional(item, "collection", collection))
return update_item(kind, text, {"collection": collection or None})
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 set_dates(kind: str, text: str, started: str | None, finished: str | None) -> dict:
return update_item(kind, text, {"started": started, "finished": finished})
def missing_details(kind: str) -> list[str]:
return media_cache.missing_identifiers(item_identifiers(kind))
def has_details(item: dict) -> bool:
return any(item["details"].get(key) for key in ["year", "genre", "plot"])
def has_stored_poster(item: dict) -> bool:
return library_api.is_stored_image(item["details"].get("poster"))
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)
details = {key: value for key, value in imdb.details(identifier).items() if key != "name"}
return library_api.save_media(identifier, details)
-92
View File
@@ -1,92 +0,0 @@
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)]
+209
View File
@@ -0,0 +1,209 @@
import argparse
import html
import json
import re
import sys
from pathlib import Path
import jsonfile
from html_page import download_page, meta_tags, page_title, plain_text
GAME_URL_PATTERN = re.compile(r"mobygames\.com/game/(\d+)")
HEADING_PATTERN = re.compile(r"<h1[^>]*>(.*?)</h1>", re.S | re.I)
METADATA_PATTERN = re.compile(r"<dt[^>]*>(.*?)</dt>\s*<dd[^>]*>(.*?)</dd>", re.S | re.I)
LINK_PATTERN = re.compile(r"<a\s[^>]*href=[\"']([^\"']*)[\"'][^>]*>(.*?)</a>", re.S | re.I)
DESCRIPTION_PATTERN = re.compile(r"<div[^>]*id=[\"']description-text[\"'][^>]*>(.*?)</div>", re.S | re.I)
COVER_PATTERN = re.compile(r"https://cdn\.mobygames\.com/covers/[^\"'\s)]+")
SCREENSHOT_PATTERN = re.compile(r"https://cdn\.mobygames\.com/screenshots/[^\"'\s)]+")
NUMBER_PATTERN = re.compile(r"\d+(?:\.\d+)?")
PERCENT_PATTERN = re.compile(r"(\d+)\s*%")
BLOCKED_PAGE_HINTS = ["just a moment", "cf-chl", "captcha", "access denied", "are you a human"]
DETAILS_FILE = "gameDetails.json"
DESCRIPTION_MAX_LENGTH = 600
class BlockedError(ValueError):
pass
def game_id(url: str) -> str | None:
match = GAME_URL_PATTERN.search(url)
return match.group(1) if match else None
def blocked_reason(page: str) -> str | None:
lowered = page.lower()
hint = next((hint for hint in BLOCKED_PAGE_HINTS if hint in lowered), None)
if hint and not METADATA_PATTERN.search(page):
return f"the page looks like a bot check (found '{hint}')"
return None
def metadata(page: str) -> dict[str, str]:
fields = {}
for label, value in METADATA_PATTERN.findall(page):
name = (plain_text(label) or "").strip().lower()
if name and name not in fields:
fields[name] = value
return fields
def field_starting_with(fields: dict[str, str], *prefixes: str) -> str:
return next((value for name, value in fields.items() if name.startswith(prefixes)), "")
def link_texts(fragment: str, path_part: str | None = None) -> list[str]:
names = []
for href, text in LINK_PATTERN.findall(fragment):
name = plain_text(text)
if name and (path_part is None or path_part in href) and name not in names:
names.append(name)
return names
def first_number(fragment: str) -> float | None:
match = NUMBER_PATTERN.search(plain_text(fragment) or "")
return float(match.group(0)) if match else None
def first_percent(fragment: str) -> int | None:
match = PERCENT_PATTERN.search(plain_text(fragment) or "")
return int(match.group(1)) if match else None
def short_description(text: str | None) -> str | None:
if not text or len(text) <= DESCRIPTION_MAX_LENGTH:
return text
return text[:DESCRIPTION_MAX_LENGTH].rsplit(" ", 1)[0] + "..."
def description_of(page: str, meta: dict) -> str | None:
match = DESCRIPTION_PATTERN.search(page)
text = plain_text(match.group(1)) if match else None
return short_description(text or meta.get("og:description") or meta.get("description"))
def title_of(page: str, meta: dict) -> str:
heading = HEADING_PATTERN.search(page)
title = plain_text(heading.group(1)) if heading else None
return title or meta.get("og:title") or page_title(page).split(" - ")[0]
def first_match(pattern: re.Pattern, page: str) -> str | None:
match = pattern.search(page)
return html.unescape(match.group(0)) if match else None
def parse_game(page: str, identifier: str, url: str) -> dict:
meta = meta_tags(page)
fields = metadata(page)
released = field_starting_with(fields, "released")
release_fields = "".join(value for name, value in fields.items() if name.startswith(("released", "releases", "platform")))
genre_fields = "".join(fields.get(name, "") for name in ["genre", "perspective", "gameplay", "setting"])
return {
"mobygamesId": identifier,
"url": url,
"title": title_of(page, meta),
"released": (link_texts(released) or [plain_text(released)])[0] if released else None,
"platforms": link_texts(release_fields, "/platform/"),
"developers": link_texts(field_starting_with(fields, "developer")),
"publishers": link_texts(field_starting_with(fields, "publisher")),
"genres": link_texts(genre_fields),
"mobyScore": first_number(field_starting_with(fields, "moby score")),
"criticScore": first_percent(field_starting_with(fields, "critic")),
"playerScore": first_number(field_starting_with(fields, "player")),
"description": description_of(page, meta),
"cover": first_match(COVER_PATTERN, page) or meta.get("og:image"),
"screenshot": first_match(SCREENSHOT_PATTERN, page),
}
def scrape_game(url: str) -> dict:
identifier = game_id(url)
if not identifier:
raise ValueError(f"not a MobyGames game url: {url}")
page = download_page(url)
reason = blocked_reason(page.text)
if reason:
raise BlockedError(f"MobyGames blocked the request ({reason}), wait a few minutes and try again")
return parse_game(page.text, identifier, url)
def save_game(game: dict) -> None:
details = jsonfile.load_data(DETAILS_FILE, {})
details[game["mobygamesId"]] = {key: value for key, value in game.items() if key != "mobygamesId"}
jsonfile.save_data(DETAILS_FILE, dict(sorted(details.items())))
def print_diagnostics(url: str, save_path: Path | None):
page = download_page(url)
print(f"Status {page.status}")
print(f"Final url {page.final_url}")
print(f"Size {len(page.text):,} characters")
print(f"Title {page_title(page.text) or '-'}")
print(f"Fields {', '.join(metadata(page.text)) or '-'}")
print(f"Meta tags {len(meta_tags(page.text))}")
print(f"Blocked {blocked_reason(page.text) or 'no sign of a bot check'}")
if save_path:
save_path.write_text(page.text, encoding="utf-8")
print(f"Saved html to {save_path}")
def print_summary(game: dict):
print(game["title"] or "unknown title")
print("-" * 60)
for label, key in [
("Released", "released"),
("Platforms", "platforms"),
("Developers", "developers"),
("Publishers", "publishers"),
("Genres", "genres"),
("Moby score", "mobyScore"),
("Critics", "criticScore"),
("Players", "playerScore"),
("Cover", "cover"),
("Screenshot", "screenshot"),
]:
value = game[key]
text = ", ".join(value) if isinstance(value, list) else value
print(f"{label:<11} {text if text not in (None, '') else '-'}")
if game["description"]:
print(f"\n{game['description']}")
missing = [key for key, value in game.items() if value in (None, "", [])]
if missing:
print(f"\nNot found on the page: {', '.join(missing)}")
def main() -> int:
parser = argparse.ArgumentParser(description="Scrape video game info from MobyGames")
parser.add_argument("urls", nargs="+", help="MobyGames game urls, for example https://www.mobygames.com/game/223472/hades-ii/")
parser.add_argument("--json", action="store_true", help="print the raw result as JSON")
parser.add_argument("--save", action="store_true", help=f"save the result into data/{DETAILS_FILE} for the site")
parser.add_argument("--debug", action="store_true", help="show what the downloaded page contains instead of parsing it")
parser.add_argument("--save-html", type=Path, help="with --debug, save the downloaded page to this file")
arguments = parser.parse_args()
exit_code = 0
for url in arguments.urls:
try:
if arguments.debug:
print_diagnostics(url, arguments.save_html)
continue
game = scrape_game(url)
except (OSError, ValueError) as error:
print(f"Error for {url}: {error}", file=sys.stderr)
exit_code = 1
continue
if arguments.save:
save_game(game)
if arguments.json:
print(json.dumps(game, indent="\t", ensure_ascii=False))
else:
print_summary(game)
print()
return exit_code
if __name__ == "__main__":
sys.exit(main())
+30 -13
View File
@@ -7,13 +7,13 @@ from textual.binding import Binding
from textual.containers import Horizontal, VerticalScroll
from textual.widgets import DataTable, Static
import book_cache
import books
import goodreads
import goodreads_series
import library_api
from tui.cache_indicator import cache_state, cache_text
from tui.content_pane import ContentPane
from tui.dialogs import ConfirmDialog, FormDialog, FormField
from tui.library_pane import LibraryPane
from tui.series_picker import SeriesPicker
REQUEST_DELAY_SECONDS = 3
@@ -29,6 +29,8 @@ def book_details(book: dict) -> str:
("Published", book.get("published")),
("Genres", ", ".join(book.get("genres") or [])),
("Added", book.get("added")),
("Started", book.get("started")),
("Finished", book.get("finished")),
("Cache", book_cache_state(book)),
]
lines = [f"[b]{escape(book.get('title', ''))}[/b]", escape(", ".join(book.get("authors") or [])), ""]
@@ -41,15 +43,16 @@ def book_details(book: dict) -> str:
def book_cache_state(book: dict) -> str:
has_details = bool(book.get("authors")) or bool(book.get("description"))
return cache_state(has_details, book_cache.is_local_cover(book.get("cover")))
return cache_state(has_details, library_api.is_stored_image(book.get("cover")))
class BooksPane(ContentPane):
class BooksPane(LibraryPane):
BINDINGS = [
Binding("a", "add", "Add"),
Binding("S", "add_series", "Add series"),
Binding("e", "rate", "Rate"),
Binding("s", "shelf", "Shelf"),
Binding("h", "dates", "Dates"),
Binding("d", "delete", "Delete"),
Binding("u", "refresh", "Refresh"),
Binding("U", "refresh_all", "Refresh all"),
@@ -57,7 +60,9 @@ class BooksPane(ContentPane):
Binding("r", "reload", "Reload"),
]
content_file = "books.json"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.books: dict[str, dict] = {}
def compose(self) -> ComposeResult:
with Horizontal(classes="split-layout"):
@@ -74,14 +79,19 @@ class BooksPane(ContentPane):
table.add_column("Cache", key="cache")
self.reload()
def current_shelves(self) -> dict[str, str]:
return {identifier: book.get("shelf", "read") for identifier, book in self.books.items()}
def book_by_id(self, identifier: str | None) -> dict | None:
return next((book for book in books.load_books() if book.get("goodreadsId") == identifier), None)
return self.books.get(identifier) if identifier else None
def reload(self, focus_key: str | None = None):
table = self.table()
table.clear()
shelf_order = {shelf: index for index, shelf in enumerate(books.SHELVES)}
sorted_books = sorted(books.load_books(), key=lambda book: (shelf_order.get(book.get("shelf"), 99), -(book.get("rating") or -1)))
loaded = self.load_entries(books.load_books)
self.books = {book["goodreadsId"]: book for book in loaded}
sorted_books = sorted(loaded, key=lambda book: (shelf_order.get(book.get("shelf"), 99), -(book.get("rating") or -1)))
for book in sorted_books:
rating = f"{book['rating']}/10" if book.get("rating") is not None else "-"
authors = ", ".join(book.get("authors") or [])
@@ -141,6 +151,13 @@ class BooksPane(ContentPane):
self.app.push_screen(FormDialog("Move to shelf", fields, "Move"), moved)
def action_dates(self):
book = self.book_by_id(self.selected_key())
if not book:
return
identifier = book["goodreadsId"]
self.edit_dates(book["title"], book, lambda started, finished: books.set_dates(identifier, started, finished), identifier)
def action_delete(self):
book = self.book_by_id(self.selected_key())
if not book:
@@ -173,7 +190,7 @@ class BooksPane(ContentPane):
@work(thread=True, exclusive=True, group="refresh")
def refresh_every_book(self):
refreshed = 0
all_books = books.load_books()
all_books = list(self.books.values())
for index, book in enumerate(all_books):
if index:
time.sleep(REQUEST_DELAY_SECONDS)
@@ -190,9 +207,9 @@ class BooksPane(ContentPane):
self.app.call_from_thread(self.notify, f"Refreshed {refreshed} of {len(all_books)} books")
def action_cache_covers(self):
missing = books.books_missing_local_covers()
missing = [book for book in self.books.values() if book.get("coverSource") and not library_api.is_stored_image(book.get("cover"))]
if not missing:
self.notify("Every cover is already saved in static/covers/books")
self.notify("Every cover is already saved in the library")
return
self.notify(f"Saving {len(missing)} covers...")
self.cache_missing_covers(missing)
@@ -202,7 +219,7 @@ class BooksPane(ContentPane):
saved = 0
for book in missing:
try:
books.cache_book_cover(book["goodreadsId"])
books.store_book_cover(book)
except (OSError, ValueError) as error:
self.app.call_from_thread(self.notify, f"{book['title']}: {error}", title="Skipped", severity="warning")
continue
@@ -225,12 +242,12 @@ class BooksPane(ContentPane):
self.notify(f"Adding {len(choices)} books from {series['name']}...")
self.add_series_books(series, choices)
self.app.push_screen(SeriesPicker(series, books.library_shelves()), picked)
self.app.push_screen(SeriesPicker(series, self.current_shelves()), picked)
@work(thread=True, exclusive=True, group="series")
def add_series_books(self, series: dict, choices: dict[str, str]):
added = 0
already_listed = set(books.library_shelves())
already_listed = set(self.current_shelves())
fetched_any = False
for entry in series["books"]:
shelf = choices.get(entry["goodreadsId"])
+2 -1
View File
@@ -7,6 +7,7 @@ from textual.widgets import DataTable
class ContentPane(Vertical):
content_file = ""
reload_message = "Reloaded from disk"
def __init__(self, **kwargs):
super().__init__(**kwargs)
@@ -40,7 +41,7 @@ class ContentPane(Vertical):
def action_reload(self):
self.reload(self.selected_key())
self.notify("Reloaded from disk")
self.notify(self.reload_message)
def run_now(self, task: Callable[[], Any], message: str | Callable[[Any], str], focus_key: str | None = None):
try:
+33
View File
@@ -0,0 +1,33 @@
from typing import Any, Callable
import library_api
from dates import check_date
from tui.content_pane import ContentPane
from tui.dialogs import FormDialog, FormField
class LibraryPane(ContentPane):
reload_message = "Reloaded from the library"
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.border_title = f"[b]kitty[/b] {library_api.api_url()}/api/library"
def load_entries(self, load: Callable[[], list[dict]]) -> list[dict]:
try:
return load()
except (OSError, ValueError) as error:
self.notify(str(error), title="Library", severity="error", timeout=10)
return []
def edit_dates(self, title: str, entry: dict, save: Callable[[str | None, str | None], Any], focus_key: str):
fields = [
FormField("started", "Started (YYYY-MM-DD, empty to clear)", value=entry.get("started", ""), placeholder="2026-01-31"),
FormField("finished", "Finished (YYYY-MM-DD, empty to clear)", value=entry.get("finished", ""), placeholder="2026-02-14"),
]
def saved(values: dict | None):
if values is not None:
self.run_now(lambda: save(check_date(values["started"]), check_date(values["finished"])), f"Saved dates for {title}", focus_key)
self.app.push_screen(FormDialog(f"History for {title}", fields), saved)
+34 -28
View File
@@ -8,10 +8,9 @@ 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
from tui.library_pane import LibraryPane
REQUEST_DELAY_SECONDS = 0.5
STATUS_STYLES = {"watching": "blue", "watched": "green", "planning": "magenta"}
@@ -21,25 +20,25 @@ def rating_text(item: dict) -> str:
return f"{item['rating']}/10" if item.get("rating") is not None else "-"
class MediaPane(ContentPane):
class MediaPane(LibraryPane):
BINDINGS = [
Binding("a", "add", "Add"),
Binding("e", "rate", "Rate"),
Binding("s", "status", "Status"),
Binding("g", "collection", "Group"),
Binding("h", "dates", "Dates"),
Binding("d", "delete", "Delete"),
Binding("u", "refresh", "Refresh"),
Binding("C", "cache_all", "Cache all"),
Binding("r", "reload", "Reload"),
]
content_file = "about.json"
def __init__(self, kind: str, **kwargs):
super().__init__(**kwargs)
self.kind = kind
self.label = media.MEDIA_KINDS[kind]["label"]
self.border_subtitle = media.MEDIA_KINDS[kind]["key"]
self.items: dict[str, dict] = {}
def compose(self) -> ComposeResult:
yield DataTable(cursor_type="row", zebra_stripes=True)
@@ -51,6 +50,7 @@ class MediaPane(ContentPane):
table.add_column("Title", key="title")
table.add_column("Group", key="collection")
table.add_column("Year", key="year")
table.add_column("Finished", key="finished")
table.add_column("Cache", key="cache")
table.add_column("IMDb", key="imdb")
self.reload()
@@ -58,30 +58,29 @@ class MediaPane(ContentPane):
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", "")
items = self.load_entries(lambda: media.list_items(self.kind))
self.items = {item["imdbId"]: item for item in items}
for item in items:
status = media.status_of(item)
details = stored.get(identifier)
state = cache_state(bool(details), bool(details) and media_cache.has_local_poster(identifier, details))
state = cache_state(media.has_details(item), media.has_stored_poster(item))
table.add_row(
Text(status, style=STATUS_STYLES[status]),
rating_text(item),
item.get("name", ""),
item.get("collection", ""),
(details or {}).get("year") or "-",
item["details"].get("year") or "-",
item.get("finished") or "-",
cache_text(state),
identifier,
key=identifier,
item["imdbId"],
key=item["imdbId"],
)
self.restore_cursor(focus_key)
def selected_item(self) -> dict | None:
identifier = self.selected_key()
return next((item for item in media.list_items(self.kind) if imdb.imdb_id(item.get("url", "")) == identifier), None)
return self.items.get(self.selected_key())
def collection_hint(self) -> str:
names = media.collection_names(self.kind)
names = sorted({item["collection"] for item in self.items.values() if item.get("collection")})
return f"Existing: {', '.join(names)}" if names else "e.g. Star Wars"
def action_add(self):
@@ -113,7 +112,7 @@ class MediaPane(ContentPane):
item = self.selected_item()
if not item:
return
identifier = imdb.imdb_id(item["url"])
identifier = item["imdbId"]
current = "" if item.get("rating") is None else str(item["rating"])
fields = [FormField("rating", f"Rating for {item['name']} (empty to clear)", kind="rating", value=current)]
@@ -127,7 +126,7 @@ class MediaPane(ContentPane):
item = self.selected_item()
if not item:
return
identifier = imdb.imdb_id(item["url"])
identifier = item["imdbId"]
fields = [FormField("status", f"Status for {item['name']}", kind="select", options=media.STATUSES, value=media.status_of(item))]
def changed(values: dict | None):
@@ -141,7 +140,7 @@ class MediaPane(ContentPane):
item = self.selected_item()
if not item:
return
identifier = imdb.imdb_id(item["url"])
identifier = item["imdbId"]
fields = [FormField("collection", f"Group for {item['name']} (empty to remove)", value=item.get("collection", ""), placeholder=self.collection_hint())]
def grouped(values: dict | None):
@@ -152,32 +151,39 @@ class MediaPane(ContentPane):
self.app.push_screen(FormDialog("Group titles", fields), grouped)
def action_delete(self):
identifier = self.selected_key()
if not identifier:
def action_dates(self):
item = self.selected_item()
if not item:
return
name = self.table().get_cell(identifier, "title")
identifier = item["imdbId"]
self.edit_dates(item["name"], item, lambda started, finished: media.set_dates(self.kind, identifier, started, finished), identifier)
def action_delete(self):
item = self.selected_item()
if not item:
return
identifier = item["imdbId"]
def confirmed(should_delete: bool | None):
if should_delete:
self.run_now(lambda: media.remove_item(self.kind, identifier), f"Removed {name}")
self.run_now(lambda: media.remove_item(self.kind, identifier), f"Removed {item['name']}")
self.app.push_screen(ConfirmDialog(f"Remove {name}?"), confirmed)
self.app.push_screen(ConfirmDialog(f"Remove {item['name']}?"), confirmed)
def action_refresh(self):
item = self.selected_item()
if not item:
return
identifier = imdb.imdb_id(item["url"])
identifier = item["imdbId"]
def refreshed(details: dict):
def refreshed(_saved: 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)
missing = [identifier for identifier, item in self.items.items() if not media.has_details(item) or not media.has_stored_poster(item)]
if not missing:
self.notify(f"Every {self.label} already has saved details and a poster")
return