126 lines
4.7 KiB
Python
126 lines
4.7 KiB
Python
import json
|
|
import mimetypes
|
|
import time
|
|
from pathlib import Path
|
|
from urllib import request
|
|
from urllib.parse import urljoin
|
|
|
|
from app.config import settings
|
|
|
|
|
|
class PaperlessClient:
|
|
def __init__(self) -> None:
|
|
self.base_url = settings.paperless_base_url.rstrip('/') + '/'
|
|
self.token = settings.paperless_api_token
|
|
|
|
def _req(self, method: str, path: str, payload: dict | None = None) -> dict:
|
|
url = urljoin(self.base_url, path.lstrip('/'))
|
|
body = None if payload is None else json.dumps(payload).encode('utf-8')
|
|
headers = {
|
|
'Authorization': f'Token {self.token}',
|
|
'Accept': 'application/json',
|
|
}
|
|
if body is not None:
|
|
headers['Content-Type'] = 'application/json'
|
|
req = request.Request(url, data=body, headers=headers, method=method)
|
|
with request.urlopen(req, timeout=120) as resp:
|
|
raw = resp.read().decode('utf-8', errors='replace')
|
|
return json.loads(raw) if raw else {}
|
|
|
|
def get_document(self, document_id: int | str) -> dict:
|
|
return self._req('GET', f'/api/documents/{document_id}/')
|
|
|
|
def update_document(self, document_id: int | str, payload: dict) -> dict:
|
|
return self._req('PATCH', f'/api/documents/{document_id}/', payload)
|
|
|
|
def add_note(self, document_id: int | str, note: str) -> dict:
|
|
return self._req('POST', f'/api/documents/{document_id}/notes/', {'note': note})
|
|
|
|
def _list_endpoint(self, path: str) -> list[dict]:
|
|
data = self._req('GET', path)
|
|
if isinstance(data, list):
|
|
return data
|
|
if isinstance(data, dict):
|
|
results = data.get('results')
|
|
if isinstance(results, list):
|
|
return results
|
|
return []
|
|
|
|
def list_tags(self) -> list[dict]:
|
|
return self._list_endpoint('/api/tags/?page_size=200')
|
|
|
|
def create_tag(self, name: str) -> dict:
|
|
return self._req('POST', '/api/tags/', {'name': name})
|
|
|
|
def get_or_create_tag(self, name: str) -> dict:
|
|
needle = name.strip().casefold()
|
|
for tag in self.list_tags():
|
|
if str(tag.get('name', '')).strip().casefold() == needle:
|
|
return tag
|
|
return self.create_tag(name)
|
|
|
|
def list_document_types(self) -> list[dict]:
|
|
return self._list_endpoint('/api/document_types/?page_size=200')
|
|
|
|
def list_correspondents(self) -> list[dict]:
|
|
return self._list_endpoint('/api/correspondents/?page_size=200')
|
|
|
|
def list_tasks(self, task_id: str | None = None) -> list[dict]:
|
|
path = '/api/tasks/?page_size=100'
|
|
if task_id:
|
|
path += f'&task_id={task_id}'
|
|
return self._list_endpoint(path)
|
|
|
|
def wait_for_task(self, task_id: str, timeout_seconds: int = 90, poll_seconds: float = 1.0) -> dict:
|
|
deadline = time.time() + timeout_seconds
|
|
last = {}
|
|
while time.time() < deadline:
|
|
tasks = self.list_tasks(task_id=task_id)
|
|
if tasks:
|
|
last = tasks[0]
|
|
status = str(last.get('status') or '').upper()
|
|
if status in {'SUCCESS', 'FAILURE', 'CANCELLED'}:
|
|
return last
|
|
time.sleep(poll_seconds)
|
|
return last
|
|
|
|
def upload_document(self, file_path: str, title: str) -> dict:
|
|
path = Path(file_path)
|
|
boundary = "----DorisSchoolhousePaperlessUpload"
|
|
mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
parts: list[bytes] = []
|
|
for key, value in {"title": title}.items():
|
|
parts.extend([
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode(),
|
|
str(value).encode(),
|
|
b"\r\n",
|
|
])
|
|
parts.extend([
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="document"; filename="{path.name}"\r\n'.encode(),
|
|
f"Content-Type: {mime}\r\n\r\n".encode(),
|
|
path.read_bytes(),
|
|
b"\r\n",
|
|
f"--{boundary}--\r\n".encode(),
|
|
])
|
|
body = b"".join(parts)
|
|
req = request.Request(
|
|
urljoin(self.base_url, "/api/documents/post_document/"),
|
|
data=body,
|
|
headers={
|
|
"Authorization": f"Token {self.token}",
|
|
"Accept": "application/json",
|
|
"Content-Type": f"multipart/form-data; boundary={boundary}",
|
|
},
|
|
method="POST",
|
|
)
|
|
with request.urlopen(req, timeout=120) as resp:
|
|
raw = resp.read().decode("utf-8", errors="replace")
|
|
if not raw:
|
|
return {}
|
|
parsed = json.loads(raw)
|
|
if isinstance(parsed, str):
|
|
return {"task_id": parsed}
|
|
return parsed
|