File size: 3,560 Bytes
1d9bd9b | 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 | """Stable case-summary contract shared by the current corpus and future re-extractions."""
import json
import os
import re
def _clean(value):
return re.sub(r"\s+", " ", str(value or "")).strip()
def _summary_text(value):
if isinstance(value, str):
return _clean(value)
if not isinstance(value, dict):
return ""
direct = value.get("text") or value.get("summary")
if direct:
return _clean(direct)
parts = []
for key, label in (
("facts", "Facts"),
("issues", "Issues"),
("holding", "Holding"),
("reasoning", "Reasoning"),
("outcome", "Outcome"),
):
item = value.get(key)
if isinstance(item, list):
item = "; ".join(_clean(x) for x in item if _clean(x))
item = _clean(item)
if item:
parts.append(f"{label}: {item}")
return " ".join(parts)
def load_case_summaries(data_dir):
"""Load the first supported extraction sidecar; absent files are a valid pre-migration state."""
for filename in ("judgment_summaries.jsonl", "case_summaries.jsonl"):
path = os.path.join(data_dir, filename)
if not os.path.exists(path):
continue
rows = {}
with open(path, encoding="utf-8") as fh:
for line in fh:
try:
row = json.loads(line)
except Exception:
continue
doc_id = row.get("doc_id")
if doc_id and _summary_text(row.get("summary") or row.get("case_summary")):
rows[doc_id] = row
return rows, filename
return {}, None
def case_summary_record(meta=None, synthetic=None, extracted=None):
"""Choose one honest summary source without falling back to arbitrary opening text."""
meta = meta or {}
synthetic = synthetic or {}
extracted = extracted or {}
text = _summary_text(extracted.get("summary") or extracted.get("case_summary"))
if text:
return {
"available": True,
"text": text[:5000],
"source": "extracted_summary",
"generated": bool(extracted.get("generated", True)),
"provider": _clean(extracted.get("provider") or extracted.get("source_provider")),
"version": _clean(extracted.get("version") or extracted.get("extraction_version")),
}
text = _summary_text(meta.get("summary") or meta.get("case_summary"))
if text:
return {
"available": True,
"text": text[:5000],
"source": "extracted_summary",
"generated": bool(meta.get("summary_generated", True)),
"provider": _clean(meta.get("source_provider")),
"version": _clean(meta.get("extraction_version")),
}
text = _clean(synthetic.get("held"))
if text:
return {
"available": True,
"text": text[:5000],
"source": "synthetic_headnote",
"generated": True,
"provider": "",
"version": _clean(synthetic.get("model")),
}
text = _clean(meta.get("held"))
if text:
return {
"available": True,
"text": text[:5000],
"source": "reporter_headnote",
"generated": False,
"provider": "",
"version": "",
}
return {
"available": False,
"text": "",
"source": "unavailable",
"generated": False,
"provider": "",
"version": "",
}
|