|
# retoor <retoor@molodetz.nl>
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Annotated
|
|
from fastapi import Depends, APIRouter, Request
|
|
from fastapi.responses import JSONResponse
|
|
from devplacepy.database import get_table, get_int_setting
|
|
from devplacepy.models import UploadUrlForm
|
|
from devplacepy.utils import require_user_api, is_admin, track_action
|
|
from devplacepy.attachments import (
|
|
store_attachment,
|
|
store_attachment_from_url,
|
|
soft_delete_attachment,
|
|
is_extension_allowed,
|
|
RemoteFetchError,
|
|
)
|
|
from urllib.parse import urlparse
|
|
from devplacepy.services.audit import record as audit
|
|
from devplacepy.dependencies import json_or_form
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
@router.post("/upload")
|
|
async def upload_file(request: Request):
|
|
user = require_user_api(request)
|
|
form = await request.form()
|
|
file = form.get("file")
|
|
|
|
if not file or not hasattr(file, "filename") or not file.filename:
|
|
return JSONResponse({"error": "No file provided"}, status_code=400)
|
|
|
|
ext = Path(file.filename).suffix.lower()
|
|
if not is_extension_allowed(ext):
|
|
return JSONResponse(
|
|
{"error": f"File type '{ext}' not allowed"}, status_code=415
|
|
)
|
|
|
|
try:
|
|
content = await file.read()
|
|
except Exception as e:
|
|
logger.warning(f"Failed to read uploaded file: {e}")
|
|
return JSONResponse({"error": "Failed to read file"}, status_code=400)
|
|
|
|
result = store_attachment(content, file.filename, user["uid"])
|
|
if result is None:
|
|
max_size_mb = get_int_setting("max_upload_size_mb", 10)
|
|
return JSONResponse(
|
|
{"error": f"File exceeds {max_size_mb}MB limit"}, status_code=413
|
|
)
|
|
|
|
logger.info(
|
|
f"File uploaded: {file.filename} ({len(content)} bytes) -> {result['url']}"
|
|
)
|
|
audit.record(
|
|
request,
|
|
"attachment.upload",
|
|
user=user,
|
|
target_type="attachment",
|
|
target_uid=result.get("uid"),
|
|
target_label=file.filename,
|
|
metadata={"size": len(content), "mime": result.get("mime_type")},
|
|
summary=f"{user['username']} uploaded attachment {file.filename}",
|
|
links=[audit.attachment_link(result.get("uid"), file.filename)],
|
|
)
|
|
track_action(user["uid"], "upload")
|
|
return JSONResponse(result, status_code=201)
|
|
|
|
@router.post("/upload-url")
|
|
async def upload_from_url(request: Request, data: Annotated[UploadUrlForm, Depends(json_or_form(UploadUrlForm))]):
|
|
user = require_user_api(request)
|
|
try:
|
|
result = await store_attachment_from_url(data.url, user["uid"], data.filename)
|
|
except RemoteFetchError as exc:
|
|
return JSONResponse({"error": exc.message}, status_code=exc.status)
|
|
|
|
logger.info(
|
|
f"URL attached: {data.url} ({result['file_size']} bytes) -> {result['url']}"
|
|
)
|
|
audit.record(
|
|
request,
|
|
"attachment.upload_url",
|
|
user=user,
|
|
target_type="attachment",
|
|
target_uid=result.get("uid"),
|
|
target_label=result.get("filename"),
|
|
metadata={"source_host": urlparse(data.url).hostname, "size": result.get("file_size"), "mime": result.get("mime_type")},
|
|
summary=f"{user['username']} attached a remote file from {urlparse(data.url).hostname}",
|
|
links=[audit.attachment_link(result.get("uid"), result.get("filename"))],
|
|
)
|
|
return JSONResponse(result, status_code=201)
|
|
|
|
@router.delete("/delete/{attachment_uid}")
|
|
async def delete_attachment_route(request: Request, attachment_uid: str):
|
|
user = require_user_api(request)
|
|
att = get_table("attachments").find_one(uid=attachment_uid, deleted_at=None)
|
|
if not att:
|
|
return JSONResponse({"error": "Attachment not found"}, status_code=404)
|
|
if att.get("user_uid") and att["user_uid"] != user["uid"] and not is_admin(user):
|
|
return JSONResponse({"error": "Not authorized"}, status_code=403)
|
|
soft_delete_attachment(attachment_uid, deleted_by=user["uid"])
|
|
logger.info(f"Attachment {attachment_uid} soft-deleted by {user['username']}")
|
|
audit.record(
|
|
request,
|
|
"attachment.delete",
|
|
user=user,
|
|
target_type="attachment",
|
|
target_uid=attachment_uid,
|
|
target_label=att.get("filename"),
|
|
summary=f"{user['username']} deleted attachment {att.get('filename') or attachment_uid}",
|
|
links=[audit.attachment_link(attachment_uid, att.get("filename"))],
|
|
)
|
|
return JSONResponse({"status": "deleted"})
|