88 lines
3.0 KiB
Python
88 lines
3.0 KiB
Python
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from app.text_extract import html_to_text, markdown_to_text
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class DocumentNotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
def _guess_content_type(key: str) -> str:
|
|
if key.endswith(".md"):
|
|
return "text/markdown"
|
|
if key.endswith(".html") or key.endswith(".htm"):
|
|
return "text/html"
|
|
return "application/octet-stream"
|
|
|
|
|
|
class DocumentService:
|
|
def __init__(self, minio_client, search_client):
|
|
self._minio_client = minio_client
|
|
self._search_client = search_client
|
|
|
|
def list_documents(self, path: str) -> dict:
|
|
prefix = path if not path or path.endswith("/") else f"{path}/"
|
|
entries = self._minio_client.list_objects(prefix)
|
|
folders = []
|
|
files = []
|
|
for entry in entries:
|
|
key = entry["key"]
|
|
if entry.get("is_dir"):
|
|
name = key[len(prefix):].rstrip("/")
|
|
if name:
|
|
folders.append(name)
|
|
else:
|
|
name = key[len(prefix):]
|
|
files.append({
|
|
"key": key,
|
|
"name": name,
|
|
"size": entry["size"],
|
|
"modifiedAt": entry["modifiedAt"],
|
|
})
|
|
return {"folders": folders, "files": files}
|
|
|
|
def get_document(self, key: str) -> dict:
|
|
try:
|
|
data = self._minio_client.get_object(key)
|
|
except FileNotFoundError as exc:
|
|
raise DocumentNotFoundError(key) from exc
|
|
return {
|
|
"key": key,
|
|
"content": data.decode("utf-8"),
|
|
"contentType": _guess_content_type(key),
|
|
}
|
|
|
|
def _extract_text(self, content: bytes, content_type: str) -> str:
|
|
text = content.decode("utf-8", errors="ignore")
|
|
if content_type == "text/html":
|
|
return html_to_text(text)
|
|
if content_type == "text/markdown":
|
|
return markdown_to_text(text)
|
|
return text
|
|
|
|
def save_document(self, key: str, content: bytes, content_type: str) -> dict:
|
|
self._minio_client.put_object(key, content, content_type)
|
|
try:
|
|
text = self._extract_text(content, content_type)
|
|
title = key.rsplit("/", 1)[-1]
|
|
path = key.rsplit("/", 1)[0] if "/" in key else ""
|
|
self._search_client.index_document(key, path, title, text)
|
|
except Exception:
|
|
logger.exception("Failed to update search index for %s", key)
|
|
return {"key": key, "updatedAt": datetime.now(timezone.utc).isoformat()}
|
|
|
|
def delete_document(self, key: str) -> None:
|
|
self._minio_client.delete_object(key)
|
|
try:
|
|
self._search_client.delete_document(key)
|
|
except Exception:
|
|
logger.exception("Failed to delete search index entry for %s", key)
|
|
|
|
def upload_document(self, key: str, data: bytes, content_type: str) -> dict:
|
|
self.save_document(key, data, content_type)
|
|
name = key.rsplit("/", 1)[-1]
|
|
return {"key": key, "name": name, "size": len(data)}
|