124 lines
3.8 KiB
Python
124 lines
3.8 KiB
Python
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
|