#!/usr/bin/env python3 """Bounded public GET capture for one frozen sample. Python 3 standard library only. Prints a fresh factual extraction; never overwrites the published observation file unless the caller explicitly supplies that path with --output. No browser, authentication, checkout, cart, or merchant writes. This is a merchant-specific capture script, not a general product-page validator. """ import argparse import datetime import hashlib import json from html.parser import HTMLParser from pathlib import Path import urllib.error import urllib.request ROOT = "https://www.studioneat.com/products/" SAMPLE = { "markone": [17522303008833, 48310475227413, 48537955041557], "marktwo": [46227276628245, 46227345277205, 48537970868501], "totebook": [39368591736897, 39388035973185], } LIMIT = 2 * 1024 * 1024 RAW_DIRECTORY = None def now(): return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z") def get(url): observation = {"requestedUrl": url, "capturedAt": now(), "status": None, "finalUrl": None, "bodySha256": None, "bytes": None, "error": None} try: request = urllib.request.Request(url, headers={"User-Agent": "iamgeorgekelly-public-evidence-study/1.0"}) with urllib.request.urlopen(request, timeout=20) as response: body = response.read(LIMIT + 1) observation.update(status=response.status, finalUrl=response.url) if len(body) > LIMIT: raise ValueError("Response exceeded the 2 MiB capture limit") observation.update(bodySha256=hashlib.sha256(body).hexdigest(), bytes=len(body)) if RAW_DIRECTORY: # A caller-selected private directory, separate from the public extraction. RAW_DIRECTORY.mkdir(parents=True, exist_ok=True) extension = ".json" if url.endswith(".js") else ".html" filename = hashlib.sha256(url.encode()).hexdigest()[:12] + "-" + observation["bodySha256"] + extension (RAW_DIRECTORY / filename).write_bytes(body) return observation, body.decode("utf-8") except (urllib.error.URLError, ValueError, UnicodeDecodeError) as error: if isinstance(error, urllib.error.HTTPError): observation["status"] = error.code observation["error"] = str(error) return observation, None class Facts(HTMLParser): def __init__(self): super().__init__() self.facts = {"canonicalUrl": None, "h1": [], "checkedOptions": [], "variantInputIds": [], "selectedVariantOptions": [], "regularPriceTexts": [], "countryCurrencyText": None, "productNodes": [], "parseErrors": []} self.capture = None self.parts = [] self.select_name = None def handle_starttag(self, tag, attrs): a = dict(attrs) if tag == "link" and a.get("rel") == "canonical": self.facts["canonicalUrl"] = a.get("href") if tag == "input": if a.get("type") == "radio" and "checked" in a: self.facts["checkedOptions"].append({"name": a.get("name"), "value": a.get("value")}) if a.get("name") == "id": self.facts["variantInputIds"].append(a.get("value")) if tag == "select": self.select_name = a.get("name") if tag == "h1": self.capture = (tag, "h1", None) self.parts = [] elif tag == "option" and "selected" in a: if self.select_name == "id": self.capture = (tag, "selectedVariantOptions", a.get("value")) self.parts = [] elif a.get("value") == "US": self.capture = (tag, "countryCurrencyText", None) self.parts = [] elif tag == "span" and "price-item--regular" in a.get("class", "").split(): self.capture = (tag, "regularPriceTexts", None) self.parts = [] elif tag == "script" and a.get("type") == "application/ld+json": self.capture = (tag, "jsonld", None) self.parts = [] def handle_data(self, data): if self.capture: self.parts.append(data) def handle_endtag(self, tag): if tag == "select": self.select_name = None if self.capture and tag == self.capture[0]: _, field, value = self.capture text = " ".join("".join(self.parts).split()) if field == "jsonld": try: def walk(node): if isinstance(node, list): for child in node: walk(child) elif isinstance(node, dict): if node.get("@type") == "Product": offers = node.get("offers", []) if isinstance(offers, dict): offers = [offers] self.facts["productNodes"].append({ key: node.get(key) for key in ["@type", "name", "url", "sku"] } | {"offers": [{key: offer.get(key) for key in ["@type", "sku", "url", "price", "priceCurrency", "availability"]} for offer in offers]}) elif "@graph" in node: walk(node["@graph"]) walk(json.loads("".join(self.parts))) except (ValueError, TypeError) as error: self.facts["parseErrors"].append(str(error)) elif field == "selectedVariantOptions": self.facts[field].append({"id": value, "text": text}) elif field == "countryCurrencyText": self.facts[field] = text else: self.facts[field].append(text) self.capture = None self.parts = [] def extract_facts(url, body): parser = Facts() parser.feed(body) for product in parser.facts["productNodes"]: offers = product["offers"] product["offerCount"] = len(offers) # Retain only evidence needed for the tested first-offer and exact-URL rules. # All exact matches remain, so duplicate matching offers are not concealed. product["offers"] = [dict(offer, arrayIndex=index) for index, offer in enumerate(offers) if index == 0 or offer.get("url") == url] return parser.facts def capture_html(url): observation, body = get(url) if body is not None: observation["facts"] = extract_facts(url, body) else: observation["facts"] = None return observation def capture(): report = { "schemaVersion": "1.0", "studyId": "product-variant-url-evidence-2026-10-04", "capturedAt": now(), "publisher": "Studio Neat", "provenance": "Observed public HTTP responses", "selection": "Convenience sample: three known Studio Neat product handles; first three variants in returned order for Mark One and Mark Two, both returned Totebook variants. IDs frozen before the final capture; no selection by observed mismatch.", "context": {"authentication": "None", "cookies": "No cookies sent or retained by this script", "requestUserAgent": "iamgeorgekelly-public-evidence-study/1.0", "timeoutSeconds": 20, "responseLimitBytes": LIMIT, "timeZone": "UTC", "localePath": "English root path", "freshness": "Capture time recorded; source update time and edge-cache age unknown"}, "catalogs": [], "basePages": [], "variantPages": [], } for handle, ids in SAMPLE.items(): observation, body = get(ROOT + handle + ".js") observation["handle"] = handle observation["facts"] = None if body is not None: try: product = json.loads(body) variants = product.get("variants", []) observation["facts"] = { "title": product.get("title"), "variantCount": len(variants), "sampleIds": ids, "returnedFirstIds": [v.get("id") for v in variants[:len(ids)]], "variants": [{k: v.get(k) for k in ["id", "title", "sku", "available", "price", "option1", "option2", "option3"]} for v in variants if v.get("id") in ids], } except ValueError as error: observation["error"] = "JSON parse: " + str(error) report["catalogs"].append(observation) report["basePages"].append({"handle": handle, **capture_html(ROOT + handle)}) for variant_id in ids: report["variantPages"].append({"handle": handle, "variantId": str(variant_id), **capture_html(ROOT + handle + "?variant=" + str(variant_id))}) report["completedAt"] = now() return report if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--output", type=Path, help="Explicit path for a new capture; default prints JSON") parser.add_argument("--raw-dir", type=Path, help="Optional PRIVATE directory for full source bodies; keep outside a public repo") args = parser.parse_args() RAW_DIRECTORY = args.raw_dir output = json.dumps(capture(), indent=2, ensure_ascii=False) + "\n" if args.output: args.output.write_text(output, encoding="utf-8") else: print(output, end="")