File size: 5,762 Bytes
8b8e59d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | """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)
|