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()