PDF workflow hub + skills ocean: backend/api/pdf_workflow_routes.py

backend/api/pdf_workflow_routes.py

@@ -0,0 +1,498 @@
+"""
+Annator PDF Workflow Hub — skills ocean + orchestrator + real PDF extract.
+
+Priority surface for HF: make PDF pipelines operable without optional OCR stacks.
+Uses PyPDF2 for text PDFs; reports when scan/OCR is needed.
+"""
+from __future__ import annotations
+
+import io
+import logging
+import re
+import uuid
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, File, Form, HTTPException, UploadFile
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger("pdf_workflow_hub")
+
+router = APIRouter(prefix="/api/pdf", tags=["PDF Workflow"])
+
+try:
+    import PyPDF2
+except ImportError:  # pragma: no cover
+    PyPDF2 = None  # type: ignore
+
+# In-memory job store (HF container; swap to DB/Neon later)
+_JOBS: Dict[str, Dict[str, Any]] = {}
+
+# ---------------------------------------------------------------------------
+# Skill ocean (PDF-domain)
+# ---------------------------------------------------------------------------
+
+PDF_SKILLS: List[Dict[str, Any]] = [
+    {
+        "id": "pdf-text-extract",
+        "name": "PDF Text Extract",
+        "category": "pdf",
+        "status": "ready",
+        "description": "Võtab tekstist loetavast PDF-ist lehekülgede teksti (PyPDF2).",
+        "inputs": ["file"],
+        "outputs": ["text", "pages"],
+    },
+    {
+        "id": "pdf-ocr",
+        "name": "PDF OCR",
+        "category": "pdf",
+        "status": "degraded",
+        "description": "OCR skannitud PDF-idele (vajab tesseract/easyocr/docling; HF peal fallback).",
+        "inputs": ["file"],
+        "outputs": ["text"],
+    },
+    {
+        "id": "bank-statement-reader",
+        "name": "Bank Statement Reader",
+        "category": "finance",
+        "status": "ready",
+        "description": "Heuristiline pangaväljavõtte tehingute tuvastus tekstist.",
+        "inputs": ["text"],
+        "outputs": ["transactions", "summary"],
+    },
+    {
+        "id": "loan-doc-classifier",
+        "name": "Loan Document Classifier",
+        "category": "finance",
+        "status": "ready",
+        "description": "Klassifitseerib laenudokumendi tüübi teksti põhjal.",
+        "inputs": ["text"],
+        "outputs": ["doc_type", "confidence"],
+    },
+    {
+        "id": "pdf-redaction-plan",
+        "name": "PDF Redaction Planner",
+        "category": "pdf",
+        "status": "ready",
+        "description": "Leiab PII/IBAN mustreid ja koostab redaktsiooniplaani (HITL).",
+        "inputs": ["text"],
+        "outputs": ["findings", "plan"],
+    },
+    {
+        "id": "pdf-orchestrator",
+        "name": "PDF Orchestrator",
+        "category": "orchestration",
+        "status": "ready",
+        "description": "Valib skill-ahela eesmärgi põhjal ja juhib töövoo samme.",
+        "inputs": ["goal", "file?"],
+        "outputs": ["plan", "execution"],
+    },
+    {
+        "id": "llm-orchestrator",
+        "name": "LLM Orchestrator",
+        "category": "ai",
+        "status": "ready",
+        "description": "Sidumine Autoflow/LLM plaaniga (plan_only kui mudel puudub).",
+        "inputs": ["goal", "context"],
+        "outputs": ["plan"],
+    },
+]
+
+WORKFLOWS: List[Dict[str, Any]] = [
+    {
+        "id": "wf-pdf-bank-analysis",
+        "name": "PDF + pangaväljavõtte analüüs",
+        "category": "pdf",
+        "status": "ready",
+        "steps": ["pdf-text-extract", "bank-statement-reader", "loan-doc-classifier", "pdf-orchestrator"],
+    },
+    {
+        "id": "wf-pdf-loan-pack",
+        "name": "Laenupaketi PDF pipeline",
+        "category": "pdf",
+        "status": "ready",
+        "steps": ["pdf-text-extract", "loan-doc-classifier", "pdf-redaction-plan", "pdf-orchestrator"],
+    },
+    {
+        "id": "wf-pdf-ocr-fallback",
+        "name": "OCR fallback skannidele",
+        "category": "pdf",
+        "status": "degraded",
+        "steps": ["pdf-text-extract", "pdf-ocr", "bank-statement-reader"],
+    },
+]
+
+
+class PlanRequest(BaseModel):
+    goal: str = Field(..., min_length=1)
+    domain: str = "pdf"
+    mode: str = "plan_only"
+
+
+class SkillExecuteRequest(BaseModel):
+    skill_id: str
+    text: Optional[str] = None
+    goal: Optional[str] = None
+    params: Dict[str, Any] = Field(default_factory=dict)
+
+
+class WorkflowRunRequest(BaseModel):
+    workflow_id: str = "wf-pdf-bank-analysis"
+    goal: str = "PDF analüüs"
+    text: Optional[str] = None  # if no file uploaded via multipart
+
+
+# ---------------------------------------------------------------------------
+# Core extractors
+# ---------------------------------------------------------------------------
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def extract_pdf_text(data: bytes) -> Dict[str, Any]:
+    if not PyPDF2:
+        raise HTTPException(status_code=503, detail="PyPDF2 not installed")
+    reader = PyPDF2.PdfReader(io.BytesIO(data))
+    pages: List[Dict[str, Any]] = []
+    chunks: List[str] = []
+    for i, page in enumerate(reader.pages):
+        try:
+            t = page.extract_text() or ""
+        except Exception as exc:  # noqa: BLE001
+            t = ""
+            logger.warning("page %s extract failed: %s", i + 1, exc)
+        pages.append({"page": i + 1, "text": t, "char_count": len(t)})
+        chunks.append(t)
+    full = "\n".join(chunks)
+    page_count = len(pages)
+    total = len(full)
+    text_ratio = min(total / max(page_count * 800, 1), 1.0)
+    return {
+        "method": "pypdf2",
+        "success": True,
+        "page_count": page_count,
+        "total_chars": total,
+        "text_ratio": text_ratio,
+        "needs_ocr": text_ratio < 0.08 or total < 80,
+        "extracted_text": full,
+        "pages": pages,
+    }
+
+
+_IBAN_RE = re.compile(r"\b[A-Z]{2}\d{2}[A-Z0-9]{10,30}\b")
+_AMOUNT_RE = re.compile(
+    r"(?<!\w)(-?\d{1,3}(?:[ \u00a0]\d{3})*(?:[.,]\d{2})|-?\d+[.,]\d{2})(?!\w)"
+)
+_DATE_RE = re.compile(r"\b(\d{1,2}[./-]\d{1,2}[./-]\d{2,4}|\d{4}-\d{2}-\d{2})\b")
+_EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
+_PERSONAL_CODE_RE = re.compile(r"\b[1-6]\d{10}\b")  # EE-like isikukood shape
+
+
+def parse_bank_statement(text: str) -> Dict[str, Any]:
+    lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
+    txs: List[Dict[str, Any]] = []
+    for ln in lines:
+        amounts = _AMOUNT_RE.findall(ln)
+        dates = _DATE_RE.findall(ln)
+        if amounts and (dates or any(k in ln.lower() for k in ("makse", "laekumine", "tasu", "transfer", "payment"))):
+            txs.append(
+                {
+                    "date": dates[0] if dates else None,
+                    "amount_raw": amounts[-1],
+                    "description": ln[:240],
+                }
+            )
+    ibans = list(dict.fromkeys(_IBAN_RE.findall(text)))
+    return {
+        "transaction_count": len(txs),
+        "transactions": txs[:200],
+        "ibans_found": ibans[:20],
+        "line_count": len(lines),
+        "summary": {
+            "approx_credit_lines": sum(1 for t in txs if not str(t["amount_raw"]).startswith("-")),
+            "approx_debit_lines": sum(1 for t in txs if str(t["amount_raw"]).startswith("-")),
+        },
+    }
+
+
+def classify_loan_doc(text: str) -> Dict[str, Any]:
+    t = text.lower()
+    scores = {
+        "bank_statement": sum(1 for k in ("konto", "väljavõte", "valjavote", "saldo", "iban", "tehing") if k in t),
+        "loan_application": sum(1 for k in ("laen", "taotlus", "sissetulek", "kohustus", "tagatis") if k in t),
+        "income_proof": sum(1 for k in ("palk", "tööleping", "tooleping", "tulumaks") if k in t),
+        "id_document": sum(1 for k in ("isikutunnistus", "pass", "id-kaart", "id kaart") if k in t),
+        "contract": sum(1 for k in ("leping", "tingimused", "allkiri", "pool") if k in t),
+        "generic_pdf": 1,
+    }
+    doc_type = max(scores, key=scores.get)
+    conf = min(0.95, 0.35 + 0.12 * scores[doc_type])
+    return {"doc_type": doc_type, "confidence": round(conf, 2), "scores": scores}
+
+
+def redaction_plan(text: str) -> Dict[str, Any]:
+    findings = []
+    for m in _IBAN_RE.finditer(text):
+        findings.append({"type": "iban", "value": m.group(0), "start": m.start(), "end": m.end()})
+    for m in _EMAIL_RE.finditer(text):
+        findings.append({"type": "email", "value": m.group(0), "start": m.start(), "end": m.end()})
+    for m in _PERSONAL_CODE_RE.finditer(text):
+        findings.append({"type": "personal_code_like", "value": m.group(0), "start": m.start(), "end": m.end()})
+    return {
+        "finding_count": len(findings),
+        "findings": findings[:100],
+        "plan": [
+            "HITL: kinnita enne redaktsiooni",
+            "Maski IBAN/email/isikukood PDF kihis (mitte ainult pealmine tekst)",
+            "Säilita auditilogi ilma plain-text PII-ta",
+        ],
+        "requires_approval": True,
+    }
+
+
+def orchestrate(goal: str, text: Optional[str] = None, extract: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+    g = goal.lower()
+    steps: List[str] = ["pdf-text-extract"]
+    if any(k in g for k in ("ocr", "skann", "scan")):
+        steps.append("pdf-ocr")
+    if any(k in g for k in ("pank", "väljavõte", "valjavote", "bank", "statement", "iban")):
+        steps.append("bank-statement-reader")
+    if any(k in g for k in ("laen", "loan", "taotlus")):
+        steps.append("loan-doc-classifier")
+    if any(k in g for k in ("redact", "pii", "isiku", "peida", "mask")):
+        steps.append("pdf-redaction-plan")
+    if "pdf-orchestrator" not in steps:
+        steps.append("pdf-orchestrator")
+
+    results: Dict[str, Any] = {}
+    body = text or (extract or {}).get("extracted_text") or ""
+    if body:
+        if "bank-statement-reader" in steps:
+            results["bank-statement-reader"] = parse_bank_statement(body)
+        if "loan-doc-classifier" in steps:
+            results["loan-doc-classifier"] = classify_loan_doc(body)
+        if "pdf-redaction-plan" in steps:
+            results["pdf-redaction-plan"] = redaction_plan(body)
+
+    return {
+        "goal": goal,
+        "selected_skills": steps,
+        "extract_meta": {
+            "page_count": (extract or {}).get("page_count"),
+            "total_chars": (extract or {}).get("total_chars"),
+            "needs_ocr": (extract or {}).get("needs_ocr"),
+            "method": (extract or {}).get("method"),
+        },
+        "skill_results": results,
+        "status": "completed" if body else "planned",
+        "warnings": (
+            ["PDF sisaldab vähe teksti — soovitatav OCR"]
+            if (extract or {}).get("needs_ocr")
+            else []
+        ),
+    }
+
+
+# ---------------------------------------------------------------------------
+# Routes
+# ---------------------------------------------------------------------------
+
+@router.get("/status")
+async def pdf_status():
+    return {
+        "status": "ok",
+        "service": "pdf-workflow-hub",
+        "pypdf2": bool(PyPDF2),
+        "skills": len(PDF_SKILLS),
+        "workflows": len(WORKFLOWS),
+        "jobs_in_memory": len(_JOBS),
+        "endpoints": {
+            "skills": "/api/pdf/skills",
+            "workflows": "/api/pdf/workflows",
+            "ocr": "POST /api/pdf/ocr",
+            "jobs": "POST /api/pdf/jobs",
+            "run": "POST /api/pdf/workflow/run",
+            "plan": "POST /api/pdf/plan",
+        },
+    }
+
+
+@router.get("/skills")
+async def list_pdf_skills():
+    return {"ok": True, "count": len(PDF_SKILLS), "skills": PDF_SKILLS}
+
+
+@router.get("/workflows")
+async def list_pdf_workflows():
+    return {"ok": True, "count": len(WORKFLOWS), "workflows": WORKFLOWS}
+
+
+@router.post("/plan")
+async def plan_pdf_workflow(req: PlanRequest):
+    plan = orchestrate(req.goal)
+    return {
+        "ok": True,
+        "mode": req.mode,
+        "domain": req.domain,
+        "selected_adapter": "pdf-orchestrator",
+        "plan": plan["selected_skills"],
+        "result": plan,
+    }
+
+
+@router.post("/ocr")
+async def pdf_ocr_upload(file: UploadFile = File(...)):
+    if not file.filename or not file.filename.lower().endswith(".pdf"):
+        raise HTTPException(status_code=400, detail="Only .pdf supported")
+    data = await file.read()
+    if not data:
+        raise HTTPException(status_code=400, detail="Empty file")
+    result = extract_pdf_text(data)
+    job_id = str(uuid.uuid4())
+    _JOBS[job_id] = {
+        "id": job_id,
+        "type": "ocr",
+        "filename": file.filename,
+        "created_at": _now(),
+        "status": "completed",
+        "result": result,
+    }
+    return {"ok": True, "job_id": job_id, **result}
+
+
+@router.post("/jobs")
+async def create_pdf_job(
+    file: UploadFile = File(...),
+    goal: str = Form("PDF analüüs"),
+    workflow_id: str = Form("wf-pdf-bank-analysis"),
+):
+    if not file.filename or not file.filename.lower().endswith(".pdf"):
+        raise HTTPException(status_code=400, detail="Only .pdf supported")
+    data = await file.read()
+    if not data:
+        raise HTTPException(status_code=400, detail="Empty file")
+
+    extract = extract_pdf_text(data)
+    orch = orchestrate(goal, extract=extract)
+    job_id = str(uuid.uuid4())
+    record = {
+        "id": job_id,
+        "type": "workflow",
+        "workflow_id": workflow_id,
+        "goal": goal,
+        "filename": file.filename,
+        "created_at": _now(),
+        "status": orch["status"],
+        "extract": {
+            k: extract[k]
+            for k in ("method", "page_count", "total_chars", "text_ratio", "needs_ocr", "success")
+        },
+        "orchestration": orch,
+        # keep text for follow-up skill calls (not logged externally)
+        "_text": extract.get("extracted_text", "")[:500_000],
+    }
+    _JOBS[job_id] = record
+    public = {k: v for k, v in record.items() if not k.startswith("_")}
+    public["text_preview"] = (extract.get("extracted_text") or "")[:1500]
+    return {"ok": True, "job": public}
+
+
+@router.get("/jobs/{job_id}")
+async def get_pdf_job(job_id: str):
+    job = _JOBS.get(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    public = {k: v for k, v in job.items() if not k.startswith("_")}
+    return {"ok": True, "job": public}
+
+
+@router.post("/skills/execute")
+async def execute_skill(req: SkillExecuteRequest):
+    skill = next((s for s in PDF_SKILLS if s["id"] == req.skill_id), None)
+    if not skill:
+        raise HTTPException(status_code=404, detail=f"Unknown skill: {req.skill_id}")
+
+    text = req.text or ""
+    if req.skill_id == "pdf-text-extract":
+        return {"ok": True, "skill": req.skill_id, "result": {"note": "Upload via POST /api/pdf/ocr"}}
+    if req.skill_id == "bank-statement-reader":
+        if not text:
+            raise HTTPException(status_code=400, detail="text required")
+        return {"ok": True, "skill": req.skill_id, "result": parse_bank_statement(text)}
+    if req.skill_id == "loan-doc-classifier":
+        if not text:
+            raise HTTPException(status_code=400, detail="text required")
+        return {"ok": True, "skill": req.skill_id, "result": classify_loan_doc(text)}
+    if req.skill_id == "pdf-redaction-plan":
+        if not text:
+            raise HTTPException(status_code=400, detail="text required")
+        return {"ok": True, "skill": req.skill_id, "result": redaction_plan(text)}
+    if req.skill_id in ("pdf-orchestrator", "llm-orchestrator"):
+        return {
+            "ok": True,
+            "skill": req.skill_id,
+            "result": orchestrate(req.goal or "PDF workflow", text=text),
+        }
+    if req.skill_id == "pdf-ocr":
+        return {
+            "ok": True,
+            "skill": req.skill_id,
+            "result": {
+                "status": "degraded",
+                "message": "Full OCR stack not installed on HF; use text PDF or attach tesseract/docling.",
+            },
+        }
+    raise HTTPException(status_code=400, detail="Skill not executable")
+
+
+@router.post("/workflow/run")
+async def run_workflow_multipart(
+    file: Optional[UploadFile] = File(None),
+    workflow_id: str = Form("wf-pdf-bank-analysis"),
+    goal: str = Form("PDF + pangaväljavõtte analüüs"),
+):
+    """End-to-end: upload PDF → extract → skill chain → orchestration result."""
+    extract = None
+    text = ""
+    filename = None
+    if file and file.filename:
+        if not file.filename.lower().endswith(".pdf"):
+            raise HTTPException(status_code=400, detail="Only .pdf supported")
+        data = await file.read()
+        extract = extract_pdf_text(data)
+        text = extract.get("extracted_text") or ""
+        filename = file.filename
+
+    wf = next((w for w in WORKFLOWS if w["id"] == workflow_id), None)
+    if not wf:
+        raise HTTPException(status_code=404, detail="Unknown workflow")
+
+    orch = orchestrate(goal, text=text, extract=extract)
+    job_id = str(uuid.uuid4())
+    record = {
+        "id": job_id,
+        "type": "workflow_run",
+        "workflow": wf,
+        "goal": goal,
+        "filename": filename,
+        "created_at": _now(),
+        "status": orch["status"],
+        "orchestration": orch,
+        "_text": text[:500_000],
+    }
+    _JOBS[job_id] = record
+    return {
+        "ok": True,
+        "job_id": job_id,
+        "workflow_id": workflow_id,
+        "status": orch["status"],
+        "selected_skills": orch["selected_skills"],
+        "skill_results": orch["skill_results"],
+        "extract_meta": orch["extract_meta"],
+        "warnings": orch["warnings"],
+        "text_preview": text[:1500],
+    }
+
+
+__all__ = ["router"]