Cache Goodreads data and slow down bulk requests

tools/content/book_cache.py

@@ -0,0 +1,75 @@
+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):
+    return SCRAPE_CACHE_DIR / f"{identifier}.json"
+
+
+def cached_scrape(identifier: str) -> dict | None:
+    path = scrape_cache_path(identifier)
+    if not path.exists():
+        return None
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except ValueError:
+        return None
+
+
+def store_scrape(scraped: dict) -> None:
+    SCRAPE_CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    scrape_cache_path(scraped["goodreadsId"]).write_text(json.dumps(scraped, ensure_ascii=False, indent=2), encoding="utf-8")
+
+
+def scrape_book(url: str, fresh: bool = False) -> dict:
+    identifier = goodreads.book_id(url)
+    cached = None if fresh or not identifier else cached_scrape(identifier)
+    if cached:
+        return cached
+    scraped = goodreads.scrape_book(url)
+    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()