Chaim
f256eddbb1
All checks were successful
Build & Deploy / build-and-deploy (push) Successful in 1m25s
The case repo is the user's backup, so anything in the dir must end up
on Gitea. Two layers:
1. Periodic sweep (every 30s) — git_sync.sweep_loop runs as a FastAPI
background task. It scans every case dir, runs git status --porcelain
on each, and commit_and_push's any dirty changes with an auto-built
Hebrew message ("אוטו: טיוטות (2) · מסמכים"). Catches files written
outside the API path: agent research artefacts, manual edits, etc.
2. Explicit commits at known write paths — DOCX export, interim draft,
apply_user_edit, revise_draft, mark-final, analysis DOCX export.
These give immediate feedback with descriptive messages instead of
waiting up to 30s for the sweep.
safe.directory injection added to _git_env so sweep + explicit commits
work even when the running uid differs from the case-dir owner (host
runs vs. uniform-root container).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
856 lines
33 KiB
Python
856 lines
33 KiB
Python
"""MCP tools for decision drafting support."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from pathlib import Path
|
||
from uuid import UUID
|
||
|
||
from legal_mcp import config
|
||
from legal_mcp.services import db, embeddings, git_sync, research_md
|
||
from legal_mcp.services.lessons import (
|
||
CITATION_GUIDANCE,
|
||
DECISION_TEMPLATES,
|
||
DISCUSSION_RULES,
|
||
GOLDEN_RATIOS,
|
||
OPENING_STRATEGIES,
|
||
PARAGRAPH_LENGTHS,
|
||
SUMMARY_STRATEGIES,
|
||
TRANSITION_PHRASES,
|
||
VALID_OUTCOMES,
|
||
format_ratios_comment,
|
||
get_lessons_for_outcome,
|
||
)
|
||
|
||
# Fallback template for cases without expected_outcome
|
||
DECISION_TEMPLATE = """# החלטה
|
||
|
||
## בפני: דפנה תמיר, יו"ר ועדת הערר מחוז ירושלים
|
||
|
||
**ערר מספר:** {case_number}
|
||
**נושא:** {subject}
|
||
**העוררים:** {appellants}
|
||
**המשיבים:** {respondents}
|
||
**כתובת הנכס:** {property_address}
|
||
|
||
---
|
||
|
||
## א. רקע עובדתי
|
||
|
||
[תיאור הרקע העובדתי של הערר]
|
||
|
||
## ב. טענות העוררים
|
||
|
||
[סיכום טענות העוררים]
|
||
|
||
## ג. טענות המשיבים
|
||
|
||
[סיכום טענות המשיבים]
|
||
|
||
## ד. דיון והכרעה
|
||
|
||
[ניתוח משפטי]
|
||
|
||
## ה. מסקנה
|
||
|
||
[מסקנת הוועדה]
|
||
|
||
## ו. החלטה
|
||
|
||
[ההחלטה הסופית]
|
||
|
||
---
|
||
ניתנה היום, {date}
|
||
דפנה תמיר, יו"ר ועדת הערר
|
||
"""
|
||
|
||
|
||
async def get_style_guide() -> str:
|
||
"""שליפת דפוסי הסגנון של דפנה - נוסחאות, ביטויים אופייניים ומבנה, כולל לקחים מעשיים."""
|
||
patterns = await db.get_style_patterns()
|
||
|
||
result = "# מדריך סגנון - דפנה תמיר\n\n"
|
||
|
||
# Part 1: DB-sourced patterns (from analyze_style)
|
||
if patterns:
|
||
type_names = {
|
||
"opening_formula": "נוסחאות פתיחה",
|
||
"transition": "ביטויי מעבר",
|
||
"citation_style": "סגנון ציטוט",
|
||
"analysis_structure": "מבנה ניתוח",
|
||
"closing_formula": "נוסחאות סיום",
|
||
"characteristic_phrase": "ביטויים אופייניים",
|
||
"argument_flow": "זרימת טיעון",
|
||
"evidence_handling": "התייחסות לראיות",
|
||
}
|
||
|
||
grouped: dict[str, list] = {}
|
||
for p in patterns:
|
||
pt = p["pattern_type"]
|
||
if pt not in grouped:
|
||
grouped[pt] = []
|
||
grouped[pt].append({
|
||
"text": p["pattern_text"],
|
||
"context": p["context"],
|
||
"frequency": p["frequency"],
|
||
})
|
||
|
||
for ptype, items in grouped.items():
|
||
result += f"## {type_names.get(ptype, ptype)}\n\n"
|
||
for item in items:
|
||
result += f"- **{item['text']}** ({item['context']}, תדירות: {item['frequency']})\n"
|
||
result += "\n"
|
||
else:
|
||
result += "_לא נמצאו דפוסים מקורפוס. יש להעלות החלטות ולהריץ /style-report._\n\n"
|
||
|
||
# Part 2: Lessons-based guidance
|
||
result += "---\n\n# לקחים מעשיים\n\n"
|
||
|
||
# Universal discussion rules
|
||
result += "## כללי דיון אוניברסליים\n\n"
|
||
for rule in DISCUSSION_RULES["universal"]:
|
||
result += f"- {rule}\n"
|
||
result += "\n"
|
||
|
||
# Citation technique
|
||
result += "## טכניקת ציטוט\n\n"
|
||
result += f"{CITATION_GUIDANCE}\n\n"
|
||
|
||
# Transition phrases
|
||
result += "## ביטויי מעבר (מהשוואת טיוטות)\n\n"
|
||
for p in TRANSITION_PHRASES:
|
||
ctx = p["context"]
|
||
outcome_note = f" [בעיקר ב-{p['outcome']}]" if p["outcome"] else ""
|
||
result += f"- **{p['phrase']}** — {ctx}{outcome_note}\n"
|
||
result += "\n"
|
||
|
||
# Paragraph lengths
|
||
result += "## אורכי פסקאות מומלצים\n\n"
|
||
result += "| סוג | מילים |\n|-----|------|\n"
|
||
labels = {
|
||
"claims": "טענות",
|
||
"discussion_regular": "דיון רגיל",
|
||
"discussion_with_citation": "דיון + ציטוט",
|
||
"discussion_average": "ממוצע דיון",
|
||
}
|
||
for key, (lo, hi) in PARAGRAPH_LENGTHS.items():
|
||
result += f"| {labels.get(key, key)} | {lo}-{hi} |\n"
|
||
result += "\n"
|
||
|
||
# Golden ratios
|
||
result += "## יחסי הזהב (אחוזי סעיפים מסך ההחלטה)\n\n"
|
||
result += "| סוג ערר | רקע | טענות | דיון | סיכום |\n"
|
||
result += "|---------|------|-------|------|-------|\n"
|
||
outcome_labels = {
|
||
"rejection": "רישוי נדחה",
|
||
"full_acceptance": "רישוי מתקבל",
|
||
"partial_acceptance": "רישוי קבלה חלקית",
|
||
"betterment_levy": "היטל השבחה",
|
||
}
|
||
for outcome in VALID_OUTCOMES:
|
||
r = GOLDEN_RATIOS[outcome]
|
||
result += (
|
||
f"| {outcome_labels[outcome]} "
|
||
f"| {r['background'][0]}-{r['background'][1]}% "
|
||
f"| {r['claims'][0]}-{r['claims'][1]}% "
|
||
f"| {r['discussion'][0]}-{r['discussion'][1]}% "
|
||
f"| {r['summary'][0]}-{r['summary'][1]}% |\n"
|
||
)
|
||
result += "\n"
|
||
|
||
# Opening and summary strategies
|
||
result += "## אסטרטגיות פתיחה וסיכום לפי תוצאה\n\n"
|
||
for outcome in VALID_OUTCOMES:
|
||
opening = OPENING_STRATEGIES[outcome]
|
||
summary = SUMMARY_STRATEGIES[outcome]
|
||
result += f"### {outcome_labels[outcome]}\n"
|
||
result += f"- **פתיחה:** {opening['description']} ({opening['paragraphs'][0]}-{opening['paragraphs'][1]} פסקאות)\n"
|
||
result += f"- **סיכום ({summary['heading']}):** {summary['description']}\n\n"
|
||
|
||
return result
|
||
|
||
|
||
async def draft_section(
|
||
case_number: str,
|
||
section: str,
|
||
instructions: str = "",
|
||
) -> str:
|
||
"""הרכבת הקשר מלא לניסוח סעיף בהחלטה - כולל עובדות מהמסמכים, תקדימים רלוונטיים ודפוסי סגנון.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
section: סוג הסעיף (facts, appellant_claims, respondent_claims, legal_analysis, conclusion, ruling)
|
||
instructions: הנחיות נוספות לניסוח
|
||
"""
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
expected_outcome = case.get("expected_outcome", "")
|
||
|
||
# 1. Get relevant chunks from case documents
|
||
section_query = {
|
||
"facts": "רקע עובדתי של התיק",
|
||
"appellant_claims": "טענות העוררים",
|
||
"respondent_claims": "טענות המשיבים",
|
||
"legal_analysis": "ניתוח משפטי ודיון",
|
||
"conclusion": "מסקנות",
|
||
"ruling": "החלטה",
|
||
}.get(section, section)
|
||
|
||
query_emb = await embeddings.embed_query(section_query)
|
||
case_chunks = await db.search_similar(
|
||
query_embedding=query_emb, limit=10, case_id=case_id
|
||
)
|
||
|
||
# 2. Get similar sections from precedents
|
||
precedent_chunks = await db.search_similar(
|
||
query_embedding=query_emb, limit=5, section_type=section
|
||
)
|
||
# Filter out chunks from the same case
|
||
precedent_chunks = [c for c in precedent_chunks if str(c["case_id"]) != case["id"]]
|
||
|
||
# 3. Get style patterns
|
||
style_patterns = await db.get_style_patterns()
|
||
|
||
# Build context
|
||
context = {
|
||
"case": {
|
||
"case_number": case["case_number"],
|
||
"title": case["title"],
|
||
"appellants": case["appellants"],
|
||
"respondents": case["respondents"],
|
||
"subject": case["subject"],
|
||
"property_address": case["property_address"],
|
||
"expected_outcome": expected_outcome,
|
||
},
|
||
"section": section,
|
||
"instructions": instructions,
|
||
"case_documents": [
|
||
{
|
||
"document": c["document_title"],
|
||
"section_type": c["section_type"],
|
||
"content": c["content"],
|
||
}
|
||
for c in case_chunks
|
||
],
|
||
"precedents": [
|
||
{
|
||
"case_number": c["case_number"],
|
||
"document": c["document_title"],
|
||
"content": c["content"][:500],
|
||
}
|
||
for c in precedent_chunks[:3]
|
||
],
|
||
"style_patterns": [
|
||
{
|
||
"type": p["pattern_type"],
|
||
"text": p["pattern_text"],
|
||
}
|
||
for p in style_patterns[:15]
|
||
],
|
||
}
|
||
|
||
# 4. Add outcome-aware drafting guidance
|
||
if expected_outcome and expected_outcome in VALID_OUTCOMES:
|
||
lessons = get_lessons_for_outcome(expected_outcome)
|
||
guidance: dict = {
|
||
"outcome": expected_outcome,
|
||
"golden_ratios": lessons["golden_ratios"],
|
||
"citation_guidance": lessons["citation_guidance"],
|
||
}
|
||
|
||
# Section-specific guidance
|
||
if section == "legal_analysis":
|
||
guidance["discussion_rules"] = lessons["discussion_rules"]
|
||
guidance["opening_strategy"] = lessons["opening_strategy"]
|
||
guidance["transition_phrases"] = lessons["transition_phrases"]
|
||
guidance["paragraph_lengths"] = lessons["paragraph_lengths"]
|
||
elif section in ("conclusion", "ruling"):
|
||
guidance["summary_strategy"] = lessons["summary_strategy"]
|
||
elif section == "facts":
|
||
guidance["target_ratio"] = lessons["golden_ratios"].get("background", "")
|
||
guidance["paragraph_lengths"] = {"claims": lessons["paragraph_lengths"].get("claims", "")}
|
||
elif section in ("appellant_claims", "respondent_claims"):
|
||
guidance["target_ratio"] = lessons["golden_ratios"].get("claims", "")
|
||
guidance["paragraph_lengths"] = {"claims": lessons["paragraph_lengths"].get("claims", "")}
|
||
|
||
context["drafting_guidance"] = guidance
|
||
|
||
return json.dumps(context, ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def get_chair_directions(case_number: str) -> str:
|
||
"""שליפת עמדות יו"ר הוועדה (דפנה) על סוגיות הערר, לצורך יצירת direction_doc
|
||
לכותב. קורא מ-analysis-and-research.md (שנוצר ע"י legal-analyst ומולא ע"י
|
||
דפנה דרך ה-UI).
|
||
|
||
מחזיר JSON עם סטטוס, כמה סוגיות מולאו וכמה עדיין ריקות, ורשימה של עמדות
|
||
מובנות — ניתן להזריק ישירות כ-direction_doc לבלוק י (דיון) ולבלוק יא (סיכום).
|
||
|
||
סטטוסים:
|
||
missing — הקובץ לא קיים
|
||
empty — הקובץ קיים אבל כל העמדות ריקות (טרם נקבעה דעה)
|
||
partial — חלק מהעמדות מולאו
|
||
complete — כל העמדות מולאו
|
||
|
||
אם המצב הוא `empty` או `missing` — הכותב צריך לעצור ולבקש מדפנה למלא
|
||
את הקובץ דרך ה-UI לפני המשך הכתיבה.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
"""
|
||
case_dir = config.find_case_dir(case_number)
|
||
file_path = case_dir / "documents" / "research" / "analysis-and-research.md"
|
||
result = research_md.extract_chair_directions(file_path)
|
||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def get_decision_template(case_number: str) -> str:
|
||
"""קבלת תבנית מבנית להחלטה מלאה עם פרטי התיק, מותאמת לסוג התוצאה הצפויה.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
"""
|
||
from datetime import date
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
expected_outcome = case.get("expected_outcome", "")
|
||
|
||
format_args = dict(
|
||
case_number=case["case_number"],
|
||
subject=case["subject"],
|
||
appellants=", ".join(case.get("appellants", [])),
|
||
respondents=", ".join(case.get("respondents", [])),
|
||
property_address=case.get("property_address", ""),
|
||
date=date.today().strftime("%d.%m.%Y"),
|
||
)
|
||
|
||
# Use outcome-specific template if available
|
||
if expected_outcome and expected_outcome in DECISION_TEMPLATES:
|
||
# Add ratio comments
|
||
format_args["ratios_background"] = format_ratios_comment(expected_outcome, "background")
|
||
format_args["ratios_claims"] = format_ratios_comment(expected_outcome, "claims")
|
||
format_args["ratios_discussion"] = format_ratios_comment(expected_outcome, "discussion")
|
||
format_args["ratios_summary"] = format_ratios_comment(expected_outcome, "summary")
|
||
|
||
template = DECISION_TEMPLATES[expected_outcome].format(**format_args)
|
||
|
||
# Add guidance header
|
||
opening = OPENING_STRATEGIES[expected_outcome]
|
||
summary = SUMMARY_STRATEGIES[expected_outcome]
|
||
header = (
|
||
f"<!-- תבנית מותאמת ל: {expected_outcome} -->\n"
|
||
f"<!-- פתיחת דיון: {opening['description']} -->\n"
|
||
f"<!-- סיכום: {summary['description']} -->\n\n"
|
||
)
|
||
return header + template
|
||
else:
|
||
# Fallback to generic template
|
||
template = DECISION_TEMPLATE.format(**format_args)
|
||
if not expected_outcome:
|
||
template = (
|
||
"<!-- לא הוגדרה תוצאה צפויה. הגדר expected_outcome בתיק לקבלת תבנית מותאמת. -->\n"
|
||
f"<!-- ערכים אפשריים: {', '.join(VALID_OUTCOMES)} -->\n\n"
|
||
) + template
|
||
return template
|
||
|
||
|
||
async def validate_decision(case_number: str) -> str:
|
||
"""בדיקת QA אוטומטית על ההחלטה — 6 בדיקות. אם נכשלת בדיקה קריטית — ייצוא חסום.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
"""
|
||
from legal_mcp.services import qa_validator
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
|
||
try:
|
||
result = await qa_validator.validate_decision(case_id)
|
||
return json.dumps(result, default=str, ensure_ascii=False, indent=2)
|
||
except ValueError as e:
|
||
return json.dumps({"status": "error", "message": str(e)}, ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def export_docx(case_number: str, output_path: str = "") -> str:
|
||
"""ייצוא החלטה לקובץ DOCX מעוצב — גופן David, RTL, כותרות, מספור סעיפים.
|
||
|
||
הקובץ נוצר עם bookmarks ב-12 הבלוקים (אנקורים ל-revisions עתידיים),
|
||
ומסומן כ-active_draft_path של התיק.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
output_path: נתיב לשמירה (אופציונלי — ברירת מחדל: תיקיית התיק)
|
||
"""
|
||
from legal_mcp.services import docx_exporter
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
|
||
try:
|
||
path = await docx_exporter.export_decision(case_id, output_path or None)
|
||
# Register this export as the new source of truth
|
||
await db.set_active_draft_path(case_id, path)
|
||
case_dir = config.find_case_dir(case_number)
|
||
if case_dir.exists():
|
||
git_sync.commit_and_push(case_dir, f"ייצוא DOCX: {Path(path).name}")
|
||
return json.dumps({
|
||
"status": "completed",
|
||
"path": path,
|
||
"active_draft_path": path,
|
||
"message": f"DOCX נוצר: {path}",
|
||
}, ensure_ascii=False, indent=2)
|
||
except ValueError as e:
|
||
return json.dumps({
|
||
"status": "error",
|
||
"message": str(e),
|
||
}, ensure_ascii=False, indent=2)
|
||
|
||
|
||
# ── Interim draft (pre-ruling) ────────────────────────────────────
|
||
|
||
# Blocks written for the interim draft, in display order.
|
||
# This is the same content the chair sees in the final decision (same template,
|
||
# same skill, same prompts) — minus opening, ruling, summary, signatures.
|
||
_INTERIM_BLOCKS = ["block-vav", "block-tet", "block-zayin", "block-chet"]
|
||
|
||
|
||
async def extract_appraiser_facts(case_number: str) -> str:
|
||
"""חילוץ תכניות והיתרים מכל השומות בתיק וזיהוי סתירות בין שמאים.
|
||
|
||
משמש כהכנה לטיוטת ביניים: בלוק ט (תכניות חלות) זקוק לעובדות מובנות
|
||
כדי לפרט תת-פרק היתרים ולסמן סתירות בנוסח ניטרלי.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
"""
|
||
from legal_mcp.services import appraiser_facts_extractor
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return json.dumps({"status": "error",
|
||
"message": f"תיק {case_number} לא נמצא."},
|
||
ensure_ascii=False, indent=2)
|
||
case_id = UUID(case["id"])
|
||
try:
|
||
result = await appraiser_facts_extractor.extract_appraiser_facts(case_id)
|
||
return json.dumps(result, default=str, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
return json.dumps({"status": "error", "message": str(e)},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def write_interim_draft(case_number: str, instructions: str = "") -> str:
|
||
"""כתיבת ארבעת הבלוקים לטיוטת ביניים: רקע (ו), תכניות+היתרים (ט),
|
||
טענות הצדדים (ז), הליכים (ח). אם לא חולצו עובדות שמאיות עדיין —
|
||
מריץ extract_appraiser_facts קודם כדי שבלוק ט יקבל פרק היתרים תקף.
|
||
|
||
הבלוקים נכתבים באותו skill, אותם prompts ואותו טמפלט כמו בטיוטה רגילה —
|
||
הסדר משתנה רק בעת הייצוא ל-DOCX (ראה export_interim_draft).
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
instructions: הנחיות נוספות (לכל הבלוקים)
|
||
"""
|
||
from legal_mcp.services import appraiser_facts_extractor, block_writer
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return json.dumps({"status": "error",
|
||
"message": f"תיק {case_number} לא נמצא."},
|
||
ensure_ascii=False, indent=2)
|
||
case_id = UUID(case["id"])
|
||
|
||
# Make sure appraiser facts exist before writing block-tet (which depends on them).
|
||
facts = await db.list_appraiser_facts(case_id)
|
||
facts_run: dict | None = None
|
||
if not facts:
|
||
try:
|
||
facts_run = await appraiser_facts_extractor.extract_appraiser_facts(case_id)
|
||
except Exception as e:
|
||
facts_run = {"status": "error", "message": str(e)}
|
||
|
||
results = []
|
||
for bid in _INTERIM_BLOCKS:
|
||
try:
|
||
r = await block_writer.write_and_store_block(case_id, bid, instructions)
|
||
results.append({
|
||
"block_id": bid,
|
||
"title": r["title"],
|
||
"word_count": r["word_count"],
|
||
"status": "completed",
|
||
})
|
||
except Exception as e:
|
||
results.append({
|
||
"block_id": bid,
|
||
"status": "error",
|
||
"error": str(e),
|
||
})
|
||
|
||
return json.dumps({
|
||
"status": "completed",
|
||
"blocks": results,
|
||
"appraiser_facts_run": facts_run,
|
||
"total_words": sum(r.get("word_count", 0) for r in results),
|
||
"completed": sum(1 for r in results if r["status"] == "completed"),
|
||
}, default=str, ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def export_interim_draft(case_number: str, output_path: str = "") -> str:
|
||
"""ייצוא טיוטת ביניים ל-DOCX — אותו עיצוב של טיוטה רגילה (David, RTL,
|
||
bookmarks), אבל בסדר חדש: רקע → תכניות+היתרים → טענות → הליכים, ללא
|
||
דיון/סיכום/חתימות. שם הקובץ: טיוטת-ביניים-v{N}.docx.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
output_path: נתיב לשמירה (אופציונלי)
|
||
"""
|
||
from legal_mcp.services import docx_exporter
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return json.dumps({"status": "error",
|
||
"message": f"תיק {case_number} לא נמצא."},
|
||
ensure_ascii=False, indent=2)
|
||
case_id = UUID(case["id"])
|
||
|
||
try:
|
||
path = await docx_exporter.export_decision(
|
||
case_id, output_path or None, mode="interim",
|
||
)
|
||
await db.set_active_draft_path(case_id, path)
|
||
case_dir = config.find_case_dir(case_number)
|
||
if case_dir.exists():
|
||
git_sync.commit_and_push(case_dir, f"טיוטת ביניים: {Path(path).name}")
|
||
return json.dumps({
|
||
"status": "completed",
|
||
"mode": "interim",
|
||
"path": path,
|
||
"active_draft_path": path,
|
||
"message": f"טיוטת ביניים נוצרה: {path}",
|
||
}, ensure_ascii=False, indent=2)
|
||
except ValueError as e:
|
||
return json.dumps({"status": "error", "message": str(e)},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def apply_user_edit(case_number: str, edit_filename: str) -> str:
|
||
"""רישום עריכה שהעלה המשתמש כמקור האמת החדש של התיק.
|
||
|
||
התהליך:
|
||
1. מאתר את הקובץ `עריכה-v*.docx` בתיקיית ה-exports
|
||
2. מזריק bookmarks רטרואקטיבית (אם אין) דרך docx_retrofit
|
||
3. מעדכן את cases.active_draft_path
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
edit_filename: שם הקובץ (למשל "עריכה-v1.docx") או נתיב מלא
|
||
"""
|
||
from legal_mcp.services import docx_retrofit
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return json.dumps({"status": "error",
|
||
"message": f"תיק {case_number} לא נמצא."},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
case_id = UUID(case["id"])
|
||
export_dir = config.find_case_dir(case_number) / "exports"
|
||
edit_path = export_dir / edit_filename if "/" not in edit_filename else Path(edit_filename)
|
||
if not edit_path.exists():
|
||
return json.dumps({"status": "error",
|
||
"message": f"קובץ לא נמצא: {edit_path}"},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
try:
|
||
retrofit_result = docx_retrofit.retrofit_bookmarks(edit_path)
|
||
await db.set_active_draft_path(case_id, str(edit_path))
|
||
case_dir = config.find_case_dir(case_number)
|
||
if case_dir.exists():
|
||
git_sync.commit_and_push(case_dir, f"גרסת עריכה: {edit_path.name}")
|
||
return json.dumps({
|
||
"status": "completed",
|
||
"active_draft_path": str(edit_path),
|
||
"bookmarks_added": retrofit_result.get("bookmarks_added", []),
|
||
"missing_blocks": retrofit_result.get("missing_blocks", []),
|
||
"structural_fallback": retrofit_result.get("structural_fallback", []),
|
||
"existing_bookmarks": retrofit_result.get("existing_bookmarks", []),
|
||
}, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
return json.dumps({"status": "error", "message": str(e)},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def list_bookmarks(case_number: str) -> str:
|
||
"""רשימת bookmarks הקיימים ב-active_draft של התיק.
|
||
|
||
משמש לסוכנים כדי לדעת אילו אנקורים זמינים לפני שליחת revisions.
|
||
"""
|
||
from legal_mcp.services import docx_reviser
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return json.dumps({"status": "error",
|
||
"message": f"תיק {case_number} לא נמצא."},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
active_path = await db.get_active_draft_path(UUID(case["id"]))
|
||
if not active_path or not Path(active_path).exists():
|
||
return json.dumps({"status": "no_active_draft",
|
||
"message": "לא נמצא active_draft. הרץ ייצוא או העלה עריכה."},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
try:
|
||
names = docx_reviser.list_bookmarks(active_path)
|
||
return json.dumps({
|
||
"status": "completed",
|
||
"active_draft_path": active_path,
|
||
"bookmarks": names,
|
||
}, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
return json.dumps({"status": "error", "message": str(e)},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def revise_draft(case_number: str, revisions_json: str,
|
||
author: str = "מערכת AI") -> str:
|
||
"""החלת revisions מסומנים כ-Track Changes על ה-active_draft של התיק.
|
||
|
||
יוצר קובץ חדש `טיוטה-v{N+1}.docx` (מגרסה הבאה בתור), ומעדכן את
|
||
active_draft_path אליו.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
revisions_json: JSON string של array עם אובייקטים:
|
||
[{"id": "r1", "type": "insert_after"|"insert_before"|"replace"|"delete",
|
||
"anchor_bookmark": "block-yod", "content": "...", "style": "body"|"heading"|"quote",
|
||
"reason": "..."}, ...]
|
||
author: מחרוזת המחבר שתופיע ב-Track Changes
|
||
"""
|
||
from legal_mcp.services import docx_reviser
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return json.dumps({"status": "error",
|
||
"message": f"תיק {case_number} לא נמצא."},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
case_id = UUID(case["id"])
|
||
active_path = await db.get_active_draft_path(case_id)
|
||
if not active_path or not Path(active_path).exists():
|
||
return json.dumps({"status": "error",
|
||
"message": "אין active_draft. הרץ ייצוא או apply_user_edit קודם."},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
try:
|
||
raw = json.loads(revisions_json) if isinstance(revisions_json, str) else revisions_json
|
||
except json.JSONDecodeError as e:
|
||
return json.dumps({"status": "error", "message": f"JSON לא תקף: {e}"},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
revisions = []
|
||
for item in raw:
|
||
revisions.append(docx_reviser.Revision(
|
||
id=item.get("id", ""),
|
||
type=item["type"],
|
||
anchor_bookmark=item["anchor_bookmark"],
|
||
content=item.get("content", ""),
|
||
style=item.get("style", "body"),
|
||
reason=item.get("reason", ""),
|
||
anchor_position=item.get("anchor_position", "end"),
|
||
))
|
||
|
||
# Determine output path — next טיוטה-v{N}.docx
|
||
export_dir = config.find_case_dir(case_number) / "exports"
|
||
export_dir.mkdir(parents=True, exist_ok=True)
|
||
existing = list(export_dir.glob("טיוטה-v*.docx"))
|
||
next_ver = 1
|
||
for p in existing:
|
||
try:
|
||
ver = int(p.stem.split("-v")[1])
|
||
next_ver = max(next_ver, ver + 1)
|
||
except (IndexError, ValueError):
|
||
pass
|
||
output_path = export_dir / f"טיוטה-v{next_ver}.docx"
|
||
|
||
try:
|
||
result = docx_reviser.apply_tracked_revisions(
|
||
active_path, output_path, revisions, author=author,
|
||
)
|
||
await db.set_active_draft_path(case_id, str(output_path))
|
||
case_dir = config.find_case_dir(case_number)
|
||
if case_dir.exists():
|
||
git_sync.commit_and_push(
|
||
case_dir,
|
||
f"revise: טיוטה-v{next_ver} ({result.applied} שינויים, {result.failed} נכשלו)",
|
||
)
|
||
return json.dumps({
|
||
"status": "completed",
|
||
"output_path": str(output_path),
|
||
"version": next_ver,
|
||
"applied": result.applied,
|
||
"failed": result.failed,
|
||
"active_draft_path": str(output_path),
|
||
"results": [
|
||
{"id": r.id, "status": r.status, "error": r.error}
|
||
for r in result.results
|
||
],
|
||
}, ensure_ascii=False, indent=2)
|
||
except Exception as e:
|
||
return json.dumps({"status": "error", "message": str(e)},
|
||
ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def get_block_context(case_number: str, block_id: str, instructions: str = "") -> str:
|
||
"""קבלת הקשר מלא לכתיבת בלוק — ללא קריאה ל-API. Claude Code כותב את הבלוק.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
block_id: מזהה הבלוק (block-he, block-vav, ..., block-yod-bet)
|
||
instructions: הנחיות נוספות
|
||
"""
|
||
from legal_mcp.services import block_writer
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
try:
|
||
ctx = await block_writer.get_block_context(case_id, block_id, instructions)
|
||
return json.dumps(ctx, default=str, ensure_ascii=False, indent=2)
|
||
except ValueError as e:
|
||
return str(e)
|
||
|
||
|
||
async def save_block_content(case_number: str, block_id: str, content: str) -> str:
|
||
"""שמירת בלוק שנכתב ע"י Claude Code ב-DB.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
block_id: מזהה הבלוק
|
||
content: הטקסט שנכתב
|
||
"""
|
||
from legal_mcp.services import block_writer
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
try:
|
||
result = await block_writer.save_block_content(case_id, block_id, content)
|
||
return json.dumps(result, default=str, ensure_ascii=False, indent=2)
|
||
except ValueError as e:
|
||
return str(e)
|
||
|
||
|
||
async def analyze_style(appeal_subtype: str = "") -> str:
|
||
"""הרצת ניתוח סגנון על קורפוס ההחלטות של דפנה. מחלץ דפוסי כתיבה ושומר אותם.
|
||
|
||
Args:
|
||
appeal_subtype: סינון לפי סוג ערר (building_permit / betterment_levy / compensation_197).
|
||
ריק = כל ההחלטות.
|
||
"""
|
||
from legal_mcp.services.style_analyzer import analyze_corpus
|
||
|
||
result = await analyze_corpus(appeal_subtype)
|
||
return json.dumps(result, ensure_ascii=False, indent=2)
|
||
|
||
|
||
async def write_block(
|
||
case_number: str,
|
||
block_id: str,
|
||
instructions: str = "",
|
||
) -> str:
|
||
"""כתיבת בלוק יחיד בהחלטה. כותב ושומר ב-DB.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
block_id: מזהה הבלוק: block-alef, block-bet, block-gimel, block-dalet, block-he, block-vav, block-zayin, block-chet, block-tet, block-yod, block-yod-alef, block-yod-bet
|
||
instructions: הנחיות נוספות לניסוח
|
||
"""
|
||
from legal_mcp.services import block_writer
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
|
||
try:
|
||
result = await block_writer.write_and_store_block(case_id, block_id, instructions)
|
||
return json.dumps(result, default=str, ensure_ascii=False, indent=2)
|
||
except ValueError as e:
|
||
return str(e)
|
||
|
||
|
||
async def write_all_blocks(
|
||
case_number: str,
|
||
start_from: str = "block-alef",
|
||
instructions: str = "",
|
||
) -> str:
|
||
"""כתיבת כל הבלוקים בהחלטה, בלוק אחרי בלוק. שומר כל בלוק מיד אחרי כתיבה.
|
||
|
||
Args:
|
||
case_number: מספר תיק הערר
|
||
start_from: מאיזה בלוק להתחיל (ברירת מחדל: block-alef)
|
||
instructions: הנחיות כלליות
|
||
"""
|
||
from legal_mcp.services import block_writer
|
||
from legal_mcp.services.block_writer import BLOCK_CONFIG
|
||
|
||
case = await db.get_case_by_number(case_number)
|
||
if not case:
|
||
return f"תיק {case_number} לא נמצא."
|
||
|
||
case_id = UUID(case["id"])
|
||
|
||
# Determine start index
|
||
start_idx = BLOCK_CONFIG.get(start_from, {}).get("index", 1)
|
||
|
||
results = []
|
||
block_order = sorted(BLOCK_CONFIG.items(), key=lambda x: x[1]["index"])
|
||
|
||
for bid, cfg in block_order:
|
||
if cfg["index"] < start_idx:
|
||
continue
|
||
|
||
try:
|
||
result = await block_writer.write_and_store_block(case_id, bid, instructions)
|
||
results.append({
|
||
"block_id": bid,
|
||
"title": result["title"],
|
||
"word_count": result["word_count"],
|
||
"status": "completed",
|
||
})
|
||
except ValueError as e:
|
||
results.append({
|
||
"block_id": bid,
|
||
"title": cfg["title"],
|
||
"status": "error",
|
||
"error": str(e),
|
||
})
|
||
# Stop on critical error (e.g., missing direction for block-yod)
|
||
if "כיוון מאושר" in str(e):
|
||
break
|
||
|
||
total_words = sum(r.get("word_count", 0) for r in results)
|
||
return json.dumps({
|
||
"blocks": results,
|
||
"total_words": total_words,
|
||
"completed": sum(1 for r in results if r["status"] == "completed"),
|
||
}, default=str, ensure_ascii=False, indent=2)
|