from __future__ import annotations import asyncio import json import re from datetime import datetime, timezone from pathlib import Path from typing import Any from uuid import UUID from fastapi import APIRouter, HTTPException, Query, Request, Response, status from pydantic import BaseModel, ConfigDict, Field, model_validator from sqlalchemy import func, or_, select from sqlalchemy.exc import IntegrityError from app.core.exceptions import NotFoundError from app.generation.models import GenerationJob, GenerationRequest from app.projects.models import Project, ProjectRenderJob from app.security.assets import CanonicalAssetNotFoundError from app.security.models import CanonicalMediaAsset, CanonicalMediaVariant router = APIRouter(prefix="/v1/assets", tags=["assets"]) class AssetMetadataUpdate(BaseModel): model_config = ConfigDict(extra="forbid") metadata: dict[str, Any] = Field(default_factory=dict) @model_validator(mode="after") def metadata_is_bounded_json(self) -> "AssetMetadataUpdate": try: encoded = json.dumps(self.metadata, ensure_ascii=False, allow_nan=False) except (TypeError, ValueError) as exc: raise ValueError("metadata must contain JSON-compatible values") from exc if len(encoded.encode("utf-8")) > 16_384: raise ValueError("metadata must not exceed 16 KiB") return self def _identity(request: Request) -> tuple[str, str]: context = getattr(request.state, "auth", None) if context is None or not context.workspace_id or not context.user_id: raise HTTPException(status_code=403, detail="An active workspace membership is required.") return context.workspace_id, context.user_id def _view(record: CanonicalMediaAsset, output_dir: Path) -> dict[str, Any]: output_root = output_dir.resolve() root = (output_root / record.request_id).resolve() path = (root / record.filename).resolve() safe = ( bool(re.fullmatch(r"[0-9a-fA-F-]{16,64}", record.request_id)) and bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{0,254}", record.filename)) and root.parent == output_root and path.parent == root ) try: available = safe and path.is_file() and path.stat().st_size == record.file_size except OSError: available = False return { "id": record.id, "workspace_id": record.workspace_id, "project_id": record.project_id, "request_id": record.request_id, "filename": record.filename, "mime_type": record.mime_type, "file_size": record.file_size, "metadata": record.metadata_json or {}, "file_available": available, "created_at": record.created_at, "updated_at": record.updated_at, } @router.get("", response_model=dict[str, Any]) async def list_assets( request: Request, project_id: UUID | None = Query(default=None), q: str | None = Query(default=None, min_length=1, max_length=200), offset: int = Query(default=0, ge=0), limit: int = Query(default=100, ge=1, le=500), ) -> dict[str, Any]: workspace_id, user_id = _identity(request) container = request.app.state.container filters = [CanonicalMediaAsset.workspace_id == workspace_id] if project_id is not None: filters.append(CanonicalMediaAsset.project_id == str(project_id)) if q: term = "%" + q.strip().replace("%", "\\%").replace("_", "\\_") + "%" filters.append(or_( CanonicalMediaAsset.filename.ilike(term, escape="\\"), CanonicalMediaAsset.mime_type.ilike(term, escape="\\"), )) async with container.security_database.tenant_session( workspace_id=workspace_id, user_id=user_id ) as session: total = await session.scalar( select(func.count()).select_from(CanonicalMediaAsset).where(*filters) ) records = (await session.scalars( select(CanonicalMediaAsset) .where(*filters) .order_by(CanonicalMediaAsset.created_at.desc(), CanonicalMediaAsset.id) .offset(offset) .limit(limit) )).all() return { "items": [_view(record, container.settings.output_dir) for record in records], "total": int(total or 0), "offset": offset, "limit": limit, } @router.get("/{asset_id}", response_model=dict[str, Any]) async def get_asset(request: Request, asset_id: UUID) -> dict[str, Any]: workspace_id, user_id = _identity(request) container = request.app.state.container try: record = await container.assets.get_owned_by_id( workspace_id=workspace_id, user_id=user_id, asset_id=str(asset_id) ) path = container.cleanup.resolve_download(record.request_id, record.filename) await container.assets.verify_file(record, path) except (CanonicalAssetNotFoundError, NotFoundError) as exc: raise HTTPException(status_code=404, detail="Asset was not found or failed integrity verification.") from exc return _view(record, container.settings.output_dir) @router.patch("/{asset_id}", response_model=dict[str, Any]) async def update_asset_metadata( request: Request, asset_id: UUID, payload: AssetMetadataUpdate ) -> dict[str, Any]: workspace_id, user_id = _identity(request) container = request.app.state.container async with container.security_database.tenant_session( workspace_id=workspace_id, user_id=user_id ) as session: record = await session.scalar(select(CanonicalMediaAsset).where( CanonicalMediaAsset.id == str(asset_id), CanonicalMediaAsset.workspace_id == workspace_id, )) if record is None: raise HTTPException(status_code=404, detail="Asset was not found in this workspace.") record.metadata_json = payload.metadata record.updated_at = datetime.now(timezone.utc) await session.commit() await session.refresh(record) return _view(record, container.settings.output_dir) @router.delete("/{asset_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_asset(request: Request, asset_id: UUID) -> Response: workspace_id, user_id = _identity(request) container = request.app.state.container file_path: Path | None = None async with container.security_database.tenant_session( workspace_id=workspace_id, user_id=user_id ) as session: record = await session.scalar( select(CanonicalMediaAsset) .where( CanonicalMediaAsset.id == str(asset_id), CanonicalMediaAsset.workspace_id == workspace_id, ) .with_for_update() ) if record is None: raise HTTPException(status_code=404, detail="Asset was not found in this workspace.") if record.project_id is not None: raise HTTPException(status_code=409, detail="Detach this asset from its project before deleting it.") references = ( select(GenerationRequest.id).where(GenerationRequest.input_asset_id == record.id).limit(1), select(GenerationJob.id).where(GenerationJob.output_asset_id == record.id).limit(1), select(Project.thumbnail_asset_id).where(Project.thumbnail_asset_id == record.id).limit(1), select(ProjectRenderJob.id).where(ProjectRenderJob.output_asset_id == record.id).limit(1), select(CanonicalMediaVariant.id).where(CanonicalMediaVariant.asset_id == record.id).limit(1), ) for statement in references: if await session.scalar(statement) is not None: raise HTTPException(status_code=409, detail="This asset is referenced by a project or job and cannot be deleted.") try: file_path = container.cleanup.resolve_download(record.request_id, record.filename) await container.assets.verify_file(record, file_path) except (CanonicalAssetNotFoundError, NotFoundError): file_path = None try: await session.delete(record) await session.commit() except IntegrityError as exc: await session.rollback() raise HTTPException(status_code=409, detail="This asset is still referenced and cannot be deleted.") from exc if file_path is not None: try: await asyncio.to_thread(file_path.unlink, missing_ok=True) except OSError: # The database record is already gone; the periodic output cleanup # reconciles any orphaned file without exposing its filesystem path. pass return Response(status_code=status.HTTP_204_NO_CONTENT)