""" Track A expansion: assemble 18 additional documents per DESIGN_v2.md §3. Adds 8 to Stratum A, 4 to Stratum B, 6 to Stratum C for target n=30. Design discipline: - URLs declared before any fetch (DESIGN_v2 §3). - Single primary URL per document; fallback map also declared. - If both primary and fallback fail, document excluded from n=30. - If total n < 27 after expansion, halt per DESIGN_v2 §7. - Updates corpus/manifest.json with new entries; preserves original 12. """ from __future__ import annotations import hashlib import json import os import re import subprocess import sys import urllib.error import urllib.request from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[2] STUDY_DIR = REPO_ROOT / "fvs_eval" / "validation_study" CORPUS_DIR = STUDY_DIR / "corpus" def _strip_html(html: str, content_selector: str | None = None) -> str: """Minimal HTML-to-text extraction. If content_selector is given as a div id or class, narrow to that div before stripping.""" if content_selector: m = re.search( rf'<(?:div|article|section)[^>]*(?:id|class)=["\'][^"\']*{re.escape(content_selector)}[^"\']*["\'][^>]*>(.*?)', html, flags=re.DOTALL | re.IGNORECASE, ) if m: html = m.group(1) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"", "\n", html, flags=re.IGNORECASE) html = re.sub(r"<(br|hr)[^>]*/?>", "\n", html, flags=re.IGNORECASE) text = re.sub(r"<[^>]+>", "", html) for old, new in [ ("&", "&"), ("<", "<"), (">", ">"), (""", '"'), ("'", "'"), ("'", "'"), (" ", " "), ("'", "'"), ("—", "-"), ("–", "-"), ("’", "'"), ("‘", "'"), ("“", '"'), ("”", '"'), ]: text = text.replace(old, new) lines = [ln.strip() for ln in text.split("\n")] lines = [ln for ln in lines if ln] return "\n\n".join(lines) def _curl(url: str, timeout: int = 30) -> str: result = subprocess.run( ["curl", "-sSL", "--max-time", str(timeout), "-A", "FrameCheck-ValidationStudy/1.0 (research)", url], capture_output=True, text=True, timeout=timeout + 15, ) if result.returncode != 0 or not result.stdout: raise RuntimeError(f"curl failed: rc={result.returncode} stderr={result.stderr[:200]}") return result.stdout def _truncate_to_paragraphs(text: str, max_chars: int = 6000) -> str: if len(text) <= max_chars: return text cut = text[:max_chars] last_para = cut.rfind("\n\n") if last_para > max_chars // 2: return cut[:last_para] return cut # ============================================================================ # Stratum A: 8 new public human-authored documents # ============================================================================ STRATUM_A_DECLARED = [ { "slug": "a03_pg_how_to_do_great_work", "primary_url": "https://www.paulgraham.com/greatwork.html", "fallback_url": "https://www.paulgraham.com/own.html", "title": "Paul Graham essay: How To Do Great Work", "notes": "Personal essay by venture capitalist and essayist. Long-form, reflective.", }, { "slug": "a04_senate_ai_oversight", "primary_url": "https://www.judiciary.senate.gov/imo/media/doc/2023-05-16_-_testimony_-_altman.pdf", "fallback_url": "https://www.congress.gov/118/chrg/CHRG-118shrg52706/CHRG-118shrg52706.htm", "title": "Senate Judiciary testimony on AI oversight (2023)", "notes": "Public congressional testimony transcript.", "plain_text_fallback": "https://en.wikipedia.org/wiki/Sam_Altman", }, { "slug": "a05_un_sg_future_generations", "primary_url": "https://www.un.org/sg/en/content/sg/statement/2024-09-22/secretary-generals-remarks-the-summit-of-the-future", "fallback_url": "https://www.un.org/en/un75/new-era-conflict-and-violence", "title": "UN Secretary-General speech (public)", "notes": "Public international policy address.", }, { "slug": "a06_whitehouse_press_statement", "primary_url": "https://www.whitehouse.gov/briefing-room/statements-releases/", "fallback_url": "https://www.whitehouse.gov/briefing-room/press-briefings/", "title": "White House press statement (most recent available)", "notes": "Public government press release.", }, { "slug": "a07_ecb_lagarde_speech", "primary_url": "https://www.ecb.europa.eu/press/key/date/2025/html/ecb.sp250130~8f2e94c4c4.en.html", "fallback_url": "https://www.ecb.europa.eu/press/key/html/index.en.html", "title": "ECB President speech (public)", "notes": "Public central bank communication.", }, { "slug": "a08_sec_8k_press_release", "primary_url": "https://investor.apple.com/news-and-events/news-details/2024/Apple-Reports-First-Quarter-Results/default.aspx", "fallback_url": "https://investor.microsoft.com/investor/news-and-events/news-details/2024/Microsoft-Cloud-Strength-Drives-Second-Quarter-Results/default.aspx", "title": "Public earnings press release (SEC-filed 8-K source)", "notes": "Public business communication.", }, { "slug": "a09_wolfram_blog_post", "primary_url": "https://writings.stephenwolfram.com/2023/02/what-is-chatgpt-doing-and-why-does-it-work/", "fallback_url": "https://writings.stephenwolfram.com/", "title": "Stephen Wolfram blog: What Is ChatGPT Doing (long-form technical essay)", "notes": "Public technical-reflective essay by Stephen Wolfram.", }, { "slug": "a10_nber_abstract", "primary_url": "https://www.nber.org/papers/w31161", "fallback_url": "https://www.nber.org/papers", "title": "NBER working paper abstract (economics)", "notes": "Public academic research abstract.", }, ] def _fetch_stratum_a_doc(spec: dict) -> str: """Try primary, then fallback, then plain_text_fallback (if listed). Return extracted text.""" for key in ("primary_url", "fallback_url", "plain_text_fallback"): url = spec.get(key) if not url: continue try: html = _curl(url, timeout=30) text = _strip_html(html) text = _truncate_to_paragraphs(text, max_chars=6000) word_count = len(text.split()) if word_count >= 200: return text, url else: continue except Exception: continue raise RuntimeError(f"all fetch paths failed for {spec['slug']}") # ============================================================================ # Stratum B: 4 new AI generations # ============================================================================ STRATUM_B_NEW_PROMPTS = [ { "slug": "b07_traditional_publishing", "topic": "creative / cultural", "prompt": "Discuss the state and outlook of traditional publishing in 2026. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b08_ai_medical_ethics", "topic": "legal / ethical", "prompt": "Analyze the ethical considerations of AI systems in medical diagnosis. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b09_online_degrees", "topic": "educational", "prompt": "Describe how universities should evaluate online-only degree programs. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b10_carbon_accounting", "topic": "environmental", "prompt": "Discuss the economic case for corporate carbon-accounting disclosure. Write a considered written analysis, approximately 400-500 words.", }, ] def _call_anthropic(prompt: str, max_tokens: int = 1200) -> dict: api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: raise RuntimeError("ANTHROPIC_API_KEY not set") system_prompt = ( "You are a helpful assistant. Respond to the user's analytical " "prompt with a considered written analysis." ) candidates = ["claude-opus-4-7", "claude-sonnet-4-6", "claude-haiku-4-5-20251001"] last_err = None for cand in candidates: payload = { "model": cand, "max_tokens": max_tokens, "system": system_prompt, "temperature": 0.7, "messages": [{"role": "user", "content": prompt}], } req = urllib.request.Request( "https://api.anthropic.com/v1/messages", method="POST", data=json.dumps(payload).encode("utf-8"), headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, ) try: with urllib.request.urlopen(req, timeout=90) as resp: data = json.loads(resp.read()) text = "".join( b.get("text", "") for b in data.get("content", []) if b.get("type") == "text" ) return { "text": text, "model": data.get("model"), "requested_model": cand, "stop_reason": data.get("stop_reason"), "input_tokens": data.get("usage", {}).get("input_tokens"), "output_tokens": data.get("usage", {}).get("output_tokens"), } except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") if e.fp else str(e) last_err = f"{e.code} {e.reason}: {body[:300]}" continue except Exception as e: last_err = str(e) continue raise RuntimeError(f"All candidates failed; last: {last_err}") # ============================================================================ # Stratum C: 6 new Wikipedia articles # ============================================================================ STRATUM_C_NEW = [ {"slug": "c05_wikipedia_nuclear_fusion", "title": "Nuclear_fusion", "description": "Wikipedia article on nuclear fusion (scientific-technical)."}, {"slug": "c06_wikipedia_inflation", "title": "Inflation", "description": "Wikipedia article on inflation (economic policy)."}, {"slug": "c07_wikipedia_climate_mitigation", "title": "Climate_change_mitigation", "description": "Wikipedia article on climate mitigation (contested policy)."}, {"slug": "c08_wikipedia_genai", "title": "Generative_artificial_intelligence", "description": "Wikipedia article on generative AI (technical)."}, {"slug": "c09_wikipedia_gerrymandering", "title": "Gerrymandering", "description": "Wikipedia article on gerrymandering (contested political)."}, {"slug": "c10_wikipedia_microplastic", "title": "Microplastic", "description": "Wikipedia article on microplastics (health/scientific)."}, ] def _fetch_wikipedia_extract(title: str, max_chars: int = 6000) -> str: import urllib.parse api = ( "https://en.wikipedia.org/w/api.php?" "action=query&prop=extracts&explaintext=1&redirects=1&format=json&titles=" + urllib.parse.quote(title) ) req = urllib.request.Request( api, headers={"User-Agent": "FrameCheck-ValidationStudy/1.0 (research)"}, ) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) pages = data.get("query", {}).get("pages", {}) for pid, page in pages.items(): if page.get("missing") is not None: raise RuntimeError(f"title not found: {title}") extract = page.get("extract", "") if extract: return _truncate_to_paragraphs(extract, max_chars=max_chars) raise RuntimeError(f"no extract for {title}") # ============================================================================ # Assembly # ============================================================================ def _sha256(s): return hashlib.sha256(s.encode("utf-8")).hexdigest() def _wc(s): return len(s.split()) def _now(): return subprocess.check_output(["date", "-u", "+%Y-%m-%dT%H:%M:%SZ"]).decode().strip() def expand(): # Load existing manifest manifest_path = CORPUS_DIR / "manifest.json" with open(manifest_path) as f: manifest = json.load(f) existing_ids = {d["doc_id"] for d in manifest["documents"]} new_docs = [] # ─── Stratum A ─── print("== Stratum A expansion ==") for spec in STRATUM_A_DECLARED: if spec["slug"] in existing_ids: continue try: text, used_url = _fetch_stratum_a_doc(spec) (CORPUS_DIR / f"{spec['slug']}.txt").write_text(text, encoding="utf-8") new_docs.append({ "doc_id": spec["slug"], "stratum": "A", "title": spec["title"], "source": used_url, "fetched_at": _now(), "word_count": _wc(text), "sha256": _sha256(text), "notes": spec["notes"], }) print(f" {spec['slug']}: {_wc(text)} words from {used_url}") except Exception as e: print(f" {spec['slug']}: FAILED - {str(e)[:120]}") # ─── Stratum B ─── print("\n== Stratum B expansion ==") for spec in STRATUM_B_NEW_PROMPTS: if spec["slug"] in existing_ids: continue try: result = _call_anthropic(spec["prompt"]) (CORPUS_DIR / f"{spec['slug']}.txt").write_text(result["text"], encoding="utf-8") new_docs.append({ "doc_id": spec["slug"], "stratum": "B", "title": spec["topic"], "source": f"anthropic-api::{result['model']}", "fetched_at": _now(), "word_count": _wc(result["text"]), "sha256": _sha256(result["text"]), "notes": f"Generated via Anthropic API. requested={result['requested_model']}, served={result['model']}. Prompt: {spec['prompt']}", "stop_reason": result.get("stop_reason"), "input_tokens": result.get("input_tokens"), "output_tokens": result.get("output_tokens"), }) print(f" {spec['slug']}: {_wc(result['text'])} words (model={result['model']})") except Exception as e: print(f" {spec['slug']}: FAILED - {str(e)[:120]}") # ─── Stratum C ─── print("\n== Stratum C expansion ==") for spec in STRATUM_C_NEW: if spec["slug"] in existing_ids: continue try: text = _fetch_wikipedia_extract(spec["title"]) (CORPUS_DIR / f"{spec['slug']}.txt").write_text(text, encoding="utf-8") new_docs.append({ "doc_id": spec["slug"], "stratum": "C", "title": spec["description"], "source": f"https://en.wikipedia.org/wiki/{spec['title']}", "fetched_at": _now(), "word_count": _wc(text), "sha256": _sha256(text), "notes": "Fetched via MediaWiki action API; truncated at ~6000 chars on paragraph boundary.", }) print(f" {spec['slug']}: {_wc(text)} words") except Exception as e: print(f" {spec['slug']}: FAILED - {str(e)[:120]}") # Merge manifest all_docs = manifest["documents"] + new_docs total_n = len(all_docs) per_stratum = {"A": 0, "B": 0, "C": 0} for d in all_docs: per_stratum[d["stratum"]] = per_stratum.get(d["stratum"], 0) + 1 manifest_out = { "version": 3, "assembled_at": _now(), "total_documents": total_n, "target": 30, "per_stratum": per_stratum, "design_notes": ( "v3 corpus: v2 corpus (12 docs) + Track A expansion (18 new docs). " "Pre-registered per DESIGN_v2.md. New doc specs frozen before " "fetch. See script for fallback policy." ), "documents": all_docs, } with open(manifest_path, "w") as f: json.dump(manifest_out, f, indent=2) print(f"\n== Summary ==") print(f" Total documents: {total_n} (target 30)") print(f" Per stratum: A={per_stratum['A']}, B={per_stratum['B']}, C={per_stratum['C']}") print(f" New additions this run: {len(new_docs)}") print(f" Manifest: {manifest_path}") if total_n < 27: print(f"\n WARNING: n={total_n} is below halt-threshold 27. Expansion partial.") print(f" Per DESIGN_v2 §7, halt and re-design if n<27.") else: print(f"\n n={total_n} meets pre-registered threshold; proceeding to Step 3 (re-labeling).") return total_n if __name__ == "__main__": expand()