"""DMS commands — file upload, delete, restore via Command pattern.""" from __future__ import annotations import hashlib import logging import os import uuid from typing import Any import redis.asyncio as aioredis from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.commands.base import BaseCommand, CommandResult from app.core.outbox import enqueue_outbox_event from app.core.storage import get_storage_backend logger = logging.getLogger(__name__) CHUNK_SIZE = 1024 * 1024 # 1MB def _sanitize_filename(filename: str) -> str: """Sanitize a filename for safe use in Content-Disposition headers.""" import re safe = os.path.basename(filename.replace("\\", "/")) safe = re.sub(r"[^a-zA-Z0-9.\-_\u00c0-\u017f\u4e00-\u9fff ]", "_", safe) safe = re.sub(r"\.{2,}", "_", safe) safe = re.sub(r" {2,}", " ", safe) safe = safe.lstrip(".").strip() if len(safe) > 200: name, ext = safe.rsplit(".", 1) if "." in safe[:200] else (safe[:200], "") safe = name[:200] + ("." + ext if ext else "") return safe or "file" class UploadFileCommand(BaseCommand): """Upload a file to DMS with chunked streaming and SHA-256 hashing.""" permission = "dms:write" def __init__(self, file_content: bytes, filename: str, mime_type: str, folder_id: str | None = None): self.file_content = file_content self.filename = _sanitize_filename(filename) self.mime_type = mime_type self.folder_id = folder_id async def run(self, db: AsyncSession, redis: aioredis.Redis, current_user: dict[str, Any]) -> CommandResult: from app.plugins.builtins.dms.models import File as DmsFile from app.plugins.builtins.dms.models import Folder tenant_id = self._tenant_id(current_user) user_id = self._user_id(current_user) # Validate folder if specified fid = None if self.folder_id: try: fid = uuid.UUID(self.folder_id) except ValueError: return CommandResult.fail("Invalid folder_id") 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: return CommandResult.fail("Folder not found") # Calculate SHA-256 sha256 = hashlib.sha256() sha256.update(self.file_content) content_hash = sha256.hexdigest() file_size = len(self.file_content) # Create file record file_id = uuid.uuid4() storage_path = f"{tenant_id}/{file_id}" # Save file storage = get_storage_backend() await storage.save(storage_path, self.file_content) dms_file = DmsFile( id=file_id, tenant_id=tenant_id, name=self.filename, folder_id=fid, uploaded_by=user_id, mime_type=self.mime_type, size_bytes=file_size, storage_path=storage_path, content_hash=content_hash, ) db.add(dms_file) await db.flush() # Enqueue outbox event await enqueue_outbox_event(db, tenant_id, "dms.file.uploaded", { "file_id": str(file_id), "name": self.filename, "size_bytes": file_size, "content_hash": content_hash, }) return CommandResult.ok({ "id": str(file_id), "name": self.filename, "size_bytes": file_size, "content_hash": content_hash, "mime_type": self.mime_type, }) async def audit(self, db: AsyncSession, current_user: dict[str, Any]) -> None: from app.core.audit import log_audit await log_audit( db, self._tenant_id(current_user), self._user_id(current_user), action="dms.file.upload", entity_type="dms_file", changes={"name": self.filename, "size": len(self.file_content)}, ) class DeleteFileCommand(BaseCommand): """Soft-delete a DMS file.""" permission = "dms:delete" def __init__(self, file_id: str): self.file_id = file_id async def run(self, db: AsyncSession, redis: aioredis.Redis, current_user: dict[str, Any]) -> CommandResult: from datetime import UTC, datetime from app.plugins.builtins.dms.models import File as DmsFile tenant_id = self._tenant_id(current_user) try: fid = uuid.UUID(self.file_id) except ValueError: return CommandResult.fail("Invalid 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: return CommandResult.fail("File not found") dms_file.deleted_at = datetime.now(UTC) await db.flush() await enqueue_outbox_event(db, tenant_id, "dms.file.deleted", {"file_id": self.file_id}) return CommandResult.ok({"id": self.file_id, "deleted": True}) async def audit(self, db: AsyncSession, current_user: dict[str, Any]) -> None: from app.core.audit import log_audit await log_audit( db, self._tenant_id(current_user), self._user_id(current_user), action="dms.file.delete", entity_type="dms_file", changes={"file_id": self.file_id}, )