WIP checkpoint: US-1: Asset Import Pipeline — glTF upload endpoint with metadata extraction

backend/app/services/gltf.py

@@ -0,0 +1,195 @@
+"""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_JSON_CHUNK_TYPE = 0x4E4F534A  # "JSON"
+_GLB_BIN_CHUNK_TYPE = 0x004E4942  # "BIN\0"
+
+
+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:
+        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.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()
+    if not stripped.startswith(b"{"):
+        raise ValueError("glTF JSON does not start with an object")
+
+    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,
+    }