Measure six ways to cut a scraping agent’s HTML input without losing card fields

if ($PSVersionTable.PSVersion -lt [version]'7.4') { throw 'PowerShell 7.4 or later required for JSON redirection' }
python -m venv .venv
.venv/Scripts/python -m pip install -r requirements.txt
.venv/Scripts/python acquire_pages.py --out owned-pages
.venv/Scripts/python -X utf8 reference_check.py owned-pages/page1.html owned-pages/page2.html owned-pages/page3.html > reader-reference.json

.venv/Scripts/python -X utf8 token_budget.py owned-pages/page1.html owned-pages/page2.html owned-pages/page3.html --state-dir reader-state --reference reader-reference.json --max-batch-input-tokens 12000 > cold.json
.venv/Scripts/python -X utf8 token_budget.py owned-pages/page1.html owned-pages/page2.html owned-pages/page3.html --state-dir reader-state --reference reader-reference.json --max-batch-input-tokens 12000 > warm.json

"""Optional reader-run acquisition of three public practice listing pages only."""
import argparse
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import time
from urllib import robotparser
from urllib.request import HTTPRedirectHandler, Request, build_opener

ORIGIN = "https://practice.scrapingcentral.com"
URLS = [ORIGIN + "/blog", ORIGIN + "/blog/page/2", ORIGIN + "/blog/page/3"]
AGENT = "BigChangeGuideReader/1.0"
MAX_BYTES = 1_000_000


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise ValueError(f"Redirect refused: HTTP {code}")


def fetch(opener, url):
    request = Request(url, headers={"User-Agent": AGENT, "Accept": "text/html,text/plain"})
    with opener.open(request, timeout=15) as response:
        if response.status != 200 or response.geturl() != url:
            raise ValueError(f"Unexpected response for {url}: {response.status}")
        body = response.read(MAX_BYTES + 1)
    if len(body) > MAX_BYTES:
        raise ValueError(f"Response exceeds one MB: {url}")
    return body


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", type=Path, required=True)
    args = ap.parse_args()
    out = args.out
    if out.exists() and any(out.iterdir()):
        raise ValueError("Use a new empty output directory")
    out.mkdir(parents=True, exist_ok=True)
    opener = build_opener(NoRedirect())
    robots_url = ORIGIN + "/robots.txt"
    robots_text = fetch(opener, robots_url).decode("utf-8", errors="strict")
    rules = robotparser.RobotFileParser()
    rules.parse(robots_text.splitlines())
    if not all(rules.can_fetch(AGENT, url) for url in URLS):
        raise ValueError("Current robots policy does not permit all three listing URLs")
    rate = rules.request_rate(AGENT)
    interval = max(1.0, float(rules.crawl_delay(AGENT) or 0),
                   float(rate.seconds / rate.requests) if rate and rate.requests else 0)
    last_start = time.monotonic()
    rows = []
    for number, url in enumerate(URLS, 1):
        time.sleep(max(0, interval - (time.monotonic() - last_start)))
        last_start = time.monotonic()
        body = fetch(opener, url)
        path = out / f"page{number}.html"
        path.write_bytes(body)
        rows.append({"url": url, "file": path.name, "httpStatus": 200,
                     "sha256": hashlib.sha256(body).hexdigest(),
                     "retrievedAtUtc": datetime.now(timezone.utc).isoformat()})
    (out / "acquisition.json").write_text(json.dumps({"robotsUrl": robots_url,
        "minimumStartSpacingSeconds": interval, "pages": rows}, indent=2) + "\n", encoding="utf-8")
    print(f"Saved {len(rows)} listing pages; inspect acquisition.json and the visible HTML")


if __name__ == "__main__":
    main()

"""Independent candidate reference from reader-owned fixed listing HTML.

Uses bounded regular-expression inspection, not token_budget's DOM or selectors.
The reader must compare every emitted field with the visible source before use.
"""
import argparse
import datetime as dt
import html
import json
from pathlib import Path
import re
from urllib.parse import urljoin, urlparse

ORIGIN = "https://practice.scrapingcentral.com"


def required(pattern, text, label):
    matches = re.findall(pattern, text, flags=re.S)
    if len(matches) != 1:
        raise ValueError(f"Expected one {label}; found {len(matches)}")
    return matches[0]


def clean(text):
    return html.unescape(re.sub(r"<[^>]*>", "", text)).strip()


def inspect_page(path):
    raw = Path(path).read_bytes()
    if len(raw) > 1_000_000:
        raise ValueError("Page exceeds one MB")
    text = raw.decode("utf-8", errors="strict")
    number = int(required(r"<title>Blog page ([1-3]) of [1-9][0-9]* - Catalog108</title>", text, "page title"))
    url = ORIGIN + ("/blog" if number == 1 else f"/blog/page/{number}")
    canonical = required(r'<link rel="canonical" href="([^"]+)">', text, "canonical")
    if canonical != url:
        raise ValueError("Canonical disagrees with page title")
    listing = required(r'<ul class="post-list">(.*?)</ul>', text, "post list")
    cards = re.findall(r'<li class="post-summary">(.*?)</li>', listing, flags=re.S)
    if not cards or re.sub(r'<li class="post-summary">.*?</li>', "", listing, flags=re.S).strip():
        raise ValueError("Unknown or empty post list")
    rows = []
    for card in cards:
        href, title = required(r'<h2><a href="([^"]+)">(.*?)</a></h2>', card, "title link")
        meta = required(r'<p class="post-meta">(.*?)</p>', card, "post meta")
        author = clean(required(r'<a[^>]*>(.*?)</a>', meta, "author"))
        date = required(r'<time datetime="([^"]+)">', meta, "date")
        dt.date.fromisoformat(date)
        excerpt = clean(required(r'<p class="post-excerpt">(.*?)</p>', card, "excerpt"))
        tags_block = required(r'<p class="post-tags">(.*?)</p>', card, "tags block")
        # Every link's visible value is retained regardless of its CSS class.
        tags = [clean(x).lstrip("#") for x in re.findall(r'<a\b[^>]*>(.*?)</a>', tags_block, flags=re.S)]
        if re.sub(r'<a\b[^>]*>.*?</a>', "", tags_block, flags=re.S).strip():
            raise ValueError("Unexpected tag-block content")
        record_url = urljoin(ORIGIN, href)
        parsed = urlparse(record_url)
        if parsed.scheme != "https" or parsed.netloc != urlparse(ORIGIN).netloc or not parsed.path.startswith("/blog/"):
            raise ValueError("Invalid record URL")
        row = {"url": record_url, "title": clean(title), "author": author,
               "publishedDate": date, "excerpt": excerpt, "tags": tags}
        if not all(row[k] for k in ("url", "title", "author", "publishedDate", "excerpt")) or any(not t for t in tags):
            raise ValueError("Empty required field")
        rows.append(row)
    if len({r["url"] for r in rows}) != len(rows):
        raise ValueError("Duplicate record URL on page")
    return f"page-{number}", rows


def build(paths):
    if not 1 <= len(paths) <= 3:
        raise ValueError("Supply one to three pages")
    result = {}
    seen = set()
    for path in paths:
        page_id, rows = inspect_page(path)
        if page_id in result or any(r["url"] in seen for r in rows):
            raise ValueError("Repeated page or record identity")
        result[page_id] = rows
        seen.update(r["url"] for r in rows)
    return result


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("pages", nargs="+", type=Path)
    args = ap.parse_args()
    print(json.dumps(build(args.pages), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

"""Offline token-input comparison for a fixed, permitted HTML listing corpus.

No model or network call. `measure` accepts reader-owned local HTML only.
"""
import argparse
import datetime as dt
import hashlib
import html
from html.parser import HTMLParser
import importlib.metadata
import json
from pathlib import Path
import re
import sys
from urllib.parse import urljoin, urlparse

import tiktoken

ORIGIN = "https://practice.scrapingcentral.com"
ENCODING = "cl100k_base"
PARSER_VERSION = "catalog108-listing-v2"
POLICY = {"fields": ["url", "title", "author", "publishedDate", "excerpt", "tags"],
          "minimumCards": 1, "allowedOrigin": ORIGIN, "strictLayout": True}
MODEL_CONFIG = {"task": "extract blog cards", "schema": "catalog108-card-v1",
                "promptVersion": "v1", "encoding": ENCODING}
INSTRUCTIONS = ("Extract every blog card. Return JSON per page ID with url, title, author, "
                "publishedDate, excerpt and tags. Preserve each page ID. Do not invent a field.\n")
VOID = {"area", "base", "br", "col", "embed", "hr", "img", "input", "link", "meta", "param", "source", "track", "wbr"}


def canonical(obj):
    return json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(",", ":"))


def sha(data):
    return hashlib.sha256(data).hexdigest()


class Node:
    def __init__(self, tag="root", attrs=()):
        self.tag, self.attrs, self.children = tag, dict(attrs), []

    def descendants(self, tag=None, class_name=None):
        for child in self.children:
            if isinstance(child, Node):
                if (tag is None or child.tag == tag) and (class_name is None or class_name in child.attrs.get("class", "").split()):
                    yield child
                yield from child.descendants(tag, class_name)

    def text(self):
        return "".join(c.text() if isinstance(c, Node) else c for c in self.children).strip()

    def markup(self):
        attrs = "".join(f' {k}="{html.escape(v or "", quote=True)}"' for k, v in self.attrs.items())
        inner = "".join(c.markup() if isinstance(c, Node) else html.escape(c) for c in self.children)
        return f"<{self.tag}{attrs}>" + ("" if self.tag in VOID else inner + f"</{self.tag}>")


class DOM(HTMLParser):
    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.root = Node()
        self.stack = [self.root]

    def handle_starttag(self, tag, attrs):
        node = Node(tag, attrs)
        self.stack[-1].children.append(node)
        if tag not in VOID:
            self.stack.append(node)

    def handle_startendtag(self, tag, attrs):
        self.handle_starttag(tag, attrs)
        if tag not in VOID:
            self.stack.pop()

    def handle_endtag(self, tag):
        for i in range(len(self.stack) - 1, 0, -1):
            if self.stack[i].tag == tag:
                del self.stack[i:]
                return

    def handle_data(self, data):
        self.stack[-1].children.append(data)


def one(node, tag, class_name=None):
    found = list(node.descendants(tag, class_name))
    if len(found) != 1:
        raise ValueError(f"Expected exactly one {tag}.{class_name or '*'}, got {len(found)}")
    return found[0]


def page_identity(raw):
    """Derive stable provenance from two independent assertions in the HTML."""
    dom = DOM()
    dom.feed(raw.decode("utf-8", errors="strict"))
    dom.close()
    title = one(dom.root, "title").text()
    match = re.fullmatch(r"Blog page ([1-3]) of [1-9][0-9]* - Catalog108", title)
    if not match:
        raise ValueError("Unknown listing page title")
    number = int(match.group(1))
    source_id = f"{ORIGIN}/blog" if number == 1 else f"{ORIGIN}/blog/page/{number}"
    canonicals = [n for n in dom.root.descendants("link") if "canonical" in n.attrs.get("rel", "").split()]
    if len(canonicals) != 1 or canonicals[0].attrs.get("href") != source_id:
        raise ValueError("Canonical URL does not match listing title")
    return f"page-{number}", source_id


def parse(raw, page_id):
    actual_id, _ = page_identity(raw)
    if actual_id != page_id:
        raise ValueError("Page identity does not match HTML")
    dom = DOM()
    dom.feed(raw.decode("utf-8", errors="strict"))
    dom.close()
    listing = one(dom.root, "ul", "post-list")
    cards = [c for c in listing.children if isinstance(c, Node)]
    if not cards or any(c.tag != "li" or "post-summary" not in c.attrs.get("class", "").split() for c in cards):
        raise ValueError("Unknown or empty post-list; refusing partial extraction")
    if any(isinstance(c, str) and c.strip() for c in listing.children):
        raise ValueError("Unexpected post-list text")
    rows = []
    for c in cards:
        link = one(one(c, "h2"), "a")
        author = one(one(c, "p", "post-meta"), "a").text()
        date = one(c, "time").attrs.get("datetime", "")
        dt.date.fromisoformat(date)
        href = link.attrs.get("href", "")
        url = urljoin(ORIGIN, href)
        p = urlparse(url)
        if p.scheme != "https" or p.netloc != urlparse(ORIGIN).netloc or not p.path.startswith("/blog/") or p.query or p.fragment:
            raise ValueError("Off-site or invalid card URL")
        tags_node = one(c, "p", "post-tags")
        tag_links = [x for x in tags_node.children if isinstance(x, Node)]
        if any(x.tag != "a" or "tag" not in x.attrs.get("class", "").split()
               or not x.attrs.get("href", "").startswith("/blog/tag/") or not x.text().startswith("#")
               for x in tag_links):
            raise ValueError("Unsupported tag link layout")
        if any(isinstance(x, str) and x.strip() for x in tags_node.children):
            raise ValueError("Unexpected tag text")
        row = {"url": url, "title": link.text(), "author": author,
               "publishedDate": date, "excerpt": one(c, "p", "post-excerpt").text(),
               "tags": [a.text().lstrip("#") for a in tag_links]}
        if not all(row[k] for k in ("url", "title", "author", "publishedDate", "excerpt")):
            raise ValueError("Blank required field")
        rows.append(row)
    if len({r["url"] for r in rows}) != len(rows):
        raise ValueError("Duplicate card identity")
    # The intermediate HTML is the fixed, selected listing, not arbitrary page chrome.
    return rows, listing.markup()


def payload(page_id, source_id, raw_hash, source, cache_key_value):
    # Count the complete bytes we would put in a model input, including
    # provenance and the cache decision metadata, as one serialized JSON value.
    return canonical({"pageId": page_id, "sourceId": source_id,
                      "rawSha256": raw_hash, "cacheKey": cache_key_value,
                      "source": source})


def input_text(parts):
    return INSTRUCTIONS + "\n".join(parts)


def tokens(s, enc):
    return len(enc.encode(s, disallowed_special=()))


def batches(entries, enc, maximum):
    groups, current = [], []
    for item in entries:
        candidate = current + [item]
        if tokens(input_text(candidate), enc) > maximum:
            if not current:
                raise ValueError("One page exceeds the batch input token limit")
            groups.append(current)
            current = [item]
            if tokens(input_text(current), enc) > maximum:
                raise ValueError("One page exceeds the batch input token limit")
        else:
            current = candidate
    if current:
        groups.append(current)
    return groups


def cache_key(page_id, raw, policy=POLICY, model=MODEL_CONFIG, parser=PARSER_VERSION):
    return sha(canonical({"pageId": page_id, "rawSha256": sha(raw), "parser": parser,
                          "policy": policy, "model": model, "instructionsSha256": sha(INSTRUCTIONS.encode()),
                          "implementationSha256": sha(Path(__file__).read_bytes()),
                          "tokenizer": {"package": "tiktoken", "version": importlib.metadata.version("tiktoken"),
                                        "encoding": ENCODING}}).encode())


def cache_digest(rows, filtered):
    return sha(canonical({"rows": rows, "filtered": filtered}).encode())


def run(paths, state_dir, max_batch_tokens=12000, policy=POLICY, model=MODEL_CONFIG, parser=PARSER_VERSION,
        reference=None):
    if not 1 <= len(paths) <= 3:
        raise ValueError("Supply one to three fixed listing pages")
    enc = tiktoken.get_encoding(ENCODING)
    state_dir.mkdir(parents=True, exist_ok=True)
    results, raw_inputs, filtered_inputs, structured_inputs, misses = [], [], [], [], []
    seen_ids = set()
    seen_pages = set()
    for path in paths:
        raw = path.read_bytes()
        if len(raw) > 1_000_000:
            raise ValueError("HTML page exceeds one MB")
        page_id, source_id = page_identity(raw)
        if page_id in seen_pages:
            raise ValueError("Duplicate listing page identity")
        seen_pages.add(page_id)
        key = cache_key(page_id, raw, policy, model, parser)
        cache = state_dir / f"{page_id}.json"
        saved = json.loads(cache.read_text(encoding="utf-8")) if cache.exists() else None
        hit = saved is not None and saved.get("key") == key
        if hit:
            rows, filtered = saved["rows"], saved["filtered"]
            if not rows or not filtered or saved.get("digest") != cache_digest(rows, filtered):
                raise ValueError("Invalid cached result")
        else:
            rows, filtered = parse(raw, page_id)
        for row in rows:
            if row["url"] in seen_ids:
                raise ValueError("Card identity appears on more than one page")
            seen_ids.add(row["url"])
        if not hit:
            # A failed parse or cross-page identity check leaves this page's old
            # entry intact; earlier validated pages can be reused on a rerun.
            tmp = cache.with_suffix(".tmp")
            tmp.write_text(canonical({"key": key, "rows": rows, "filtered": filtered,
                                      "digest": cache_digest(rows, filtered)}), encoding="utf-8")
            tmp.replace(cache)
            misses.append(page_id)
        results.append({"pageId": page_id, "sourceId": source_id, "rawSha256": sha(raw),
                        "rowCount": len(rows), "rows": rows, "cacheHit": hit})
        raw_inputs.append(payload(page_id, source_id, sha(raw), raw.decode("utf-8"), key))
        filtered_inputs.append(payload(page_id, source_id, sha(raw), filtered, key))
        structured_inputs.append(payload(page_id, source_id, sha(raw), rows, key))
    if reference is not None:
        expected = json.loads(Path(reference).read_text(encoding="utf-8"))
        actual = {p["pageId"]: p["rows"] for p in results}
        if actual != expected:
            raise ValueError("Extraction differs from independently checked reference")
    groups = batches(structured_inputs, enc, max_batch_tokens)
    miss_entries = [structured_inputs[i] for i, r in enumerate(results) if not r["cacheHit"]]
    miss_groups = batches(miss_entries, enc, max_batch_tokens)
    count = lambda groups: sum(tokens(input_text(g), enc) for g in groups)
    metrics = {"rawPerPage": count([[x] for x in raw_inputs]),
               "filteredPerPage": count([[x] for x in filtered_inputs]),
               "fixedSelectorJsonPerPage": count([[x] for x in structured_inputs]),
               "structuredJsonBatched": count(groups),
               "incrementalChangedOnly": count(miss_groups),
               "unchangedPagesSkipped": len(results) - len(misses),
               "resultCacheHits": len(results) - len(misses)}
    return {"measurement": "serialized-input-token proxy only; no model calls, output tokens, billing or answer-accuracy measurement",
            "tokenizer": {"package": "tiktoken", "version": importlib.metadata.version("tiktoken"), "encoding": ENCODING},
            "modelConfig": model, "policy": policy, "parserVersion": parser,
            "maxBatchInputTokens": max_batch_tokens, "batchPageIds": [[json.loads(x)["pageId"] for x in g] for g in groups],
            "changedPageIds": misses, "metrics": metrics, "pages": results}


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("pages", nargs="+", type=Path)
    ap.add_argument("--state-dir", type=Path, required=True)
    ap.add_argument("--max-batch-input-tokens", type=int, default=12000)
    ap.add_argument("--reference", type=Path)
    args = ap.parse_args()
    result = run(args.pages, args.state_dir, args.max_batch_input_tokens, reference=args.reference)
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

tiktoken==0.14.0