shira-hermes/mcp_server/tools/document_tools.py

"""Document tools: list, read, read_multiple, batch_rename, generate_from_template."""

from __future__ import annotations

import json
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from mcp_server.espocrm_client import EspoCrmClient

from mcp_server.tools._helpers import ok, fail


def register_document_tools(
    tools: dict,
    crm: "EspoCrmClient",
    case_id: str | None,
    context: dict,
):
    """Register document-related tools."""

    async def list_documents(refresh: bool = False) -> str:
        docs = context.get("documents", {})
        if refresh and case_id:
            try:
                docs = await crm.post("SmartAssistant/action/listDocuments", {"caseId": case_id})
                context["documents"] = docs
            except Exception as e:
                return fail(f"שגיאה ברענון רשימת מסמכים: {e}")

        # Handle legacy shape where documents is a flat list of files
        if isinstance(docs, list):
            if not docs:
                return ok("לא נמצאו מסמכים בתיק.")
            lines = [f"נמצאו {len(docs)} מסמכים:"]
            for f in docs:
                if isinstance(f, dict):
                    lines.append(f'📄 {f.get("name", "?")} — {f.get("path", "?")}')
                else:
                    lines.append(f'📄 {f}')
            return ok("\n".join(lines))

        if not docs.get("totalFiles"):
            parts = ["לא נמצאו מסמכים בתיק."]
            resolved = docs.get("resolvedPath")
            source = docs.get("pathSource")
            exists = docs.get("folderExists")
            diag = docs.get("diagnostic")
            if resolved:
                parts.append(f"נתיב מחושב: '{resolved}' (מקור: {source or 'unknown'}, קיים: {exists}).")
            else:
                parts.append("לא ניתן לחשב נתיב תיקייה לתיק.")
            if diag:
                parts.append(f"סיבה: {diag}")
            parts.append("הצעה: find_case_folder(query=<שם איש קשר או מספר תיק>) כדי לאתר את התיקייה, ואז set_case_folder_path(path=...) לשמור.")
            return ok(" ".join(parts))
        return ok(json.dumps(docs, ensure_ascii=False))

    async def browse_folder(path: str) -> str:
        try:
            result = await crm.post("SmartAssistant/action/browseFolder", {"path": path})
            entries = result.get("entries", [])
            if not entries:
                return ok(f'תיקייה "{path}" ריקה או לא נמצאה.')
            lines = [f'תוכן "{path}" ({len(entries)} פריטים):']
            for e in entries:
                kind = "📁" if e.get("type") == "folder" else "📄"
                size = f' ({e["size"]} bytes)' if e.get("size") else ""
                lines.append(f'{kind} {e.get("name")} — {e.get("path")}{size}')
            return ok("\n".join(lines))
        except Exception as e:
            return fail(f"שגיאה בסריקת תיקייה: {e}")

    async def find_case_folder(query: str, limit: int = 20) -> str:
        try:
            result = await crm.post("SmartAssistant/action/findCaseFolder", {"query": query, "limit": limit})
            matches = result.get("matches", [])
            if not matches:
                return ok(f'לא נמצאו תיקיות התואמות "{query}" ב-root של האחסון.')
            lines = [f'נמצאו {len(matches)} תיקיות התואמות "{query}":']
            for m in matches:
                lines.append(f'• {m.get("name")} — {m.get("path")} (רמה {m.get("level")}, score={m.get("score")})')
            return ok("\n".join(lines))
        except Exception as e:
            return fail(f"שגיאה בחיפוש תיקייה: {e}")

    async def set_case_folder_path(path: str) -> str:
        if not case_id:
            return fail("צריך להיות בתוך תיק כדי להגדיר נתיב תיקייה.")
        try:
            result = await crm.post("SmartAssistant/action/setCaseFolderPath", {"caseId": case_id, "path": path})
            if not result.get("success"):
                return fail(f"לא הצלחתי לשמור נתיב: {result}")
            context["documents"] = {}
            return ok(f'נתיב התיקייה של התיק עודכן ל-"{path}". הפעל list_documents(refresh=true) כדי לטעון את הרשימה.')
        except Exception as e:
            return fail(f"שגיאה בשמירת נתיב: {e}")

    async def _ocr_fallback(filePath: str, reason: str) -> str:
        """Fetch file bytes and OCR via Claude Vision."""
        from api.services.ocr import extract_text_from_bytes
        try:
            bytes_result = await crm.post("SmartAssistant/action/getDocumentBytes", {"filePath": filePath})
        except Exception as e:
            return fail(f'לא ניתן לחלץ טקסט מ-"{filePath}" ({reason}). כשל ב-getDocumentBytes: {e}')

        if not bytes_result.get("success") or not bytes_result.get("base64"):
            err = bytes_result.get("error", "unknown")
            return fail(f'לא ניתן לחלץ טקסט מ-"{filePath}" ({reason}). בייטי הקובץ: {err}')

        file_name = bytes_result.get("fileName", filePath)
        mime_type = bytes_result.get("mimeType", "")
        try:
            text = await extract_text_from_bytes(
                bytes_result["base64"], mime_type, file_name
            )
            return ok(f'=== {file_name} (OCR via Vision) ===\n\n{text}')
        except Exception as e:
            return fail(f'שגיאה ב-OCR של "{file_name}": {e}')

    async def read_document(filePath: str) -> str:
        try:
            result = await crm.post("SmartAssistant/action/readDocument", {"filePath": filePath})
            if not result.get("success") or not result.get("text"):
                # Fallback to OCR via Claude Vision
                return await _ocr_fallback(filePath, "חילוץ טקסט רגיל נכשל")
            return ok(f'=== {result["fileName"]} ({result.get("charCount", "?")} תווים) ===\n\n{result["text"]}')
        except Exception as e:
            return fail(f"שגיאה בקריאת מסמך: {e}")

    async def read_document_ocr(filePath: str) -> str:
        """Force OCR via Claude Vision — skip regular text extraction."""
        return await _ocr_fallback(filePath, "OCR מבוקש במפורש")

    async def read_multiple_documents(filePaths: list[str], maxCharsPerFile: int = 50000) -> str:
        try:
            result = await crm.post("SmartAssistant/action/readMultipleDocuments", {
                "filePaths": filePaths,
                "maxCharsPerFile": maxCharsPerFile,
            })
            lines = []
            for doc in result.get("documents", []):
                lines.append(f'=== {doc["fileName"]} ({doc.get("charCount", "?")} תווים) ===')
                if doc.get("error"):
                    lines.append(f'שגיאה: {doc["error"]}')
                elif not doc.get("text"):
                    lines.append("לא ניתן לחלץ טקסט")
                else:
                    lines.append(doc["text"])
                lines.append("")
            return ok("\n".join(lines))
        except Exception as e:
            return fail(f"שגיאה בקריאת מסמכים: {e}")

    async def batch_rename_documents(renames: list[dict]) -> str:
        try:
            result = await crm.post("SmartAssistant/action/batchRename", {"renames": renames})
            lines = [f'שינוי שמות: {result.get("successCount", 0)}/{result.get("totalCount", 0)} הצליחו']
            for r in result.get("results", []):
                if r.get("success"):
                    lines.append(f'✅ "{r.get("oldName", "?")}" → "{r.get("newName", "?")}"')
                else:
                    lines.append(f'❌ "{r.get("oldName", "?")}": {r.get("error", "unknown")}')
            return ok("\n".join(lines))
        except Exception as e:
            return fail(f"שגיאה בשינוי שמות: {e}")

    async def generate_from_template(
        templateId: str,
        documentSubject: str = "",
        customPlaceholders: dict | None = None,
    ) -> str:
        if not case_id:
            return fail("צריך להיות בתוך תיק כדי לייצר מסמך מתבנית")
        try:
            result = await crm.post("SmartAssistant/action/generateFromTemplate", {
                "templateId": templateId,
                "caseId": case_id,
                "documentSubject": documentSubject or None,
                "customPlaceholders": customPlaceholders or {},
            })
            return ok(result.get("message", "✅ מסמך נוצר בהצלחה"))
        except Exception as e:
            return fail(f"שגיאה ביצירת מסמך מתבנית: {e}")

    # Register
    tools["list_documents"] = {
        "description": "List documents in the case folder. Shows folder structure with file names and paths. When totalFiles is 0, returns a diagnostic explaining why (missing path, folder doesn't exist, etc.). Pass refresh=true to re-fetch from EspoCRM after the folder path changes.",
        "parameters": {
            "type": "object",
            "properties": {
                "refresh": {"type": "boolean", "description": "Re-query EspoCRM instead of using cached context (default false)"},
            },
        },
        "handler": list_documents,
    }

    tools["browse_folder"] = {
        "description": "List the immediate contents of an arbitrary folder path on network storage. Escape hatch when list_documents returns no results — e.g. browse_folder('30-שוחר תום') to see what's really there.",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Folder path relative to the storage root (e.g. '33-שוחר תום' or 'Documents/2026')"},
            },
            "required": ["path"],
        },
        "handler": browse_folder,
    }

    tools["find_case_folder"] = {
        "description": "Search network-storage for folders whose name contains the query (e.g. contact name or case number). Scans root + one level. Returns ranked matches. Use when list_documents shows the wrong or missing path.",
        "parameters": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Substring to match (Hebrew or English; case-insensitive)."},
                "limit": {"type": "integer", "description": "Max matches to return (default 20, max 50)"},
            },
            "required": ["query"],
        },
        "handler": find_case_folder,
    }

    tools["set_case_folder_path"] = {
        "description": "Persist the given path as the case's networkStorageFolderPath. After this, list_documents(refresh=true) will read from the new path. Use only after find_case_folder or browse_folder confirms the path exists and matches this case.",
        "parameters": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "Storage-relative folder path confirmed to exist."},
            },
            "required": ["path"],
        },
        "handler": set_case_folder_path,
    }

    tools["read_document"] = {
        "description": "Read and extract text content from a document (PDF, DOCX, TXT, images). Requires the file path from list_documents. Automatically falls back to Claude Vision OCR if regular extraction fails (e.g. scanned PDFs, image-only docx).",
        "parameters": {
            "type": "object",
            "properties": {
                "filePath": {"type": "string", "description": "Full path of the document (from list_documents results)"},
            },
            "required": ["filePath"],
        },
        "handler": read_document,
    }

    tools["read_document_ocr"] = {
        "description": "Force OCR extraction via Claude Vision — for images (PNG/JPG/TIFF), scanned PDFs, or any file where you want AI vision to read it. Skips regular text extraction and sends the file straight to Claude Vision.",
        "parameters": {
            "type": "object",
            "properties": {
                "filePath": {"type": "string", "description": "Full path of the document (from list_documents results)"},
            },
            "required": ["filePath"],
        },
        "handler": read_document_ocr,
    }

    tools["read_multiple_documents"] = {
        "description": "Read and extract text from multiple documents at once.",
        "parameters": {
            "type": "object",
            "properties": {
                "filePaths": {"type": "array", "items": {"type": "string"}, "description": "Array of file paths to read"},
                "maxCharsPerFile": {"type": "integer", "description": "Max chars per file (default 50000)"},
            },
            "required": ["filePaths"],
        },
        "handler": read_multiple_documents,
    }

    tools["batch_rename_documents"] = {
        "description": "Rename multiple documents at once.",
        "parameters": {
            "type": "object",
            "properties": {
                "renames": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "sourcePath": {"type": "string", "description": "Current file path"},
                            "newName": {"type": "string", "description": "New file name (without path)"},
                        },
                        "required": ["sourcePath", "newName"],
                    },
                    "description": "Array of rename operations",
                },
            },
            "required": ["renames"],
        },
        "handler": batch_rename_documents,
    }

    tools["generate_from_template"] = {
        "description": "Generate a document from any template. Templates are listed in context under availableTemplates.",
        "parameters": {
            "type": "object",
            "properties": {
                "templateId": {"type": "string", "description": "DocumentTemplate ID (from availableTemplates in context)"},
                "documentSubject": {"type": "string", "description": "Custom document name"},
                "customPlaceholders": {"type": "object", "description": "Extra placeholder values"},
            },
            "required": ["templateId"],
        },
        "handler": generate_from_template,
    }