80 lines
4 KiB
Python
80 lines
4 KiB
Python
"""Abgeglichene Ordner: verbinden, Dateiliste, Abgleich aus dem Browser (Liste, Datei, Abschluss), löschen."""
|
|
|
|
import json
|
|
from typing import Annotated
|
|
|
|
from fastapi import APIRouter, Form, Request, UploadFile
|
|
from fastapi.concurrency import run_in_threadpool
|
|
from fastapi.responses import HTMLResponse, JSONResponse, Response
|
|
|
|
from app import web
|
|
from app.core.errors import Invalid
|
|
from app.documents import service as documents
|
|
from app.workspaces import service as workspaces
|
|
|
|
Text = Annotated[str, Form()]
|
|
|
|
|
|
def router() -> APIRouter:
|
|
r = APIRouter()
|
|
|
|
@r.post("/workspaces/{workspace_id}/folders")
|
|
def create(request: Request, principal: web.Checked, workspace_id: int, name: Text, scope: Text = "private",
|
|
exposure_confirmed: Annotated[bool, Form()] = False) -> Response:
|
|
with web.db(request).session() as session:
|
|
created = documents.create(session, principal, workspace_id, name, scope, exposure_confirmed)
|
|
folder_id = created.id
|
|
return web.redirect(f"/folders/{folder_id}")
|
|
|
|
@r.get("/folders/{folder_id}")
|
|
def page(request: Request, principal: web.User, folder_id: int) -> HTMLResponse:
|
|
with web.db(request).session() as session:
|
|
found = documents.folder(session, principal, folder_id)
|
|
membership = workspaces.get(session, principal, found.workspace_id)
|
|
settings = web.settings(request)
|
|
context = {"folder": found, "workspace": membership.workspace, "current_workspace": membership.workspace,
|
|
"membership": membership, "files": documents.files(session, found.id),
|
|
"summary": documents.summaries(session, [found.id])[found.id],
|
|
"owner": found.owner_id == principal.user_id, "scopes": documents.SCOPES,
|
|
"suffixes": ",".join(sorted(documents.TEXT | documents.EXTRACTED)),
|
|
"max_mb": settings.doc_max_mb, "max_files": settings.doc_max_files,
|
|
"extractor": documents.extractor(session) is not None}
|
|
return web.render(request, "doc_folder.html", context)
|
|
|
|
@r.get("/folders/{folder_id}/manifest")
|
|
def manifest(request: Request, principal: web.User, folder_id: int) -> JSONResponse:
|
|
with web.db(request).session() as session:
|
|
return JSONResponse(documents.manifest(session, principal, folder_id))
|
|
|
|
@r.post("/folders/{folder_id}/files")
|
|
async def put(request: Request, principal: web.Checked, folder_id: int, file: UploadFile, path: Text,
|
|
modified: Annotated[int, Form()] = 0) -> JSONResponse:
|
|
def work():
|
|
with web.db(request).session() as session:
|
|
stored = documents.put(session, principal, web.settings(request), folder_id, path, modified,
|
|
file.file)
|
|
return {"path": stored.path, "status": stored.status}
|
|
try:
|
|
return JSONResponse(await run_in_threadpool(work))
|
|
except Invalid as e:
|
|
return JSONResponse({"error": str(e)}, status_code=400)
|
|
|
|
@r.post("/folders/{folder_id}/prune")
|
|
async def prune(request: Request, principal: web.Checked, folder_id: int) -> JSONResponse:
|
|
try:
|
|
paths = json.loads(await request.body() or b"{}").get("paths", [])
|
|
except (ValueError, AttributeError):
|
|
paths = None
|
|
if not isinstance(paths, list) or not all(isinstance(p, str) for p in paths):
|
|
return JSONResponse({"error": "Erwartet: {\"paths\": [...]}"}, status_code=400)
|
|
with web.db(request).session() as session:
|
|
removed = documents.prune(session, principal, web.settings(request), folder_id, paths)
|
|
return JSONResponse({"removed": removed})
|
|
|
|
@r.post("/folders/{folder_id}/delete")
|
|
def delete(request: Request, principal: web.Checked, folder_id: int) -> Response:
|
|
with web.db(request).session() as session:
|
|
workspace_id = documents.delete(session, principal, web.settings(request), folder_id)
|
|
return web.redirect(f"/workspaces/{workspace_id}/data#folders")
|
|
|
|
return r
|