[books] Fix amazon scrapper to use actual AWS endpoints

vrobbler/apps/books/amazon.py

@@ -1,128 +0,0 @@
-from enum import Enum
-from typing import Optional
-from bs4 import BeautifulSoup
-import requests
-import logging
-
-logger = logging.getLogger(__name__)
-
-USER_AGENT = "Mozilla/5.0 (Android 4.4; Mobile; rv:41.0) Gecko/41.0 Firefox/41.0"
-AMAZON_SEARCH_URL = "https://www.amazon.com/s?k={amazon_id}"
-
-
-class AmazonAttribute(Enum):
-    SERIES = 0
-    PAGES = 1
-    LANGUAGE = 2
-    PUBLISHER = 3
-    PUB_DATE = 4
-    DIMENSIONS = 5
-    ISBN_10 = 6
-    ISBN_13 = 7
-
-
-def strip_and_clean(text):
-    return text.strip("\n").rstrip().lstrip()
-
-
-def get_rating_from_soup(soup) -> Optional[int]:
-    rating = None
-    try:
-        potential_rating = soup.find("div", class_="allmusic-rating")
-        if potential_rating:
-            rating = int(strip_and_clean(potential_rating.get_text()))
-    except ValueError:
-        pass
-    return rating
-
-
-def get_review_from_soup(soup) -> str:
-    review = ""
-    try:
-        potential_text = soup.find("div", class_="text")
-        if potential_text:
-            review = strip_and_clean(potential_text.get_text())
-    except ValueError:
-        pass
-    return review
-
-
-def scrape_data_from_amazon(url) -> dict:
-    data_dict = {}
-    headers = {"User-Agent": USER_AGENT}
-    r = requests.get(url, headers=headers)
-    if r.status_code == 200:
-        soup = BeautifulSoup(r.text, "html.parser")
-        # TODO Fix this scraper
-        data_dict["rating"] = get_rating_from_soup(soup)
-        data_dict["review"] = get_review_from_soup(soup)
-    return data_dict
-
-
-def get_amazon_product_dict(amazon_id: str) -> dict:
-    data_dict = {}
-    url = ""
-
-    search_url = AMAZON_SEARCH_URL.format(amazon_id=amazon_id)
-    headers = {
-        "user-agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
-        "accept-language": "en-GB,en;q=0.9",
-    }
-
-    response = requests.get(search_url, headers=headers)
-
-    if response.status_code != 200:
-        logger.info(f"Bad http response from Amazon {response}")
-        return data_dict
-
-    soup = BeautifulSoup(response.text, "html.parser")
-    results = soup.find("a", class_="a-link-normal")
-
-    if not results:
-        logger.info(f"No search results for {amazon_id}")
-        return data_dict
-
-    product_url = "https://www.amazon.com" + str(results.get("href", ""))
-
-    data_dict = {}
-    response = requests.get(product_url, headers=headers)
-
-    if response.status_code != 200:
-        logger.info(f"Bad http response from Amazon {response}")
-        return data_dict
-
-    soup = BeautifulSoup(response.text, "html.parser")
-    try:
-        data_dict["title"] = soup.findAll("span", class_="celwidget")[1].text.strip()
-        data_dict["cover_url"] = soup.find("img", class_="frontImage").get("src")
-        data_dict["summary"] = soup.findAll("div", class_="a-expander-content")[1].text
-        meta = soup.findAll("div", class_="rpi-attribute-value")
-        data_dict["isbn"] = meta[AmazonAttribute.ISBN_10.value].text.strip()
-        pages = meta[AmazonAttribute.PAGES.value].text
-        if "pages" in pages:
-            data_dict["pages"] = (
-                meta[AmazonAttribute.PAGES.value].text.split("pages")[0].strip()
-            )
-    except IndexError as e:
-        logger.error(f"Amazon lookup is failing for this product {amazon_id}: {e}")
-    except AttributeError as e:
-        logger.error(f"Amazon lookup is failing for this product {amazon_id}: {e}")
-
-    return data_dict
-
-
-def lookup_book_from_amazon(amazon_id: str) -> dict:
-    top = {}
-
-    return {
-        "title": top.get("title"),
-        "isbn": isbn,
-        "openlibrary_id": ol_id,
-        "goodreads_id": get_first("id_goodreads", top),
-        "first_publish_year": top.get("first_publish_year"),
-        "first_sentence": first_sentence,
-        "pages": top.get("number_of_pages_median", None),
-        "cover_url": COVER_URL.format(id=ol_id),
-        "ol_author_id": ol_author_id,
-        "subject_key_list": top.get("subject_key", []),
-    }

vrobbler/apps/books/constants.py

@@ -17,6 +17,7 @@ class MediaSourceTag(str, Enum):
     LOCG = "source_locg"
     KOREADER = "source_koreader"
     SEMANTIC_SCHOLAR = "source_semantic_scholar"
+    AMAZON = "source_amazon"
 
     @classmethod
     def choices(cls):

vrobbler/apps/books/models.py

@@ -23,6 +23,7 @@ from books.sources.comicvine import (
     lookup_issue_by_comicvine_id,
 )
 from books.sources.google import lookup_book_from_google
+from books.sources.amazon import lookup_book_from_amazon
 from books.sources.openlibrary import (
     lookup_book_from_openlibrary as lookup_book_from_ol,
 )
@@ -260,6 +261,7 @@ class Book(LongPlayScrobblableMixin):
         url: str = "",
         enrich: bool = True,
         commit: bool = True,
+        amazon_id: str | None = None,
     ):
         """Given a title, get a Book instance.
 
@@ -321,6 +323,13 @@ class Book(LongPlayScrobblableMixin):
                         book_dict.setdefault(k, v)
                 source_tag = MediaSourceTag.COMICVINE
 
+        # Try Amazon PAAPI as a fallback when given an ASIN
+        if amazon_id and not book_dict:
+            amazon_data = lookup_book_from_amazon(amazon_id)
+            if amazon_data:
+                book_dict.update(amazon_data)
+                source_tag = MediaSourceTag.AMAZON
+
         if not book_dict:
             logger.warning(
                 "No book found in any source, using data as is",

vrobbler/apps/books/sources/amazon.py

@@ -0,0 +1,123 @@
+import logging
+import warnings
+
+from django.conf import settings
+
+with warnings.catch_warnings():
+    warnings.simplefilter("ignore", DeprecationWarning)
+    from amazon_paapi import AmazonApi
+
+logger = logging.getLogger(__name__)
+
+_amazon_client = None
+
+
+def _get_client() -> AmazonApi | None:
+    global _amazon_client
+    if _amazon_client is not None:
+        return _amazon_client
+
+    key = settings.AMAZON_PAAPI_ACCESS_KEY
+    secret = settings.AMAZON_PAAPI_SECRET_KEY
+    tag = settings.AMAZON_PAAPI_ASSOCIATE_TAG
+    country = settings.AMAZON_PAAPI_COUNTRY
+
+    if not all([key, secret, tag]):
+        logger.warning("Amazon PAAPI credentials not configured")
+        return None
+
+    _amazon_client = AmazonApi(key, secret, tag, country)
+    return _amazon_client
+
+
+def lookup_book_from_amazon(asin: str) -> dict:
+    book_dict: dict = {}
+
+    client = _get_client()
+    if not client:
+        return book_dict
+
+    try:
+        items = client.get_items(
+            items=[asin],
+            Condition="New",
+            LanguagesOfPreference=["en_US"],
+        )
+    except Exception as e:
+        logger.warning(f"Amazon PAAPI lookup failed for {asin}: {e}")
+        return book_dict
+
+    if not items:
+        logger.info(f"No Amazon item found for {asin}")
+        return book_dict
+
+    item = items[0]
+    raw = item.to_dict()
+    item_info = raw.get("item_info", {}) or {}
+
+    book_dict["title"] = _get_nested(item_info, "title", "display_value")
+    if not book_dict.get("title"):
+        book_dict["title"] = _get_nested(item_info, "title", "value")
+
+    contributors = _get_nested(item_info, "by_line_info", "contributors") or []
+    authors = [
+        c["name"]
+        for c in contributors
+        if c.get("role", "").lower() in ("author", "artist", "writer")
+    ]
+    if authors:
+        book_dict["authors"] = authors
+
+    publisher = _get_nested(item_info, "by_line_info", "manufacturer")
+    if publisher:
+        book_dict["publisher"] = publisher
+
+    isb_ns = _get_nested(item_info, "external_ids", "isb_ns")
+    if isb_ns and isinstance(isb_ns, list):
+        for isb in isb_ns:
+            if isinstance(isb, dict):
+                if isb.get("type") == "ISBN_13":
+                    book_dict["isbn_13"] = isb.get("value")
+                elif isb.get("type") == "ISBN_10":
+                    book_dict["isbn_10"] = isb.get("value")
+
+    pages_count = _get_nested(item_info, "content_info", "pages_count")
+    if pages_count and isinstance(pages_count, dict):
+        book_dict["pages"] = pages_count.get("value") or pages_count.get("display_value")
+
+    languages = _get_nested(item_info, "content_info", "languages") or []
+    if languages and isinstance(languages, list):
+        lang = languages[0]
+        if isinstance(lang, dict):
+            book_dict["language"] = lang.get("display_value") or lang.get("value")
+
+    pub_date = _get_nested(item_info, "content_info", "publication_date")
+    if not pub_date:
+        pub_date = _get_nested(item_info, "product_info", "release_date")
+    if pub_date and isinstance(pub_date, dict):
+        book_dict["publish_date"] = pub_date.get("display_value") or pub_date.get("value")
+
+    features = item_info.get("features") or []
+    if features and isinstance(features, list):
+        book_dict["summary"] = " ".join(features[:5])
+
+    images = raw.get("images", {}) or {}
+    primary = images.get("primary", {}) or {}
+    for size in ("large", "hi_res", "medium"):
+        candidate = primary.get(size, {}) or {}
+        url = candidate.get("url")
+        if url:
+            book_dict["cover_url"] = url
+            break
+
+    book_dict["detail_page_url"] = raw.get("detail_page_url")
+
+    return book_dict
+
+
+def _get_nested(d: dict, *keys):
+    for key in keys:
+        if not isinstance(d, dict):
+            return None
+        d = d.get(key)
+    return d

vrobbler/settings.py

@@ -82,6 +82,11 @@ TODOIST_CLIENT_SECRET = os.getenv("VROBBLER_TODOIST_CLIENT_SECRET", "")
 GOOGLE_API_KEY = os.getenv("VROBBLER_GOOGLE_API_KEY", "")
 LICHESS_API_KEY = os.getenv("VROBBLER_LICHESS_API_KEY", "")
 
+AMAZON_PAAPI_ACCESS_KEY = os.getenv("VROBBLER_AMAZON_PAAPI_ACCESS_KEY", "")
+AMAZON_PAAPI_SECRET_KEY = os.getenv("VROBBLER_AMAZON_PAAPI_SECRET_KEY", "")
+AMAZON_PAAPI_ASSOCIATE_TAG = os.getenv("VROBBLER_AMAZON_PAAPI_ASSOCIATE_TAG", "")
+AMAZON_PAAPI_COUNTRY = os.getenv("VROBBLER_AMAZON_PAAPI_COUNTRY", "US")
+
 DEFAULT_TASK_CONTEXT_TAGS = [
     "Dev",
     "Home",