catalog-extract/scripts/nx.py · py · download
#!/usr/bin/env python3
"""Nimble Extract helper: the single fetch path for every script in this skill.
fetch(url, ...) -> Page one page through Nimble Extract
fetch_json(url, ...) -> object|None a JSON endpoint (unwraps the <pre> shell)
browser_fetch(page, reqs) -> [object] in-browser fetch() calls (POST / exact headers)
warm_cookies(host) -> [cookie] cookie jar from a homepage visit, sanitised
Everything here is plain `client.extract` from the `nimble_python` SDK. No
Extraction Templates, no Web Search Agents.
The one rule that matters: `status == "success"` does NOT prove the body is
complete. Nimble can return a cleanly truncated page (closing tags intact) when
the render races. Every fetch therefore takes a `sentinel` (a substring or
regex that must appear in the HTML) and/or `expect_json=True`, and retries
until the guard passes. A row written from a truncated page looks complete and
is silently wrong, so prefer a counted failure over a hollow row.
NIMBLE_API_KEY comes from the environment or the nearest `.env` (cwd upwards).
CLI (recon / one-off probes):
python nx.py URL [--render] [--wait 4000] [--sentinel 'application/ld+json']
[--json] [--driver vx10-pro] [--country US] [--out page.html]
"""
from __future__ import annotations
import argparse
import html as _htmllib
import json
import logging
import os
import re
import sys
import threading
import time
import warnings
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path
logger = logging.getLogger("nx")
logging.getLogger("httpx").setLevel(logging.WARNING) # one INFO line per request otherwise
# Nimble's Pydantic model types `parsing` as dict; list-shaped payloads warn on
# every serialise. Harmless; silence by message prefix.
warnings.filterwarnings("ignore", message="Pydantic serializer warnings:")
MAX_TRIES = 3
BACKOFF_BASE = 2.0 # seconds; sleep = BACKOFF_BASE ** attempt
CLIENT_TIMEOUT_S = 600.0 # render=True + browser actions can take minutes
DEFAULT_REQUEST_TIMEOUT_MS = 120_000
DEFAULT_DRIVER = "vx10-pro"
_billed = 0
_billed_lock = threading.Lock()
def billed_requests() -> int:
"""Every extract attempt is billed, retries included. Log this at the end
of a run so refresh cost is recorded rather than estimated."""
return _billed
def _count() -> None:
global _billed
with _billed_lock:
_billed += 1
@dataclass
class Page:
ok: bool
html: str = ""
fetch_results: list = field(default_factory=list)
cookies: list = field(default_factory=list)
task_id: str | None = None
error: str | None = None
attempts: int = 0
@property
def size(self) -> int:
return len(self.html or "")
def _load_env() -> None:
if os.environ.get("NIMBLE_API_KEY"):
return
here = Path.cwd()
for d in [here, *here.parents]:
p = d / ".env"
if p.is_file():
for line in p.read_text(encoding="utf-8").splitlines():
line = line.strip()
if line.startswith("NIMBLE_API_KEY=") and "=" in line:
os.environ.setdefault("NIMBLE_API_KEY", line.split("=", 1)[1].strip().strip('"').strip("'"))
return
@lru_cache(maxsize=1)
def get_client():
from nimble_python import Nimble # imported lazily so --help works without the SDK
_load_env()
key = os.environ.get("NIMBLE_API_KEY")
if not key:
raise RuntimeError("NIMBLE_API_KEY is not set (env or a .env in cwd / a parent).")
return Nimble(api_key=key, timeout=CLIENT_TIMEOUT_S)
def _guard_ok(html: str, sentinel, expect_json: bool) -> tuple[bool, str]:
"""Return (passes, reason). The guard is what makes a fetch trustworthy."""
if expect_json:
if json_from_html(html) is None:
return False, "body is not parseable JSON (truncated or blocked)"
if sentinel:
if isinstance(sentinel, re.Pattern):
if not sentinel.search(html):
return False, f"sentinel /{sentinel.pattern}/ absent in {len(html)} bytes"
elif sentinel not in html:
return False, f"sentinel {sentinel!r} absent in {len(html)} bytes"
return True, ""
def fetch(
url: str,
*,
render: bool = False,
wait_ms: int | None = None,
sentinel: str | re.Pattern | None = None,
expect_json: bool = False,
browser_actions: list | None = None,
driver: str = DEFAULT_DRIVER,
country: str = "US",
locale: str = "en",
formats: list[str] | None = None,
headers: dict | None = None,
cookies: list | None = None,
referrer_type: str | None = None,
request_timeout_ms: int = DEFAULT_REQUEST_TIMEOUT_MS,
tries: int = MAX_TRIES,
) -> Page:
"""Fetch one URL through Nimble Extract with retry and a completeness guard.
render=False cheap, raw server HTML (SSR pages, sitemaps, JSON endpoints)
render=True headless browser, JS runs (client-hydrated PDPs, grids)
wait_ms adds a {"wait": ms} browser action. ALWAYS set it on large
JSON (8000) and JS-hydrated PDPs (4000-5000); it is the main
defence against truncation.
sentinel substring/regex that proves the body arrived (a spec block
selector's class name, 'application/ld+json', '__NEXT_DATA__').
expect_json the body must parse as JSON (after <pre> unwrap).
Never raises for a failed fetch: inspect `.ok` / `.error`.
"""
client = get_client()
actions: list = []
if wait_ms:
actions.append({"wait": {"duration": int(wait_ms)}})
if browser_actions:
actions.extend(browser_actions)
kwargs: dict = dict(
url=url, render=render, formats=formats or ["html"], country=country,
locale=locale, driver=driver, request_timeout=request_timeout_ms,
)
# An empty browser_actions list is rejected (400 "must be a non-empty array"):
# only send the key when there is something to run.
if actions:
kwargs["browser_actions"] = actions
if headers:
kwargs["headers"] = headers
if cookies:
kwargs["cookies"] = cookies
if referrer_type:
kwargs["referrer_type"] = referrer_type
last_err = None
for attempt in range(1, tries + 1):
try:
_count()
resp = client.extract(**kwargs)
if resp.status == "success":
data = resp.data
html = (data.html or "") if data else ""
passes, why = _guard_ok(html, sentinel, expect_json)
if passes:
return Page(
ok=True, html=html,
fetch_results=list(getattr(data, "fetch", None) or []) if data else [],
cookies=list(getattr(data, "cookies", None) or []) if data else [],
task_id=resp.task_id, attempts=attempt,
)
last_err = f"guard failed: {why}"
else:
last_err = (f"status={resp.status!r} status_code={resp.status_code!r} "
f"warnings={resp.warnings!r}")
except Exception as exc: # noqa: BLE001 - transport errors vary by SDK version
last_err = f"{type(exc).__name__}: {exc}"
if attempt < tries:
pause = BACKOFF_BASE ** attempt
logger.warning("retry %d/%d %s in %.0fs (%s)", attempt, tries, url, pause, last_err)
time.sleep(pause)
logger.error("FAILED %s: %s", url, last_err)
return Page(ok=False, error=last_err, attempts=tries)
_PRE_RE = re.compile(r"<pre[^>]*>(.*)</pre>", re.DOTALL)
def json_from_html(html: str | None):
"""A JSON endpoint fetched through Extract arrives HTML-wrapped:
`<html>...<pre>{json}</pre></html>`, HTML-escaped. Unwrap and parse.
Falls back to parsing the whole body. Returns None if nothing parses."""
if not html:
return None
m = _PRE_RE.search(html)
raw = _htmllib.unescape(m.group(1)) if m else html
for candidate in (raw, html):
try:
return json.loads(candidate)
except (json.JSONDecodeError, TypeError):
continue
return None
def fetch_json(url: str, *, wait_ms: int | None = 8000, render: bool = False, **kw):
"""GET a JSON endpoint and return the parsed object, or None.
Defaults to an 8s wait: large JSON bodies are the most truncation-prone."""
page = fetch(url, render=render, wait_ms=wait_ms, expect_json=True, **kw)
if not page.ok:
return None
return json_from_html(page.html)
def browser_fetch(
page_url: str,
requests: list[dict],
*,
render: bool = True,
wait_ms: int = 4000,
between_ms: int = 800,
per_request_timeout_ms: int = 30_000,
request_timeout_ms: int = 240_000,
**kw,
) -> list:
"""Load `page_url`, then run in-browser fetch() calls from that origin.
Use when the target API needs a POST body, exact headers (Accept:
application/json, an X-Algolia-* key read from the page), or only answers
a real XHR. `requests` = [{url, method?, headers?, body?}]. Returns a list
aligned to `requests`: parsed JSON or None per call.
"""
actions: list = []
for req in requests:
f = {"url": req["url"], "method": req.get("method", "GET"),
"timeout": per_request_timeout_ms}
if req.get("headers"):
f["headers"] = req["headers"]
if req.get("body") is not None:
f["body"] = req["body"] if isinstance(req["body"], str) else json.dumps(req["body"])
actions.append({"fetch": f})
actions.append({"wait": {"duration": between_ms}})
page = fetch(page_url, render=render, wait_ms=wait_ms, browser_actions=actions,
request_timeout_ms=request_timeout_ms, **kw)
if not page.ok:
return [None] * len(requests)
out = []
for fr in page.fetch_results:
fr = fr if isinstance(fr, dict) else _to_dict(fr)
body, status = fr.get("body"), fr.get("status")
if status == 200 and body:
try:
out.append(json.loads(body) if isinstance(body, str) else body)
continue
except (json.JSONDecodeError, TypeError):
pass
logger.warning("browser_fetch: %s -> status=%s, %d bytes, not JSON",
fr.get("url"), status, len(body or ""))
out.append(None)
while len(out) < len(requests):
out.append(None)
return out
def clean_cookie(raw) -> dict:
"""Nimble's own get_cookies output fails its cookies= input validation
verbatim (sameSite capitalised, expires in a rejected shape). Keep only
what the API accepts; the jar is used immediately so expiry is moot."""
d = raw if isinstance(raw, dict) else _to_dict(raw)
out = {"name": d.get("name"), "value": d.get("value"),
"domain": d.get("domain"), "path": d.get("path") or "/"}
same_site = str(d.get("sameSite") or d.get("same_site") or "").lower()
if same_site in ("strict", "lax", "none"):
out["sameSite"] = same_site
if d.get("secure") is not None:
out["secure"] = bool(d.get("secure"))
return out
def warm_cookies(host: str, *, wait_ms: int = 6000, min_bytes: int = 50_000, **kw) -> list[dict]:
"""Visit `host` (https://www.example.com) with a browser, return a sanitised
cookie jar to pass as cookies= on later fetches. This replaces sticky
sessions (which hung 8-12 min then 500'd) for Akamai-style edges that serve
an empty app shell to a cold client."""
bare = host.split("//", 1)[-1].strip("/")
page = fetch(host.rstrip("/") + "/", render=True, wait_ms=wait_ms,
browser_actions=[{"get_cookies": {"domains": [bare, "." + bare]}}],
tries=1, **kw)
jar = [clean_cookie(c) for c in (page.cookies or [])]
jar = [c for c in jar if c.get("name")]
if not (page.ok and page.size >= min_bytes and jar):
logger.warning("warm_cookies(%s): ok=%s bytes=%d cookies=%d", host, page.ok, page.size, len(jar))
return jar
def _to_dict(obj):
if isinstance(obj, dict):
return obj
for m in ("model_dump", "to_dict", "dict"):
fn = getattr(obj, m, None)
if callable(fn):
try:
out = fn()
if isinstance(out, dict):
return out
except TypeError:
pass
return {k: v for k, v in vars(obj).items() if not k.startswith("_")} if hasattr(obj, "__dict__") else {}
# ----------------------------------------------------------------------------- CLI
def _main(argv=None) -> int:
ap = argparse.ArgumentParser(description="Probe one URL through Nimble Extract.")
ap.add_argument("url")
ap.add_argument("--render", action="store_true")
ap.add_argument("--wait", type=int, default=None, help="wait browser action, ms")
ap.add_argument("--sentinel", default=None, help="substring that must be present")
ap.add_argument("--json", action="store_true", help="body must parse as JSON")
ap.add_argument("--driver", default=DEFAULT_DRIVER)
ap.add_argument("--country", default="US")
ap.add_argument("--timeout", type=int, default=DEFAULT_REQUEST_TIMEOUT_MS)
ap.add_argument("--out", default=None, help="save HTML (or unwrapped JSON) here")
a = ap.parse_args(argv)
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
t0 = time.time()
page = fetch(a.url, render=a.render, wait_ms=a.wait, sentinel=a.sentinel,
expect_json=a.json, driver=a.driver, country=a.country,
request_timeout_ms=a.timeout)
took = time.time() - t0
html = page.html or ""
report = {
"ok": page.ok, "error": page.error, "attempts": page.attempts, "seconds": round(took, 1),
"bytes": page.size, "task_id": page.task_id,
"has_json_ld_product": bool(re.search(r'"@type"\s*:\s*"?\[?"?Product', html)),
"has_next_data": '__NEXT_DATA__' in html,
"has_pre_json": json_from_html(html) is not None if "<pre" in html[:2000] else False,
"title": (re.search(r"<title[^>]*>(.*?)</title>", html, re.S | re.I) or [None, ""])[1].strip()[:120]
if "<title" in html else "",
}
if a.out:
obj = json_from_html(html) if a.json else None
Path(a.out).parent.mkdir(parents=True, exist_ok=True)
Path(a.out).write_text(json.dumps(obj, indent=2) if obj is not None else html, encoding="utf-8")
report["saved"] = a.out
print(json.dumps(report, indent=2))
return 0 if page.ok else 1
if __name__ == "__main__":
sys.exit(_main())