Arag / app /services /book_url_scraper.py
AuthorBot
Add GOOGLE_BOOKS_API_KEY for authenticated Books API lookups.
02b30d1
Raw
History Blame Contribute Delete
164 kB
"""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"<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