132 lines
3.9 KiB
Python
132 lines
3.9 KiB
Python
import os
|
|
import re
|
|
import shutil
|
|
from pathlib import Path
|
|
from typing import List
|
|
|
|
from fastapi import FastAPI, HTTPException, UploadFile, File
|
|
from fastapi.responses import FileResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
PROJECTS_DIR = Path(os.environ.get("PROJECTS_DIR", "/data/projects")).resolve()
|
|
PROJECTS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
ALLOWED_EXTENSIONS = {".stl", ".3mf"}
|
|
NAME_PATTERN = re.compile(r"^[A-Za-z0-9 _\-]+$")
|
|
|
|
app = FastAPI(title="3D Project Manager")
|
|
|
|
|
|
def safe_project_path(name: str) -> Path:
|
|
"""Validate a project name and return its resolved path, guarding
|
|
against path traversal / invalid characters."""
|
|
if not name or not NAME_PATTERN.match(name):
|
|
raise HTTPException(400, "Invalid project name")
|
|
path = (PROJECTS_DIR / name).resolve()
|
|
if path.parent != PROJECTS_DIR:
|
|
raise HTTPException(400, "Invalid project name")
|
|
return path
|
|
|
|
|
|
def safe_filename(filename: str) -> str:
|
|
name = Path(filename).name
|
|
if not name or name in (".", ".."):
|
|
raise HTTPException(400, "Invalid file name")
|
|
return name
|
|
|
|
|
|
class ProjectCreate(BaseModel):
|
|
name: str
|
|
|
|
|
|
@app.get("/api/projects")
|
|
def list_projects():
|
|
projects = []
|
|
for entry in sorted(PROJECTS_DIR.iterdir()):
|
|
if entry.is_dir():
|
|
file_count = sum(
|
|
1 for f in entry.iterdir()
|
|
if f.is_file() and f.suffix.lower() in ALLOWED_EXTENSIONS
|
|
)
|
|
projects.append({"name": entry.name, "file_count": file_count})
|
|
return projects
|
|
|
|
|
|
@app.post("/api/projects")
|
|
def create_project(payload: ProjectCreate):
|
|
path = safe_project_path(payload.name)
|
|
if path.exists():
|
|
raise HTTPException(409, "A project with this name already exists")
|
|
path.mkdir(parents=True)
|
|
return {"name": payload.name}
|
|
|
|
|
|
@app.delete("/api/projects/{name}")
|
|
def delete_project(name: str):
|
|
path = safe_project_path(name)
|
|
if not path.exists():
|
|
raise HTTPException(404, "Project not found")
|
|
shutil.rmtree(path)
|
|
return {"status": "deleted"}
|
|
|
|
|
|
@app.get("/api/projects/{name}/files")
|
|
def list_files(name: str):
|
|
path = safe_project_path(name)
|
|
if not path.exists():
|
|
raise HTTPException(404, "Project not found")
|
|
files = []
|
|
for f in sorted(path.iterdir()):
|
|
if f.is_file() and f.suffix.lower() in ALLOWED_EXTENSIONS:
|
|
files.append({
|
|
"name": f.name,
|
|
"size": f.stat().st_size,
|
|
"type": f.suffix.lower().lstrip("."),
|
|
})
|
|
return files
|
|
|
|
|
|
@app.post("/api/projects/{name}/upload")
|
|
async def upload_files(name: str, files: List[UploadFile] = File(...)):
|
|
path = safe_project_path(name)
|
|
if not path.exists():
|
|
raise HTTPException(404, "Project not found")
|
|
|
|
saved, skipped = [], []
|
|
for upload in files:
|
|
ext = Path(upload.filename).suffix.lower()
|
|
if ext not in ALLOWED_EXTENSIONS:
|
|
skipped.append(upload.filename)
|
|
continue
|
|
filename = safe_filename(upload.filename)
|
|
dest = path / filename
|
|
with open(dest, "wb") as out:
|
|
shutil.copyfileobj(upload.file, out)
|
|
saved.append(filename)
|
|
|
|
return {"saved": saved, "skipped": skipped}
|
|
|
|
|
|
@app.get("/api/projects/{name}/files/{filename}")
|
|
def get_file(name: str, filename: str):
|
|
path = safe_project_path(name)
|
|
target = path / safe_filename(filename)
|
|
if not target.exists():
|
|
raise HTTPException(404, "File not found")
|
|
return FileResponse(target)
|
|
|
|
|
|
@app.delete("/api/projects/{name}/files/{filename}")
|
|
def delete_file(name: str, filename: str):
|
|
path = safe_project_path(name)
|
|
target = path / safe_filename(filename)
|
|
if not target.exists():
|
|
raise HTTPException(404, "File not found")
|
|
target.unlink()
|
|
return {"status": "deleted"}
|
|
|
|
|
|
# Serve the frontend last, so it doesn't shadow the /api routes above.
|
|
app.mount("/", StaticFiles(directory="/app/static", html=True), name="static")
|