feat(analysis): דגל "לא-נותח" + ניתוח-מחדש מאחד + בדיקת-השפעה (#201)
All checks were successful
G12 Leak-Guard / leak-guard (pull_request) Successful in 5s
Lint — undefined names / undefined-names (pull_request) Successful in 11s

WS2b של עיצוב-מחדש זרימת-העבודה. מאפשר ניתוח כתב-הערר לבד ואז איחוד
מסמך-עיקרי שנוסף מאוחר, בלי force-delete גורף, עם diff ליו"ר.

- SCHEMA_V49: documents.claims_extracted_at + claims_extraction_status
  (אירוע-חילוץ per-מסמך, לא נגזר מ-doc_type), + אינדקס חלקי
  idx_documents_claims_pending. idempotent (ADD COLUMN IF NOT EXISTS).
  V48 שמור ל-#357.
- claims_extractor חותם את המסמך אחרי שמירת/אי-מציאת טענות.
- db.primary_docs_not_analyzed (is_primary V47 + claims_extracted_at IS NULL)
  + mark_document_claims_extracted; _row_to_doc חושף claims_analyzed.
- reanalyze_claims (כלי-MCP): snapshot→חילוץ-מאחד רק למסמכים חדשים/לא-נותחו
  (store_claims מחליף per-source ⇒ טענות אחרות נשמרות)→aggregate force=True
  (מסלול קנוני, מוחק רק legal_arguments)→_impact_diff before↔after ליו"ר.
- workflow_status חושף primary_docs_not_analyzed + next-step.
- ספ: 02-data-model §2ג (דגל לא-נותח), 04-analysis-writing §1.3.
- 5 בדיקות-יחידה ל-_impact_diff/_snapshot (פונקציות טהורות).

Invariants:
- G1 (נרמול-במקור): claims_extracted_at נחתם בנקודת-החילוץ, claims_analyzed
  נגזר ממנו — לא תיקון-בקריאה.
- G2 (מסלול קנוני יחיד / merge-not-fork): reanalyze מרחיב את מסלול
  claims_extractor+argument_aggregator הקיים, לא forks; האיחוד דרך
  store_claims per-source; aggregate force=True מוחק רק נגזר (legal_arguments).
- INV-DM (מודל-נתונים): דגל-אירוע per-מסמך, אינדקס חלקי, מקור-אמת יחיד.
- G10 (שער-אנושי): בדיקת-ההשפעה מוצגת ליו"ר, לא מוחלת אוטומטית.
- INV-TOOL idempotency: SCHEMA_V49 idempotent; ניתוח-מחדש חוזר על מסמך
  כבר-נותח הוא refresh נקי (store_methods replace per-source).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-30 12:14:54 +00:00
parent 99574f0eba
commit dfb2ffe7ce
8 changed files with 375 additions and 11 deletions

View File

@@ -4,9 +4,18 @@ from __future__ import annotations
from uuid import UUID
from legal_mcp.services import argument_aggregator, db
from legal_mcp.services import argument_aggregator, claims_extractor, db
from legal_mcp.tools.envelope import empty, err, ok # GAP-48: SSoT envelope
# Party labels reused across this module (display + impact diff).
_PARTY_HE = {
"appellant": "עוררים",
"respondent": "משיבים",
"committee": "ועדה מקומית",
"permit_applicant": "מבקשי היתר",
"unknown": "צד לא מזוהה",
}
async def aggregate_claims_to_arguments(
case_number: str,
@@ -55,16 +64,9 @@ async def get_legal_arguments(
)
# Group by party for nicer display.
party_he = {
"appellant": "עוררים",
"respondent": "משיבים",
"committee": "ועדה מקומית",
"permit_applicant": "מבקשי היתר",
"unknown": "צד לא מזוהה",
}
by_party: dict[str, list[dict]] = {}
for a in args:
label = party_he.get(a["party"], a["party"])
label = _PARTY_HE.get(a["party"], a["party"])
by_party.setdefault(label, []).append(a)
return ok({
@@ -72,3 +74,147 @@ async def get_legal_arguments(
"total": len(args),
"by_party": by_party,
})
# ── Re-analysis with merge + impact diff (WS2 / #201) ───────────────
def _snapshot(args: list[dict]) -> dict:
"""Build a comparable snapshot of aggregated arguments for the impact diff.
Captures, per party: the set of argument titles and a priority histogram —
enough to show the chair what changed without dumping full bodies. The
priority mix is the deterministic "recommendation" signal: it summarises the
balance of threshold/substantive/procedural/relief arguments per side.
"""
by_party: dict[str, dict] = {}
for a in args:
party = a.get("party", "unknown")
bucket = by_party.setdefault(party, {"titles": [], "priorities": {}})
bucket["titles"].append((a.get("argument_title") or "").strip())
pr = a.get("priority", "substantive")
bucket["priorities"][pr] = bucket["priorities"].get(pr, 0) + 1
return {"total": len(args), "by_party": by_party}
def _impact_diff(before: dict, after: dict) -> dict:
"""Diff two argument snapshots → a chair-facing before↔after summary.
Reports, per party, which argument titles were added/removed and how the
priority mix (the "recommendation" balance) shifted. ``changed`` is a quick
boolean the UI/chair can gate on. Nothing here is auto-applied — it is purely
surfaced for the human gate (G10).
"""
parties = sorted(set(before["by_party"]) | set(after["by_party"]))
per_party: list[dict] = []
any_change = False
for party in parties:
b = before["by_party"].get(party, {"titles": [], "priorities": {}})
a = after["by_party"].get(party, {"titles": [], "priorities": {}})
b_titles, a_titles = set(b["titles"]), set(a["titles"])
added = sorted(a_titles - b_titles)
removed = sorted(b_titles - a_titles)
prio_before = dict(sorted(b["priorities"].items()))
prio_after = dict(sorted(a["priorities"].items()))
changed = bool(added or removed or prio_before != prio_after)
any_change = any_change or changed
per_party.append({
"party": party,
"party_he": _PARTY_HE.get(party, party),
"count_before": len(b["titles"]),
"count_after": len(a["titles"]),
"added_arguments": added,
"removed_arguments": removed,
"priority_before": prio_before,
"priority_after": prio_after,
"changed": changed,
})
return {
"changed": any_change,
"total_before": before["total"],
"total_after": after["total"],
"by_party": per_party,
}
async def reanalyze_claims(
case_number: str,
reanalyze_all_primary: bool = False,
) -> str:
"""ניתוח-מחדש מאחד של טענות — מחלץ ממסמכים-עיקריים חדשים/לא-נותחו בלבד,
מאחד עם הטענות הקיימות (לא מוחק הכול), מריץ צבירה-מחדש, ומחזיר בדיקת-השפעה
(diff טיעונים/המלצה לפני↔אחרי) ליו"ר.
Args:
case_number: מספר תיק הערר.
reanalyze_all_primary: True = לחלץ מחדש מכל המסמכים-העיקריים (ולא רק
מאלה שטרם-נותחו). ברירת-מחדל False = רק מסמכים-עיקריים חדשים/לא-נותחו.
מנגנון האיחוד (לא force-delete): כל מסמך מחולץ דרך
``claims_extractor.extract_and_store_claims``, ש-``store_claims`` שלו מחליף רק
את טענות *אותו* מסמך (לפי ``source_document``) — כך טענות ממסמכים שכבר-נותחו
נשמרות. הצבירה-מחדש (``aggregate_claims_to_arguments(force=True)``) מחשבת את
הטיעונים מחדש מתוך **מערך-הטענות המאוחד השלם** — מסלול-החישוב הקנוני, לא מסלול
מקביל (G2).
"""
case = await db.get_case_by_number(case_number)
if not case:
return err(f"תיק {case_number} לא נמצא.")
case_id = UUID(case["id"])
# 1. BEFORE snapshot — current aggregated arguments (the impact baseline).
before_args = await argument_aggregator.get_legal_arguments(case_id)
before = _snapshot(before_args)
# 2. Select PRIMARY documents to (re)extract. Default: only those not yet
# analysed (the not-analysed flag). reanalyze_all_primary widens to every
# primary doc. Source of truth for "primary" = is_primary (V47).
if reanalyze_all_primary:
all_docs = await db.list_documents(case_id)
targets = [d for d in all_docs if d.get("is_primary")]
else:
targets = await db.primary_docs_not_analyzed(case_id)
if not targets:
return ok({
"case_number": case_number,
"status": "no_pending_documents",
"message": "אין מסמכים-עיקריים חדשים/לא-נותחו. לא בוצע ניתוח-מחדש.",
"documents_analyzed": [],
"impact": _impact_diff(before, before),
})
# 3. Extract+merge claims from the selected documents only. Each call replaces
# only that document's claims (merge), never the whole case (no force-delete).
analyzed: list[dict] = []
for doc in targets:
text = await db.get_document_text(UUID(doc["id"]))
if not text:
await db.mark_document_claims_extracted(UUID(doc["id"]), status="no_claims")
analyzed.append({"document": doc["title"], "status": "empty_text", "total": 0})
continue
res = await claims_extractor.extract_and_store_claims(
case_id=case_id,
document_id=UUID(doc["id"]),
text=text,
doc_type=doc["doc_type"],
)
analyzed.append({"document": doc["title"], **res})
# 4. Re-aggregate from the now-complete merged claim set. force=True is the
# canonical recompute (deletes only legal_arguments, NOT claims) — same
# path aggregate_claims_to_arguments always uses, not a fork.
agg = await argument_aggregator.aggregate_claims_to_arguments(case_id, force=True)
# 5. AFTER snapshot + impact diff for the chair.
after_args = await argument_aggregator.get_legal_arguments(case_id)
after = _snapshot(after_args)
impact = _impact_diff(before, after)
return ok({
"case_number": case_number,
"status": "completed",
"documents_analyzed": analyzed,
"aggregation": agg,
"impact": impact,
})