1"""Claim check (Lane C, c3). Checks one seller claim against the live page.
2
3check_claim(claim, fetch=None, history=None) ->
4 {"result": "verified"|"unverified"|"false", "evidence_url", "checked_at" (ISO-8601 UTC), "method",
5 "evidence_hash" (sha256 of the page bytes), "reason"}
6Input is a Claim (contracts/Claim.schema.json); `result` maps to Claim.check. The page is claim["evidence_url"].
7Optional claim["field"] and claim["value"] skip text parsing. Page facts come from our SIM pages'
8<script id="listing"> JSON, else schema.org JSON-LD. A page we cannot read is "unverified", never "false".
9Wording is neutral: "differs from the listing at time T", "did not match the price history", "could not verify".
10"""
11import hashlib
12import json
13import re
14from datetime import datetime, timezone
15from urllib.parse import urlparse
16
17from data.fit_checker.tables import MICROARCH_SOCKET
18
19PATTERNS = [
20 ("was_price", r"\bwas\s*\$?\s*([\d,]+(?:\.\d+)?)", lambda m: _num(m.group(1))),
21 ("socket", r"\bsocket\s*(AM\d|LGA\s?\d{3,4})", lambda m: m.group(1).replace(" ", "").upper()),
22 ("return_days", r"(\d+)[- ]day returns?", lambda m: int(m.group(1))),
23 ("stock", r"\b(out of stock|not in stock|unavailable|sold out)\b", lambda m: False),
24 ("stock", r"\bin stock\b", lambda m: True),
25 ("warranty_months", r"(\d+)[- ]?(year|yr|month|mo)s?\b[- ]?(?:limited\s+)?warranty", lambda m: _warranty_months(int(m.group(1)), m.group(2))),
26 ("condition", r"\bopen[- ]box\b", lambda m: "open-box"),
27 ("condition", r"\brefurb", lambda m: "refurbished"),
28 ("condition", r"\b(used|pre-owned)\b", lambda m: "used"),
29 ("condition", r"\bnew\b", lambda m: "new"),
30 ("price", None, None),
31]
32
33
34NOT_PRICE_BEFORE = re.compile(r"\b(over|save|saves|saving|under|shipping|ship)\s*$", re.I)
35NOT_PRICE_AFTER = re.compile(r"^\s*(off|fee|fees)\b", re.I)
36
37OFFER_BEFORE = re.compile(r"\b(orders? of|up to|rebate|coupon|gift card|credit)\s*$", re.I)
38OFFER_AFTER = re.compile(r"^\s*(or more|gift card|rebate|coupon|credit|back)\b", re.I)
39
40LD_CONDITION = {"NewCondition": "new", "UsedCondition": "used", "RefurbishedCondition": "refurbished", "DamagedCondition": "used"}
41
42
43def _num(s):
44 return float(str(s).replace(",", ""))
45
46
47def _money(v):
48 """Page price -> float, or None when the page gives no usable number ("", null, "N/A")."""
49 if isinstance(v, bool) or v in (None, ""):
50 return None
51 if isinstance(v, (int, float)):
52 return float(v)
53 m = re.search(r"\d[\d,]*\.?\d*", str(v))
54 return _num(m.group(0)) if m else None
55
56
57def _price(text):
58 """(amount,) for the claimed price; (None,) when it is ambiguous (two different amounts, or an offer amount such as a
59 gift card or an order minimum); () when no amount is the price."""
60 amounts, offer = [], False
61 for m in re.finditer(r"\$\s?([\d,]+(?:\.\d+)?)", text):
62 before, after = text[:m.start()], text[m.end():]
63 if OFFER_BEFORE.search(before) or OFFER_AFTER.search(after):
64 offer = True
65 elif not (NOT_PRICE_BEFORE.search(before) or NOT_PRICE_AFTER.search(after)):
66 amounts.append(_num(m.group(1)))
67 if offer or len(set(amounts)) > 1:
68 return (None,)
69 return (amounts[0],) if amounts else ()
70
71
72def parse_claims(claim):
73 """Every field the claim states, in PATTERNS order. A "was $X" amount is not also read as the current price."""
74 if claim.get("field"):
75 return [(claim["field"], claim.get("value"))]
76 text = claim.get("text") or ""
77 found = {}
78 for field, rx, val in PATTERNS:
79 if field == "price":
80 for v in _price(text):
81 found.setdefault("price", v)
82 continue
83 m = re.search(rx, text, re.I)
84 if m and field not in found:
85 found[field] = val(m)
86 if field == "was_price":
87 text = text[:m.start()] + text[m.end():]
88 return list(found.items())
89
90
91def parse_claim(claim):
92 """The first field the claim states (kept for callers that read one field)."""
93 return (parse_claims(claim) or [(None, None)])[0]
94
95
96from .fetch import guarded_get as default_fetch
97
98
99_ASCII_LOWER = str.maketrans("ABCDEFGHIJKLMNOPQRSTUVWXYZ", "abcdefghijklmnopqrstuvwxyz")
100LISTING_TAG = '<script type="application/json" id="listing">'
101PRODUCT_TYPES = {"Product", "IndividualProduct", "SomeProducts", "ProductModel", "ProductGroup"}
102
103
104def script_blocks(html):
105 """[(opening tag, body)] for every <script ...>...</script> in one linear scan (#317: the old regex was cubic on
106 "<script x application/ld+json" * N). Each search starts where the last one ended, so no character is rescanned."""
107 low, out, i = html.translate(_ASCII_LOWER), [], 0
108 while True:
109 s = low.find("<script", i)
110 if s < 0:
111 break
112 if low[s + 7:s + 8] not in ("", ">", "/") and not low[s + 7:s + 8].isspace():
113 i = s + 7
114 continue
115 e = low.find(">", s + 7)
116 if e < 0:
117 break
118 c = low.find("</script", e + 1)
119 if c < 0:
120 break
121 out.append((html[s:e + 1], html[e + 1:c]))
122 i = c + 8
123 return out
124
125
126def _loads(text, what):
127 try:
128 return json.loads(text)
129 except RecursionError as e:
130 raise ValueError(f"{what} is nested too deeply") from e
131
132
133def _is_product(node):
134 t = node.get("@type")
135 types = [t] if isinstance(t, str) else t if isinstance(t, list) else []
136 return any(isinstance(x, str) and re.sub(r"^(https?://schema\.org/|schema:)", "", x.strip()) in PRODUCT_TYPES for x in types)
137
138
139def _same_url(a, b):
140 norm = lambda u: u.strip().split("#", 1)[0].rstrip("/") if isinstance(u, str) else None
141 return norm(a) is not None and norm(a) == norm(b)
142
143
144def _page_key(u, base):
145 """(host, path) of u resolved against base, ignoring scheme, query, fragment and a trailing slash; None if not a URL."""
146 from urllib.parse import urljoin, urlparse
147 if not isinstance(u, str) or not u.strip():
148 return None
149 p = urlparse(urljoin(base, u.strip()))
150 host = (p.hostname or "").lower()
151 host = host[4:] if host.startswith("www.") else host
152 return (host, p.path.rstrip("/")) if p.hostname else None
153
154
155def _names_another_page(node, url):
156 """True when the Product's url is a different page (a related product or a carousel item, not this listing)."""
157 there, here = _page_key(node.get("url"), url), _page_key(url, url)
158 return there is not None and here is not None and there != here
159
160
161def ld_product(html, url=None, notes=None):
162 """The schema.org Product node an agent reads: the one whose @id or url is this page (a leading www. ignored), else the
163 first with offers, else the first; {} when there is none. With the page url given, a Product whose url is another page
164 is not read when another Product is there (#334); when it is the only Product (a canonical on another path) it is kept
165 and a note is appended to `notes` (#397). Raises ValueError for JSON-LD too deeply nested to parse."""
166 nodes = []
167 for tag, body in script_blocks(html):
168 if "application/ld+json" not in tag.translate(_ASCII_LOWER):
169 continue
170 try:
171 data = _loads(body, "JSON-LD")
172 except ValueError as e:
173 if isinstance(e.__cause__, RecursionError):
174 raise
175 continue
176 items = data if isinstance(data, list) else data.get("@graph", [data]) if isinstance(data, dict) else []
177 nodes += [n for n in (items if isinstance(items, list) else [items]) if isinstance(n, dict) and _is_product(n)]
178 if url:
179 here = _page_key(url, url)
180 for n in nodes:
181 if _same_url(n.get("@id"), url) or _same_url(n.get("url"), url) or \
182 (here is not None and here in (_page_key(n.get("@id"), url), _page_key(n.get("url"), url))):
183 return n
184 kept = [n for n in nodes if not _names_another_page(n, url)]
185 if not kept and len(nodes) == 1:
186 if notes is not None:
187 notes.append(f"the only Product names another url ({str(nodes[0].get('url'))[:120]}), read as this listing")
188 return nodes[0]
189 nodes = kept
190 return next((n for n in nodes if n.get("offers")), nodes[0] if nodes else {})
191
192
193def ld_offer(product):
194 offer = product.get("offers") or {}
195 offer = offer[0] if isinstance(offer, list) and offer else offer
196 return offer if isinstance(offer, dict) else {}
197
198
199def page_facts(html, url=None):
200 """(facts, how) from our SIM <script id="listing"> JSON, else the page's schema.org Product (see ld_product).
201 Raises ValueError when the listing JSON is not a JSON object or the page's JSON is too deeply nested to parse."""
202 body = next((b for tag, b in script_blocks(html) if tag == LISTING_TAG), None)
203 if body is not None:
204 data = _loads(body, "listing JSON")
205 if not isinstance(data, dict):
206 raise ValueError("listing JSON is not an object")
207 if "warranty_months" not in data:
208 wm = _text_warranty(data.get("warranty"), data.get("description"))
209 if wm is not None:
210 data["warranty_months"] = wm
211 return data, "page listing JSON"
212 notes = []
213 node = ld_product(html, url, notes)
214 if not node:
215 return {}, "no structured data"
216 offer = ld_offer(node)
217 cond = str(offer.get("itemCondition") or "").rsplit("/", 1)[-1]
218 avail = str(offer.get("availability") or "").rsplit("/", 1)[-1]
219 facts = {"title": node.get("name"), "price": _money(offer.get("price")),
220 "condition": LD_CONDITION.get(cond), "stock": avail or None,
221 "return_policy": _ld_returns(offer.get("hasMerchantReturnPolicy") or node.get("hasMerchantReturnPolicy")),
222 "warranty_months": _ld_warranty(node.get("warranty") or offer.get("warranty"))
223 if (node.get("warranty") or offer.get("warranty")) else _text_warranty(node.get("description"))}
224 return {k: v for k, v in facts.items() if v is not None}, "page JSON-LD" + "".join(f"; {n}" for n in notes)
225
226
227def _return_days(days):
228 """Finite, non-negative, whole-number days only (a bool, Infinity, -30 or 14.5 is not a policy)."""
229 if isinstance(days, str) and days.strip().isdigit():
230 days = int(days)
231 if isinstance(days, bool) or not isinstance(days, (int, float)):
232 return None
233 if isinstance(days, float) and (days != days or days in (float("inf"), float("-inf")) or days != int(days)):
234 return None
235 return int(days) if days >= 0 else None
236
237
238def _one_return_policy(policy):
239 if not isinstance(policy, dict):
240 return None
241 if str(policy.get("returnPolicyCategory") or "").endswith("MerchantReturnNotPermitted"):
242 return "returns not permitted"
243 days = _return_days(policy.get("merchantReturnDays"))
244 return None if days is None else f"{days}-day returns"
245
246
247def _warranty_months(n, unit):
248 """N years or months -> whole months, or None when out of range (finite, non-negative, whole, at most 120 months)."""
249 n = _return_days(n)
250 if n is None:
251 return None
252 months = n * 12 if str(unit).lower().startswith(("y", "ann")) else n if str(unit).lower().startswith(("m", "mon")) else None
253 return months if months is not None and months <= 120 else None
254
255
256def _one_warranty(w):
257 d = w.get("durationOfWarranty") if isinstance(w, dict) else None
258 if not isinstance(d, dict):
259 return None
260 return _warranty_months(d.get("value"), d.get("unitCode") or d.get("unitText"))
261
262
263def _ld_warranty(w):
264 """schema.org WarrantyPromise -> months. A list counts only when every promise agrees."""
265 if isinstance(w, list):
266 found = {_one_warranty(x) for x in w}
267 return found.pop() if len(found) == 1 else None
268 return _one_warranty(w)
269
270
271def _text_warranty(*texts):
272 for t in texts:
273 m = re.search(PATTERNS[next(i for i, p in enumerate(PATTERNS) if p[0] == "warranty_months")][1], str(t or ""), re.I)
274 if m:
275 return _warranty_months(int(m.group(1)), m.group(2))
276 return None
277
278
279def _ld_returns(policy):
280 """schema.org MerchantReturnPolicy -> "N-day returns", "returns not permitted", or None when it says neither.
281 A list of policies counts only when every one agrees (we do not guess which country applies)."""
282 if isinstance(policy, list):
283 found = {_one_return_policy(p) for p in policy}
284 return found.pop() if len(found) == 1 else None
285 return _one_return_policy(policy)
286
287
288SCHEMA_AVAILABLE = {"instock", "limitedavailability", "onlineonly", "instoreonly"}
289SCHEMA_UNAVAILABLE = {"outofstock", "soldout", "discontinued", "preorder", "presale", "backorder"}
290
291
292def in_stock(v):
293 if isinstance(v, bool):
294 return v
295 s = str(v).lower()
296 if s.replace(" ", "") in SCHEMA_AVAILABLE:
297 return True
298 if s.replace(" ", "") in SCHEMA_UNAVAILABLE:
299 return False
300 if re.search(r"out of stock|outofstock|not in stock|unavailable|sold out|soldout", s) or s.startswith("0 "):
301 return False
302 if "in stock" in s or "instock" in s or re.match(r"\d+ (units|in stock|remaining)", s):
303 return True
304 return None
305
306
307LIAR_PAGE_HOSTS = {"localhost:8765", "127.0.0.1:8765"}
308
309
310def is_our_liar_page(url):
311 """SIM is decided from the evidence URL (our own hosted pages), never from anything a page says about itself."""
312 p = urlparse(url or "")
313 return p.scheme == "http" and p.netloc in LIAR_PAGE_HOSTS and p.path.endswith(".html")
314
315
316def _history(url, facts, fetch, history, observations=None):
317 """Price history for a was-price claim, from our own records only: the caller's history, our recorded SIM series for a
318 SIM page, or our dated MIRROR observations of this exact URL. Nothing is read from the page being judged."""
319 if history is not None:
320 return history
321 if is_our_liar_page(url) and facts.get("sku"):
322 try:
323 from data.liar_pages.build import PRICE_HISTORY
324 except ImportError:
325 return None
326 return PRICE_HISTORY.get(facts["sku"])
327 if observations is None:
328 try:
329 from data.baselines import observations as load_observations
330 except ImportError:
331 return None
332 observations = load_observations()
333 series = sorted(({"date": str(r["observed_at"])[:10], "price": r["price"]} for r in observations if r.get("url") == url),
334 key=lambda p: p["date"])
335 return series or None
336
337
338def judge(field, value, facts, at, series):
339 """-> (result, reason)"""
340 if field == "condition":
341 page = facts.get("condition")
342 if not page:
343 return "unverified", "could not verify: the page states no condition"
344 return ("verified", f"the page shows {page}") if page == value else \
345 ("false", f"claim says {value}; differs from the listing at {at}: the page shows {page}")
346 if field == "price" and value is None:
347 return "unverified", "could not verify: the claim states more than one amount (another price, or an offer such as a gift card or order minimum)"
348 if field == "price":
349 page = _money(facts.get("price"))
350 if page is None:
351 return "unverified", "could not verify: the page states no price"
352 return ("verified", f"the page shows ${page:,.2f}") if abs(page - value) < 0.01 else \
353 ("false", f"claim says ${value:,.2f}; differs from the listing at {at}: the page shows ${page:,.2f}")
354 if field == "was_price":
355 if not series:
356 return "unverified", "could not verify: no dated price history from our own captures"
357 top = max(series, key=lambda p: p["price"])
358 span = f"{series[0]['date']} to {series[-1]['date']}"
359 if value <= top["price"] + 0.01:
360 return "verified", f"price history shows ${top['price']:,.2f} on {top['date']}"
361 return "false", f"the price claim did not match the price history: highest recorded ${top['price']:,.2f} ({span})"
362 if field == "socket":
363 specs = facts.get("specs") or {}
364 arch = specs.get("Microarchitecture")
365 truth = MICROARCH_SOCKET.get(arch) if arch else None
366 if truth:
367 return ("verified", f"maker table maps {arch} to {truth}") if truth == value else \
368 ("false", f"claim says {value}; the maker table maps {arch} to {truth}")
369 page = specs.get("Socket")
370 if not page:
371 return "unverified", "could not verify: no socket or microarchitecture on the page"
372 return ("verified", f"the page lists socket {page}") if page == value else \
373 ("false", f"claim says {value}; differs from the listing at {at}: the page lists {page}")
374 if field == "stock":
375 page = in_stock(facts.get("stock"))
376 if page is None:
377 return "unverified", "could not verify: stock not stated on the page"
378 word = lambda b: "in stock" if b else "out of stock"
379 raw = str(facts.get("stock"))
380 shown = f"{raw} (not available now)" if raw.lower().replace(" ", "") in SCHEMA_UNAVAILABLE - {"outofstock", "soldout"} else word(page)
381 return ("verified", f"the page shows {shown}") if page == value else \
382 ("false", f"claim says {word(value)}; differs from the listing at {at}: the page shows {shown}")
383 if field == "warranty_months":
384 page = facts.get("warranty_months")
385 if page is None:
386 return "unverified", "could not verify: no warranty term on the page"
387 return ("verified", f"the page shows a {page}-month warranty") if page == value else \
388 ("false", f"claim says {value} months of warranty; differs from the listing at {at}: the page shows {page} months")
389 if field == "return_days":
390 policy = str(facts.get("return_policy") or "")
391 if "not permitted" in policy:
392 return "false", f"claim says {value}-day returns; differs from the listing at {at}: the page says returns are not permitted"
393 m = re.search(r"(\d+)[- ]day", policy)
394 if not m:
395 return "unverified", "could not verify: no return window on the page"
396 days = int(m.group(1))
397 return ("verified", f"the page shows {days}-day returns") if days == value else \
398 ("false", f"claim says {value}-day returns; differs from the listing at {at}: the page shows {days}-day")
399 return "unverified", f"could not verify: {field} claims are not checked yet"
400
401
402def check_claim(claim, fetch=None, history=None, observations=None):
403 fetch = fetch or default_fetch
404 url = claim.get("evidence_url") or ""
405 unverified = lambda why: {**out, "method": f"could not verify: {why}", "reason": "could not verify"}
406 at = datetime.now(timezone.utc).isoformat(timespec="seconds")
407 out = {"result": "unverified", "evidence_url": url, "checked_at": at, "evidence_hash": ""}
408 fields = parse_claims(claim)
409 if not fields:
410 return unverified("claim text not recognized")
411 if urlparse(url).scheme not in ("http", "https"):
412 return unverified("only http and https pages are checked")
413 try:
414 body = fetch(url)
415 except Exception as e:
416 return unverified(f"page fetch failed ({e if isinstance(e, ValueError) else type(e).__name__})")
417 if isinstance(body, str):
418 body = body.encode("utf-8")
419 if not isinstance(body, (bytes, bytearray)):
420 return unverified("page body was not text")
421 out["evidence_hash"] = hashlib.sha256(body).hexdigest()
422 try:
423 facts, source = page_facts(body.decode("utf-8", "replace"), url)
424 results = [(f, *judge(f, v, facts, at, _history(url, facts, fetch, history, observations) if f == "was_price" else None))
425 for f, v in fields]
426 except (ValueError, TypeError, KeyError, AttributeError, OverflowError, RecursionError) as e:
427 return unverified(f"page could not be parsed ({type(e).__name__})")
428 rank = {"false": 0, "unverified": 1, "verified": 2}
429 result = min((r for _, r, _ in results), key=rank.get)
430 names = ", ".join(f for f, _, _ in results)
431 reason = "; ".join(why for _, r, why in results if r == result)
432 rules = f"{names} rule{'s' if len(results) > 1 else ''}"
433 if result == "unverified":
434 return {**out, "result": result, "method": f"could not verify: {source}, {rules}", "reason": reason}
435 return {**out, "result": result, "method": f"fetched page, {source}, {rules}", "reason": reason}
