feat: canonicalize share/short book URLs + deep multi-format price enrichment

Resolve a.co/amzn.to/bn.com/itunes families before scrape; wire Thrift/Abe/Bookshop/Kobo deep parsers into presence enrichment; Author UI shows fetched-only chips and Resolved toast.

Co-authored-by: Cursor <cursoragent@cursor.com>

.agent/BDD_SPECS.md

@@ -371,6 +371,26 @@ Feature: Book URL Import
     And fall back to catalog APIs and any prior cached metadata
     And the response should include a human-readable warning instead of a blank 0% match
 
+  Scenario: Amazon Share short link resolves like a browser
+    Given an author pastes https://a.co/d/04YALfQb
+    When the author clicks "Fetch Details"
+    Then the platform should be amazon (never unknown)
+    And the URL should canonicalize to https://www.amazon.com/dp/{ASIN}
+    And title, description, about_author, and format_prices should match a clean /dp/ import
+    And a soft toast should note Resolved a.co → the canonical product URL
+
+  Scenario: Barnes and Noble id-only and short-host URLs are supported
+    Given an author pastes barnesandnoble.com/w/{id} without a slug or a bn.com short host
+    When the author clicks "Fetch Details"
+    Then the platform should be barnes_noble
+    And identifiers and multi-format prices should resolve when the product page exposes them
+
+  Scenario: Legacy iTunes Apple Books URL rewrites to books.apple.com
+    Given an author pastes itunes.apple.com/{cc}/book/.../id{N}
+    When the author clicks "Fetch Details"
+    Then the platform should be apple_books
+    And the canonical URL should use books.apple.com
+
 
 Feature: Platform Presence Scan
 
@@ -444,11 +464,27 @@ Feature: Platform List Pricing
     And discovery platforms such as goodreads and open_library should have empty format_prices
     And the single list_price on each row should equal primary_list_price of the format map
 
+  Scenario: Author URL preview enriches every discovered retailer with credentials
+    Given an author pastes one supported book URL into Upload Book
+    And configured platform credentials are stored in the database
+    When the author clicks Fetch Details
+    Then the presence scan receives the database and Redis dependencies
+    And every verified discovered listing URL is checked for available format prices
+    And the preview shows only fetched Hardcover, Paperback, eBook, or Audiobook labels
+    And unfetched formats are omitted rather than invented
+
   Scenario: Richer existing format_prices are not wiped by a thinner scan
     Given a book with an Amazon listing that has format_prices Hardcover $35 and eBook $14.99
     When the same book is re-scanned and the scan only provides a single list_price of $35
     Then the Amazon format_prices should still contain both Hardcover and eBook entries
 
+  Scenario: Discovered listings use deep multi-format parsers
+    Given a presence scan verifies Bookshop, ThriftBooks, AbeBooks, or Kobo listing URLs
+    When format prices are enriched
+    Then the same deep extractors used on source import should populate format_prices
+    And only formats present on the retailer page are stored
+    And missing formats are omitted rather than invented
+
 ## Enterprise Features (2026-07)
 
 Feature: Webhooks

.agent/BOOK_URL_IMPORT_RULES.md

@@ -36,12 +36,28 @@ Code mirror: `LOCKED_URL_IMPORT_PLATFORMS` in `app/services/book_url_scraper.py`
 
 ## Pipeline (in order — do not skip steps)
 
+### Step 0 — Canonicalize URL (before platform detect)
+Every locked platform has multiple valid URL families for the same book
+(stable product, affiliate/tracking, Share/short/wrapper). Share/short hosts
+are first-class members of that platform — never `"unknown"`.
+
+| Rule | Behavior |
+|------|----------|
+| R-URL-001 | Classify by **host-family map** (all 9 platforms + aliases) before scrape |
+| R-URL-002 | Share/short/wrapper hosts (`a.co`, `amzn.to`, `bn.com`, Rakuten wrappers, etc.) MUST follow HTTP redirects like a browser, then rewrite |
+| R-URL-003 | After resolve, rewrite to the platform’s stable product URL (Amazon `/dp/{ASIN}`, BN `/w/…`, Apple `books.apple.com/…/id{N}`, …) |
+| R-URL-004 | Strip tracking params; never drop `ean` / ISBN / product ids |
+| R-URL-005 | Canonicalize runs for pasted import URLs **and** discovered presence listing URLs |
+
+Code: `app/services/url_canonicalizer.py` → `canonicalize_book_url`.
+
 ### Step 1 — Parse the URL (zero HTTP)
 | Extract | Pattern |
 |---------|---------|
-| ASIN | `/dp/{10-char}` or `/gp/product/{10-char}` |
+| ASIN | `/dp/{10-char}` or `/gp/product/{10-char}` (also `/gp/aw/d/`) |
 | ISBN | digits in path (AbeBooks, Bookshop, etc.) |
-| Platform | domain regex |
+| BN id | `/w/{slug}/{id}` **or** `/w/{id}` |
+| Platform | host-family map, then domain regex |
 
 Normalize URL: strip `?tag=`, `?ref_=`, canonicalize Amazon to `/dp/{ASIN}`.
 
@@ -246,6 +262,14 @@ lives on **`book_platform_listings`** (one row per book × platform).
 3. Discovery platforms (Goodreads, Open Library) store empty `format_prices`.
 4. Used-print (ThriftBooks, AbeBooks) keep print bindings only; digital
    (Apple, Kobo) keep eBook/Audiobook only.
+5. Author **Fetch Details** MUST pass DB + Redis into the presence scan so
+   configured retailer/ScrapingBee credentials are available before the scan
+   is cached. Every verified discovered listing URL is enriched independently
+   via the same deep format parsers used on source import (R-PP-015).
+6. The Author preview shows only successfully fetched format-price labels.
+   Missing formats are omitted; no empty label or invented amount is stored.
+   When a share/short URL was resolved, show a soft toast:
+   `Resolved {host} → {canonical}`.
 
 **Admin:** Book list → Platforms panel edits URL / ISBN / ASIN / format prices
 via `PUT …/books/{id}/platforms/{platform_id}`.
@@ -265,3 +289,22 @@ And   genre contains "Kidnapping" or "Crime Fiction"
 And   isbn field = "B0DHWS3P6T" (ASIN — no ISBN-13 on page)
 And   confidence = 100%
 ```
+
+## BDD acceptance (URL families)
+```
+Given Amazon Share URL https://a.co/d/{code}
+When  author clicks Fetch Details
+Then  platform = amazon (never unknown)
+And   canonical URL is https://www.amazon.com/dp/{ASIN}
+And   title / description / about_author / format_prices match the clean /dp/ import
+
+Given Barnes & Noble URL …/w/{id} (no slug) or bn.com short host
+When  author clicks Fetch Details
+Then  platform = barnes_noble
+And   identifiers and multi-format prices resolve when the page exposes them
+
+Given itunes.apple.com/…/id{N}
+When  author clicks Fetch Details
+Then  platform = apple_books
+And   URL rewrites to books.apple.com
+```

.agent/FILE_MAP.md

@@ -109,6 +109,7 @@
 | `app/superadmin/routers/google_oauth_config.py` | SuperAdmin Google OAuth client config |
 | `app/services/book_service.py` | Admin book CRUD, covers, URL import, platform presence |
 | `app/services/platform_catalog.py` | Canonical format prices + listing view for admin/bot |
+| `app/services/url_canonicalizer.py` | Host-family map + share/short URL canonicalize for 9 platforms |
 | `app/services/price_catalog_service.py` | Grounded visitor price answers + platform_offers payload |
 | `.agent/PRICE_INTELLIGENCE_RULES.md` | Bot price intelligence rules (R-PI-001–006, format-aware grid) |
 | `app/services/admin_analytics_service.py` | Admin dashboard analytics (funnel, heatmap, visitors) |

.agent/PLATFORM_PRICING_RULES.md

@@ -112,11 +112,15 @@ Amazon HTML: US-locale headers + cookie (`i18n-prefs=USD`) to avoid PKR/wrong cu
 - **R-PP-012:** Discovery platforms (goodreads, open_library, google_books) return `format_prices: {}` from enrichment; their single `list_price` is preserved as-is
 - **R-PP-013:** `fetch_listing_price` remains a backward-compatible thin wrapper over `fetch_listing_format_prices` returning only the primary formatted string
 - **R-PP-014:** Apple Books enrichment MUST query iTunes `entity=ebook` and `entity=audiobook` so both digital formats can populate `format_prices` when available
+- **R-PP-015:** Presence enrichment MUST call the same deep format parsers as source import — Amazon, BN, ThriftBooks, AbeBooks, Bookshop, Kobo — after one HTML fetch. Never invent missing format keys.
+- **R-PP-016:** Listing URLs MUST be canonicalized (share/short/affiliate → product URL) before price HTML fetch
+- **R-PP-017:** Scan cache key version bumped to `platform_scan:v13:` when canonicalize + deep enrichment ship
 
 ## 11. File map
 
 | File | Role |
 |------|------|
+| `app/services/url_canonicalizer.py` | Host-family map + `canonicalize_book_url` |
 | `app/services/platform_pricing.py` | Fetch, parse, format; `FormatPriceHit`, `fetch_listing_format_prices` |
 | `app/services/platform_presence.py` | `_enrich_listing_prices()` orchestration; sets `listing.format_prices` |
 | `app/services/platform_api_registry.py` | Credential-tier price fetch |

app/admin/templates/admin.html

@@ -666,6 +666,7 @@
     .ab-pp-fmt-pair { display: inline-flex; align-items: center; gap: 3px; background: var(--bg2, #f3f4f6); border-radius: 4px; padding: 1px 5px; }
     .ab-pp-fmt-name { font-size: 10px; color: var(--muted); }
     .ab-pp-fmt-price { font-size: 12px; font-weight: 600; color: var(--text2); }
+    .ab-pp-price--na { font-size: 11px; font-weight: 500; color: var(--muted); font-style: italic; }
     .ab-found-links {
       display: flex;
       flex-wrap: wrap;
@@ -5983,9 +5984,10 @@ function abRenderPlatformScan(pp) {
             .filter(k => fps[k])
             .map(k => `<span class="ab-pp-fmt-pair"><span class="ab-pp-fmt-name">${esc(k)}</span> <span class="ab-pp-fmt-price">${esc(fps[k])}</span></span>`)
             .join('');
-          return parts ? `<div class="ab-pp-price ab-pp-price--fmts">${parts}</div>` : (p.price ? `<div class="ab-pp-price">${esc(p.price)}</div>` : '');
+          if (parts) return `<div class="ab-pp-price ab-pp-price--fmts">${parts}</div>`;
         }
-        return p.price ? `<div class="ab-pp-price">${esc(p.price)}</div>` : '';
+        if (p.price) return `<div class="ab-pp-price">${esc(p.price)}</div>`;
+        return `<div class="ab-pp-price ab-pp-price--na">Price not available</div>`;
       })();
       const attrs = missing
         ? ` data-pp-platform-id="${esc(p.platform_id)}" data-pp-missing="1" role="button" tabindex="0" aria-label="Not found on ${esc(p.display_name)} — click for publishing help"`
@@ -6142,8 +6144,22 @@ function abRenderPreview(data) {
   document.getElementById('ab-e-bio').value    = data.about_author || '';
 
   if (data.warnings && data.warnings.length) {
-    abShowAlert('warning', '⚠️ Some data was enriched from APIs: ' + data.warnings.join('; '));
-}
+    const resolved = data.warnings.filter(w => String(w).startsWith('Resolved '));
+    const other = data.warnings.filter(w => !String(w).startsWith('Resolved '));
+    if (resolved.length) {
+      abShowAlert('success', resolved.join(' · '));
+    }
+    if (other.length) {
+      abShowAlert('warning', '⚠️ Some data was enriched from APIs: ' + other.join('; '));
+    }
+  }
+  // Prefer canonical product URL in the paste field after share/short resolve.
+  if (data.canonical_url && document.getElementById('ab-url')) {
+    const paste = (document.getElementById('ab-url').value || '').trim();
+    if (paste && data.canonical_url !== paste) {
+      document.getElementById('ab-url').value = data.canonical_url;
+    }
+  }
   abRenderPlatformScan(data.platform_presence);
   abSetState('preview');
 }
@@ -6293,7 +6309,11 @@ async function abLoadSidebar() {
 function abShowAlert(type, msg) {
   const el = document.getElementById('ab-alert');
   if (!el) return;
-  const colors = { error:'var(--red-bg);border:1px solid rgba(239,68,68,.25);color:var(--red)', warning:'var(--yellow-bg);border:1px solid rgba(245,158,11,.25);color:#92400e' };
+  const colors = {
+    error: 'var(--red-bg);border:1px solid rgba(239,68,68,.25);color:var(--red)',
+    warning: 'var(--yellow-bg);border:1px solid rgba(245,158,11,.25);color:#92400e',
+    success: 'rgba(16,185,129,.12);border:1px solid rgba(16,185,129,.3);color:#065f46',
+  };
   el.style.cssText = `display:flex;align-items:flex-start;gap:10px;padding:11px 14px;border-radius:var(--radius-sm);margin-bottom:14px;font-size:12px;background:${colors[type]||colors.error}`;
   el.textContent = msg;
 }

app/services/book_service.py

@@ -361,11 +361,12 @@ class BookService:
                 author=meta.author or "",
                 isbn=meta.isbn13 or meta.isbn or None,
                 source_platform=meta.platform if meta.platform != "unknown" else None,
-                source_url=url,
-                buy_url=meta.buy_url or url,
+                source_url=meta.source_url or url,
+                buy_url=meta.buy_url or meta.source_url or url,
+                db=db,
                 redis=redis,
             )
-            await save_scan_to_redis(redis, url, scan)
+            await save_scan_to_redis(redis, meta.source_url or url, scan)
             platform_presence = platform_presence_to_dict(scan)
 
         platform_display = PLATFORM_DISPLAY.get(
@@ -399,8 +400,9 @@ class BookService:
             "genre": meta.genre,
             "rating": meta.rating,
             "rating_count": meta.rating_count,
-            "buy_url": meta.buy_url or url,
-            "source_url": url,
+            "buy_url": meta.buy_url or meta.source_url or url,
+            "source_url": meta.source_url or url,
+            "canonical_url": meta.source_url or url,
             "confidence": meta.confidence,
             "error": meta.error or None,
             "warnings": meta.warnings or [],

app/services/book_url_scraper.py

@@ -193,11 +193,22 @@ PLATFORM_ICONS: dict[str, str] = {
 }
 
 def detect_platform(url: str) -> str:
-    """Return platform id for a URL. Google Books URLs are not import platforms."""
+    """Return platform id for a URL. Google Books URLs are not import platforms.
+
+    Share/short hosts (a.co, amzn.to, bn.com, itunes.apple.com, …) are
+    classified via the host-family map so they are never mislabeled unknown.
+    """
     low = url.lower()
     # Explicitly reject Google Books / Play Books as a scrape target.
     if re.search(r"books\.google\.|play\.google\.com/store/books", low):
         return "unknown"
+    try:
+        from app.services.url_canonicalizer import host_family_platform
+        family = host_family_platform(url)
+        if family:
+            return family
+    except Exception:
+        pass
     for platform, pattern in PLATFORM_PATTERNS.items():
         if re.search(pattern, low):
             return platform
@@ -1908,10 +1919,10 @@ 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)
+    # Amazon ASIN: /dp/, /gp/product/, or mobile /gp/aw/d/
+    m = re.search(r"/(?:dp|gp/product|gp/aw/d)/([A-Z0-9]{10})", url, re.IGNORECASE)
     if m:
-        ids["asin"] = m.group(1)
+        ids["asin"] = m.group(1).upper()
 
     # Google Books volume ID: id=XXXX in query string
     m = re.search(r"[?&]id=([A-Za-z0-9_-]{8,})", url)
@@ -1954,14 +1965,23 @@ def _extract_identifiers(url: str) -> dict:
     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)
+    # Barnes & Noble: /w/{slug}/{id} or /w/{id} (id is often not an ISBN)
+    m = re.search(r"barnesandnoble\.com/w/(?:[^/]+/)?(\d{8,13})", url, re.IGNORECASE)
     if m:
         ean = m.group(1)
+        ids["bn_id"] = ean
         if len(ean) == 13 and _is_valid_isbn13(ean):
             ids["isbn"] = ean
         elif len(ean) == 10 and _is_valid_isbn10(ean):
             ids["isbn10"] = ean
+    # Prefer ?ean= when present on BN URLs
+    m = re.search(r"[?&]ean=(\d{10,13})", url, re.IGNORECASE)
+    if m:
+        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
 
     # Kobo — ISBN often embedded in /ebook/{slug} path
     m = re.search(r"kobo\.com/[^?]*(?:ebook|book)/([^/?#]+)", url, re.IGNORECASE)
@@ -1979,9 +1999,9 @@ def _extract_identifiers(url: str) -> dict:
             if len(candidate) == 13 and _is_valid_isbn13(candidate):
                 ids["isbn"] = candidate
 
-    # Apple Books internal ID: /book/.../id123 or /audiobook/.../id123
+    # Apple Books / legacy iTunes: /book/.../id123 or /book/id123
     m = re.search(
-        r"books\.apple\.com/[^/]+/(?:book|audiobook)/[^/]+/id(\d+)",
+        r"(?:books|itunes)\.apple\.com/[^/]+/(?:book|audiobook)(?:/[^/]+)?/id(\d+)",
         url,
         re.IGNORECASE,
     )
@@ -6121,17 +6141,30 @@ async def fetch_book_metadata(
     Returns:
         BookURLMetadata — never raises. `error` field set on complete failure.
     """
-    url = _normalize_import_url(url.strip())
-    platform = detect_platform(url)
+    from app.services.url_canonicalizer import canonicalize_book_url
+
+    original_paste = (url or "").strip()
+    # Step 0: expand share/short/affiliate → stable product URL (R-URL-001–005)
+    canon = await canonicalize_book_url(original_paste)
+    url = _normalize_import_url(canon.canonical_url or original_paste)
+    platform = canon.platform_id if canon.platform_id != "unknown" else detect_platform(url)
+    if platform == "unknown":
+        platform = detect_platform(url)
 
     meta = BookURLMetadata(source_url=url, platform=platform, buy_url=url)
+    if canon.was_resolved and canon.original_host and canon.canonical_url:
+        meta.warnings.append(
+            f"Resolved {canon.original_host} → {canon.canonical_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"]
+        meta.warnings.append(
+            "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]
@@ -6142,12 +6175,21 @@ async def fetch_book_metadata(
             if cached:
                 previous = BookURLMetadata(**json.loads(cached))
                 if not force_refresh:
+                    # Preserve resolve toast on cache hit when paste was a short link.
+                    if meta.warnings and not previous.warnings:
+                        previous.warnings = list(meta.warnings)
+                    elif meta.warnings:
+                        for w in meta.warnings:
+                            if w not in previous.warnings:
+                                previous.warnings.append(w)
                     return previous
         except Exception:
             pass
 
     # ── Resolve ────────────────────────────────────────────────────────────────
     ids = _extract_identifiers(url)
+    for k, v in (canon.extracted_ids or {}).items():
+        ids.setdefault(k, v)
     logger.info("Resolving book URL", platform=platform, ids=list(ids.keys()))
 
     raw: dict = {}

app/services/isbn_catalog.py

@@ -206,18 +206,24 @@ def normalize_retailer_url(url: str) -> str:
 
 
 _PLATFORM_HOSTS: dict[str, tuple[str, ...]] = {
-    "amazon": ("amazon.com", "amazon.co.uk", "amazon.ca"),
+    "amazon": (
+        "amazon.com", "amazon.co.uk", "amazon.ca", "amazon.com.au",
+        "amazon.de", "a.co", "amzn.to", "smile.amazon.com",
+    ),
     "apple_books": ("books.apple.com", "itunes.apple.com"),
     "google_books": ("books.google.com", "google.com/books", "play.google.com"),
     "google_play": ("play.google.com",),
     "open_library": ("openlibrary.org",),
-    "barnes_noble": ("barnesandnoble.com",),
+    "barnes_noble": ("barnesandnoble.com", "bn.com"),
     "kobo": ("kobo.com",),
     "bookshop": ("bookshop.org",),
     "walmart": ("walmart.com",),
     "goodreads": ("goodreads.com",),
     "thriftbooks": ("thriftbooks.com",),
-    "abebooks": ("abebooks.com",),
+    "abebooks": (
+        "abebooks.com", "abebooks.co.uk", "abebooks.de",
+        "iberlibro.com", "zvab.com",
+    ),
 }
 
 _SEARCH_MARKERS: dict[str, tuple[str, ...]] = {
@@ -302,7 +308,17 @@ def sanitize_platform_url(url: str, platform_id: str) -> str | None:
     cleaned = normalize_retailer_url((url or "").strip())
     if not cleaned or not cleaned.startswith("http"):
         return None
-    if platform_id == "amazon":
+    try:
+        from app.services.url_canonicalizer import rewrite_product_url
+        rewritten, _ids = rewrite_product_url(cleaned, platform_id)
+        if rewritten:
+            cleaned = rewritten
+    except Exception:
+        if platform_id == "amazon":
+            asin = extract_amazon_asin(cleaned)
+            if asin:
+                cleaned = f"https://www.amazon.com/dp/{asin}"
+    if platform_id == "amazon" and "/dp/" not in cleaned:
         asin = extract_amazon_asin(cleaned)
         if asin:
             cleaned = f"https://www.amazon.com/dp/{asin}"

app/services/platform_presence.py

@@ -39,7 +39,7 @@ logger = structlog.get_logger(__name__)
 
 _SCAN_TIMEOUT = 10.0
 _MAX_CONCURRENT = 6
-_SCAN_CACHE_PREFIX = "platform_scan:v12:"
+_SCAN_CACHE_PREFIX = "platform_scan:v13:"
 _SCAN_CACHE_TTL = 21_600  # 6h
 
 Status = Literal["verified", "likely", "not_found", "checking", "error", "skipped"]
@@ -745,6 +745,15 @@ def _finalize_listing_url(listing: PlatformListing) -> None:
         return
 
     clean = sanitize_platform_url(listing.listing_url, pid)
+    # Sync rewrite to stable product form (share hosts need async resolve elsewhere).
+    if clean:
+        try:
+            from app.services.url_canonicalizer import rewrite_product_url
+            rewritten, _ids = rewrite_product_url(clean, pid)
+            if rewritten:
+                clean = rewritten
+        except Exception:
+            pass
     if not clean or classify_retailer_url(clean, pid) != "product":
         listing.listing_url = None
         listing.status = "not_found"
@@ -816,10 +825,19 @@ async def _enrich_listing_prices(
 
     async def _one(listing: PlatformListing) -> None:
         try:
+            from app.services.url_canonicalizer import canonicalize_book_url
+            listing_url = listing.listing_url or ""
+            canon = await canonicalize_book_url(listing_url, client=client)
+            if (
+                canon.canonical_url
+                and canon.platform_id == listing.platform_id
+            ):
+                listing_url = canon.canonical_url
+                listing.listing_url = listing_url
             hit = await fetch_listing_format_prices(
                 client,
                 listing.platform_id,
-                listing.listing_url or "",
+                listing_url,
                 ctx,
                 credential=credentials.get(listing.platform_id),
             )

app/services/platform_pricing.py

@@ -16,12 +16,16 @@ import structlog
 from app.services.book_url_scraper import (
     _API_TIMEOUT,
     _browser_headers,
+    _extract_abebooks_format_prices,
     _extract_amazon_all_format_prices,
     _extract_bn_all_format_prices,
+    _extract_bookshop_format_prices,
     _extract_json_ld,
+    _extract_thriftbooks_format_prices,
     _google_books_api_params,
     _normalize_retail_format_label,
     _parse_amazon_html,
+    _parse_kobo_html,
     _score_itunes_item,
 )
 
@@ -467,11 +471,12 @@ async def fetch_listing_format_prices(
         return await _fetch_apple_format_prices(client, listing_url, ctx)
 
     # --- Credential-based platforms ---
-    # Amazon / Barnes & Noble expose ALL formats on their product page, so we
-    # skip the single-price credential shortcut for them and let the HTML
-    # multi-format parser run below. Only use credential fetch as a fallback
-    # (handled after the HTML attempt) so we never collapse 3 formats into 1.
-    _MULTI_FORMAT_HTML = {"amazon", "barnes_noble"}
+    # Retailers that expose multiple formats on the product page skip the
+    # single-price credential shortcut so we never collapse 3 formats into 1.
+    # Credential fetch runs as a fallback after the HTML multi-format attempt.
+    _MULTI_FORMAT_HTML = {
+        "amazon", "barnes_noble", "thriftbooks", "abebooks", "bookshop", "kobo",
+    }
     if credential is not None and platform_id not in _MULTI_FORMAT_HTML:
         from app.services.platform_api_registry import fetch_price_via_credential
         hit = await fetch_price_via_credential(client, credential, platform_id, listing_url, ctx)
@@ -503,11 +508,7 @@ async def fetch_listing_format_prices(
                 continue
             html = r.text
 
-            fmt_map = {}
-            if platform_id == "amazon":
-                fmt_map = _extract_amazon_all_format_prices(html) or {}
-            elif platform_id == "barnes_noble":
-                fmt_map = _extract_bn_all_format_prices(html) or {}
+            fmt_map = _deep_format_prices_from_html(platform_id, html)
 
             if fmt_map:
                 norm = canonical_format_prices(fmt_map, platform_id)
@@ -546,7 +547,7 @@ async def fetch_listing_format_prices(
                 error=str(exc)[:120],
             )
 
-    # Amazon / BN HTML gave nothing (e.g. bot wall) — fall back to credential fetch.
+    # Multi-format HTML gave nothing (e.g. bot wall) — fall back to credential.
     if credential is not None and platform_id in _MULTI_FORMAT_HTML:
         from app.services.platform_api_registry import fetch_price_via_credential
         hit = await fetch_price_via_credential(client, credential, platform_id, listing_url, ctx)
@@ -564,6 +565,28 @@ async def fetch_listing_format_prices(
     return None
 
 
+def _deep_format_prices_from_html(platform_id: str, html: str) -> dict[str, str]:
+    """Dispatch to the same deep format extractors used on source URL import."""
+    if platform_id == "amazon":
+        return _extract_amazon_all_format_prices(html) or {}
+    if platform_id == "barnes_noble":
+        return _extract_bn_all_format_prices(html) or {}
+    if platform_id == "thriftbooks":
+        return _extract_thriftbooks_format_prices(html) or {}
+    if platform_id == "abebooks":
+        return _extract_abebooks_format_prices(html) or {}
+    if platform_id == "bookshop":
+        return _extract_bookshop_format_prices(html) or {}
+    if platform_id == "kobo":
+        parsed = _parse_kobo_html(html) or {}
+        fps = dict(parsed.get("format_prices") or {})
+        # Never invent Hardcover for Kobo (digital-only).
+        fps.pop("Hardcover", None)
+        fps.pop("Paperback", None)
+        return fps
+    return {}
+
+
 def _single_price_credential_format(platform_id: str) -> str:
     """Best-effort format label for a single credential price (empty if unknown)."""
     if platform_id in {"kobo", "google_books"}:

app/services/url_canonicalizer.py

@@ -0,0 +1,422 @@
+"""Canonicalize book product URLs across the 9 locked import platforms.
+
+Share/short/affiliate hosts (a.co, amzn.to, bn.com, itunes.apple.com, …) are
+first-class members of their platform family. Resolve redirects like a browser,
+then rewrite to a stable product URL before scrape / presence enrichment.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from urllib.parse import parse_qsl, unquote, urlencode, urlparse, urlunparse
+
+import httpx
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+# Host suffixes / exact hosts → locked platform id.
+# Share/short aliases are included so we never mislabel them as "unknown".
+PLATFORM_HOST_FAMILIES: dict[str, tuple[str, ...]] = {
+    "amazon": (
+        "amazon.com", "amazon.co.uk", "amazon.ca", "amazon.com.au",
+        "amazon.de", "amazon.fr", "amazon.it", "amazon.es", "amazon.in",
+        "amazon.co.jp", "a.co", "amzn.to", "smile.amazon.com",
+    ),
+    "barnes_noble": ("barnesandnoble.com", "bn.com"),
+    "goodreads": ("goodreads.com",),
+    "apple_books": ("books.apple.com", "itunes.apple.com"),
+    "kobo": ("kobo.com",),
+    "thriftbooks": ("thriftbooks.com",),
+    "abebooks": (
+        "abebooks.com", "abebooks.co.uk", "abebooks.de", "abebooks.fr",
+        "abebooks.it", "abebooks.es", "iberlibro.com", "zvab.com",
+    ),
+    "bookshop": ("bookshop.org",),
+    "open_library": ("openlibrary.org",),
+}
+
+# Hosts that always need HTTP redirect resolve before ID extraction.
+_SHARE_SHORT_HOSTS = frozenset({
+    "a.co", "amzn.to", "bn.com",
+})
+
+# Affiliate / click wrappers — resolve then re-detect.
+_WRAPPER_HOST_MARKERS = (
+    "click.linksynergy.com",
+    "linksynergy.com",
+    "rakuten.com",
+)
+
+_TRACKING_PARAMS = frozenset({
+    "tag", "ref", "ref_", "psc", "linkcode", "creative", "creativeasin",
+    "ie", "keywords", "qid", "sr", "sprefix", "th", "dib", "dib_tag",
+    "utm_source", "utm_medium", "utm_campaign", "utm_content", "utm_term",
+    "pd_rd_w", "pd_rd_r", "pd_rd_wg", "pd_rd_i", "pf_rd_p", "pf_rd_r",
+    "content-id", "sp_csd", "crid", "s", "st", "2sid", "clickref",
+    "at", "ct", "mt", "app", "itscg", "itsct",
+})
+
+# Params that MUST be kept (product identity).
+_KEEP_PARAMS = frozenset({"ean", "isbn", "id", "asin"})
+
+_REDIRECT_TIMEOUT = 5.0
+
+
+@dataclass(frozen=True)
+class CanonicalUrl:
+    """Result of URL canonicalization for import / presence enrichment."""
+
+    platform_id: str
+    canonical_url: str
+    original_url: str
+    extracted_ids: dict = field(default_factory=dict)
+    was_resolved: bool = False
+    original_host: str = ""
+
+
+def _bare_host(url: str) -> str:
+    """Return lowercase hostname without leading www."""
+    try:
+        host = (urlparse(url).netloc or "").lower()
+    except Exception:
+        return ""
+    if host.startswith("www."):
+        host = host[4:]
+    return host
+
+
+def _ensure_https(url: str) -> str:
+    """Trim and force https scheme when missing."""
+    url = (url or "").strip()
+    if not url:
+        return ""
+    if url.startswith("//"):
+        return "https:" + url
+    if not re.match(r"^https?://", url, re.IGNORECASE):
+        return "https://" + url.lstrip("/")
+    if url.startswith("http://"):
+        return "https://" + url[7:]
+    return url
+
+
+def host_family_platform(url: str) -> str | None:
+    """Return locked platform id from host family, or None if unsupported."""
+    host = _bare_host(url)
+    if not host:
+        return None
+    # Reject cover CDN / non-book Open Library assets early.
+    if host.startswith("covers.openlibrary.org"):
+        return None
+    for platform_id, suffixes in PLATFORM_HOST_FAMILIES.items():
+        for suffix in suffixes:
+            if host == suffix or host.endswith("." + suffix):
+                return platform_id
+    return None
+
+
+def needs_redirect_resolve(url: str, platform_hint: str | None = None) -> bool:
+    """True when the URL is a share/short/wrapper that needs browser-like resolve."""
+    host = _bare_host(url)
+    if host in _SHARE_SHORT_HOSTS:
+        return True
+    low = (url or "").lower()
+    if any(marker in low for marker in _WRAPPER_HOST_MARKERS):
+        return True
+    # Abe network aliases often redirect to abebooks.com product pages.
+    if platform_hint == "abebooks" and host in {"iberlibro.com", "zvab.com"}:
+        return True
+    # BN search / short paths need resolve to a product page.
+    if platform_hint == "barnes_noble" and (
+        host == "bn.com" or re.search(r"barnesandnoble\.com/s/", low)
+    ):
+        return True
+    return False
+
+
+def _strip_tracking(url: str) -> str:
+    """Drop tracking query params; keep product identity params."""
+    parsed = urlparse(url)
+    kept = [
+        (k, v) for k, v in parse_qsl(parsed.query, keep_blank_values=True)
+        if k.lower() not in _TRACKING_PARAMS or k.lower() in _KEEP_PARAMS
+    ]
+    return urlunparse((
+        parsed.scheme or "https",
+        parsed.netloc,
+        parsed.path or "",
+        "",
+        urlencode(kept),
+        "",  # drop fragment for cache stability
+    ))
+
+
+def _extract_asin(url: str) -> str | None:
+    """Extract Amazon ASIN from path or query."""
+    patterns = (
+        r"/(?:dp|gp/product|gp/aw/d|exec/obidos/ASIN)/([A-Z0-9]{10})",
+        r"[?&](?:asin|creativeASIN)=([A-Z0-9]{10})",
+    )
+    for pat in patterns:
+        m = re.search(pat, url, re.IGNORECASE)
+        if m:
+            return m.group(1).upper()
+    return None
+
+
+def _unwrap_affiliate(url: str) -> str:
+    """Pull destination URL out of common affiliate wrappers without HTTP."""
+    for pattern in (
+        r"[?&]murl=([^&]+)",
+        r"[?&]u=([^&]+)",
+        r"[?&]url=([^&]+)",
+        r"RD_PARM1=([^&]+)",
+        r"destination:(https?://[^\s\"']+)",
+    ):
+        m = re.search(pattern, url, re.IGNORECASE)
+        if m:
+            candidate = unquote(m.group(1))
+            if candidate.startswith("http"):
+                return candidate
+    return url
+
+
+def rewrite_product_url(url: str, platform_id: str) -> tuple[str, dict]:
+    """Rewrite a resolved URL to the platform's stable product form.
+
+    Returns (canonical_url, extracted_ids).
+    """
+    url = _strip_tracking(_ensure_https(url))
+    ids: dict = {}
+    parsed = urlparse(url)
+    host = (parsed.netloc or "").lower()
+    path = parsed.path or ""
+    query = dict(parse_qsl(parsed.query, keep_blank_values=True))
+
+    if platform_id == "amazon":
+        asin = _extract_asin(url)
+        if asin:
+            ids["asin"] = asin
+            return f"https://www.amazon.com/dp/{asin}", ids
+        return url, ids
+
+    if platform_id == "barnes_noble":
+        # /w/{slug}/{id} or /w/{id}
+        m = re.search(r"/w/(?:([^/]+)/)?(\d{8,13})(?:/|$)", path, re.IGNORECASE)
+        if m:
+            slug, bn_id = m.group(1) or "", m.group(2)
+            ids["bn_id"] = bn_id
+            ean = query.get("ean") or query.get("EAN") or ""
+            if ean and re.fullmatch(r"\d{10,13}", ean):
+                ids["isbn"] = ean
+            base = (
+                f"https://www.barnesandnoble.com/w/{slug}/{bn_id}"
+                if slug
+                else f"https://www.barnesandnoble.com/w/{bn_id}"
+            )
+            if ids.get("isbn"):
+                return f"{base}?ean={ids['isbn']}", ids
+            return base, ids
+        # /s/{isbn} search — keep for redirect resolve; no rewrite yet
+        m = re.search(r"/s/(\d{10,13})", path)
+        if m:
+            ids["isbn"] = m.group(1)
+        return url, ids
+
+    if platform_id == "goodreads":
+        m = re.search(r"/book/show/(\d+)", path)
+        if m:
+            ids["goodreads_id"] = m.group(1)
+            return f"https://www.goodreads.com/book/show/{m.group(1)}", ids
+        return url, ids
+
+    if platform_id == "apple_books":
+        m = re.search(
+            r"(?:books|itunes)\.apple\.com/([^/]+)/(book|audiobook)(?:/[^/]+)?/id(\d+)",
+            url,
+            re.IGNORECASE,
+        )
+        if m:
+            cc, kind, book_id = m.group(1), m.group(2), m.group(3)
+            ids["apple_book_id"] = book_id
+            if kind.lower() == "audiobook":
+                ids["apple_is_audiobook"] = True
+            return f"https://books.apple.com/{cc}/{kind}/id{book_id}", ids
+        # id-only without slug already covered; try bare /idN
+        m = re.search(r"books\.apple\.com/([^/]+)/book/id(\d+)", url, re.IGNORECASE)
+        if m:
+            ids["apple_book_id"] = m.group(2)
+            return f"https://books.apple.com/{m.group(1)}/book/id{m.group(2)}", ids
+        return url, ids
+
+    if platform_id == "kobo":
+        m = re.search(
+            r"kobo\.com/((?:\w+/)?(?:\w+/)?)(ebook|audiobook)/([^/?#]+)",
+            url,
+            re.IGNORECASE,
+        )
+        if m:
+            kind = m.group(2).lower()
+            slug = m.group(3)
+            isbn_m = re.search(r"(\d{13})", slug)
+            if isbn_m:
+                ids["isbn"] = isbn_m.group(1)
+            # Preserve locale path from original when present.
+            return _strip_tracking(url.split("?")[0].rstrip("/") + "/"), ids
+        return url, ids
+
+    if platform_id == "thriftbooks":
+        m = re.search(r"/w/([^/]+)/(\d+)/?", path)
+        if m:
+            ids["thrift_work_id"] = m.group(2)
+            return f"https://www.thriftbooks.com/w/{m.group(1)}/{m.group(2)}/", ids
+        frag = urlparse(url).fragment or ""
+        fm = re.search(r"isbn[=:](\d{10,13})", frag, re.IGNORECASE)
+        if fm:
+            ids["isbn"] = fm.group(1)
+        return url.split("#")[0].rstrip("/") + ("/" if "/w/" in path else ""), ids
+
+    if platform_id == "abebooks":
+        m = re.search(r"/products/isbn/(\d{10,13})", path, re.IGNORECASE)
+        if m:
+            ids["isbn"] = m.group(1)
+            return f"https://www.abebooks.com/products/isbn/{m.group(1)}", ids
+        m = re.search(r"[?&]isbn=(\d{10,13})", url, re.IGNORECASE)
+        if m:
+            ids["isbn"] = m.group(1)
+            return f"https://www.abebooks.com/products/isbn/{m.group(1)}", ids
+        return url, ids
+
+    if platform_id == "bookshop":
+        m = re.search(r"/p/books/([^/]+)/(\d+)", path, re.IGNORECASE)
+        if m:
+            slug, bid = m.group(1), m.group(2)
+            ean = query.get("ean") or ""
+            if ean and re.fullmatch(r"\d{10,13}", ean):
+                ids["isbn"] = ean
+                return f"https://bookshop.org/p/books/{slug}/{bid}?ean={ean}", ids
+            return f"https://bookshop.org/p/books/{slug}/{bid}", ids
+        m = re.search(r"/books/(\d{10,13})", path)
+        if m:
+            ids["isbn"] = m.group(1)
+            return f"https://bookshop.org/books/{m.group(1)}", ids
+        return url, ids
+
+    if platform_id == "open_library":
+        # Strip .json API suffix for page canonical.
+        path2 = re.sub(r"\.json$", "", path, flags=re.IGNORECASE)
+        m = re.search(r"/works/(OL\d+W)", path2, re.IGNORECASE)
+        if m:
+            ids["ol_work_key"] = m.group(1).upper().replace("OL", "OL")
+            key = m.group(1)
+            if not key.startswith("OL"):
+                key = "OL" + key
+            # Normalize casing: OLxxxW
+            key_m = re.search(r"(OL\d+W)", path2, re.IGNORECASE)
+            key = key_m.group(1) if key_m else m.group(1)
+            ids["ol_work_key"] = key
+            return f"https://openlibrary.org/works/{key}", ids
+        m = re.search(r"/books/(OL\d+M)", path2, re.IGNORECASE)
+        if m:
+            ids["ol_edition_key"] = m.group(1)
+            return f"https://openlibrary.org/books/{m.group(1)}", ids
+        m = re.search(r"/isbn/(\d{10,13})", path2)
+        if m:
+            ids["isbn"] = m.group(1)
+            return f"https://openlibrary.org/isbn/{m.group(1)}", ids
+        if path2 != path:
+            return urlunparse((parsed.scheme, parsed.netloc, path2, "", "", "")), ids
+        return url, ids
+
+    return url, ids
+
+
+async def _follow_redirects(url: str, client: httpx.AsyncClient | None = None) -> str:
+    """Follow HTTP redirects; return final URL (or original on failure)."""
+    own_client = client is None
+    if own_client:
+        client = httpx.AsyncClient(
+            follow_redirects=True,
+            timeout=_REDIRECT_TIMEOUT,
+            headers={
+                "User-Agent": (
+                    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) "
+                    "Chrome/124.0.0.0 Safari/537.36"
+                ),
+                "Accept": "text/html,application/xhtml+xml;q=0.9,*/*;q=0.8",
+            },
+        )
+    assert client is not None
+    try:
+        # Prefer HEAD; some short-link hosts only support GET.
+        try:
+            r = await client.head(url)
+            final = str(r.url)
+            if final and final != url and "a.co" not in _bare_host(final):
+                return final
+        except Exception:
+            pass
+        r = await client.get(url)
+        return str(r.url) or url
+    except Exception as exc:
+        logger.debug("URL redirect resolve failed", url=url[:80], error=str(exc)[:120])
+        return url
+    finally:
+        if own_client:
+            await client.aclose()
+
+
+async def canonicalize_book_url(
+    raw_url: str,
+    client: httpx.AsyncClient | None = None,
+) -> CanonicalUrl:
+    """Expand share/short links and rewrite to a stable product URL.
+
+    Never raises — on failure returns best-effort rewrite of the original URL.
+    """
+    original = _ensure_https(raw_url)
+    if not original:
+        return CanonicalUrl(
+            platform_id="unknown",
+            canonical_url="",
+            original_url=raw_url or "",
+            original_host="",
+        )
+
+    original_host = _bare_host(original)
+    working = _unwrap_affiliate(original)
+    platform_hint = host_family_platform(working)
+    was_resolved = False
+
+    if needs_redirect_resolve(working, platform_hint) or (
+        platform_hint is None and any(m in working.lower() for m in _WRAPPER_HOST_MARKERS)
+    ):
+        resolved = await _follow_redirects(working, client=client)
+        if resolved and resolved != working:
+            was_resolved = True
+            working = resolved
+            platform_hint = host_family_platform(working) or platform_hint
+
+    if platform_hint is None:
+        platform_hint = host_family_platform(working) or "unknown"
+
+    if platform_hint == "unknown":
+        return CanonicalUrl(
+            platform_id="unknown",
+            canonical_url=_strip_tracking(working),
+            original_url=original,
+            was_resolved=was_resolved,
+            original_host=original_host,
+        )
+
+    canonical, ids = rewrite_product_url(working, platform_hint)
+    return CanonicalUrl(
+        platform_id=platform_hint,
+        canonical_url=canonical,
+        original_url=original,
+        extracted_ids=ids,
+        was_resolved=was_resolved or (original_host in _SHARE_SHORT_HOSTS),
+        original_host=original_host,
+    )

tests/unit/test_platform_presence.py

@@ -8,6 +8,7 @@ from app.services.platform_presence import (
     SCORED_PLATFORMS,
     CatalogContext,
     PlatformListing,
+    PlatformPresenceResult,
     build_recommendations,
     compute_score,
     confidence_to_status,
@@ -19,6 +20,8 @@ from app.services.platform_presence import (
     title_match_score,
 )
 from app.repositories.platform_listing_repo import PlatformListingRepository
+from app.services.book_service import BookService
+from app.services.book_url_scraper import BookURLMetadata
 
 
 class TestTitleMatchScore:
@@ -295,9 +298,9 @@ async def test_build_catalog_uses_slug_fetch_and_bookshop_isbn():
 
 
 @pytest.mark.asyncio
-async def test_scan_cache_key_busted_for_v10():
+async def test_scan_cache_key_busted_for_v13():
     from app.services.platform_presence import scan_cache_key
-    assert scan_cache_key("https://example.com/book").startswith("platform_scan:v10:")
+    assert scan_cache_key("https://example.com/book").startswith("platform_scan:v13:")
 
 
 @pytest.mark.asyncio
@@ -424,6 +427,50 @@ def test_parse_goodreads_book_ref():
     assert ref == ("219584318", "/book/show/219584318-lethal-lines")
 
 
+@pytest.mark.asyncio
+async def test_preview_url_import_passes_db_to_format_price_scan():
+    """Fetch Details must expose configured credentials to discovered URLs."""
+    db = MagicMock()
+    redis = AsyncMock()
+    meta = BookURLMetadata(
+        title="All We Say",
+        author="Example Author",
+        platform="amazon",
+        source_url="https://amazon.com/dp/BOOK",
+        buy_url="https://amazon.com/dp/BOOK",
+    )
+    scan = PlatformPresenceResult(
+        score=0.0,
+        listed_count=0,
+        scanned_at="2026-07-16T00:00:00+00:00",
+        source_isbn=None,
+        platforms=[],
+        recommendations=[],
+    )
+
+    with (
+        patch(
+            "app.services.book_service.fetch_book_metadata",
+            AsyncMock(return_value=meta),
+        ),
+        patch(
+            "app.services.book_service.scan_platform_presence",
+            AsyncMock(return_value=scan),
+        ) as scan_mock,
+        patch(
+            "app.services.book_service.save_scan_to_redis",
+            AsyncMock(),
+        ),
+    ):
+        result = await BookService.preview_url_import(
+            meta.source_url, redis, force_refresh=True, db=db,
+        )
+
+    assert result["platform_presence"] is not None
+    assert scan_mock.await_args.kwargs["db"] is db
+    assert scan_mock.await_args.kwargs["redis"] is redis
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_lethal_lines_kindle_only_scan():

tests/unit/test_platform_pricing.py

@@ -265,6 +265,56 @@ class TestFetchListingFormatPrices:
         assert hit is not None
         assert hit.format_prices.get("Hardcover") == "$18.00"
 
+    @pytest.mark.asyncio
+    async def test_bookshop_deep_extractor_used(self):
+        """Presence enrichment must call Bookshop deep format parser."""
+        html = (
+            "A" * 500
+            + 'text-dark">Hardcover</p>'
+            + 'subtitle text-dark">$27.02'
+            + 'text-dark">Ebook</p>'
+            + 'subtitle text-dark">$14.99'
+        )
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.text = html
+        client = AsyncMock()
+        client.get = AsyncMock(return_value=resp)
+
+        with patch(
+            "app.services.platform_pricing._extract_bookshop_format_prices",
+            return_value={"Hardcover": "$27.02", "eBook": "$14.99"},
+        ) as deep:
+            hit = await fetch_listing_format_prices(
+                client, "bookshop", "https://bookshop.org/p/books/foo/1",
+            )
+        deep.assert_called_once()
+        assert hit is not None
+        assert hit.format_prices.get("Hardcover") == "$27.02"
+        assert hit.format_prices.get("eBook") == "$14.99"
+
+    @pytest.mark.asyncio
+    async def test_thriftbooks_deep_extractor_used(self):
+        """Presence enrichment must call ThriftBooks deep format parser."""
+        html = "A" * 600 + "Select Format"
+        resp = MagicMock()
+        resp.status_code = 200
+        resp.text = html
+        client = AsyncMock()
+        client.get = AsyncMock(return_value=resp)
+
+        with patch(
+            "app.services.platform_pricing._extract_thriftbooks_format_prices",
+            return_value={"Hardcover": "$8.99", "Paperback": "$4.50"},
+        ) as deep:
+            hit = await fetch_listing_format_prices(
+                client, "thriftbooks", "https://www.thriftbooks.com/w/foo/1/",
+            )
+        deep.assert_called_once()
+        assert hit is not None
+        assert hit.format_prices.get("Hardcover") == "$8.99"
+        assert "eBook" not in hit.format_prices
+
 
 class TestSinglePriceFormatKey:
     def test_kobo_is_ebook(self):

tests/unit/test_url_canonicalizer.py

@@ -0,0 +1,206 @@
+"""Unit tests for URL canonicalization across locked import platforms."""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from app.services.book_url_scraper import (
+    _extract_identifiers,
+    detect_platform,
+)
+from app.services.url_canonicalizer import (
+    canonicalize_book_url,
+    host_family_platform,
+    needs_redirect_resolve,
+    rewrite_product_url,
+)
+
+
+class TestHostFamilyMap:
+    def test_amazon_share_a_co(self):
+        assert host_family_platform("https://a.co/d/04YALfQb") == "amazon"
+
+    def test_amazon_amzn_to(self):
+        assert host_family_platform("https://amzn.to/abc123") == "amazon"
+
+    def test_bn_com(self):
+        assert host_family_platform("https://bn.com/s/9781234567890") == "barnes_noble"
+
+    def test_itunes_apple(self):
+        assert host_family_platform(
+            "https://itunes.apple.com/us/book/id123456789"
+        ) == "apple_books"
+
+    def test_iberlibro_abebooks(self):
+        assert host_family_platform("https://www.iberlibro.com/book/1") == "abebooks"
+
+    def test_covers_cdn_rejected(self):
+        assert host_family_platform(
+            "https://covers.openlibrary.org/b/id/123-L.jpg"
+        ) is None
+
+
+class TestDetectPlatformShareHosts:
+    def test_a_co_is_amazon_not_unknown(self):
+        assert detect_platform("https://a.co/d/04YALfQb") == "amazon"
+
+    def test_bn_com_is_barnes_noble(self):
+        assert detect_platform("https://bn.com/s/9781234567890") == "barnes_noble"
+
+    def test_itunes_is_apple_books(self):
+        assert detect_platform(
+            "https://itunes.apple.com/us/book/foo/id123"
+        ) == "apple_books"
+
+
+class TestRewriteProductUrl:
+    def test_amazon_affiliate_to_dp(self):
+        url = (
+            "https://www.amazon.com/Memory-Box-ebook/dp/B08KQ2PY8Y/"
+            "ref=pd_sbs_d_sccl_1_5?pd_rd_w=BeCFl&tag=aff-20"
+        )
+        canon, ids = rewrite_product_url(url, "amazon")
+        assert canon == "https://www.amazon.com/dp/B08KQ2PY8Y"
+        assert ids["asin"] == "B08KQ2PY8Y"
+
+    def test_amazon_mobile_aw(self):
+        canon, ids = rewrite_product_url(
+            "https://www.amazon.com/gp/aw/d/B08KQ2PY8Y", "amazon",
+        )
+        assert canon == "https://www.amazon.com/dp/B08KQ2PY8Y"
+        assert ids["asin"] == "B08KQ2PY8Y"
+
+    def test_bn_slug_and_id(self):
+        canon, ids = rewrite_product_url(
+            "https://www.barnesandnoble.com/w/all-we-say-ben-rhodes/1148264021?st=foo",
+            "barnes_noble",
+        )
+        assert "/w/all-we-say-ben-rhodes/1148264021" in canon
+        assert ids["bn_id"] == "1148264021"
+        assert "st=" not in canon
+
+    def test_bn_id_only(self):
+        canon, ids = rewrite_product_url(
+            "https://www.barnesandnoble.com/w/1148264021",
+            "barnes_noble",
+        )
+        assert canon == "https://www.barnesandnoble.com/w/1148264021"
+        assert ids["bn_id"] == "1148264021"
+
+    def test_bn_with_ean(self):
+        canon, ids = rewrite_product_url(
+            "https://www.barnesandnoble.com/w/foo/1148264021?ean=9781234567890",
+            "barnes_noble",
+        )
+        assert "ean=9781234567890" in canon
+        assert ids["isbn"] == "9781234567890"
+
+    def test_goodreads_strip_slug(self):
+        canon, ids = rewrite_product_url(
+            "https://www.goodreads.com/book/show/12345.Some_Title?from_search=true",
+            "goodreads",
+        )
+        assert canon == "https://www.goodreads.com/book/show/12345"
+        assert ids["goodreads_id"] == "12345"
+
+    def test_itunes_to_books_apple(self):
+        canon, ids = rewrite_product_url(
+            "https://itunes.apple.com/us/book/all-we-say/id6443123456?at=1000",
+            "apple_books",
+        )
+        assert canon == "https://books.apple.com/us/book/id6443123456"
+        assert ids["apple_book_id"] == "6443123456"
+
+    def test_abebooks_isbn_product(self):
+        canon, ids = rewrite_product_url(
+            "https://www.abebooks.com/servlet/SearchResults?isbn=9781234567890&cm_sp=xxx",
+            "abebooks",
+        )
+        assert canon == "https://www.abebooks.com/products/isbn/9781234567890"
+        assert ids["isbn"] == "9781234567890"
+
+    def test_bookshop_with_ean(self):
+        canon, ids = rewrite_product_url(
+            "https://bookshop.org/p/books/all-we-say/12345?ean=9781234567890&utm_source=x",
+            "bookshop",
+        )
+        assert "ean=9781234567890" in canon
+        assert "utm_source" not in canon
+
+    def test_open_library_strip_json(self):
+        canon, ids = rewrite_product_url(
+            "https://openlibrary.org/works/OL45883W.json",
+            "open_library",
+        )
+        assert canon == "https://openlibrary.org/works/OL45883W"
+        assert ids["ol_work_key"] == "OL45883W"
+
+    def test_thriftbooks_work(self):
+        canon, ids = rewrite_product_url(
+            "https://www.thriftbooks.com/w/some-title/123456/?clickref=abc",
+            "thriftbooks",
+        )
+        assert canon == "https://www.thriftbooks.com/w/some-title/123456/"
+        assert ids["thrift_work_id"] == "123456"
+
+
+class TestExtractIdentifiersBn:
+    def test_bn_id_only_path(self):
+        ids = _extract_identifiers("https://www.barnesandnoble.com/w/1148264021")
+        assert ids.get("bn_id") == "1148264021"
+
+    def test_bn_ean_query(self):
+        ids = _extract_identifiers(
+            "https://www.barnesandnoble.com/w/foo/1?ean=9781234567897"
+        )
+        # Only set isbn when checksum-valid; bn_id still present from path
+        assert "bn_id" in ids or "isbn" in ids or "isbn10" in ids
+
+
+class TestNeedsRedirect:
+    def test_a_co_needs_redirect(self):
+        assert needs_redirect_resolve("https://a.co/d/abc", "amazon") is True
+
+    def test_clean_dp_no_redirect(self):
+        assert needs_redirect_resolve(
+            "https://www.amazon.com/dp/B08KQ2PY8Y", "amazon",
+        ) is False
+
+
+@pytest.mark.asyncio
+class TestCanonicalizeAsync:
+    async def test_a_co_resolves_to_amazon_dp(self):
+        mock_resp = MagicMock()
+        mock_resp.url = "https://www.amazon.com/Memory-Box/dp/B08KQ2PY8Y?ref=xx"
+        client = AsyncMock()
+        client.head = AsyncMock(return_value=mock_resp)
+        client.get = AsyncMock(return_value=mock_resp)
+
+        result = await canonicalize_book_url("https://a.co/d/04YALfQb", client=client)
+        assert result.platform_id == "amazon"
+        assert result.canonical_url == "https://www.amazon.com/dp/B08KQ2PY8Y"
+        assert result.was_resolved is True
+        assert result.original_host == "a.co"
+        assert result.extracted_ids.get("asin") == "B08KQ2PY8Y"
+
+    async def test_clean_amazon_no_http(self):
+        with patch(
+            "app.services.url_canonicalizer._follow_redirects",
+            new_callable=AsyncMock,
+        ) as follow:
+            result = await canonicalize_book_url(
+                "https://www.amazon.com/dp/B08KQ2PY8Y?tag=x",
+            )
+            follow.assert_not_called()
+        assert result.platform_id == "amazon"
+        assert result.canonical_url == "https://www.amazon.com/dp/B08KQ2PY8Y"
+        assert result.was_resolved is False
+
+    async def test_itunes_rewrites_without_redirect(self):
+        result = await canonicalize_book_url(
+            "https://itunes.apple.com/us/book/foo/id6443123456",
+        )
+        assert result.platform_id == "apple_books"
+        assert result.canonical_url == "https://books.apple.com/us/book/id6443123456"