"""DMS plugin routes — folders, files, preview, Collabora, internal sharing, search, bulk.""" from __future__ import annotations import uuid from fastapi import ( APIRouter, Depends, File, Form, HTTPException, Response, UploadFile, status, ) from fastapi.responses import StreamingResponse from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.db import get_db from app.core.storage import LocalStorage, get_storage_backend from app.core.visibility import apply_visibility_filter, check_single_entity_access from app.deps import get_current_user, require_permission # BUG-018 God-Object-Split: Helper/Konstanten leben jetzt in common.py; # Re-Exports sichern Import- und Patch-Kompatibilitaet # (tests patchen app.plugins.builtins.dms.routes.MAX_FILE_SIZE fuer den Upload). from app.plugins.builtins.dms.common import ( # noqa: F401 ALLOWED_MIME_PREFIXES, BLOCKED_EXTENSIONS, CHUNK_SIZE, MAX_FILE_SIZE, OFFICE_EXTENSIONS, _file_storage_path, _get_file_extension, _is_blocked_filetype, _parse_uuid, _sanitize_filename, chunk_size, ) from app.plugins.builtins.dms.folders_routes import router as folders_router from app.plugins.builtins.dms.models import File as DmsFile from app.plugins.builtins.dms.models import Folder from app.plugins.builtins.dms.schemas import ( FileMetadataResponse, FileUpdate, ) from app.plugins.builtins.dms.search_bulk_routes import router as search_bulk_router from app.plugins.builtins.dms.sharing_routes import router as sharing_router from app.plugins.builtins.permissions.contracts import get_contract as get_perms_contract # Get Permission model from the permissions contract _perms_contract = get_perms_contract() Permission = _perms_contract.Permission router = APIRouter(prefix="/api/v1/dms", tags=["dms"]) # Office file extensions mapped to Collabora file types @router.post("/files/upload", status_code=status.HTTP_201_CREATED, response_model=FileMetadataResponse, dependencies=[Depends(require_permission("dms:write"))]) async def upload_file( file: UploadFile = File(...), folder_id: str | None = Form(None), db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """AC5: POST /api/v1/dms/files/upload → 201, file stored + metadata.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) fid = _parse_uuid(folder_id, "folder_id") if folder_id else None # Rate limit — UPLOAD policy from app.core.rate_limit import RateLimitPolicy, check_rate_limit_policy await check_rate_limit_policy( f"rate:upload:dms:{tenant_id}:{user_id}", RateLimitPolicy.UPLOAD, ) # Check for blocked file types if _is_blocked_filetype(file.filename or ""): raise HTTPException( 400, detail={"detail": "File type not allowed", "code": "blocked_filetype"}, ) # MIME type validation: verify content_type against allowlist mime_type = file.content_type or "application/octet-stream" if not any(mime_type.startswith(prefix) for prefix in ALLOWED_MIME_PREFIXES): raise HTTPException( 400, detail={"detail": f"MIME type '{mime_type}' not allowed", "code": "blocked_mimetype"}, ) # Validate folder exists if specified if fid is not None: folder_result = await db.execute( select(Folder).where( Folder.id == fid, Folder.tenant_id == tenant_id, Folder.deleted_at.is_(None), ) ) if folder_result.scalar_one_or_none() is None: raise HTTPException(404, detail={"detail": "Folder not found", "code": "not_found"}) # Stream file to storage — avoid loading entire file into RAM import hashlib chunk_size = 1024 * 1024 # noqa: F811 (Original-Shadowing im Original auch so) sha256 = hashlib.sha256() file_size = 0 async def chunk_stream(): nonlocal file_size while True: chunk = await file.read(chunk_size) if not chunk: break file_size += len(chunk) if file_size > MAX_FILE_SIZE: raise HTTPException( 413, detail={"detail": "File too large (max 100MB)", "code": "file_too_large"} ) sha256.update(chunk) yield chunk # ── Hook: dms.before_upload (Filter) — can modify filename ── from app.core.hooks import apply_filters upload_data = { "filename": file.filename or "unnamed", "mime_type": file.content_type or "application/octet-stream", } upload_data = await apply_filters("dms.before_upload", upload_data) # Create file record file_id = uuid.uuid4() storage_path = _file_storage_path(tenant_id, file_id) # Save file via storage backend (true streaming, no RAM accumulation) storage = get_storage_backend() await storage.save_stream(storage_path, chunk_stream()) content_hash = sha256.hexdigest() mime_type = upload_data["mime_type"] # Tenant-local deduplication: check for existing file with same content_hash existing_q = await db.execute( select(DmsFile).where( DmsFile.tenant_id == tenant_id, DmsFile.content_hash == content_hash, DmsFile.deleted_at.is_(None), ).limit(1) ) existing_file = existing_q.scalar_one_or_none() if existing_file: # Deduplicate: reuse existing file, remove the duplicate we just saved await storage.delete(storage_path) dms_file = existing_file else: dms_file = DmsFile( id=file_id, tenant_id=tenant_id, name=upload_data["filename"], folder_id=fid, uploaded_by=user_id, mime_type=mime_type, size_bytes=file_size, storage_path=storage_path, content_hash=content_hash, ) db.add(dms_file) await db.flush() # Record history for new file only (D-PLUG) from app.services.entity_history_service import record_history await record_history(db, tenant_id, user_id, "dms_file", dms_file.id, "create", snapshot_after={ "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, }) # Lifecycle hook: dms.after_upload from app.core.hooks import do_action await do_action("dms.after_upload", {'id': str(dms_file.id), 'name': dms_file.name, 'folder_id': str(dms_file.folder_id) if dms_file.folder_id else None, 'mime_type': dms_file.mime_type, 'size_bytes': dms_file.size_bytes}, db=db, tenant_id=tenant_id, user_id=user_id) await do_action("dms_file.after_create", {'id': str(dms_file.id), 'name': dms_file.name, 'tenant_id': str(tenant_id)}, db=db, tenant_id=tenant_id, user_id=user_id) # Outbox event: file.created from app.core.outbox import enqueue_outbox_event await enqueue_outbox_event(db, tenant_id, 'file.created', {'file_id': str(dms_file.id), 'tenant_id': str(tenant_id), 'name': dms_file.name, 'mime_type': dms_file.mime_type, 'size_bytes': dms_file.size_bytes}, aggregate_type='dms_file', aggregate_id=dms_file.id) return { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, "content_hash": dms_file.content_hash, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, } @router.get("/files/{file_id}", dependencies=[Depends(require_permission("dms:read"))]) async def get_file( file_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """AC6: GET /api/v1/dms/files/{id} → 200 + file metadata.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" fid = _parse_uuid(file_id, "file_id") result = await db.execute( select(DmsFile).where( DmsFile.id == fid, DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_(None), ) ) dms_file = result.scalar_one_or_none() if dms_file is None: raise HTTPException(404, detail={"detail": "File not found", "code": "not_found"}) if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "read", is_system_admin): raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"}) return { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, } @router.get("/files", dependencies=[Depends(require_permission("dms:read"))]) async def list_all_files( db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """List all non-deleted files for the current tenant.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" query = select(DmsFile).where( DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_(None), ) query = await apply_visibility_filter( db, query, "dms_file", DmsFile, user_id, tenant_id, is_system_admin ) result = await db.execute(query) files = result.scalars().all() return [ { "id": str(f.id), "name": f.name, "folder_id": str(f.folder_id) if f.folder_id else None, "uploaded_by": str(f.uploaded_by), "mime_type": f.mime_type, "size_bytes": f.size_bytes, "deleted_at": None, "created_at": f.created_at.isoformat() if f.created_at else None, "updated_at": f.updated_at.isoformat() if f.updated_at else None, } for f in files ] @router.get("/folders/{folder_id}/files", dependencies=[Depends(require_permission("dms:read"))]) async def list_files_in_folder( folder_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """List all non-deleted files in a specific folder (non-recursive).""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" fid = _parse_uuid(folder_id, "folder_id") # Validate folder exists folder_result = await db.execute( select(Folder).where( Folder.id == fid, Folder.tenant_id == tenant_id, Folder.deleted_at.is_(None), ) ) if folder_result.scalar_one_or_none() is None: raise HTTPException(404, detail={"detail": "Folder not found", "code": "not_found"}) query = select(DmsFile).where( DmsFile.tenant_id == tenant_id, DmsFile.folder_id == fid, DmsFile.deleted_at.is_(None), ) query = await apply_visibility_filter( db, query, "dms_file", DmsFile, user_id, tenant_id, is_system_admin ) result = await db.execute(query) files = result.scalars().all() return [ { "id": str(f.id), "name": f.name, "folder_id": str(f.folder_id) if f.folder_id else None, "uploaded_by": str(f.uploaded_by), "mime_type": f.mime_type, "size_bytes": f.size_bytes, "deleted_at": None, "created_at": f.created_at.isoformat() if f.created_at else None, "updated_at": f.updated_at.isoformat() if f.updated_at else None, } for f in files ] @router.patch("/files/{file_id}", dependencies=[Depends(require_permission("dms:write"))]) async def update_file( file_id: str, body: FileUpdate, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """AC7: PATCH /api/v1/dms/files/{id} → 200, rename/move.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" fid = _parse_uuid(file_id, "file_id") result = await db.execute( select(DmsFile).where( DmsFile.id == fid, DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_(None), ) ) dms_file = result.scalar_one_or_none() if dms_file is None: raise HTTPException(404, detail={"detail": "File not found", "code": "not_found"}) if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "write", is_system_admin): raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"}) # Lifecycle hook: dms.before_update from app.core.hooks import do_action await do_action("dms.before_update", body.model_dump(exclude_unset=True), db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) data = body.model_dump(exclude_unset=True) # Capture snapshot before update (D-PLUG) snapshot_before = { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, } if "name" in data and data["name"] is not None: dms_file.name = data["name"] if "folder_id" in data: new_folder = _parse_uuid(data["folder_id"], "folder_id") if data["folder_id"] else None if new_folder is not None: folder_result = await db.execute( select(Folder).where( Folder.id == new_folder, Folder.tenant_id == tenant_id, Folder.deleted_at.is_(None), ) ) if folder_result.scalar_one_or_none() is None: raise HTTPException(404, detail={"detail": "Folder not found", "code": "not_found"}) dms_file.folder_id = new_folder await db.flush() await db.refresh(dms_file) snapshot_after = { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, } # Compute changes diff (D-PLUG) changes: dict = {} for key, new_val in snapshot_after.items(): old_val = snapshot_before.get(key) if old_val != new_val: changes[key] = {"old": old_val, "new": new_val} # Record history (D-PLUG) from app.services.entity_history_service import record_history await record_history(db, tenant_id, user_id, "dms_file", dms_file.id, "update", snapshot_before=snapshot_before, snapshot_after=snapshot_after, changes=changes or None) # Lifecycle hook: dms.after_update from app.core.hooks import do_action await do_action("dms.after_update", {'id': str(dms_file.id), 'name': dms_file.name}, db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) await do_action("dms_file.after_update", {'id': str(dms_file.id), 'name': dms_file.name, 'tenant_id': str(tenant_id)}, db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) return { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, } @router.delete("/files/{file_id}", status_code=status.HTTP_204_NO_CONTENT, dependencies=[Depends(require_permission("dms:delete"))]) async def delete_file( file_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """AC8: DELETE /api/v1/dms/files/{id} → 204, soft-delete.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" fid = _parse_uuid(file_id, "file_id") result = await db.execute( select(DmsFile).where( DmsFile.id == fid, DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_(None), ) ) dms_file = result.scalar_one_or_none() if dms_file is None: raise HTTPException(404, detail={"detail": "File not found", "code": "not_found"}) if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "delete", is_system_admin): raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"}) # Lifecycle hook: dms.before_delete from app.core.hooks import do_action await do_action("dms.before_delete", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) # Capture snapshot before delete (D-PLUG) snapshot_before = { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, } from datetime import UTC, datetime dms_file.deleted_at = datetime.now(UTC) await db.flush() # Record history (D-PLUG) from app.services.entity_history_service import record_history await record_history(db, tenant_id, user_id, "dms_file", dms_file.id, "delete", snapshot_before=snapshot_before) # Lifecycle hook: dms.after_delete await do_action("dms.after_delete", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) await do_action("dms_file.after_delete", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) # Outbox event: file.deleted from app.core.outbox import enqueue_outbox_event await enqueue_outbox_event(db, tenant_id, 'file.deleted', {'file_id': str(fid), 'tenant_id': str(tenant_id)}, aggregate_type='dms_file', aggregate_id=fid) return Response(status_code=status.HTTP_204_NO_CONTENT) @router.post("/files/{file_id}/restore", dependencies=[Depends(require_permission("dms:write"))]) async def restore_file( file_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """AC9: POST /api/v1/dms/files/{id}/restore → 200, restored from trash.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" fid = _parse_uuid(file_id, "file_id") result = await db.execute( select(DmsFile).where( DmsFile.id == fid, DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_not(None), ) ) dms_file = result.scalar_one_or_none() if dms_file is None: raise HTTPException(404, detail={"detail": "Deleted file not found", "code": "not_found"}) if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "write", is_system_admin): raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"}) # Lifecycle hook: dms.before_restore from app.core.hooks import do_action await do_action("dms.before_restore", db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) dms_file.deleted_at = None await db.flush() await db.refresh(dms_file) # Lifecycle hook: dms.after_restore await do_action("dms.after_restore", {'id': str(dms_file.id), 'name': dms_file.name}, db=db, tenant_id=tenant_id, user_id=user_id, file_id=file_id) # Outbox event: file.restored from app.core.outbox import enqueue_outbox_event await enqueue_outbox_event(db, tenant_id, 'file.restored', {'file_id': str(fid), 'tenant_id': str(tenant_id)}, aggregate_type='dms_file', aggregate_id=fid) return { "id": str(dms_file.id), "name": dms_file.name, "folder_id": str(dms_file.folder_id) if dms_file.folder_id else None, "uploaded_by": str(dms_file.uploaded_by), "mime_type": dms_file.mime_type, "size_bytes": dms_file.size_bytes, "deleted_at": None, "created_at": dms_file.created_at.isoformat() if dms_file.created_at else None, "updated_at": dms_file.updated_at.isoformat() if dms_file.updated_at else None, } # ─── Preview & Edit ─── @router.get("/files/{file_id}/preview", dependencies=[Depends(require_permission("dms:read"))]) async def preview_file( file_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """AC10: GET /api/v1/dms/files/{id}/preview → 200 + PDF stream.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_system_admin = current_user.get("role") == "admin" fid = _parse_uuid(file_id, "file_id") result = await db.execute( select(DmsFile).where( DmsFile.id == fid, DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_(None), ) ) dms_file = result.scalar_one_or_none() if dms_file is None: raise HTTPException(404, detail={"detail": "File not found", "code": "not_found"}) if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "read", is_system_admin): raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"}) if dms_file.mime_type != "application/pdf": raise HTTPException( 400, detail={"detail": "Only PDF files can be previewed", "code": "not_pdf"} ) storage = get_storage_backend() if not await storage.exists(dms_file.storage_path): raise HTTPException( 404, detail={"detail": "File not found on disk", "code": "file_missing"} ) # Stream file directly from storage without loading into RAM from fastapi.responses import FileResponse as FastApiFileResponse if isinstance(storage, LocalStorage): # LocalStorage: use FileResponse for automatic streaming full_path = storage._full_path(dms_file.storage_path) return FastApiFileResponse( path=full_path, media_type="application/pdf", headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'}, ) else: # S3 or other: fall back to read (TODO: implement S3 streaming) content = await storage.read(dms_file.storage_path) def _stream(): yield content return StreamingResponse( _stream(), media_type="application/pdf", headers={"Content-Disposition": f'inline; filename="{dms_file.name}"'}, ) @router.get("/files/{file_id}/download", dependencies=[Depends(require_permission("dms:read"))]) async def download_file( file_id: str, db: AsyncSession = Depends(get_db), current_user: dict = Depends(get_current_user), ): """Download any file type — streams directly from storage without loading into RAM.""" tenant_id = uuid.UUID(current_user["tenant_id"]) user_id = uuid.UUID(current_user["user_id"]) is_admin = current_user.get("role") == "admin" fid = _parse_uuid(file_id, "file_id") result = await db.execute( select(DmsFile).where( DmsFile.id == fid, DmsFile.tenant_id == tenant_id, DmsFile.deleted_at.is_(None), ) ) dms_file = result.scalar_one_or_none() if dms_file is None: raise HTTPException(404, detail={"detail": "File not found", "code": "not_found"}) if not await check_single_entity_access(db, "dms_file", fid, user_id, tenant_id, "read", is_admin): raise HTTPException(403, detail={"detail": "Access denied", "code": "forbidden"}) storage = get_storage_backend() if not await storage.exists(dms_file.storage_path): raise HTTPException(404, detail={"detail": "File not found on disk", "code": "file_missing"}) from fastapi.responses import FileResponse as FastApiFileResponse if isinstance(storage, LocalStorage): full_path = storage._full_path(dms_file.storage_path) return FastApiFileResponse( path=full_path, media_type=dms_file.mime_type or "application/octet-stream", filename=dms_file.name, ) else: content = await storage.read(dms_file.storage_path) def _stream(): yield content return StreamingResponse( _stream(), media_type=dms_file.mime_type or "application/octet-stream", headers={"Content-Disposition": f'attachment; filename="{dms_file.name}"'}, ) # Sub-Router einbinden (BUG-018 Split): folders, sharing/collabora, search/bulk router.include_router(folders_router) router.include_router(sharing_router) router.include_router(search_bulk_router)