"""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 ───────────────────────────────────────────────────────────────── @dataclass 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"", "\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 ", "", html_fragment, flags=re.DOTALL | re.IGNORECASE) cleaned = re.sub(r"]*>.*?", "", cleaned, flags=re.DOTALL | re.IGNORECASE) cleaned = re.sub(r"]*>.*?", "", 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']*(?:name|property)=["\'](?:og:|twitter:)?{re.escape(name)}["\']\s[^>]*content=["\'](.*?)["\']', rf']*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']*type=["\']application/ld\+json["\'][^>]*>(.*?)', 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' 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']*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"[^>]*>(.*?)', html, re.DOTALL | re.IGNORECASE, ) if not block: return "" links = re.findall(r"]*>([^<]+)", 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*([^<]+)', 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.*?(\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*([^<]+)', 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*([^<]+).*?' r'rpi-attribute-value.*?([^<]+)', 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 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: 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"[^>]*>.*?]*>\s*([A-Z][^<]{2,80}?)\s*', # 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"[^>]*>.*?]*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"[^>]*>(.*?)(?=]*id=| 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