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

"""Auto-extract per-decision metadata for a style_corpus row.

Populates the fields that the upload flow leaves empty — summary, outcome,
key_principles, appeal_subtype, practice_area — by asking Claude (via the
local CLI session) to read the proofread full_text and return a structured
JSON blob.

Caller policy (``apply_to_corpus``): by default we **only fill empty
columns**, so chair-edited values are preserved across re-runs. The chair
can force a refresh by passing ``overwrite=True``.

Why this is a separate module from ``precedent_metadata_extractor``:
that one fills the *external* case_law corpus (court rulings, third-party
committee decisions). This one fills the *style* corpus — Daphna's own
decisions used to teach the writer the in-house voice. The two corpora
have different schemas, different prompts, and different downstream
consumers, so coupling them would have been the wrong shortcut.
"""

from __future__ import annotations

import logging
from uuid import UUID

from legal_mcp.services import claude_session, db

logger = logging.getLogger(__name__)


# A single decision typically runs 200K-650K chars. We sample the head
# (where outcome + parties + framing live) and the tail (where the
# operative ruling sits). Picking from both edges keeps the prompt under
# 60K chars — comfortable for any Claude tier.
_HEAD_CHARS = 25_000
_TAIL_CHARS = 15_000


def _build_text_window(full_text: str) -> str:
    if len(full_text) <= _HEAD_CHARS + _TAIL_CHARS:
        return full_text
    head = full_text[:_HEAD_CHARS]
    tail = full_text[-_TAIL_CHARS:]
    return (
        f"{head}\n\n"
        f"[... חתך: {len(full_text) - _HEAD_CHARS - _TAIL_CHARS:,} תווים מהאמצע "
        f"הושמטו — שמרנו על ההתחלה (טענות + רקע) ועל הסוף (הכרעה + הוצאות) ...]"
        f"\n\n{tail}"
    )


# Static instructions — go via ``system`` so the SDK path can cache them
# across batch enrichment runs (24+ decisions in one pass).
METADATA_PROMPT = """אתה מסייע משפטי שמקטלג את הקורפוס הסגנוני של דפנה תמיר (יו"ר ועדת ערר).

תפקידך: לקרוא החלטה אחת ולחלץ מטא-דאטה ל-style_corpus — שדות שהמשתמש לא הזין בעת ההעלאה.

**אל תמציא**. אם המידע לא מופיע בטקסט, השאר מחרוזת ריקה או מערך ריק. אסור להסיק עובדות שלא כתובות.

## פלט נדרש

החזר JSON אחד (object אחד — לא array, לא markdown, לא הסברים):

{
  "summary": "תקציר עניני ב-2-3 משפטים: מי העורר, מה דרש, מה הוכרע. סגנון יבש, ניטרלי, ללא שיפוט. דוגמה: 'ערר על דחיית בקשה להיתר לתוספת מרפסת בקומה ג׳. דפנה קיבלה את הערר חלקית — אישרה את המרפסת בהקטנה ל-12 מ״ר.'",

  "outcome": "התוצאה התמציתית. אחד מאלה (או צירוף קצר): 'קבלה' / 'קבלה חלקית' / 'דחייה' / 'הסתלקות' / 'החזרה לוועדה המקומית'. אם זה לא ברור — מחרוזת ריקה.",

  "key_principles": [
    "עיקרון משפטי 1 שעולה מההחלטה — משפט אחד, ניסוח מופשט. למשל 'שיקול דעת מוגבל לחריגות בנייה קטנות'.",
    "עיקרון 2",
    "..."
  ],

  "appeal_subtype": "תת-סוג ערר. ערכים מותרים: 'building_permit' (היתר בנייה / רישוי), 'betterment_levy' (היטל השבחה), 'compensation_197' (פיצויים ס׳ 197), 'use_change' (שימוש חורג), 'tama_38' (תמ\\"א 38), או מחרוזת ריקה אם לא ברור.",

  "practice_area": "תחום משפט גנרי. ברירת מחדל: 'appeals_committee'. אם זה במובהק 'planning_law' — סמן.",

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

  "parties_respondent": "שם המשיב/ים. ברירת מחדל לעררי 1xxx ו-8xxx: 'הוועדה המקומית לתכנון ובניה ירושלים' או דומה. השאר ריק אם לא ברור."
}

## כללי איכות

1. **summary** — חייב להזכיר את התוצאה. בלי 'בית המשפט קבע ש...' (אנחנו לא בית משפט). בלי הערכת אישית.
2. **outcome** — קבלה / קבלה חלקית / דחייה / הסתלקות / החזרה לוועדה המקומית. אם דפנה הכריעה חלקית — 'קבלה חלקית'. אסור 'התקבל' או 'נדחה' בלשון פעולה — רק שם פעולה.
3. **key_principles** — 2-5 עקרונות מקסימום. כל אחד משפט אחד. לא ציטוטים מילוליים, אלא תמצות העיקרון.
4. **appeal_subtype** — תמיד פעולה אחת. אם החלטה מערבת כמה תת-סוגים — בחר את העיקרי.
5. **parties_appellant / parties_respondent** — שם בלבד, בלי 'נ׳' או 'נגד'.

החזר רק את ה-JSON. אל תכתוב שום דבר לפניו או אחריו.
"""


async def extract_decision_metadata(corpus_id: UUID | str) -> dict:
    """Run Claude over the row's full_text and return suggested fields.

    Does NOT touch the DB. The caller decides what to apply.
    """
    if isinstance(corpus_id, str):
        corpus_id = UUID(corpus_id)
    row = await db.get_style_corpus_row(corpus_id)
    if not row:
        return {}
    full_text = (row.get("full_text") or "").strip()
    if not full_text:
        return {}

    context = (
        f"מספר החלטה: {row.get('decision_number') or '—'}\n"
        f"תאריך: {row.get('decision_date') or '—'}\n"
        f"תת-סוג נוכחי: {row.get('appeal_subtype') or '—'}\n"
        f"נושאים מתויגים: {row.get('subject_categories') or '—'}"
    )
    window = _build_text_window(full_text)
    user_msg = (
        f"## הקלט\n{context}\n\n"
        f"--- תחילת ההחלטה ---\n{window}\n--- סוף ההחלטה ---"
    )

    try:
        result = await claude_session.query_json(user_msg, system=METADATA_PROMPT, tools="")  # no tool_use → no error_max_turns
    except Exception as e:
        logger.warning("style_metadata_extractor: query failed: %s", e)
        return {}

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

    out: dict = {}
    if isinstance(result.get("summary"), str):
        out["summary"] = result["summary"].strip()
    if isinstance(result.get("outcome"), str):
        out["outcome"] = result["outcome"].strip()
    kp = result.get("key_principles") or []
    if isinstance(kp, list):
        out["key_principles"] = [str(p).strip() for p in kp if str(p).strip()]
    if isinstance(result.get("appeal_subtype"), str):
        st = result["appeal_subtype"].strip()
        # Open enum — but log values outside the documented list so we can
        # tighten the prompt later if needed.
        known = {
            "building_permit", "betterment_levy", "compensation_197",
            "use_change", "tama_38", "",
        }
        if st not in known:
            logger.info("style_metadata: unknown appeal_subtype=%r (kept)", st)
        out["appeal_subtype"] = st
    if isinstance(result.get("practice_area"), str):
        out["practice_area"] = result["practice_area"].strip()
    # Parties: not stored in the schema today, but worth surfacing in the
    # extractor's return value so callers (and the UI's drawer) can display
    # them. The list endpoint extracts via regex; LLM output is the
    # higher-quality fallback when regex fails.
    if isinstance(result.get("parties_appellant"), str):
        out["parties_appellant"] = result["parties_appellant"].strip()
    if isinstance(result.get("parties_respondent"), str):
        out["parties_respondent"] = result["parties_respondent"].strip()
    return out


async def extract_and_apply(
    corpus_id: UUID | str, *, overwrite: bool = False,
) -> dict:
    """Convenience: extract → apply → return summary of what changed.

    Idempotent under default ``overwrite=False`` — re-runs only fill empty
    fields. Use ``overwrite=True`` to refresh values the chair (or a prior
    extraction) already wrote.
    """
    if isinstance(corpus_id, str):
        corpus_id = UUID(corpus_id)
    suggested = await extract_decision_metadata(corpus_id)
    if not suggested:
        return {"extracted": False, "applied": False, "reason": "no suggestion"}

    update_result = await db.update_style_corpus_metadata(
        corpus_id,
        summary=suggested.get("summary"),
        outcome=suggested.get("outcome"),
        key_principles=suggested.get("key_principles"),
        appeal_subtype=suggested.get("appeal_subtype"),
        practice_area=suggested.get("practice_area"),
        overwrite=overwrite,
    )
    return {
        "extracted": True,
        "applied": update_result.get("updated", False),
        "fields_set": update_result.get("fields", []),
        "suggested": suggested,
    }