shira-hermes/api/routes/legal_aid.py

"""Legal Aid PDF parser endpoint.

Parses a monthly Legal Aid (סיוע משפטי) payment-report PDF and returns the
JSON shape that GreenInvoiceBilling's LegalAidPaymentReportService.ingest()
expects.

v3 (2026-05-03): switched extraction from pdfplumber to Claude Vision via
ai-gateway. The Legal Aid PDF format has multi-line cells, RTL Hebrew column
ordering quirks, and amounts that wrap across rows ("1478.5\\n4" → 1478.54).
pdfplumber's table detection couldn't get past those cleanly. Claude reads
the rendered page image natively and returns structured rows.

Pattern mirrors api/services/ocr.py — render each PDF page with PyMuPDF,
send the PNG to ai-gateway with a JSON-output prompt, then assemble the
result. Auth is X-Api-Key as for the other admin endpoints.
"""

from __future__ import annotations

import asyncio
import base64
import io
import json
import logging
import os
import re
import time
import uuid
from typing import Any

from fastapi import APIRouter, File, HTTPException, Query, Request, UploadFile

logger = logging.getLogger("shira.legal_aid")

router = APIRouter(prefix="/api/legal-aid", tags=["legal-aid"])


def _verify_admin(request: Request) -> None:
    expected = os.environ.get("ADMIN_API_KEY") or os.environ.get("API_KEY", "")
    if not expected:
        raise HTTPException(status_code=503, detail="Admin API key not configured")
    provided = request.headers.get("X-Admin-Key") or request.headers.get("X-Api-Key") or ""
    if provided != expected:
        raise HTTPException(status_code=401, detail="Unauthorized")


_PDF_RENDER_DPI = 180
_PDF_MAX_PAGES = 8  # safety cap; real reports are 1-2 pages

# --- Async job store ---------------------------------------------------
# Claude Vision via ai-gateway's claude-code OAuth provider runs ~140s per
# rendered page (a 3-page report measured at 417s). That far exceeds the
# EspoCRM controller's synchronous curl timeout, so manual uploads run as a
# background job: POST ?mode=async returns a jobId immediately and the caller
# polls GET /parse-pdf-report/jobs/{jobId}. Jobs are persisted as files under
# /opt/data (a named volume) so they survive across uvicorn workers and short
# restarts. n8n / other server-side callers can still use the default sync mode.
_JOBS_DIR = os.environ.get("LEGAL_AID_JOBS_DIR", "/opt/data/legal-aid-jobs")
_JOB_TTL_SECONDS = 3600  # finished jobs are reaped after 1h
_running_jobs: set[asyncio.Task] = set()


def _jobs_dir() -> str:
    os.makedirs(_JOBS_DIR, exist_ok=True)
    return _JOBS_DIR


def _job_path(job_id: str) -> str:
    # job_id is a server-generated uuid hex; sanitize defensively anyway so a
    # crafted id from the status route can never escape the jobs directory.
    safe = re.sub(r"[^a-f0-9]", "", job_id.lower())
    if not safe:
        safe = "_invalid_"
    return os.path.join(_jobs_dir(), f"{safe}.json")


def _write_job(job_id: str, data: dict[str, Any]) -> None:
    path = _job_path(job_id)
    tmp = f"{path}.tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False)
    os.replace(tmp, path)  # atomic publish


def _read_job(job_id: str) -> dict[str, Any] | None:
    path = _job_path(job_id)
    if not os.path.exists(path):
        return None
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except (OSError, ValueError):
        return None


def _cleanup_old_jobs() -> None:
    try:
        now = time.time()
        for name in os.listdir(_jobs_dir()):
            if not name.endswith(".json"):
                continue
            p = os.path.join(_JOBS_DIR, name)
            try:
                if now - os.path.getmtime(p) > _JOB_TTL_SECONDS:
                    os.remove(p)
            except OSError:
                pass
    except OSError:
        pass


_EXTRACT_PROMPT = """\
אתה מחלץ נתונים מדוח תשלומים חודשי של הסיוע המשפטי (Legal Aid, עברית RTL).

תוציא JSON תקני בלבד (בלי backticks, בלי טקסט נוסף) בסכמה הזו:
{
  "issuerOffice": "Jerusalem" | "TelAviv" | "Haifa" | "BeerSheva" | "Nazareth" | "Center" | "National" | null,
  "reportDate": "YYYY-MM-DD" | null,
  "reportMonth": "YYYY-MM" | null,
  "totalExclVat": number,
  "totalVat": number,
  "totalInclVat": number,
  "lines": [
    {
      "invoiceNumber": "<16 ספרות מתוך 'מספר חשבונית'>",
      "requestNumber": "<7 ספרות מתוך 'מספר בקשה'>",
      "legalAidCaseNumber": "<7 ספרות מתוך 'מספר תיק'>",
      "clientName": "<שם לקוח בעברית, מאוחד אם נשבר לשני שורות>",
      "activityCategory": "<עמודת 'פעולה' — קטגוריית האב, למשל: הודעה, לימוד ענין, התיעצות / פגישה, בקשה, דיון>",
      "activitySubTypeHebrew": "<עמודת 'תת סוג פעולה' — תווית מלאה, למשל: הודעה בהתאם להחלטה, לימוד ענין מלא, התייעצות עם מסמכים>",
      "activityDate": "<עמודת 'תאריך פעילות' בפורמט YYYY-MM-DD>",
      "amountExclVat": <עמודת 'סכום' (ללא מע\"מ) כמספר עשרוני>,
      "vat": <עמודת 'מע\"מ' כמספר עשרוני>,
      "amountInclVat": <עמודת 'סה\"כ' כמספר עשרוני>
    }
  ],
  "travelReimbursementLines": []
}

הנחיות חשובות:
- הטבלה הראשית בעמוד 1 היא "פעולות חייבות במע\"מ" — חלץ ממנה את כל השורות.
- אם תא בטבלה משתרע על שתי שורות (כמו "הודעה בהתאם" + "להחלטה"), אחד אותם לטקסט אחד עם רווח.
- אם סכום נשבר לשורות (כמו "1478.5" ו-"4"), אחד אותם ל-1478.54.
- "מספר חשבונית" הוא תמיד 16 ספרות; "מספר בקשה" ו-"מספר תיק" 7 ספרות כל אחד.
- שדה reportMonth מתוך "עבור חודש: M/YYYY" → "YYYY-MM".
- שדה reportDate מתוך התאריך בכותרת הראשית של הדוח (DD/MM/YYYY → YYYY-MM-DD).
- אם אין שדה — null. אם הטבלה ריקה (כמו טבלת נסיעות בעמוד 2) — אל תכניס את שורותיה ל-lines.
- ה-issuerOffice לפי "מחוז ירושלים" → "Jerusalem" וכו'.

החזר JSON בלבד.
"""


async def _parse_pdf_to_payload(
    pdf_bytes: bytes, subject: str | None, debug: bool
) -> dict[str, Any]:
    """Render the PDF pages and run Claude Vision to produce the ingest payload.

    Shared by both the synchronous route and the async job runner. Raises
    HTTPException on any failure (the job runner converts those to a stored
    error status).
    """
    if not pdf_bytes:
        raise HTTPException(status_code=400, detail="Empty file")
    if pdf_bytes[:4] != b"%PDF":
        raise HTTPException(status_code=400, detail="File is not a PDF")

    try:
        import fitz  # PyMuPDF
    except ImportError as e:
        raise HTTPException(status_code=503, detail=f"PyMuPDF not installed: {e}")

    images_b64: list[tuple[str, str]] = []
    page_count = 0
    try:
        with fitz.open(stream=pdf_bytes, filetype="pdf") as doc:
            page_count = min(len(doc), _PDF_MAX_PAGES)
            for i in range(page_count):
                pix = doc[i].get_pixmap(dpi=_PDF_RENDER_DPI)
                # JPEG (not PNG): a rendered table page is mostly white, and
                # PNG of it is several × larger than a q85 JPEG. Bundling all
                # pages as PNG blew past ai-gateway's body limit (413). JPEG
                # q85 keeps the table text crisp enough for Claude Vision.
                jpg = pix.tobytes("jpeg", jpg_quality=85)
                images_b64.append(("image/jpeg", base64.b64encode(jpg).decode("ascii")))
    except Exception as e:
        logger.exception("Legal Aid PDF render failed")
        raise HTTPException(status_code=422, detail=f"PDF render error: {e}")

    if not images_b64:
        raise HTTPException(status_code=422, detail="PDF contained no pages")

    raw_response = await _call_vision_for_json(images_b64, _EXTRACT_PROMPT)
    parsed = _parse_json_response(raw_response)

    sysref = None
    if subject:
        m = re.search(r"\[סימוכין\s+(\d+)\]", subject)
        if m:
            sysref = m.group(1)

    out: dict[str, Any] = {
        "subject": subject,
        "sysref": sysref,
        "issuerOffice": parsed.get("issuerOffice"),
        "reportDate": parsed.get("reportDate"),
        "reportMonth": parsed.get("reportMonth"),
        "totalExclVat": _to_float(parsed.get("totalExclVat")),
        "totalVat": _to_float(parsed.get("totalVat")),
        "totalInclVat": _to_float(parsed.get("totalInclVat")),
        "lines": _normalize_lines(parsed.get("lines") or []),
        "travelReimbursementLines": parsed.get("travelReimbursementLines") or [],
    }

    if debug:
        out["_debug"] = {
            "pageCount": page_count,
            "rawResponse": raw_response,
        }

    return out


async def _run_parse_job(
    job_id: str, pdf_bytes: bytes, subject: str | None, debug: bool
) -> None:
    """Background runner: parse the PDF and persist the result/error to disk."""
    try:
        result = await _parse_pdf_to_payload(pdf_bytes, subject, debug)
        _write_job(
            job_id,
            {"jobId": job_id, "status": "done", "result": result, "finishedAt": time.time()},
        )
    except HTTPException as e:
        _write_job(
            job_id,
            {"jobId": job_id, "status": "error", "detail": str(e.detail), "finishedAt": time.time()},
        )
    except Exception as e:  # noqa: BLE001 — anything unexpected becomes a stored error
        logger.exception("Legal Aid async parse job %s failed", job_id)
        _write_job(
            job_id,
            {"jobId": job_id, "status": "error", "detail": str(e), "finishedAt": time.time()},
        )


@router.post("/parse-pdf-report")
async def parse_pdf_report(
    request: Request,
    file: UploadFile = File(..., description="Legal Aid monthly payment report PDF"),
    subject: str | None = None,
    mode: str = Query(default="sync", description="'sync' (inline) or 'async' (job + poll)"),
    debug: bool = Query(default=False),
) -> dict[str, Any]:
    """Parse a Legal Aid monthly payment-report PDF via Claude Vision.

    Auth: X-Api-Key.
    Body: multipart with `file` (the PDF bytes).
    Optional `subject` query: if supplied, used to extract `[סימוכין NNN]` →
        sysref. Manual uploads typically omit this.
    Optional `mode=async`: returns `{jobId, status:"pending"}` immediately and
        runs the parse in the background; poll `/parse-pdf-report/jobs/{jobId}`.
        Default `sync` runs inline (the vision call can take several minutes —
        only safe for callers without a short request timeout, e.g. n8n).
    Optional `debug=true`: also returns the raw model response + page count.
    """
    _verify_admin(request)

    pdf_bytes = await file.read()
    # Validate up front so an obviously-bad upload fails fast in BOTH modes
    # (rather than surfacing as a background-job error a moment later).
    if not pdf_bytes:
        raise HTTPException(status_code=400, detail="Empty file")
    if pdf_bytes[:4] != b"%PDF":
        raise HTTPException(status_code=400, detail="File is not a PDF")

    if mode == "async":
        _cleanup_old_jobs()
        job_id = uuid.uuid4().hex
        _write_job(job_id, {"jobId": job_id, "status": "pending", "createdAt": time.time()})
        task = asyncio.create_task(_run_parse_job(job_id, pdf_bytes, subject, debug))
        _running_jobs.add(task)  # keep a strong ref so the task isn't GC'd mid-run
        task.add_done_callback(_running_jobs.discard)
        return {"jobId": job_id, "status": "pending"}

    return await _parse_pdf_to_payload(pdf_bytes, subject, debug)


@router.get("/parse-pdf-report/jobs/{job_id}")
async def get_parse_job(request: Request, job_id: str) -> dict[str, Any]:
    """Poll an async parse job. Returns `{status: pending|done|error, ...}`.

    `done` carries `result` (the ingest payload); `error` carries `detail`.
    """
    _verify_admin(request)
    job = _read_job(job_id)
    if job is None:
        raise HTTPException(status_code=404, detail="Job not found (expired or invalid)")
    return job


# ----------------------------------------------------------------------
# AI Gateway / Claude Vision call
# ----------------------------------------------------------------------


async def _call_vision_for_json(
    images_b64: list[tuple[str, str]], prompt: str
) -> str:
    """Send page images to Claude Vision via ai-gateway and return raw text.

    Mirrors the pattern in api/services/ocr.py — uses the OpenAI-compatible
    chat completions endpoint that ai-gateway exposes at /v1.
    """
    try:
        from openai import AsyncOpenAI
    except ImportError as e:
        raise HTTPException(status_code=503, detail=f"openai sdk missing: {e}")

    base_url = os.environ.get("AI_GATEWAY_URL", "http://localhost:3000").rstrip("/") + "/v1"
    api_key = os.environ.get("AI_GATEWAY_API_KEY", "")
    if not api_key:
        raise HTTPException(
            status_code=503,
            detail="AI_GATEWAY_API_KEY is not configured on shira-hermes",
        )

    client = AsyncOpenAI(base_url=base_url, api_key=api_key)

    content: list[dict[str, Any]] = [{"type": "text", "text": prompt}]
    for mime, b64 in images_b64:
        content.append(
            {
                "type": "image_url",
                "image_url": {"url": f"data:{mime};base64,{b64}"},
            }
        )

    try:
        response = await client.chat.completions.create(
            model=os.environ.get("CLAUDE_MODEL", "sonnet"),
            messages=[{"role": "user", "content": content}],
            max_tokens=4096,
            temperature=0,
        )
    except Exception as e:
        logger.exception("Legal Aid: ai-gateway call failed")
        raise HTTPException(status_code=502, detail=f"ai-gateway error: {e}")

    return (response.choices[0].message.content or "").strip()


# ----------------------------------------------------------------------
# Response parsing + normalization
# ----------------------------------------------------------------------


def _parse_json_response(text: str) -> dict[str, Any]:
    """Extract a JSON object from the model's raw text.

    Handles a few common Claude failure modes:
      - markdown fences (```json ... ```)
      - trailing commentary after the JSON block
      - smart-quotes in keys/values
    """
    if not text:
        return {}

    # Strip code fences first.
    fence = re.match(r"^\s*```(?:json)?\s*\n(.*?)\n?\s*```\s*$", text, re.DOTALL)
    if fence:
        text = fence.group(1)

    # Take the substring from first `{` to matching last `}`.
    start = text.find("{")
    end = text.rfind("}")
    if start == -1 or end == -1 or end <= start:
        logger.warning("Legal Aid: no JSON object found in model response: %r", text[:300])
        return {}

    candidate = text[start : end + 1]

    try:
        return json.loads(candidate)
    except json.JSONDecodeError:
        # Try one fix-up pass: replace smart quotes, then retry.
        cleaned = (
            candidate.replace("“", '"')
            .replace("”", '"')
            .replace("‘", "'")
            .replace("’", "'")
        )
        try:
            return json.loads(cleaned)
        except json.JSONDecodeError as e:
            logger.warning(
                "Legal Aid: model response is not valid JSON: %s; raw=%r",
                e,
                candidate[:300],
            )
            return {}


def _normalize_lines(lines: list[Any]) -> list[dict[str, Any]]:
    out: list[dict[str, Any]] = []
    for raw in lines:
        if not isinstance(raw, dict):
            continue
        # Drop rows that don't carry the minimum identifying info.
        if not (raw.get("invoiceNumber") or raw.get("legalAidCaseNumber")):
            continue
        out.append(
            {
                "invoiceNumber": _str_or_none(raw.get("invoiceNumber")),
                "requestNumber": _str_or_none(raw.get("requestNumber")),
                "legalAidCaseNumber": _str_or_none(raw.get("legalAidCaseNumber")),
                "clientName": _str_or_none(raw.get("clientName")),
                "activityCategory": _str_or_none(raw.get("activityCategory")),
                "activitySubTypeHebrew": _str_or_none(raw.get("activitySubTypeHebrew")),
                "activityDate": _str_or_none(raw.get("activityDate")),
                "amountExclVat": _to_float(raw.get("amountExclVat")),
                "vat": _to_float(raw.get("vat")),
                "amountInclVat": _to_float(raw.get("amountInclVat")),
            }
        )
    return out


# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------


def _to_float(v: Any) -> float:
    if v is None:
        return 0.0
    if isinstance(v, (int, float)):
        return float(v)
    s = str(v).replace(",", "").replace("₪", "").strip()
    try:
        return float(s)
    except ValueError:
        return 0.0


def _str_or_none(v: Any) -> str | None:
    if v is None:
        return None
    s = str(v).strip()
    return s or None