| 1 | """Amazon product and review signals via the Bright Data CLI. |
| 2 | |
| 3 | Two-stage source, following the digg discover-then-enrich shape: |
| 4 | |
| 5 | 1. **Discovery** -- one ``amazon_product_search`` per run turns a |
| 6 | model-supplied product keyword into product records carrying live |
| 7 | aggregate stats (rating, rating count, price). Cheap and fast. |
| 8 | 2. **Enrichment** -- ``amazon_product_reviews`` pulls a capped sample of |
| 9 | written reviews for the top few surviving products, in parallel, under |
| 10 | a lane deadline. Reviews ride on their product item as metadata |
| 11 | comments and feed community-voice weaving. |
| 12 | |
| 13 | The signature signal is the fusion of those two: an all-time rating from |
| 14 | thousands of ratings, set against the average of just the reviews inside |
| 15 | the last-30-day window. When those disagree, something changed this month, |
| 16 | and the review text says what. No Amazon page shows that. |
| 17 | |
| 18 | Metering (R13): one credit per pipeline request regardless of records |
| 19 | returned, so the caps here bound paid-tier *records*, not credits. A |
| 20 | default run is 1 search + up to 3 review pulls = 4 requests. |
| 21 | |
| 22 | Field names and quirks below are verified against live payloads pulled |
| 23 | 2026-08-13; see the plan's schema block. Three fields arrive doubled |
| 24 | (``review_posted_date``, ``review_header``, ``badge``) and are repaired |
| 25 | here rather than downstream. |
| 26 | """ |
| 27 | |
| 28 | from __future__ import annotations |
| 29 | |
| 30 | import re |
| 31 | import time |
| 32 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 33 | from datetime import datetime, timezone |
| 34 | from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple |
| 35 | from urllib.parse import urlparse |
| 36 | |
| 37 | from . import brightdata, log |
| 38 | from .relevance import token_overlap_relevance |
| 39 | |
| 40 | |
| 41 | SEARCH_PIPELINE = "amazon_product_search" |
| 42 | REVIEWS_PIPELINE = "amazon_product_reviews" |
| 43 | |
| 44 | DEFAULT_DOMAIN = "https://www.amazon.com" |
| 45 | |
| 46 | # Reviews requested per pull. Uniform across topic shapes and depths by |
| 47 | # decision: billing is per *request*, not per record, so a bigger cap is |
| 48 | # free on the monthly credit tier, and the in-window sample is what the |
| 49 | # drift signal rests on. Live-verified that latency does not scale with |
| 50 | # this number (50 reviews in 22s vs 20 reviews in 115s on a slower SKU). |
| 51 | # |
| 52 | # It is a ceiling, never a quota -- a SKU with 31 total reviews returns 31. |
| 53 | MAX_REVIEWS = 50 |
| 54 | |
| 55 | # How many products get a review pull, per depth. Quick spends one credit |
| 56 | # on discovery only: aggregate stats with no recent window. |
| 57 | DEPTH_CONFIG = { |
| 58 | "quick": 0, |
| 59 | "default": 3, |
| 60 | "deep": 5, |
| 61 | } |
| 62 | |
| 63 | SEARCH_TIMEOUT = 90 |
| 64 | REVIEW_TIMEOUT = 180 |
| 65 | |
| 66 | # Wall-clock ceiling for the whole parallel review lane. Pulls that miss it |
| 67 | # are abandoned, and their products degrade to `quiet` rather than |
| 68 | # disappearing (a slow SKU is real and unrelated to the cap: one live pull |
| 69 | # took 115s). |
| 70 | LANE_DEADLINE = 180 |
| 71 | |
| 72 | # The engine's foreground contract. The lane deadline is clamped against |
| 73 | # whatever remains of it, minus room to render. |
| 74 | FOREGROUND_CONTRACT = 300 |
| 75 | RENDER_MARGIN = 20 |
| 76 | |
| 77 | # Minimum useful budget for the review lane. Below this threshold, Bright |
| 78 | # Data pulls reliably time out (cli_timeout = max(5, timeout-10), so budget |
| 79 | # 11s → CLI timeout 1s). Crumbs are a skip, not a short timeout: firing |
| 80 | # doomed pulls still spends 3 credits with no reviews returned. |
| 81 | MIN_USEFUL_REVIEW_BUDGET = 90 |
| 82 | |
| 83 | # Minimum dated reviews inside the window before a drift arrow is honest. |
| 84 | # Live census: a 50-cap pull returned 31 records of which only 5 were |
| 85 | # inside 30 days, so an unguarded arrow would routinely publish a "trend" |
| 86 | # computed from one or two reviews. |
| 87 | MIN_DRIFT_SAMPLE = 5 |
| 88 | |
| 89 | RECENT_WINDOW_DAYS = 30 |
| 90 | |
| 91 | # Product names run long and pipe-delimited; the footer needs a scannable |
| 92 | # handle, not a title. |
| 93 | SHORT_NAME_MAX = 18 |
| 94 | |
| 95 | _STAR_FIELDS = ( |
| 96 | ("one_star", 1), |
| 97 | ("two_star", 2), |
| 98 | ("three_star", 3), |
| 99 | ("four_star", 4), |
| 100 | ("five_star", 5), |
| 101 | ) |
| 102 | |
| 103 | |
| 104 | def _log(msg: str) -> None: |
| 105 | log.source_log("Amazon", msg, tty_only=False) |
| 106 | |
| 107 | |
| 108 | def _today() -> datetime: |
| 109 | return datetime.now(timezone.utc) |
| 110 | |
| 111 | |
| 112 | # --------------------------------------------------------------- parsing |
| 113 | |
| 114 | |
| 115 | def undouble(text: str) -> str: |
| 116 | """Repair the CLI's doubled string fields. |
| 117 | |
| 118 | Observed live: ``review_header`` arrives as ``"Best Box!Best Box!"`` and |
| 119 | ``badge`` as ``"Verified Purchase, Verified Purchase"``. Handles the |
| 120 | exact-repeat case and the comma-joined repeat, and leaves anything else |
| 121 | untouched -- a genuinely repetitive title must survive intact. |
| 122 | """ |
| 123 | value = (text or "").strip() |
| 124 | if not value: |
| 125 | return "" |
| 126 | half, odd = divmod(len(value), 2) |
| 127 | # Only treat an exact repeat as doubling when the halves are substantial |
| 128 | # and look like a phrase rather than a syllable -- otherwise a real title |
| 129 | # of "ByeBye" or "NoNo" gets silently truncated to half of itself. The |
| 130 | # observed artifact doubles whole headlines, so requiring some length and |
| 131 | # either whitespace or terminal punctuation keeps the repair targeted. |
| 132 | if not odd and half >= 6 and value[:half] == value[half:]: |
| 133 | first = value[:half] |
| 134 | if " " in first or first[-1] in ".!?": |
| 135 | return first.strip() |
| 136 | parts = [p.strip() for p in value.split(",")] |
| 137 | if len(parts) == 2 and parts[0] and parts[0] == parts[1]: |
| 138 | return parts[0] |
| 139 | return value |
| 140 | |
| 141 | |
| 142 | _DATE_HEAD = re.compile(r"^([A-Z][a-z]+ \d{1,2}, \d{4})") |
| 143 | |
| 144 | |
| 145 | def parse_review_date(raw: Any) -> Optional[str]: |
| 146 | """Pull the ISO date out of the CLI's prose-wrapped date field. |
| 147 | |
| 148 | Live shape: ``"August 3, 2026Reviewed in the United States on August 3, |
| 149 | 2026"``. Only the leading ``%B %d, %Y`` is trustworthy; the tail is |
| 150 | localized prose that varies by marketplace. |
| 151 | |
| 152 | Returns ``YYYY-MM-DD`` or None. |
| 153 | """ |
| 154 | match = _DATE_HEAD.match(str(raw or "").strip()) |
| 155 | if not match: |
| 156 | return None |
| 157 | try: |
| 158 | return datetime.strptime(match.group(1), "%B %d, %Y").date().isoformat() |
| 159 | except ValueError: |
| 160 | return None |
| 161 | |
| 162 | |
| 163 | def short_name(name: str, brand: str = "") -> str: |
| 164 | """Derive a scannable footer handle from a long product name. |
| 165 | |
| 166 | Live names are pipe-delimited marketing strings with the brand carried |
| 167 | in its own field rather than as a prefix ("Chill Max Leak-Proof XL |
| 168 | Bento-Style Lunch Box | Included Ice Pack Keeps Food Cold"). Take the |
| 169 | segment before the first delimiter, drop a leading brand token if one |
| 170 | did sneak in, and clip to a scannable width on a word boundary. |
| 171 | """ |
| 172 | text = re.split(r"[|(–—]", str(name or ""), maxsplit=1)[0].strip(" -,") |
| 173 | brand_token = str(brand or "").strip() |
| 174 | if brand_token: |
| 175 | # Word-boundary anchored: a bare startswith() eats into sub-brands and |
| 176 | # coincidental prefixes ("AnkerWork" under brand "Anker" would become |
| 177 | # "Work", "Chillax" under "Chill" would become "ax"). |
| 178 | stripped = re.sub( |
| 179 | rf"^{re.escape(brand_token)}\b[\s\-,]*", "", text, count=1, flags=re.IGNORECASE |
| 180 | ) |
| 181 | if stripped: |
| 182 | text = stripped.strip(" -,") |
| 183 | if len(text) <= SHORT_NAME_MAX: |
| 184 | return text |
| 185 | clipped = text[:SHORT_NAME_MAX].rsplit(" ", 1)[0].strip(" -,") |
| 186 | return clipped or text[:SHORT_NAME_MAX].strip() |
| 187 | |
| 188 | |
| 189 | def _as_float(value: Any) -> Optional[float]: |
| 190 | try: |
| 191 | result = float(value) |
| 192 | except (TypeError, ValueError): |
| 193 | return None |
| 194 | return result |
| 195 | |
| 196 | |
| 197 | def _as_int(value: Any) -> int: |
| 198 | try: |
| 199 | return int(value) |
| 200 | except (TypeError, ValueError): |
| 201 | return 0 |
| 202 | |
| 203 | |
| 204 | def _is_sponsored(value: Any) -> bool: |
| 205 | """The flag arrives as the string 'true'/'false', not a bool. |
| 206 | |
| 207 | Recorded in metadata but never used to filter (R4): its distribution |
| 208 | swings hard with keyword phrasing, so filtering on it can blank the |
| 209 | lane on an unlucky query. |
| 210 | """ |
| 211 | if isinstance(value, bool): |
| 212 | return value |
| 213 | return str(value or "").strip().lower() == "true" |
| 214 | |
| 215 | |
| 216 | def _valid_product_url(url: str, domain: str) -> bool: |
| 217 | """Accept only https URLs on the configured Amazon host.""" |
| 218 | try: |
| 219 | parsed = urlparse(url) |
| 220 | expected = urlparse(domain or DEFAULT_DOMAIN) |
| 221 | except ValueError: |
| 222 | return False |
| 223 | if parsed.scheme != "https" or not parsed.netloc: |
| 224 | return False |
| 225 | host = parsed.netloc.lower().removeprefix("www.") |
| 226 | want = (expected.netloc or "").lower().removeprefix("www.") |
| 227 | return bool(want) and host == want |
| 228 | |
| 229 | |
| 230 | # Amazon ASINs are a fixed shape. Validating it matters because the value |
| 231 | # is interpolated into a URL that is then refetched through the CLI *and* |
| 232 | # rendered as a link in the report -- two sinks, one unvalidated API field. |
| 233 | _ASIN_RE = re.compile(r"^[A-Za-z0-9]{10}$") |
| 234 | |
| 235 | |
| 236 | def _valid_asin(asin: str) -> bool: |
| 237 | return bool(_ASIN_RE.match(asin or "")) |
| 238 | |
| 239 | |
| 240 | def canonical_product_url(url: str, asin: str, domain: str) -> str: |
| 241 | """Strip Amazon's tracking tail down to a stable /dp/<asin> link. |
| 242 | |
| 243 | Search records carry 200+ character URLs with session-scoped ``dib`` |
| 244 | tokens. Those work but are unreadable in a report and unstable across |
| 245 | runs, which breaks dedupe on re-runs of the same topic. |
| 246 | |
| 247 | Falls back to the (already host-validated) original URL if the ASIN is |
| 248 | not well-formed, so a malformed record can never shape the rebuilt URL. |
| 249 | """ |
| 250 | if not _valid_asin(asin): |
| 251 | return url |
| 252 | base = (domain or DEFAULT_DOMAIN).rstrip("/") |
| 253 | return f"{base}/dp/{asin}" |
| 254 | |
| 255 | |
| 256 | # ------------------------------------------------------------- discovery |
| 257 | |
| 258 | |
| 259 | def search_products( |
| 260 | keyword: str, |
| 261 | *, |
| 262 | domain: str = DEFAULT_DOMAIN, |
| 263 | config: Optional[Dict[str, Any]] = None, |
| 264 | timeout: int = SEARCH_TIMEOUT, |
| 265 | ) -> Dict[str, Any]: |
| 266 | """Run one product search. Never raises; returns the adapter envelope.""" |
| 267 | query = (keyword or "").strip() |
| 268 | if not query: |
| 269 | return {"records": []} |
| 270 | # A leading dash would be parsed as a CLI option rather than a search |
| 271 | # term. The keyword is model-supplied and can be influenced by |
| 272 | # pre-research over untrusted web content, so reject rather than |
| 273 | # sanitize -- a keyword starting with '-' is never a real product. |
| 274 | if query.startswith("-"): |
| 275 | _log(f"rejecting option-shaped keyword: {query!r}") |
| 276 | return {"records": [], "error": "amazon keyword may not begin with '-'"} |
| 277 | _log(f"search '{query}' on {domain}") |
| 278 | response = brightdata.run_pipeline( |
| 279 | SEARCH_PIPELINE, [query, domain or DEFAULT_DOMAIN], |
| 280 | timeout=timeout, config=config, |
| 281 | ) |
| 282 | if response.get("error"): |
| 283 | _log(f"search failed: {response['error']}") |
| 284 | else: |
| 285 | _log(f"search returned {len(response.get('records') or [])} records") |
| 286 | return response |
| 287 | |
| 288 | |
| 289 | def parse_search_response( |
| 290 | response: Dict[str, Any], |
| 291 | keyword: str, |
| 292 | *, |
| 293 | domain: str = DEFAULT_DOMAIN, |
| 294 | min_relevance: float = 0.15, |
| 295 | ) -> List[Dict[str, Any]]: |
| 296 | """Turn raw search records into deduped, relevance-gated product dicts. |
| 297 | |
| 298 | Dedupe is by ASIN: live payloads repeat a single product up to five |
| 299 | times across the result set (64 unique of 66 records on one pull). |
| 300 | Relevance is scored against the *supplied keyword*, not the run topic, |
| 301 | because the model may search "June Oven" on a topic about a person. |
| 302 | """ |
| 303 | records = response.get("records") if isinstance(response, dict) else None |
| 304 | if not isinstance(records, list): |
| 305 | return [] |
| 306 | |
| 307 | today = _today().date().isoformat() |
| 308 | seen: Dict[str, Dict[str, Any]] = {} |
| 309 | for record in records: |
| 310 | if not isinstance(record, dict): |
| 311 | continue |
| 312 | asin = str(record.get("asin") or "").strip() |
| 313 | raw_url = str(record.get("url") or "").strip() |
| 314 | if not _valid_asin(asin) or not _valid_product_url(raw_url, domain): |
| 315 | continue |
| 316 | |
| 317 | name = str(record.get("name") or "").strip() |
| 318 | brand = str(record.get("brand") or "").strip() |
| 319 | if not name: |
| 320 | continue |
| 321 | |
| 322 | relevance = token_overlap_relevance(keyword, f"{brand} {name}".strip()) |
| 323 | if relevance < min_relevance: |
| 324 | continue |
| 325 | |
| 326 | num_ratings = _as_int(record.get("num_ratings")) |
| 327 | existing = seen.get(asin) |
| 328 | # Duplicates of one ASIN can disagree on rating count (variant-level |
| 329 | # records); keep the richest. |
| 330 | if existing and _as_int(existing.get("num_ratings")) >= num_ratings: |
| 331 | continue |
| 332 | |
| 333 | seen[asin] = { |
| 334 | "asin": asin, |
| 335 | # Current-date stamped (KTD6, trustpilot precedent): a live |
| 336 | # aggregate rating is a fact about now, not about the product's |
| 337 | # launch date, so it must not be dropped by the 30-day filter. |
| 338 | "date": today, |
| 339 | "name": name, |
| 340 | "short_name": short_name(name, brand), |
| 341 | "brand": brand, |
| 342 | "url": canonical_product_url(raw_url, asin, domain), |
| 343 | "rating": _as_float(record.get("rating")), |
| 344 | "num_ratings": num_ratings, |
| 345 | "price": _as_float(record.get("final_price")), |
| 346 | "currency": str(record.get("currency") or "").strip(), |
| 347 | "badge": undouble(str(record.get("badge") or "")), |
| 348 | "sponsored": _is_sponsored(record.get("sponsored")), |
| 349 | "bought_past_month": _as_int(record.get("bought_past_month")), |
| 350 | "rank_on_page": _as_int(record.get("rank_on_page")), |
| 351 | "relevance": relevance, |
| 352 | } |
| 353 | |
| 354 | products = sorted( |
| 355 | seen.values(), |
| 356 | key=lambda p: (p["num_ratings"], p["relevance"]), |
| 357 | reverse=True, |
| 358 | ) |
| 359 | _log(f"{len(products)} unique on-keyword products after dedupe") |
| 360 | return products |
| 361 | |
| 362 | |
| 363 | def infer_brand(products: Sequence[Dict[str, Any]], keyword: str) -> str: |
| 364 | """Detect a brand topic by matching record brands against the keyword. |
| 365 | |
| 366 | This is the guard against paying to review a competitor. Rival brands |
| 367 | buy ads against a brand keyword and can outrank the brand's own catalog |
| 368 | on raw rating count: on a live "bentgo lunch box" search a competitor |
| 369 | held the top two slots and would have taken two of the three review |
| 370 | pulls, putting a rival's reviews in a Bentgo report. |
| 371 | |
| 372 | Matching the *keyword's own tokens*, rather than picking the most |
| 373 | common brand in the results, is what keeps category topics unfiltered. |
| 374 | "best bluetooth speaker" names no brand, so nothing is constrained and |
| 375 | the top products across brands compete on merit -- which is exactly |
| 376 | what that topic shape wants. |
| 377 | """ |
| 378 | normalized_keyword = " ".join(re.findall(r"[a-z0-9]+", (keyword or "").lower())) |
| 379 | if not normalized_keyword: |
| 380 | return "" |
| 381 | keyword_tokens = set(normalized_keyword.split()) |
| 382 | |
| 383 | # Keyed by the lowercased brand so one vendor spelled two ways ("Bentgo" |
| 384 | # and "BENTGO" in the same result set) reads as one candidate. Without |
| 385 | # this the set has two members, the function bails, and the guard it |
| 386 | # exists to provide silently turns off. |
| 387 | candidates: Dict[str, str] = {} |
| 388 | for product in products: |
| 389 | brand = str(product.get("brand") or "").strip() |
| 390 | if not brand: |
| 391 | continue |
| 392 | brand_tokens = re.findall(r"[a-z0-9]+", brand.lower()) |
| 393 | if not brand_tokens: |
| 394 | continue |
| 395 | # Multi-word brands ("Hydro Flask") can never match a single-token |
| 396 | # test, so compare the brand's whole token sequence against the |
| 397 | # keyword's -- otherwise the guard is off for every two-word brand. |
| 398 | if len(brand_tokens) == 1: |
| 399 | matched = brand_tokens[0] in keyword_tokens and len(brand_tokens[0]) > 2 |
| 400 | else: |
| 401 | matched = " ".join(brand_tokens) in normalized_keyword |
| 402 | if matched: |
| 403 | # First spelling wins, so the result is deterministic across runs. |
| 404 | candidates.setdefault(brand.lower(), brand) |
| 405 | return next(iter(candidates.values())) if len(candidates) == 1 else "" |
| 406 | |
| 407 | |
| 408 | def select_enrichment_targets( |
| 409 | products: Sequence[Dict[str, Any]], |
| 410 | *, |
| 411 | limit: int, |
| 412 | brand: str = "", |
| 413 | keyword: str = "", |
| 414 | ) -> List[Dict[str, Any]]: |
| 415 | """Pick which products get a review pull. |
| 416 | |
| 417 | Ranked by rating count, which is a coarse signal: search records carry |
| 418 | variant-level counts that can undercount badly (84 on a record whose |
| 419 | review pull reported 8,446). The review pull's own |
| 420 | ``product_rating_count`` is authoritative once available. |
| 421 | |
| 422 | Two filters run before the cut: |
| 423 | |
| 424 | * **Brand**, supplied or inferred from the keyword (see ``infer_brand``). |
| 425 | The record's own ``brand`` field does the work, which also solves |
| 426 | accessory contamination outright -- a "grill brush for Weber" carries |
| 427 | the brush maker's brand, not Weber. A front-anchored name match covers |
| 428 | the few records where ``brand`` is null. |
| 429 | * **Variant collapse.** Live results repeat one product across colors |
| 430 | and sizes under distinct ASINs with near-identical names. Two of those |
| 431 | would burn two of three pulls on the same product and render as |
| 432 | duplicate footer entries, so only the best-ranked of each short-name |
| 433 | group stays eligible. |
| 434 | """ |
| 435 | if limit <= 0: |
| 436 | return [] |
| 437 | pool = list(products) |
| 438 | |
| 439 | wanted = (brand or "").strip().lower() or infer_brand(pool, keyword).lower() |
| 440 | if wanted: |
| 441 | matched = [ |
| 442 | p for p in pool |
| 443 | if (p.get("brand") or "").strip().lower() == wanted |
| 444 | or (not (p.get("brand") or "").strip() |
| 445 | and str(p.get("name") or "").strip().lower().startswith(wanted)) |
| 446 | ] |
| 447 | if matched: |
| 448 | pool = matched |
| 449 | |
| 450 | deduped: List[Dict[str, Any]] = [] |
| 451 | seen_names: set[str] = set() |
| 452 | for product in pool: |
| 453 | key = (product.get("short_name") or "").strip().lower() |
| 454 | if key and key in seen_names: |
| 455 | continue |
| 456 | if key: |
| 457 | seen_names.add(key) |
| 458 | deduped.append(product) |
| 459 | return deduped[:limit] |
| 460 | |
| 461 | |
| 462 | # ------------------------------------------------------------ enrichment |
| 463 | |
| 464 | |
| 465 | def fetch_reviews( |
| 466 | product_url: str, |
| 467 | *, |
| 468 | max_reviews: int = MAX_REVIEWS, |
| 469 | config: Optional[Dict[str, Any]] = None, |
| 470 | timeout: int = REVIEW_TIMEOUT, |
| 471 | ) -> Dict[str, Any]: |
| 472 | """Pull a capped review sample for one product. Never raises.""" |
| 473 | if not product_url: |
| 474 | return {"records": []} |
| 475 | return brightdata.run_pipeline( |
| 476 | REVIEWS_PIPELINE, [product_url, str(max_reviews)], |
| 477 | timeout=timeout, config=config, |
| 478 | ) |
| 479 | |
| 480 | |
| 481 | def parse_reviews(response: Dict[str, Any]) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]: |
| 482 | """Split a review payload into comment dicts and product-level stats. |
| 483 | |
| 484 | Product-level fields (``product_rating``, ``product_rating_count``, the |
| 485 | ``product_rating_object`` star distribution) ride on *every* review |
| 486 | record, so they are read off the first one. |
| 487 | |
| 488 | Comments are built directly in the shared score/excerpt shape rather |
| 489 | than routed through ``normalize._remap_comments``, which strips keys it |
| 490 | does not know -- and rating, date, and verified are exactly the keys |
| 491 | this source needs to keep. Sorted newest first so the woven sample |
| 492 | favors recent voices. |
| 493 | """ |
| 494 | records = response.get("records") if isinstance(response, dict) else None |
| 495 | if not isinstance(records, list) or not records: |
| 496 | return [], {} |
| 497 | |
| 498 | first = records[0] |
| 499 | distribution = first.get("product_rating_object") |
| 500 | stats: Dict[str, Any] = { |
| 501 | "product_rating": _as_float(first.get("product_rating")), |
| 502 | "product_rating_count": _as_int(first.get("product_rating_count")), |
| 503 | "star_distribution": distribution if isinstance(distribution, dict) else {}, |
| 504 | } |
| 505 | |
| 506 | comments: List[Dict[str, Any]] = [] |
| 507 | for record in records: |
| 508 | if not isinstance(record, dict): |
| 509 | continue |
| 510 | body = str(record.get("review_text") or "").strip() |
| 511 | header = undouble(str(record.get("review_header") or "")) |
| 512 | excerpt = body or header |
| 513 | if not excerpt: |
| 514 | continue |
| 515 | comments.append( |
| 516 | { |
| 517 | # Shared comment shape: downstream weaving reads score/excerpt. |
| 518 | "score": _as_int(record.get("helpful_count")), |
| 519 | "excerpt": excerpt, |
| 520 | "author": str(record.get("author_name") or "").strip(), |
| 521 | "rating": _as_int(record.get("rating")), |
| 522 | "date": parse_review_date(record.get("review_posted_date")), |
| 523 | "verified": bool(record.get("is_verified")), |
| 524 | "vine": bool(record.get("is_amazon_vine")), |
| 525 | "title": header, |
| 526 | } |
| 527 | ) |
| 528 | |
| 529 | # Newest first; undated records sink rather than disappear (R2a). |
| 530 | comments.sort(key=lambda c: (c["date"] or "", c["score"]), reverse=True) |
| 531 | return comments, stats |
| 532 | |
| 533 | |
| 534 | def _remaining_lane_budget(elapsed: float) -> int: |
| 535 | """Compute the review lane's wall-clock budget. |
| 536 | |
| 537 | Returns the lesser of LANE_DEADLINE and whatever remains of the foreground |
| 538 | contract. If the remaining time is below MIN_USEFUL_REVIEW_BUDGET, returns |
| 539 | 0 (skip the lane entirely) rather than firing doomed short pulls that spend |
| 540 | Bright Data credits without returning reviews. |
| 541 | """ |
| 542 | remaining = FOREGROUND_CONTRACT - elapsed - RENDER_MARGIN |
| 543 | if remaining < MIN_USEFUL_REVIEW_BUDGET: |
| 544 | return 0 |
| 545 | return int(min(LANE_DEADLINE, remaining)) |
| 546 | |
| 547 | |
| 548 | def enrich_with_reviews( |
| 549 | products: Sequence[Dict[str, Any]], |
| 550 | *, |
| 551 | depth: str = "default", |
| 552 | config: Optional[Dict[str, Any]] = None, |
| 553 | elapsed: float = 0.0, |
| 554 | max_reviews: int = MAX_REVIEWS, |
| 555 | brand: str = "", |
| 556 | keyword: str = "", |
| 557 | fetcher=None, |
| 558 | ) -> Tuple[List[Dict[str, Any]], Optional[str]]: |
| 559 | """Attach review samples to the top products, in parallel, under a deadline. |
| 560 | |
| 561 | Every product is returned either way. A product whose pull is dropped |
| 562 | by the deadline keeps its search-record stats and simply carries no |
| 563 | review sample -- it renders as ``quiet`` rather than vanishing, because |
| 564 | losing a top product entirely is a worse failure than losing its |
| 565 | recent-window read. The dropped pull has spent its credit regardless. |
| 566 | |
| 567 | Returns (enriched_products, status_detail). status_detail is None when |
| 568 | enrichment succeeded normally, or a string describing a degraded outcome: |
| 569 | - ``"review lane skipped (budget 0s)"`` -- crumb budget, lane did not run |
| 570 | - ``"review lane timed out"`` -- all pulls dropped by the deadline |
| 571 | """ |
| 572 | enriched = [dict(p) for p in products] |
| 573 | pull_count = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 574 | if pull_count <= 0: |
| 575 | _log(f"depth={depth}: discovery only, no review pulls") |
| 576 | return enriched, None |
| 577 | |
| 578 | budget = _remaining_lane_budget(elapsed) |
| 579 | if budget <= 0: |
| 580 | _log(f"review lane skipped (budget {budget}s, floor {MIN_USEFUL_REVIEW_BUDGET}s)") |
| 581 | return enriched, "review lane skipped (budget 0s)" |
| 582 | |
| 583 | targets = select_enrichment_targets( |
| 584 | enriched, limit=pull_count, brand=brand, keyword=keyword |
| 585 | ) |
| 586 | if not targets: |
| 587 | return enriched, None |
| 588 | |
| 589 | by_asin = {p["asin"]: p for p in enriched} |
| 590 | pull = fetcher or ( |
| 591 | lambda url: fetch_reviews( |
| 592 | url, max_reviews=max_reviews, config=config, |
| 593 | timeout=min(REVIEW_TIMEOUT, budget), |
| 594 | ) |
| 595 | ) |
| 596 | |
| 597 | _log(f"pulling up to {max_reviews} reviews for {len(targets)} products (budget {budget}s)") |
| 598 | started = time.monotonic() |
| 599 | completed_count = 0 |
| 600 | dropped_count = 0 |
| 601 | # Not a `with` block on purpose. Every future is already running (one |
| 602 | # worker per target), so `future.cancel()` can never succeed, and |
| 603 | # ThreadPoolExecutor's context-manager exit calls shutdown(wait=True) -- |
| 604 | # which would block on the very straggler the deadline just declared |
| 605 | # dropped, making the deadline advisory rather than real. Shutting down |
| 606 | # without waiting lets the abandoned thread finish and discard its result |
| 607 | # in the background while the run proceeds. |
| 608 | pool = ThreadPoolExecutor(max_workers=max(1, len(targets))) |
| 609 | try: |
| 610 | futures = {pool.submit(pull, t["url"]): t["asin"] for t in targets} |
| 611 | try: |
| 612 | for future in as_completed(futures, timeout=budget): |
| 613 | asin = futures[future] |
| 614 | try: |
| 615 | response = future.result() |
| 616 | except Exception as exc: # never let one pull kill siblings |
| 617 | _log(f"review pull failed for {asin}: {exc}") |
| 618 | continue |
| 619 | if response.get("error"): |
| 620 | _log(f"review pull error for {asin}: {response['error']}") |
| 621 | continue |
| 622 | comments, stats = parse_reviews(response) |
| 623 | product = by_asin.get(asin) |
| 624 | if product is None: |
| 625 | continue |
| 626 | product["top_comments"] = comments |
| 627 | product.update({k: v for k, v in stats.items() if v}) |
| 628 | completed_count += 1 |
| 629 | except TimeoutError: |
| 630 | dropped_count = sum(1 for f in futures if not f.done()) |
| 631 | _log(f"lane deadline {budget}s hit; dropped {dropped_count} straggling pull(s)") |
| 632 | finally: |
| 633 | pool.shutdown(wait=False, cancel_futures=True) |
| 634 | |
| 635 | _log(f"review lane finished in {time.monotonic() - started:.0f}s") |
| 636 | |
| 637 | # Report degraded outcome if all pulls dropped (none completed) |
| 638 | status_detail = None |
| 639 | if completed_count == 0 and dropped_count > 0: |
| 640 | status_detail = "review lane timed out" |
| 641 | |
| 642 | return enriched, status_detail |
| 643 | |
| 644 | |
| 645 | def enrich_source_items( |
| 646 | items: List[Any], |
| 647 | *, |
| 648 | depth: str = "default", |
| 649 | config: Optional[Dict[str, Any]] = None, |
| 650 | keyword: str = "", |
| 651 | elapsed: float = 0.0, |
| 652 | max_reviews: int = MAX_REVIEWS, |
| 653 | fetcher=None, |
| 654 | ) -> List[Any]: |
| 655 | """Attach review samples to the amazon SourceItems that survived dedupe. |
| 656 | |
| 657 | Reads product identity out of ``metadata`` and writes ``top_comments`` |
| 658 | plus the computed stat block back into it, in place. Runs from |
| 659 | ``pipeline._finalize_items_by_source`` so the review budget is spent on |
| 660 | the products the brief will actually show, not on the top of the raw |
| 661 | fanout (the digg enrichment precedent). |
| 662 | """ |
| 663 | products: List[Dict[str, Any]] = [] |
| 664 | by_asin: Dict[str, Any] = {} |
| 665 | for item in items: |
| 666 | if getattr(item, "source", None) != "amazon": |
| 667 | continue |
| 668 | metadata = getattr(item, "metadata", None) or {} |
| 669 | asin = str(metadata.get("asin") or "").strip() |
| 670 | if not asin or metadata.get("top_comments"): |
| 671 | continue |
| 672 | products.append( |
| 673 | { |
| 674 | "asin": asin, |
| 675 | "url": getattr(item, "url", "") or metadata.get("url", ""), |
| 676 | "name": metadata.get("name") or getattr(item, "title", ""), |
| 677 | "short_name": metadata.get("short_name") or "", |
| 678 | "brand": metadata.get("brand") or "", |
| 679 | "num_ratings": metadata.get("num_ratings") or 0, |
| 680 | "rating": metadata.get("rating"), |
| 681 | } |
| 682 | ) |
| 683 | by_asin[asin] = item |
| 684 | |
| 685 | if not products: |
| 686 | return items |
| 687 | |
| 688 | enriched, _status = enrich_with_reviews( |
| 689 | products, depth=depth, config=config, elapsed=elapsed, |
| 690 | max_reviews=max_reviews, keyword=keyword, fetcher=fetcher, |
| 691 | ) |
| 692 | for product in enriched: |
| 693 | item = by_asin.get(product["asin"]) |
| 694 | if item is None: |
| 695 | continue |
| 696 | metadata = getattr(item, "metadata", None) |
| 697 | if metadata is None: |
| 698 | continue |
| 699 | if product.get("top_comments"): |
| 700 | metadata["top_comments"] = product["top_comments"] |
| 701 | stats = product_stats(product) |
| 702 | metadata["stats"] = stats |
| 703 | # The review pull's product_rating_count supersedes the search |
| 704 | # record's, which is variant-level and can undercount by orders of |
| 705 | # magnitude (84 on a record whose pull reported 8,446). Normalization |
| 706 | # ran before enrichment, so refresh the surfaces that already baked |
| 707 | # the old number in -- otherwise one product shows two different |
| 708 | # rating counts in the same report. |
| 709 | for key in ("product_rating", "product_rating_count", "star_distribution"): |
| 710 | if product.get(key): |
| 711 | metadata[key] = product[key] |
| 712 | authoritative = stats.get("ratings_total") or 0 |
| 713 | if authoritative and getattr(item, "engagement", None) is not None: |
| 714 | item.engagement["ratings"] = authoritative |
| 715 | metadata["num_ratings"] = authoritative |
| 716 | _refresh_title(item, stats) |
| 717 | return items |
| 718 | |
| 719 | |
| 720 | def _refresh_title(item: Any, stats: Dict[str, Any]) -> None: |
| 721 | """Rewrite the trailing "- 4.4/5 (N ratings)" headline after enrichment.""" |
| 722 | title = getattr(item, "title", "") or "" |
| 723 | rating = stats.get("all_time") |
| 724 | total = stats.get("ratings_total") or 0 |
| 725 | if not title or rating is None or not total: |
| 726 | return |
| 727 | headline = f"{rating}/5 ({total:,} ratings)" |
| 728 | base = title.rsplit(" - ", 1)[0] if " - " in title else title |
| 729 | item.title = f"{base} - {headline}" |
| 730 | |
| 731 | |
| 732 | # ------------------------------------------------------------------ stats |
| 733 | |
| 734 | |
| 735 | def stats_from_item(item: Any, *, today: Optional[datetime] = None) -> Dict[str, Any]: |
| 736 | """Compute the stat block for a rendered SourceItem. |
| 737 | |
| 738 | Enrichment stores a precomputed block, but mock runs and replayed |
| 739 | fixtures skip enrichment entirely, so render recomputes from metadata |
| 740 | when it is absent. Cheap and pure -- all the inputs already live on |
| 741 | the item. |
| 742 | """ |
| 743 | metadata = getattr(item, "metadata", None) or {} |
| 744 | cached = metadata.get("stats") |
| 745 | if isinstance(cached, dict) and cached: |
| 746 | return cached |
| 747 | return product_stats( |
| 748 | { |
| 749 | "short_name": metadata.get("short_name") or "", |
| 750 | "name": metadata.get("name") or getattr(item, "title", ""), |
| 751 | "url": getattr(item, "url", "") or "", |
| 752 | "rating": metadata.get("rating"), |
| 753 | "num_ratings": metadata.get("num_ratings") or 0, |
| 754 | "product_rating": metadata.get("product_rating"), |
| 755 | "product_rating_count": metadata.get("product_rating_count") or 0, |
| 756 | "star_distribution": metadata.get("star_distribution") or {}, |
| 757 | "top_comments": metadata.get("top_comments") or [], |
| 758 | }, |
| 759 | today=today, |
| 760 | ) |
| 761 | |
| 762 | |
| 763 | def footer_entry(stats: Dict[str, Any], *, quote: str = "") -> str: |
| 764 | """Render one product's segment of the emoji-footer line (R1c). |
| 765 | |
| 766 | Shapes, by drift state:: |
| 767 | |
| 768 | Chill Max XL 4.4★→3.8★ ↓ "the lid jams" negative drift (+ quote) |
| 769 | Deluxe Bag 4.7★→5.0★ positive or flat drift |
| 770 | Spirit E-325 4.4★ quiet too few in-window reviews |
| 771 | BLUEY Set new no all-time baseline |
| 772 | |
| 773 | The ``↓`` is asymmetric on purpose: a sagging product is the alarm |
| 774 | worth catching at a glance, and a healthy one needs no decoration. |
| 775 | """ |
| 776 | name = stats.get("short_name") or "Product" |
| 777 | all_time = stats.get("all_time") |
| 778 | recent = stats.get("recent_avg") |
| 779 | drift = stats.get("drift") |
| 780 | |
| 781 | if drift == "new" or all_time is None: |
| 782 | return f"{name} new" |
| 783 | if drift == "quiet" or recent is None: |
| 784 | return f"{name} {all_time}★ quiet" |
| 785 | |
| 786 | entry = f"{name} {all_time}★→{recent}★" |
| 787 | if drift == "down": |
| 788 | entry += " ↓" |
| 789 | if quote: |
| 790 | entry += f' "{quote}"' |
| 791 | return entry |
| 792 | |
| 793 | |
| 794 | def five_star_share(distribution: Dict[str, Any]) -> Optional[float]: |
| 795 | """Share of ratings that are 5-star, from the star-distribution object.""" |
| 796 | if not isinstance(distribution, dict) or not distribution: |
| 797 | return None |
| 798 | total = sum(_as_int(distribution.get(key)) for key, _ in _STAR_FIELDS) |
| 799 | if total <= 0: |
| 800 | return None |
| 801 | return _as_int(distribution.get("five_star")) / total |
| 802 | |
| 803 | |
| 804 | def recent_window_stats( |
| 805 | comments: Iterable[Dict[str, Any]], |
| 806 | *, |
| 807 | today: Optional[datetime] = None, |
| 808 | window_days: int = RECENT_WINDOW_DAYS, |
| 809 | ) -> Dict[str, Any]: |
| 810 | """Average rating and sample size inside the recent window.""" |
| 811 | reference = (today or _today()).date() |
| 812 | ratings: List[int] = [] |
| 813 | for comment in comments or []: |
| 814 | iso = comment.get("date") |
| 815 | if not iso: |
| 816 | continue |
| 817 | try: |
| 818 | posted = datetime.strptime(iso, "%Y-%m-%d").date() |
| 819 | except (TypeError, ValueError): |
| 820 | continue |
| 821 | if 0 <= (reference - posted).days <= window_days: |
| 822 | rating = _as_int(comment.get("rating")) |
| 823 | if rating: |
| 824 | ratings.append(rating) |
| 825 | if not ratings: |
| 826 | return {"recent_n": 0, "recent_avg": None} |
| 827 | return {"recent_n": len(ratings), "recent_avg": sum(ratings) / len(ratings)} |
| 828 | |
| 829 | |
| 830 | def product_stats( |
| 831 | product: Dict[str, Any], |
| 832 | *, |
| 833 | today: Optional[datetime] = None, |
| 834 | ) -> Dict[str, Any]: |
| 835 | """Compute the render-facing stat block for one product. |
| 836 | |
| 837 | ``drift`` is one of: |
| 838 | * ``"new"`` -- no all-time baseline to move away from |
| 839 | * ``"quiet"`` -- baseline exists but the window has too few dated |
| 840 | reviews to average honestly (below MIN_DRIFT_SAMPLE) |
| 841 | * ``"up"`` / ``"down"`` / ``"flat"`` -- a real, sample-backed move |
| 842 | |
| 843 | The engine owns every number here; the model owns the words (R1b). |
| 844 | """ |
| 845 | # The review pull's rating count supersedes the search record's, which |
| 846 | # can be variant-level and badly low. |
| 847 | all_time = product.get("product_rating") |
| 848 | if all_time is None: |
| 849 | all_time = product.get("rating") |
| 850 | ratings_total = product.get("product_rating_count") or product.get("num_ratings") or 0 |
| 851 | |
| 852 | window = recent_window_stats(product.get("top_comments") or [], today=today) |
| 853 | recent_avg = window["recent_avg"] |
| 854 | recent_n = window["recent_n"] |
| 855 | |
| 856 | if all_time is None: |
| 857 | drift = "new" |
| 858 | elif recent_n < MIN_DRIFT_SAMPLE or recent_avg is None: |
| 859 | drift = "quiet" |
| 860 | elif round(recent_avg, 1) > round(float(all_time), 1): |
| 861 | drift = "up" |
| 862 | elif round(recent_avg, 1) < round(float(all_time), 1): |
| 863 | drift = "down" |
| 864 | else: |
| 865 | drift = "flat" |
| 866 | |
| 867 | return { |
| 868 | "short_name": product.get("short_name") or short_name(product.get("name", "")), |
| 869 | "url": product.get("url", ""), |
| 870 | "all_time": round(float(all_time), 1) if all_time is not None else None, |
| 871 | "ratings_total": _as_int(ratings_total), |
| 872 | "five_star_share": five_star_share(product.get("star_distribution") or {}), |
| 873 | "recent_avg": round(recent_avg, 1) if recent_avg is not None else None, |
| 874 | "recent_n": recent_n, |
| 875 | "reviews_pulled": len(product.get("top_comments") or []), |
| 876 | "drift": drift, |
| 877 | } |
| 878 |