287 lines
8.8 KiB
Python
287 lines
8.8 KiB
Python
"""Asset management router."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, UploadFile, File, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.config import settings
|
|
from app.database import get_db
|
|
from app.dependencies import CurrentUser
|
|
from app.models.asset import Asset, AssetType
|
|
from app.models.world import World
|
|
from pydantic import BaseModel, Field
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
# --- Pydantic schemas ---
|
|
|
|
class AssetCreate(BaseModel):
|
|
"""Schema for creating an asset reference."""
|
|
name: str = Field(..., min_length=1, max_length=255)
|
|
description: Optional[str] = None
|
|
asset_type: AssetType
|
|
world_id: Optional[int] = None
|
|
|
|
|
|
class AssetResponse(BaseModel):
|
|
"""Schema for asset responses."""
|
|
id: int
|
|
name: str
|
|
description: Optional[str]
|
|
asset_type: AssetType
|
|
file_path: str
|
|
file_size: Optional[int]
|
|
mime_type: Optional[str]
|
|
metadata_json: Optional[dict]
|
|
world_id: Optional[int]
|
|
owner_id: int
|
|
is_public: bool
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
model_config = {"from_attributes": True}
|
|
|
|
|
|
class AssetUpdate(BaseModel):
|
|
"""Schema for updating an asset."""
|
|
name: Optional[str] = None
|
|
description: Optional[str] = None
|
|
is_public: Optional[bool] = None
|
|
|
|
|
|
# --- Helpers ---
|
|
|
|
def _sanitize_filename(filename: str) -> str:
|
|
"""Strip path components and unsafe characters from a filename."""
|
|
# Remove any directory components (path traversal protection).
|
|
base = os.path.basename(filename.replace("\\", "/"))
|
|
# Keep only safe characters.
|
|
safe = "".join(c for c in base if c.isalnum() or c in "._-")
|
|
return safe or "file"
|
|
|
|
|
|
def _extract_gltf_metadata(content: bytes) -> Optional[dict]:
|
|
"""Extract metadata from a glTF/GLB file.
|
|
|
|
Returns a dict with mesh_count, vertex_count, has_animation, texture_refs
|
|
if the file is a valid glTF/GLB, or None otherwise.
|
|
"""
|
|
try:
|
|
import gltflib
|
|
except ImportError:
|
|
return None
|
|
|
|
try:
|
|
model = gltflib.GLFModel.from_json(content)
|
|
except Exception:
|
|
return None
|
|
|
|
# Count meshes
|
|
meshes = model.meshes or []
|
|
mesh_count = len(meshes)
|
|
|
|
# Count total vertices across all primitives
|
|
vertex_count = 0
|
|
for mesh in meshes:
|
|
primitives = mesh.primitives or []
|
|
for primitive in primitives:
|
|
indices = primitive.indices
|
|
if indices is not None:
|
|
# Use index count
|
|
accessor = model.accessors[indices] if indices < len(model.accessors or []) else None
|
|
if accessor and accessor.count is not None:
|
|
vertex_count += accessor.count
|
|
else:
|
|
vertex_count += 1 # fallback
|
|
else:
|
|
# No indices — count from POSITION attribute
|
|
if primitive.attributes and primitive.attributes.POSITION is not None:
|
|
accessor = model.accessors[primitive.attributes.POSITION] if primitive.attributes.POSITION < len(model.accessors or []) else None
|
|
if accessor and accessor.count is not None:
|
|
vertex_count += accessor.count
|
|
else:
|
|
vertex_count += 1 # fallback
|
|
|
|
# Check for animations
|
|
animations = model.animations or []
|
|
has_animation = len(animations) > 0
|
|
|
|
# Collect texture references
|
|
texture_refs = []
|
|
textures = model.textures or []
|
|
for texture in textures:
|
|
if texture.source is not None and texture.source < len(model.images or []):
|
|
image = model.images[texture.source]
|
|
ref = {}
|
|
if image.name:
|
|
ref["name"] = image.name
|
|
if image.uri:
|
|
ref["uri"] = image.uri
|
|
if image.mimeType:
|
|
ref["mime_type"] = image.mimeType
|
|
texture_refs.append(ref)
|
|
|
|
return {
|
|
"mesh_count": mesh_count,
|
|
"vertex_count": vertex_count,
|
|
"has_animation": has_animation,
|
|
"texture_refs": texture_refs,
|
|
}
|
|
|
|
|
|
# --- Endpoints ---
|
|
|
|
@router.post("/", response_model=AssetResponse, status_code=201)
|
|
async def create_asset(
|
|
payload: AssetCreate,
|
|
current_user: CurrentUser,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Create a new asset reference."""
|
|
# Verify world ownership if world_id is provided
|
|
if payload.world_id is not None:
|
|
world = await db.get(World, payload.world_id)
|
|
if world is None:
|
|
raise HTTPException(status_code=404, detail="World not found")
|
|
if world.owner_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail="Not owner of this world")
|
|
|
|
asset = Asset(
|
|
name=payload.name,
|
|
description=payload.description,
|
|
asset_type=payload.asset_type,
|
|
file_path="", # Will be set after upload
|
|
owner_id=current_user.id,
|
|
world_id=payload.world_id,
|
|
is_public=False,
|
|
)
|
|
db.add(asset)
|
|
await db.commit()
|
|
await db.refresh(asset)
|
|
return asset
|
|
|
|
|
|
@router.get("/", response_model=list[AssetResponse])
|
|
async def list_assets(
|
|
current_user: CurrentUser,
|
|
world_id: Optional[int] = Query(None, description="Filter by world"),
|
|
asset_type: Optional[AssetType] = Query(None),
|
|
limit: int = Query(50, ge=1, le=100),
|
|
offset: int = Query(0, ge=0),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""List assets with optional filters."""
|
|
stmt = select(Asset).where(Asset.owner_id == current_user.id)
|
|
|
|
if world_id is not None:
|
|
stmt = stmt.where(Asset.world_id == world_id)
|
|
if asset_type is not None:
|
|
stmt = stmt.where(Asset.asset_type == asset_type)
|
|
|
|
stmt = stmt.order_by(Asset.created_at.desc()).limit(limit).offset(offset)
|
|
result = await db.execute(stmt)
|
|
assets = result.scalars().all()
|
|
return assets
|
|
|
|
|
|
@router.get("/{asset_id}", response_model=AssetResponse)
|
|
async def get_asset(
|
|
asset_id: int,
|
|
current_user: CurrentUser,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Get a single asset by ID."""
|
|
asset = await db.get(Asset, asset_id)
|
|
if asset is None or asset.owner_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
return asset
|
|
|
|
|
|
@router.patch("/{asset_id}", response_model=AssetResponse)
|
|
async def update_asset(
|
|
asset_id: int,
|
|
payload: AssetUpdate,
|
|
current_user: CurrentUser,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Update an asset."""
|
|
asset = await db.get(Asset, asset_id)
|
|
if asset is None or asset.owner_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
|
|
update_data = payload.model_dump(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(asset, field, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(asset)
|
|
return asset
|
|
|
|
|
|
@router.delete("/{asset_id}", status_code=204)
|
|
async def delete_asset(
|
|
asset_id: int,
|
|
current_user: CurrentUser,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Delete an asset."""
|
|
asset = await db.get(Asset, asset_id)
|
|
if asset is None or asset.owner_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
|
|
await db.delete(asset)
|
|
await db.commit()
|
|
|
|
|
|
@router.post("/{asset_id}/upload", response_model=AssetResponse)
|
|
async def upload_asset_file(
|
|
asset_id: int,
|
|
current_user: CurrentUser,
|
|
file: UploadFile = File(...),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Upload a file for an existing asset."""
|
|
asset = await db.get(Asset, asset_id)
|
|
if asset is None or asset.owner_id != current_user.id:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
|
|
# Sanitize filename to prevent path traversal.
|
|
safe_filename = _sanitize_filename(file.filename or f"asset_{asset_id}")
|
|
|
|
# Validate file size against configured maximum.
|
|
max_bytes = settings.MAX_UPLOAD_SIZE_MB * 1024 * 1024
|
|
content = await file.read()
|
|
file_size = len(content)
|
|
if file_size > max_bytes:
|
|
raise HTTPException(
|
|
status_code=413,
|
|
detail=f"File size {file_size} exceeds maximum allowed size {max_bytes} bytes",
|
|
)
|
|
|
|
# Write the file to the upload directory.
|
|
upload_dir = Path(settings.UPLOAD_DIR)
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
stored_name = f"{asset_id}_{safe_filename}"
|
|
file_path = upload_dir / stored_name
|
|
file_path.write_bytes(content)
|
|
|
|
asset.file_path = str(file_path)
|
|
asset.mime_type = file.content_type
|
|
asset.file_size = file_size
|
|
|
|
# Extract glTF metadata for glTF/GLB files.
|
|
filename_lower = (file.filename or "").lower()
|
|
if filename_lower.endswith(".glb") or filename_lower.endswith(".gltf"):
|
|
metadata = _extract_gltf_metadata(content)
|
|
if metadata is not None:
|
|
asset.metadata_json = str(metadata)
|
|
|
|
await db.commit()
|
|
await db.refresh(asset)
|
|
return asset
|