Spaces:
Running
Running
| """Author RAG Chatbot SaaS — Book URL Metadata Scraper (v3 — Never-Blocked Edition). | |
| Architecture: API-First, HTML-Last. | |
| Resolution pipeline for EVERY URL: | |
| 1. URL parsing — extract identifiers (ASIN/ISBN/volume-id/OL-key) from the URL itself | |
| 2. ASIN→ISBN — Amazon ASINs for books are often ISBN-10; convert to ISBN-13 instantly | |
| 3. Google Books API (by ISBN or volume-id) — comprehensive, free, NO auth, NEVER blocks | |
| 4. Open Library API (by ISBN or OL-key) — free, NO auth, NEVER blocks, adds author bio | |
| 5. Google Books search (title+author) — fallback if ISBN unavailable | |
| 6. Minimal HTML fetch (OG/JSON-LD only) — last resort; UA-rotated, retried 3× | |
| 7. Score & cache in Redis (6h TTL) | |
| Why this never gets blocked: | |
| - Steps 1–5 never touch the retailer's HTML at all for most books. | |
| - Google Books API + Open Library: both are public JSON APIs with no rate limits for | |
| reasonable usage; they return the same data as the product page, often richer. | |
| - HTML fallback (step 6) uses 10-rotation of realistic browser UA strings, proper | |
| Referer/Accept headers, retry on 403/429/503 with exponential backoff. | |
| - No Playwright / Selenium — pure httpx, zero extra dependencies. | |
| Supported platforms: | |
| 1. Amazon (amazon.com, .co.uk, .ca, .com.au, .de, .fr, .it, .es, .in, .co.jp) | |
| 2. Barnes & Noble (barnesandnoble.com) | |
| 3. Goodreads (goodreads.com) | |
| 4. Google Books (books.google.com / play.google.com/store/books) | |
| 5. Apple Books (books.apple.com) | |
| 6. Kobo (kobo.com) | |
| 7. ThriftBooks (thriftbooks.com) | |
| 8. AbeBooks (abebooks.com) | |
| 9. WorldCat (worldcat.org) | |
| 10. Open Library (openlibrary.org) | |
| 11. Bookshop.org (bookshop.org) | |
| 12. Lulu (lulu.com) | |
| 13. Walmart Books (walmart.com) | |
| 14. Waterstones (waterstones.com) | |
| 15. Books-A-Million (booksamillion.com) | |
| 16. Powell's (powells.com) | |
| 17. Indigo / Chapters (chapters.indigo.ca) | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import hashlib | |
| import html as html_module | |
| import json | |
| import os | |
| import re | |
| import structlog | |
| from dataclasses import asdict, dataclass, field | |
| from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse | |
| import httpx | |
| logger = structlog.get_logger(__name__) | |
| # ── Constants ────────────────────────────────────────────────────────────────── | |
| _CACHE_TTL = 21_600 # 6h Redis cache | |
| _CACHE_PREFIX = "book_v3:" | |
| _API_TIMEOUT = 10.0 # Google/OL JSON APIs | |
| _HTML_TIMEOUT = 14.0 # HTML pages — more lenient | |
| _MAX_DESC = 3000 | |
| _MAX_BIO = 500 | |
| _MAX_RETRIES = 3 # HTML retry attempts on block | |
| def _google_books_api_params(**extra: str) -> dict[str, str]: | |
| """Build Google Books API query params, including API key when configured.""" | |
| params = {k: str(v) for k, v in extra.items()} | |
| key = os.environ.get("GOOGLE_BOOKS_API_KEY", "").strip() | |
| if not key: | |
| try: | |
| from app.config import get_settings | |
| key = (get_settings().GOOGLE_BOOKS_API_KEY or "").strip() | |
| except Exception: | |
| pass | |
| if key: | |
| params["key"] = key | |
| return params | |
| # ── Rotating User-Agent pool ─────────────────────────────────────────────────── | |
| # Covers Chrome on Windows, Mac, Linux and Safari on Mac/iPhone. | |
| # Rotated on every HTML request and on retry. | |
| _USER_AGENTS = [ | |
| # Chrome 124 Windows | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| # Chrome 124 macOS | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", | |
| # Firefox 125 Windows | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0", | |
| # Firefox 125 macOS | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 14.4; rv:125.0) Gecko/20100101 Firefox/125.0", | |
| # Safari 17 macOS | |
| "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4_1) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15", | |
| # Edge 124 Windows | |
| "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36 Edg/124.0.0.0", | |
| # Chrome 123 Ubuntu | |
| "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36", | |
| # Chrome 122 Android | |
| "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.6367.82 Mobile Safari/537.36", | |
| # Safari iOS | |
| "Mozilla/5.0 (iPhone; CPU iPhone OS 17_4_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Mobile/15E148 Safari/604.1", | |
| # Googlebot (some sites serve clean HTML to bots — useful fallback) | |
| "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)", | |
| ] | |
| _ua_index = 0 | |
| def _next_ua() -> str: | |
| """Cycle through the UA pool.""" | |
| global _ua_index | |
| ua = _USER_AGENTS[_ua_index % len(_USER_AGENTS)] | |
| _ua_index += 1 | |
| return ua | |
| def _browser_headers(referer: str = "") -> dict: | |
| """Realistic browser headers with rotating UA.""" | |
| h = { | |
| "User-Agent": _next_ua(), | |
| "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8", | |
| "Accept-Language": "en-US,en;q=0.9", | |
| "Accept-Encoding": "gzip, deflate, br", | |
| "DNT": "1", | |
| "Connection": "keep-alive", | |
| "Upgrade-Insecure-Requests": "1", | |
| "Sec-Fetch-Dest": "document", | |
| "Sec-Fetch-Mode": "navigate", | |
| "Sec-Fetch-Site": "none", | |
| "Sec-Fetch-User": "?1", | |
| "Cache-Control": "max-age=0", | |
| } | |
| if referer: | |
| h["Referer"] = referer | |
| h["Sec-Fetch-Site"] = "same-origin" | |
| return h | |
| # ── Platform detection ───────────────────────────────────────────────────────── | |
| PLATFORM_PATTERNS: dict[str, str] = { | |
| "amazon": r"amazon\.(com|co\.uk|ca|com\.au|de|fr|it|es|in|co\.jp)", | |
| "barnes_noble": r"barnesandnoble\.com", | |
| "goodreads": r"goodreads\.com", | |
| "google_books": r"(books\.google\.|play\.google\.com/store/books)", | |
| "apple_books": r"books\.apple\.com", | |
| "kobo": r"kobo\.com", | |
| "thriftbooks": r"thriftbooks\.com", | |
| "abebooks": r"abebooks\.(com|co\.uk|de|fr|it|es)", | |
| "worldcat": r"worldcat\.org", | |
| "open_library": r"openlibrary\.org", | |
| "bookshop": r"bookshop\.org", | |
| "lulu": r"lulu\.com", | |
| "walmart": r"walmart\.com", | |
| "waterstones": r"waterstones\.com", | |
| "booksamillion": r"booksamillion\.com", | |
| "powells": r"powells\.com", | |
| "indigo": r"(chapters\.)?indigo\.ca", | |
| } | |
| PLATFORM_DISPLAY: dict[str, str] = { | |
| "amazon": "Amazon", "barnes_noble": "Barnes & Noble", | |
| "goodreads": "Goodreads", "google_books": "Google Books", | |
| "apple_books": "Apple Books", "kobo": "Kobo", | |
| "thriftbooks": "ThriftBooks", "abebooks": "AbeBooks", | |
| "worldcat": "WorldCat", "open_library": "Open Library", | |
| "bookshop": "Bookshop.org", "lulu": "Lulu", | |
| "walmart": "Walmart", "waterstones": "Waterstones", | |
| "booksamillion": "Books-A-Million", "powells": "Powell's", | |
| "indigo": "Indigo", "generic": "Author Website", | |
| } | |
| PLATFORM_ICONS: dict[str, str] = { | |
| "amazon": "🛍", "barnes_noble": "📕", "goodreads": "⭐", | |
| "google_books": "🔍", "apple_books": "🍎", "kobo": "📖", | |
| "thriftbooks": "💸", "abebooks": "🏗️", "worldcat": "🌍", | |
| "open_library": "🏗", "bookshop": "📚", "lulu": "🖨", | |
| "walmart": "🛒", "waterstones": "☕", "booksamillion": "📗", | |
| "powells": "🏪", "indigo": "🍁", "generic": "🔗", | |
| } | |
| def detect_platform(url: str) -> str: | |
| for platform, pattern in PLATFORM_PATTERNS.items(): | |
| if re.search(pattern, url.lower()): | |
| return platform | |
| return "unknown" | |
| # ── Data model ───────────────────────────────────────────────────────────────── | |
| class BookURLMetadata: | |
| title: str = "" | |
| author: str = "" | |
| about_author: str = "" | |
| description: str = "" | |
| genre: str = "" | |
| isbn: str = "" # canonical — prefers ISBN-13 | |
| isbn10: str = "" | |
| isbn13: str = "" | |
| asin: str = "" | |
| publisher: str = "" | |
| publish_year: str = "" | |
| edition: str = "" | |
| book_format: str = "" # paperback / hardcover / ebook / kindle | |
| series: str = "" | |
| language: str = "" | |
| page_count: int = 0 | |
| price: str = "" | |
| rating: str = "" | |
| rating_count: str = "" | |
| cover_url: str = "" | |
| platform: str = "" | |
| source_url: str = "" | |
| buy_url: str = "" | |
| confidence: float = 0.0 | |
| error: str = "" | |
| warnings: list = field(default_factory=list) | |
| # ── Text utilities ───────────────────────────────────────────────────────────── | |
| def _strip_html(text: str) -> str: | |
| text = re.sub(r"<br\s*/?>", "\n", text, flags=re.IGNORECASE) | |
| text = re.sub(r"<[^>]+>", "", text) | |
| return re.sub(r"[ \t]+", " ", text).strip() | |
| def _strip_scripts(html_fragment: str) -> str: | |
| """Remove <script>, <style>, and <noscript> blocks then strip remaining tags. | |
| Amazon embeds JS initialisation code directly inside content divs, which | |
| bleeds through _strip_html because that only removes individual tags, not | |
| full script blocks and their contents. | |
| """ | |
| # Remove full script / style / noscript blocks (content + tags) | |
| cleaned = re.sub(r"<script[^>]*>.*?</script>", "", html_fragment, flags=re.DOTALL | re.IGNORECASE) | |
| cleaned = re.sub(r"<style[^>]*>.*?</style>", "", cleaned, flags=re.DOTALL | re.IGNORECASE) | |
| cleaned = re.sub(r"<noscript[^>]*>.*?</noscript>", "", cleaned, flags=re.DOTALL | re.IGNORECASE) | |
| return _strip_html(cleaned) | |
| def _extract_meta(html: str, *names: str) -> str: | |
| for name in names: | |
| for pat in [ | |
| rf'<meta\s[^>]*(?:name|property)=["\'](?:og:|twitter:)?{re.escape(name)}["\']\s[^>]*content=["\'](.*?)["\']', | |
| rf'<meta\s[^>]*content=["\'](.*?)["\']\s[^>]*(?:name|property)=["\'](?:og:|twitter:)?{re.escape(name)}["\']', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE | re.DOTALL) | |
| if m: | |
| return _strip_html(m.group(1)).strip() | |
| return "" | |
| def _extract_json_ld(html: str) -> list[dict]: | |
| results = [] | |
| for m in re.finditer( | |
| r'<script\s[^>]*type=["\']application/ld\+json["\'][^>]*>(.*?)</script>', | |
| html, re.DOTALL | re.IGNORECASE | |
| ): | |
| try: | |
| obj = json.loads(m.group(1).strip()) | |
| results.extend(obj if isinstance(obj, list) else [obj]) | |
| except Exception: | |
| pass | |
| return results | |
| def _find_isbn(text: str) -> str: | |
| m = re.search(r"97[89][- ]?(?:\d[- ]?){9}\d", text) | |
| if m: | |
| candidate = re.sub(r"[- ]", "", m.group(0)) | |
| if _is_valid_isbn13(candidate): | |
| return candidate | |
| m = re.search(r"\b(?:ISBN[:\s-]*)?(\d{9}[\dX])\b", text, re.IGNORECASE) | |
| if m and _is_valid_isbn10(m.group(1)): | |
| return m.group(1) | |
| return "" | |
| def _is_valid_isbn13(isbn: str) -> bool: | |
| """Return True only when the 13-digit string passes the ISBN-13 checksum.""" | |
| if not re.fullmatch(r"97[89]\d{10}", isbn): | |
| return False | |
| total = sum(int(d) * (1 if i % 2 == 0 else 3) for i, d in enumerate(isbn[:12])) | |
| return (10 - (total % 10)) % 10 == int(isbn[12]) | |
| def _is_valid_isbn10(isbn: str) -> bool: | |
| """Return True only when the 10-digit string passes the ISBN-10 checksum.""" | |
| if not re.fullmatch(r"\d{9}[\dX]", isbn.upper()): | |
| return False | |
| total = sum((10 - i) * (10 if isbn[i] == "X" else int(isbn[i])) for i in range(10)) | |
| return total % 11 == 0 | |
| def _is_hex_hash(text: str) -> bool: | |
| """True when text looks like a Bookshop-style internal hash slug.""" | |
| cleaned = text.replace("-", "").replace(" ", "") | |
| return bool(re.fullmatch(r"[a-f0-9]{12,}", cleaned, re.IGNORECASE)) | |
| def _is_valid_author(name: str) -> bool: | |
| """Reject empty, URL, or obviously garbage author strings.""" | |
| if not name or len(name.strip()) < 2: | |
| return False | |
| low = name.lower().strip() | |
| if low.startswith("http") or "://" in low or "www." in low: | |
| return False | |
| if low in {"visit amazon", "amazon", "follow", "see all", "kindle store", "books"}: | |
| return False | |
| return True | |
| def _clean_title(title: str) -> str: | |
| """Normalize titles polluted by URL slugs or page paths.""" | |
| title = (title or "").strip() | |
| if not title: | |
| return "" | |
| title = re.sub(r"\.html?$", "", title, flags=re.IGNORECASE) | |
| title = title.replace("_", " ") | |
| title = re.sub(r"\s+", " ", title).strip() | |
| if _is_hex_hash(title): | |
| return "" | |
| return title | |
| def _is_latin_text(text: str) -> bool: | |
| """True when text is mostly Latin-script (English-friendly display).""" | |
| letters = [c for c in text if c.isalpha()] | |
| if not letters: | |
| return False | |
| latin = sum(1 for c in letters if ord(c) < 0x250) | |
| return latin / len(letters) >= 0.8 | |
| def _author_tokens(name: str) -> set[str]: | |
| """Lowercase word tokens from an author name for fuzzy matching.""" | |
| return {t for t in re.findall(r"[a-z]+", name.lower()) if len(t) > 1} | |
| def _authors_match(hint: str, candidate: str) -> bool: | |
| """True when surname or given-name tokens overlap between hint and candidate.""" | |
| if not hint or not candidate: | |
| return True | |
| hint_t = _author_tokens(hint) | |
| cand_t = _author_tokens(candidate) | |
| if not hint_t: | |
| return True | |
| return bool(hint_t & cand_t) | |
| def _resolve_author(catalog_author: str, slug_author: str) -> str: | |
| """Pick the best author string using URL slug hints vs catalog data.""" | |
| slug_author = (slug_author or "").strip() | |
| catalog_author = (catalog_author or "").strip() | |
| if slug_author and _is_latin_text(slug_author): | |
| if not catalog_author: | |
| return slug_author | |
| if not _is_latin_text(catalog_author): | |
| return slug_author | |
| if not _authors_match(slug_author, catalog_author): | |
| return slug_author | |
| if slug_author.lower() not in catalog_author.lower(): | |
| # Catalog lists translators/editors — keep URL primary author | |
| if len(catalog_author) > len(slug_author) + 10: | |
| return slug_author | |
| if catalog_author and _is_valid_author(catalog_author): | |
| return catalog_author | |
| return slug_author or "" | |
| def _score_gbvolume(volume: dict, title: str, author_hint: str) -> float: | |
| """Rank Google Books hits — prefer title match, author hint, English, recency.""" | |
| info = volume.get("volumeInfo", {}) | |
| score = 0.0 | |
| vtitle = (info.get("title") or "").lower() | |
| t = title.lower() | |
| if vtitle == t: | |
| score += 10.0 | |
| elif vtitle.rstrip("s") == t.rstrip("s") and vtitle != t: | |
| score -= 6.0 | |
| elif t in vtitle or vtitle in t: | |
| score += 5.0 | |
| else: | |
| score -= 8.0 | |
| for name in info.get("authors") or []: | |
| if author_hint and _authors_match(author_hint, name): | |
| score += 15.0 | |
| if info.get("language") in ("en", "eng"): | |
| score += 3.0 | |
| year = str(info.get("publishedDate", ""))[:4] | |
| if year.isdigit(): | |
| score += (int(year) - 1900) / 50.0 | |
| score += min(int(info.get("ratingsCount") or 0) / 5000.0, 5.0) | |
| return score | |
| def _score_ol_doc(doc: dict, title: str, author_hint: str) -> float: | |
| """Rank Open Library search hits.""" | |
| score = 0.0 | |
| dt = (doc.get("title") or "").lower() | |
| t = title.lower() | |
| if dt == t: | |
| score += 10.0 | |
| elif dt.rstrip("s") == t.rstrip("s"): | |
| score -= 6.0 | |
| elif t in dt or dt in t: | |
| score += 3.0 | |
| else: | |
| score -= 8.0 | |
| for name in doc.get("author_name") or []: | |
| if author_hint and _authors_match(author_hint, name): | |
| score += 15.0 | |
| elif author_hint and not _authors_match(author_hint, name): | |
| score -= 5.0 | |
| langs = doc.get("language") or [] | |
| if "eng" in langs or "en" in langs: | |
| score += 4.0 | |
| year = int(doc.get("first_publish_year") or 0) | |
| if year >= 2015: | |
| score += (year - 2015) / 2.0 | |
| else: | |
| score += year / 200.0 | |
| score += min(int(doc.get("ratings_count") or 0) / 200.0, 6.0) | |
| return score | |
| def _synthesize_description(result: dict) -> str: | |
| """Build a minimal description when no catalog blurb exists (indie/self-published).""" | |
| title = result.get("title", "").strip() | |
| if not title: | |
| return "" | |
| parts = [f"{title}"] | |
| if result.get("author"): | |
| parts[0] += f" by {result['author']}" | |
| parts[0] += "." | |
| extras = [] | |
| if result.get("publisher"): | |
| extras.append(f"Published by {result['publisher']}.") | |
| if result.get("page_count"): | |
| extras.append(f"{result['page_count']} pages.") | |
| if result.get("publish_year"): | |
| extras.append(f"Published in {result['publish_year']}.") | |
| text = " ".join([parts[0], *extras]) | |
| return text[:_MAX_DESC] if len(text) >= 40 else "" | |
| def _title_looks_bad(title: str) -> bool: | |
| """True when a scraped title should be replaced by catalog API data.""" | |
| title = _clean_title(title) | |
| if not title or len(title) < 2: | |
| return True | |
| if _is_hex_hash(title): | |
| return True | |
| if re.search(r"\.html?$", title, re.IGNORECASE): | |
| return True | |
| if re.match(r"^\d+q\d+[_\s]", title, re.IGNORECASE): | |
| return True | |
| low = title.lower() | |
| if low in {"imageblock", "producttitle", "booktitle", "title"}: | |
| return True | |
| # Amazon/React internal component ids (e.g. imageBlock, bylineInfo) | |
| if re.fullmatch(r"[a-z]+[A-Z][a-zA-Z0-9]*", title): | |
| return True | |
| if "kindle edition" in low and len(title) > 80: | |
| return True | |
| return False | |
| def _normalize_isbn(value: str) -> str: | |
| """Return a checksum-valid ISBN-10/13 or empty string.""" | |
| raw = re.sub(r"[-\s]", "", (value or "").strip()) | |
| if len(raw) == 13 and _is_valid_isbn13(raw): | |
| return raw | |
| if len(raw) == 10 and _is_valid_isbn10(raw.upper()): | |
| return raw.upper() | |
| return "" | |
| def _isbn13_to_isbn10(isbn13: str) -> str: | |
| """Convert a 978-prefixed ISBN-13 to ISBN-10 when possible.""" | |
| raw = re.sub(r"[-\s]", "", (isbn13 or "").strip()) | |
| if not raw.startswith("978") or len(raw) != 13 or not _is_valid_isbn13(raw): | |
| return "" | |
| body = raw[3:12] | |
| total = sum((10 - i) * int(d) for i, d in enumerate(body)) | |
| check = (11 - (total % 11)) % 11 | |
| return body + ("X" if check == 10 else str(check)) | |
| def _extract_isbn_pair(text: str) -> tuple[str, str]: | |
| """Extract ISBN-10 and ISBN-13 from arbitrary text or a single identifier.""" | |
| if not text: | |
| return "", "" | |
| isbn10 = "" | |
| isbn13 = "" | |
| for m in re.finditer( | |
| r"ISBN[- ]?13[:\s]*([0-9][- 0-9]{11,17}[0-9X])", text, re.IGNORECASE, | |
| ): | |
| candidate = _normalize_isbn(re.sub(r"[-\s]", "", m.group(1))) | |
| if len(candidate) == 13: | |
| isbn13 = candidate | |
| for m in re.finditer( | |
| r"ISBN[- ]?10[:\s]*([0-9][- 0-9]{8,12}[0-9X])", text, re.IGNORECASE, | |
| ): | |
| candidate = _normalize_isbn(re.sub(r"[-\s]", "", m.group(1))) | |
| if len(candidate) == 10: | |
| isbn10 = candidate | |
| if not isbn13: | |
| m = re.search(r"97[89][- ]?(?:\d[- ]?){9}\d", text) | |
| if m: | |
| candidate = re.sub(r"[- ]", "", m.group(0)) | |
| if _is_valid_isbn13(candidate): | |
| isbn13 = candidate | |
| if not isbn10: | |
| m = re.search(r"\b(?:ISBN[:\s-]*)?(\d{9}[\dX])\b", text, re.IGNORECASE) | |
| if m and _is_valid_isbn10(m.group(1)): | |
| isbn10 = m.group(1).upper() | |
| single = _normalize_isbn(text) | |
| if single: | |
| if len(single) == 13 and not isbn13: | |
| isbn13 = single | |
| elif len(single) == 10 and not isbn10: | |
| isbn10 = single | |
| if isbn10 and not isbn13: | |
| isbn13 = _isbn10_to_isbn13(isbn10) | |
| elif isbn13 and not isbn10: | |
| isbn10 = _isbn13_to_isbn10(isbn13) | |
| return isbn10, isbn13 | |
| def _apply_isbn_pair(result: dict, isbn10: str = "", isbn13: str = "") -> None: | |
| """Write isbn10/isbn13 and canonical isbn into a flat result dict.""" | |
| if isbn10: | |
| result["isbn10"] = isbn10 | |
| if isbn13: | |
| result["isbn13"] = isbn13 | |
| canonical = isbn13 or isbn10 or result.get("isbn", "") | |
| if canonical: | |
| result["isbn"] = canonical | |
| def _merge_isbn_fields(result: dict) -> None: | |
| """Normalize isbn / isbn10 / isbn13 keys on a resolver result dict.""" | |
| i10 = str(result.get("isbn10", "") or "") | |
| i13 = str(result.get("isbn13", "") or "") | |
| raw = str(result.get("isbn", "") or "") | |
| if not i10 and not i13 and raw: | |
| i10, i13 = _extract_isbn_pair(raw) | |
| elif i10 or i13: | |
| _apply_isbn_pair(result, i10, i13) | |
| return | |
| _apply_isbn_pair(result, i10, i13) | |
| _TRACKING_QUERY_PARAMS = frozenset({ | |
| "tag", "ref", "psc", "linkcode", "creative", "creativeasin", | |
| "ie", "keywords", "qid", "sr", "sprefix", "th", "dib", "dib_tag", | |
| "utm_source", "utm_medium", "utm_campaign", "utm_content", "utm_term", | |
| }) | |
| def _normalize_import_url(url: str) -> str: | |
| """Canonicalize book import URLs for stable cache keys and ASIN extraction.""" | |
| url = (url or "").strip() | |
| if not url: | |
| return url | |
| parsed = urlparse(url) | |
| host = (parsed.netloc or "").lower() | |
| if host.startswith("www."): | |
| host = host[4:] | |
| kept = [ | |
| (k, v) for k, v in parse_qsl(parsed.query, keep_blank_values=True) | |
| if k.lower() not in _TRACKING_QUERY_PARAMS | |
| ] | |
| path = parsed.path or "" | |
| m = re.search(r"/(?:dp|gp/product)/([A-Z0-9]{10})", path, re.IGNORECASE) | |
| if m and "amazon." in host: | |
| asin = m.group(1).upper() | |
| path = re.sub( | |
| r"/(?:dp|gp/product)/[A-Z0-9]{10}[^/]*", | |
| f"/dp/{asin}", | |
| path, | |
| count=1, | |
| flags=re.IGNORECASE, | |
| ) | |
| scheme = parsed.scheme or "https" | |
| return urlunparse((scheme, host, path.rstrip("/") or path, "", urlencode(kept), "")) | |
| def _is_placeholder_cover(url: str) -> bool: | |
| """True when cover URL is a generic placeholder, not a real jacket.""" | |
| if not url: | |
| return True | |
| low = url.lower() | |
| return any( | |
| token in low | |
| for token in ( | |
| "avatar_book", "no_cover", "no-cover", "nocover", | |
| "placeholder", "default_cover", "image-not-available", | |
| ) | |
| ) | |
| def _normalize_genre(genre: str) -> str: | |
| """Deduplicate and trim noisy genre/subject strings.""" | |
| if not genre: | |
| return "" | |
| parts: list[str] = [] | |
| for chunk in re.split(r"[,/]", genre): | |
| label = chunk.strip() | |
| if not label or len(label) < 2: | |
| continue | |
| if not _genre_label_ok(label): | |
| continue | |
| if label.lower() in {p.lower() for p in parts}: | |
| continue | |
| parts.append(label) | |
| return ", ".join(parts[:6]) | |
| def _genre_label_ok(label: str) -> bool: | |
| """True when a genre/subject label is worth showing to authors.""" | |
| low = label.lower().strip() | |
| if len(low) < 2 or len(low) > 80: | |
| return False | |
| if low.startswith("nyt:") or "fictitious character" in low: | |
| return False | |
| if re.search(r"=\d{4}-\d{2}-\d{2}", label): | |
| return False | |
| if "bestseller" in low or "enslaved" in low: | |
| return False | |
| if re.match(r"^(serie|series|saga|collection)\s*:", low): | |
| return False | |
| if not _is_latin_text(label): | |
| return False | |
| _FOREIGN = {"del", "las", "los", "der", "die", "und", "het", "ein", | |
| "une", "des", "les", "el", "la", "il", "et"} | |
| if _FOREIGN & set(low.split()): | |
| return False | |
| if low in {"accessible book", "protected daisy", "in library", "overdrive", "nook", "kindle"}: | |
| return False | |
| if re.fullmatch(r"[A-Z][a-z]+", label) and low not in { | |
| "fiction", "classics", "poetry", "fantasy", "romance", "horror", "mystery", "thriller", | |
| }: | |
| return False | |
| return True | |
| def _genre_looks_noisy(genre: str) -> bool: | |
| """True when scraped genres are catalog noise rather than real categories.""" | |
| if not genre: | |
| return False | |
| labels = [g.strip() for g in genre.split(",") if g.strip()] | |
| if not labels: | |
| return False | |
| for label in labels: | |
| low = label.lower() | |
| if ( | |
| low.startswith("nyt:") | |
| or "fictitious character" in low | |
| or "enslaved persons" in low | |
| or re.search(r"=\d{4}-\d{2}-\d{2}", label) | |
| ): | |
| return True | |
| ok_count = sum(1 for label in labels if _genre_label_ok(label)) | |
| return ok_count < max(1, len(labels) // 2) | |
| def _clean_retailer_title(title: str) -> str: | |
| """Strip store suffixes and decode HTML entities from retailer page titles.""" | |
| title = html_module.unescape((title or "").strip()) | |
| title = re.sub(r"\s*\|[^|]*$", "", title) | |
| title = re.sub(r"®", "", title).strip() | |
| return _clean_title(title) | |
| def _extract_page_meta_description(html: str) -> str: | |
| """Prefer meta name=description over og:description (B&N og:description is site boilerplate).""" | |
| for pat in ( | |
| r'<meta\s+name="description"\s+content="([^"]*)"', | |
| r'<meta\s+content="([^"]*)"\s+name="description"', | |
| ): | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| return html_module.unescape(m.group(1)).strip() | |
| return "" | |
| def _contributor_slug_to_name(slug: str) -> str: | |
| """Convert B&N contributor URL slug to a display name.""" | |
| if not slug: | |
| return "" | |
| return " ".join(part.capitalize() for part in slug.split("-") if part) | |
| def _is_retailer_cover(url: str) -> bool: | |
| """True when cover URL came from a retailer CDN (prefer over catalog fallbacks).""" | |
| if not url: | |
| return False | |
| low = url.lower() | |
| return any( | |
| token in low | |
| for token in ( | |
| "cdn.shopify.com/s/files", | |
| "m.media-amazon.com/images", | |
| "images-na.ssl-images-amazon.com", | |
| "images.amazon.com", | |
| ) | |
| ) | |
| def _decode_json_string(text: str) -> str: | |
| """Decode escaped JSON / JS string blobs from retailer HTML.""" | |
| if not text: | |
| return "" | |
| text = text.replace('\\"', '"').replace("\\'", "'") | |
| text = text.replace("\\n", "\n").replace("\\r", "").replace("\\t", "\t") | |
| text = re.sub(r"\\(?=\s)", "", text) | |
| text = text.replace("\\", "") | |
| return html_module.unescape(text) | |
| def _description_looks_bad(text: str) -> bool: | |
| """True when description is retailer boilerplate or corrupted escapes.""" | |
| if not text or len(text.strip()) < 40: | |
| return True | |
| low = text.lower() | |
| if _looks_like_structured_data(text): | |
| return True | |
| if "barnesandnoble.com" in low and "online bookstore" in low: | |
| return True | |
| if "shop music, movies, toys" in low: | |
| return True | |
| if "great selection of related books" in low or "available now at abebooks" in low: | |
| return True | |
| if "a great selection of new" in low and "used" in low: | |
| return True | |
| if text.count("\\") >= 4: | |
| return True | |
| if "\ufffd" in text: | |
| return True | |
| if re.search(r"\\ +\S", text): | |
| return True | |
| return False | |
| def _sanitize_book_description(text: str) -> str: | |
| """Strip retailer boilerplate and fix escaped characters in descriptions.""" | |
| if not text: | |
| return "" | |
| if _looks_like_structured_data(text): | |
| return "" | |
| d = _decode_json_string(text) | |
| d = _strip_html(d).strip() | |
| d = re.sub( | |
| r"^Read\s+[\d,]+\s+reviews?\s+from\s+the\s+world.+?readers\.\s*", | |
| "", | |
| d, | |
| flags=re.IGNORECASE, | |
| ) | |
| d = re.sub( | |
| r"^(?:Buy a cheap copy of|Free Shipping on all orders).+$", | |
| "", | |
| d, | |
| flags=re.IGNORECASE | re.MULTILINE, | |
| ) | |
| # Pull-quote blurbs: keep text but normalize whitespace around em-dashes | |
| d = re.sub(r"\s*—\s*", " — ", d) | |
| d = re.sub(r"[ \t]+", " ", d) | |
| d = re.sub(r"\n{3,}", "\n\n", d) | |
| if len(d) < 40 or _description_looks_bad(d): | |
| return "" | |
| return d[:_MAX_DESC] | |
| def _looks_like_structured_data(text: str) -> bool: | |
| """True when text is leaked JSON-LD / schema.org markup, not real prose.""" | |
| if not text: | |
| return False | |
| low = text.lower() | |
| markers = ("@context", "schema.org", '"@type"', "@type", "reviewbody", | |
| "reviewrating", "itemreviewed", '"datepublished"', "aggregaterating") | |
| hits = sum(1 for m in markers if m in low) | |
| if hits >= 2: | |
| return True | |
| if text.lstrip().startswith("{") and ("@" in text or '":' in text): | |
| return True | |
| return False | |
| def _valid_publish_year(year: str) -> str: | |
| """Return a plausible 4-digit publication year, else empty string.""" | |
| m = re.search(r"\b(1[4-9]\d{2}|20[0-4]\d)\b", str(year or "")) | |
| return m.group(1) if m else "" | |
| def _trim_prose(text: str, max_len: int = _MAX_BIO) -> str: | |
| """Trim prose to a display-friendly length, preferring sentence boundaries.""" | |
| text = (text or "").strip() | |
| if len(text) <= max_len: | |
| return text | |
| cut = text[:max_len] | |
| dot = max(cut.rfind(". "), cut.rfind("! "), cut.rfind("? ")) | |
| if dot >= max_len // 2: | |
| return cut[: dot + 1].strip() | |
| return cut.rstrip(" ,;") + "…" | |
| def _sanitize_about_author(bio: str, description: str = "") -> str: | |
| """Reject author bios that are actually book blurbs or review snippets.""" | |
| bio = _strip_html(bio or "").strip() | |
| if len(bio) < 50: | |
| return "" | |
| if _looks_like_structured_data(bio): | |
| return "" | |
| low = bio.lower() | |
| if "visit " in low and "barnes" in low and "page" in low: | |
| return "" | |
| if "shop all" in low and "books" in low and "explore books by author" in low: | |
| return "" | |
| if "online bookstore" in low and "barnes" in low: | |
| return "" | |
| if description and bio[:120] == description[:120]: | |
| return "" | |
| if re.match(r"^Read\s+[\d,]+\s+reviews?", bio, re.IGNORECASE): | |
| return "" | |
| if len(bio) > 400 and description and bio[:200] in description: | |
| return "" | |
| return _trim_prose(bio, _MAX_BIO) | |
| def _first_author_name(author: str) -> str: | |
| """First credited author for bio lookup (co-authors separated by and/,).""" | |
| if not author: | |
| return "" | |
| chunk = re.split(r",|\band\b", author, maxsplit=1)[0].strip() | |
| if "," in chunk and not chunk.lower().startswith("http"): | |
| parts = [p.strip() for p in chunk.split(",", 1)] | |
| if len(parts) == 2 and len(parts[0]) < len(parts[1]): | |
| chunk = f"{parts[1]} {parts[0]}" | |
| return chunk | |
| def _bn_slug_title_author(slug: str) -> tuple[str, str]: | |
| """Parse B&N /w/{slug}/{id} URLs — author is usually the last two hyphen segments.""" | |
| parts = [p for p in slug.lower().split("-") if p and not (len(p) == 13 and p.startswith("978"))] | |
| if len(parts) < 2: | |
| return slug.replace("-", " ").title(), "" | |
| for n in (2, 3): | |
| if len(parts) <= n: | |
| continue | |
| author = " ".join(parts[-n:]).title() | |
| title = " ".join(parts[:-n]).title() | |
| if title and _is_valid_author(author): | |
| return title, author | |
| return parts[0].title(), " ".join(parts[1:]).title() | |
| def _bn_shopify_cover_url(isbn: str) -> str: | |
| """B&N Shopify CDN cover pattern (works without scraping the product page).""" | |
| normalized = _normalize_isbn(isbn) | |
| if not normalized: | |
| return "" | |
| return f"https://cdn.shopify.com/s/files/1/0674/5433/7265/files/{normalized}_p0.jpg" | |
| async def _fetch_bn_html(url: str, client: httpx.AsyncClient) -> str: | |
| """Fetch B&N product HTML — Googlebot first, then standard UA rotation.""" | |
| for ua in ( | |
| "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)", | |
| _next_ua(), | |
| _next_ua(), | |
| ): | |
| try: | |
| headers = _browser_headers() | |
| headers["User-Agent"] = ua | |
| r = await client.get(url, headers=headers, timeout=_HTML_TIMEOUT) | |
| if r.status_code == 200 and len(r.text) > 5000: | |
| if "barnesandnoble" in r.text.lower() and ( | |
| "shopify" in r.text.lower() or "og:title" in r.text.lower() | |
| ): | |
| return r.text | |
| except Exception: | |
| continue | |
| return await _fetch_html(url, client) | |
| def _amazon_html_looks_blocked(html: str) -> bool: | |
| """True when Amazon returned a captcha / continue-shopping interstitial.""" | |
| if not html or len(html) < 15_000: | |
| low = html.lower() | |
| if "opfcaptcha" in low or "continue shopping" in low: | |
| return True | |
| if 'id="producttitle"' not in low and '"producttitle"' not in low: | |
| return len(html) < 15_000 | |
| return False | |
| def _amazon_product_urls(url: str, asin: str = "") -> list[str]: | |
| """Build Amazon product-page URL candidates, preferring formats that bypass /dp/ captchas.""" | |
| asin = (asin or _extract_identifiers(url).get("asin", "")).upper() | |
| host = "www.amazon.com" | |
| m = re.search(r"(amazon\.(?:com|co\.uk|ca|com\.au|de|fr|it|es|in|co\.jp))", url, re.I) | |
| if m: | |
| host = m.group(1).lower() | |
| if not host.startswith("www.") and not host.startswith("m."): | |
| host = f"www.{host}" | |
| candidates: list[str] = [] | |
| if asin: | |
| candidates.extend([ | |
| f"https://{host}/gp/product/{asin}", | |
| f"https://{host}/gp/aw/d/{asin}", | |
| f"https://m.{host.replace('www.', '')}/dp/{asin}", | |
| f"https://{host}/dp/{asin}", | |
| ]) | |
| norm = _normalize_import_url(url) | |
| if norm and norm not in candidates: | |
| candidates.append(norm) | |
| if url.split("?")[0] not in candidates: | |
| candidates.append(url.split("?")[0]) | |
| seen: set[str] = set() | |
| ordered: list[str] = [] | |
| for candidate in candidates: | |
| key = candidate.lower().rstrip("/") | |
| if key not in seen: | |
| seen.add(key) | |
| ordered.append(candidate) | |
| return ordered | |
| async def _fetch_amazon_html(url: str, client: httpx.AsyncClient, asin: str = "") -> str: | |
| """Fetch Amazon product HTML, trying alternate URL shapes when /dp/ is captcha-blocked.""" | |
| best = "" | |
| for candidate in _amazon_product_urls(url, asin): | |
| html = await _fetch_html(candidate, client) | |
| if not html: | |
| continue | |
| if not _amazon_html_looks_blocked(html): | |
| logger.debug("Amazon HTML fetched", url=candidate[:80], size=len(html)) | |
| return html | |
| if len(html) > len(best): | |
| best = html | |
| if best: | |
| logger.debug("Amazon HTML fallback (partial)", size=len(best)) | |
| return best | |
| def _title_author_from_url_slug(url: str) -> tuple[str, str]: | |
| """Guess title and author from common retailer URL slug patterns.""" | |
| path = url.split("?")[0].lower() | |
| m = re.search(r"thriftbooks\.com/w/([^/]+)/", path, re.IGNORECASE) | |
| if m: | |
| raw = m.group(1) | |
| if "_" in raw: | |
| title_part, author_part = raw.split("_", 1) | |
| return ( | |
| title_part.replace("-", " ").title(), | |
| author_part.replace("-", " ").title(), | |
| ) | |
| return raw.replace("-", " ").replace("_", " ").title(), "" | |
| m = re.search(r"lulu\.com/shop/([^/]+)/([^/]+)/", path, re.IGNORECASE) | |
| if m: | |
| return m.group(2).replace("-", " ").title(), m.group(1).replace("-", " ").title() | |
| m = re.search(r"bookshop\.org/(?:p/)?books/([^/]+)/", path, re.IGNORECASE) | |
| if m: | |
| slug = m.group(1) | |
| if not _is_hex_hash(slug): | |
| slug = re.sub(r"-\d+$", "", slug) | |
| parts = [p for p in slug.split("-") if p] | |
| if len(parts) >= 4: | |
| return ( | |
| " ".join(parts[:-2]).title(), | |
| " ".join(parts[-2:]).title(), | |
| ) | |
| if len(parts) == 3: | |
| return parts[0].title(), " ".join(parts[1:]).title() | |
| return " ".join(parts).title(), "" | |
| m = re.search(r"barnesandnoble\.com/w/([^/]+)/", path, re.IGNORECASE) | |
| if m: | |
| return _bn_slug_title_author(m.group(1)) | |
| # Amazon — /Title-Words-Slug/dp/ASIN (common for Kindle ebooks) | |
| m = re.search(r"amazon\.[^/]+/([^/]+)/dp/[a-z0-9]{10}", path, re.IGNORECASE) | |
| if m: | |
| slug = m.group(1) | |
| if slug.lower() not in {"dp", "gp", "product"}: | |
| slug = re.sub(r"-(ebook|kindle|paperback|hardcover|audiobook)$", "", slug, flags=re.I) | |
| parts = [p for p in slug.split("-") if p] | |
| if parts: | |
| return " ".join(parts).title(), "" | |
| m = re.search(r"booksamillion\.com/p/[^/]+/([^/]+)/", path, re.IGNORECASE) | |
| if m: | |
| return "", m.group(1).replace("-", " ").title() | |
| slug = "" | |
| for pat in ( | |
| r"/w/([^/]+)/\d", | |
| r"/ebook/([^/]+)", | |
| r"/book/([^/]+)/id\d", | |
| r"/shop/[^/]+/([^/]+)/", | |
| ): | |
| m = re.search(pat, path, re.IGNORECASE) | |
| if m: | |
| slug = m.group(1) | |
| break | |
| if not slug or _is_hex_hash(slug): | |
| return "", "" | |
| slug = re.sub(r"-?\d{13}$", "", slug) | |
| slug = re.sub(r"-?\d{10}$", "", slug) | |
| slug = re.sub(r"-\d+$", "", slug) | |
| slug = slug.strip("-") | |
| parts = [p for p in slug.split("-") if p and p not in ("ebook", "paperback", "hardcover")] | |
| if not parts: | |
| return "", "" | |
| if len(parts) < 3: | |
| return " ".join(parts).title(), "" | |
| author_start = len(parts) | |
| for i in range(len(parts) - 2, 0, -1): | |
| if len(parts[i]) == 1 and parts[i].isalpha(): | |
| author_start = i | |
| break | |
| if author_start < len(parts) - 1: | |
| title = " ".join(parts[:author_start]).title() | |
| author = " ".join(parts[author_start:]).title() | |
| return title, author | |
| if len(parts) <= 3: | |
| return " ".join(parts).title(), "" | |
| author_tokens = parts[-2:] | |
| title_tokens = parts[:-2] | |
| return " ".join(title_tokens).title(), " ".join(author_tokens).title() | |
| def _is_digital_listing(meta: BookURLMetadata) -> bool: | |
| """True for Kindle/ebook listings where ISBN/publisher/price are often absent.""" | |
| fmt = (meta.book_format or "").lower() | |
| asin = (meta.asin or "").upper() | |
| return "kindle" in fmt or "ebook" in fmt or asin.startswith("B0") | |
| def _score(meta: BookURLMetadata) -> float: | |
| """Completeness score — 100% when every field applicable to this listing is filled. | |
| Physical books expect ISBN + publisher; Kindle/B0-ASIN listings use ASIN as the | |
| identifier and omit publisher/price from the denominator when not on the page. | |
| """ | |
| weights = { | |
| "title": 0.20, | |
| "author": 0.12, | |
| "description": 0.15, | |
| "cover_url": 0.10, | |
| "identifier": 0.10, | |
| "genre": 0.06, | |
| "about_author": 0.04, | |
| "page_count": 0.05, | |
| "language": 0.04, | |
| "book_format": 0.04, | |
| "series": 0.02, | |
| "publisher": 0.04, | |
| "price": 0.04, | |
| } | |
| digital = _is_digital_listing(meta) | |
| def satisfied(field: str) -> bool: | |
| if field == "identifier": | |
| return bool(meta.isbn13 or meta.isbn10 or meta.isbn or meta.asin) | |
| if field == "page_count": | |
| return int(meta.page_count or 0) > 0 | |
| return bool(getattr(meta, field, "")) | |
| def applicable(field: str) -> bool: | |
| if field == "series": | |
| return bool(meta.series) | |
| if field == "publisher": | |
| return bool(meta.publisher) or not digital | |
| if field == "price": | |
| return bool(meta.price) or not digital | |
| return True | |
| earned = 0.0 | |
| total = 0.0 | |
| for field, weight in weights.items(): | |
| if not applicable(field): | |
| continue | |
| total += weight | |
| if satisfied(field): | |
| earned += weight | |
| if total <= 0: | |
| return 0.0 | |
| return round(min(earned / total, 1.0), 2) | |
| # ── Step 1: URL → Identifiers (zero HTTP calls) ─────────────────────────────── | |
| def _extract_identifiers(url: str) -> dict: | |
| """Extract all book identifiers directly from the URL — no HTTP needed.""" | |
| ids: dict = {} | |
| # Amazon ASIN: /dp/XXXXXXXXXX or /gp/product/XXXXXXXXXX | |
| m = re.search(r"/(?:dp|gp/product)/([A-Z0-9]{10})", url) | |
| if m: | |
| ids["asin"] = m.group(1) | |
| # Google Books volume ID: id=XXXX in query string | |
| m = re.search(r"[?&]id=([A-Za-z0-9_-]{8,})", url) | |
| if m: | |
| ids["google_volume_id"] = m.group(1) | |
| # Play Store Books: /details/xxx?id=XXXX | |
| m = re.search(r"play\.google\.com/store/books/details/[^?]+\?id=([A-Za-z0-9_-]+)", url) | |
| if m: | |
| ids["google_volume_id"] = m.group(1) | |
| # Open Library work key: /works/OL123W | |
| m = re.search(r"/works/(OL\d+W)", url) | |
| if m: | |
| ids["ol_work_key"] = m.group(1) | |
| # Open Library edition key: /books/OL123M | |
| m = re.search(r"/books/(OL\d+M)", url) | |
| if m: | |
| ids["ol_edition_key"] = m.group(1) | |
| # ISBN directly in URL (some platforms embed it) | |
| isbn = _find_isbn(url) | |
| if isbn: | |
| ids["isbn"] = isbn | |
| # AbeBooks — /products/isbn/9781234567890 | |
| m = re.search( | |
| r"abebooks\.com/products/isbn/(\d{13}|\d{9}[\dX])(?:[/?#]|$)", | |
| url, re.IGNORECASE, | |
| ) | |
| if m and not ids.get("isbn"): | |
| candidate = m.group(1) | |
| normalized = _normalize_isbn(candidate) | |
| if normalized: | |
| ids["isbn"] = normalized | |
| # Goodreads book ID: /book/show/12345 | |
| m = re.search(r"goodreads\.com/book/show/(\d+)", url) | |
| if m: | |
| ids["goodreads_id"] = m.group(1) | |
| # Barnes & Noble EAN: /w/title/XXXXXXXXXX (often a product ID, not ISBN) | |
| m = re.search(r"barnesandnoble\.com/w/[^/]+/(\d{10,13})", url) | |
| if m: | |
| ean = m.group(1) | |
| if len(ean) == 13 and _is_valid_isbn13(ean): | |
| ids["isbn"] = ean | |
| elif len(ean) == 10 and _is_valid_isbn10(ean): | |
| ids["isbn10"] = ean | |
| # Kobo — ISBN often embedded in /ebook/{slug} path | |
| m = re.search(r"kobo\.com/[^?]*(?:ebook|book)/([^/?#]+)", url, re.IGNORECASE) | |
| if m and not ids.get("isbn"): | |
| isbn = _find_isbn(m.group(1)) | |
| if isbn: | |
| ids["isbn"] = isbn | |
| if not ids.get("isbn"): | |
| m = re.search( | |
| r"kobo\.com/(?:\w+/){1,3}(?:ebook|book)/(?:[^/]+-)?(\d{10,13})", | |
| url, re.IGNORECASE, | |
| ) | |
| if m: | |
| candidate = m.group(1) | |
| if len(candidate) == 13 and _is_valid_isbn13(candidate): | |
| ids["isbn"] = candidate | |
| # Apple Books internal ID: /book/.../id1234567890 | |
| m = re.search(r"books\.apple\.com/[^/]+/book/[^/]+/id(\d+)", url, re.IGNORECASE) | |
| if m: | |
| ids["apple_book_id"] = m.group(1) | |
| # Bookshop.org — ISBN often in path: /books/9781234567890-title | |
| m = re.search(r"bookshop\.org/(?:books|a|p)/(?:[^/]+/)?(\d{10,13})", url, re.IGNORECASE) | |
| if m and not ids.get("isbn"): | |
| candidate = m.group(1) | |
| if len(candidate) == 13 and _is_valid_isbn13(candidate): | |
| ids["isbn"] = candidate | |
| elif len(candidate) == 10 and _is_valid_isbn10(candidate): | |
| ids["isbn10"] = candidate | |
| # Lulu — ISBN in product URL slug or query | |
| m = re.search(r"lulu\.com/shop/[^/]+/[^/]+/paperback/product-(\d{10,13})", url, re.IGNORECASE) | |
| if m and not ids.get("isbn"): | |
| ids["isbn"] = m.group(1) | |
| # Walmart / Waterstones / BAM — ISBN in URL path segments | |
| for retailer in (r"walmart\.com", r"waterstones\.com", r"booksamillion\.com", r"powells\.com"): | |
| m = re.search(retailer + r"/[^?]*?(\d{13}|\d{9}[\dX])(?:[/?#]|$)", url, re.IGNORECASE) | |
| if m and not ids.get("isbn"): | |
| candidate = m.group(1) | |
| if len(candidate) == 13 and _is_valid_isbn13(candidate): | |
| ids["isbn"] = candidate | |
| elif _is_valid_isbn10(candidate): | |
| ids["isbn10"] = candidate | |
| break | |
| m = re.search(r"indigo\.ca/[^?]*?(\d{13})(?:[/?#]|$)", url, re.IGNORECASE) | |
| if m and not ids.get("isbn") and _is_valid_isbn13(m.group(1)): | |
| ids["isbn"] = m.group(1) | |
| # Bookshop.org ?ean= query parameter | |
| m = re.search(r"[?&]ean=(\d{13})", url, re.IGNORECASE) | |
| if m and not ids.get("isbn") and _is_valid_isbn13(m.group(1)): | |
| ids["isbn"] = m.group(1) | |
| # WorldCat OCLC number: /title/1302336640 | |
| m = re.search(r"worldcat\.org/title/(\d+)", url, re.IGNORECASE) | |
| if m: | |
| ids["oclc"] = m.group(1) | |
| return ids | |
| # ── Step 2: ASIN → ISBN-13 conversion ───────────────────────────────────────── | |
| def _asin_to_isbn13(asin: str) -> str: | |
| """Convert an Amazon ASIN (which for books is often ISBN-10) to ISBN-13. | |
| Books sold on Amazon before ~2007 use ISBN-10 ASINs (all digits or ends in X). | |
| This lets us skip Amazon HTML entirely and go straight to the free APIs. | |
| """ | |
| if not re.match(r"^\d{9}[\dX]$", asin.upper()): | |
| return "" # ASIN is not an ISBN-10 (it's a non-book ASIN) | |
| body = "978" + asin[:9] | |
| total = sum(int(d) * (1 if i % 2 == 0 else 3) for i, d in enumerate(body)) | |
| check = (10 - (total % 10)) % 10 | |
| return body + str(check) | |
| def _isbn10_to_isbn13(isbn10: str) -> str: | |
| """Convert a valid ISBN-10 to ISBN-13.""" | |
| if not _is_valid_isbn10(isbn10): | |
| return "" | |
| return _asin_to_isbn13(isbn10.upper()) | |
| # ── Step 3–4: Google Books + Open Library APIs (never blocked) ──────────────── | |
| def _parse_google_volume(volume: dict) -> dict: | |
| """Parse a Google Books volume object into a flat metadata dict.""" | |
| info = volume.get("volumeInfo", {}) | |
| isbn13 = "" | |
| isbn10 = "" | |
| for id_obj in info.get("industryIdentifiers", []): | |
| if id_obj.get("type") == "ISBN_13": | |
| isbn13 = id_obj.get("identifier", "") | |
| elif id_obj.get("type") == "ISBN_10" and not isbn10: | |
| isbn10 = id_obj.get("identifier", "") | |
| isbn10 = _normalize_isbn(isbn10) if isbn10 else "" | |
| isbn13 = _normalize_isbn(isbn13) if isbn13 else "" | |
| if isbn10 and not isbn13: | |
| isbn13 = _isbn10_to_isbn13(isbn10) | |
| elif isbn13 and not isbn10: | |
| isbn10 = _isbn13_to_isbn10(isbn13) | |
| isbn = isbn13 or isbn10 | |
| imgs = info.get("imageLinks", {}) | |
| cover = (imgs.get("extraLarge") or imgs.get("large") or | |
| imgs.get("medium") or imgs.get("thumbnail") or "") | |
| cover = cover.replace("http://", "https://") | |
| # Upgrade zoom for larger image | |
| cover = re.sub(r"[?&]zoom=\d+", lambda m: m.group(0).replace(m.group(0)[-1], "3"), cover) | |
| rating = str(info.get("averageRating", "")) if info.get("averageRating") else "" | |
| rating_count = str(info.get("ratingsCount", "")) if info.get("ratingsCount") else "" | |
| return { | |
| "title": info.get("title", ""), | |
| "subtitle": info.get("subtitle", ""), | |
| "author": ", ".join(info.get("authors") or []), | |
| "description": _sanitize_book_description(info.get("description", "")), | |
| "genre": _normalize_genre(", ".join(info.get("categories") or [])), | |
| "isbn": isbn, | |
| "isbn10": isbn10, | |
| "isbn13": isbn13, | |
| "publisher": info.get("publisher", ""), | |
| "publish_year": str(info.get("publishedDate", ""))[:4], | |
| "language": info.get("language", ""), | |
| "page_count": int(info.get("pageCount") or 0), | |
| "rating": rating, | |
| "rating_count": rating_count, | |
| "cover_url": cover, | |
| "volume_id": volume.get("id", ""), | |
| } | |
| async def _apple_itunes_lookup(book_id: str, client: httpx.AsyncClient) -> dict: | |
| """Fetch Apple Books metadata via the public iTunes Lookup API (never blocked).""" | |
| try: | |
| r = await client.get( | |
| "https://itunes.apple.com/lookup", | |
| params={"id": book_id, "entity": "ebook"}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code != 200: | |
| return {} | |
| results = r.json().get("results", []) | |
| if not results: | |
| r = await client.get( | |
| "https://itunes.apple.com/lookup", | |
| params={"id": book_id}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| results = r.json().get("results", []) if r.status_code == 200 else [] | |
| if not results: | |
| return {} | |
| item = results[0] | |
| cover = ( | |
| item.get("artworkUrl100") | |
| or item.get("artworkUrl60") | |
| or item.get("artworkUrl512") | |
| or "" | |
| ).replace("http://", "https://") | |
| if cover: | |
| cover = re.sub(r"\d+x\d+bb", "600x600bb", cover) | |
| genres = item.get("genres") or item.get("genre") or [] | |
| if isinstance(genres, str): | |
| genres = [genres] | |
| return { | |
| "title": item.get("trackName") or item.get("collectionName") or "", | |
| "author": item.get("artistName") or "", | |
| "description": _sanitize_book_description(item.get("description", "")), | |
| "genre": _normalize_genre(", ".join(g for g in genres if g and g != "Books")), | |
| "publisher": item.get("sellerName", ""), | |
| "publish_year": str(item.get("releaseDate", ""))[:4], | |
| "cover_url": cover, | |
| "isbn": _normalize_isbn(str(item.get("isbn") or "")), | |
| } | |
| except Exception as e: | |
| logger.debug("Apple iTunes lookup failed", book_id=book_id, error=str(e)) | |
| return {} | |
| def _score_itunes_item(item: dict, title: str, author_hint: str) -> float: | |
| """Rank iTunes search hits — penalize summaries and study guides.""" | |
| name = (item.get("trackName") or "").lower() | |
| artist = (item.get("artistName") or "").lower() | |
| t = title.lower() | |
| score = 0.0 | |
| if name == t: | |
| score += 15.0 | |
| elif t in name or name in t: | |
| score += 8.0 | |
| else: | |
| score -= 6.0 | |
| if author_hint and _authors_match(author_hint, artist): | |
| score += 12.0 | |
| spam = ("summary", "study guide", "reading guide", "after reading", "glossary", "thesis") | |
| if any(s in name for s in spam): | |
| score -= 25.0 | |
| if item.get("description") and len(str(item["description"])) > 80: | |
| score += 3.0 | |
| return score | |
| async def _apple_itunes_search(title: str, author_hint: str, client: httpx.AsyncClient) -> dict: | |
| """Search Apple Books/iTunes catalog by title+author (free API, never blocked).""" | |
| if not title: | |
| return {} | |
| term = f"{title} {author_hint}".strip() if author_hint else title | |
| try: | |
| r = await client.get( | |
| "https://itunes.apple.com/search", | |
| params={"term": term, "entity": "ebook", "limit": "10", "country": "US"}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code != 200: | |
| return {} | |
| results = r.json().get("results", []) | |
| if not results: | |
| return {} | |
| best = max(results, key=lambda item: _score_itunes_item(item, title, author_hint)) | |
| if _score_itunes_item(best, title, author_hint) < 4.0: | |
| return {} | |
| genres = [g for g in (best.get("genres") or []) if g and g != "Books"] | |
| cover = (best.get("artworkUrl100") or best.get("artworkUrl60") or "") | |
| if cover: | |
| cover = re.sub(r"\d+x\d+bb", "600x600bb", cover).replace("http://", "https://") | |
| desc = _sanitize_book_description(_strip_html(best.get("description", ""))) | |
| return { | |
| "title": (best.get("trackName") or "").strip(), | |
| "author": (best.get("artistName") or "").strip(), | |
| "description": desc, | |
| "genre": _normalize_genre(", ".join(genres)), | |
| "cover_url": cover, | |
| "publish_year": str(best.get("releaseDate", ""))[:4], | |
| } | |
| except Exception as e: | |
| logger.debug("Apple iTunes search failed", term=term[:50], error=str(e)) | |
| return {} | |
| def _apply_bn_slug_fallback(result: dict, url: str, ids: dict) -> None: | |
| """Seed B&N title/author/cover from URL slug + ISBN before catalog gap-fill.""" | |
| if "barnesandnoble.com" not in url.lower(): | |
| return | |
| bn_title, bn_author = _title_author_from_url_slug(url) | |
| if not result.get("title") and bn_title and not _title_looks_bad(bn_title): | |
| result["title"] = bn_title | |
| if not _is_valid_author(result.get("author", "")) and bn_author: | |
| result["author"] = bn_author | |
| isbn_bn = _normalize_isbn(result.get("isbn", "") or ids.get("isbn", "")) | |
| if isbn_bn: | |
| result["isbn"] = isbn_bn | |
| if not result.get("cover_url"): | |
| result["cover_url"] = _bn_shopify_cover_url(isbn_bn) | |
| async def _google_by_volume_id(vid: str, client: httpx.AsyncClient) -> dict: | |
| try: | |
| r = await client.get( | |
| f"https://www.googleapis.com/books/v1/volumes/{vid}", | |
| params=_google_books_api_params(), | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| return _parse_google_volume(r.json()) | |
| except Exception as e: | |
| logger.debug("GB volume lookup failed", vid=vid, error=str(e)) | |
| return {} | |
| async def _google_by_isbn(isbn: str, client: httpx.AsyncClient, author_hint: str = "") -> dict: | |
| try: | |
| r = await client.get( | |
| "https://www.googleapis.com/books/v1/volumes", | |
| params=_google_books_api_params(q=f"isbn:{isbn}", maxResults="5"), | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| items = r.json().get("items", []) | |
| if not items: | |
| return {} | |
| if len(items) == 1: | |
| return _parse_google_volume(items[0]) | |
| title = items[0].get("volumeInfo", {}).get("title", "") | |
| best = max(items, key=lambda v: _score_gbvolume(v, title, author_hint)) | |
| return _parse_google_volume(best) | |
| except Exception as e: | |
| logger.debug("GB ISBN lookup failed", isbn=isbn, error=str(e)) | |
| return {} | |
| async def _google_search_best(title: str, author_hint: str, client: httpx.AsyncClient) -> dict: | |
| """Search Google Books and pick the best-ranked volume.""" | |
| q = f'intitle:"{title}"' | |
| if author_hint: | |
| q += f' inauthor:"{author_hint}"' | |
| try: | |
| r = await client.get( | |
| "https://www.googleapis.com/books/v1/volumes", | |
| params=_google_books_api_params(q=q, maxResults="8", langRestrict="en"), | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| items = r.json().get("items", []) | |
| if items: | |
| t_lower = title.lower() | |
| exact = [ | |
| v for v in items | |
| if (v.get("volumeInfo", {}).get("title") or "").lower() == t_lower | |
| ] | |
| pool = exact if exact else items | |
| best = max(pool, key=lambda v: _score_gbvolume(v, title, author_hint)) | |
| return _parse_google_volume(best) | |
| except Exception as e: | |
| logger.debug("GB search failed", query=q[:60], error=str(e)) | |
| return {} | |
| async def _google_search(query: str, client: httpx.AsyncClient) -> dict: | |
| """Search Google Books by any query (title, author, ISBN).""" | |
| try: | |
| r = await client.get( | |
| "https://www.googleapis.com/books/v1/volumes", | |
| params=_google_books_api_params(q=query, maxResults="1", langRestrict="en"), | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| items = r.json().get("items", []) | |
| if items: | |
| return _parse_google_volume(items[0]) | |
| except Exception as e: | |
| logger.debug("GB search failed", query=query, error=str(e)) | |
| return {} | |
| async def _ol_by_isbn(isbn: str, client: httpx.AsyncClient) -> dict: | |
| """Open Library Books API — rich structured data + author key.""" | |
| try: | |
| r = await client.get( | |
| "https://openlibrary.org/api/books", | |
| params={"bibkeys": f"ISBN:{isbn}", "format": "json", "jscmd": "data"}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| data = r.json().get(f"ISBN:{isbn}", {}) | |
| if data: | |
| covers = data.get("cover", {}) | |
| cover = covers.get("large") or covers.get("medium") or covers.get("small") or "" | |
| authors = data.get("authors", []) | |
| author_name = ", ".join(a.get("name", "") for a in authors if a.get("name")) | |
| author_keys = [a.get("key", "") for a in authors if a.get("key")] | |
| subjects = data.get("subjects", []) | |
| genre = ", ".join( | |
| (s.get("name", "") if isinstance(s, dict) else str(s)) | |
| for s in subjects[:5] | |
| ) | |
| return { | |
| "title": data.get("title", ""), | |
| "author": author_name, | |
| "author_keys": author_keys, | |
| "genre": genre, | |
| "publisher": ", ".join(p.get("name","") for p in data.get("publishers",[]) if p.get("name")), | |
| "publish_year":str(data.get("publish_date",""))[-4:], | |
| "page_count": int(data.get("number_of_pages") or 0), | |
| "cover_url": cover, | |
| } | |
| except Exception as e: | |
| logger.debug("OL ISBN failed", isbn=isbn, error=str(e)) | |
| return {} | |
| async def _ol_by_work_key(key: str, client: httpx.AsyncClient) -> dict: | |
| """Open Library /works/{key}.json — used for Open Library URLs.""" | |
| try: | |
| r = await client.get( | |
| f"https://openlibrary.org/works/{key}.json", | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| data = r.json() | |
| desc = data.get("description", "") | |
| if isinstance(desc, dict): | |
| desc = desc.get("value", "") | |
| subjects = data.get("subjects", [])[:6] | |
| genre = ", ".join(s if isinstance(s, str) else s.get("name","") for s in subjects) | |
| cover_id = str(data["covers"][0]) if data.get("covers") else "" | |
| cover = f"https://covers.openlibrary.org/b/id/{cover_id}-L.jpg" if cover_id else "" | |
| author_keys = [ | |
| (a.get("author") or a).get("key","") for a in data.get("authors",[]) | |
| ] | |
| return { | |
| "title": data.get("title",""), | |
| "author_keys": author_keys, | |
| "description": _strip_html(str(desc))[:_MAX_DESC], | |
| "genre": genre, | |
| "cover_url": cover, | |
| "isbn": _find_isbn(str(data)), | |
| } | |
| except Exception as e: | |
| logger.debug("OL work key failed", key=key, error=str(e)) | |
| return {} | |
| async def _ol_by_edition_key(key: str, client: httpx.AsyncClient) -> dict: | |
| """Open Library /books/{key}.json — for edition URLs.""" | |
| try: | |
| r = await client.get( | |
| f"https://openlibrary.org/books/{key}.json", | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| data = r.json() | |
| isbn = (data.get("isbn_13", [""])[0] | |
| or data.get("isbn_10", [""])[0] | |
| or _find_isbn(str(data))) | |
| cover_id = str(data["covers"][0]) if data.get("covers") else "" | |
| cover = f"https://covers.openlibrary.org/b/id/{cover_id}-L.jpg" if cover_id else "" | |
| author_keys = [a.get("key","") for a in data.get("authors",[]) if a.get("key")] | |
| return { | |
| "title": data.get("title",""), | |
| "author_keys": author_keys, | |
| "isbn": isbn, | |
| "publisher": ", ".join(data.get("publishers",[])), | |
| "publish_year":str(data.get("publish_date",""))[-4:], | |
| "page_count": int(data.get("number_of_pages") or 0), | |
| "cover_url": cover, | |
| } | |
| except Exception as e: | |
| logger.debug("OL edition key failed", key=key, error=str(e)) | |
| return {} | |
| async def _ol_description_by_isbn(isbn: str, client: httpx.AsyncClient) -> str: | |
| """Fetch book description from Open Library edition record.""" | |
| try: | |
| r = await client.get( | |
| f"https://openlibrary.org/isbn/{isbn}.json", | |
| timeout=_API_TIMEOUT, | |
| follow_redirects=True, | |
| ) | |
| if r.status_code != 200: | |
| return "" | |
| data = r.json() | |
| desc = data.get("description", "") | |
| if isinstance(desc, dict): | |
| desc = desc.get("value", "") | |
| if desc and len(str(desc)) > 40: | |
| return _strip_html(str(desc))[:_MAX_DESC] | |
| works = data.get("works") or [] | |
| if works: | |
| work_key = works[0].get("key", "").split("/")[-1] | |
| if work_key: | |
| wr = await client.get( | |
| f"https://openlibrary.org/works/{work_key}.json", | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if wr.status_code == 200: | |
| wdesc = wr.json().get("description", "") | |
| if isinstance(wdesc, dict): | |
| wdesc = wdesc.get("value", "") | |
| if wdesc: | |
| return _strip_html(str(wdesc))[:_MAX_DESC] | |
| except Exception as e: | |
| logger.debug("OL description lookup failed", isbn=isbn, error=str(e)) | |
| return "" | |
| async def _fetch_genre_fallback(title: str, author: str, client: httpx.AsyncClient) -> str: | |
| """Dedicated genre resolver — runs when all other strategies missed genre. | |
| Queries Open Library Search API and Google Books in parallel using | |
| title + author. Open Library returns rich subject lists; Google Books | |
| returns category breadcrumbs. Works for ANY book on ANY platform. | |
| """ | |
| if not title: | |
| return "" | |
| genre_parts: list[str] = [] | |
| async def _ol_search() -> list[str]: | |
| """Open Library ranked search — subject_facet gives clean genres.""" | |
| try: | |
| doc = await _ol_search_book(title, author, client) | |
| if not doc: | |
| return [] | |
| genre = doc.get("genre", "") | |
| if genre: | |
| return [g.strip() for g in genre.split(",") if g.strip()][:6] | |
| except Exception: | |
| pass | |
| return [] | |
| async def _gb_search() -> list[str]: | |
| """Google Books ranked search — categories field.""" | |
| try: | |
| gb = await _google_search_best(title, author, client) | |
| if gb.get("genre"): | |
| parts = [] | |
| for c in gb["genre"].split(","): | |
| parts.extend(p.strip() for p in c.split("/") if p.strip()) | |
| return list(dict.fromkeys(parts))[:6] | |
| except Exception: | |
| pass | |
| return [] | |
| async def _itunes_search() -> list[str]: | |
| try: | |
| apple = await _apple_itunes_search(title, author, client) | |
| if apple.get("genre"): | |
| return [g.strip() for g in apple["genre"].split(",") if g.strip()][:6] | |
| except Exception: | |
| pass | |
| return [] | |
| ol_genres, gb_genres, itunes_genres = await asyncio.gather( | |
| _ol_search(), _gb_search(), _itunes_search(), | |
| ) | |
| # Merge: prefer OL subjects, then iTunes, then Google Books categories | |
| seen: set[str] = set() | |
| for g in ol_genres + itunes_genres + gb_genres: | |
| gl = g.lower() | |
| if gl not in seen: | |
| seen.add(gl) | |
| genre_parts.append(g) | |
| if len(genre_parts) >= 5: | |
| break | |
| return _normalize_genre(", ".join(genre_parts)) | |
| async def _fill_catalog_gaps( | |
| result: dict, client: httpx.AsyncClient, author_hint: str = "", | |
| ) -> None: | |
| """Final pass — fill author, description, genre, bio, and cover from catalog APIs.""" | |
| isbn = _normalize_isbn(result.get("isbn", "")) | |
| if isbn: | |
| result["isbn"] = isbn | |
| elif result.get("isbn"): | |
| result.pop("isbn", None) | |
| title = result.get("title", "") | |
| hint = (author_hint or result.get("author") or "").strip() | |
| if _title_looks_bad(title) and isbn: | |
| gb_fix = await _google_by_isbn(isbn, client, author_hint=hint) | |
| if gb_fix.get("title") and not _title_looks_bad(gb_fix["title"]): | |
| result["title"] = _clean_title(gb_fix["title"]) | |
| if not _is_valid_author(result.get("author", "")) and (isbn or title): | |
| if isbn: | |
| gb, ol = await asyncio.gather( | |
| _google_by_isbn(isbn, client, author_hint=hint), | |
| _ol_by_isbn(isbn, client), | |
| ) | |
| else: | |
| gb, ol = await asyncio.gather( | |
| _google_search_best(title, hint, client), | |
| _ol_search_book(title, hint, client), | |
| ) | |
| for src in (gb, ol): | |
| if src.get("author"): | |
| resolved = _resolve_author(src["author"], hint) | |
| if _is_valid_author(resolved): | |
| result["author"] = resolved | |
| if ol.get("author_keys"): | |
| result.setdefault("author_keys", ol["author_keys"]) | |
| if not result.get("description") or _description_looks_bad(result.get("description", "")): | |
| if isbn: | |
| desc = await _ol_description_by_isbn(isbn, client) | |
| if desc and not _description_looks_bad(desc): | |
| result["description"] = desc | |
| if (not result.get("description") or _description_looks_bad(result.get("description", ""))) and title: | |
| gb = await _google_search_best(title, result.get("author", hint), client) | |
| if gb.get("description") and not _description_looks_bad(gb["description"]): | |
| result["description"] = gb["description"] | |
| if (not result.get("description") or _description_looks_bad(result.get("description", ""))) and title: | |
| apple = await _apple_itunes_search(title, result.get("author", hint), client) | |
| if apple.get("description"): | |
| result["description"] = apple["description"] | |
| for key in ("genre", "publish_year"): | |
| if apple.get(key) and not result.get(key): | |
| result[key] = apple[key] | |
| if apple.get("cover_url") and _is_placeholder_cover(result.get("cover_url", "")): | |
| result["cover_url"] = apple["cover_url"] | |
| if not result.get("genre") or _genre_looks_noisy(result.get("genre", "")): | |
| if isbn: | |
| gb_g = await _google_by_isbn(isbn, client, author_hint=hint) | |
| if gb_g.get("genre"): | |
| result["genre"] = gb_g["genre"] | |
| if not result.get("genre") or _genre_looks_noisy(result.get("genre", "")): | |
| genre = await _fetch_genre_fallback(title, result.get("author", ""), client) | |
| if genre: | |
| result["genre"] = genre | |
| if (not result.get("genre") or _genre_looks_noisy(result.get("genre", ""))) and title: | |
| apple = await _apple_itunes_search(title, result.get("author", hint), client) | |
| if apple.get("genre"): | |
| result["genre"] = apple["genre"] | |
| if not result.get("about_author"): | |
| for ak in (result.get("author_keys") or [])[:2]: | |
| _, bio = await _ol_author_bio(ak, client) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if bio: | |
| result["about_author"] = bio | |
| break | |
| if not result.get("about_author") and result.get("author"): | |
| name = _first_author_name(result["author"]) | |
| wiki = await _wikipedia_author_bio(name, client) | |
| wiki = _sanitize_about_author(wiki, result.get("description", "")) | |
| if wiki: | |
| result["about_author"] = wiki | |
| if not result.get("about_author") and result.get("author"): | |
| name = _first_author_name(result["author"]) | |
| _, bio = await _ol_author_search_by_name(name, client) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if bio: | |
| result["about_author"] = bio | |
| if _is_placeholder_cover(result.get("cover_url", "")): | |
| if isbn: | |
| ol = await _ol_by_isbn(isbn, client) | |
| if ol.get("cover_url") and not _is_placeholder_cover(ol["cover_url"]): | |
| result["cover_url"] = ol["cover_url"] | |
| if result.get("genre"): | |
| result["genre"] = _normalize_genre(result["genre"]) | |
| if result.get("description"): | |
| result["description"] = _sanitize_book_description(result["description"]) | |
| if result.get("about_author"): | |
| result["about_author"] = _sanitize_about_author( | |
| result["about_author"], result.get("description", ""), | |
| ) | |
| # Fields we try to fill from catalog APIs after HTML scraping. | |
| _CATALOG_FIELDS = ( | |
| "title", "author", "description", "genre", "publisher", "publish_year", | |
| "page_count", "language", "cover_url", "rating", "rating_count", | |
| "isbn", "isbn10", "isbn13", "edition", "book_format", "series", | |
| ) | |
| async def _ol_search_book(title: str, author: str, client: httpx.AsyncClient) -> dict: | |
| """Open Library search — returns flat metadata when ISBN lookup is unavailable.""" | |
| if not title: | |
| return {} | |
| try: | |
| author_surname = author.split()[-1] if author else "" | |
| params: dict = { | |
| "title": title, | |
| "limit": "10", | |
| "fields": ( | |
| "key,title,author_name,author_key,isbn,first_publish_year," | |
| "number_of_pages_median,publisher,subject,subject_facet,cover_i,language," | |
| "first_sentence,ratings_count" | |
| ), | |
| } | |
| if author_surname: | |
| params["author"] = author_surname | |
| r = await client.get( | |
| "https://openlibrary.org/search.json", | |
| params=params, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code != 200: | |
| return {} | |
| docs = r.json().get("docs", []) | |
| if not docs: | |
| # Fallback: free-text query | |
| r = await client.get( | |
| "https://openlibrary.org/search.json", | |
| params={ | |
| "q": f"{title} {author}".strip(), | |
| "limit": "3", | |
| "fields": params["fields"], | |
| }, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| docs = r.json().get("docs", []) if r.status_code == 200 else [] | |
| if not docs: | |
| return {} | |
| t_lower = title.lower() | |
| exact = [d for d in docs if (d.get("title") or "").lower() == t_lower] | |
| pool = exact if exact else docs | |
| doc = max(pool, key=lambda d: _score_ol_doc(d, title, author)) | |
| isbn_list = doc.get("isbn") or [] | |
| isbn = next( | |
| (i for i in isbn_list if len(i) == 13 and i.startswith("978")), | |
| next((i for i in isbn_list if len(i) == 13), isbn_list[0] if isbn_list else ""), | |
| ) | |
| cover_id = doc.get("cover_i") | |
| cover = f"https://covers.openlibrary.org/b/id/{cover_id}-L.jpg" if cover_id else "" | |
| subjects = doc.get("subject_facet") or doc.get("subject") or [] | |
| genre = _normalize_genre( | |
| ", ".join(s for s in subjects[:8] if _genre_label_ok(str(s))) | |
| ) | |
| publishers = doc.get("publisher") or [] | |
| author_keys = doc.get("author_key") or [] | |
| description = "" | |
| work_key = doc.get("key", "") | |
| first_sentence = doc.get("first_sentence") | |
| if isinstance(first_sentence, list) and first_sentence: | |
| description = _strip_html(str(first_sentence[0]))[:_MAX_DESC] | |
| elif isinstance(first_sentence, str) and first_sentence: | |
| description = _strip_html(first_sentence)[:_MAX_DESC] | |
| if work_key and not description: | |
| try: | |
| wr = await client.get( | |
| f"https://openlibrary.org{work_key}.json", | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if wr.status_code == 200: | |
| wdesc = wr.json().get("description", "") | |
| if isinstance(wdesc, dict): | |
| wdesc = wdesc.get("value", "") | |
| if wdesc: | |
| description = _strip_html(str(wdesc))[:_MAX_DESC] | |
| except Exception: | |
| pass | |
| return { | |
| "title": doc.get("title", ""), | |
| "author": ", ".join(doc.get("author_name") or []), | |
| "author_keys": [f"/authors/{k}" for k in author_keys], | |
| "isbn": isbn, | |
| "genre": genre, | |
| "publisher": ", ".join(publishers[:2]), | |
| "publish_year": str(doc.get("first_publish_year", ""))[:4], | |
| "page_count": int(doc.get("number_of_pages_median") or 0), | |
| "cover_url": cover, | |
| "description": description, | |
| } | |
| except Exception as e: | |
| logger.debug("OL search failed", title=title[:40], error=str(e)) | |
| return {} | |
| async def _wikipedia_author_bio(name: str, client: httpx.AsyncClient) -> str: | |
| """Fetch an author biography from Wikipedia (reliable, never blocked).""" | |
| if not name or len(name) < 3: | |
| return "" | |
| ua = ("AuthorBotRAG/1.0 (https://huggingface.co/spaces/testingweb647/Arag; " | |
| "metadata-import)") | |
| try: | |
| r = await client.get( | |
| "https://en.wikipedia.org/w/api.php", | |
| params={ | |
| "action": "query", "prop": "extracts", "exsentences": "4", | |
| "explaintext": "1", "redirects": "1", "format": "json", | |
| "titles": name, | |
| }, | |
| headers={"User-Agent": ua}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code != 200: | |
| return "" | |
| pages = r.json().get("query", {}).get("pages", {}) | |
| for page in pages.values(): | |
| extract = (page.get("extract") or "").strip() | |
| low = extract.lower() | |
| # Reject disambiguation pages and non-author entries | |
| if "may refer to" in low or "disambiguation" in low: | |
| continue | |
| if extract and any( | |
| w in low for w in ("author", "writer", "novelist", "wrote", "books", "poet") | |
| ): | |
| return _trim_prose(_strip_html(extract), _MAX_BIO) | |
| except Exception as e: | |
| logger.debug("Wikipedia bio lookup failed", name=name[:40], error=str(e)) | |
| return "" | |
| async def _ol_author_search_by_name(name: str, client: httpx.AsyncClient) -> tuple[str, str]: | |
| """Resolve author bio via Open Library author search when no author key exists.""" | |
| if not name or len(name) < 3: | |
| return "", "" | |
| try: | |
| r = await client.get( | |
| "https://openlibrary.org/search/authors.json", | |
| params={"q": name, "limit": "1"}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code != 200: | |
| return "", "" | |
| docs = r.json().get("docs", []) | |
| if not docs: | |
| return "", "" | |
| key = docs[0].get("key", "") | |
| if key: | |
| return await _ol_author_bio(key, client) | |
| except Exception: | |
| pass | |
| return "", "" | |
| async def _enrich_from_catalog( | |
| result: dict, client: httpx.AsyncClient, author_hint: str = "", | |
| ) -> None: | |
| """Backfill missing fields from Google Books + Open Library (never blocked). | |
| Runs after HTML scraping so JS-heavy stores (Apple Books, Kobo) still get | |
| full title, description, author bio, genre, and cover via ISBN or search. | |
| """ | |
| isbn = result.get("isbn", "") | |
| title = result.get("title", "") | |
| author = result.get("author", "") | |
| hint = (author_hint or author or "").strip() | |
| locked_isbn = isbn | |
| locked_isbn10 = str(result.get("isbn10", "") or "") | |
| locked_isbn13 = str(result.get("isbn13", "") or "") | |
| if not isbn and not title: | |
| return | |
| missing = [f for f in _CATALOG_FIELDS if not result.get(f)] | |
| if _description_looks_bad(result.get("description", "")): | |
| if "description" not in missing: | |
| missing.append("description") | |
| if _genre_looks_noisy(result.get("genre", "")): | |
| result.pop("genre", None) | |
| if "genre" not in missing: | |
| missing.append("genre") | |
| if not missing and result.get("about_author"): | |
| return | |
| gb: dict = {} | |
| ol: dict = {} | |
| if isbn: | |
| gb, ol = await asyncio.gather( | |
| _google_by_isbn(isbn, client, author_hint=hint), | |
| _ol_by_isbn(isbn, client), | |
| ) | |
| elif title: | |
| gb, ol = await asyncio.gather( | |
| _google_search_best(title, hint, client), | |
| _ol_search_book(title, hint, client), | |
| ) | |
| if gb.get("isbn") and not isbn: | |
| isbn = gb["isbn"] | |
| ol_isbn = await _ol_by_isbn(isbn, client) | |
| for key in ("genre", "publisher", "publish_year", "page_count", "cover_url", "author_keys"): | |
| if ol_isbn.get(key) and not ol.get(key): | |
| ol[key] = ol_isbn[key] | |
| elif ol.get("isbn") and not isbn: | |
| isbn = ol["isbn"] | |
| ol_isbn = await _ol_by_isbn(isbn, client) | |
| for key in ("genre", "publisher", "publish_year", "page_count", "cover_url", "author_keys"): | |
| if ol_isbn.get(key) and not ol.get(key): | |
| ol[key] = ol_isbn[key] | |
| if not ol.get("description"): | |
| desc = await _ol_description_by_isbn(isbn, client) | |
| if desc: | |
| ol["description"] = desc | |
| for source in (gb, ol): | |
| for key in _CATALOG_FIELDS: | |
| if key == "isbn": | |
| continue | |
| if key == "title" and source.get(key) and _title_looks_bad(result.get("title", "")): | |
| result["title"] = _clean_title(source[key]) | |
| continue | |
| if key == "author" and source.get(key) and not _is_valid_author(result.get("author", "")): | |
| result["author"] = source[key] | |
| continue | |
| if key == "cover_url" and _is_retailer_cover(result.get("cover_url", "")): | |
| continue | |
| if source.get(key) and not result.get(key): | |
| result[key] = source[key] | |
| if gb.get("description") and _description_looks_bad(result.get("description", "")): | |
| result["description"] = gb["description"] | |
| elif not result.get("description") and gb.get("description"): | |
| result["description"] = gb["description"] | |
| if gb.get("genre") and (not result.get("genre") or _genre_looks_noisy(result.get("genre", ""))): | |
| result["genre"] = gb["genre"] | |
| if hint: | |
| resolved = _resolve_author(result.get("author", ""), hint) | |
| if resolved: | |
| result["author"] = resolved | |
| if locked_isbn10 or locked_isbn13 or locked_isbn: | |
| _apply_isbn_pair(result, locked_isbn10, locked_isbn13) | |
| if locked_isbn: | |
| result["isbn"] = locked_isbn | |
| elif gb.get("isbn"): | |
| _apply_isbn_pair(result, *_extract_isbn_pair(gb["isbn"])) | |
| elif ol.get("isbn"): | |
| _apply_isbn_pair(result, *_extract_isbn_pair(ol["isbn"])) | |
| # Author bio — prefer OL author keys, then name search | |
| if not result.get("about_author"): | |
| author_keys = result.get("author_keys") or ol.get("author_keys") or [] | |
| for ak in author_keys[:2]: | |
| _, bio = await _ol_author_bio(ak, client) | |
| if bio: | |
| result["about_author"] = bio | |
| break | |
| if not result.get("about_author") and (author or result.get("author")): | |
| name = _first_author_name(author or result["author"]) | |
| wiki = await _wikipedia_author_bio(name, client) | |
| wiki = _sanitize_about_author(wiki, result.get("description", "")) | |
| if wiki: | |
| result["about_author"] = wiki | |
| else: | |
| _, bio = await _ol_author_search_by_name(name, client) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if bio: | |
| result["about_author"] = bio | |
| # Google Books description often richer than OL when HTML missed it | |
| if not result.get("description"): | |
| if gb.get("description"): | |
| result["description"] = gb["description"] | |
| elif locked_isbn or result.get("isbn"): | |
| desc = await _ol_description_by_isbn(locked_isbn or result["isbn"], client) | |
| if desc: | |
| result["description"] = desc | |
| async def _ol_by_oclc(oclc: str, client: httpx.AsyncClient) -> dict: | |
| """Open Library lookup by WorldCat OCLC number.""" | |
| try: | |
| r = await client.get( | |
| "https://openlibrary.org/api/books", | |
| params={"bibkeys": f"OCLC:{oclc}", "format": "json", "jscmd": "data"}, | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code != 200: | |
| return {} | |
| data = r.json().get(f"OCLC:{oclc}", {}) | |
| if not data: | |
| return {} | |
| covers = data.get("cover", {}) | |
| cover = covers.get("large") or covers.get("medium") or covers.get("small") or "" | |
| authors = data.get("authors", []) | |
| author_name = ", ".join(a.get("name", "") for a in authors if a.get("name")) | |
| author_keys = [a.get("key", "") for a in authors if a.get("key")] | |
| subjects = data.get("subjects", []) | |
| genre = ", ".join( | |
| (s.get("name", "") if isinstance(s, dict) else str(s)) | |
| for s in subjects[:5] | |
| ) | |
| isbn_list = data.get("identifiers", {}).get("isbn_13", []) or data.get("isbn_13", []) | |
| if not isbn_list: | |
| isbn_list = data.get("identifiers", {}).get("isbn_10", []) or data.get("isbn_10", []) | |
| isbn = isbn_list[0] if isbn_list else _find_isbn(str(data)) | |
| desc = data.get("notes", "") or data.get("subtitle", "") | |
| if isinstance(desc, dict): | |
| desc = desc.get("value", "") | |
| return { | |
| "title": data.get("title", ""), | |
| "author": author_name, | |
| "author_keys": author_keys, | |
| "description": _strip_html(str(desc))[:_MAX_DESC] if desc else "", | |
| "genre": genre, | |
| "publisher": ", ".join(p.get("name", "") for p in data.get("publishers", []) if p.get("name")), | |
| "publish_year":str(data.get("publish_date", ""))[-4:], | |
| "page_count": int(data.get("number_of_pages") or 0), | |
| "cover_url": cover, | |
| "isbn": isbn, | |
| } | |
| except Exception as e: | |
| logger.debug("OL OCLC lookup failed", oclc=oclc, error=str(e)) | |
| return {} | |
| async def _ol_author_bio(key: str, client: httpx.AsyncClient) -> tuple[str, str]: | |
| """Returns (author_name, author_bio) from Open Library author endpoint.""" | |
| key_short = key.strip("/").split("/")[-1] | |
| try: | |
| r = await client.get( | |
| f"https://openlibrary.org/authors/{key_short}.json", | |
| timeout=_API_TIMEOUT, | |
| ) | |
| if r.status_code == 200: | |
| data = r.json() | |
| name = data.get("name", "") | |
| bio = data.get("bio", "") | |
| if isinstance(bio, dict): | |
| bio = bio.get("value", "") | |
| return name, _strip_html(str(bio))[:_MAX_BIO] | |
| except Exception: | |
| pass | |
| return "", "" | |
| async def _fetch_author_page_bio(author_url: str, client: httpx.AsyncClient, patterns: list[str]) -> str: | |
| """Fetch an author profile page and extract biography text.""" | |
| if not author_url: | |
| return "" | |
| html = await _fetch_html(author_url, client) | |
| if not html: | |
| return "" | |
| for pat in patterns: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| bio_raw = m.group(1).replace("\\n", "\n").replace('\\"', '"') | |
| bio = _strip_scripts(bio_raw).strip() | |
| if len(bio) > 50: | |
| return bio[:_MAX_BIO] | |
| return "" | |
| # ── Step 6: HTML fetch with retry + UA rotation ──────────────────────────────── | |
| async def _fetch_html(url: str, client: httpx.AsyncClient, referer: str = "") -> str: | |
| """Fetch a URL with UA rotation and up to 3 retries on blocks. | |
| Returns empty string on complete failure. | |
| """ | |
| RETRYABLE = {403, 429, 503, 520, 521, 522, 523, 524} | |
| last_error = "" | |
| for attempt in range(_MAX_RETRIES): | |
| try: | |
| if attempt > 0: | |
| await asyncio.sleep(1.5 * attempt) # back-off: 0s, 1.5s, 3s | |
| headers = _browser_headers(referer) | |
| r = await client.get(url, headers=headers, timeout=_HTML_TIMEOUT) | |
| if r.status_code == 200: | |
| return r.text | |
| if r.status_code in RETRYABLE: | |
| logger.debug("HTML fetch blocked, retrying", url=url[:60], | |
| status=r.status_code, attempt=attempt + 1) | |
| last_error = f"HTTP {r.status_code}" | |
| continue # retry with new UA | |
| # Non-retryable error (404, 410, etc.) | |
| last_error = f"HTTP {r.status_code}" | |
| break | |
| except (httpx.TimeoutException, httpx.ConnectError) as e: | |
| last_error = str(e) | |
| logger.debug("HTML fetch network error", url=url[:60], error=last_error) | |
| if attempt < _MAX_RETRIES - 1: | |
| await asyncio.sleep(1.5 * (attempt + 1)) | |
| logger.warning("HTML fetch failed after retries", url=url[:60], error=last_error) | |
| return "" | |
| def _parse_html_og(html: str) -> dict: | |
| """Extract OG/Twitter meta tags + JSON-LD from HTML.""" | |
| result: dict = { | |
| "title": _extract_meta(html, "title"), | |
| "description": _sanitize_book_description(_extract_meta(html, "description")), | |
| "cover_url": _extract_meta(html, "image"), | |
| "isbn": _find_isbn(html), | |
| } | |
| # JSON-LD enrichment | |
| for block in _extract_json_ld(html): | |
| if not isinstance(block, dict): | |
| continue | |
| btype = block.get("@type", "") | |
| if btype not in ("Book", "Product", "CreativeWork"): | |
| continue | |
| author_obj = block.get("author", {}) | |
| author = "" | |
| if isinstance(author_obj, list): | |
| author = ", ".join( | |
| a.get("name","") if isinstance(a,dict) else str(a) for a in author_obj | |
| ) | |
| elif isinstance(author_obj, dict): | |
| author = author_obj.get("name","") | |
| elif isinstance(author_obj, str): | |
| author = author_obj | |
| rating_obj = block.get("aggregateRating", {}) | |
| publisher_obj = block.get("publisher", {}) | |
| result.update({k: v for k, v in { | |
| "title": block.get("name","").strip() or result["title"], | |
| "author": author, | |
| "description": _strip_html(block.get("description",""))[:_MAX_DESC] or result["description"], | |
| "isbn": block.get("isbn","") or result["isbn"], | |
| "publisher": publisher_obj.get("name","") if isinstance(publisher_obj,dict) else str(publisher_obj or ""), | |
| "publish_year": str(block.get("datePublished",""))[:4], | |
| "page_count": int(block.get("numberOfPages") or 0), | |
| "language": block.get("inLanguage",""), | |
| "rating": str(rating_obj.get("ratingValue","")) if rating_obj else "", | |
| "rating_count": str(rating_obj.get("ratingCount","")) if rating_obj else "", | |
| "cover_url": (block.get("image","") or result["cover_url"]), | |
| "genre": (", ".join(block.get("genre",[])) | |
| if isinstance(block.get("genre"), list) | |
| else str(block.get("genre",""))), | |
| "book_format": str(block.get("bookFormat") or block.get("binding") or ""), | |
| "edition": str(block.get("bookEdition") or block.get("edition") or ""), | |
| }.items() if v}) | |
| if block.get("isbn"): | |
| i10, i13 = _extract_isbn_pair(str(block.get("isbn"))) | |
| _apply_isbn_pair(result, i10, i13) | |
| break # Use first Book-type block | |
| return result | |
| return result | |
| def _parse_amazon_meta_title(html: str) -> dict: | |
| """Parse Amazon's meta title — reliable when productTitle is absent (bot pages).""" | |
| result: dict = {} | |
| m = re.search( | |
| r'<meta\s[^>]*name=["\']title["\'][^>]*content=["\']([^"\']+)["\']', | |
| html, re.IGNORECASE, | |
| ) | |
| if not m: | |
| return result | |
| content = html_module.unescape(m.group(1)) | |
| # Amazon.com: Book Title eBook : Author, Name: Kindle Store | |
| em = re.search( | |
| r"Amazon\.com:\s*(.+?)\s+eBook\s*:\s*([^:]+?)\s*:\s*Kindle", | |
| content, re.IGNORECASE, | |
| ) | |
| if em: | |
| title = _clean_title(em.group(1).strip()) | |
| author_raw = em.group(2).strip() | |
| if "," in author_raw: | |
| last, first = [p.strip() for p in author_raw.split(",", 1)] | |
| author = f"{first} {last}".strip() if first and last else author_raw | |
| else: | |
| author = author_raw | |
| if title and not _title_looks_bad(title): | |
| result["title"] = title | |
| if _is_valid_author(author): | |
| result["author"] = author | |
| result.setdefault("book_format", "Kindle Edition") | |
| return result | |
| def _parse_amazon_breadcrumb_genre(html: str) -> str: | |
| """Genre from Amazon category breadcrumbs (Kindle Store › … › Kidnapping).""" | |
| block = re.search( | |
| r'id="wayfinding-breadcrumbs_feature_div"[^>]*>(.*?)</ul>', | |
| html, re.DOTALL | re.IGNORECASE, | |
| ) | |
| if not block: | |
| return "" | |
| links = re.findall(r"<a[^>]*>([^<]+)</a>", block.group(1)) | |
| skip = {"kindle store", "kindle ebooks", "books"} | |
| cats = [ | |
| _strip_html(link).strip() | |
| for link in links | |
| if _strip_html(link).strip().lower() not in skip | |
| ] | |
| if not cats: | |
| return "" | |
| if len(cats) >= 2: | |
| return _normalize_genre(f"{cats[-1]}, {cats[-2]}")[:300] | |
| return _normalize_genre(cats[-1])[:300] | |
| def _parse_amazon_rpi_carousel(html: str) -> dict: | |
| """Parse Amazon's rich-product-information (RPI) carousel — modern Kindle layout.""" | |
| result: dict = {} | |
| m = re.search( | |
| r'rpi-attribute-book_details-publication_date.*?rpi-attribute-value[^>]*>\s*<span>([^<]+)', | |
| html, re.DOTALL | re.IGNORECASE, | |
| ) | |
| if m: | |
| yr = re.search(r"(\d{4})", m.group(1)) | |
| if yr: | |
| result["publish_year"] = yr.group(1) | |
| m = re.search( | |
| r'rpi-attribute-book_details-ebook_pages.*?rpi-attribute-value.*?<span>(\d+)\s*pages', | |
| html, re.DOTALL | re.IGNORECASE, | |
| ) | |
| if m: | |
| result["page_count"] = int(m.group(1)) | |
| m = re.search( | |
| r'id="rpi-attribute-language".*?rpi-attribute-value[^>]*>\s*<span>([^<]+)', | |
| html, re.DOTALL | re.IGNORECASE, | |
| ) | |
| if m: | |
| result["language"] = m.group(1).strip()[:50] | |
| m = re.search( | |
| r'rpi-attribute-book_details-series".*?rpi-attribute-label[^>]*>\s*<span>([^<]+)</span>.*?' | |
| r'rpi-attribute-value.*?<span>([^<]+)</span>', | |
| html, re.DOTALL | re.IGNORECASE, | |
| ) | |
| if m: | |
| result["series"] = f"{m.group(1).strip()}: {m.group(2).strip()}"[:120] | |
| # Kindle pages reference print edition as "ISBN B0XXXXXXXX" (really an ASIN) | |
| m = re.search(r"print edition \(ISBN ([A-Z0-9]{10})\)", html, re.IGNORECASE) | |
| if m: | |
| result["print_edition_asin"] = m.group(1).upper() | |
| return result | |
| def _parse_amazon_html(html: str) -> dict: | |
| """Deep-parse Amazon product page for any fields the APIs didn't return. | |
| Amazon's OG <meta> tags set title/description to the store page title | |
| (e.g. 'Amazon.com: Book Title: Author: Kindle Store'), so we MUST | |
| pull from the actual product-page DOM elements instead. | |
| """ | |
| result: dict = {} | |
| # ── Meta title fallback (works even when DOM is stripped) ───────────────── | |
| for k, v in _parse_amazon_meta_title(html).items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| # ── RPI carousel (modern Kindle book details) ───────────────────────────── | |
| for k, v in _parse_amazon_rpi_carousel(html).items(): | |
| if k == "print_edition_asin": | |
| continue | |
| if v and not result.get(k): | |
| result[k] = v | |
| # ── Title: <span id="productTitle"> is the reliable source ─────────────── | |
| for pat in [ | |
| r'id="productTitle"[^>]*>\s*([^<]{3,400})', | |
| r'"name"\s*:\s*"([^"]{3,300})"', # JSON-LD name | |
| ]: | |
| m = re.search(pat, html) | |
| if m: | |
| t = m.group(1).strip() | |
| # Skip if it looks like the full page title with 'Amazon.com:' prefix | |
| if t and not t.lower().startswith("amazon") and not _title_looks_bad(t): | |
| result["title"] = t | |
| break | |
| # Split "Title (Series Name, Book N)" — common on Amazon Kindle listings | |
| if result.get("title"): | |
| paren = re.match(r"^(.+?)\s*\(([^)]+)\)\s*$", result["title"]) | |
| if paren: | |
| inner = paren.group(2).strip() | |
| if re.search(r"\b(book|series|vol\.?|volume|#)\b", inner, re.I) or len(inner) > 12: | |
| result["title"] = paren.group(1).strip() | |
| result.setdefault("series", inner[:120]) | |
| # ── Author: bylineInfo section or JSON-LD ──────────────────────────────── | |
| for pat in [ | |
| # New Amazon layout — contributorNameID span | |
| r'id="bylineInfo"[^>]*>.*?class="[^"]*contributorNameID[^"]*"[^>]*>([^<]+)', | |
| # Byline author link text | |
| r'id="bylineInfo"[^>]*>.*?<a[^>]*>\s*([A-Z][^<]{2,80}?)\s*</a>', | |
| # JSON-LD author.name | |
| r'"author"\s*:\s*\{[^}]*"name"\s*:\s*"([^"]{2,100})"', | |
| # JSON embedded in page scripts | |
| r'"author"\s*:\s*"([A-Z][^"]{2,80})"', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL) | |
| if m: | |
| author = _strip_html(m.group(1)).strip() | |
| # Exclude obvious non-author strings | |
| bad = {"visit amazon", "amazon", "follow", "see all", "kindle store", "books"} | |
| if author and author.lower() not in bad and len(author) > 2: | |
| result["author"] = author | |
| break | |
| # ── Author URL (for bio fallback) ──────────────────────────────────────── | |
| for pat in [ | |
| r'id="bylineInfo"[^>]*>.*?<a[^>]*href="([^"]*/(?:e|author)/[^"]+)"', | |
| r'class="[^"]*contributorNameID[^"]*"[^>]*href="([^"]*/(?:e|author)/[^"]+)"', | |
| r'class="[^"]*follow-the-author[^"]*"[^>]*href="([^"]*/(?:e|author)/[^"]+)"', | |
| # Fallback to the first author-like link on the page (almost always the primary author) | |
| r'href="([^"]*/(?:e|author)/[^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| url_part = m.group(1) | |
| # Make sure it's not a generic link (like a review or customer-preferences) | |
| if not "customer-preferences" in url_part and not "/review/" in url_part: | |
| if url_part.startswith("/"): | |
| url_part = "https://www.amazon.com" + url_part | |
| result["author_url"] = url_part | |
| break | |
| # ── Description: bookDescription_feature_div is the canonical source ───── | |
| # Strategy 1: grab the whole div then strip scripts — most reliable | |
| m = re.search( | |
| r'id="bookDescription_feature_div"[^>]*>(.*?)(?=<div\s[^>]*id=|<section\s|$)', | |
| html, re.DOTALL, | |
| ) | |
| if m: | |
| raw_desc = m.group(1) | |
| d = _strip_scripts(raw_desc) | |
| # Amazon adds a "Read more" link and trailing metadata — truncate there | |
| d = re.split(r'\s*Read\s+more\s*$', d, flags=re.IGNORECASE)[0].strip() | |
| d = re.sub(r'\s{3,}', '\n\n', d) # collapse excessive whitespace | |
| if len(d) > 40: | |
| result["description"] = _sanitize_book_description(d) | |
| # Strategy 2: JSON key bookDescription in page JS blobs | |
| if not result.get("description"): | |
| for pat in [ | |
| r'"bookDescription"\s*:\s*"([^"]{30,})"', | |
| r'"description"\s*:\s*"([^"]{50,})"', | |
| ]: | |
| m2 = re.search(pat, html, re.DOTALL) | |
| if m2: | |
| d2 = m2.group(1).replace('\\n', '\n').replace('\\"', '"') | |
| d2 = _strip_html(d2) | |
| if len(d2) > 40: | |
| result["description"] = d2[:_MAX_DESC] | |
| break | |
| # ── Author Bio ──────────────────────────────────────────────────────────── | |
| # Amazon renders the About the Author section dynamically via JS widgets. | |
| # The static HTML contains only JS init code — NOT the bio text. | |
| # However, the bio IS embedded in a JSON blob inside a <script> tag that | |
| # the page injects to pre-populate the widget. | |
| bio_found = False | |
| # Strategy 1: JSON keys inside embedded <script> blobs | |
| for bio_pat in [ | |
| r'"authorBiography"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| r'"authorBio"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| r'"biography"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| r'"bio"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| # Amazon's author-profile data blob | |
| r'"description"\s*:\s*"((?:[^"\\]|\\.){80,})"', | |
| ]: | |
| m = re.search(bio_pat, html, re.DOTALL) | |
| if m: | |
| bio_raw = m.group(1).replace('\\n', '\n').replace('\\"', '"').replace('\\u003c', '<').replace('\\u003e', '>').replace('\\u0026', '&') | |
| bio = _strip_html(bio_raw).strip() | |
| # Reject if it looks like it's the book description (overlap check) | |
| book_desc = result.get("description", "") | |
| if len(bio) > 50 and (not book_desc or bio[:60] not in book_desc): | |
| result["about_author"] = bio[:_MAX_BIO] | |
| bio_found = True | |
| break | |
| # Strategy 2: Static authorBio_feature_div (present on some non-Kindle pages) | |
| if not bio_found: | |
| for pat in [ | |
| r'id="authorBio_feature_div"[^>]*>(.*?)</div>\s*(?:</div>|<div)', | |
| r'id="authorBio_feature_div"[^>]*>(.*?)</div>', | |
| r'class="[^"]*author-bio[^"]*"[^>]*>(.*?)</div>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| bio = _strip_scripts(m.group(1)).strip() | |
| if len(bio) > 50 and not bio.lower().startswith(("if(", "p.when", "p.load")): | |
| result["about_author"] = bio[:_MAX_BIO] | |
| bio_found = True | |
| break | |
| # ── Page count ──────────────────────────────────────────────────────────── | |
| if not result.get("page_count"): | |
| for pat in [ | |
| r'>Print length</span>.*?(\d+)\s+pages', | |
| r'>Page numbers source[^<]*</span>.*?(\d+)\s+pages', | |
| r'"numberOfPages":\s*(\d+)', | |
| r'(\d+)\s+pages.*?Print length', | |
| r'rpi-attribute-book_details-ebook_pages.*?<span>(\d+)\s*pages', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| result["page_count"] = int(m.group(1)) | |
| break | |
| # ── Publisher + publish year ────────────────────────────────────────────── | |
| for pat in [ | |
| r'>Publisher</span>.*?<span[^>]*>([^<]{3,120})', | |
| r'>Imprint</span>.*?<span[^>]*>([^<]{3,120})', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| pub_raw = _strip_html(m.group(1)).strip() | |
| yr = re.search(r"\(.*?(\d{4}).*?\)", pub_raw) | |
| if yr: | |
| result.setdefault("publish_year", yr.group(1)) | |
| pub_raw = pub_raw[:pub_raw.index("(")].strip() | |
| elif re.search(r"\b(\d{4})\b", pub_raw): | |
| result.setdefault("publish_year", re.search(r"\b(\d{4})\b", pub_raw).group(1)) | |
| result["publisher"] = pub_raw[:200] | |
| break | |
| # Publication date standalone | |
| if not result.get("publish_year"): | |
| for pat in ( | |
| r'>Publication date</span>.*?<span[^>]*>([^<]+)', | |
| r'rpi-attribute-book_details-publication_date.*?rpi-attribute-value[^>]*>\s*<span>([^<]+)', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| yr = re.search(r"(\d{4})", m.group(1)) | |
| if yr: | |
| result["publish_year"] = yr.group(1) | |
| break | |
| # ── Genre — breadcrumbs first, then Best Sellers Rank ───────────────────── | |
| genre = _parse_amazon_breadcrumb_genre(html) | |
| if genre: | |
| result["genre"] = genre | |
| if not result.get("genre"): | |
| m = re.search( | |
| r'Best Sellers Rank.*?in\s+<a[^>]*>([^<]{3,80})</a>', | |
| html, re.DOTALL | re.IGNORECASE, | |
| ) | |
| if m: | |
| result["genre"] = m.group(1).strip() | |
| if not result.get("genre"): | |
| for block in _extract_json_ld(html): | |
| g = block.get("genre", "") | |
| if g: | |
| result["genre"] = (", ".join(g) if isinstance(g, list) else str(g))[:300] | |
| break | |
| # ── Cover — hi-res product image ───────────────────────────────────────── | |
| for pat in [ | |
| r'"hiRes"\s*:\s*"(https://[^"]+)"', | |
| r'"large"\s*:\s*"(https://[^"]+)"', | |
| r'id="imgBlkFront"[^>]*src="([^"]+)"', | |
| r'id="landingImage"[^>]*src="([^"]+)"', | |
| r'data-old-hires="(https://[^"]+)"', | |
| r'"mainImage"[^}]*"url"\s*:\s*"(https://[^"]+)"', | |
| ]: | |
| m = re.search(pat, html) | |
| if m and "amazon" in m.group(1): | |
| result["cover_url"] = m.group(1) | |
| break | |
| # ── ISBN-10 / ISBN-13 (detail bullets + embedded JSON) ─────────────────── | |
| i10, i13 = _extract_isbn_pair(html) | |
| if i10 or i13: | |
| _apply_isbn_pair(result, i10, i13) | |
| # Kindle print-edition reference: "ISBN B0XXXXXXXX" in RPI popover | |
| if not result.get("isbn") and not result.get("isbn13"): | |
| rpi = _parse_amazon_rpi_carousel(html) | |
| print_asin = rpi.get("print_edition_asin", "") | |
| if print_asin: | |
| result["isbn"] = print_asin | |
| result.setdefault("warnings_note", "print_edition_asin") | |
| # ── ASIN ───────────────────────────────────────────────────────────────── | |
| for pat in ( | |
| r'"asin"\s*:\s*"([A-Z0-9]{10})"', | |
| r'data-asin="([A-Z0-9]{10})"', | |
| r'"productAsin"\s*:\s*"([A-Z0-9]{10})"', | |
| ): | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["asin"] = m.group(1).upper() | |
| break | |
| # ── Format / binding ───────────────────────────────────────────────────── | |
| for pat in ( | |
| r'>Format</span>.*?<span[^>]*>([^<]{2,80})', | |
| r'>Print length</span>.*?<span[^>]*>([^<]{2,80})', | |
| r'"binding"\s*:\s*"([^"]{2,80})"', | |
| r'"bookFormat"\s*:\s*"([^"]{2,80})"', | |
| r'"format"\s*:\s*"([^"]{2,80})"', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| fmt = _strip_html(m.group(1)).strip() | |
| if fmt and "page" not in fmt.lower(): | |
| result["book_format"] = fmt[:80] | |
| break | |
| if not result.get("book_format"): | |
| low = html.lower() | |
| for label in ("kindle edition", "paperback", "hardcover", "audiobook", "mass market paperback"): | |
| if label in low: | |
| result["book_format"] = label.title() | |
| break | |
| # ── Edition ────────────────────────────────────────────────────────────── | |
| for pat in ( | |
| r'>Edition</span>.*?<span[^>]*>([^<]{2,120})', | |
| r'"edition"\s*:\s*"([^"]{2,120})"', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| edition = _strip_html(m.group(1)).strip() | |
| if edition: | |
| result["edition"] = edition[:120] | |
| break | |
| # ── Series ─────────────────────────────────────────────────────────────── | |
| for pat in ( | |
| r'id="seriesBulletWidget_feature_div"[^>]*>.*?<a[^>]*>([^<]{2,120})', | |
| r'Book \d+ of \d+:\s*</span>\s*<a[^>]*>([^<]+)', | |
| r'"seriesName"\s*:\s*"([^"]{2,120})"', | |
| r'"series"\s*:\s*\{[^}]*"name"\s*:\s*"([^"]{2,120})"', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| series = _strip_html(m.group(1)).strip() | |
| if series: | |
| result["series"] = series[:120] | |
| break | |
| # ── Language ───────────────────────────────────────────────────────────── | |
| if not result.get("language"): | |
| for pat in ( | |
| r'>Language</span>.*?<span[^>]*>([^<]{2,50})', | |
| r'"language"\s*:\s*"([^"]{2,10})"', | |
| r'"inLanguage"\s*:\s*"([^"]{2,10})"', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| lang = _strip_html(m.group(1)).strip() | |
| if lang: | |
| result["language"] = lang[:50] | |
| break | |
| # ── Price ──────────────────────────────────────────────────────────────── | |
| for pat in ( | |
| r'class="[^"]*a-price[^"]*"[^>]*>.*?<span[^>]*class="[^"]*a-offscreen[^"]*"[^>]*>\$([^<]+)', | |
| r'"price"\s*:\s*"?([0-9]+(?:\.[0-9]{2})?)"?', | |
| r'"buyingPrice"\s*:\s*"?([0-9]+(?:\.[0-9]{2})?)"?', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| price = m.group(1).strip().replace(",", "") | |
| if price: | |
| result["price"] = price if price.startswith("$") else f"${price}" | |
| break | |
| return result | |
| def _parse_goodreads_html(html: str) -> dict: | |
| """Deep-parse a Goodreads book page — genre tags, description, cover, bio.""" | |
| result: dict = {} | |
| # Title | |
| m = re.search(r'<h1[^>]*data-testid="bookTitle"[^>]*>([^<]+)', html, re.IGNORECASE) | |
| if not m: | |
| m = re.search(r'<h1[^>]*class="[^"]*Text Text__title1[^"]*"[^>]*>([^<]+)', html, re.IGNORECASE) | |
| if m: | |
| result["title"] = m.group(1).strip() | |
| # Author | |
| m = re.search(r'data-testid="name"[^>]*>([^<]+)', html, re.IGNORECASE) | |
| if not m: | |
| m = re.search(r'class="authorName"[^>]*>\s*<span[^>]*>([^<]+)', html, re.IGNORECASE) | |
| if m: | |
| result["author"] = m.group(1).strip() | |
| # Genre — Goodreads shows genre tags as clickable links | |
| genres = re.findall(r'data-testid="genre"[^>]*>([^<]+)</span>', html, re.IGNORECASE) | |
| if not genres: | |
| genres = re.findall(r'itemprop="genre"[^>]*>([^<]+)', html, re.IGNORECASE) | |
| if not genres: | |
| genres = re.findall(r'class="[^"]*BookPageMetadataSection__genre[^"]*"[^>]*>([^<]+)', html, re.IGNORECASE) | |
| if not genres: | |
| genres = re.findall(r'/genres/([^"?\s;]+)', html) | |
| genres = [g.replace("-", " ").title() for g in genres if len(g) > 2][:5] | |
| if not genres: | |
| # Old Goodreads layout: genre shelf links | |
| genres = re.findall(r'goodreads\.com/shelf/show/([a-z0-9-]+)"', html) | |
| genres = [g.replace("-", " ").title() for g in genres[:5]] | |
| if genres: | |
| result["genre"] = _normalize_genre(", ".join(genres[:5])) | |
| # Description | |
| for pat in [ | |
| r'data-testid="description"[^>]*>(.*?)</div>', | |
| r'id="description"[^>]*>(.*?)</div>', | |
| r'class="readable stacked"[^>]*>(.*?)</div>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = _strip_html(m.group(1)) | |
| d = _sanitize_book_description(d) | |
| if len(d) >= 40: | |
| result["description"] = d | |
| break | |
| # About the author | |
| for pat in [ | |
| r'data-testid="contentContainer"[^>]*>(.*?)</div>', | |
| r'class="[^"]*aboutAuthorContainer[^"]*"[^>]*>(.*?)</div>', | |
| r'id="freeTextauthor[^"]*"[^>]*>(.*?)</span>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| bio = _strip_html(m.group(1)) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if len(bio) >= 50: | |
| result["about_author"] = bio | |
| break | |
| # Cover | |
| for pat in [ | |
| r'data-testid="bookCover"[^>]*src="([^"]+)"', | |
| r'class="[^"]*BookCover[^"]*"[^>]*src="([^"]+)"', | |
| r'id="coverImage"[^>]*src="([^"]+)"', | |
| r'"image"\s*:\s*"(https://[^"]+goodreads[^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["cover_url"] = m.group(1) | |
| break | |
| # Rating | |
| m = re.search(r'data-testid="ratingsSummary".*?(\d[\d.]+)\s*(?:avg|rating|stars)', html, re.DOTALL | re.IGNORECASE) | |
| if not m: | |
| m = re.search(r'"ratingValue"\s*:\s*"?([\d.]+)"?', html) | |
| if m: | |
| result["rating"] = m.group(1) | |
| # Rating count | |
| m = re.search(r'([\d,]+)\s+rating', html, re.IGNORECASE) | |
| if m: | |
| result["rating_count"] = m.group(1).replace(",", "") | |
| # Pages | |
| m = re.search(r'(\d+)\s+pages?', html, re.IGNORECASE) | |
| if m: | |
| result["page_count"] = int(m.group(1)) | |
| # Publisher / year | |
| m = re.search(r'Published[^<]*?(\w+ \d+, \d{4}|\w+ \d{4}|\d{4})', html, re.IGNORECASE) | |
| if m: | |
| yr = re.search(r"(\d{4})", m.group(1)) | |
| if yr: | |
| result["publish_year"] = yr.group(1) | |
| # Kindle ASIN from Goodreads affiliate buy links | |
| for pat in (r"creativeASIN=([A-Z0-9]{10})", r"gp/product/([A-Z0-9]{10})"): | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| asin = m.group(1).upper() | |
| result["asin"] = asin | |
| result["buy_url"] = f"https://www.amazon.com/dp/{asin}" | |
| break | |
| return result | |
| def _parse_barnes_noble_html(html: str) -> dict: | |
| """Deep-parse Barnes & Noble product page (legacy DOM + Shopify Oxygen SSR).""" | |
| result: dict = {} | |
| # ── Shopify Oxygen SSR (2024+): meta tags carry title, synopsis, cover ── | |
| og_title = _extract_meta(html, "title") | |
| if not og_title: | |
| m = re.search(r"<title>([^<]+)</title>", html, re.IGNORECASE) | |
| og_title = m.group(1) if m else "" | |
| title = _clean_retailer_title(og_title) | |
| if title and not _title_looks_bad(title): | |
| result["title"] = title | |
| meta_desc = _extract_page_meta_description(html) | |
| if meta_desc: | |
| desc = _sanitize_book_description(meta_desc) | |
| if desc and not _description_looks_bad(desc): | |
| result["description"] = desc | |
| cover = _extract_meta(html, "image") | |
| if not cover: | |
| m = re.search( | |
| r'(https://cdn\.shopify\.com/s/files/[^"\']+\.(?:jpg|jpeg|png|webp)[^"\']*)', | |
| html, re.IGNORECASE, | |
| ) | |
| cover = m.group(1) if m else "" | |
| if cover and not cover.endswith(".svg"): | |
| result["cover_url"] = cover | |
| isbn_in_cover = re.search(r"/files/(\d{13})_", cover) | |
| if isbn_in_cover: | |
| result["isbn"] = isbn_in_cover.group(1) | |
| m = re.search(r'>By\s*(?:<span[^>]*>\s*)?<a[^>]*href="([^"]*(?:contributorId=|/authors/)[^"]+)"[^>]*>([^<]+)</a>', html, re.IGNORECASE) | |
| if m: | |
| url = html_module.unescape(m.group(1)) | |
| if url.startswith('/'): | |
| url = 'https://www.barnesandnoble.com' + url | |
| result["author_url"] = url | |
| author_name = _strip_html(m.group(2)).strip() | |
| if _is_valid_author(author_name): | |
| result["author"] = author_name | |
| else: | |
| m = re.search(r'/b/contributor/([a-z0-9-]+)', html, re.IGNORECASE) | |
| if m: | |
| slug = m.group(1) | |
| result["author_url"] = f"https://www.barnesandnoble.com/b/contributor/{slug}" | |
| author_name = _contributor_slug_to_name(slug) | |
| if _is_valid_author(author_name): | |
| result["author"] = author_name | |
| # JSON-LD Book block — legacy B&N pages | |
| for block in _extract_json_ld(html): | |
| types = block.get("@type", "") | |
| type_list = types if isinstance(types, list) else [types] if types else [] | |
| if not any(t == "Book" for t in type_list): | |
| continue | |
| if block.get("name") and not result.get("title"): | |
| result["title"] = _clean_retailer_title(_strip_html(str(block["name"]))) | |
| authors = block.get("author") | |
| if authors and not result.get("author"): | |
| if isinstance(authors, list): | |
| names = [ | |
| a.get("name", "") if isinstance(a, dict) else str(a) | |
| for a in authors | |
| ] | |
| author = ", ".join(n for n in names if n and _is_valid_author(n)) | |
| if author: | |
| result["author"] = author | |
| elif isinstance(authors, dict): | |
| name = authors.get("name", "") | |
| if _is_valid_author(name): | |
| result["author"] = name | |
| elif isinstance(authors, str) and _is_valid_author(authors): | |
| result["author"] = authors | |
| genre_val = block.get("genre") | |
| if genre_val and not result.get("genre"): | |
| if isinstance(genre_val, list): | |
| result["genre"] = _normalize_genre(", ".join(str(g) for g in genre_val)) | |
| else: | |
| result["genre"] = _normalize_genre(str(genre_val)) | |
| if block.get("description"): | |
| desc = _sanitize_book_description(_decode_json_string(str(block["description"]))) | |
| if desc and not _description_looks_bad(desc): | |
| if not result.get("description") or len(desc) > len(result["description"]): | |
| result["description"] = desc | |
| image = block.get("image") | |
| if image and not result.get("cover_url"): | |
| if isinstance(image, str): | |
| result["cover_url"] = image | |
| elif isinstance(image, list) and image: | |
| result["cover_url"] = image[0] if isinstance(image[0], str) else image[0].get("url", "") | |
| elif isinstance(image, dict): | |
| result["cover_url"] = image.get("url", "") | |
| isbn = _normalize_isbn(str(block.get("isbn") or block.get("gtin13") or "")) | |
| if isbn: | |
| result["isbn"] = isbn | |
| # Legacy DOM selectors (pre-Shopify storefront) | |
| if not result.get("title"): | |
| for pat in ( | |
| r'class="[^"]*product-info-title[^"]*"[^>]*>([^<]+)', | |
| r'<h1[^>]*itemprop="name"[^>]*>([^<]+)', | |
| r'"productName"\s*:\s*"([^"]{3,300})"', | |
| ): | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["title"] = _strip_html(m.group(1)).strip() | |
| break | |
| if not result.get("author"): | |
| for pat in ( | |
| r'class="[^"]*contributor-name[^"]*"[^>]*>([^<]+)', | |
| r'class="[^"]*contributor[^"]*"[^>]*>\s*<a[^>]*>([^<]+)', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| author = _strip_html(m.group(1)).strip() | |
| if _is_valid_author(author): | |
| result["author"] = author | |
| break | |
| if not result.get("author_url"): | |
| m = re.search( | |
| r'href="(https?://[^"]*barnesandnoble\.com/b/contributor/[^"]+)"', | |
| html, re.IGNORECASE, | |
| ) | |
| if m: | |
| result["author_url"] = m.group(1) | |
| if not result.get("description"): | |
| for pat in ( | |
| r'id="[^"]*productDetails[^"]*"[^>]*>(.*?)</div>\s*</div>', | |
| r'class="[^"]*synopsis[^"]*"[^>]*>(.*?)</div>', | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|span|p)>', | |
| r'"description"\s*:\s*"((?:[^"\\]|\\.){40,})"', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| raw = m.group(1) | |
| d = _sanitize_book_description( | |
| _decode_json_string(raw) if "\\" in raw else _strip_scripts(raw) | |
| ) | |
| if len(d) >= 40 and not _description_looks_bad(d): | |
| result["description"] = d | |
| break | |
| for pat in ( | |
| r'class="[^"]*author-bio[^"]*"[^>]*>(.*?)</div>', | |
| r'About the Author[^<]*</[^>]+>(.*?)</(?:div|section)>', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| bio = _sanitize_about_author( | |
| _strip_scripts(m.group(1)), result.get("description", ""), | |
| ) | |
| if bio: | |
| result["about_author"] = bio | |
| break | |
| if not result.get("cover_url"): | |
| for pat in ( | |
| r'"image"\s*:\s*"(https://[^"]+barnesandnoble[^"]+)"', | |
| r'itemprop="image"[^>]*(?:src|content)="([^"]+)"', | |
| r'id="[^"]*pdp-image[^"]*"[^>]*src="([^"]+)"', | |
| ): | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["cover_url"] = m.group(1) | |
| break | |
| if not result.get("genre"): | |
| genres = re.findall( | |
| r'(?:itemprop="genre"|data-category|breadcrumb)[^>]*>([^<]{3,60})', | |
| html, re.IGNORECASE, | |
| ) | |
| genres += re.findall(r'"category(?:Name)?"\s*:\s*"([^"]{3,60})"', html, re.IGNORECASE) | |
| genres += re.findall(r'class="[^"]*category[^"]*"[^>]*>([^<]{3,60})', html, re.IGNORECASE) | |
| if genres: | |
| result["genre"] = _normalize_genre(", ".join(dict.fromkeys(g.strip() for g in genres[:6]))) | |
| m = re.search(r'(\d+)\s+pages?', html, re.IGNORECASE) | |
| if m: | |
| result["page_count"] = int(m.group(1)) | |
| m = re.search(r'Publisher[^<]*</[^>]+>\s*([^<]{3,120})', html, re.IGNORECASE) | |
| if m: | |
| pub = _strip_html(m.group(1)).strip() | |
| yr = re.search(r"(\d{4})", pub) | |
| if yr: | |
| result["publish_year"] = yr.group(1) | |
| result["publisher"] = re.sub(r"\(.*?\)", "", pub).strip()[:200] | |
| isbn = _normalize_isbn(result.get("isbn") or _find_isbn(html)) | |
| if isbn: | |
| result["isbn"] = isbn | |
| return result | |
| def _parse_kobo_html(html: str) -> dict: | |
| """Deep-parse Kobo ebook product page.""" | |
| result: dict = {} | |
| for pat in [ | |
| r'<h1[^>]*class="[^"]*title[^"]*"[^>]*>([^<]+)', | |
| r'itemprop="name"[^>]*>([^<]+)', | |
| r'"title"\s*:\s*"([^"]{3,300})"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["title"] = _strip_html(m.group(1)).strip() | |
| break | |
| for pat in [ | |
| r'class="[^"]*contributor[^"]*"[^>]*>\s*<a[^>]*>([^<]+)', | |
| r'itemprop="author"[^>]*>([^<]+)', | |
| r'"author"\s*:\s*\{[^}]*"name"\s*:\s*"([^"]+)"', | |
| r'"authorName"\s*:\s*"([^"]{2,100})"', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| result["author"] = _strip_html(m.group(1)).strip() | |
| break | |
| for pat in [ | |
| r'class="[^"]*synopsis[^"]*"[^>]*>(.*?)</div>', | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|p)>', | |
| r'id="[^"]*book-description[^"]*"[^>]*>(.*?)</div>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = _strip_scripts(m.group(1)) | |
| if len(d) > 40: | |
| result["description"] = d[:_MAX_DESC] | |
| break | |
| for pat in [ | |
| r'class="[^"]*author-biography[^"]*"[^>]*>(.*?)</div>', | |
| r'About the author[^<]*</[^>]+>(.*?)</(?:div|section)>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| bio = _strip_scripts(m.group(1)) | |
| if len(bio) > 50: | |
| result["about_author"] = bio[:_MAX_BIO] | |
| break | |
| for pat in [ | |
| r'class="[^"]*cover-image[^"]*"[^>]*src="([^"]+)"', | |
| r'itemprop="image"[^>]*(?:src|content)="([^"]+)"', | |
| r'"coverImageUrl"\s*:\s*"(https://[^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["cover_url"] = m.group(1) | |
| break | |
| genres = re.findall(r'class="[^"]*category[^"]*"[^>]*>([^<]{3,60})', html, re.IGNORECASE) | |
| if genres: | |
| result["genre"] = ", ".join(dict.fromkeys(g.strip() for g in genres[:5])) | |
| m = re.search(r'(\d+)\s+pages?', html, re.IGNORECASE) | |
| if m: | |
| result["page_count"] = int(m.group(1)) | |
| result.setdefault("isbn", _find_isbn(html)) | |
| return result | |
| def _parse_apple_books_html(html: str) -> dict: | |
| """Deep-parse Apple Books page — heavy JS; extract embedded JSON when present.""" | |
| result: dict = {} | |
| for pat in [ | |
| r'"name"\s*:\s*"([^"]{3,300})"', | |
| r'<h1[^>]*>([^<]{3,300})</h1>', | |
| r'property="og:title"\s+content="([^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| title = _strip_html(m.group(1)).strip() | |
| if title and "apple books" not in title.lower(): | |
| result["title"] = title | |
| break | |
| for pat in [ | |
| r'"artistName"\s*:\s*"([^"]{2,100})"', | |
| r'"author"\s*:\s*"([^"]{2,100})"', | |
| r'itemprop="author"[^>]*>([^<]+)', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["author"] = _strip_html(m.group(1)).strip() | |
| break | |
| for pat in [ | |
| r'"description"\s*:\s*"((?:[^"\\]|\\.){40,})"', | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|p)>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = m.group(1).replace("\\n", "\n").replace('\\"', '"') | |
| d = _strip_html(d) | |
| if len(d) > 40: | |
| result["description"] = d[:_MAX_DESC] | |
| break | |
| for pat in [ | |
| r'"artworkUrl\d+x\d+"\s*:\s*"(https://[^"]+)"', | |
| r'"artwork"\s*:\s*\{[^}]*"url"\s*:\s*"(https://[^"]+)"', | |
| r'property="og:image"\s+content="([^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| cover = m.group(1).replace("http://", "https://") | |
| result["cover_url"] = cover | |
| break | |
| genres = re.findall(r'"genreNames"\s*:\s*\[([^\]]+)\]', html) | |
| if genres: | |
| names = re.findall(r'"([^"]+)"', genres[0]) | |
| if names: | |
| result["genre"] = ", ".join(names[:5]) | |
| result.setdefault("isbn", _find_isbn(html)) | |
| return result | |
| def _clean_abebooks_author(author: str) -> str: | |
| """Strip brackets and trailing series names from AbeBooks author metadata. | |
| AbeBooks often encodes author as "[Ian Fleming / Bondiana]" where the | |
| text after "/" is a series/imprint name, not a co-author. | |
| """ | |
| author = (author or "").strip().strip("[]").strip() | |
| author = author.split("/")[0].strip() | |
| return author | |
| def _parse_abebooks_html(html: str) -> dict: | |
| """Deep-parse AbeBooks product or search listing page.""" | |
| result: dict = {} | |
| # AbeBooks search/listing pages expose title+author only in the boilerplate | |
| # meta description: "TITLE by [AUTHOR] and a great selection of related books…" | |
| meta_desc = _extract_meta(html, "description") | |
| bm = re.match( | |
| r"^(.{2,200}?)\s+by\s+\[?([^\]/]+?)[\]/].*?(?:and a great selection|available now)", | |
| meta_desc, re.IGNORECASE, | |
| ) | |
| if not bm: | |
| bm = re.match( | |
| r"^(.{2,200}?)\s+by\s+(.{2,80}?)\s+and a great selection", | |
| meta_desc, re.IGNORECASE, | |
| ) | |
| if bm: | |
| cand_title = _clean_title(bm.group(1).strip()) | |
| cand_author = bm.group(2).strip() | |
| if cand_title and not _title_looks_bad(cand_title): | |
| result["title"] = cand_title | |
| if _is_valid_author(cand_author): | |
| result["author"] = cand_author | |
| # Multiple itemprop="name" meta tags exist (site org name "AbeBooks" plus | |
| # the actual book name) — take the first candidate that isn't the org name. | |
| for cand in re.findall(r'<meta[^>]*itemprop="name"[^>]*content="([^"]+)"', html, re.IGNORECASE): | |
| cand = _strip_html(cand).strip() | |
| if not cand or cand.lower() == "abebooks" or cand.startswith("["): | |
| continue | |
| if not result.get("title"): | |
| result["title"] = cand | |
| break | |
| for pat in [ | |
| r'<h1[^>]*itemprop="name"[^>]*>([^<]+)', | |
| r'<span[^>]*itemprop="name"[^>]*>([^<]+)', | |
| r'property="og:title"\s+content="([^"]+)"', | |
| r'"name"\s*:\s*"([^"]{3,300})"', | |
| ]: | |
| if result.get("title"): | |
| break | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| title = _strip_html(m.group(1)).strip() | |
| title = re.sub(r"\s*[-|]\s*AbeBooks.*$", "", title, flags=re.IGNORECASE).strip() | |
| if title and not title.lower().startswith("abebooks") and not result.get("title"): | |
| result["title"] = title | |
| break | |
| for pat in [ | |
| r'<meta[^>]*itemprop="author"[^>]*content="([^"]+)"', | |
| r'itemprop="author"[^>]*>([^<]+)', | |
| r'class="[^"]*author[^"]*"[^>]*>\s*<a[^>]*>([^<]+)', | |
| r'"author"\s*:\s*"([^"]{2,100})"', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| author = _clean_abebooks_author(_strip_html(m.group(1)).strip()) | |
| if _is_valid_author(author): | |
| result["author"] = author | |
| break | |
| for pat in [ | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|span|p)>', | |
| r'class="[^"]*book-description[^"]*"[^>]*>(.*?)</div>', | |
| r'property="og:description"\s+content="([^"]{40,})"', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = _sanitize_book_description(m.group(1)) | |
| if len(d) >= 40: | |
| result["description"] = d | |
| break | |
| # Individual listing pages: <meta name="description"> is the seller's own | |
| # condition/synopsis note (not the generic search-page boilerplate) — use | |
| # it when nothing better was found and it isn't the marketing blurb. | |
| if not result.get("description") and meta_desc and not _description_looks_bad(meta_desc): | |
| d = _sanitize_book_description(meta_desc) | |
| title = result.get("title", "") | |
| if title: | |
| d = re.sub(r"\s*-\s*" + re.escape(title) + r"\s*$", "", d, flags=re.IGNORECASE).strip() | |
| if len(d) >= 40: | |
| result["description"] = d | |
| for pat in [ | |
| r'itemprop="image"[^>]*(?:src|content)="([^"]+)"', | |
| r'property="og:image"\s+content="([^"]+)"', | |
| r'id="main-image"[^>]*src="([^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m and not _is_placeholder_cover(m.group(1)): | |
| result["cover_url"] = m.group(1) | |
| break | |
| isbn = _normalize_isbn(_find_isbn(html)) | |
| if isbn: | |
| result["isbn"] = isbn | |
| return result | |
| def _parse_bookshop_html(html: str) -> dict: | |
| """Deep-parse Bookshop.org product page.""" | |
| result: dict = {} | |
| for pat in [ | |
| r'<h1[^>]*class="[^"]*book-title[^"]*"[^>]*>([^<]+)', | |
| r'itemprop="name"[^>]*>([^<]+)', | |
| r'property="og:title"\s+content="([^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["title"] = _strip_html(m.group(1)).strip() | |
| break | |
| for pat in [ | |
| r'itemprop="author"[^>]*>([^<]+)', | |
| r'class="[^"]*book-author[^"]*"[^>]*>([^<]+)', | |
| r'"author"\s*:\s*"([^"]{2,100})"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["author"] = _strip_html(m.group(1)).strip() | |
| break | |
| for pat in [ | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|p)>', | |
| r'class="[^"]*book-description[^"]*"[^>]*>(.*?)</div>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = _strip_scripts(m.group(1)) | |
| if len(d) > 40: | |
| result["description"] = d[:_MAX_DESC] | |
| break | |
| for pat in [ | |
| r'itemprop="image"[^>]*(?:src|content)="([^"]+)"', | |
| r'property="og:image"\s+content="([^"]+)"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| result["cover_url"] = m.group(1) | |
| break | |
| result.setdefault("isbn", _find_isbn(html)) | |
| return result | |
| def _parse_thriftbooks_html(html: str) -> dict: | |
| """Deep-parse ThriftBooks: 'Book Overview' synopsis + author profile link.""" | |
| result = _parse_generic_html(html, "thriftbooks") | |
| result.pop("about_author", None) # TB injects review JSON-LD here — never trust it | |
| # Description lives under the "Book Overview" heading | |
| if not result.get("description") or _description_looks_bad(result.get("description", "")): | |
| for pat in ( | |
| r'Book Overview\s*</[^>]+>(.*?)</(?:div|section)>', | |
| r'class="[^"]*WorkMeta-overview[^"]*"[^>]*>(.*?)</div>\s*</div>', | |
| r'class="[^"]*Expandable[^"]*"[^>]*>(.*?)</div>', | |
| r'id="[^"]*BookOverview[^"]*"[^>]*>(.*?)</div>', | |
| ): | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = _sanitize_book_description(_strip_scripts(m.group(1))) | |
| if len(d) >= 40 and not _description_looks_bad(d): | |
| result["description"] = d | |
| break | |
| # Author profile link: /a/{slug}/{id}/ | |
| m = re.search(r'href="(/a/[a-z0-9-]+/\d+/?)"', html, re.IGNORECASE) | |
| if m: | |
| result["author_url"] = "https://www.thriftbooks.com" + m.group(1) | |
| return result | |
| def _deep_parse_html(html: str, platform: str) -> dict: | |
| """Route HTML to the best platform-specific parser.""" | |
| parsers = { | |
| "amazon": _parse_amazon_html, | |
| "goodreads": _parse_goodreads_html, | |
| "barnes_noble": _parse_barnes_noble_html, | |
| "kobo": _parse_kobo_html, | |
| "apple_books": _parse_apple_books_html, | |
| "bookshop": _parse_bookshop_html, | |
| "abebooks": _parse_abebooks_html, | |
| "thriftbooks": _parse_thriftbooks_html, | |
| } | |
| parser = parsers.get(platform, lambda h: _parse_generic_html(h, platform)) | |
| return parser(html) | |
| def _parse_generic_html(html: str, platform: str) -> dict: | |
| """Generic deep-parse for Barnes & Noble, Kobo, Apple Books, ThriftBooks, AbeBooks, etc.""" | |
| result: dict = {} | |
| # ── Cover ───────────────────────────────────────────────────────────────── | |
| # Most book pages have a prominent product image with alt text containing the title | |
| for pat in [ | |
| r'class="[^"]*product[^"]*cover[^"]*"[^>]*src="([^"]+)"', | |
| r'class="[^"]*book[^"]*cover[^"]*"[^>]*src="([^"]+)"', | |
| r'class="[^"]*cover[^"]*image[^"]*"[^>]*src="([^"]+)"', | |
| r'id="[^"]*cover[^"]*"[^>]*src="([^"]+)"', | |
| r'itemprop="image"[^>]*(?:src|content)="([^"]+)"', | |
| r'"thumbnailUrl"\s*:\s*"(https://[^"]+)"', | |
| r'"image"\s*:\s*"(https?://[^"]{20,})"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m and not m.group(1).endswith(".svg"): | |
| result["cover_url"] = m.group(1) | |
| break | |
| # ── Genre ───────────────────────────────────────────────────────────────── | |
| for pat in [ | |
| r'(?:Genre|Category|Subject)[s]?\s*[:\|]?\s*</[^>]+>\s*<[^>]*>([^<]{3,80})', | |
| r'itemprop="genre"[^>]*>([^<]+)', | |
| r'class="[^"]*genre[^"]*"[^>]*>([^<]{3,60})', | |
| r'class="[^"]*category[^"]*"[^>]*>([^<]{3,60})', | |
| r'"genre"\s*:\s*"([^"]{3,100})"', | |
| ]: | |
| m = re.search(pat, html, re.IGNORECASE) | |
| if m: | |
| g = _strip_html(m.group(1)).strip() | |
| if len(g) > 2 and g.lower() not in ("see all", "more", "books"): | |
| result["genre"] = g[:300] | |
| break | |
| # JSON-LD genre (works on many stores) | |
| if not result.get("genre"): | |
| for block in _extract_json_ld(html): | |
| g = block.get("genre", "") | |
| if g: | |
| result["genre"] = (", ".join(g) if isinstance(g, list) else str(g))[:300] | |
| break | |
| # ── Description ──────────────────────────────────────────────────────────── | |
| for pat in [ | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|section|p)>', | |
| r'class="[^"]*product[^"]*description[^"]*"[^>]*>(.*?)</div>', | |
| r'class="[^"]*book[^"]*description[^"]*"[^>]*>(.*?)</div>', | |
| r'id="[^"]*description[^"]*"[^>]*>(.*?)</div>', | |
| r'"description"\s*:\s*"([^"]{30,})"', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| d = _strip_html(m.group(1)) | |
| d = _sanitize_book_description(d) | |
| if len(d) >= 40: | |
| result["description"] = d | |
| break | |
| # ── Author bio ───────────────────────────────────────────────────────────── | |
| for pat in [ | |
| r'(?:About the Author|Author Bio)[^<]*</[^>]+>(.*?)</(?:div|section)>', | |
| r'class="[^"]*author[^"]*bio[^"]*"[^>]*>(.*?)</div>', | |
| r'itemprop="author".*?(?:description|bio)[^>]*>(.*?)</div>', | |
| ]: | |
| m = re.search(pat, html, re.DOTALL | re.IGNORECASE) | |
| if m: | |
| bio = _strip_html(m.group(1)) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if len(bio) >= 50: | |
| result["about_author"] = bio | |
| break | |
| # ── Pages / Publisher ────────────────────────────────────────────────────── | |
| m = re.search(r'(\d+)\s+pages?', html, re.IGNORECASE) | |
| if m: | |
| result["page_count"] = int(m.group(1)) | |
| m = re.search(r'(?:Publisher|Imprint)\s*:?\s*</[^>]+>\s*([^<]{3,100})', html, re.IGNORECASE) | |
| if m: | |
| result["publisher"] = _strip_html(m.group(1)).strip()[:200] | |
| yr = re.search(r'(?:Publish(?:ed)?|Publication)[^<]*(\d{4})', html, re.IGNORECASE) | |
| if yr: | |
| result["publish_year"] = yr.group(1) | |
| # Barnes & Noble specific | |
| if platform == "barnes_noble": | |
| m = re.search(r'"productFormat"\s*:\s*"([^"]+)"', html) | |
| if m and "page" in m.group(1).lower(): | |
| pg = re.search(r'(\d+)', m.group(1)) | |
| if pg: | |
| result.setdefault("page_count", int(pg.group(1))) | |
| return result | |
| # ── Master resolution function ───────────────────────────────────────────────── | |
| async def _resolve(url: str, platform: str, ids: dict, client: httpx.AsyncClient) -> dict: | |
| """Multi-strategy resolver. APIs first (never blocked), then HTML enrichment. | |
| Stages: | |
| 1. Google Books by volume ID (Google Books URLs) | |
| 2. Open Library by work/edition key (OL URLs) | |
| 3. Google Books + Open Library by ISBN (parallel, covers most books) | |
| 4. HTML fetch → OG/JSON-LD + platform-specific deep parse | |
| 5. Google Books text search (last resort) | |
| Stage 4 (HTML) now runs as an ENRICHMENT step even after a successful API hit, | |
| filling any gaps in genre, cover, description, and author bio. | |
| """ | |
| result: dict = {"buy_url": url.split("?")[0], "warnings": []} | |
| slug_title, slug_author = _title_author_from_url_slug(url) | |
| skip_slug = bool( | |
| ids.get("isbn") or ids.get("google_volume_id") or ids.get("oclc") | |
| or ids.get("ol_work_key") or ids.get("ol_edition_key") | |
| ) | |
| if slug_title and not skip_slug: | |
| result.setdefault("title", slug_title) | |
| if slug_author and not skip_slug: | |
| result.setdefault("author", slug_author) | |
| # ── Apple Books: iTunes Lookup API (never blocked) ─────────────────────── | |
| if ids.get("apple_book_id"): | |
| apple = await _apple_itunes_lookup(ids["apple_book_id"], client) | |
| for k, v in apple.items(): | |
| if not v: | |
| continue | |
| if k == "title" and not _title_looks_bad(v): | |
| result.setdefault("title", _clean_title(v)) | |
| elif k == "author" and _is_valid_author(v): | |
| result.setdefault("author", v) | |
| elif not result.get(k): | |
| result[k] = v | |
| if ids.get("google_volume_id"): | |
| gb = await _google_by_volume_id(ids["google_volume_id"], client) | |
| if gb: | |
| for k, v in gb.items(): | |
| if not v: | |
| continue | |
| if k == "title": | |
| result["title"] = _clean_title(v) | |
| elif k == "author" and _is_valid_author(v): | |
| result["author"] = v | |
| else: | |
| result[k] = v | |
| if gb.get("isbn"): | |
| ol = await _ol_by_isbn(gb["isbn"], client) | |
| for ak in (ol.get("author_keys") or [])[:1]: | |
| _, bio = await _ol_author_bio(ak, client) | |
| if bio: | |
| result.setdefault("about_author", bio) | |
| break | |
| # ── STRATEGY 2: Open Library work/edition key (OL URLs) ────────────────── | |
| if ids.get("ol_work_key"): | |
| ol = await _ol_by_work_key(ids["ol_work_key"], client) | |
| result.update({k: v for k, v in ol.items() if v and not result.get(k)}) | |
| if ids.get("ol_edition_key"): | |
| ol_e = await _ol_by_edition_key(ids["ol_edition_key"], client) | |
| result.update({k: v for k, v in ol_e.items() if v and not result.get(k)}) | |
| for ak in result.get("author_keys", [])[:2]: | |
| name, bio = await _ol_author_bio(ak, client) | |
| if name and not result.get("author"): | |
| result["author"] = name | |
| if bio and not result.get("about_author"): | |
| result["about_author"] = bio | |
| if result.get("author") and result.get("about_author"): | |
| break | |
| # ── STRATEGY 2b: WorldCat OCLC → Open Library (never blocked) ───────────── | |
| if ids.get("oclc"): | |
| ol_oclc = await _ol_by_oclc(ids["oclc"], client) | |
| for k, v in ol_oclc.items(): | |
| if not v: | |
| continue | |
| if k == "title" and (_title_looks_bad(result.get("title", "")) or not result.get("title")): | |
| result["title"] = _clean_title(v) | |
| elif k == "author" and (_is_valid_author(v) and not _is_valid_author(result.get("author", ""))): | |
| result["author"] = v | |
| elif not result.get(k): | |
| result[k] = v | |
| if ol_oclc.get("isbn"): | |
| result.setdefault("isbn", ol_oclc["isbn"]) | |
| for ak in ol_oclc.get("author_keys", [])[:2]: | |
| _, bio = await _ol_author_bio(ak, client) | |
| if bio: | |
| result.setdefault("about_author", bio) | |
| break | |
| if not result.get("description") and ol_oclc.get("isbn"): | |
| desc = await _ol_description_by_isbn(ol_oclc["isbn"], client) | |
| if desc: | |
| result["description"] = desc | |
| # ── STRATEGY 3: ISBN from URL → Google Books + Open Library (parallel) ─── | |
| isbn = ids.get("isbn", "") | |
| if not isbn and ids.get("isbn10"): | |
| isbn = _isbn10_to_isbn13(ids["isbn10"]) or ids["isbn10"] | |
| if not isbn and ids.get("asin"): | |
| isbn = _asin_to_isbn13(ids["asin"]) | |
| if isbn: | |
| logger.debug("Converted ASIN to ISBN-13", asin=ids["asin"], isbn=isbn) | |
| else: | |
| result["warnings"].append("ASIN is not an ISBN-10 (B0-style ASIN)") | |
| result.setdefault("asin", ids["asin"]) | |
| if isbn: | |
| _apply_isbn_pair(result, *_extract_isbn_pair(isbn)) | |
| result["isbn"] = isbn | |
| if isbn and not result.get("title"): | |
| gb, ol = await asyncio.gather( | |
| _google_by_isbn(isbn, client, author_hint=slug_author), | |
| _ol_by_isbn(isbn, client), | |
| ) | |
| result.update({k: v for k, v in gb.items() if v}) | |
| for key in ("title", "author", "author_keys", "genre", "publisher", | |
| "publish_year", "page_count", "cover_url"): | |
| if ol.get(key) and not result.get(key): | |
| result[key] = ol[key] | |
| for ak in result.get("author_keys", [])[:1]: | |
| _, bio = await _ol_author_bio(ak, client) | |
| if bio: | |
| result.setdefault("about_author", bio) | |
| break | |
| elif isbn: | |
| # ISBN known, title already set — just enrich missing fields | |
| gb, ol = await asyncio.gather( | |
| _google_by_isbn(isbn, client, author_hint=slug_author), | |
| _ol_by_isbn(isbn, client), | |
| ) | |
| for key in ("genre", "description", "cover_url", "rating", "rating_count", | |
| "page_count", "language", "publisher", "publish_year"): | |
| if gb.get(key) and not result.get(key): | |
| result[key] = gb[key] | |
| if gb.get("title") and _title_looks_bad(result.get("title", "")): | |
| result["title"] = _clean_title(gb["title"]) | |
| if gb.get("author") and not _is_valid_author(result.get("author", "")): | |
| result["author"] = gb["author"] | |
| elif slug_author and gb.get("author"): | |
| result["author"] = _resolve_author(gb["author"], slug_author) | |
| for key in ("genre", "cover_url", "publisher", "publish_year", "page_count"): | |
| if ol.get(key) and not result.get(key): | |
| result[key] = ol[key] | |
| for ak in (ol.get("author_keys") or [])[:1]: | |
| _, bio = await _ol_author_bio(ak, client) | |
| if bio: | |
| result.setdefault("about_author", bio) | |
| break | |
| # ── STRATEGY 4: HTML fetch — always run for enrichment ─────────────────── | |
| # Even when APIs returned a title, the product page has: | |
| # • High-res cover image • Genre tags • Full synopsis | |
| # • Author bio section • Publisher/year • Platform-specific fields | |
| _FIELDS_TO_FILL = ("genre", "cover_url", "description", "about_author", | |
| "publisher", "publish_year", "page_count", "rating", | |
| "rating_count", "author", "title") | |
| needs_html = any(not result.get(f) for f in _FIELDS_TO_FILL) | |
| if platform == "barnes_noble": | |
| needs_html = True | |
| if needs_html: | |
| if platform == "barnes_noble": | |
| html = await _fetch_bn_html(url, client) | |
| elif platform == "amazon": | |
| html = await _fetch_amazon_html(url, client, asin=ids.get("asin", "")) | |
| else: | |
| html = await _fetch_html(url, client) | |
| if html: | |
| deep = _deep_parse_html(html, platform) | |
| if platform == "amazon": | |
| # For Amazon, deep-parse results WIN over OG meta because | |
| # Amazon's OG <meta> tags are set to the store page title | |
| # (e.g. "Amazon.com: Book Title: Author: Kindle Store") | |
| # rather than the actual book title/description. | |
| for k, v in deep.items(): | |
| if v: | |
| result[k] = v | |
| og = _parse_html_og(html) | |
| for k, v in og.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| author_url = result.get("author_url") | |
| if not result.get("about_author") and author_url: | |
| logger.info("Fetching Amazon author page for bio", author_url=author_url) | |
| bio = await _fetch_author_page_bio( | |
| author_url, client, | |
| [ | |
| r'data-testid="aboutAuthorText"[^>]*>(.*?)</', | |
| r'class="[^"]*about-author[^"]*"[^>]*>(.*?)</', | |
| r'"biography"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| r'"bio"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| r'id="ap-bio"[^>]*>(.*?)</div>', | |
| ], | |
| ) | |
| if bio: | |
| result["about_author"] = bio | |
| elif platform == "goodreads": | |
| og = _parse_html_og(html) | |
| for k, v in og.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| for k, v in deep.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| elif k == "cover_url" and v: | |
| result[k] = v | |
| if not result.get("about_author"): | |
| m = re.search( | |
| r'href="(https?://[^"]*goodreads\.com/author/show/\d+[^"]*)"', | |
| html, re.IGNORECASE, | |
| ) | |
| if m: | |
| bio = await _fetch_author_page_bio( | |
| m.group(1), client, | |
| [ | |
| r'data-testid="contentContainer"[^>]*>(.*?)</div>', | |
| r'class="[^"]*aboutAuthorInfo[^"]*"[^>]*>(.*?)</div>', | |
| r'id="freeTextauthor[^"]*"[^>]*>(.*?)</span>', | |
| r'itemprop="description"[^>]*>(.*?)</span>', | |
| ], | |
| ) | |
| if bio: | |
| result["about_author"] = bio | |
| elif platform == "barnes_noble": | |
| # B&N Shopify SSR: meta description + og:image have book data; | |
| # og:description is generic site copy. Deep-parse wins over catalog. | |
| for k, v in deep.items(): | |
| if not v or k == "author_url": | |
| continue | |
| if k == "author" and not _is_valid_author(v): | |
| continue | |
| result[k] = v | |
| og = _parse_html_og(html) | |
| if og.get("title"): | |
| t = _clean_retailer_title(og["title"]) | |
| if t and not _title_looks_bad(t): | |
| result["title"] = t | |
| if og.get("cover_url") and _is_retailer_cover(og["cover_url"]): | |
| result["cover_url"] = og["cover_url"] | |
| meta_desc = _extract_page_meta_description(html) | |
| desc = _sanitize_book_description(meta_desc) | |
| if desc and not _description_looks_bad(desc): | |
| result["description"] = desc | |
| author_url = deep.get("author_url") or result.get("author_url") | |
| if not result.get("about_author") and author_url: | |
| bio = await _fetch_author_page_bio( | |
| author_url, client, | |
| [ | |
| r'class="[^"]*author-bio[^"]*"[^>]*>(.*?)</div>', | |
| r'About the Author[^<]*</[^>]+>(.*?)</(?:div|section)>', | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|p)>', | |
| r'"biography"\s*:\s*"((?:[^"\\]|\\.){50,})"', | |
| r'name="description" content="([^"]{80,})', | |
| ], | |
| ) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if bio: | |
| result["about_author"] = bio | |
| elif platform == "thriftbooks": | |
| og = _parse_html_og(html) | |
| for k, v in og.items(): | |
| if k in ("about_author",): | |
| continue | |
| if v and not result.get(k): | |
| result[k] = v | |
| for k, v in deep.items(): | |
| if not v or k == "author_url": | |
| continue | |
| if v and not result.get(k): | |
| result[k] = v | |
| elif k == "cover_url" and v: | |
| result[k] = v | |
| author_url = deep.get("author_url") | |
| if not result.get("about_author") and author_url: | |
| bio = await _fetch_author_page_bio( | |
| author_url, client, | |
| [ | |
| r'class="[^"]*AuthorOverview-description[^"]*"[^>]*>(.*?)</div>', | |
| r'class="[^"]*Bio[^"]*"[^>]*>(.*?)</div>', | |
| r'About [^<]+</[^>]+>(.*?)</(?:div|section)>', | |
| r'itemprop="description"[^>]*>(.*?)</(?:div|p|span)>', | |
| r'name="description" content="([^"]{80,})', | |
| ], | |
| ) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if bio: | |
| result["about_author"] = bio | |
| elif platform == "abebooks": | |
| # Deep-parse already rejects marketing boilerplate descriptions; | |
| # legit per-listing seller notes (condition/synopsis) pass through. | |
| for k, v in deep.items(): | |
| if not v or k == "author_url": | |
| continue | |
| if k == "author" and not _is_valid_author(v): | |
| continue | |
| if not result.get(k): | |
| result[k] = v | |
| og = _parse_html_og(html) | |
| if og.get("title") and not result.get("title"): | |
| t = re.sub(r"\s*[-|]\s*AbeBooks.*$", "", og["title"], flags=re.IGNORECASE).strip() | |
| t = _clean_title(t) | |
| if t and not _title_looks_bad(t): | |
| result["title"] = t | |
| if og.get("cover_url") and not _is_placeholder_cover(og["cover_url"]): | |
| result["cover_url"] = og["cover_url"] | |
| if og.get("isbn") and not result.get("isbn"): | |
| result["isbn"] = og["isbn"] | |
| else: | |
| og = _parse_html_og(html) | |
| for k, v in og.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| for k, v in deep.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| elif k == "cover_url" and v: | |
| result[k] = v | |
| isbn = _normalize_isbn(result.get("isbn") or _find_isbn(html)) | |
| if isbn: | |
| i10, i13 = _extract_isbn_pair(isbn) | |
| _apply_isbn_pair(result, i10, i13) | |
| elif result.get("isbn") and not _normalize_isbn(result["isbn"]): | |
| result.pop("isbn", None) | |
| result.pop("isbn10", None) | |
| result.pop("isbn13", None) | |
| # ── STRATEGY 4b: Catalog API enrichment (GB + Open Library) ───────────── | |
| # Fills description, author bio, genre, cover for JS-heavy stores where | |
| # static HTML only exposes title/author (Apple Books, Kobo, B&N, etc.). | |
| await _enrich_from_catalog(result, client, author_hint=slug_author) | |
| # Apple Books — supplement iTunes data when ISBN still missing after HTML | |
| if ids.get("apple_book_id") and not result.get("isbn"): | |
| slug = re.sub(r"[^a-z0-9 ]", " ", url.lower()) | |
| words = [w for w in slug.split() if len(w) > 3 and w not in ("books", "apple", "book")] | |
| if words: | |
| gb = await _google_search(" ".join(words[-6:]), client) | |
| for k, v in gb.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| await _enrich_from_catalog(result, client, author_hint=slug_author) | |
| # ── Final catalog gap-fill (author, description, genre, bio, cover) ───── | |
| _apply_bn_slug_fallback(result, url, ids) | |
| await _fill_catalog_gaps(result, client, author_hint=slug_author) | |
| # ── STRATEGY 5: Google Books text search (absolute last resort) ─────────── | |
| if not result.get("title"): | |
| title = result.get("title", "") | |
| author = result.get("author", "") | |
| if title or author: | |
| q = f'intitle:"{title}"' if title else "" | |
| if author: | |
| q += f' inauthor:"{author}"' | |
| gb = await _google_search(q.strip(), client) | |
| else: | |
| # Guess from URL slug | |
| slug = re.sub(r"[^a-z0-9 ]", " ", url.lower().split("?")[0]) | |
| words = [w for w in slug.split() if len(w) > 3 | |
| and w not in ("https", "http", "www", "com", "books", | |
| "book", "show", "detail", "product", "store")] | |
| q = " ".join(words[-5:]) | |
| gb = await _google_search(q, client) if q else {} | |
| for k, v in gb.items(): | |
| if v and not result.get(k): | |
| result[k] = v | |
| # ── GENRE FALLBACK: OL Search + Google Books in parallel ───────────────── | |
| # Runs only when genre is still empty after all 5 strategies. | |
| # Works for any platform — JavaScript-rendered pages (Apple Books, Kobo, | |
| # B&N) never expose genre in static HTML; APIs always do. | |
| if not result.get("genre") and result.get("title"): | |
| genre = await _fetch_genre_fallback( | |
| title=result.get("title", ""), | |
| author=result.get("author", ""), | |
| client=client, | |
| ) | |
| if genre: | |
| result["genre"] = genre | |
| logger.debug("Genre resolved via fallback search", genre=genre[:60]) | |
| if not result.get("description") and platform == "lulu" and result.get("title"): | |
| synth = _synthesize_description(result) | |
| if synth: | |
| result["description"] = synth | |
| result["warnings"].append( | |
| "Lulu page blocked — description synthesized from catalog metadata; edit before saving." | |
| ) | |
| if not result.get("genre"): | |
| result["genre"] = "Fiction" | |
| if platform == "barnes_noble": | |
| _apply_bn_slug_fallback(result, url, ids) | |
| if platform in ("barnes_noble", "thriftbooks", "abebooks"): | |
| if not result.get("description") or not result.get("genre"): | |
| apple = await _apple_itunes_search( | |
| result.get("title", ""), | |
| result.get("author", ""), | |
| client, | |
| ) | |
| if apple.get("description") and not result.get("description"): | |
| result["description"] = apple["description"] | |
| if apple.get("genre") and not result.get("genre"): | |
| result["genre"] = apple["genre"] | |
| if not result.get("about_author") and result.get("author"): | |
| name = _first_author_name(result["author"]) | |
| wiki = await _wikipedia_author_bio(name, client) | |
| wiki = _sanitize_about_author(wiki, result.get("description", "")) | |
| if wiki: | |
| result["about_author"] = wiki | |
| if not result.get("about_author") and result.get("author"): | |
| name = _first_author_name(result["author"]) | |
| _, bio = await _ol_author_search_by_name(name, client) | |
| bio = _sanitize_about_author(bio, result.get("description", "")) | |
| if bio: | |
| result["about_author"] = bio | |
| if result.get("title"): | |
| result["title"] = _clean_title(result["title"]) | |
| if slug_author: | |
| resolved = _resolve_author(result.get("author", ""), slug_author) | |
| if resolved: | |
| result["author"] = resolved | |
| if not _is_valid_author(result.get("author", "")): | |
| result["author"] = "" | |
| if ids.get("asin"): | |
| result["asin"] = ids["asin"].upper() | |
| _merge_isbn_fields(result) | |
| return result | |
| # ── Public entry point ───────────────────────────────────────────────────────── | |
| async def fetch_book_metadata(url: str, redis=None, force_refresh: bool = False) -> BookURLMetadata: | |
| """Fetch complete book metadata from any supported platform URL. | |
| Uses a 5-strategy pipeline. Strategies 1–3 (Google Books API, Open Library | |
| API, ASIN→ISBN conversion) NEVER hit retailer HTML, so they're never blocked. | |
| Strategies 4–5 are HTML fallbacks with UA rotation and retry logic. | |
| Args: | |
| url: Full book product URL (any supported platform). | |
| redis: Optional Redis client for 6h result caching. | |
| force_refresh: If True, bypass cache read but still use cache as fallback | |
| when a live re-fetch returns worse/empty metadata. | |
| Returns: | |
| BookURLMetadata — never raises. `error` field set on complete failure. | |
| """ | |
| url = _normalize_import_url(url.strip()) | |
| platform = detect_platform(url) | |
| meta = BookURLMetadata(source_url=url, platform=platform, buy_url=url) | |
| if platform == "unknown": | |
| # Don't give up! Try generic HTML + Google Books search for any URL. | |
| # Author personal sites, indie stores, custom URLs all work this way. | |
| platform = "generic" | |
| meta.platform = "generic" | |
| meta.warnings = ["Unrecognized platform — using generic fetch + Google Books search"] | |
| # ── Redis cache (always load — used for fast path + re-fetch fallback) ─── | |
| cache_key = _CACHE_PREFIX + hashlib.sha256(url.encode()).hexdigest()[:20] | |
| previous: BookURLMetadata | None = None | |
| if redis is not None: | |
| try: | |
| cached = await redis.get(cache_key) | |
| if cached: | |
| previous = BookURLMetadata(**json.loads(cached)) | |
| if not force_refresh: | |
| return previous | |
| except Exception: | |
| pass | |
| # ── Resolve ──────────────────────────────────────────────────────────────── | |
| ids = _extract_identifiers(url) | |
| logger.info("Resolving book URL", platform=platform, ids=list(ids.keys())) | |
| raw: dict = {} | |
| try: | |
| async with httpx.AsyncClient( | |
| follow_redirects=True, | |
| timeout=_HTML_TIMEOUT, | |
| ) as client: | |
| raw = await _resolve(url, platform, ids, client) | |
| except Exception as e: | |
| logger.warning("Book resolution failed", url=url[:80], error=str(e)) | |
| raw["error"] = str(e) | |
| # ── Populate meta ────────────────────────────────────────────────────────── | |
| title = _clean_title((raw.get("title") or "").strip()) | |
| if _title_looks_bad(title): | |
| title = "" | |
| subtitle = (raw.get("subtitle") or "").strip() | |
| if subtitle and title and not title.endswith(subtitle): | |
| title = f"{title}: {subtitle}" if title else subtitle | |
| author_raw = str(raw.get("author", "")).strip() | |
| meta.title = title[:500] | |
| meta.author = author_raw[:200] if _is_valid_author(author_raw) else "" | |
| meta.description = _sanitize_book_description(str(raw.get("description", ""))) | |
| meta.about_author = _sanitize_about_author( | |
| str(raw.get("about_author", "")), meta.description, | |
| ) | |
| meta.genre = _normalize_genre(str(raw.get("genre", "")))[:300] | |
| meta.isbn10 = _normalize_isbn(str(raw.get("isbn10", ""))) | |
| meta.isbn13 = _normalize_isbn(str(raw.get("isbn13", ""))) | |
| if meta.isbn10 and not meta.isbn13: | |
| meta.isbn13 = _isbn10_to_isbn13(meta.isbn10) | |
| elif meta.isbn13 and not meta.isbn10: | |
| meta.isbn10 = _isbn13_to_isbn10(meta.isbn13) | |
| meta.isbn = meta.isbn13 or meta.isbn10 or _normalize_isbn(str(raw.get("isbn", ""))) | |
| meta.asin = str(ids.get("asin") or raw.get("asin", ""))[:10].upper() | |
| # Kindle-only listings often have no ISBN-13 — use ASIN so the ISBN field is never blank | |
| if not meta.isbn and meta.asin: | |
| meta.isbn = meta.asin | |
| meta.publisher = str(raw.get("publisher", ""))[:200] | |
| meta.publish_year = _valid_publish_year(raw.get("publish_year", "")) | |
| meta.edition = str(raw.get("edition", ""))[:120] | |
| meta.book_format = str(raw.get("book_format", ""))[:80] | |
| meta.series = str(raw.get("series", ""))[:120] | |
| meta.language = str(raw.get("language", ""))[:50] | |
| meta.page_count = int(raw.get("page_count") or 0) | |
| meta.price = str(raw.get("price", ""))[:20] | |
| meta.rating = str(raw.get("rating", ""))[:10] | |
| meta.rating_count = str(raw.get("rating_count", ""))[:20] | |
| # Always HTTPS — some APIs (Google Books, OL) return http:// cover URLs | |
| cover_raw = str(raw.get("cover_url", "")) | |
| if _is_placeholder_cover(cover_raw): | |
| cover_raw = "" | |
| meta.cover_url = cover_raw.replace("http://", "https://", 1) if cover_raw else "" | |
| meta.buy_url = str(raw.get("buy_url", url)) | |
| meta.warnings = list(raw.get("warnings", [])) | |
| meta.error = raw.get("error", "") | |
| meta.confidence = _score(meta) | |
| # ── Re-fetch fallback: never degrade below a good cached result ──────────── | |
| if ( | |
| previous | |
| and previous.title | |
| and not _title_looks_bad(previous.title) | |
| and ( | |
| not meta.title | |
| or meta.confidence + 0.05 < previous.confidence | |
| ) | |
| ): | |
| logger.info( | |
| "Using cached book metadata after weaker re-fetch", | |
| url=url[:60], | |
| cached_confidence=previous.confidence, | |
| fresh_confidence=meta.confidence, | |
| ) | |
| previous.warnings = list(previous.warnings or []) + [ | |
| "Live re-fetch returned less metadata; showing the previous successful result.", | |
| ] | |
| meta = previous | |
| logger.info( | |
| "Book resolved", | |
| platform=platform, | |
| title=meta.title[:50] or "(none)", | |
| isbn=meta.isbn, | |
| isbn13=meta.isbn13, | |
| confidence=meta.confidence, | |
| strategy_warnings=meta.warnings, | |
| ) | |
| # ── Cache good results so repeat fetches stay idempotent ─────────────────── | |
| if ( | |
| redis is not None | |
| and meta.title | |
| and not _title_looks_bad(meta.title) | |
| and meta.confidence >= 0.75 | |
| ): | |
| try: | |
| await redis.setex(cache_key, _CACHE_TTL, json.dumps(asdict(meta))) | |
| except Exception: | |
| pass | |
| return meta | |