""" Automatic first-pass trim of UNTRIMMED fetched blocks. Uses propose_cuts.py (the case-anchored proposer) to find the paragraph run that talks about this case, then widens it backwards across short non-junk gaps so a lede is not lost to a one-line subhead, and drops interior lines that are plainly site furniture (ALSO READ teasers, share prompts, advertisements). Every kept piece is verified as a verbatim substring of the staged file before it is written. The full extract stays on disk in staging/, so any cut can be revisited. This is a machine cut, not a review. Blocks trimmed here are marked trimmed_by = "auto (propose_cuts, unreviewed)" and the dossier prints a banner saying so. Pages where the proposer cannot find the case (fewer than MIN_HITS case-term hits) are left UNTRIMMED and listed in auto-trim-review.md; they are candidates for a wrong-source verdict and need a human. Run build_cases.py --all and verify_all.py afterwards. """ import argparse import hashlib import io import json import re import sys from pathlib import Path from propose_cuts import JUNK, PAYWALL, case_terms, score from trim import slice_paras BASE = Path(__file__).resolve().parent STORE = BASE / "sourced-articles.json" STAGING = BASE / "staging" JSONF = BASE.parent / "attacks-export-Gart-website.json" DECISIONS_OUT = BASE / "trim-decisions-auto.json" REVIEW_OUT = BASE / "auto-trim-review.md" MIN_HITS = 3 # at or above this the page is confidently about the case # 1-2 hits: trimmed, but listed as WEAK on the review list (thin records such as # "30 y/o man, Sweden" cannot score higher). 0 hits: left UNTRIMMED as possible OFF-CASE. BACK_GAP = 4 # merge an earlier run if separated by at most this many short lines # Interior furniture: short lines that sit inside an article body on many news sites. # Anchored and short-only so a real sentence that happens to contain "share" survives. INTERIOR = re.compile( r"^(also read|read (more|also|next)|related( (articles?|stories|news|coverage))?|" r"see also|more (on|from) |advertisement|advertising|publicidade|publicit\u00e9|" r"a lire aussi|\u00e0 lire aussi|lire aussi|leia tamb\u00e9m|lea tambi\u00e9n|siehe auch|" r"sign up|subscribe|share( this| via)?|loading|image[: ]|photo[: ]|picture[: ]|" r"illustration( of|:)|file photo|getty images|\(image|credit[: ]|source[: ]|" r"watch[: ]|listen[: ]|video[: ]|" r"recommended|trending|most read|top stories|editor'?s picks|popular|" r"follow us|join us|download the app|click here|tap here|" r"this article (is|was)|this story (is|was)|updated at|published (on|at)|" r"(mon|tue|wed|thu|fri|sat|sun)[a-z]*, [a-z]+\.? \d{1,2}, \d{4}|" r"editorial process|trusted editorial|ad disclosure|" r"let us know|got an opinion|email your thoughts|story tips|" r"poster sur|tweeter|envoyer via|partager|accueil \u00bb|compartilh|compartir|teilen|" r"by registering|for signing up|thank you for (signing|subscribing)|" r"copyright|all rights reserved|\u00a9)", re.I) # Teaser headline with a date suffix: "Some headline - 4. september 2025" / "- May 3, 2024" DATED_TEASER = re.compile(r" [-\u2013] (\d{1,2}\.? [A-Za-z\u00e6\u00f8\u00e5]+ \d{4}|[A-Z][a-z]+ \d{1,2}, \d{4})$") TAIL_WORDS = re.compile(r"(inbox|newsletter|briefing|unsubscribe|privacy policy|terms of (use|service)|" r"retningslinjer|debattskikk|^vi i document|^kj\u00f8p .* fra document|" r"cookie|comments? below|in the comments|support local news|pay what you want|" r"starting at us\$|join a community|become a member|donate|" r"^every (mon|tues|wednes|thurs|fri|satur|sun)day)", re.I) def listy(p): """A newline-packed paragraph of short fragments is a nav bar or a tag list.""" segs = [x for x in p.split("\n") if x.strip()] return len(segs) >= 4 and sum(len(x) for x in segs) / len(segs) < 30 def is_furniture(p): p = p.strip() if not p or listy(p) or p.startswith(("http://", "https://", "www.")) or DATED_TEASER.search(p): return True return len(p) <= 400 and bool(INTERIOR.match(p) or JUNK.search(p) or TAIL_WORDS.search(p)) MIN_KEEP = 600 # a cut that keeps less than this from a page 3x larger is refused GAP = 4 # furniture lines tolerated between two body runs def bodyish(p): """Body text, relaxed from propose_cuts: a short sentence is still a sentence. A 70-char line ending in a full stop is a closing sentence far more often than furniture (teasers are headlines and headlines do not end in full stops); the furniture that does end in a full stop is caught by JUNK / INTERIOR. """ p = p.strip() if not p or is_furniture(p): return False if len(p) >= 140: return True return len(p) >= 40 and bool(re.search(r"[.!?;:\u201d\u00bb\"')]$", p)) def gap_ok(p): p = p.strip() return len(p) <= 200 and not JUNK.search(p) def find_range(paras, terms): """Choose the body run that talks about this case and widen it across short gaps. Runs are maximal stretches of body paragraphs. The best run is the one with the most distinct case terms (density, then length, break ties), as in propose_cuts. It is then merged with neighbouring runs when the gap between them is at most GAP short non-junk lines: backwards unconditionally (a lede sits before a subhead), forwards only when the next run mentions the case or is short (a closing sentence or two), so a trailing block of sentence-shaped teasers is not absorbed. """ body = [bodyish(p) for p in paras] runs, cur = [], [] for i, b in enumerate(body): if b: cur.append(i) elif cur: runs.append(cur) cur = [] if cur: runs.append(cur) if not runs: return None, None def run_score(r): text = " ".join(paras[i] for i in r).lower() distinct = sum(1 for t in terms if t in text) chars = sum(len(paras[i]) for i in r) or 1 # Distinct terms first, then LENGTH. propose_cuts used density as the tie-break, # which let a six-line teaser list carrying one "Oslo" beat the 18-paragraph # article carrying the same one "Oslo" (Document.no, case 248). return (distinct, chars) k = max(range(len(runs)), key=lambda i: run_score(runs[i])) start, end = runs[k][0], runs[k][-1] def gap_between(a_end, b_start): between = list(range(a_end + 1, b_start)) return between if len(between) <= GAP and all(gap_ok(paras[i]) for i in between) else None j = k while j - 1 >= 0 and gap_between(runs[j - 1][-1], runs[j][0]) is not None: j -= 1 start = runs[j][0] j = k while j + 1 < len(runs) and gap_between(runs[j][-1], runs[j + 1][0]) is not None: nxt = runs[j + 1] gap = gap_between(runs[j][-1], nxt[0]) hits = sum(score(paras[i], terms) for i in nxt) # A one- or two-line gap is a subheading or a pull quote inside the article; # merge regardless of hits (later sections often carry no record term). A # longer gap looks like a teaser block, so the next run must earn its place. if len(gap) > 2 and hits == 0 and len(nxt) > 3: break j += 1 end = nxt[-1] while end > start and not bodyish(paras[end]): end -= 1 while start < end and not bodyish(paras[start]): start += 1 return start, end def reset_auto(store, only=()): """Put every auto-trimmed block back to UNTRIMMED from its staged file.""" n = 0 for cid, blocks in store.items(): if only and int(cid) not in only: continue for b in blocks: if b.get("verdict") == "KEEP" and str(b.get("trimmed_by", "")).startswith("auto"): path = STAGING / f"{int(cid):03d}" / b["staged_file"] text = path.read_text(encoding="utf-8") b.update(verdict="UNTRIMMED", text=text, chars=len(text), sha256=hashlib.sha256(text.encode("utf-8")).hexdigest(), paragraphs_kept=f"ALL {len(text.split(chr(10) * 2))} paragraphs, untrimmed", cut_note="NOT TRIMMED. Full page extract; trim on read-through.") b.pop("trimmed_by", None) n += 1 return n def decisions_from_store(store): """Reconstruct every auto-trim decision from the block fields the trimmer wrote (paragraphs_kept, cut_note, staged_file), plus anchors read from the staged file.""" out = {} for cid_s, blocks in store.items(): cid = int(cid_s) for b in blocks: if b.get("verdict") != "KEEP" or not str(b.get("trimmed_by", "")).startswith("auto"): continue pk = b.get("paragraphs_kept", "") m = re.search(r"(\d+)-(\d+)(?: excl \[([\d, ]*)\])? of (\d+)", pk) if not m: continue start, end, excl, total = int(m.group(1)), int(m.group(2)), m.group(3), int(m.group(4)) exclude = [int(x) for x in excl.split(",")] if excl else [] title = 0 if pk.startswith("headline [0]") else None h = re.search(r"hits on page: (\d+)", b.get("cut_note", "")) path = STAGING / f"{cid:03d}" / b["staged_file"] full = path.read_text(encoding="utf-8") if path.exists() else "" paras = full.split("\n\n") out.setdefault(cid_s, []).append({ "field": b["field"], "staged_file": b["staged_file"], "start": start, "end": end, "title": title, "exclude": exclude, "hits": int(h.group(1)) if h else None, "starts_with": paras[start][:45] if start < len(paras) else None, "ends_with": paras[end][-55:] if end < len(paras) else None, "chars_before": len(full), "chars_after": b.get("chars"), "paywall": "PAYWALL" in b.get("cut_note", "")}) return out def write_review(store, decisions): """Rebuild auto-trim-review.md from the whole store: every UNTRIMMED block with its reason, every auto-trimmed block cut on a weak anchor. Regenerated on each run, never appended.""" records = {r["id"]: r for r in json.loads(JSONF.read_text(encoding="utf-8"))} rows = [] for cid_s, blocks in store.items(): cid = int(cid_s) terms = case_terms(records[cid]) for b in blocks: if b.get("verdict") == "UNTRIMMED" and b.get("staged_file"): path = STAGING / f"{cid:03d}" / b["staged_file"] full = path.read_text(encoding="utf-8") paras = full.split("\n\n") hits = sum(score(p, terms) for p in paras) start, end = find_range(paras, terms) if start is None: reason = "no body run found" elif hits == 0: reason = "OFF-CASE? no case term appears anywhere on the page" else: reason = (f"TOO SMALL: the only body run found is tiny on a {len(full):,}-char page; " "page may be a script shell or list-form article") rows.append((cid, b["field"], b.get("source"), reason, b["url"])) for d in decisions.get(cid_s, []): if d.get("hits") is not None and d["hits"] < MIN_HITS: blk = next((b for b in blocks if b.get("field") == d["field"]), {}) rows.append((cid, d["field"], blk.get("source"), f"WEAK anchor: trimmed on only {d['hits']} case-term hit(s); confirm the page is about this case", blk.get("url"))) lines = ["# Auto-trim review list", "", "Regenerated from the whole store on every auto_trim.py run. OFF-CASE? blocks were left UNTRIMMED", "because no term from the record appears on the page; the linked URL may be the wrong source.", "TOO SMALL blocks were left UNTRIMMED because almost no body text was found on a large page.", "WEAK blocks were trimmed on one or two hits only, usually because the record is thin; confirm the", "page is about this case. The full extract stays in staging/ either way.", ""] for cid, field, src, reason, url in sorted(rows, key=lambda r: (r[0], r[1])): lines.append(f"- **case {cid}** `{field}` {src}: {reason}\n <{url}>") REVIEW_OUT.write_text("\n".join(lines) + "\n", encoding="utf-8") return len(rows) def main(): sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") ap = argparse.ArgumentParser() ap.add_argument("cases", nargs="*", type=int) ap.add_argument("--dry-run", action="store_true") ap.add_argument("--reset", action="store_true", help="first restore every auto-trimmed block to UNTRIMMED from staging/") args = ap.parse_args() records = {r["id"]: r for r in json.loads(JSONF.read_text(encoding="utf-8"))} store = json.loads(STORE.read_text(encoding="utf-8")) decisions, review, applied, skipped = {}, [], 0, 0 if args.reset: print(f"reset {reset_auto(store, args.cases)} auto-trimmed block(s) to UNTRIMMED") for cid_s, blocks in store.items(): cid = int(cid_s) if args.cases and cid not in args.cases: continue terms = case_terms(records[cid]) for b in blocks: if b.get("verdict") != "UNTRIMMED" or not b.get("staged_file"): continue path = STAGING / f"{cid:03d}" / b["staged_file"] full = path.read_text(encoding="utf-8") paras = full.split("\n\n") hits = sum(score(p, terms) for p in paras) start, end = find_range(paras, terms) if start is None or hits == 0: reason = ("no body run found" if start is None else "OFF-CASE? no case term appears anywhere on the page") review.append((cid, b["field"], b.get("source"), reason, b["url"])) skipped += 1 continue weak = hits < MIN_HITS if weak: review.append((cid, b["field"], b.get("source"), f"WEAK anchor: trimmed on only {hits} case-term hit(s); confirm " "the page is about this case", b["url"])) exclude = [i for i in range(start, end + 1) if is_furniture(paras[i])] title = None t0 = paras[0].strip() if start > 0 and 20 < len(t0) < 200 and not JUNK.search(t0): title = 0 text, pieces = slice_paras(paras, start, end, title, exclude) if any(p not in full for p in pieces): review.append((cid, b["field"], b.get("source"), "slice not a substring (bug)", b["url"])) skipped += 1 continue if len(text) < MIN_KEEP and len(full) > 3 * len(text): review.append((cid, b["field"], b.get("source"), f"TOO SMALL: the only body run found is {len(text)} chars of a " f"{len(full):,}-char page; page may be a script shell or list-form " "article. Left UNTRIMMED.", b["url"])) skipped += 1 continue dropped_head = start - (1 if title is not None else 0) dropped_tail = len(paras) - 1 - end pay = [p.strip()[:60] for p in paras if PAYWALL.search(p)] cut_note = (f"AUTO-TRIMMED, unreviewed: kept paragraphs {start}-{end}" + (f" plus headline [0]" if title is not None else "") + f"; dropped {dropped_head} leading and {dropped_tail} trailing paragraphs" + (f" and {len(exclude)} interior furniture line(s)" if exclude else "") + f". Case-term hits on page: {hits}" + (" (WEAK anchor, confirm the page is about this case)." if weak else ".") + (" PAYWALL marker seen on page." if pay else "")) decisions.setdefault(cid_s, []).append({ "field": b["field"], "staged_file": b["staged_file"], "start": start, "end": end, "title": title, "exclude": exclude, "hits": hits, "starts_with": paras[start][:45], "ends_with": paras[end][-55:], "chars_before": len(full), "chars_after": len(text), "paywall": bool(pay)}) if not args.dry_run: b.update(verdict="KEEP", text=text, chars=len(text), sha256=hashlib.sha256(text.encode("utf-8")).hexdigest(), paragraphs_kept=(f"headline [0] + " if title is not None else "") + f"{start}-{end}" + (f" excl {exclude}" if exclude else "") + f" of {len(paras)}", cut_note=cut_note, trimmed_by="auto (propose_cuts, unreviewed)") b.pop("carried_over", None) applied += 1 if not args.dry_run: STORE.write_text(json.dumps(store, ensure_ascii=False, indent=2), encoding="utf-8") # The audit trail is rebuilt from the store every run, so a run limited to a few # cases never drops the decisions for the rest. alld = decisions_from_store(store) DECISIONS_OUT.write_text(json.dumps(alld, ensure_ascii=False, indent=2), encoding="utf-8") n_rev = write_review(store, alld) print(f"audit trail: {sum(len(v) for v in alld.values())} auto-trim decisions across {len(alld)} cases; " f"{n_rev} entries on the review list") before = sum(d["chars_before"] for ds in decisions.values() for d in ds) after = sum(d["chars_after"] for ds in decisions.values() for d in ds) print(f"{'would trim' if args.dry_run else 'trimmed'} {applied} block(s): " f"{before:,} -> {after:,} chars; left UNTRIMMED for review: {skipped}") remaining = sum(1 for c in store.values() for b in c if b.get("verdict") == "UNTRIMMED") print(f"UNTRIMMED blocks in store now: {remaining}") if __name__ == "__main__": main()