#!/usr/bin/env python3 """Append one real public HTTP read per exact product. Python 3, standard library only. Run: python3 observe-public-offers.py No credentials, sleeps, schedules, historical fabrication, checkout, or image downloads. """ import copy import argparse import hashlib import json from datetime import datetime, timezone from html.parser import HTMLParser from pathlib import Path from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError ROOT = Path(__file__).resolve().parent argument_parser = argparse.ArgumentParser(description=__doc__) argument_parser.add_argument('dataset', nargs='?', type=Path, default=(ROOT.parent / 'data/public-products.json' if (ROOT.parent / 'data/public-products.json').exists() else ROOT / 'public-products.json')) DATA = argument_parser.parse_args().dataset.resolve() ROOT = DATA.parent TARGETS = [('anker-a2669113', 'A2669113'), ('belkin-wch013dqwh', 'WCH013dqWH')] CACHE_HEADERS = ['Date', 'Age', 'Cache-Control', 'Expires', 'ETag', 'Last-Modified', 'CF-Cache-Status', 'X-Cache', 'X-Cache-Hits', 'Via', 'Vary'] class StructuredData(HTMLParser): def __init__(self): super().__init__() self.in_ld = False self.in_title = False self.buffer = [] self.blocks = [] self.title = [] def handle_starttag(self, tag, attrs): attrs = dict(attrs) if tag == 'script' and attrs.get('type', '').lower() == 'application/ld+json': self.in_ld = True self.buffer = [] if tag == 'title': self.in_title = True def handle_endtag(self, tag): if tag == 'script' and self.in_ld: self.blocks.append(''.join(self.buffer)) self.in_ld = False if tag == 'title': self.in_title = False def handle_data(self, text): if self.in_ld: self.buffer.append(text) if self.in_title: self.title.append(text) def objects(node): if isinstance(node, dict): yield node for value in node.values(): yield from objects(value) elif isinstance(node, list): for value in node: yield from objects(value) def utcnow(): return datetime.now(timezone.utc).isoformat() def main(): data = json.loads(DATA.read_text()) reports = [] for product_id, exact_sku in TARGETS: prior = [o for o in data['offer_observations'] if o['product_id'] == product_id][-1] url = prior['source_url'] stamp = utcnow() base_source_id = next(o['source_id'] for o in data['offer_observations'] if o['product_id'] == product_id) seq = 1 + len([s for s in data['sources'] if s['id'].startswith(base_source_id + '-read-')]) source_id = base_source_id + '-read-' + str(seq + 1).zfill(2) evidence = {'source_id': source_id, 'url': url, 'request_started_at_utc': stamp, 'expected_exact_sku': exact_sku, 'request_cache_control': 'no-cache', 'cache_limitation': 'A new HTTP request is not proof of a fresh origin computation. CDN/server caches may supply content; origin freshness is unknown. No cache-busting parameter was added.'} try: request = Request(url, headers={'User-Agent': 'Mozilla/5.0 (compatible; PublicCatalogEvidence/1.0)', 'Accept': 'text/html', 'Cache-Control': 'no-cache'}) try: response = urlopen(request, timeout=45) except HTTPError as error: response = error body = response.read() stamp = utcnow() status = response.status evidence.update(retrieved_at_utc=stamp, http_status=status, final_url=response.geturl(), response_cache_headers={h: response.headers.get(h) for h in CACHE_HEADERS}, response_body_sha256=hashlib.sha256(body).hexdigest(), response_bytes=len(body)) if status != 200: raise ValueError('HTTP ' + str(status)) parser = StructuredData() parser.feed(body.decode('utf-8', errors='replace')) evidence['title'] = ''.join(parser.title).strip() matched = [] for block in parser.blocks: try: document = json.loads(block) except json.JSONDecodeError: continue matched.extend(n for n in objects(document) if str(n.get('sku', '')) == exact_sku and n.get('offers')) if not matched: raise ValueError('Exact expected SKU with direct offers not found in JSON-LD') offers = [] for match in matched: candidates = match['offers'] if isinstance(match['offers'], list) else [match['offers']] offers.extend((match, o) for o in candidates if isinstance(o, dict) and o.get('price') is not None and o.get('priceCurrency')) values = {(str(o['price']), o['priceCurrency'], o.get('availability'), o.get('url')) for _, o in offers} if len(values) != 1: raise ValueError('Missing or conflicting direct exact-SKU offers; manual review required') product, offer = offers[0] observed = copy.deepcopy(prior) observed.update(id='offer-' + product_id + '-' + str(1 + len([o for o in data['offer_observations'] if o['product_id'] == product_id])).zfill(2), source_id=source_id, observed_at_utc=stamp, price=float(offer['price']), currency=offer['priceCurrency'], availability_evidence=offer.get('availability'), availability={'https://schema.org/InStock':'in_stock','http://schema.org/InStock':'in_stock', 'https://schema.org/OutOfStock':'out_of_stock','http://schema.org/OutOfStock':'out_of_stock'}.get(offer.get('availability'), 'unknown'), fetch_status='success', http_status=status, variant_url=offer.get('url') or prior['variant_url'], exact_sku_verified=exact_sku, response_body_sha256=evidence['response_body_sha256'], cache_note=evidence['cache_limitation']) # Preserve the established market/unknowns: a new USD amount does not prove a country. evidence['matched_product'] = {k: product.get(k) for k in ['@type', 'name', 'sku', 'mpn', 'gtin', 'color'] if k in product} evidence['matched_offer'] = {k: offer.get(k) for k in ['@type','url','price','priceCurrency','availability','priceValidUntil'] if k in offer} evidence['observation'] = observed data['offer_observations'].append(observed) read_status = 'read' report = {'product_id': product_id, 'observed_at_utc': stamp, 'status': status, 'price': observed['price'], 'currency': observed['currency'], 'availability': observed['availability'], 'prior_observed_at_utc': prior['observed_at_utc'], 'price_changed': observed['price'] != prior['price'], 'cache_headers': evidence['response_cache_headers']} except (URLError, ValueError, TimeoutError, OSError) as error: stamp = evidence.get('retrieved_at_utc', utcnow()) failure = {'source_id':source_id,'product_id':product_id,'url':url,'observed_at_utc':stamp, 'http_status':evidence.get('http_status'),'fetch_status':'failed','availability':'unknown', 'price':None,'currency':None,'reason':str(error)} data['failed_fetches'].append(failure) evidence['failure'] = failure read_status = 'fetch_failed' report = failure (ROOT / 'public-source-excerpts').mkdir(exist_ok=True) evidence_path = ROOT / 'public-source-excerpts' / (source_id + '.json') evidence_path.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + '\n') data['sources'].append({'id':source_id,'url':url,'final_url':evidence.get('final_url'), 'retrieved_at_utc':stamp,'http_status':evidence.get('http_status'), 'title':evidence.get('title'),'read_status':read_status, 'excerpt_file':str(evidence_path.relative_to(ROOT)), 'response_cache_headers':evidence.get('response_cache_headers',{})}) reports.append(report) old_warning = 'One observation time per successful retail variant; no invented historical changes or multi-day monitoring results.' if old_warning in data['warnings']: data['warnings'].remove(old_warning) warning = 'Offer observations are individual real HTTP reads with distinct retrieval timestamps, not evidence of multi-day monitoring or fresh origin data. Prices may be unchanged. CDN/server cache freshness remains unverified.' if warning not in data['warnings']: data['warnings'].append(warning) data['last_public_offer_read_at_utc'] = utcnow() temp = DATA.with_suffix('.json.tmp') temp.write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n') temp.replace(DATA) print(json.dumps(reports, ensure_ascii=False, indent=2)) if __name__ == '__main__': main()