feat(metadata): choose the Amazon store used by the Amazon provider

cps/metadata_provider/amazon.py

@@ -23,6 +23,51 @@
 from operator import itemgetter
 log = logger.create()
 
+# Amazon stores selectable in the CWA settings (metadata_amazon_domain), with
+# the Accept-Language header sent to each one. Japan has its own provider (amazonjp).
+AMAZON_DOMAINS = {
+    "com": "en-US,en;q=0.9",
+    "co.uk": "en-GB,en;q=0.9",
+    "ca": "en-CA,en;q=0.9,fr-CA;q=0.8",
+    "com.au": "en-AU,en;q=0.9",
+    "in": "en-IN,en;q=0.9",
+    "de": "de-DE,de;q=0.9,en;q=0.5",
+    "fr": "fr-FR,fr;q=0.9,en;q=0.5",
+    "it": "it-IT,it;q=0.9,en;q=0.5",
+    "es": "es-ES,es;q=0.9,en;q=0.5",
+    "nl": "nl-NL,nl;q=0.9,en;q=0.5",
+    "com.br": "pt-BR,pt;q=0.9,en;q=0.5",
+    "com.mx": "es-MX,es;q=0.9,en;q=0.5",
+}
+DEFAULT_AMAZON_DOMAIN = "com"
+
+# "Book 2 of 5: Series", "Libro 2 di 5: Serie", "Buch 2 von 5: Reihe", "Tome 2 sur 5 : Série"...
+SERIES_RE = re.compile(
+    r'(?:Book|Libro|Buch|Band|Tome|Livre|Boek|Deel|Livro)\s+(\d+)'
+    r'(?:\s+(?:of|di|de|von|aus|sur|van|do)\s+\d+)?\s*:\s*(.+)',
+    re.IGNORECASE,
+)
+
+
+def normalize_amazon_domain(value) -> str:
+    """Return a supported Amazon domain suffix ("com", "it", "co.uk"...), defaulting to amazon.com."""
+    value = str(value or "").strip().lower()
+    for prefix in ("https://", "http://", "www.", "amazon."):
+        if value.startswith(prefix):
+            value = value[len(prefix):]
+    value = value.strip("/")
+    return value if value in AMAZON_DOMAINS else DEFAULT_AMAZON_DOMAIN
+
+
+def get_amazon_domain() -> str:
+    """Read the configured Amazon store from the CWA settings."""
+    try:
+        from scripts.cwa_db import CWA_DB  # imported here to avoid circular imports
+        return normalize_amazon_domain(CWA_DB().cwa_settings.get("metadata_amazon_domain"))
+    except Exception as e:
+        log.debug(f"Could not read metadata_amazon_domain, using amazon.{DEFAULT_AMAZON_DOMAIN}: {e}")
+        return DEFAULT_AMAZON_DOMAIN
+
 
 class Amazon(Metadata):
     __name__ = "Amazon"
@@ -38,16 +83,20 @@ class Amazon(Metadata):
     def search(
         self, query: str, generic_cover: str = "", locale: str = "en"
     ) -> Optional[List[MetaRecord]]:
+        domain = get_amazon_domain()
+        store_url = f"https://www.amazon.{domain}"
+        request_headers = {"Accept-Language": AMAZON_DOMAINS[domain]}
+
         def inner(link, index) -> [dict, int]:
-            if link.startswith("/sspa/"):
-                return []  # sspa links are not book pages
+            if link.startswith("/sspa/") or link.startswith(("http://", "https://")):
+                return []  # sponsored/ad links are not book pages
 
             # Ensure link starts with / for proper URL construction
             if not link.startswith('/'):
                 link = f"/{link}"
 
             try:
-                r = self.session.get(f"https://www.amazon.com{link}", timeout=10)
+                r = self.session.get(f"{store_url}{link}", headers=request_headers, timeout=10)
                 r.raise_for_status()
             except Exception as ex:
                 log.warning(ex)
@@ -63,7 +112,7 @@ def inner(link, index) -> [dict, int]:
             if soup2 is None:
                 return []
             if soup2.find("input", attrs={"name": "submit.preorder"}) is not None:
-                log.debug(f"Skipping pre-order page: https://www.amazon.com{link}")
+                log.debug(f"Skipping pre-order page: {store_url}{link}")
                 return []  # pre-order page, ignore
             try:
                 match = MetaRecord(
@@ -72,9 +121,9 @@ def inner(link, index) -> [dict, int]:
                     source=MetaSourceInfo(
                         id=self.__id__,
                         description="Amazon Books",
-                        link="https://amazon.com/"
+                        link=f"https://amazon.{domain}/"
                     ),
-                    url = f"https://www.amazon.com{link}",
+                    url = f"{store_url}{link}",
                     #the more searches the slower, these are too hard to find in reasonable time or might not even exist
                     publisher= "",  # very unreliable
                     publishedDate= "",  # very unreliable
@@ -123,7 +172,7 @@ def inner(link, index) -> [dict, int]:
                         if series_text:
                             series_str = str(series_text.text).strip()
                             # Match patterns like "Book X of Y: Series Title" or "Book X: Series Title"
-                            series_match = re.search(r'Book\s+(\d+)(?:\s+of\s+\d+)?:\s*(.+)', series_str, re.IGNORECASE)
+                            series_match = SERIES_RE.search(series_str)
                             if series_match:
                                 match.series_index = int(series_match.group(1))
                                 match.series = series_match.group(2).strip()
@@ -171,9 +220,9 @@ def inner(link, index) -> [dict, int]:
 
             try:
                 results = self.session.get(
-                    "https://www.amazon.com/s",
+                    f"{store_url}/s",
                     params=q,
-                    # headers=self.headers,
+                    headers=request_headers,
                     timeout=10,
                 )
                 results.raise_for_status()
@@ -193,7 +242,7 @@ def inner(link, index) -> [dict, int]:
                     if base_url not in links_list:
                         links_list.append(base_url)
             if len(links_list) == 0:
-                log.info(f"No Amazon search results found for query: {query}")
+                log.info(f"No amazon.{domain} search results found for query: {query}")
                 return []
             with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
                 fut = {executor.submit(inner, link, index) for index, link in enumerate(links_list[:3])}

cps/templates/cwa_settings.html

@@ -543,6 +543,28 @@ <h4 class="settings-section-header">{{_('Enabled Metadata Providers')}}</h4>
           <input type="hidden" name="metadata_providers_enabled" id="metadata_providers_enabled_hidden" value="{}">
         </div>
       </div>
+
+      <!-- Amazon store (region) used by the Amazon metadata provider -->
+      <div class="form-group">
+        <h4 class="settings-section-header">{{_('Amazon Store')}}</h4>
+        <p class="settings-description">
+          {{_('Amazon store used by the Amazon metadata provider. Pick your country to get titles and descriptions of local editions, like the region setting of Calibre desktop.')}}
+        </p>
+        {% set amazon_stores = [
+          ('com', 'amazon.com'), ('co.uk', 'amazon.co.uk'), ('ca', 'amazon.ca'), ('com.au', 'amazon.com.au'),
+          ('in', 'amazon.in'), ('de', 'amazon.de'), ('fr', 'amazon.fr'), ('it', 'amazon.it'),
+          ('es', 'amazon.es'), ('nl', 'amazon.nl'), ('com.br', 'amazon.com.br'), ('com.mx', 'amazon.com.mx')
+        ] %}
+        <label for="metadata_amazon_domain" style="padding-right: 10px;">{{_('Store:')}}</label>
+        <select class="cwa-settings-select" name="metadata_amazon_domain" id="metadata_amazon_domain">
+          {% for value, label in amazon_stores %}
+            <option value="{{ value }}" {% if cwa_settings.get('metadata_amazon_domain', 'com') == value %}selected{% endif %}>{{ label }}</option>
+          {% endfor %}
+        </select>
+        <div class="cwa-settings-tip">
+          <small class="settings-explanation">💡 <strong>{{_('Tip')}}:</strong> {{_('Amazon Japan has its own provider (AmazonJp).')}}</small>
+        </div>
+      </div>
     </div>
 
     <!-- Hardcover Auto-Fetch Settings -->

scripts/cwa_schema.sql

@@ -72,6 +72,7 @@ CREATE TABLE IF NOT EXISTS cwa_settings(
     cover_download_max_mb INTEGER DEFAULT 15 NOT NULL,
     metadata_provider_hierarchy TEXT DEFAULT '["ibdb","google","dnb"]' NOT NULL,
     metadata_providers_enabled TEXT DEFAULT '{}' NOT NULL,
+    metadata_amazon_domain TEXT DEFAULT 'com' NOT NULL,
     auto_send_delay_minutes INTEGER DEFAULT 5 NOT NULL,
     duplicate_detection_title SMALLINT DEFAULT 1 NOT NULL,
     duplicate_detection_author SMALLINT DEFAULT 1 NOT NULL,

tests/unit/test_amazon_provider.py

@@ -0,0 +1,82 @@
+# -*- coding: utf-8 -*-
+# Calibre-Web Automated – fork of Calibre-Web
+# Copyright (C) 2024-2026 Calibre-Web Automated contributors
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+import pytest
+
+from cps.metadata_provider import amazon
+from cps.metadata_provider.amazon import Amazon, SERIES_RE, normalize_amazon_domain
+
+
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("it", "it"),
+        ("co.uk", "co.uk"),
+        (" DE ", "de"),
+        ("amazon.fr", "fr"),
+        ("https://www.amazon.es/", "es"),
+        ("", "com"),
+        (None, "com"),
+        ("co.jp", "com"),  # Japan has its own provider
+        ("evil.example", "com"),
+    ],
+)
+def test_normalize_amazon_domain(value, expected):
+    assert normalize_amazon_domain(value) == expected
+
+
+@pytest.mark.parametrize(
+    "text, index, series",
+    [
+        ("Book 2 of 5: Dungeon Crawler Carl", "2", "Dungeon Crawler Carl"),
+        ("Book 3: The Stormlight Archive", "3", "The Stormlight Archive"),
+        ("Libro 3 di 7: Dungeon Crawler Carl", "3", "Dungeon Crawler Carl"),
+        ("Buch 1 von 4: Die Tribute von Panem", "1", "Die Tribute von Panem"),
+        ("Band 2 aus 3: Reihe", "2", "Reihe"),
+        ("Tome 2 sur 5 : Le Trône de verre", "2", "Le Trône de verre"),
+        ("Libro 4 de 6: Trono de cristal", "4", "Trono de cristal"),
+    ],
+)
+def test_series_regex_handles_store_languages(text, index, series):
+    match = SERIES_RE.search(text)
+    assert match is not None
+    assert match.group(1) == index
+    assert match.group(2).strip() == series
+
+
+class _FakeResponse:
+    text = "<html><body></body></html>"
+
+    def raise_for_status(self):
+        return None
+
+
+def test_search_uses_configured_store(monkeypatch):
+    calls = []
+
+    def fake_get(url, **kwargs):
+        calls.append((url, kwargs.get("headers") or {}))
+        return _FakeResponse()
+
+    monkeypatch.setattr(amazon, "get_amazon_domain", lambda: "it")
+    provider = Amazon()
+    monkeypatch.setattr(provider, "session", type("S", (), {"get": staticmethod(fake_get)})())
+    provider.active = True
+
+    assert provider.search("Il ricettario dell'anarchico") == []
+    assert calls, "search request was not sent"
+    url, headers = calls[0]
+    assert url == "https://www.amazon.it/s"
+    assert headers.get("Accept-Language", "").startswith("it-IT")
+
+
+def test_get_amazon_domain_falls_back_to_com(monkeypatch):
+    import scripts.cwa_db as cwa_db
+
+    def broken_db():
+        raise RuntimeError("no database")
+
+    monkeypatch.setattr(cwa_db, "CWA_DB", broken_db)
+    assert amazon.get_amazon_domain() == "com"