"""Database-backed global upload-policy API. ``GET /api/upload-policy`` is available to authenticated users so every upload surface can render the same effective limits. ``PUT`` replaces the document atomically and is restricted to administrators. Before the first save the built-in defaults are returned without requiring a seed row. """ from __future__ import annotations from copy import deepcopy from datetime import datetime from typing import Literal from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel, Field, field_validator, model_validator from app.gateway.auth.disabled_mode import is_auth_disabled from app.gateway.deps import get_optional_user_from_request from deerflow.persistence.upload_policy import UploadPolicyStore from deerflow.uploads.policy import MIB, default_upload_policy, normalize_extension, set_runtime_upload_policy router = APIRouter(prefix="/api/upload-policy", tags=["upload-policy"]) MAX_CONFIGURED_FILES = 100 MAX_CONFIGURED_TOTAL_BYTES = 200 * MIB MAX_CONFIGURED_FILE_BYTES = 100 * MIB MAX_MODEL_RESULT_CHARS = 32_000 _SUPPORTED_BY_GROUP = { "document": {".txt", ".md", ".doc", ".docx", ".ppt", ".pptx", ".pdf"}, "image": {".jpg", ".jpeg", ".png"}, "spreadsheet": {".xls", ".xlsx", ".csv"}, } class UploadGroupPolicy(BaseModel): max_bytes: int = Field(..., ge=1, le=MAX_CONFIGURED_FILE_BYTES) extensions: list[str] = Field(default_factory=list) @field_validator("extensions") @classmethod def _normalize_extensions(cls, values: list[str]) -> list[str]: normalized: list[str] = [] seen: set[str] = set() for value in values: extension = normalize_extension(value) if not extension or len(extension) > 16 or not extension[1:].isalnum(): raise ValueError(f"Invalid file extension: {value!r}") if extension not in seen: normalized.append(extension) seen.add(extension) return normalized class UploadGroups(BaseModel): document: UploadGroupPolicy image: UploadGroupPolicy spreadsheet: UploadGroupPolicy class LargeFilePolicy(BaseModel): threshold_bytes: int = Field(default=1 * MIB, ge=1, le=MAX_CONFIGURED_TOTAL_BYTES) max_result_chars: int = Field(default=12_000, ge=1024, le=MAX_MODEL_RESULT_CHARS) hard_max_result_chars: int = Field(default=MAX_MODEL_RESULT_CHARS, ge=1024, le=MAX_MODEL_RESULT_CHARS) @model_validator(mode="after") def _soft_limit_not_above_hard_limit(self) -> LargeFilePolicy: if self.max_result_chars > self.hard_max_result_chars: raise ValueError("large_file.max_result_chars must not exceed hard_max_result_chars") return self class UploadPolicyDocument(BaseModel): max_files_per_message: int = Field(default=20, ge=1, le=MAX_CONFIGURED_FILES) max_total_bytes: int = Field(default=100 * MIB, ge=1, le=MAX_CONFIGURED_TOTAL_BYTES) groups: UploadGroups large_file: LargeFilePolicy = Field(default_factory=LargeFilePolicy) @model_validator(mode="after") def _validate_groups(self) -> UploadPolicyDocument: seen: set[str] = set() for group_name in ("document", "image", "spreadsheet"): group = getattr(self.groups, group_name) invalid = set(group.extensions) - _SUPPORTED_BY_GROUP[group_name] if invalid: raise ValueError( f"Unsupported {group_name} extension(s): {', '.join(sorted(invalid))}" ) overlap = seen.intersection(group.extensions) if overlap: raise ValueError(f"Extensions cannot appear in multiple groups: {', '.join(sorted(overlap))}") seen.update(group.extensions) if group.max_bytes > self.max_total_bytes: raise ValueError(f"groups.{group_name}.max_bytes must not exceed max_total_bytes") if self.large_file.threshold_bytes > self.max_total_bytes: raise ValueError("large_file.threshold_bytes must not exceed max_total_bytes") return self class UploadPolicyResponse(UploadPolicyDocument): source: Literal["default", "database"] = "default" revision: int = 0 updated_by: str | None = None updated_at: datetime | str | None = None def _get_store(request: Request) -> UploadPolicyStore: store = getattr(request.app.state, "upload_policy_store", None) if store is None: raise HTTPException(status_code=503, detail="Upload policy store not available") return store async def _actor(request: Request): user = await get_optional_user_from_request(request) if user is None and not is_auth_disabled(): raise HTTPException(status_code=401, detail="Not authenticated") return user async def get_effective_upload_policy(request: Request) -> tuple[dict, dict]: """Load the latest policy on every request, falling back to defaults.""" row = await _get_store(request).get_policy() if not row: policy = default_upload_policy() set_runtime_upload_policy(policy) return policy, { "source": "default", "revision": 0, "updated_by": None, "updated_at": None, } try: policy = UploadPolicyDocument.model_validate(row.get("policy") or {}).model_dump() except Exception: # A future/older malformed row must never disable upload enforcement. policy = default_upload_policy() set_runtime_upload_policy(policy) return policy, { "source": "default", "revision": int(row.get("revision") or 0), "updated_by": row.get("updated_by"), "updated_at": row.get("updated_at"), } set_runtime_upload_policy(policy) return policy, { "source": "database", "revision": int(row.get("revision") or 1), "updated_by": row.get("updated_by"), "updated_at": row.get("updated_at"), } def _response(policy: dict, metadata: dict) -> UploadPolicyResponse: return UploadPolicyResponse(**deepcopy(policy), **metadata) @router.get("", response_model=UploadPolicyResponse) async def read_upload_policy(request: Request) -> UploadPolicyResponse: await _actor(request) policy, metadata = await get_effective_upload_policy(request) return _response(policy, metadata) @router.put("", response_model=UploadPolicyResponse) async def replace_upload_policy(request: Request, body: UploadPolicyDocument) -> UploadPolicyResponse: user = await _actor(request) if user is not None and getattr(user, "system_role", None) != "admin": raise HTTPException(status_code=403, detail="上传设置仅限管理员") user_id = str(user.id) if user is not None else None row = await _get_store(request).set_policy(body.model_dump(), updated_by=user_id) policy = UploadPolicyDocument.model_validate(row["policy"]).model_dump() set_runtime_upload_policy(policy) return _response( policy, { "source": "database", "revision": int(row.get("revision") or 1), "updated_by": row.get("updated_by"), "updated_at": row.get("updated_at"), }, ) @router.post("/reset", response_model=UploadPolicyResponse) async def reset_upload_policy(request: Request) -> UploadPolicyResponse: """Restore the built-in defaults as a new persisted revision.""" user = await _actor(request) if user is not None and getattr(user, "system_role", None) != "admin": raise HTTPException(status_code=403, detail="上传设置仅限管理员") user_id = str(user.id) if user is not None else None row = await _get_store(request).set_policy(default_upload_policy(), updated_by=user_id) set_runtime_upload_policy(row["policy"]) return _response( row["policy"], { "source": "database", "revision": int(row.get("revision") or 1), "updated_by": row.get("updated_by"), "updated_at": row.get("updated_at"), }, )