From 661bb8c1872210bd28bc04a2efad83cad8307717 Mon Sep 17 00:00:00 2001 From: CREDO23 Date: Sun, 19 Jul 2026 08:19:22 +0200 Subject: [PATCH] feat(walmart): add JSON parsers --- .../proprietary/platforms/walmart/parsers.py | 231 ++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 surfsense_backend/app/proprietary/platforms/walmart/parsers.py diff --git a/surfsense_backend/app/proprietary/platforms/walmart/parsers.py b/surfsense_backend/app/proprietary/platforms/walmart/parsers.py new file mode 100644 index 000000000..9a0d070a1 --- /dev/null +++ b/surfsense_backend/app/proprietary/platforms/walmart/parsers.py @@ -0,0 +1,231 @@ +"""Pure parsers for Walmart's hidden ``__NEXT_DATA__`` JSON. + +Product, listing, and review data all live in the same Next.js state tree; these +functions navigate it defensively (via :func:`.next_data.dig`) and normalize the +fields into the stable output shape. Missing sections yield ``None``/``[]`` so an +isolated schema change never discards an otherwise usable record. +""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import urljoin + +from .next_data import dig, initial_data + +_WALMART_ORIGIN = "https://www.walmart.com" + + +# --------------------------------------------------------------------------- # +# Shared helpers. # +# --------------------------------------------------------------------------- # + + +def _price(node: Any) -> dict[str, Any] | None: + """Normalize a Walmart price node into ``{value, currency}``.""" + if not isinstance(node, dict): + return None + value = node.get("price") + currency = node.get("currencyUnit") + if value is None and currency is None: + return None + return {"value": value, "currency": currency} + + +def _seller(raw: dict[str, Any]) -> dict[str, Any] | None: + name = raw.get("sellerName") or raw.get("sellerDisplayName") + seller_id = raw.get("sellerId") + if not name and not seller_id: + return None + is_walmart = bool(name) and "walmart" in name.lower() + return { + "id": seller_id, + "name": name, + "type": "WALMART" if is_walmart else "MARKETPLACE", + } + + +def _absolute(url: str | None) -> str | None: + return urljoin(_WALMART_ORIGIN, url) if url else None + + +# --------------------------------------------------------------------------- # +# Listing / search cards. # +# --------------------------------------------------------------------------- # + + +def _listing_card(raw: dict[str, Any]) -> dict[str, Any] | None: + """Normalize one search/category result item into a product card.""" + item_id = raw.get("usItemId") or raw.get("id") + name = raw.get("name") + if not item_id or not name: + return None + price_info = raw.get("priceInfo") or {} + availability = raw.get("availabilityStatusV2") or {} + image = raw.get("imageInfo") or {} + return { + "usItemId": str(item_id), + "name": name, + "brand": raw.get("brand"), + "url": _absolute(raw.get("canonicalUrl")), + "price": _price(price_info.get("currentPrice")) + or ({"value": raw.get("price")} if raw.get("price") is not None else None), + "listPrice": _price(price_info.get("wasPrice")), + "stars": raw.get("averageRating"), + "reviewsCount": raw.get("numberOfReviews"), + "seller": _seller(raw), + "availabilityStatus": availability.get("value") + or ("OUT_OF_STOCK" if raw.get("isOutOfStock") else None), + "inStock": (availability.get("value") == "IN_STOCK") + if availability.get("value") + else (raw.get("isOutOfStock") is False if "isOutOfStock" in raw else None), + "thumbnailImage": image.get("thumbnailUrl") or raw.get("image"), + "sponsored": raw.get("isSponsoredFlag"), + } + + +def parse_listing_page(next_data: dict[str, Any]) -> list[dict[str, Any]]: + """Extract normalized product cards from a search/category/browse page.""" + data = initial_data(next_data) + if data is None: + return [] + stacks = dig(data, "searchResult", "itemStacks") + if not isinstance(stacks, list): + return [] + cards: list[dict[str, Any]] = [] + seen: set[str] = set() + for stack in stacks: + for raw in (stack or {}).get("items") or []: + if not isinstance(raw, dict) or raw.get("__typename") not in ( + None, + "Product", + ): + continue + card = _listing_card(raw) + if card and card["usItemId"] not in seen: + seen.add(card["usItemId"]) + cards.append(card) + return cards + + +# --------------------------------------------------------------------------- # +# Product detail. # +# --------------------------------------------------------------------------- # + + +def _images(image_info: dict[str, Any]) -> list[str]: + images: list[str] = [] + for entry in image_info.get("allImages") or []: + url = entry.get("url") if isinstance(entry, dict) else None + if url and url not in images: + images.append(url) + return images + + +def _reviews_sample( + reviews: dict[str, Any] | None, limit: int = 10 +) -> dict[str, Any] | None: + if not isinstance(reviews, dict): + return None + customer = reviews.get("customerReviews") or [] + return { + "averageOverallRating": reviews.get("averageOverallRating"), + "totalReviewCount": reviews.get("totalReviewCount"), + "aspects": reviews.get("aspects") or [], + "topReviews": [ + normalize_review(r) for r in customer[:limit] if isinstance(r, dict) + ], + } + + +def parse_product( + next_data: dict[str, Any], *, url: str, include_reviews_sample: bool = True +) -> dict[str, Any]: + """Extract normalized product fields from a product detail page.""" + data = initial_data(next_data) + product = dig(data, "data", "product") if data else None + if not isinstance(product, dict): + return {} + idml = dig(data, "data", "idml") if data else None + reviews = dig(data, "data", "reviews") if data else None + price_info = product.get("priceInfo") or {} + image_info = product.get("imageInfo") or {} + availability = product.get("availabilityStatus") + + fields: dict[str, Any] = { + "usItemId": str(product.get("usItemId") or product.get("id") or "") or None, + "name": product.get("name"), + "brand": product.get("brand"), + "url": url, + "price": _price(price_info.get("currentPrice")), + "listPrice": _price(price_info.get("wasPrice")), + "currency": dig(price_info, "currentPrice", "currencyUnit"), + "availabilityStatus": availability, + "inStock": availability == "IN_STOCK" if availability else None, + "stars": product.get("averageRating"), + "reviewsCount": product.get("numberOfReviews"), + "seller": _seller(product), + "manufacturerName": product.get("manufacturerName"), + "shortDescription": product.get("shortDescription"), + "longDescription": (idml or {}).get("longDescription") + if isinstance(idml, dict) + else None, + "thumbnailImage": image_info.get("thumbnailUrl"), + "images": _images(image_info), + "category": dig(product, "category", "path"), + "variants": product.get("variantCriteria") or [], + } + if include_reviews_sample: + fields["reviewsSample"] = _reviews_sample(reviews) + return {key: value for key, value in fields.items() if value is not None} + + +# --------------------------------------------------------------------------- # +# Reviews (deep pagination). # +# --------------------------------------------------------------------------- # + + +def normalize_review(raw: dict[str, Any]) -> dict[str, Any]: + """Normalize one Walmart ``customerReviews`` record into a ``ReviewItem`` dict.""" + badges = raw.get("badges") or [] + verified = any( + isinstance(b, dict) and b.get("id") == "VerifiedPurchaser" for b in badges + ) + photos = raw.get("photos") or raw.get("media") or [] + images: list[str] = [] + for photo in photos: + if not isinstance(photo, dict): + continue + url = photo.get("normalUrl") or dig(photo, "sizes", "normal", "url") + if url and url not in images: + images.append(url) + responses = raw.get("clientResponses") or [] + seller_response = None + if responses and isinstance(responses[0], dict): + seller_response = responses[0].get("response") or responses[0].get( + "responseText" + ) + return { + "reviewId": raw.get("reviewId"), + "rating": raw.get("rating"), + "title": raw.get("reviewTitle"), + "text": raw.get("reviewText"), + "submissionTime": raw.get("reviewSubmissionTime"), + "author": raw.get("userNickname"), + "verifiedPurchase": verified, + "positiveFeedback": raw.get("positiveFeedback"), + "negativeFeedback": raw.get("negativeFeedback"), + "images": images, + "syndicated": bool(raw.get("syndicationSource")), + "sellerResponse": seller_response, + } + + +def parse_reviews_page(next_data: dict[str, Any]) -> list[dict[str, Any]]: + """Extract normalized reviews from one ``/reviews/product/{id}`` page.""" + data = initial_data(next_data) + reviews = dig(data, "data", "reviews") if data else None + customer = reviews.get("customerReviews") if isinstance(reviews, dict) else None + if not isinstance(customer, list): + return [] + return [normalize_review(r) for r in customer if isinstance(r, dict)]