"""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": "שם הערכאה כפי שהוא מופיע בכותרת (למשל 'בית המשפט העליון', 'בית המשפט המחוזי בירושלים בשבתו כבית משפט לעניינים מנהליים', 'ועדת הערר לתכנון ובניה פיצויים והיטלי השבחה — מחוז ירושלים'). מחרוזת ריקה אם לא ניתן לזהות.", "case_number_clean": "מספר הערר/תיק כפי שמופיע בכותרת — רק הספרות והאלכסון, למשל '1062/24' או '8031/21'. ללא המילה 'ערר', ללא שם הצדדים, ללא סוגריים. אם יש כמה עררים מאוחדים — הרשום הראשון. מחרוזת ריקה אם לא ניתן לזהות." } ## כללי איכות 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() if isinstance(result.get("case_number_clean"), str): out["case_number_clean"] = result["case_number_clean"].strip() return out async def apply_to_record( case_law_id: UUID | str, suggested: dict, overwrite_case_number: bool = False, ) -> 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. overwrite_case_number: when True, update case_number from case_number_clean even if the field already has a value (used for one-time migration enrichment). """ 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 [] # Treat character-by-character corruption as empty. Early ingest # pipelines stored a JSON string (`'["היטל השבחה"]'`) into a TEXT[] # column, which Postgres split into individual chars: # `['[', '"', 'ה', 'י', 'ט', 'ל', ' ', 'ה', 'ש', ...]`. Detection: # 3+ elements where every element is at most 2 chars (legitimate # tags are multi-character Hebrew words like `היטל_השבחה`). is_corrupt = ( len(cur_tags) >= 3 and all(isinstance(t, str) and len(t) <= 2 for t in cur_tags) ) if not cur_tags or is_corrupt: 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 overwrite_case_number: cn = (suggested.get("case_number_clean") or "").strip() if cn: fields_to_update["case_number"] = cn 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, overwrite_case_number: bool = False, ) -> 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, overwrite_case_number=overwrite_case_number) return { "status": "completed" if result["updated"] else "no_changes", "fields": result["fields"], "suggested": suggested, }