app/services/platform_presence.py

"""Platform Presence Scanner — discover where a book is listed across retailers.

Scores distribution coverage out of 10, saves admin-only listing URLs, and
suggests missing platforms with realistic benefit copy.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import re
import urllib.parse
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from typing import Literal

import httpx
import structlog

from app.services.book_url_scraper import (
    PLATFORM_ICONS,
    _browser_headers,
    _normalize_isbn,
)
from app.services.catalog_hub import (
    build_catalog_snapshot,
    discover_platform_hit,
    normalize_source_platform,
)
from app.services.isbn_catalog import (
    classify_retailer_url,
    extract_amazon_asin,
    is_confirmed_isbn13,
    sanitize_platform_url,
)

logger = structlog.get_logger(__name__)

_SCAN_TIMEOUT = 10.0
_MAX_CONCURRENT = 6
_SCAN_CACHE_PREFIX = "platform_scan:v10:"
_SCAN_CACHE_TTL = 21_600  # 6h

Status = Literal["verified", "likely", "not_found", "checking", "error", "skipped"]

VERIFIED_THRESHOLD = 0.85
LIKELY_THRESHOLD = 0.70

PLATFORM_BENEFITS: dict[str, str] = {
    "amazon": "Largest US ebook and print discovery channel",
    "barnes_noble": "US brick-and-mortar and Nook readers",
    "goodreads": "Social discovery and reader reviews",
    "google_books": "Google Books catalog and preview reach",
    "apple_books": "Reach readers on iPhone, iPad, and Mac",
    "kobo": "Strong global indie and Rakuten audience",
    "thriftbooks": "Secondary-market demand signal",
    "abebooks": "Used, rare, and long-tail collector buyers",
    "bookshop": "Supports indie bookstores with affiliate margin",
    "open_library": "Free open catalog and lending discovery",
}

SCORED_PLATFORMS: list[dict] = [
    {"platform_id": "amazon", "display_name": "Amazon", "weight": 2.0},
    {"platform_id": "barnes_noble", "display_name": "Barnes & Noble", "weight": 1.0},
    {"platform_id": "goodreads", "display_name": "Goodreads", "weight": 1.0},
    {"platform_id": "google_books", "display_name": "Google Books", "weight": 1.2},
    {"platform_id": "apple_books", "display_name": "Apple Books", "weight": 1.2},
    {"platform_id": "kobo", "display_name": "Kobo", "weight": 1.0},
    {"platform_id": "thriftbooks", "display_name": "ThriftBooks", "weight": 0.8},
    {"platform_id": "abebooks", "display_name": "AbeBooks", "weight": 0.6},
    {"platform_id": "bookshop", "display_name": "Bookshop.org", "weight": 0.8},
    {"platform_id": "open_library", "display_name": "Open Library", "weight": 0.8},
]


@dataclass
class PlatformListing:
    """Single platform scan result."""

    platform_id: str
    display_name: str
    status: Status
    confidence: float
    listing_url: str | None
    benefit: str
    weight: float
    icon: str = ""
    price: str | None = None
    price_currency: str | None = None


@dataclass
class PlatformPresenceResult:
    """Full scan output for a book."""

    score: float
    listed_count: int
    scanned_at: str
    source_isbn: str | None
    platforms: list[PlatformListing] = field(default_factory=list)
    recommendations: list[dict] = field(default_factory=list)
    distribution_note: str | None = None


def parse_goodreads_book_ref(url: str | None) -> tuple[str, str] | None:
    """Parse Goodreads book show URL into (book_id, book_path)."""
    if not url:
        return None
    m = re.search(
        r"goodreads\.com/book/show/(\d+)(?:[/-]([a-z0-9-]+))?",
        url,
        re.IGNORECASE,
    )
    if not m:
        return None
    book_id = m.group(1)
    slug = (m.group(2) or "").strip("-")
    path = f"/book/show/{book_id}-{slug}" if slug else f"/book/show/{book_id}"
    return book_id, path


def infer_distribution_note(
    platforms: list[PlatformListing],
    source_isbn: str | None = None,
) -> str | None:
    """Explain Kindle-only / narrow distribution so authors aren't misled."""
    from app.services.isbn_catalog import is_confirmed_isbn13

    listed = {p.platform_id for p in platforms if p.status == "verified"}
    if not listed:
        return None
    if is_confirmed_isbn13(source_isbn):
        return None
    if listed <= {"amazon", "goodreads"} and "amazon" in listed:
        return (
            "This title appears to be Amazon Kindle / KDP only. "
            "Goodreads buy buttons are affiliate search links — not verified listings on "
            "Barnes & Noble, Kobo, or Apple Books."
        )
    return None


def scan_cache_key(url: str) -> str:
    """Redis key for caching scan results by source URL."""
    digest = hashlib.sha256(url.strip().encode()).hexdigest()[:20]
    return f"{_SCAN_CACHE_PREFIX}{digest}"


def search_title(raw: str) -> str:
    """Strip series/edition suffixes retailers omit from listing titles.

    Goodreads and similar sources often append series info like
    ``Her Deadly Homecoming (Carolina McKay, #1)`` — store pages use
    ``Her Deadly Homecoming`` only.
    """
    title = re.sub(r"\s+", " ", (raw or "").strip())
    if not title:
        return ""

    prev = None
    while prev != title:
        prev = title
        title = re.sub(r"\s*\([^)]+\)\s*$", "", title).strip()
        title = re.sub(r"\s*\[[^\]]+\]\s*$", "", title).strip()

    title = re.sub(
        r"\s*[:\-–—]\s*(Book|Bk\.?|Volume|Vol\.?)\s+\d+\s*$",
        "",
        title,
        flags=re.IGNORECASE,
    ).strip()
    title = re.sub(r"\s+#\d+\s*$", "", title).strip()

    title = _strip_marketing_subtitle(title)

    return title or (raw or "").strip()


def _strip_marketing_subtitle(title: str) -> str:
    """Drop Amazon-style ': The gripping…' taglines that break catalog search."""
    if not title:
        return ""
    if ":" in title:
        head, tail = title.split(":", 1)
        tail = tail.strip()
        if (
            len(tail) >= 28
            or re.match(r"^(the|a|an)\s", tail, re.IGNORECASE)
        ):
            return head.strip()
    for sep in (" – ", " — ", " - "):
        if sep in title:
            head, tail = title.split(sep, 1)
            if len(tail.strip()) >= 28:
                return head.strip()
    return title


def _title_tokens(text: str) -> list[str]:
    """Normalize title into meaningful tokens for fuzzy matching."""
    text = search_title(text)
    if not text:
        return []
    cleaned = re.sub(r"[^\w\s]", " ", text.lower())
    stop = {"the", "a", "an", "and", "or", "of", "in", "on", "at", "to", "for"}
    return [t for t in cleaned.split() if len(t) > 2 and t not in stop]


def _author_tokens(author: str) -> list[str]:
    """Extract surname-weighted author tokens."""
    if not author:
        return []
    parts = re.split(r"[,;&]", author.lower())
    tokens: list[str] = []
    for part in parts:
        words = [w for w in re.sub(r"[^\w\s]", " ", part).split() if len(w) > 2]
        if words:
            tokens.append(words[-1])
            tokens.extend(words[:-1])
    return list(dict.fromkeys(tokens))


def title_match_score(
    title: str,
    found_title: str,
    author: str = "",
    found_author: str = "",
) -> float:
    """Compute 0.0–1.0 overlap between expected and discovered metadata."""
    title = search_title(title)
    found_title = search_title(found_title)
    title_tokens = _title_tokens(title)
    if not title_tokens:
        return 0.0
    found_lower = found_title.lower()
    matched = sum(1 for t in title_tokens if t in found_lower)
    score = matched / len(title_tokens)

    if author:
        auth_tokens = _author_tokens(author)
        if auth_tokens:
            found_auth = found_author.lower()
            auth_hit = sum(1 for t in auth_tokens if t in found_auth)
            score = score * 0.65 + (auth_hit / len(auth_tokens)) * 0.35
    return min(score, 1.0)


def confidence_to_status(confidence: float, isbn_match: bool = False) -> Status:
    """Binary presence: verified (listed) or not_found."""
    if isbn_match or confidence >= VERIFIED_THRESHOLD:
        return "verified"
    return "not_found"


def compute_score(platforms: list[PlatformListing]) -> float:
    """Count verified platforms out of 10 (same as listed_count)."""
    return float(sum(1 for p in platforms if p.status == "verified"))


def build_recommendations(platforms: list[PlatformListing]) -> list[dict]:
    """Missing platforms sorted by weight with benefit copy."""
    missing = [
        p for p in platforms if p.status in ("not_found", "error", "skipped")
    ]
    missing.sort(key=lambda p: p.weight, reverse=True)
    return [
        {
            "platform_id": p.platform_id,
            "display_name": p.display_name,
            "weight": p.weight,
            "benefit": p.benefit,
            "icon": p.icon,
        }
        for p in missing
    ]


def _search_query(title: str, author: str, isbn: str | None) -> str:
    """Build a retailer search query from metadata."""
    if isbn:
        return isbn
    core = search_title(title) or title.strip()
    parts = [core]
    if author.strip():
        parts.append(author.strip())
    return " ".join(parts)


@dataclass
class CatalogContext:
    """Cross-platform hints from Goodreads autocomplete + buy links."""

    resolved_author: str = ""
    goodreads_id: str | None = None
    goodreads_url: str | None = None
    retailer_links: dict[str, str] = field(default_factory=dict)


# Goodreads BookLink name → platform_id
_BOOKLINK_PLATFORM_KEYS: list[tuple[str, tuple[str, ...]]] = [
    ("amazon", ("amazon",)),
    ("barnes_noble", ("barnes", "noble")),
    ("kobo", ("kobo",)),
    ("google_books", ("google books", "google play")),
    ("google_play", ("google play",)),
    ("apple_books", ("apple books", "ibooks")),
    ("bookshop", ("bookshop",)),
    ("thriftbooks", ("thriftbooks",)),
    ("abebooks", ("abebooks",)),
    ("walmart", ("walmart",)),
    ("goodreads", ("goodreads",)),
]


def _unescape_js_string(value: str) -> str:
    """Decode Goodreads embedded JSON string escapes."""
    return (
        value.replace("\\u0026", "&")
        .replace("\\u0027", "'")
        .replace("\\/", "/")
        .replace('\\"', '"')
    )


def parse_goodreads_book_links(html: str) -> dict[str, str]:
    """Extract retailer buy/search URLs from a Goodreads book page."""
    if not html:
        return {}
    links: dict[str, str] = {}
    for match in re.finditer(
        r'BookLink","name":"((?:[^"\\]|\\.)*)","url":"((?:[^"\\]|\\.)*)"',
        html,
    ):
        name = _unescape_js_string(match.group(1)).lower()
        url = _unescape_js_string(match.group(2))
        if not url.startswith("http"):
            continue
        for platform_id, keys in _BOOKLINK_PLATFORM_KEYS:
            if platform_id in links:
                continue
            if any(key in name for key in keys):
                links[platform_id] = url
                break
    return links


async def _fetch_goodreads_book_page(
    client: httpx.AsyncClient,
    book_id: str,
    book_path: str = "",
) -> str:
    """Fetch Goodreads book HTML with slug/referer fallbacks for datacenter IPs."""
    if not book_id:
        return ""

    candidates: list[str] = []
    if book_path:
        candidates.append(f"https://www.goodreads.com{book_path}")
    candidates.append(f"https://www.goodreads.com/book/show/{book_id}")
    candidates.append(f"https://m.goodreads.com/book/show/{book_id}")

    retryable = {403, 429, 503, 202, 520, 521, 522, 523, 524}
    referer = "https://www.goodreads.com/"

    for url in candidates:
        for attempt in range(3):
            try:
                if attempt > 0:
                    await asyncio.sleep(1.5 * attempt)
                headers = _browser_headers(referer)
                headers["Referer"] = referer
                resp = await client.get(url, headers=headers, timeout=_SCAN_TIMEOUT)
                body = resp.text or ""
                if body and "BookLink" in body:
                    return body
                if resp.status_code == 200 and body:
                    return body
                if resp.status_code in retryable:
                    logger.debug(
                        "Goodreads page blocked, retrying",
                        url=url[:60],
                        status=resp.status_code,
                        attempt=attempt + 1,
                    )
                    continue
                break
            except (httpx.TimeoutException, httpx.ConnectError) as exc:
                logger.debug("Goodreads page fetch error", url=url[:60], error=str(exc))
                if attempt < 2:
                    await asyncio.sleep(1.5 * (attempt + 1))

    logger.warning("Goodreads book page fetch failed", book_id=book_id)
    return ""


def merge_retailer_links(
    base: dict[str, str],
    extra: dict[str, str],
) -> dict[str, str]:
    """Merge retailer URLs; first source wins."""
    merged = dict(base)
    for pid, url in extra.items():
        if pid not in merged and url:
            merged[pid] = url
    return merged


async def _goodreads_autocomplete_by_isbn(
    client: httpx.AsyncClient,
    isbn13: str,
) -> dict | None:
    """Resolve Goodreads book via ISBN when title autocomplete fails."""
    if not is_confirmed_isbn13(isbn13):
        return None
    try:
        resp = await client.get(
            "https://www.goodreads.com/book/auto_complete",
            params={"format": "json", "q": isbn13},
            headers={
                "Accept": "application/json",
                "User-Agent": _browser_headers()["User-Agent"],
            },
            timeout=_SCAN_TIMEOUT,
        )
        if resp.status_code != 200:
            return None
        for item in resp.json()[:5]:
            book_id = str(item.get("bookId") or "")
            book_path = item.get("bookUrl") or ""
            if book_id and book_path:
                return {
                    "book_id": book_id,
                    "book_path": book_path,
                    "author_name": (item.get("author") or {}).get("name") or "",
                    "title_bare": item.get("bookTitleBare") or item.get("title") or "",
                }
    except Exception as exc:
        logger.debug("Goodreads ISBN autocomplete failed", error=str(exc))
    return None


async def _goodreads_autocomplete_best(
    client: httpx.AsyncClient,
    title: str,
    author: str,
) -> dict | None:
    """Match title via Goodreads public autocomplete JSON API."""
    core = search_title(title) or title
    if not core:
        return None
    query = f"{core} {author}".strip() if author.strip() else core
    try:
        resp = await client.get(
            "https://www.goodreads.com/book/auto_complete",
            params={"format": "json", "q": query},
            headers={
                "Accept": "application/json",
                "User-Agent": _browser_headers()["User-Agent"],
            },
            timeout=_SCAN_TIMEOUT,
        )
        if resp.status_code != 200:
            return None
        for item in resp.json()[:10]:
            bare = item.get("bookTitleBare") or item.get("title") or ""
            auth = (item.get("author") or {}).get("name") or ""
            if title_match_score(core, bare, author, auth) >= LIKELY_THRESHOLD:
                book_path = item.get("bookUrl") or ""
                return {
                    "book_id": str(item.get("bookId") or ""),
                    "book_path": book_path,
                    "author_name": auth,
                    "title_bare": bare,
                }
    except Exception as exc:
        logger.debug("Goodreads autocomplete failed", error=str(exc))
    return None


async def _build_catalog_context(
    client: httpx.AsyncClient,
    title: str,
    author: str,
    isbn: str | None = None,
) -> CatalogContext:
    """Resolve author and retailer links via Goodreads (hub for indie books)."""
    from app.services.isbn_catalog import (
        bookshop_url_from_isbn,
        resolve_isbn_open_library,
    )

    ctx = CatalogContext(resolved_author=author.strip())
    hit = await _goodreads_autocomplete_best(client, title, author)
    if not hit:
        return ctx

    if not ctx.resolved_author and hit.get("author_name"):
        ctx.resolved_author = hit["author_name"]
    book_id = hit.get("book_id") or ""
    book_path = hit.get("book_path") or ""
    if book_id:
        ctx.goodreads_id = book_id
        ctx.goodreads_url = f"https://www.goodreads.com{book_path}" if book_path else None
        page_html = await _fetch_goodreads_book_page(client, book_id, book_path)
        if page_html:
            gr_links = {
                pid: url
                for pid, raw in parse_goodreads_book_links(page_html).items()
                if (url := sanitize_platform_url(raw, pid))
            }
            ctx.retailer_links = merge_retailer_links(ctx.retailer_links, gr_links)
    if ctx.goodreads_url:
        ctx.retailer_links = merge_retailer_links(
            ctx.retailer_links,
            {"goodreads": ctx.goodreads_url},
        )

    scan_author = ctx.resolved_author or author
    resolved_isbn = _normalize_isbn(isbn) if isbn else None
    if not resolved_isbn:
        resolved_isbn = await resolve_isbn_open_library(client, title, scan_author)

    if resolved_isbn and "bookshop" not in ctx.retailer_links:
        bs_url = bookshop_url_from_isbn(resolved_isbn, title)
        if bs_url:
            ctx.retailer_links["bookshop"] = bs_url

    return ctx


def _core_phrase(title: str) -> str:
    """Lowercase core title phrase for substring checks in HTML."""
    return re.sub(r"\s+", " ", (search_title(title) or title).lower()).strip()


def _build_search_url(platform_id: str, title: str, author: str, isbn: str | None) -> str:
    """Construct canonical search URL per retailer."""
    q = urllib.parse.quote(_search_query(title, author, isbn))
    urls = {
        "amazon": f"https://www.amazon.com/s?k={q}",
        "barnes_noble": f"https://www.barnesandnoble.com/s/{q}",
        "kobo": f"https://www.kobo.com/us/en/search?query={q}",
        "walmart": f"https://www.walmart.com/search?q={q}",
        "bookshop": f"https://bookshop.org/search?keywords={q}",
        "thriftbooks": f"https://www.thriftbooks.com/browse/?b.search={q}",
        "abebooks": f"https://www.abebooks.com/servlet/SearchResults?kn={q}",
        "goodreads": f"https://www.goodreads.com/search?q={q}",
        "google_play": f"https://play.google.com/store/search?q={q}&c=books",
    }
    return urls.get(platform_id, "")


def _listing_from_config(cfg: dict, status: Status = "checking") -> PlatformListing:
    """Seed a PlatformListing from platform config."""
    pid = cfg["platform_id"]
    return PlatformListing(
        platform_id=pid,
        display_name=cfg["display_name"],
        status=status,
        confidence=0.0,
        listing_url=None,
        benefit=PLATFORM_BENEFITS.get(pid, ""),
        weight=cfg["weight"],
        icon=PLATFORM_ICONS.get(pid, "🔗"),
    )


def platform_presence_to_dict(result: PlatformPresenceResult) -> dict:
    """Serialize scan result for API/Redis."""
    return asdict(result)


def platform_presence_from_dict(data: dict) -> PlatformPresenceResult:
    """Deserialize scan result from Redis."""
    platforms = [PlatformListing(**p) for p in data.get("platforms", [])]
    return PlatformPresenceResult(
        score=data.get("score", 0.0),
        listed_count=data.get("listed_count", 0),
        scanned_at=data.get("scanned_at", ""),
        source_isbn=data.get("source_isbn"),
        platforms=platforms,
        recommendations=data.get("recommendations", []),
        distribution_note=data.get("distribution_note"),
    )


def presence_dict_from_db_listings(
    listings: list,
    source_isbn: str | None = None,
) -> dict:
    """Rebuild platform_presence dict from persisted BookPlatformListing rows."""
    cfg_by_id = {p["platform_id"]: p for p in SCORED_PLATFORMS}
    platforms: list[PlatformListing] = []
    for row in listings:
        pid = row.platform_id
        cfg = cfg_by_id.get(pid, {})
        platforms.append(PlatformListing(
            platform_id=pid,
            display_name=cfg.get("display_name", pid.replace("_", " ").title()),
            status=row.status,
            confidence=row.confidence,
            listing_url=row.listing_url,
            benefit=PLATFORM_BENEFITS.get(pid, ""),
            weight=cfg.get("weight", 0.5),
            icon=PLATFORM_ICONS.get(pid, "🔗"),
            price=row.list_price,
            price_currency=row.price_currency,
        ))
    listed_count = sum(1 for p in platforms if p.status == "verified")
    scanned_at = ""
    if listings and listings[0].last_scanned_at:
        scanned_at = listings[0].last_scanned_at.isoformat()
    result = PlatformPresenceResult(
        score=compute_score(platforms),
        listed_count=listed_count,
        scanned_at=scanned_at,
        source_isbn=source_isbn,
        platforms=platforms,
        recommendations=build_recommendations(platforms),
    )
    return platform_presence_to_dict(result)


def count_missing_listings(listings: list) -> int:
    """Count platforms with not_found/skipped/error status."""
    return sum(
        1 for row in listings
        if row.status in ("not_found", "skipped", "error")
    )


async def save_scan_to_redis(redis, url: str, result: PlatformPresenceResult) -> None:
    """Cache scan result for import confirm."""
    if redis is None:
        return
    try:
        await redis.setex(
            scan_cache_key(url),
            _SCAN_CACHE_TTL,
            json.dumps(platform_presence_to_dict(result)),
        )
    except Exception as exc:
        logger.debug("Platform scan cache write failed", error=str(exc))


async def load_scan_from_redis(redis, url: str) -> PlatformPresenceResult | None:
    """Load cached scan result if available."""
    if redis is None:
        return None
    try:
        raw = await redis.get(scan_cache_key(url))
        if raw:
            return platform_presence_from_dict(json.loads(raw))
    except Exception as exc:
        logger.debug("Platform scan cache read failed", error=str(exc))
    return None


async def _load_goodreads_hub(
    client: httpx.AsyncClient,
    title: str,
    author: str,
    source_url: str | None = None,
    isbn: str | None = None,
) -> tuple[str, str | None, dict[str, str]]:
    """Goodreads book page + buy links (prefer explicit source URL when provided)."""
    resolved_author = author.strip()
    gr_url: str | None = None
    links: dict[str, str] = {}

    ref = parse_goodreads_book_ref(source_url)
    if ref:
        book_id, book_path = ref
        gr_url = f"https://www.goodreads.com{book_path}"
        links["goodreads"] = gr_url
        page_html = await _fetch_goodreads_book_page(client, book_id, book_path)
        if page_html:
            for pid, raw in parse_goodreads_book_links(page_html).items():
                if pid == "google_play":
                    if "books.google.com" not in raw and "/store/books/details" not in raw:
                        continue
                    pid = "google_books"
                clean = sanitize_platform_url(raw, pid)
                if clean:
                    links[pid] = clean
        return resolved_author, gr_url, links

    hit = await _goodreads_autocomplete_best(client, title, author)
    if not hit and isbn:
        hit = await _goodreads_autocomplete_by_isbn(client, isbn)
    if not hit:
        return resolved_author, None, links

    if hit.get("author_name"):
        resolved_author = hit["author_name"]
    book_id = hit.get("book_id") or ""
    book_path = hit.get("book_path") or ""
    if book_path:
        gr_url = f"https://www.goodreads.com{book_path}"
        links["goodreads"] = gr_url
    if book_id:
        page_html = await _fetch_goodreads_book_page(client, book_id, book_path)
        if page_html:
            for pid, raw in parse_goodreads_book_links(page_html).items():
                if pid == "google_play":
                    if "books.google.com" not in raw and "/store/books/details" not in raw:
                        continue
                    pid = "google_books"
                clean = sanitize_platform_url(raw, pid)
                if clean:
                    links[pid] = clean
    return resolved_author, gr_url, links


def _extract_asin(isbn: str | None) -> str | None:
    """Pull Kindle ASIN from identifier field when present."""
    if not isbn:
        return None
    raw = isbn.strip().upper()
    if raw.startswith("B0") and len(raw) == 10:
        return raw
    return extract_amazon_asin(isbn)


def _finalize_listing_url(listing: PlatformListing) -> None:
    """Normalize URL — search pages are not listings; clear them."""
    pid = listing.platform_id
    if not listing.listing_url:
        listing.status = "not_found"
        listing.confidence = 0.0
        return

    clean = sanitize_platform_url(listing.listing_url, pid)
    if not clean or classify_retailer_url(clean, pid) != "product":
        listing.listing_url = None
        listing.status = "not_found"
        listing.confidence = 0.0
        return

    listing.listing_url = clean
    if listing.confidence >= VERIFIED_THRESHOLD:
        listing.status = "verified"
    else:
        listing.status = "not_found"
        listing.listing_url = None
        listing.confidence = 0.0


async def _scan_one_platform(
    cfg: dict,
    snap,
    source_platform: str | None,
    source_url: str | None,
    buy_url: str | None,
) -> PlatformListing:
    """Run API-first discovery for a single platform."""
    listing = _listing_from_config(cfg, status="checking")
    pid = cfg["platform_id"]
    hit = discover_platform_hit(
        pid,
        snap,
        source_platform=source_platform,
        source_url=source_url,
        buy_url=buy_url,
    )
    if hit.listing_url and (
        hit.isbn_match or hit.confidence >= VERIFIED_THRESHOLD
    ):
        listing.confidence = round(hit.confidence, 2)
        listing.status = confidence_to_status(hit.confidence, hit.isbn_match)
        listing.listing_url = hit.listing_url
        _finalize_listing_url(listing)
    else:
        listing.status = "not_found"
        listing.confidence = 0.0
        listing.listing_url = None
    return listing


async def _enrich_listing_prices(
    client: httpx.AsyncClient,
    platforms: list[PlatformListing],
    *,
    isbn13: str | None = None,
    title: str | None = None,
    author: str | None = None,
    db=None,
    redis=None,
) -> None:
    """Fetch list prices for verified listings (best-effort, parallel)."""
    from app.services.platform_api_registry import load_all_credentials
    from app.services.platform_pricing import PriceContext, fetch_listing_price

    ctx = PriceContext(isbn13=isbn13, title=title, author=author)
    credentials = await load_all_credentials(db=db, redis=redis)
    targets = [
        p for p in platforms
        if p.status == "verified" and p.listing_url
    ]
    if not targets:
        return

    async def _one(listing: PlatformListing) -> None:
        try:
            hit = await fetch_listing_price(
                client,
                listing.platform_id,
                listing.listing_url or "",
                ctx,
                credential=credentials.get(listing.platform_id),
            )
            if hit:
                listing.price = hit.formatted
                listing.price_currency = hit.currency
        except Exception as exc:
            logger.debug(
                "Platform price enrichment failed",
                platform=listing.platform_id,
                error=str(exc)[:120],
            )

    await asyncio.gather(*[_one(p) for p in targets], return_exceptions=True)

    gr = next(
        (p for p in platforms if p.platform_id == "goodreads" and p.status == "verified"),
        None,
    )
    if gr and not gr.price:
        retail = [
            p for p in platforms
            if p.platform_id not in ("goodreads", "open_library")
            and p.status == "verified" and p.price and p.price != "Free"
        ]
        if retail:
            amazon = next((p for p in retail if p.platform_id == "amazon"), None)
            ref = amazon or retail[0]
            gr.price = f"from {ref.price}"
            gr.price_currency = ref.price_currency


async def scan_platform_presence(
    *,
    title: str,
    author: str = "",
    isbn: str | None = None,
    source_platform: str | None = None,
    source_url: str | None = None,
    buy_url: str | None = None,
    client: httpx.AsyncClient | None = None,
    db=None,
    redis=None,
) -> PlatformPresenceResult:
    """Scan top-10 platforms and return scored presence result."""
    raw_isbn = _normalize_isbn(isbn) if isbn else None
    norm_isbn = raw_isbn if is_confirmed_isbn13(raw_isbn) else None
    scanned_at = datetime.now(timezone.utc).isoformat()
    core_title = search_title(title) or title.strip()

    if not core_title:
        empty = [_listing_from_config(c, status="skipped") for c in SCORED_PLATFORMS]
        return PlatformPresenceResult(
            score=0.0,
            listed_count=0,
            scanned_at=scanned_at,
            source_isbn=norm_isbn,
            platforms=empty,
            recommendations=build_recommendations(empty),
        )

    sem = asyncio.Semaphore(_MAX_CONCURRENT)

    async def _run_scan(client: httpx.AsyncClient) -> tuple[list[PlatformListing], str | None]:
        scan_isbn = norm_isbn

        async def _gr_hub() -> tuple[str, str | None, dict[str, str]]:
            return await _load_goodreads_hub(
                client, core_title, author,
                source_url=source_url, isbn=scan_isbn,
            )

        async def _base_snapshot() -> "CatalogSnapshot":
            from app.services.catalog_hub import build_catalog_snapshot
            resolved_asin = _extract_asin(scan_isbn or isbn)
            return await build_catalog_snapshot(
                client,
                title=core_title,
                author=author,
                isbn=scan_isbn,
                asin=resolved_asin,
                goodreads_links={},
                goodreads_url=None,
                goodreads_confidence=0.0,
                resolved_author=author,
                skip_enrich=True,
            )

        (resolved_author, gr_url, gr_links), snap = await asyncio.gather(
            _gr_hub(), _base_snapshot(),
        )

        if scan_isbn and is_confirmed_isbn13(scan_isbn):
            snap.isbn13 = scan_isbn
        elif snap.isbn13 and is_confirmed_isbn13(snap.isbn13):
            scan_isbn = snap.isbn13

        if gr_url:
            snap.goodreads_url = gr_url
            snap.goodreads_confidence = 0.95
        if resolved_author:
            snap.author = resolved_author
        snap.retailer_links = merge_retailer_links(gr_links, snap.retailer_links)

        from app.services.isbn_catalog import enrich_retailer_links_from_isbn
        snap.retailer_links = await enrich_retailer_links_from_isbn(
            client,
            snap.retailer_links,
            isbn13=snap.isbn13 or scan_isbn,
            title=core_title,
            author=snap.author or author,
        )

        if not snap.asin:
            resolved_asin = _extract_asin(scan_isbn or isbn)
            if not resolved_asin and gr_links.get("amazon"):
                resolved_asin = extract_amazon_asin(gr_links["amazon"])
            if resolved_asin:
                snap.asin = resolved_asin

        src = normalize_source_platform(source_platform)

        async def _guarded(cfg: dict) -> PlatformListing:
            async with sem:
                return await _scan_one_platform(
                    cfg, snap, src, source_url, buy_url,
                )

        return list(await asyncio.gather(*[_guarded(c) for c in SCORED_PLATFORMS])), scan_isbn

    if client is not None:
        platforms, resolved_scan_isbn = await _run_scan(client)
        await _enrich_listing_prices(
            client, platforms,
            isbn13=resolved_scan_isbn or norm_isbn,
            title=core_title,
            author=author,
            db=db,
            redis=redis,
        )
    else:
        async with httpx.AsyncClient(follow_redirects=True) as shared:
            platforms, resolved_scan_isbn = await _run_scan(shared)
            await _enrich_listing_prices(
                shared, platforms,
                isbn13=resolved_scan_isbn or norm_isbn,
                title=core_title,
                author=author,
                db=db,
                redis=redis,
            )

    listed_count = sum(1 for p in platforms if p.status == "verified")
    score = compute_score(platforms)
    recommendations = build_recommendations(platforms)

    return PlatformPresenceResult(
        score=score,
        listed_count=listed_count,
        scanned_at=scanned_at,
        source_isbn=resolved_scan_isbn or norm_isbn,
        platforms=platforms,
        recommendations=recommendations,
        distribution_note=infer_distribution_note(
            platforms, resolved_scan_isbn or norm_isbn,
        ),
    )