209 lines
9.0 KiB
Python
209 lines
9.0 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.core.soundfont_scanner import SoundFontAutoScanner
|
|
from app.api.v1.auth import get_current_user, enforce_password_changed
|
|
|
|
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
|
|
_scanner = 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
|
|
|
|
def get_scanner():
|
|
global _scanner
|
|
if _scanner is None:
|
|
_scanner = SoundFontAutoScanner(system_sf_dir=SYSTEM_SF_DIR, upload_sf_dir=UPLOAD_SF_DIR)
|
|
_scanner.scan_once()
|
|
return _scanner
|
|
|
|
|
|
@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)):
|
|
scanner = get_scanner()
|
|
full_catalog = scanner.get_catalog()
|
|
inspector = get_inspector()
|
|
condensed_catalog = inspector.get_condensed_catalog_summary(full_catalog)
|
|
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)
|
|
):
|
|
enforce_password_changed(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")
|
|
|
|
# Stream upload in chunks with a hard size cap (SGM-class fonts can exceed
|
|
# 500MB; reading the whole body into RAM would OOM the server).
|
|
MAX_SF_UPLOAD_BYTES = 2 * 1024 * 1024 * 1024 # 2 GB
|
|
contents = bytearray()
|
|
while True:
|
|
chunk = await file.read(1024 * 1024)
|
|
if not chunk:
|
|
break
|
|
contents.extend(chunk)
|
|
if len(contents) > MAX_SF_UPLOAD_BYTES:
|
|
raise HTTPException(status_code=413, detail="SoundFont quá lớn (giới hạn 2GB)")
|
|
|
|
if not PluginManager.validate_sf2_header(bytes(contents[:4096])):
|
|
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)
|
|
|
|
scanner = get_scanner()
|
|
if background_tasks:
|
|
background_tasks.add_task(scanner.scan_once)
|
|
else:
|
|
scanner.scan_once()
|
|
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")
|
|
get_scanner().scan_once()
|
|
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
|
|
# Cũng tìm trong static/soundfonts (font bundled theo deployment) — trước
|
|
# đây chỉ UPLOAD + SYSTEM → font bundled 404 → incognito (IndexedDB rỗng)
|
|
# không tải được font → instrument CÂM (browser thường dùng cache nên OK).
|
|
static_sf_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "static", "soundfonts")
|
|
for base_dir in [UPLOAD_SF_DIR, SYSTEM_SF_DIR, static_sf_dir]:
|
|
if not os.path.isdir(base_dir):
|
|
continue
|
|
# Prefer SF2: the client FluidSynth WASM cannot decode SF3 (Ogg Vorbis)
|
|
# samples, so any SF3 would play silence in the browser.
|
|
for fname in os.listdir(base_dir):
|
|
fbase, fext = os.path.splitext(fname)
|
|
if fext.lower() == ".sf2" and fbase.lower() == clean_id.lower():
|
|
full = os.path.join(base_dir, fname)
|
|
return FileResponse(full, media_type="application/octet-stream", filename="soundfont.sf2")
|
|
# Only an SF3 exists -> decompress it to a playable SF2 on demand (cached)
|
|
for fname in os.listdir(base_dir):
|
|
fbase, fext = os.path.splitext(fname)
|
|
if fext.lower() == ".sf3" and fbase.lower() == clean_id.lower():
|
|
full = os.path.join(base_dir, fname)
|
|
try:
|
|
from app.core.soundfont_converter import SoundFontConverter
|
|
sf2_path = os.path.join(UPLOAD_SF_DIR, clean_id + ".sf2")
|
|
if os.path.exists(sf2_path) and os.path.getmtime(sf2_path) >= os.path.getmtime(full):
|
|
return FileResponse(sf2_path, media_type="application/octet-stream", filename="soundfont.sf2")
|
|
result = SoundFontConverter().sf3_to_sf2(full, sf2_path)
|
|
if result != full and os.path.exists(result):
|
|
return FileResponse(result, media_type="application/octet-stream", filename="soundfont.sf2")
|
|
except Exception as e:
|
|
print(f"[soundfont-download] SF3->SF2 conversion failed for {full}: {e}")
|
|
return FileResponse(full, media_type="application/octet-stream", filename="soundfont.sf3")
|
|
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)
|
|
):
|
|
enforce_password_changed(current_user)
|
|
engine = PythonRenderEngine()
|
|
# Prevent path traversal: strip any directory components and force .wav.
|
|
safe_name = os.path.basename((req.output_filename or "render_output.wav").replace("\\", "/"))
|
|
if not safe_name.lower().endswith(".wav"):
|
|
safe_name += ".wav"
|
|
output_path = os.path.join(settings.PROCESSED_DIR, safe_name)
|
|
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)}")
|