""" Stage source text for one or more cases. Raw HTTP first, because that path is deterministic: the bytes come off the wire and the extracted text is hashed, so it can be re-verified later. Sites that refuse a raw request are recorded as BLOCKED and left for the WebFetch path, which is a model transcription and must be labelled as such. Nothing here writes to case files. Output goes to merge-output/staging/. """ import argparse import gzip import hashlib import json import re import urllib.error import urllib.request import zlib from datetime import datetime, timezone from html.parser import HTMLParser from pathlib import Path BASE = Path(__file__).resolve().parent JSONF = BASE.parent / "attacks-export-Gart-website.json" STAGING = BASE / "staging" UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36") NOT_ARTICLE = re.compile(r"(youtube\.com|youtu\.be|x\.com|twitter\.com|instagram\.com|" r"facebook\.com|tiktok\.com|t\.me)", re.I) # Furniture that survives tag-stripping and has to go by content, not markup. JUNK = re.compile(r"^(facebook|instagram|linkedin|rss|twitter|x|whatsapp|telegram|share|" r"compartilhar|publicidade|advertisement|newsletter|cookies?|assine|" r"leia (também|mais)|read more|related( articles)?|tags?|" r"siga o|follow us|subscribe)\b", re.I) class Extract(HTMLParser): """Pull block-level text. Deliberately simple. Suppression is tracked with a stack of the tag names that opened it, so a container that never closes cleanly cannot swallow the rest of the document. Furniture is removed by content afterwards, not by guessing at CSS class names, because class conventions differ on every site and a wrong guess here fails silently by returning an empty page. """ DROP = {"script", "style", "noscript", "nav", "header", "footer", "aside", "form", "svg", "button", "figure", "iframe", "select"} BLOCK = {"p", "h1", "h2", "h3", "h4", "li", "blockquote", "div", "section", "br"} def __init__(self): super().__init__(convert_charrefs=True) self.stack = [] self.cur = [] self.blocks = [] def handle_starttag(self, tag, attrs): if tag in self.DROP: self.stack.append(tag) return if tag in self.BLOCK: self._flush() def handle_endtag(self, tag): if self.stack and self.stack[-1] == tag: self.stack.pop() return if not self.stack and tag in self.BLOCK: self._flush() def handle_data(self, data): if not self.stack: self.cur.append(data) def _flush(self): if self.cur: t = re.sub(r"[ \t\xa0]+", " ", "".join(self.cur)).strip() if t: self.blocks.append(t) self.cur = [] def text(self): self._flush() out, seen = [], set() for t in self.blocks: if len(t) < 40 and not re.search(r"[.!?]$", t): continue # nav labels, buttons, breadcrumbs if JUNK.match(t): continue if t in seen: continue seen.add(t) out.append(t) return "\n\n".join(out) def fetch_raw(url, timeout=30): req = urllib.request.Request(url, headers={ "User-Agent": UA, "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en,pt;q=0.9,fr;q=0.8,id;q=0.7,es;q=0.6", }) with urllib.request.urlopen(req, timeout=timeout) as r: raw = r.read() enc = (r.headers.get("Content-Encoding") or "").lower() if enc == "gzip": raw = gzip.decompress(raw) elif enc == "deflate": raw = zlib.decompress(raw) charset = r.headers.get_content_charset() or "utf-8" return r.status, raw.decode(charset, errors="replace"), r.geturl() WAYBACK_AVAIL = "https://archive.org/wayback/available?url=" def fetch_wayback(url, timeout=40): """Fetch the closest Wayback Machine snapshot of a URL. Returns (timestamp, snapshot_url, html) or (None, None, None) when nothing is archived. The `id_` flag asks for the original bytes without the Wayback toolbar, so extraction runs on the same markup a live fetch would have seen. A snapshot is a third party's copy taken on a given date, so anything staged from it must say so. """ import urllib.parse q = WAYBACK_AVAIL + urllib.parse.quote(url, safe="") with urllib.request.urlopen(urllib.request.Request(q, headers={"User-Agent": UA}), timeout=timeout) as r: closest = json.loads(r.read().decode("utf-8")).get("archived_snapshots", {}).get("closest") if not closest or not closest.get("available"): return None, None, None ts = closest["timestamp"] raw_url = f"https://web.archive.org/web/{ts}id_/{url}" status, html, final = fetch_raw(raw_url, timeout=timeout) return ts, closest["url"], html def stage_case(rec, extra_urls=(), new_only=False): cid = rec["id"] out = STAGING / f"{cid:03d}" out.mkdir(parents=True, exist_ok=True) entries = [] # Re-fetching a source that is already staged is pure cost, and worse: live pages # add and drop teasers, so the paragraph indices in an approved cut silently shift. # With --new-only, anything already fetched is carried over untouched. prior = {} idx_path = out / "index.json" if new_only and idx_path.exists(): prior = {e["field"]: e for e in json.loads(idx_path.read_text(encoding="utf-8"))} urls = [(f, rec[f]) for f in ("url", "url_2", "url_3", "url_4", "url_5") if rec.get(f)] urls += [(f"extra_{i+1}", u) for i, u in enumerate(extra_urls)] for n, (field, url) in enumerate(urls, 1): host = re.sub(r"^https?://(www\.)?([^/]+).*", r"\2", url) if field in prior and prior[field].get("verdict") in ("FETCHED", "NOT AN ARTICLE"): carried = dict(prior[field]) carried["carried_over"] = True entries.append(carried) continue e = {"case": cid, "n": n, "field": field, "url": url, "host": host, "fetched_at": datetime.now(timezone.utc).isoformat(timespec="seconds")} if NOT_ARTICLE.search(url): e.update(verdict="NOT AN ARTICLE", method="none", note="video or social post; recorded as a source reference, not transcribed") entries.append(e) continue try: status, html, final = fetch_raw(url) text = Extract() text.feed(html) body = text.text() if len(body) < 400: e.update(verdict="THIN", method="raw", http=status, chars=len(body), note="extractor recovered very little; needs the WebFetch path or a " "site-specific rule") else: e.update(verdict="FETCHED", method="raw", http=status, chars=len(body), sha256=hashlib.sha256(body.encode()).hexdigest(), final_url=final if final != url else None) (out / f"{n:02d}_{re.sub(r'[^a-z0-9]+', '-', host.lower())}.txt").write_text( body, encoding="utf-8") except urllib.error.HTTPError as ex: e.update(verdict="BLOCKED" if ex.code in (401, 403, 429) else "DEAD", method="raw", http=ex.code, note="raw request refused; try the WebFetch path, which is a model " "transcription and must be labelled as such") except Exception as ex: e.update(verdict="ERROR", method="raw", note=f"{type(ex).__name__}: {str(ex)[:120]}") entries.append(e) (out / "index.json").write_text(json.dumps(entries, ensure_ascii=False, indent=2), encoding="utf-8") return entries def main(): ap = argparse.ArgumentParser() ap.add_argument("cases", nargs="+", type=int) ap.add_argument("--extra", action="append", default=[], help="CASEID=URL for a source not on the record") ap.add_argument("--new-only", action="store_true", help="skip sources already staged; use this whenever adding a source to " "a case that has approved cuts, or their paragraph indices will drift") args = ap.parse_args() extra = {} for spec in args.extra: cid, _, url = spec.partition("=") extra.setdefault(int(cid), []).append(url) records = {r["id"]: r for r in json.loads(JSONF.read_text(encoding="utf-8"))} STAGING.mkdir(exist_ok=True) for cid in args.cases: entries = stage_case(records[cid], extra.get(cid, ()), new_only=args.new_only) print(f"\ncase {cid}") for e in entries: bits = "carried" if e.get("carried_over") else (f"HTTP {e.get('http')}" if e.get("http") else "") size = f"{e['chars']:,} chars" if e.get("chars") else "" print(f" [{e['verdict']:<15}] {e['field']:<8} {e['host']:<26} {bits:<9} {size}") if e.get("note"): print(f" {e['note']}") if __name__ == "__main__": main()