legal-ai/mcp-server/src/legal_mcp/services/precedent_metadata_extractor.py

"""Auto-extract precedent metadata from a freshly-uploaded ruling.

Runs after chunking. Reads the precedent's full_text and asks Claude to
fill in the metadata fields that an upload form usually leaves empty:
short case_name, summary, headnote, key_quote, subject_tags,
appeal_subtype, decision_date, precedent_level, court.

Caller policy: only empty user-supplied fields are filled. Anything the
chair already typed in the upload form is preserved. This is enforced
in ``apply_to_record``.
"""

from __future__ import annotations

import logging
from datetime import date as date_type
from uuid import UUID

from legal_mcp.config import parse_llm_json
from legal_mcp.services import claude_session, db

logger = logging.getLogger(__name__)


# The prompt is short — we only need the first 12K chars of the ruling
# (header + opening of discussion is enough for naming + summary). For
# subject tags we sample the discussion section too.
_HEAD_CHARS = 12_000
_TAIL_CHARS = 6_000


# Note: this template is concatenated with f-strings at call-time rather
# than using .format(), because the JSON example below contains '{' / '}'
# which str.format would interpret as placeholders and crash with
# KeyError on the field names.
METADATA_EXTRACTION_PROMPT = """אתה מסייע משפטי בכיר. קרא את פסק הדין/ההחלטה הבא וחלץ ממנו מטא-דאטה לקטלוג הקורפוס.

המטרה: למלא שדות בטופס העלאה שהמשתמש הזין באופן חלקי. **אל תמציא** — אם המידע לא מופיע בטקסט, השאר ריק (מחרוזת ריקה / מערך ריק).

## פלט נדרש
החזר JSON אחד (object — לא array) בפורמט הבא, ללא markdown וללא הסברים:

{
  "case_name_short": "שם קצר ל-3-6 מילים (למשל 'אהרון ברק' או 'ב. קרן-נכסים'). אל תכלול מספר תיק. שם המבקש/העורר העיקרי. אם זו החלטה מאוחדת — שם הצד המוביל.",
  "appeal_subtype": "תת-סוג ספציפי בתוך תחום המשפט (למשל 'תכנית רחביה', 'מימוש במכר', 'תמ\\"א 38', 'שימוש חורג', 'סופיות ההחלטה'). מילה אחת או צירוף קצר.",
  "summary": "תקציר עניני 2-3 משפטים: מה הייתה השאלה, מה הוכרע. בלי שיפוט.",
  "headnote": "headnote בסגנון נבו: 1-2 משפטים שמסכמים את העיקרון שנקבע/יושם בפסק. למשל 'תכנית רחביה — היטל השבחה במימוש במכר — אין לחייב כשהזכויות צפות'.",
  "key_quote": "ציטוט מילולי בודד, 30-100 מילים, שמייצג את לב הפסק. חייב להופיע מילה במילה בטקסט. אם אין ציטוט מתאים — מחרוזת ריקה.",
  "subject_tags": ["תגיות", "נושא", "בעברית"],
  "decision_date_iso": "YYYY-MM-DD — תאריך מתן ההחלטה כפי שמופיע בטקסט (בכותרת או בחתימה הסופית). אם לא ניתן לזהות במדויק — מחרוזת ריקה.",
  "precedent_level": "אחד מ-4: 'עליון' / 'מנהלי' / 'ועדת_ערר_ארצית' / 'ועדת_ערר_מחוזית'. בחר לפי הערכאה שמסומנת בכותרת הפסק. אם לא ברור — מחרוזת ריקה.",
  "source_type": "אחד מ-2: 'court_ruling' (פסק דין של בית משפט — עליון/מנהלי) / 'appeals_committee' (החלטה של ועדת ערר). אם לא ברור — מחרוזת ריקה.",
  "court": "שם הערכאה כפי שהוא מופיע בכותרת (למשל 'בית המשפט העליון', 'בית המשפט המחוזי בירושלים בשבתו כבית משפט לעניינים מנהליים', 'ועדת הערר לתכנון ובניה פיצויים והיטלי השבחה — מחוז ירושלים'). מחרוזת ריקה אם לא ניתן לזהות."
}

## כללי איכות
1. **case_name_short** — שם בולט וקצר. בלי 'נ\\'' / 'נגד' / מספרי תיק.
2. **appeal_subtype** — אופציונלי. אם הסוגיה רחבה ולא מסווגת — השאר ריק.
3. **summary** — תיאור ניטרלי, גוף שלישי.
4. **headnote** — לא מצטטים, מסכמים. סגנון נבו: ביטוי קצר אחד.
5. **key_quote** — חייב להיות הדבקה מילולית מהקלט. אם אין ציטוט בולט — השאר ריק.
6. **subject_tags** — 3-7 תגיות בעברית, snake_case (חניה, קווי_בניין, שיקול_דעת, פגם_פרוצדורלי, סמכות, מועדים, פגיעה_במקרקעין, ירידת_ערך, תכנית_רחביה, מימוש_במכר, וכד'). שייך לתחום של ועדת ערר תכנון ובניה.
7. **decision_date_iso** — תאריך מדויק בלבד. אם בטקסט יש "ניתנה היום, ט' באלול תשפ"א, 5 בספטמבר 2022" — הפלט: "2022-09-05".
8. **precedent_level** — קבע לפי הערכאה: בית המשפט העליון = "עליון"; בית משפט מחוזי בשבתו כבית משפט לעניינים מנהליים = "מנהלי"; ועדת ערר ארצית = "ועדת_ערר_ארצית"; ועדת ערר מחוזית (כמו ועדות תכנון ובניה ירושלים/מחוז המרכז וכד') = "ועדת_ערר_מחוזית". השתמש ב-underscore כפי שמופיע — לא ברווח.
9. **source_type** — שני ערכים בלבד: "court_ruling" כשהמסמך הוא פסק דין/החלטה של בית משפט (עליון/בג"ץ/מנהלי/מחוזי); "appeals_committee" כשהמסמך הוא החלטה של ועדת ערר (ארצית או מחוזית). זה משלים את `precedent_level` — שני השדות צריכים להיות תואמים.
10. **court** — מהכותרת הראשית של הפסק. ניסוח מלא (לא קיצור). מחרוזת ריקה אם לא ניתן לזהות.
"""


def _build_text_window(full_text: str) -> str:
    """Return the head + tail of the ruling, with a marker if truncated.

    Most rulings have the parties/subject in the head and the conclusion
    in the tail; the middle is the discussion which is captured via the
    halacha extractor independently. Sending head+tail keeps the prompt
    cheap while preserving naming and conclusion context.
    """
    if len(full_text) <= _HEAD_CHARS + _TAIL_CHARS:
        return full_text
    return (
        full_text[:_HEAD_CHARS]
        + "\n\n[... חלק האמצע הושמט עקב אורך — ראה את החלק האחרון של הפסק להלן ...]\n\n"
        + full_text[-_TAIL_CHARS:]
    )


async def extract_metadata(case_law_id: UUID | str) -> dict:
    """Run metadata extraction. Returns a dict with the suggested values.

    Does NOT write to the DB — caller decides what to merge.
    """
    if isinstance(case_law_id, str):
        case_law_id = UUID(case_law_id)

    record = await db.get_case_law(case_law_id)
    if not record:
        return {}
    full_text = (record.get("full_text") or "").strip()
    if not full_text:
        return {}

    citation = record.get("case_number") or ""
    court = record.get("court") or ""
    date_str = str(record.get("date") or "")
    practice_area = record.get("practice_area") or ""

    context = (
        f"מראה מקום: {citation}\n"
        f"ערכאה: {court}\n"
        f"תאריך: {date_str}\n"
        f"תחום: {practice_area}"
    )
    text_window = _build_text_window(full_text)
    # Static instructions go via `system` so the SDK path can cache them
    # across uploads. Per-precedent content goes in the user prompt.
    user_msg = (
        f"## הקלט\n{context}\n\n"
        f"--- תחילת הטקסט ---\n{text_window}\n--- סוף הטקסט ---"
    )

    try:
        result = await claude_session.query_json(
            user_msg, system=METADATA_EXTRACTION_PROMPT,
        )
    except Exception as e:
        logger.warning("precedent_metadata_extractor: query failed: %s", e)
        return {}

    if not isinstance(result, dict):
        logger.warning(
            "precedent_metadata_extractor: expected dict, got %s",
            type(result).__name__,
        )
        return {}

    # Normalize keys / types
    out: dict = {}
    if isinstance(result.get("case_name_short"), str):
        out["case_name_short"] = result["case_name_short"].strip()
    if isinstance(result.get("appeal_subtype"), str):
        out["appeal_subtype"] = result["appeal_subtype"].strip()
    if isinstance(result.get("summary"), str):
        out["summary"] = result["summary"].strip()
    if isinstance(result.get("headnote"), str):
        out["headnote"] = result["headnote"].strip()
    if isinstance(result.get("key_quote"), str):
        out["key_quote"] = result["key_quote"].strip()
    tags = result.get("subject_tags") or []
    if isinstance(tags, list):
        out["subject_tags"] = [str(t).strip() for t in tags if str(t).strip()]
    if isinstance(result.get("decision_date_iso"), str):
        out["decision_date_iso"] = result["decision_date_iso"].strip()
    if isinstance(result.get("precedent_level"), str):
        # Validate against the closed enum used elsewhere in the system
        lvl = result["precedent_level"].strip()
        if lvl in {"עליון", "מנהלי", "ועדת_ערר_ארצית", "ועדת_ערר_מחוזית"}:
            out["precedent_level"] = lvl
    if isinstance(result.get("source_type"), str):
        st = result["source_type"].strip()
        if st in {"court_ruling", "appeals_committee"}:
            out["source_type"] = st
    if isinstance(result.get("court"), str):
        out["court"] = result["court"].strip()
    return out


async def apply_to_record(
    case_law_id: UUID | str,
    suggested: dict,
) -> dict:
    """Merge suggested metadata into the case_law row, filling ONLY empty fields.

    Empty rules:
      - string field == "" → fill from suggested
      - list field == [] → fill from suggested
      - if suggested key is missing or empty, skip

    case_name has special handling: if the current case_name equals the
    case_number (a tell-tale sign of the upload form sending the long
    citation into both fields), treat it as empty and overwrite.
    """
    if isinstance(case_law_id, str):
        case_law_id = UUID(case_law_id)
    record = await db.get_case_law(case_law_id)
    if not record:
        return {"updated": False, "fields": []}

    fields_to_update: dict = {}

    cur_case_name = (record.get("case_name") or "").strip()
    cur_case_number = (record.get("case_number") or "").strip()
    suggested_case_name = (suggested.get("case_name_short") or "").strip()
    if suggested_case_name and (
        not cur_case_name or cur_case_name == cur_case_number
    ):
        fields_to_update["case_name"] = suggested_case_name

    if not (record.get("appeal_subtype") or "").strip():
        s = (suggested.get("appeal_subtype") or "").strip()
        if s:
            fields_to_update["appeal_subtype"] = s

    if not (record.get("summary") or "").strip():
        s = (suggested.get("summary") or "").strip()
        if s:
            fields_to_update["summary"] = s

    if not (record.get("headnote") or "").strip():
        s = (suggested.get("headnote") or "").strip()
        if s:
            fields_to_update["headnote"] = s

    if not (record.get("key_quote") or "").strip():
        s = (suggested.get("key_quote") or "").strip()
        if s:
            fields_to_update["key_quote"] = s

    cur_tags = record.get("subject_tags") or []
    if not cur_tags:
        sug_tags = suggested.get("subject_tags") or []
        if sug_tags:
            fields_to_update["subject_tags"] = sug_tags

    # decision_date — only fill if currently null. The DB column is DATE,
    # so we parse the LLM's ISO string into a date object before passing
    # it to update_case_law (asyncpg won't coerce a string to DATE).
    if record.get("date") is None:
        iso = (suggested.get("decision_date_iso") or "").strip()
        if iso:
            try:
                fields_to_update["date"] = date_type.fromisoformat(iso[:10])
            except ValueError:
                logger.debug(
                    "metadata_extractor: ignoring invalid decision_date_iso=%r",
                    iso,
                )

    if not (record.get("precedent_level") or "").strip():
        lvl = (suggested.get("precedent_level") or "").strip()
        if lvl:
            fields_to_update["precedent_level"] = lvl

    if not (record.get("source_type") or "").strip():
        st = (suggested.get("source_type") or "").strip()
        if st:
            fields_to_update["source_type"] = st

    if not (record.get("court") or "").strip():
        c = (suggested.get("court") or "").strip()
        if c:
            fields_to_update["court"] = c

    if not fields_to_update:
        return {"updated": False, "fields": []}

    await db.update_case_law(case_law_id, **fields_to_update)
    return {"updated": True, "fields": list(fields_to_update.keys())}


async def extract_and_apply(case_law_id: UUID | str) -> dict:
    """Convenience wrapper: extract → merge into row → return summary."""
    suggested = await extract_metadata(case_law_id)
    if not suggested:
        return {"status": "no_metadata", "fields": []}
    result = await apply_to_record(case_law_id, suggested)
    return {
        "status": "completed" if result["updated"] else "no_changes",
        "fields": result["fields"],
        "suggested": suggested,
    }