Files
SonicForgeStudio/app/api/v1/plugins.py
T
3dtours 2ad29226c4 fix: deduplicate soundfont scan + source label + delete only uploads
- _scan_soundfonts: dict-keyed by base_id to prevent duplicates
- Add 'source' field ('system' | 'upload') to each soundfont entry
- Delete endpoint: only allow deleting upload soundfonts (403 for system)
- Frontend: show (system)/(upload) tag, hide Delete for system fonts
2026-07-27 09:36:58 +07:00

162 lines
6.3 KiB
Python

import os, uuid, json, tempfile
from fastapi import APIRouter, HTTPException, Depends, UploadFile, File, BackgroundTasks
from fastapi.responses import FileResponse
from pydantic import BaseModel
from typing import Optional, Any
from app.config import settings
from app.core.vst_engine import PluginManager, HAS_PEDALBOARD, HAS_PYFLUIDSYNTH
from app.core.render_engine import PythonRenderEngine
from app.core.soundfont_inspector import SoundFontInspector
from app.core.soundfont_converter import SoundFontConverter
from app.api.v1.auth import get_current_user
router = APIRouter()
UPLOAD_SF_DIR = os.path.join(settings.STORAGE_DIR, "soundfonts")
os.makedirs(UPLOAD_SF_DIR, exist_ok=True)
SYSTEM_SF_DIR = "/opt/daw_engine/soundfonts"
_inspector = None
def get_inspector():
global _inspector
if _inspector is None:
_inspector = SoundFontInspector(system_sf_dir=SYSTEM_SF_DIR, upload_sf_dir=UPLOAD_SF_DIR)
return _inspector
@router.get("/available")
async def list_plugins(current_user: dict = Depends(get_current_user)):
pm = PluginManager(upload_sf_dir=UPLOAD_SF_DIR)
return pm.list_available()
@router.get("/default-soundfonts")
async def list_default_soundfonts():
static_sf_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "static", "soundfonts")
result = []
if os.path.isdir(static_sf_dir):
for f in os.listdir(static_sf_dir):
if f.endswith(".sf2") or f.endswith(".sf3"):
result.append({
"id": os.path.splitext(f)[0],
"name": f,
"file": f,
"url": f"/soundfonts/{f}"
})
return result
@router.get("/soundfonts/catalog")
async def soundfont_catalog(current_user: dict = Depends(get_current_user)):
inspector = get_inspector()
full_catalog = inspector.get_catalog()
condensed_catalog = inspector.get_condensed_catalog_summary()
return {"full_catalog": full_catalog, "condensed_catalog": condensed_catalog}
@router.get("/soundfont-instruments/{sf_id}")
async def list_soundfont_instruments(sf_id: str, current_user: dict = Depends(get_current_user)):
pm = PluginManager(upload_sf_dir=UPLOAD_SF_DIR)
presets = pm.list_soundfont_instruments(sf_id)
return {"presets": presets, "count": len(presets)}
@router.post("/upload-soundfont")
async def upload_soundfont(
file: UploadFile = File(...),
background_tasks: BackgroundTasks = None,
current_user: dict = Depends(get_current_user)
):
if not (file.filename and (file.filename.endswith(".sf2") or file.filename.endswith(".sf3"))):
raise HTTPException(status_code=400, detail="Only .sf2 / .sf3 files are allowed")
contents = await file.read()
if not PluginManager.validate_sf2_header(contents):
raise HTTPException(status_code=400, detail="Invalid SoundFont file: missing RIFF/sfbk header")
file_ext = os.path.splitext(file.filename)[1]
file_uuid = str(uuid.uuid4())
# Store original name in a sidecar file
base_name = os.path.splitext(file.filename)[0].replace('/', '_').replace('\\', '_')
file_id = file_uuid + file_ext
dest_path = os.path.join(UPLOAD_SF_DIR, file_id)
with open(dest_path, "wb") as f:
f.write(contents)
# Save metadata with original name
meta_path = os.path.join(UPLOAD_SF_DIR, file_uuid + ".meta")
with open(meta_path, "w", encoding="utf-8") as f:
import json
json.dump({"original_name": file.filename, "uuid": file_uuid, "file": file_id}, f)
inspector = get_inspector()
inspector.invalidate_catalog_cache()
if background_tasks:
background_tasks.add_task(inspector.generate_full_catalog)
return {"id": file_id, "name": file.filename, "path": dest_path, "size_bytes": len(contents)}
@router.delete("/soundfont/{sf_id}")
async def delete_soundfont(sf_id: str, current_user: dict = Depends(get_current_user)):
base_id = sf_id.replace("sf_", "")
deleted = False
for d in [UPLOAD_SF_DIR, os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "static", "soundfonts"), SYSTEM_SF_DIR]:
if not os.path.isdir(d):
continue
for f in os.listdir(d):
if os.path.splitext(f)[0] == base_id:
# Skip system dir — only allow deleting uploads
if d == SYSTEM_SF_DIR:
raise HTTPException(status_code=403, detail="System soundfonts cannot be deleted via this endpoint")
path = os.path.join(d, f)
os.remove(path)
meta_path = os.path.join(d, os.path.splitext(f)[0] + ".meta")
if os.path.isfile(meta_path):
os.remove(meta_path)
deleted = True
break
if deleted:
break
if not deleted:
raise HTTPException(status_code=404, detail="SoundFont not found")
inspector = get_inspector()
inspector.invalidate_catalog_cache()
return {"deleted": True, "sf_id": sf_id}
@router.get("/soundfonts/download/{sf_id}")
async def download_soundfont_asset(sf_id: str):
clean_id = sf_id.replace("sf_", "") if sf_id.startswith("sf_") else sf_id
for base_dir in [UPLOAD_SF_DIR, SYSTEM_SF_DIR]:
if not os.path.isdir(base_dir):
continue
for ext in [".sf2", ".sf3"]:
for fname in os.listdir(base_dir):
fbase, fext = os.path.splitext(fname)
if fext.lower() == ext and fbase.lower() == clean_id.lower():
full = os.path.join(base_dir, fname)
return FileResponse(full, media_type="application/octet-stream", filename=f"soundfont{ext}")
raise HTTPException(status_code=404, detail="SoundFont asset not found")
class RenderRequest(BaseModel):
project_json: dict
output_filename: Optional[str] = "render_output.wav"
@router.post("/render")
async def render_project(
req: RenderRequest,
current_user: dict = Depends(get_current_user)
):
engine = PythonRenderEngine()
output_path = os.path.join(settings.PROCESSED_DIR, req.output_filename or "render_output.wav")
try:
result_path = engine.render_project(req.project_json, output_path)
return {"url": f"/static/audio/processed/{os.path.basename(result_path)}", "path": result_path}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Render failed: {str(e)}")