369 lines
12 KiB
Python
369 lines
12 KiB
Python
"""Asset management router."""
|
|
|
|
import json
|
|
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}
|
|
|
|
@classmethod
|
|
def model_validate(cls, obj, **kwargs):
|
|
"""Override to handle metadata_json stored as JSON string."""
|
|
if isinstance(obj, dict):
|
|
obj = dict(obj)
|
|
if "metadata_json" in obj and isinstance(obj["metadata_json"], str):
|
|
try:
|
|
obj["metadata_json"] = json.loads(obj["metadata_json"])
|
|
except (json.JSONDecodeError, TypeError):
|
|
obj["metadata_json"] = None
|
|
return super().model_validate(obj, **kwargs)
|
|
|
|
|
|
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.
|
|
|
|
Uses only the standard library (struct, json) so no extra dependencies
|
|
are required.
|
|
"""
|
|
if not content:
|
|
return None
|
|
|
|
# --- GLB detection ---
|
|
# GLB magic: 0x676C5442 ("glTF") followed by version 2 and payload length.
|
|
if len(content) >= 12:
|
|
magic = int.from_bytes(content[0:4], "little")
|
|
if magic == 0x46546C67: # "glTF"
|
|
version = int.from_bytes(content[4:8], "little")
|
|
if version == 2:
|
|
return _parse_glb(content)
|
|
|
|
# --- .gltf JSON detection ---
|
|
# Try parsing as JSON; if it has a top-level "scene" or "scenes" key,
|
|
# it's likely a .gltf file.
|
|
try:
|
|
data = json.loads(content)
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
return None
|
|
|
|
if not isinstance(data, dict):
|
|
return None
|
|
|
|
if "scene" not in data and "scenes" not in data:
|
|
return None
|
|
|
|
return _parse_gltf_json(data)
|
|
|
|
|
|
def _parse_glb(content: bytes) -> Optional[dict]:
|
|
"""Parse a binary GLB chunk structure."""
|
|
# Header: magic(4) + version(4) + length(4) = 12 bytes
|
|
# Then chunks: length(4) + type(4) + data...
|
|
offset = 12
|
|
json_start = None
|
|
json_length = 0
|
|
binary_start = None
|
|
binary_length = 0
|
|
|
|
while offset < len(content):
|
|
if offset + 8 > len(content):
|
|
break
|
|
chunk_length = int.from_bytes(content[offset:offset + 4], "little")
|
|
chunk_type = int.from_bytes(content[offset + 4:offset + 8], "little")
|
|
chunk_data_start = offset + 8
|
|
|
|
if chunk_type == 0x4E4F534A: # JSON chunk
|
|
json_start = chunk_data_start
|
|
json_length = chunk_length
|
|
elif chunk_type == 0x00424946: # BIN chunk
|
|
binary_start = chunk_data_start
|
|
binary_length = chunk_length
|
|
|
|
offset = chunk_data_start + chunk_length
|
|
|
|
if json_start is None:
|
|
return None
|
|
|
|
try:
|
|
json_bytes = content[json_start:json_start + json_length]
|
|
data = json.loads(json_bytes)
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
return None
|
|
|
|
return _parse_gltf_json(data)
|
|
|
|
|
|
def _parse_gltf_json(data: dict) -> Optional[dict]:
|
|
"""Parse a glTF JSON structure and extract metadata."""
|
|
meshes = data.get("meshes") or []
|
|
animations = data.get("animations") or []
|
|
textures = data.get("textures") or []
|
|
images = data.get("images") or []
|
|
accessors = data.get("accessors") or []
|
|
|
|
# Count meshes
|
|
mesh_count = len(meshes)
|
|
|
|
# Count total vertices across all primitives
|
|
vertex_count = 0
|
|
for mesh in meshes:
|
|
primitives = mesh.get("primitives") or []
|
|
for primitive in primitives:
|
|
indices = primitive.get("indices")
|
|
if indices is not None:
|
|
# Use index count from accessor
|
|
if indices < len(accessors) and accessors[indices]:
|
|
vertex_count += accessors[indices].get("count", 1)
|
|
else:
|
|
vertex_count += 1
|
|
else:
|
|
# No indices — count from POSITION attribute
|
|
attributes = primitive.get("attributes") or {}
|
|
pos_accessor = attributes.get("POSITION")
|
|
if pos_accessor is not None:
|
|
if pos_accessor < len(accessors) and accessors[pos_accessor]:
|
|
vertex_count += accessors[pos_accessor].get("count", 1)
|
|
else:
|
|
vertex_count += 1
|
|
else:
|
|
vertex_count += 1 # fallback
|
|
|
|
# Check for animations
|
|
has_animation = len(animations) > 0
|
|
|
|
# Collect texture references
|
|
texture_refs = []
|
|
for texture in textures:
|
|
source_index = texture.get("source")
|
|
ref = {}
|
|
if source_index is not None and source_index < len(images):
|
|
image = images[source_index]
|
|
if isinstance(image, dict):
|
|
if image.get("name"):
|
|
ref["name"] = image["name"]
|
|
if image.get("uri"):
|
|
ref["uri"] = image["uri"]
|
|
if image.get("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 = json.dumps(metadata)
|
|
|
|
await db.commit()
|
|
await db.refresh(asset)
|
|
return asset
|