| """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": "", |
| } |
|
|
