Compare commits

..

8 Commits

8 changed files with 490 additions and 156 deletions
+1
View File
@@ -30,6 +30,7 @@ class Settings(BaseSettings):
# File uploads # File uploads
UPLOAD_DIR: str = "uploads" UPLOAD_DIR: str = "uploads"
MAX_UPLOAD_SIZE_MB: int = 100 MAX_UPLOAD_SIZE_MB: int = 100
GLTF_MAX_FILE_SIZE_MB: int = 200
model_config = SettingsConfigDict( model_config = SettingsConfigDict(
env_file=".env", env_file=".env",
+39
View File
@@ -0,0 +1,39 @@
"""Script model — game logic scripts owned by users and associated with worlds."""
from typing import Optional
from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey, func
from sqlalchemy.dialects.postgresql import ENUM, JSONB
from sqlalchemy.orm import relationship
import enum
from app.database import Base
class ScriptType(str, enum.Enum):
"""Type of script storage."""
GDSCRIPT = "gdscript"
DSL = "dsl"
class Script(Base):
"""A game script — either raw GDScript source or a DSL payload."""
__tablename__ = "scripts"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), nullable=False)
world_id = Column(Integer, ForeignKey("worlds.id", ondelete="CASCADE"), nullable=True)
owner_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
source_code = Column(Text, nullable=True)
version = Column(Integer, nullable=False, server_default="1")
script_type = Column(
ENUM(ScriptType, name="script_type", create_type=False),
nullable=False,
default=ScriptType.GDSCRIPT,
)
dsl_payload = Column(JSONB, nullable=True)
created_at = Column(DateTime(timezone=True), server_default=func.now())
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
owner = relationship("User", back_populates="scripts")
world = relationship("World", back_populates="scripts")
+31 -156
View File
@@ -14,6 +14,7 @@ from app.database import get_db
from app.dependencies import CurrentUser from app.dependencies import CurrentUser
from app.models.asset import Asset, AssetType from app.models.asset import Asset, AssetType
from app.models.world import World from app.models.world import World
from app.services import gltf as gltf_service
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
router = APIRouter() router = APIRouter()
@@ -38,7 +39,6 @@ class AssetResponse(BaseModel):
file_path: str file_path: str
file_size: Optional[int] file_size: Optional[int]
mime_type: Optional[str] mime_type: Optional[str]
metadata_json: Optional[dict]
world_id: Optional[int] world_id: Optional[int]
owner_id: int owner_id: int
is_public: bool is_public: bool
@@ -47,18 +47,6 @@ class AssetResponse(BaseModel):
model_config = {"from_attributes": True} 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): class AssetUpdate(BaseModel):
"""Schema for updating an asset.""" """Schema for updating an asset."""
@@ -78,142 +66,12 @@ def _sanitize_filename(filename: str) -> str:
return safe or "file" return safe or "file"
def _extract_gltf_metadata(content: bytes) -> Optional[dict]: def _detect_asset_type(filename: Optional[str], content: bytes) -> AssetType:
"""Extract metadata from a glTF/GLB file. """Detect the asset type from filename and content."""
if filename and gltf_service.looks_like_gltf(content, filename):
Returns a dict with mesh_count, vertex_count, has_animation, texture_refs return AssetType.MODEL_3D
if the file is a valid glTF/GLB, or None otherwise. # Fallback: let the caller decide based on mime_type or default.
return AssetType.OTHER
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 --- # --- Endpoints ---
@@ -335,10 +193,16 @@ async def upload_asset_file(
# Sanitize filename to prevent path traversal. # Sanitize filename to prevent path traversal.
safe_filename = _sanitize_filename(file.filename or f"asset_{asset_id}") safe_filename = _sanitize_filename(file.filename or f"asset_{asset_id}")
# Validate file size against configured maximum. # Read the file content.
max_bytes = settings.MAX_UPLOAD_SIZE_MB * 1024 * 1024
content = await file.read() content = await file.read()
file_size = len(content) file_size = len(content)
# Determine which size limit to apply.
if gltf_service.looks_like_gltf(content, file.filename):
max_bytes = settings.GLTF_MAX_FILE_SIZE_MB * 1024 * 1024
else:
max_bytes = settings.MAX_UPLOAD_SIZE_MB * 1024 * 1024
if file_size > max_bytes: if file_size > max_bytes:
raise HTTPException( raise HTTPException(
status_code=413, status_code=413,
@@ -352,16 +216,27 @@ async def upload_asset_file(
file_path = upload_dir / stored_name file_path = upload_dir / stored_name
file_path.write_bytes(content) file_path.write_bytes(content)
# Detect asset type from content/filename.
detected_type = _detect_asset_type(file.filename, content)
# Update asset record.
asset.file_path = str(file_path) asset.file_path = str(file_path)
asset.mime_type = file.content_type asset.mime_type = file.content_type
asset.file_size = file_size asset.file_size = file_size
# Extract glTF metadata for glTF/GLB files. # If this is a glTF file, extract and store metadata.
filename_lower = (file.filename or "").lower() if detected_type == AssetType.MODEL_3D:
if filename_lower.endswith(".glb") or filename_lower.endswith(".gltf"): try:
metadata = _extract_gltf_metadata(content) metadata = gltf_service.extract_gltf_metadata(content)
if metadata is not None:
asset.metadata_json = json.dumps(metadata) asset.metadata_json = json.dumps(metadata)
# Also set the asset type to model_3d if it was auto-detected.
if asset.asset_type == AssetType.OTHER:
asset.asset_type = AssetType.MODEL_3D
except ValueError as exc:
raise HTTPException(
status_code=400,
detail=f"Invalid glTF file: {exc}",
)
await db.commit() await db.commit()
await db.refresh(asset) await db.refresh(asset)
View File
+197
View File
@@ -0,0 +1,197 @@
"""Lightweight glTF / GLB metadata extraction.
This module intentionally avoids heavyweight dependencies such as trimesh or
pygltflib. For the metadata fields the asset upload pipeline needs (mesh and
vertex counts, animation presence, texture references, and the bounding box)
a small amount of glTF-specific parsing is sufficient and keeps the backend
dependency-free.
"""
from __future__ import annotations
import json
import struct
from typing import Any, Optional
# GLB binary header magic bytes ("glTF").
_GLB_MAGIC = b"glTF"
_GLB_MAGIC_INT = 0x46546C67 # b"glTF" interpreted as little-endian uint32
_GLB_JSON_CHUNK_TYPE = 0x4E4F534A # "JSON"
_GLB_BIN_CHUNK_TYPE = 0x004E4942 # "BIN\0"
# Whitespace allowed around a GLB JSON chunk. The spec pads with spaces (0x20),
# but be lenient with null bytes and other whitespace that real exporters emit.
_JSON_CHUNK_TRIM = b" \t\r\n\x00"
def looks_like_gltf(content: bytes, filename: Optional[str] = None) -> bool:
"""Return True when the bytes or filename indicate a glTF/GLB asset."""
if content.startswith(_GLB_MAGIC):
return True
if filename:
lowered = filename.lower()
if lowered.endswith(".gltf") or lowered.endswith(".glb"):
return True
# A bare JSON buffer (e.g. uploaded without an extension) that has the
# glTF asset marker is also a glTF file.
stripped = content.lstrip()
if stripped.startswith(b"{"):
try:
payload = json.loads(stripped.decode("utf-8"))
except (ValueError, UnicodeDecodeError):
return False
return isinstance(payload, dict) and "asset" in payload and "version" in payload.get("asset", {})
return False
def _parse_glb(content: bytes) -> dict[str, Any]:
"""Parse the JSON chunk of a GLB binary container."""
if len(content) < 12:
raise ValueError("GLB file is too short")
magic, _version, _length = struct.unpack_from("<III", content, 0)
if magic != _GLB_MAGIC_INT:
raise ValueError("invalid GLB magic")
offset = 12
json_chunk: Optional[dict[str, Any]] = None
while offset + 8 <= len(content):
chunk_length, chunk_type = struct.unpack_from("<II", content, offset)
offset += 8
chunk_end = offset + chunk_length
if chunk_end > len(content):
raise ValueError("GLB chunk overruns file boundary")
chunk_data = content[offset:chunk_end]
if chunk_type == _GLB_JSON_CHUNK_TYPE:
try:
json_chunk = json.loads(chunk_data.strip(_JSON_CHUNK_TRIM).decode("utf-8"))
except (ValueError, UnicodeDecodeError) as exc:
raise ValueError(f"invalid GLB JSON chunk: {exc}") from exc
offset = chunk_end
if json_chunk is None:
raise ValueError("GLB file does not contain a JSON chunk")
return json_chunk
def _parse_gltf_json(content: bytes) -> dict[str, Any]:
"""Parse a plain .gltf JSON file."""
stripped = content.lstrip()
try:
payload = json.loads(stripped.decode("utf-8"))
except (ValueError, UnicodeDecodeError) as exc:
raise ValueError(f"invalid glTF JSON: {exc}") from exc
if not isinstance(payload, dict):
raise ValueError("glTF JSON root must be an object")
return payload
def _position_accessor_counts(gltf: dict[str, Any]) -> list[int]:
"""Return POSITION accessor counts for every mesh primitive."""
counts: list[int] = []
accessors = gltf.get("accessors", [])
for mesh in gltf.get("meshes", []):
for primitive in mesh.get("primitives", []):
attributes = primitive.get("attributes", {})
position_index = attributes.get("POSITION")
if position_index is None:
continue
try:
accessor = accessors[int(position_index)]
except (IndexError, TypeError, ValueError):
continue
count = accessor.get("count")
if isinstance(count, int):
counts.append(count)
return counts
def _position_bounds(gltf: dict[str, Any]) -> list[list[float]]:
"""Return min/max bounds for every POSITION accessor that declares them."""
bounds: list[list[float]] = []
accessors = gltf.get("accessors", [])
for mesh in gltf.get("meshes", []):
for primitive in mesh.get("primitives", []):
attributes = primitive.get("attributes", {})
position_index = attributes.get("POSITION")
if position_index is None:
continue
try:
accessor = accessors[int(position_index)]
except (IndexError, TypeError, ValueError):
continue
minimum = accessor.get("min")
maximum = accessor.get("max")
if isinstance(minimum, list) and isinstance(maximum, list):
bounds.append([float(x) for x in minimum])
bounds.append([float(x) for x in maximum])
return bounds
def _texture_refs(gltf: dict[str, Any]) -> list[str]:
"""Collect external texture source references from the image array."""
refs: list[str] = []
for index, image in enumerate(gltf.get("images", [])):
uri = image.get("uri")
if isinstance(uri, str) and not uri.startswith("data:"):
refs.append(uri)
continue
name = image.get("name")
if isinstance(name, str) and name:
refs.append(name)
continue
if isinstance(uri, str):
# Embedded data URI; keep a stable, non-huge reference.
refs.append(f"data:{index}")
continue
buffer_view = image.get("bufferView")
if buffer_view is not None:
refs.append(f"bufferView:{buffer_view}")
continue
refs.append(f"image:{index}")
return refs
def extract_gltf_metadata(content: bytes) -> dict[str, Any]:
"""Extract upload-relevant metadata from glTF/GLB bytes.
Raises ValueError for malformed glTF content. The returned dict contains:
mesh_count, vertex_count, has_animation, texture_refs,
bounding_box_min, bounding_box_max.
"""
if content.startswith(_GLB_MAGIC):
gltf = _parse_glb(content)
else:
gltf = _parse_gltf_json(content)
position_counts = _position_accessor_counts(gltf)
bounds = _position_bounds(gltf)
if bounds:
bounding_box_min = [min(axis) for axis in zip(*bounds[::2])]
bounding_box_max = [max(axis) for axis in zip(*bounds[1::2])]
else:
bounding_box_min = [0.0, 0.0, 0.0]
bounding_box_max = [0.0, 0.0, 0.0]
return {
"mesh_count": len(gltf.get("meshes", [])),
"vertex_count": sum(position_counts),
"has_animation": bool(gltf.get("animations")),
"texture_refs": _texture_refs(gltf),
"bounding_box_min": bounding_box_min,
"bounding_box_max": bounding_box_max,
}
+220
View File
@@ -0,0 +1,220 @@
"""Tests for the lightweight glTF metadata extraction service."""
import json
import struct
import pytest
from app.services.gltf import (
extract_gltf_metadata,
looks_like_gltf,
_parse_glb,
_parse_gltf_json,
)
# ---------------------------------------------------------------------------
# Test data helpers
# ---------------------------------------------------------------------------
def _make_gltf_json(
meshes: int = 1,
vertices_per_mesh: int = 24,
has_animations: bool = False,
images: list[dict] | None = None,
) -> dict:
"""Build a minimal glTF JSON structure."""
accessors = []
meshes_data = []
for i in range(meshes):
acc_index = len(accessors)
accessors.append({
"bufferView": 0,
"componentType": 5126, # FLOAT
"count": vertices_per_mesh,
"type": "VEC3",
"max": [1.0, 1.0, 1.0],
"min": [-1.0, -1.0, -1.0],
})
meshes_data.append({
"primitives": [{
"attributes": {"POSITION": acc_index},
}],
})
gltf = {
"asset": {"version": "2.0"},
"accessors": accessors,
"meshes": meshes_data,
}
if has_animations:
gltf["animations"] = [{"name": "anim1", "channels": [], "samplers": []}]
if images:
gltf["images"] = images
return gltf
def _make_glb(gltf_json: dict) -> bytes:
"""Wrap *gltf_json* in a minimal GLB binary container."""
json_bytes = json.dumps(gltf_json).encode("utf-8")
json_length = len(json_bytes)
# Pad JSON chunk to 4-byte boundary.
json_padding = (4 - json_length % 4) % 4
json_chunk = json_bytes + b"\x00" * json_padding
json_chunk_type = 0x4E4F534A # "JSON"
# No BIN chunk.
bin_length = 0
bin_chunk_type = 0x004E4942
total_length = 12 + 8 + len(json_chunk) + 8 + bin_length
header = struct.pack("<III", 0x46546C67, 2, total_length) # magic, version, length
json_header = struct.pack("<II", len(json_chunk), json_chunk_type)
if bin_length > 0:
bin_header = struct.pack("<II", bin_length, bin_chunk_type)
else:
bin_header = b""
return header + json_header + json_chunk + bin_header
# ---------------------------------------------------------------------------
# looks_like_gltf
# ---------------------------------------------------------------------------
class TestLooksLikeGltf:
def test_glb_magic(self):
assert looks_like_gltf(b"glTF\x00\x00\x00\x00") is True
def test_gltf_extension(self):
assert looks_like_gltf(b'{"asset":{"version":"2.0"}}', "model.gltf") is True
def test_glb_extension(self):
assert looks_like_gltf(b"some binary", "model.glb") is True
def test_json_with_asset_version(self):
payload = json.dumps({"asset": {"version": "2.0"}}).encode()
assert looks_like_gltf(payload) is True
def test_plain_json_no_asset(self):
payload = json.dumps({"foo": "bar"}).encode()
assert looks_like_gltf(payload) is False
def test_non_json_bytes(self):
assert looks_like_gltf(b"\x89PNG\r\n\x1a\n") is False
# ---------------------------------------------------------------------------
# _parse_glb
# ---------------------------------------------------------------------------
class TestParseGlb:
def test_parses_valid_glb(self):
gltf_json = _make_gltf_json()
glb = _make_glb(gltf_json)
result = _parse_glb(glb)
assert result["asset"]["version"] == "2.0"
assert len(result["meshes"]) == 1
def test_rejects_too_short(self):
with pytest.raises(ValueError, match="too short"):
_parse_glb(b"glTF\x00\x00")
def test_rejects_invalid_magic(self):
with pytest.raises(ValueError, match="invalid GLB magic"):
_parse_glb(b"NOTglTF\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00")
# ---------------------------------------------------------------------------
# _parse_gltf_json
# ---------------------------------------------------------------------------
class TestParseGltfJson:
def test_parses_valid_gltf(self):
gltf_json = _make_gltf_json()
result = _parse_gltf_json(json.dumps(gltf_json).encode())
assert result["asset"]["version"] == "2.0"
def test_rejects_non_object(self):
with pytest.raises(ValueError, match="root must be an object"):
_parse_gltf_json(b"[1, 2, 3]")
# ---------------------------------------------------------------------------
# extract_gltf_metadata (GLB path)
# ---------------------------------------------------------------------------
class TestExtractGltfMetadataGlb:
def test_basic_glb(self):
gltf_json = _make_gltf_json(meshes=1, vertices_per_mesh=24)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["mesh_count"] == 1
assert meta["vertex_count"] == 24
assert meta["has_animation"] is False
assert meta["bounding_box_min"] == [-1.0, -1.0, -1.0]
assert meta["bounding_box_max"] == [1.0, 1.0, 1.0]
assert meta["texture_refs"] == []
def test_multiple_meshes(self):
gltf_json = _make_gltf_json(meshes=3, vertices_per_mesh=12)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["mesh_count"] == 3
assert meta["vertex_count"] == 36
def test_with_animation(self):
gltf_json = _make_gltf_json(has_animations=True)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["has_animation"] is True
def test_with_external_texture(self):
gltf_json = _make_gltf_json(images=[{"uri": "textures/rock.png"}])
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["texture_refs"] == ["textures/rock.png"]
def test_with_named_texture(self):
gltf_json = _make_gltf_json(images=[{"name": "my_texture"}])
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["texture_refs"] == ["my_texture"]
def test_empty_meshes(self):
gltf_json = _make_gltf_json(meshes=0)
glb = _make_glb(gltf_json)
meta = extract_gltf_metadata(glb)
assert meta["mesh_count"] == 0
assert meta["vertex_count"] == 0
assert meta["bounding_box_min"] == [0.0, 0.0, 0.0]
assert meta["bounding_box_max"] == [0.0, 0.0, 0.0]
# ---------------------------------------------------------------------------
# extract_gltf_metadata (plain .gltf JSON path)
# ---------------------------------------------------------------------------
class TestExtractGltfMetadataGltfJson:
def test_plain_gltf(self):
gltf_json = _make_gltf_json(meshes=2, vertices_per_mesh=8)
raw = json.dumps(gltf_json).encode()
meta = extract_gltf_metadata(raw)
assert meta["mesh_count"] == 2
assert meta["vertex_count"] == 16
assert meta["has_animation"] is False
def test_plain_gltf_with_animation(self):
gltf_json = _make_gltf_json(has_animations=True)
raw = json.dumps(gltf_json).encode()
meta = extract_gltf_metadata(raw)
assert meta["has_animation"] is True
# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------
class TestExtractGltfMetadataErrors:
def test_invalid_glb(self):
with pytest.raises(ValueError):
extract_gltf_metadata(b"not a glb")
def test_invalid_gltf_json(self):
with pytest.raises(ValueError):
_parse_gltf_json(b"[1, 2, 3]")
+1
View File
@@ -0,0 +1 @@
uid://cjd0uhll7iylj
+1
View File
@@ -0,0 +1 @@
uid://bgnc3iamfgf6k