content block. The page structure (verified on live fetch): the main content is in a series of

that appears AFTER the share-buttons block. We isolate that block by finding the first

tag after the "Implementation Note" link is preceded in the page; more robust extraction: take the 4-6

tags between the article-share block and the "Last Update" footer. """ url = "https://www.federalreserve.gov/newsevents/pressreleases/monetary20260318a.htm" result = subprocess.run( ["curl", "-sSL", "--max-time", "30", url], capture_output=True, text=True, timeout=45, ) html = result.stdout # Narrow to the col-xs-12 col-sm-8 col-md-8 that carries the press # release body. There are two such divs on the page; the second (by # source order) holds the body. divs = re.findall( r'

(.*?)

', html, flags=re.DOTALL, ) if len(divs) < 1: raise RuntimeError("FOMC: could not isolate press-release div") # Largest div (by length) is typically the body. body = max(divs, key=len) # Cut off anything after "Implementation Note" link section impl = body.find("Implementation Note") if impl > 500: body = body[:impl] text = _strip_html(body) # Sanity filter: remove navigation fragments that sometimes leak text = re.sub(r"(?im)^\s*(share|print|email|last update|home|press releases?)\s*$", "", text) text = re.sub(r"\n{3,}", "\n\n", text).strip() return text # ============================================================================ # Stratum B: freshly generated AI analyses (pre-declared prompts) # ============================================================================ STRATUM_B_PROMPTS = [ { "slug": "b01_nvidia_investment", "topic": "business / investment", "prompt": "Analyze the investment case for a major technology company entering the AI accelerator market in 2026. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b02_automation_policy", "topic": "policy / governance", "prompt": "Describe how a national government might approach long-term structural unemployment caused by automation. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b03_social_media_adolescents", "topic": "social / behavioral", "prompt": "Discuss the relationship between social media use and adolescent mental health. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b04_llm_customer_support", "topic": "technical", "prompt": "Analyze whether a company should deploy large language models in customer support. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b05_remote_work_productivity", "topic": "organizational", "prompt": "Discuss how a mid-sized company should evaluate remote work policies. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "b06_quantum_computing_outlook", "topic": "scientific / research", "prompt": "Describe the current state and near-term outlook of quantum computing research. 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( block.get("text", "") for block in data.get("content", []) if block.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[:400]}" continue except Exception as e: last_err = str(e) continue raise RuntimeError(f"All candidates failed; last: {last_err}") # ============================================================================ # Stratum C: Wikipedia articles via action API # ============================================================================ STRATUM_C_URLS = [ { "slug": "c01_wikipedia_semaglutide", "title": "Semaglutide", "description": "Wikipedia article on semaglutide (medical topic).", }, { "slug": "c02_wikipedia_eu_ai_act", "title": "Artificial_Intelligence_Act", "description": "Wikipedia article on the EU AI Act (policy topic).", }, { "slug": "c03_wikipedia_quantum_supremacy", "title": "Quantum_supremacy", "description": "Wikipedia article on quantum supremacy (technical topic).", }, { "slug": "c04_wikipedia_ubi", "title": "Universal_basic_income", "description": "Wikipedia article on universal basic income (contested topic).", }, ] def _fetch_wikipedia_extract(title: str, max_chars: int = 6000) -> str: """Fetch Wikipedia plain-text extract via MediaWiki action API. The /api/rest_v1/page/plain endpoint returns 404 (deprecated); the action API's prop=extracts with explaintext=1 is the current supported path and returns the intro + lead sections in plain text. """ 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", {}) if not pages: raise RuntimeError(f"Wikipedia: no pages returned for {title}") # Pages is keyed by pageid; take the first non-missing. for pid, page in pages.items(): if page.get("missing") is not None: raise RuntimeError(f"Wikipedia: title not found: {title}") extract = page.get("extract", "") if not extract: continue if len(extract) > max_chars: cut = extract[:max_chars] last_para = cut.rfind("\n\n") if last_para > max_chars // 2: extract = cut[:last_para] else: extract = cut return extract.strip() raise RuntimeError(f"Wikipedia: no extract content for {title}") # ============================================================================ # Corpus assembly # ============================================================================ def _sha256(s: str) -> str: return hashlib.sha256(s.encode("utf-8")).hexdigest() def _word_count(s: str) -> int: return len(s.split()) def _now_iso() -> str: return subprocess.check_output(["date", "-u", "+%Y-%m-%dT%H:%M:%SZ"]).decode().strip() def assemble_corpus(): manifest = [] print("== Stratum A: public human-authored documents ==") for slug, fetcher, title, notes in [ ("a01_altman_intelligence_age", _altman_essay_text, "The Intelligence Age (Sam Altman, Sept 2024)", "AI-company founder essay (human-authored)."), ("a02_fomc_march_2026", _fomc_march_2026_text, "FOMC Press Release, March 2026", "Policy document; central bank rate decision."), ]: try: text = fetcher() if _word_count(text) < 150: raise RuntimeError(f"too short ({_word_count(text)} words)") (CORPUS_DIR / f"{slug}.txt").write_text(text, encoding="utf-8") manifest.append({ "doc_id": slug, "stratum": "A", "title": title, "source": "https://ia.samaltman.com/" if "altman" in slug else "https://www.federalreserve.gov/newsevents/pressreleases/monetary20260318a.htm", "fetched_at": _now_iso(), "word_count": _word_count(text), "sha256": _sha256(text), "notes": notes, }) print(f" {slug}: {_word_count(text)} words") except Exception as e: print(f" {slug}: FAILED - {e}") print("\n== Stratum B: freshly generated AI analyses ==") for spec in STRATUM_B_PROMPTS: try: result = _call_anthropic(spec["prompt"]) (CORPUS_DIR / f"{spec['slug']}.txt").write_text(result["text"], encoding="utf-8") manifest.append({ "doc_id": spec["slug"], "stratum": "B", "title": spec["topic"], "source": f"anthropic-api::{result['model']}", "fetched_at": _now_iso(), "word_count": _word_count(result["text"]), "sha256": _sha256(result["text"]), "notes": f"Generated via Anthropic API. Model 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']}: {_word_count(result['text'])} words (model={result['model']})") except Exception as e: print(f" {spec['slug']}: FAILED - {e}") print("\n== Stratum C: Wikipedia articles ==") for spec in STRATUM_C_URLS: try: text = _fetch_wikipedia_extract(spec["title"]) if _word_count(text) < 200: raise RuntimeError(f"too short ({_word_count(text)} words)") (CORPUS_DIR / f"{spec['slug']}.txt").write_text(text, encoding="utf-8") manifest.append({ "doc_id": spec["slug"], "stratum": "C", "title": spec["description"], "source": f"https://en.wikipedia.org/wiki/{spec['title']}", "fetched_at": _now_iso(), "word_count": _word_count(text), "sha256": _sha256(text), "notes": "Fetched via MediaWiki action API (prop=extracts, explaintext=1); truncated at ~6000 chars on paragraph boundary.", }) print(f" {spec['slug']}: {_word_count(text)} words") except Exception as e: print(f" {spec['slug']}: FAILED - {e}") manifest_path = CORPUS_DIR / "manifest.json" with open(manifest_path, "w", encoding="utf-8") as f: json.dump({ "version": 2, "assembled_at": _now_iso(), "total_documents": len(manifest), "target": 12, "design_notes": ( "v2 corpus after v1 extraction failures. See script " "docstring for revisions. Stratum A reduced to 2 " "documents (Altman + FOMC); Stratum B expanded to 6 " "fresh AI generations; Stratum C is 4 Wikipedia " "articles. Total target 12, actual varies on fetch " "success." ), "documents": manifest, }, f, indent=2) print(f"\nManifest: {manifest_path}") print(f"Total documents: {len(manifest)}") return manifest if __name__ == "__main__": assemble_corpus()