| """CaseFile notebook (journalism suite layer 7). |
| |
| A durable, auditable case notebook: sources, claims, evidence, timeline |
| events, threads, and findings in one JSONL file per case under data/casefiles/. |
| Every finding carries a thread, verdict, confidence, and the source ids it |
| rests on (chain-of-custody). The notebook is the deliverable a journalist can |
| show a grants panel or an editor — not just chat history. |
| |
| Usage: |
| from research.casefile import CaseFile |
| cf = CaseFile("bridge_case") |
| cf.add_source("s1", "DOT filing 2010", tier="verified-leak") |
| cf.add_finding("main", "bridge opened 2010", "supports", "HIGH", ["s1"]) |
| md = cf.export_markdown() |
| """ |
| import json |
| import time |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| CASE_DIR = ROOT / "data" / "casefiles" |
|
|
|
|
| class CaseFile: |
| def __init__(self, name="case"): |
| self.name = name |
| self.path = CASE_DIR / f"{name}.jsonl" |
| self.records = [] |
| self._load() |
|
|
| def _load(self): |
| if not self.path.exists(): |
| return |
| for line in self.path.read_text(encoding="utf-8").splitlines(): |
| line = line.strip() |
| if line: |
| try: |
| self.records.append(json.loads(line)) |
| except ValueError: |
| pass |
|
|
| def _append(self, rec): |
| self.records.append(rec) |
| self.path.parent.mkdir(parents=True, exist_ok=True) |
| with self.path.open("a", encoding="utf-8") as fh: |
| fh.write(json.dumps(rec, ensure_ascii=False) + "\n") |
| return rec |
|
|
| def add_source(self, source_id, title, tier="unverified", url="", date="", |
| retrievable=True, independent=True, origin="", |
| content_sha256="", triage=None, retrieved_at=""): |
| return self._append({"type": "source", "source_id": source_id, |
| "title": title, "tier": tier, "url": url, |
| "date": date, "retrievable": retrievable, |
| "independent": independent, "origin": origin, |
| "content_sha256": content_sha256, |
| "triage": triage or {}, "retrieved_at": retrieved_at, |
| "ts": time.strftime("%Y-%m-%d")}) |
|
|
| def add_claim(self, claim, source_ids, verdict="", confidence="", source_policy=None): |
| return self._append({"type": "claim", "claim": claim, |
| "source_ids": list(source_ids), |
| "verdict": verdict, "confidence": confidence, |
| "source_policy": source_policy or {}, |
| "ts": time.strftime("%Y-%m-%d")}) |
|
|
| def add_evidence(self, claim, source_id, quote, supports=True): |
| return self._append({"type": "evidence", "claim": claim, |
| "source_id": source_id, "quote": quote, |
| "supports": supports, |
| "ts": time.strftime("%Y-%m-%d")}) |
|
|
| def add_event(self, when, what, source_id="-"): |
| return self._append({"type": "event", "when": when, "what": what, |
| "source_id": source_id}) |
|
|
| def add_thread(self, name, question): |
| return self._append({"type": "thread", "name": name, |
| "question": question, |
| "ts": time.strftime("%Y-%m-%d")}) |
|
|
| def add_finding(self, thread, claim, verdict, confidence, source_ids, |
| missing="", source_policy=None): |
| return self._append({"type": "finding", "thread": thread, "claim": claim, |
| "verdict": verdict, "confidence": confidence, |
| "source_ids": list(source_ids), "missing": missing, |
| "source_policy": source_policy or {}, |
| "ts": time.strftime("%Y-%m-%d")}) |
|
|
| def sources(self): |
| return [r for r in self.records if r["type"] == "source"] |
|
|
| def claims(self): |
| return [r for r in self.records if r["type"] == "claim"] |
|
|
| def findings(self): |
| return [r for r in self.records if r["type"] == "findings" |
| or r["type"] == "finding"] |
|
|
| def threads(self): |
| seen = {} |
| for r in self.records: |
| if r["type"] == "thread": |
| seen[r["name"]] = r["question"] |
| return seen |
|
|
| def export_markdown(self): |
| lines = [f"# CaseFile: {self.name}", ""] |
| lines.append("## Sources") |
| for s in self.sources(): |
| lines.append(f"- `{s['source_id']}` [{s['tier']}] {s['title']} " |
| f"{s.get('url', '')}") |
| lines.append("") |
| lines.append("## Threads") |
| for name, q in self.threads().items(): |
| lines.append(f"- **{name}**: {q}") |
| lines.append("") |
| lines.append("## Claims") |
| for c in self.claims(): |
| lines.append(f"- {c['claim']} -> {c.get('verdict', '')} " |
| f"{c.get('confidence', '')} " |
| f"(sources: {', '.join(c['source_ids'])})") |
| lines.append("") |
| lines.append("## Findings") |
| for f in self.findings(): |
| lines.append(f"- [{f.get('thread', '-')}] {f['claim']} -> " |
| f"{f['verdict']} {f['confidence']} " |
| f"(sources: {', '.join(f['source_ids'])})") |
| if f.get("missing"): |
| lines.append(f" - missing: {f['missing']}") |
| lines.append("") |
| lines.append("## Timeline") |
| for e in [r for r in self.records if r["type"] == "event"]: |
| lines.append(f"- {e['when']}: {e['what']} ({e['source_id']})") |
| return "\n".join(lines) |
|
|
