feat: implementar 12 mejoras, tests, docs y optimizaciones
- Fase A: license templates, search history, cost estimator - Fase B: import URL, bulk ZIP, batch download - Fase C: comparison mode, mesh validation, measurement tool - Fase D: cross-section clipping, overhang heatmap, layer animation - Refactor Pydantic/SQLAlchemy warnings - 24 tests pytest - README actualizado - WebP thumbnails, lazy loading, cache headers
This commit is contained in:
Consultoria AS
0
app/routers/__init__.py
Normal file
0
app/routers/__init__.py
Normal file
739
app/routers/models.py
Normal file
739
app/routers/models.py
Normal file
@@ -0,0 +1,739 @@
|
||||
import os
|
||||
import shutil
|
||||
import hashlib
|
||||
import json
|
||||
import zipfile
|
||||
import qrcode
|
||||
import io
|
||||
import tempfile
|
||||
import urllib.parse
|
||||
import httpx
|
||||
from fastapi import APIRouter, Depends, UploadFile, File, Form, HTTPException, Query, Request, Body
|
||||
from fastapi.responses import FileResponse, StreamingResponse, JSONResponse
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import Optional, List
|
||||
|
||||
def _cache_headers(seconds: int):
|
||||
return {"Cache-Control": f"public, max-age={seconds}"}
|
||||
|
||||
from app.database import get_db
|
||||
from app.models import Model3D, Tag, ModelFile, Rating, Comment, Collection
|
||||
from app.schemas import (
|
||||
Model3DResponse, Model3DUpdate, RatingResponse, CommentResponse,
|
||||
CollectionResponse, CollectionCreate, CollectionDetailResponse
|
||||
)
|
||||
from app.parsers import parse_model_file, generate_thumbnail, generate_generic_thumbnail
|
||||
|
||||
router = APIRouter(prefix="/api/models", tags=["models"])
|
||||
|
||||
UPLOAD_DIR = "uploads"
|
||||
THUMBNAIL_DIR = "thumbnails"
|
||||
IMAGES_DIR = "images"
|
||||
os.makedirs(UPLOAD_DIR, exist_ok=True)
|
||||
os.makedirs(THUMBNAIL_DIR, exist_ok=True)
|
||||
os.makedirs(IMAGES_DIR, exist_ok=True)
|
||||
|
||||
|
||||
def _file_hash(file_path: str) -> str:
|
||||
h = hashlib.sha256()
|
||||
with open(file_path, 'rb') as f:
|
||||
while chunk := f.read(8192):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def _get_or_create_tags(db: Session, tag_names: List[str]) -> List[Tag]:
|
||||
tags = []
|
||||
for name in tag_names:
|
||||
name = name.strip().lower()
|
||||
if not name:
|
||||
continue
|
||||
tag = db.query(Tag).filter(Tag.name == name).first()
|
||||
if not tag:
|
||||
tag = Tag(name=name)
|
||||
db.add(tag)
|
||||
db.flush()
|
||||
tags.append(tag)
|
||||
return tags
|
||||
|
||||
|
||||
def _calc_avg_rating(model: Model3D) -> Optional[float]:
|
||||
if not model.ratings:
|
||||
return None
|
||||
return round(sum(r.stars for r in model.ratings) / len(model.ratings), 1)
|
||||
|
||||
|
||||
# --- Cost estimator helpers ---
|
||||
def _estimate_print_time_seconds(volume_cm3: float) -> int:
|
||||
# Rough estimate: ~1 hour per 10cm3 at standard speed
|
||||
return int(volume_cm3 * 360)
|
||||
|
||||
|
||||
# --- Process a single STL file into a model ---
|
||||
def _process_single_file(db: Session, file_path: str, title: str, description: str,
|
||||
author: str, license_str: str, tags_str: str, category: str,
|
||||
part_name: Optional[str] = None, is_primary: bool = False) -> Model3D:
|
||||
file_hash = _file_hash(file_path)
|
||||
existing = db.query(Model3D).filter(Model3D.file_hash == file_hash).first()
|
||||
if existing:
|
||||
os.remove(file_path)
|
||||
raise HTTPException(status_code=409,
|
||||
detail=f"File already exists as '{existing.title}' (ID: {existing.id})")
|
||||
|
||||
try:
|
||||
model_data = parse_model_file(file_path)
|
||||
except Exception as e:
|
||||
os.remove(file_path)
|
||||
raise HTTPException(status_code=400, detail=f"Failed to parse: {str(e)}")
|
||||
|
||||
thumbnail_path = os.path.join(THUMBNAIL_DIR, f"{os.path.splitext(os.path.basename(file_path))[0]}.webp")
|
||||
generate_thumbnail(model_data['vertices'], thumbnail_path)
|
||||
|
||||
db_model = Model3D(
|
||||
title=title,
|
||||
filename=os.path.basename(file_path),
|
||||
description=description,
|
||||
author=author,
|
||||
license=license_str,
|
||||
category=category,
|
||||
file_size=os.path.getsize(file_path),
|
||||
file_hash=file_hash,
|
||||
width=model_data['width'],
|
||||
height=model_data['height'],
|
||||
depth=model_data['depth'],
|
||||
faces=model_data['faces'],
|
||||
thumbnail_path=thumbnail_path,
|
||||
)
|
||||
db.add(db_model)
|
||||
db.flush()
|
||||
|
||||
db_file = ModelFile(
|
||||
model_id=db_model.id,
|
||||
filename=os.path.basename(file_path),
|
||||
file_path=file_path,
|
||||
file_type='stl' if file_path.lower().endswith('.stl') else '3mf',
|
||||
part_name=part_name,
|
||||
is_primary=is_primary,
|
||||
file_size=os.path.getsize(file_path),
|
||||
file_hash=file_hash,
|
||||
)
|
||||
db.add(db_file)
|
||||
|
||||
tag_names = [t.strip() for t in (tags_str or '').split(',') if t.strip()]
|
||||
db_model.tags = _get_or_create_tags(db, tag_names)
|
||||
db.commit()
|
||||
db.refresh(db_model)
|
||||
return db_model
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# STATIC ROUTES (no path parameters) - MUST come before dynamic routes
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/", response_model=List[Model3DResponse])
|
||||
def list_models(
|
||||
search: Optional[str] = None, category: Optional[str] = None, tag: Optional[str] = None,
|
||||
min_width: Optional[float] = None, max_width: Optional[float] = None,
|
||||
min_height: Optional[float] = None, max_height: Optional[float] = None,
|
||||
min_depth: Optional[float] = None, max_depth: Optional[float] = None,
|
||||
min_faces: Optional[int] = None, max_faces: Optional[int] = None,
|
||||
date_from: Optional[str] = None, date_to: Optional[str] = None,
|
||||
sort_by: Optional[str] = Query("newest", enum=["newest", "oldest", "most_downloaded", "largest", "most_faces", "highest_rated"]),
|
||||
skip: int = Query(0, ge=0), limit: int = Query(24, ge=1, le=100),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
query = db.query(Model3D)
|
||||
if search: query = query.filter(Model3D.title.contains(search))
|
||||
if category: query = query.filter(Model3D.category == category)
|
||||
if tag: query = query.join(Model3D.tags).filter(Tag.name.contains(tag.lower()))
|
||||
if min_width: query = query.filter(Model3D.width >= min_width)
|
||||
if max_width: query = query.filter(Model3D.width <= max_width)
|
||||
if min_height: query = query.filter(Model3D.height >= min_height)
|
||||
if max_height: query = query.filter(Model3D.height <= max_height)
|
||||
if min_depth: query = query.filter(Model3D.depth >= min_depth)
|
||||
if max_depth: query = query.filter(Model3D.depth <= max_depth)
|
||||
if min_faces: query = query.filter(Model3D.faces >= min_faces)
|
||||
if max_faces: query = query.filter(Model3D.faces <= max_faces)
|
||||
if date_from: query = query.filter(Model3D.created_at >= date_from)
|
||||
if date_to: query = query.filter(Model3D.created_at <= date_to)
|
||||
|
||||
if sort_by == "newest": query = query.order_by(Model3D.created_at.desc())
|
||||
elif sort_by == "oldest": query = query.order_by(Model3D.created_at.asc())
|
||||
elif sort_by == "most_downloaded": query = query.order_by(Model3D.download_count.desc())
|
||||
elif sort_by == "largest": query = query.order_by(Model3D.file_size.desc())
|
||||
elif sort_by == "most_faces": query = query.order_by(Model3D.faces.desc())
|
||||
elif sort_by == "highest_rated":
|
||||
models = query.all()
|
||||
models.sort(key=lambda m: _calc_avg_rating(m) or 0, reverse=True)
|
||||
return models[skip:skip+limit]
|
||||
|
||||
models = query.offset(skip).limit(limit).all()
|
||||
return JSONResponse(content=[Model3DResponse.model_validate(m).model_dump(mode='json') for m in models], headers=_cache_headers(60))
|
||||
|
||||
|
||||
@router.get("/tags", response_model=List[dict])
|
||||
def list_tags(search: Optional[str] = None, db: Session = Depends(get_db)):
|
||||
query = db.query(Tag)
|
||||
if search: query = query.filter(Tag.name.contains(search.lower()))
|
||||
tags = query.order_by(Tag.name).all()
|
||||
data = [{"id": t.id, "name": t.name, "count": len(t.models)} for t in tags]
|
||||
return JSONResponse(content=data, headers=_cache_headers(300))
|
||||
|
||||
|
||||
@router.get("/collections/all", response_model=List[CollectionResponse])
|
||||
def list_collections(db: Session = Depends(get_db)):
|
||||
collections = db.query(Collection).order_by(Collection.created_at.desc()).all()
|
||||
result = []
|
||||
for c in collections:
|
||||
resp = CollectionResponse.model_validate(c)
|
||||
resp.model_count = len(c.models)
|
||||
result.append(resp)
|
||||
return result
|
||||
|
||||
|
||||
@router.post("/collections", response_model=CollectionResponse)
|
||||
def create_collection(data: CollectionCreate, db: Session = Depends(get_db)):
|
||||
coll = Collection(name=data.name, description=data.description)
|
||||
db.add(coll)
|
||||
db.commit()
|
||||
db.refresh(coll)
|
||||
resp = CollectionResponse.model_validate(coll)
|
||||
resp.model_count = 0
|
||||
return resp
|
||||
|
||||
|
||||
@router.get("/system/backup")
|
||||
def backup_all(db: Session = Depends(get_db)):
|
||||
from datetime import datetime
|
||||
import glob
|
||||
zip_buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
db_path = "stl_repo.db"
|
||||
if os.path.exists(db_path): zf.write(db_path, arcname="backup/stl_repo.db")
|
||||
for f in glob.glob("uploads/*"):
|
||||
if os.path.isfile(f): zf.write(f, arcname=f"backup/{f}")
|
||||
for f in glob.glob("thumbnails/*"):
|
||||
if os.path.isfile(f): zf.write(f, arcname=f"backup/{f}")
|
||||
for f in glob.glob("images/*"):
|
||||
if os.path.isfile(f): zf.write(f, arcname=f"backup/{f}")
|
||||
models = db.query(Model3D).all()
|
||||
metadata = []
|
||||
for m in models:
|
||||
metadata.append({
|
||||
"id": m.id, "title": m.title, "filename": m.filename,
|
||||
"description": m.description, "author": m.author, "license": m.license,
|
||||
"category": m.category, "tags": [t.name for t in m.tags],
|
||||
"faces": m.faces, "width": m.width, "height": m.height, "depth": m.depth,
|
||||
"created_at": m.created_at.isoformat() if m.created_at else None,
|
||||
"download_count": m.download_count,
|
||||
"files": [{"filename": f.filename, "file_type": f.file_type, "part_name": f.part_name, "is_primary": f.is_primary} for f in m.files]
|
||||
})
|
||||
zf.writestr("backup/metadata.json", json.dumps(metadata, indent=2, default=str))
|
||||
zip_buffer.seek(0)
|
||||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
return StreamingResponse(zip_buffer, media_type="application/zip",
|
||||
headers={"Content-Disposition": f"attachment; filename=stl_repo_backup_{timestamp}.zip"})
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# MODEL CREATION
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/", response_model=Model3DResponse)
|
||||
async def create_model(
|
||||
title: str = Form(...),
|
||||
description: Optional[str] = Form(None),
|
||||
author: Optional[str] = Form(None),
|
||||
license: Optional[str] = Form(None),
|
||||
tags: Optional[str] = Form(None),
|
||||
category: Optional[str] = Form(None),
|
||||
files: List[UploadFile] = File(default=[]),
|
||||
images: List[UploadFile] = File(default=[]),
|
||||
part_names: Optional[str] = Form(None),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
if not files:
|
||||
raise HTTPException(status_code=400, detail="At least one 3D file is required")
|
||||
|
||||
part_names_map = {}
|
||||
if part_names:
|
||||
try:
|
||||
part_names_map = json.loads(part_names)
|
||||
except:
|
||||
pass
|
||||
|
||||
saved_3d_files = []
|
||||
total_faces = 0
|
||||
|
||||
for idx, file in enumerate(files):
|
||||
if not file.filename:
|
||||
continue
|
||||
lower_name = file.filename.lower()
|
||||
if not (lower_name.endswith('.stl') or lower_name.endswith('.3mf')):
|
||||
for sf in saved_3d_files:
|
||||
if os.path.exists(sf['path']): os.remove(sf['path'])
|
||||
raise HTTPException(status_code=400, detail=f"Unsupported format: {file.filename}")
|
||||
|
||||
file_path = os.path.join(UPLOAD_DIR, file.filename)
|
||||
with open(file_path, "wb") as buffer:
|
||||
shutil.copyfileobj(file.file, buffer)
|
||||
|
||||
file_hash = _file_hash(file_path)
|
||||
existing = db.query(Model3D).filter(Model3D.file_hash == file_hash).first()
|
||||
if existing:
|
||||
os.remove(file_path)
|
||||
for sf in saved_3d_files:
|
||||
if os.path.exists(sf['path']): os.remove(sf['path'])
|
||||
raise HTTPException(status_code=409,
|
||||
detail=f"File '{file.filename}' already exists as '{existing.title}' (ID: {existing.id})")
|
||||
|
||||
try:
|
||||
model_data = parse_model_file(file_path)
|
||||
except Exception as e:
|
||||
os.remove(file_path)
|
||||
for sf in saved_3d_files:
|
||||
if os.path.exists(sf['path']): os.remove(sf['path'])
|
||||
raise HTTPException(status_code=400, detail=f"Failed to parse {file.filename}: {str(e)}")
|
||||
|
||||
saved_3d_files.append({
|
||||
'path': file_path, 'filename': file.filename, 'hash': file_hash,
|
||||
'size': os.path.getsize(file_path), 'data': model_data,
|
||||
'is_primary': idx == 0, 'part_name': part_names_map.get(str(idx), None),
|
||||
})
|
||||
if idx == 0:
|
||||
total_faces += model_data['faces']
|
||||
|
||||
if not saved_3d_files:
|
||||
raise HTTPException(status_code=400, detail="No valid 3D files provided")
|
||||
|
||||
primary = saved_3d_files[0]
|
||||
thumbnail_path = os.path.join(THUMBNAIL_DIR, f"{os.path.splitext(primary['filename'])[0]}.webp")
|
||||
generate_thumbnail(primary['data']['vertices'], thumbnail_path)
|
||||
|
||||
db_model = Model3D(
|
||||
title=title, filename=primary['filename'], description=description,
|
||||
author=author, license=license, category=category,
|
||||
file_size=sum(f['size'] for f in saved_3d_files),
|
||||
file_hash=primary['hash'], width=primary['data']['width'],
|
||||
height=primary['data']['height'], depth=primary['data']['depth'],
|
||||
faces=total_faces, thumbnail_path=thumbnail_path,
|
||||
)
|
||||
db.add(db_model)
|
||||
db.flush()
|
||||
|
||||
for sf in saved_3d_files:
|
||||
db_file = ModelFile(
|
||||
model_id=db_model.id, filename=sf['filename'], file_path=sf['path'],
|
||||
file_type='stl' if sf['filename'].lower().endswith('.stl') else '3mf',
|
||||
part_name=sf['part_name'], is_primary=sf['is_primary'],
|
||||
file_size=sf['size'], file_hash=sf['hash'],
|
||||
)
|
||||
db.add(db_file)
|
||||
|
||||
for img in images:
|
||||
if not img.filename: continue
|
||||
lower_name = img.filename.lower()
|
||||
if not (lower_name.endswith('.jpg') or lower_name.endswith('.jpeg') or lower_name.endswith('.png')):
|
||||
continue
|
||||
img_path = os.path.join(IMAGES_DIR, img.filename)
|
||||
with open(img_path, "wb") as buffer:
|
||||
shutil.copyfileobj(img.file, buffer)
|
||||
db.add(ModelFile(
|
||||
model_id=db_model.id, filename=img.filename, file_path=img_path,
|
||||
file_type='image', is_primary=False, file_size=os.path.getsize(img_path),
|
||||
))
|
||||
|
||||
tag_names = [t.strip() for t in (tags or '').split(',') if t.strip()]
|
||||
db_model.tags = _get_or_create_tags(db, tag_names)
|
||||
db.commit()
|
||||
db.refresh(db_model)
|
||||
response_data = Model3DResponse.model_validate(db_model)
|
||||
response_data.avg_rating = _calc_avg_rating(db_model)
|
||||
return response_data
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# IMPORT / BULK / BATCH
|
||||
# =============================================================================
|
||||
|
||||
@router.post("/import-url", response_model=Model3DResponse)
|
||||
async def import_from_url(
|
||||
url: str = Form(...),
|
||||
title: str = Form(...),
|
||||
description: Optional[str] = Form(None),
|
||||
author: Optional[str] = Form(None),
|
||||
license: Optional[str] = Form(None),
|
||||
tags: Optional[str] = Form(None),
|
||||
category: Optional[str] = Form(None),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
MAX_SIZE = 50 * 1024 * 1024 # 50MB
|
||||
try:
|
||||
parsed = urllib.parse.urlparse(url)
|
||||
if not parsed.scheme or not parsed.netloc:
|
||||
raise HTTPException(status_code=400, detail="Invalid URL")
|
||||
except Exception:
|
||||
raise HTTPException(status_code=400, detail="Invalid URL")
|
||||
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
|
||||
r = await client.get(url)
|
||||
r.raise_for_status()
|
||||
if len(r.content) > MAX_SIZE:
|
||||
raise HTTPException(status_code=400, detail="File too large (max 50MB)")
|
||||
except httpx.RequestError as e:
|
||||
raise HTTPException(status_code=400, detail=f"Download failed: {str(e)}")
|
||||
|
||||
filename = os.path.basename(parsed.path) or "downloaded.stl"
|
||||
if not filename.lower().endswith(('.stl', '.3mf')):
|
||||
filename += ".stl"
|
||||
|
||||
file_path = os.path.join(UPLOAD_DIR, filename)
|
||||
with open(file_path, 'wb') as f:
|
||||
f.write(r.content)
|
||||
|
||||
db_model = _process_single_file(db, file_path, title, description, author, license, tags, category, is_primary=True)
|
||||
response_data = Model3DResponse.model_validate(db_model)
|
||||
response_data.avg_rating = _calc_avg_rating(db_model)
|
||||
return response_data
|
||||
|
||||
|
||||
@router.post("/bulk-zip", response_model=List[Model3DResponse])
|
||||
async def bulk_zip_upload(
|
||||
zip_file: UploadFile = File(...),
|
||||
description: Optional[str] = Form(None),
|
||||
author: Optional[str] = Form(None),
|
||||
license: Optional[str] = Form(None),
|
||||
tags: Optional[str] = Form(None),
|
||||
category: Optional[str] = Form(None),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
if not zip_file.filename or not zip_file.filename.lower().endswith('.zip'):
|
||||
raise HTTPException(status_code=400, detail="Only ZIP files are allowed")
|
||||
|
||||
tmp_path = os.path.join(tempfile.gettempdir(), zip_file.filename)
|
||||
with open(tmp_path, 'wb') as f:
|
||||
shutil.copyfileobj(zip_file.file, f)
|
||||
|
||||
extract_dir = tempfile.mkdtemp()
|
||||
try:
|
||||
with zipfile.ZipFile(tmp_path, 'r') as zf:
|
||||
zf.extractall(extract_dir)
|
||||
except zipfile.BadZipFile:
|
||||
os.remove(tmp_path)
|
||||
raise HTTPException(status_code=400, detail="Invalid ZIP file")
|
||||
|
||||
results = []
|
||||
for root, dirs, files in os.walk(extract_dir):
|
||||
for fname in files:
|
||||
lower = fname.lower()
|
||||
if lower.endswith('.stl') or lower.endswith('.3mf'):
|
||||
src = os.path.join(root, fname)
|
||||
dst = os.path.join(UPLOAD_DIR, fname)
|
||||
# avoid collisions
|
||||
if os.path.exists(dst):
|
||||
base, ext = os.path.splitext(fname)
|
||||
fname = f"{base}_{hashlib.md5(src.encode()).hexdigest()[:6]}{ext}"
|
||||
dst = os.path.join(UPLOAD_DIR, fname)
|
||||
shutil.move(src, dst)
|
||||
try:
|
||||
db_model = _process_single_file(
|
||||
db, dst, os.path.splitext(fname)[0],
|
||||
description, author, license, tags, category, is_primary=True
|
||||
)
|
||||
response_data = Model3DResponse.model_validate(db_model)
|
||||
response_data.avg_rating = _calc_avg_rating(db_model)
|
||||
results.append(response_data)
|
||||
except HTTPException:
|
||||
pass # skip duplicates
|
||||
|
||||
os.remove(tmp_path)
|
||||
shutil.rmtree(extract_dir, ignore_errors=True)
|
||||
return results
|
||||
|
||||
|
||||
@router.post("/batch-download")
|
||||
def batch_download(
|
||||
model_ids: List[int] = Body(...),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
if not model_ids:
|
||||
raise HTTPException(status_code=400, detail="No model IDs provided")
|
||||
|
||||
zip_buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
for mid in model_ids:
|
||||
model = db.query(Model3D).filter(Model3D.id == mid).first()
|
||||
if not model:
|
||||
continue
|
||||
for mf in model.files:
|
||||
if mf.file_type in ('stl', '3mf') and os.path.exists(mf.file_path):
|
||||
zf.write(mf.file_path, arcname=f"{model.title.replace(' ', '_')}/{mf.filename}")
|
||||
|
||||
zip_buffer.seek(0)
|
||||
return StreamingResponse(
|
||||
zip_buffer, media_type="application/zip",
|
||||
headers={"Content-Disposition": "attachment; filename=batch_download.zip"}
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DYNAMIC MODEL ROUTES
|
||||
# =============================================================================
|
||||
|
||||
@router.get("/{model_id}", response_model=Model3DResponse)
|
||||
def get_model(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
response_data = Model3DResponse.model_validate(model)
|
||||
response_data.avg_rating = _calc_avg_rating(model)
|
||||
return response_data
|
||||
|
||||
|
||||
@router.put("/{model_id}", response_model=Model3DResponse)
|
||||
def update_model(model_id: int, data: Model3DUpdate, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
if data.title is not None: model.title = data.title
|
||||
if data.description is not None: model.description = data.description
|
||||
if data.author is not None: model.author = data.author
|
||||
if data.license is not None: model.license = data.license
|
||||
if data.category is not None: model.category = data.category
|
||||
if data.tag_names is not None: model.tags = _get_or_create_tags(db, data.tag_names)
|
||||
db.commit()
|
||||
db.refresh(model)
|
||||
response_data = Model3DResponse.model_validate(model)
|
||||
response_data.avg_rating = _calc_avg_rating(model)
|
||||
return response_data
|
||||
|
||||
|
||||
@router.delete("/{model_id}")
|
||||
def delete_model(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
for mf in model.files:
|
||||
if os.path.exists(mf.file_path): os.remove(mf.file_path)
|
||||
if model.thumbnail_path and os.path.exists(model.thumbnail_path): os.remove(model.thumbnail_path)
|
||||
db.delete(model)
|
||||
db.commit()
|
||||
return {"detail": "Model deleted"}
|
||||
|
||||
|
||||
# --- Validation & Estimation ---
|
||||
|
||||
@router.get("/{model_id}/validate")
|
||||
def validate_mesh(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model:
|
||||
raise HTTPException(status_code=404, detail="Model not found")
|
||||
|
||||
primary = next((f for f in model.files if f.is_primary and f.file_type in ('stl', '3mf')), None)
|
||||
if not primary:
|
||||
raise HTTPException(status_code=404, detail="No 3D file found")
|
||||
|
||||
try:
|
||||
import trimesh
|
||||
mesh = trimesh.load(primary.file_path, force='mesh')
|
||||
volume = abs(float(mesh.volume)) if mesh.is_watertight else 0.0
|
||||
area = float(mesh.area)
|
||||
is_watertight = bool(mesh.is_watertight)
|
||||
is_winding_consistent = bool(mesh.is_winding_consistent)
|
||||
bounds = mesh.bounds.tolist()
|
||||
euler = int(mesh.euler_number)
|
||||
|
||||
# Count holes via euler characteristic
|
||||
holes = 0 if is_watertight else max(0, 2 - euler)
|
||||
|
||||
return {
|
||||
"is_watertight": is_watertight,
|
||||
"is_winding_consistent": is_winding_consistent,
|
||||
"volume_cm3": round(volume / 1000, 3) if volume else 0,
|
||||
"surface_area_cm2": round(area / 100, 2),
|
||||
"bounds_mm": bounds,
|
||||
"euler_number": euler,
|
||||
"estimated_holes": holes,
|
||||
"face_count": len(mesh.faces),
|
||||
"vertex_count": len(mesh.vertices),
|
||||
}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Validation failed: {str(e)}")
|
||||
|
||||
|
||||
@router.get("/{model_id}/estimate")
|
||||
def estimate_print(model_id: int, price_per_kg: float = Query(20.0), material_density: float = Query(1.24), db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model:
|
||||
raise HTTPException(status_code=404, detail="Model not found")
|
||||
|
||||
primary = next((f for f in model.files if f.is_primary and f.file_type in ('stl', '3mf')), None)
|
||||
if not primary:
|
||||
raise HTTPException(status_code=404, detail="No 3D file found")
|
||||
|
||||
try:
|
||||
import trimesh
|
||||
mesh = trimesh.load(primary.file_path, force='mesh')
|
||||
volume_cm3 = abs(float(mesh.volume)) / 1000.0
|
||||
grams = volume_cm3 * material_density
|
||||
cost = (grams / 1000.0) * price_per_kg
|
||||
seconds = _estimate_print_time_seconds(volume_cm3)
|
||||
hours = seconds // 3600
|
||||
mins = (seconds % 3600) // 60
|
||||
|
||||
return {
|
||||
"volume_cm3": round(volume_cm3, 2),
|
||||
"grams": round(grams, 2),
|
||||
"cost": round(cost, 2),
|
||||
"estimated_time": f"{hours}h {mins}m",
|
||||
"estimated_seconds": seconds,
|
||||
"price_per_kg": price_per_kg,
|
||||
"material_density": material_density,
|
||||
}
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Estimation failed: {str(e)}")
|
||||
|
||||
|
||||
# --- Ratings ---
|
||||
|
||||
@router.post("/{model_id}/ratings", response_model=RatingResponse)
|
||||
def create_rating(model_id: int, stars: int = Query(..., ge=1, le=5), db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
rating = Rating(model_id=model_id, stars=stars)
|
||||
db.add(rating)
|
||||
db.commit()
|
||||
db.refresh(rating)
|
||||
return rating
|
||||
|
||||
|
||||
@router.get("/{model_id}/ratings", response_model=List[RatingResponse])
|
||||
def list_ratings(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
return db.query(Rating).filter(Rating.model_id == model_id).order_by(Rating.created_at.desc()).all()
|
||||
|
||||
|
||||
# --- Comments ---
|
||||
|
||||
@router.post("/{model_id}/comments", response_model=CommentResponse)
|
||||
def create_comment(model_id: int, text: str = Query(..., min_length=1), author_name: Optional[str] = Query(None), db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
comment = Comment(model_id=model_id, text=text, author_name=author_name)
|
||||
db.add(comment)
|
||||
db.commit()
|
||||
db.refresh(comment)
|
||||
return comment
|
||||
|
||||
|
||||
@router.get("/{model_id}/comments", response_model=List[CommentResponse])
|
||||
def list_comments(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
return db.query(Comment).filter(Comment.model_id == model_id).order_by(Comment.created_at.desc()).all()
|
||||
|
||||
|
||||
# --- Collections (dynamic) ---
|
||||
|
||||
@router.get("/collections/{collection_id}", response_model=CollectionDetailResponse)
|
||||
def get_collection(collection_id: int, db: Session = Depends(get_db)):
|
||||
coll = db.query(Collection).filter(Collection.id == collection_id).first()
|
||||
if not coll: raise HTTPException(status_code=404, detail="Collection not found")
|
||||
resp = CollectionDetailResponse.model_validate(coll)
|
||||
resp.model_count = len(coll.models)
|
||||
return resp
|
||||
|
||||
|
||||
@router.post("/collections/{collection_id}/add/{model_id}")
|
||||
def add_to_collection(collection_id: int, model_id: int, db: Session = Depends(get_db)):
|
||||
coll = db.query(Collection).filter(Collection.id == collection_id).first()
|
||||
if not coll: raise HTTPException(status_code=404, detail="Collection not found")
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
if model not in coll.models:
|
||||
coll.models.append(model)
|
||||
db.commit()
|
||||
return {"detail": "Model added to collection"}
|
||||
|
||||
|
||||
@router.delete("/collections/{collection_id}/remove/{model_id}")
|
||||
def remove_from_collection(collection_id: int, model_id: int, db: Session = Depends(get_db)):
|
||||
coll = db.query(Collection).filter(Collection.id == collection_id).first()
|
||||
if not coll: raise HTTPException(status_code=404, detail="Collection not found")
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
if model in coll.models:
|
||||
coll.models.remove(model)
|
||||
db.commit()
|
||||
return {"detail": "Model removed from collection"}
|
||||
|
||||
|
||||
@router.delete("/collections/{collection_id}")
|
||||
def delete_collection(collection_id: int, db: Session = Depends(get_db)):
|
||||
coll = db.query(Collection).filter(Collection.id == collection_id).first()
|
||||
if not coll: raise HTTPException(status_code=404, detail="Collection not found")
|
||||
db.delete(coll)
|
||||
db.commit()
|
||||
return {"detail": "Collection deleted"}
|
||||
|
||||
|
||||
# --- Downloads & QR ---
|
||||
|
||||
@router.get("/{model_id}/download")
|
||||
def download_model(model_id: int, file_id: Optional[int] = None, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
if file_id:
|
||||
mf = db.query(ModelFile).filter(ModelFile.id == file_id, ModelFile.model_id == model_id).first()
|
||||
if not mf: raise HTTPException(status_code=404, detail="File not found")
|
||||
if not os.path.exists(mf.file_path): raise HTTPException(status_code=404, detail="File not found on disk")
|
||||
model.download_count += 1; db.commit()
|
||||
return FileResponse(path=mf.file_path, filename=mf.filename, media_type="application/octet-stream")
|
||||
primary = next((f for f in model.files if f.is_primary and f.file_type in ('stl', '3mf')), None)
|
||||
file_path = primary.file_path if primary else os.path.join(UPLOAD_DIR, model.filename)
|
||||
if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found on disk")
|
||||
model.download_count += 1; db.commit()
|
||||
return FileResponse(path=file_path, filename=model.filename, media_type="application/octet-stream")
|
||||
|
||||
|
||||
@router.get("/{model_id}/download-all")
|
||||
def download_all(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
model_files = [f for f in model.files if f.file_type in ('stl', '3mf')]
|
||||
if not model_files: raise HTTPException(status_code=404, detail="No 3D files found")
|
||||
zip_buffer = io.BytesIO()
|
||||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
|
||||
for mf in model_files:
|
||||
if os.path.exists(mf.file_path): zf.write(mf.file_path, arcname=mf.filename)
|
||||
zip_buffer.seek(0)
|
||||
model.download_count += 1; db.commit()
|
||||
return StreamingResponse(zip_buffer, media_type="application/zip",
|
||||
headers={"Content-Disposition": f"attachment; filename={model.title.replace(' ', '_')}_all_parts.zip"})
|
||||
|
||||
|
||||
@router.get("/{model_id}/thumbnail")
|
||||
def get_thumbnail(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
# Try WEBP first, fallback to PNG
|
||||
if model.thumbnail_path:
|
||||
if os.path.exists(model.thumbnail_path):
|
||||
media_type = "image/webp" if model.thumbnail_path.lower().endswith('.webp') else "image/png"
|
||||
return FileResponse(path=model.thumbnail_path, media_type=media_type, headers=_cache_headers(86400))
|
||||
# Fallback to PNG with same base name
|
||||
png_path = os.path.splitext(model.thumbnail_path)[0] + '.png'
|
||||
if os.path.exists(png_path):
|
||||
return FileResponse(path=png_path, media_type="image/png", headers=_cache_headers(86400))
|
||||
raise HTTPException(status_code=404, detail="Thumbnail not found")
|
||||
|
||||
|
||||
@router.get("/{model_id}/qr")
|
||||
def get_qr(model_id: int, db: Session = Depends(get_db)):
|
||||
model = db.query(Model3D).filter(Model3D.id == model_id).first()
|
||||
if not model: raise HTTPException(status_code=404, detail="Model not found")
|
||||
url = f"http://192.168.10.104:8000/model/{model_id}"
|
||||
qr = qrcode.QRCode(version=1, box_size=10, border=2)
|
||||
qr.add_data(url); qr.make(fit=True)
|
||||
img = qr.make_image(fill_color="#0f172a", back_color="#ffffff")
|
||||
buf = io.BytesIO(); img.save(buf, format='PNG'); buf.seek(0)
|
||||
return StreamingResponse(buf, media_type="image/png")
|
||||
Reference in New Issue
Block a user