141 lines
4.4 KiB
Python
141 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Paperless-ngx REST API client — supporto token per utente MCP."""
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
LOGGER = logging.getLogger("loogle_mcp.paperless")
|
|
|
|
PAPERLESS_URL = os.environ.get("PAPERLESS_URL", "https://docs.loogle.it").rstrip("/")
|
|
MCP_USERS = ("daniele", "lucia", "davide", "luca")
|
|
_tokens_cache: Optional[dict[str, str]] = None
|
|
|
|
|
|
def _load_tokens() -> dict[str, str]:
|
|
"""Carica token Paperless per utente MCP.
|
|
|
|
Priorità per ogni utente:
|
|
1. PAPERLESS_API_TOKEN_{USERNAME} (es. PAPERLESS_API_TOKEN_LUCIA)
|
|
2. Chiavi in PAPERLESS_API_TOKENS JSON (es. {"daniele":"...", "lucia":"..."})
|
|
3. PAPERLESS_API_TOKEN globale (fallback per tutti, tipico account admin)
|
|
"""
|
|
global _tokens_cache
|
|
if _tokens_cache is not None:
|
|
return _tokens_cache
|
|
|
|
tokens: dict[str, str] = {}
|
|
json_map = os.environ.get("PAPERLESS_API_TOKENS", "").strip()
|
|
if json_map:
|
|
try:
|
|
parsed = json.loads(json_map)
|
|
if isinstance(parsed, dict):
|
|
tokens.update({k.lower(): v for k, v in parsed.items() if v})
|
|
except json.JSONDecodeError:
|
|
LOGGER.warning("PAPERLESS_API_TOKENS non è JSON valido")
|
|
|
|
fallback = os.environ.get("PAPERLESS_API_TOKEN", "").strip()
|
|
for user in MCP_USERS:
|
|
env_key = f"PAPERLESS_API_TOKEN_{user.upper()}"
|
|
token = os.environ.get(env_key, "").strip()
|
|
if token:
|
|
tokens[user] = token
|
|
elif user not in tokens and fallback:
|
|
tokens[user] = fallback
|
|
|
|
if not tokens and fallback:
|
|
tokens["daniele"] = fallback
|
|
|
|
_tokens_cache = tokens
|
|
return tokens
|
|
|
|
|
|
def list_configured_users() -> list[str]:
|
|
return list(_load_tokens().keys())
|
|
|
|
|
|
def _headers(username: Optional[str] = None) -> dict:
|
|
tokens = _load_tokens()
|
|
if not tokens:
|
|
raise RuntimeError(
|
|
"Nessun token Paperless configurato. "
|
|
"Imposta PAPERLESS_API_TOKEN o PAPERLESS_API_TOKEN_{USER} in .env"
|
|
)
|
|
user = (username or "daniele").lower()
|
|
token = tokens.get(user) or tokens.get("daniele") or next(iter(tokens.values()))
|
|
return {"Authorization": f"Token {token}"}
|
|
|
|
|
|
def list_documents(
|
|
page: int = 1,
|
|
page_size: int = 25,
|
|
ordering: str = "-modified",
|
|
username: Optional[str] = None,
|
|
) -> dict:
|
|
with httpx.Client(timeout=60.0, verify=True) as client:
|
|
resp = client.get(
|
|
f"{PAPERLESS_URL}/api/documents/",
|
|
headers=_headers(username),
|
|
params={"page": page, "page_size": page_size, "ordering": ordering},
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def get_document(doc_id: int, username: Optional[str] = None) -> dict:
|
|
with httpx.Client(timeout=60.0, verify=True) as client:
|
|
resp = client.get(
|
|
f"{PAPERLESS_URL}/api/documents/{doc_id}/",
|
|
headers=_headers(username),
|
|
)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
|
|
def download_document_text(doc_id: int, username: Optional[str] = None) -> str:
|
|
doc = get_document(doc_id, username=username)
|
|
content = (doc.get("content") or "").strip()
|
|
if content:
|
|
return content
|
|
title = doc.get("title") or f"Documento {doc_id}"
|
|
return f"# {title}\n\n(Nessun testo OCR disponibile)"
|
|
|
|
|
|
def document_visibility(doc: dict, owner_username: str) -> str:
|
|
tags = doc.get("tags") or []
|
|
tag_names = []
|
|
for t in tags:
|
|
if isinstance(t, dict):
|
|
tag_names.append((t.get("name") or "").lower())
|
|
elif isinstance(t, int):
|
|
continue
|
|
else:
|
|
tag_names.append(str(t).lower())
|
|
if any(t in ("personal", "privato", "private") for t in tag_names):
|
|
return "personal"
|
|
if any(t in ("admin-only", "admin") for t in tag_names):
|
|
return "admin"
|
|
return "family"
|
|
|
|
|
|
def document_owner(doc: dict, default: str = "daniele") -> str:
|
|
owner = doc.get("owner")
|
|
if isinstance(owner, int):
|
|
pass
|
|
owner_username = doc.get("owner_username") or doc.get("owner_name")
|
|
if isinstance(owner_username, str):
|
|
name = owner_username.lower()
|
|
for user in MCP_USERS:
|
|
if user in name:
|
|
return user
|
|
correspondent = doc.get("correspondent")
|
|
if isinstance(correspondent, dict):
|
|
name = (correspondent.get("name") or "").lower()
|
|
for user in MCP_USERS:
|
|
if user in name:
|
|
return user
|
|
return default
|