foodlinkk-command-center/docling_routes.py

"""Docling API routes for doc-ingest service."""
from __future__ import annotations

import os
from pathlib import Path
from typing import Any, Optional

from fastapi import APIRouter, HTTPException, Query
from pydantic import BaseModel, Field

from app.docling_service import capabilities, convert_nas_batch, convert_nas_file, get_result, list_history

router = APIRouter(prefix="/docling", tags=["docling"])

NAS_ROOT = Path(os.getenv("NAS_ROOT", "/nas"))


class DoclingConvertBody(BaseModel):
    path: str = Field(..., min_length=1, description="Relatief NAS-pad")
    ocr: bool = True
    tables: bool = True
    page_images: bool = False
    picture_images: bool = True
    force_full_page_ocr: bool = False
    formats: list[str] = Field(default_factory=lambda: ["markdown", "html", "json", "text"])


class DoclingBatchBody(BaseModel):
    paths: list[str] = Field(default_factory=list)
    limit: int = Field(default=10, ge=1, le=50)
    ext: Optional[str] = None
    ocr: bool = True
    tables: bool = True
    page_images: bool = False
    picture_images: bool = True
    force_full_page_ocr: bool = False
    formats: list[str] = Field(default_factory=lambda: ["markdown", "json"])


def _options(body: DoclingConvertBody | DoclingBatchBody) -> dict[str, Any]:
    return {
        "ocr": body.ocr,
        "tables": body.tables,
        "page_images": body.page_images,
        "picture_images": body.picture_images,
        "force_full_page_ocr": body.force_full_page_ocr,
        "formats": body.formats,
    }


@router.get("/capabilities")
def docling_capabilities() -> dict[str, Any]:
    return {"ok": True, **capabilities()}


@router.post("/convert")
def docling_convert(body: DoclingConvertBody) -> dict[str, Any]:
    if not NAS_ROOT.is_dir():
        raise HTTPException(503, f"NAS not mounted at {NAS_ROOT}")
    caps = capabilities()
    if not caps.get("installed"):
        raise HTTPException(503, "Docling not installed on doc-ingest service")
    out = convert_nas_file(NAS_ROOT, body.path.lstrip("/"), _options(body))
    if not out.get("ok"):
        raise HTTPException(502, out.get("error") or "Conversion failed")
    return out


@router.post("/batch")
def docling_batch(body: DoclingBatchBody) -> dict[str, Any]:
    if not NAS_ROOT.is_dir():
        raise HTTPException(503, f"NAS not mounted at {NAS_ROOT}")
    caps = capabilities()
    if not caps.get("installed"):
        raise HTTPException(503, "Docling not installed on doc-ingest service")
    paths = list(body.paths or [])
    if not paths:
        from app.ingest import scan_share_files

        files = scan_share_files(NAS_ROOT)
        if body.ext:
            want = body.ext if body.ext.startswith(".") else f".{body.ext}"
            files = [f for f in files if f.suffix.lower() == want.lower()]
        paths = [str(f.relative_to(NAS_ROOT)) for f in files[: body.limit]]
    if not paths:
        return {"ok": True, "converted": 0, "total": 0, "results": []}
    return convert_nas_batch(NAS_ROOT, paths, _options(body))


@router.get("/history")
def docling_history(limit: int = Query(default=30, ge=1, le=100)) -> dict[str, Any]:
    return {"ok": True, "items": list_history(limit)}


@router.get("/result")
def docling_result(
    path: str = Query(..., min_length=1),
    file_sig: Optional[str] = None,
) -> dict[str, Any]:
    row = get_result(path.lstrip("/"), file_sig)
    if not row:
        raise HTTPException(404, "No Docling result for this file")
    return {"ok": True, "result": row}


class NasWriteBody(BaseModel):
    path: str = Field(..., min_length=1, description="Bronbestand (voor bestandsnaam)")
    content: str = ""
    subdir: str = Field(default="Telegram/Exports", max_length=256)


@router.get("/nas/diagnostics")
def docling_nas_diagnostics() -> dict[str, Any]:
    if not NAS_ROOT.is_dir():
        return {"ok": False, "nas_mounted": False, "error": f"NAS not at {NAS_ROOT}"}
    from app.ingest import scan_share_files, SKIP_DIR_NAMES

    all_files = scan_share_files(NAS_ROOT)
    folder_counts: dict[str, int] = {}
    for p in all_files:
        top = p.relative_to(NAS_ROOT).parts[0] if p.relative_to(NAS_ROOT).parts else "."
        folder_counts[top] = folder_counts.get(top, 0) + 1
    empty_tops = []
    for child in NAS_ROOT.iterdir():
        if not child.is_dir() or child.name.startswith(("#", ".")) or child.name in SKIP_DIR_NAMES:
            continue
        if child.name not in folder_counts:
            empty_tops.append(child.name)
    return {
        "ok": True,
        "nas_mounted": True,
        "total_files": len(all_files),
        "by_top_folder": folder_counts,
        "empty_folder_trees": empty_tops,
        "hint": (
            "Mappen zonder zichtbare bestanden = Synology ACL probleem. "
            "Geef SMB-gebruiker aissa Read/Write op share + submappen in DSM."
        ) if empty_tops else "",
    }


TEXT_SUFFIXES = {
    ".txt", ".md", ".csv", ".json", ".xml", ".html", ".htm", ".yaml", ".yml", ".log", ".rtf",
}
IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp", ".tiff", ".tif"}
PREVIEW_SUFFIXES = TEXT_SUFFIXES | IMAGE_SUFFIXES | {".pdf"}


@router.get("/nas/read")
def docling_nas_read(path: str = Query(..., min_length=1)) -> dict[str, Any]:
    if not NAS_ROOT.is_dir():
        raise HTTPException(503, f"NAS not mounted at {NAS_ROOT}")
    rel = path.lstrip("/")
    full = (NAS_ROOT / rel).resolve()
    if not str(full).startswith(str(NAS_ROOT.resolve())) or not full.is_file():
        raise HTTPException(404, "File not found")
    ext = full.suffix.lower()
    if ext not in TEXT_SUFFIXES and ext not in {".csv", ".rtf"}:
        raise HTTPException(400, f"Not a plain-text file: {ext}")
    try:
        content = full.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        content = full.read_text(encoding="latin-1", errors="replace")
    return {
        "ok": True,
        "path": rel,
        "filename": full.name,
        "extension": ext,
        "content": content,
        "bytes": full.stat().st_size,
        "editable": True,
    }


@router.get("/nas/file")
def docling_nas_file(path: str = Query(..., min_length=1)):
    from fastapi.responses import FileResponse

    if not NAS_ROOT.is_dir():
        raise HTTPException(503, f"NAS not mounted at {NAS_ROOT}")
    rel = path.lstrip("/")
    full = (NAS_ROOT / rel).resolve()
    if not str(full).startswith(str(NAS_ROOT.resolve())) or not full.is_file():
        raise HTTPException(404, "File not found")
    media = "application/octet-stream"
    ext = full.suffix.lower()
    if ext == ".pdf":
        media = "application/pdf"
    elif ext in IMAGE_SUFFIXES:
        media = f"image/{ext.lstrip('.')}"
        if ext == ".jpg":
            media = "image/jpeg"
    elif ext == ".html":
        media = "text/html"
    return FileResponse(full, media_type=media, filename=full.name)


@router.post("/nas/write")
def docling_nas_write(body: NasWriteBody) -> dict[str, Any]:
    if not NAS_ROOT.is_dir():
        raise HTTPException(503, f"NAS not mounted at {NAS_ROOT}")
    subdir = body.subdir.strip().strip("/")
    out_dir = (NAS_ROOT / subdir).resolve()
    if not str(out_dir).startswith(str(NAS_ROOT.resolve())):
        raise HTTPException(400, "Invalid subdir")
    out_dir.mkdir(parents=True, exist_ok=True)
    stem = Path(body.path.lstrip("/")).stem
    fname = f"{stem}-edited.md"
    full = out_dir / fname
    full.write_text(body.content or "", encoding="utf-8")
    rel = str(full.relative_to(NAS_ROOT))
    return {"ok": True, "path": rel, "filename": fname, "bytes": len(body.content or "")}