mirror of
https://github.com/jamiepine/voicebox.git
synced 2026-09-29 07:05:14 -07:00
Add profile export and import functionality
- Implemented API endpoints for exporting and importing voice profiles as ZIP archives. - Enhanced the frontend with new hooks and components for profile export and import, including file handling and user dialogs. - Integrated Tauri plugins for file system access and dialog interactions to facilitate seamless user experience. - Updated ProfileCard and ProfileList components to support new export and import features, improving overall functionality. - Added necessary error handling and validation for file operations to ensure robustness.
This commit is contained in:
Jamie Pine
@@ -0,0 +1,213 @@
|
||||
"""
|
||||
Voice profile export/import module.
|
||||
|
||||
Handles exporting profiles to ZIP archives and importing them back.
|
||||
"""
|
||||
|
||||
import json
|
||||
import zipfile
|
||||
import io
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from .models import VoiceProfileResponse
|
||||
from .database import VoiceProfile as DBVoiceProfile, ProfileSample as DBProfileSample
|
||||
from .profiles import create_profile, add_profile_sample
|
||||
from .models import VoiceProfileCreate
|
||||
from . import config
|
||||
|
||||
|
||||
def _get_profiles_dir() -> Path:
|
||||
"""Get profiles directory from config."""
|
||||
return config.get_profiles_dir()
|
||||
|
||||
|
||||
def _get_unique_profile_name(name: str, db: Session) -> str:
|
||||
"""
|
||||
Get a unique profile name by appending a number if needed.
|
||||
|
||||
Args:
|
||||
name: Original profile name
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Unique profile name
|
||||
"""
|
||||
base_name = name
|
||||
counter = 1
|
||||
|
||||
while True:
|
||||
existing = db.query(DBVoiceProfile).filter_by(name=name).first()
|
||||
if not existing:
|
||||
return name
|
||||
|
||||
name = f"{base_name} ({counter})"
|
||||
counter += 1
|
||||
|
||||
|
||||
def export_profile_to_zip(profile_id: str, db: Session) -> bytes:
|
||||
"""
|
||||
Export a voice profile to a ZIP archive.
|
||||
|
||||
Args:
|
||||
profile_id: Profile ID to export
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
ZIP file contents as bytes
|
||||
|
||||
Raises:
|
||||
ValueError: If profile not found or has no samples
|
||||
"""
|
||||
# Get profile
|
||||
profile = db.query(DBVoiceProfile).filter_by(id=profile_id).first()
|
||||
if not profile:
|
||||
raise ValueError(f"Profile {profile_id} not found")
|
||||
|
||||
# Get all samples
|
||||
samples = db.query(DBProfileSample).filter_by(profile_id=profile_id).all()
|
||||
if not samples:
|
||||
raise ValueError(f"Profile {profile_id} has no samples")
|
||||
|
||||
# Create ZIP in memory
|
||||
zip_buffer = io.BytesIO()
|
||||
|
||||
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
|
||||
# Create manifest.json
|
||||
manifest = {
|
||||
"version": "1.0",
|
||||
"profile": {
|
||||
"name": profile.name,
|
||||
"description": profile.description,
|
||||
"language": profile.language,
|
||||
}
|
||||
}
|
||||
zip_file.writestr("manifest.json", json.dumps(manifest, indent=2))
|
||||
|
||||
# Create samples.json mapping
|
||||
samples_data = {}
|
||||
profile_dir = _get_profiles_dir() / profile_id
|
||||
|
||||
for sample in samples:
|
||||
# Get filename from audio_path (should be {sample_id}.wav)
|
||||
audio_path = Path(sample.audio_path)
|
||||
filename = audio_path.name
|
||||
|
||||
# Read audio file
|
||||
if not audio_path.exists():
|
||||
raise ValueError(f"Audio file not found: {audio_path}")
|
||||
|
||||
# Add to samples directory in ZIP
|
||||
zip_path = f"samples/{filename}"
|
||||
zip_file.write(audio_path, zip_path)
|
||||
|
||||
# Map filename to reference text
|
||||
samples_data[filename] = sample.reference_text
|
||||
|
||||
zip_file.writestr("samples.json", json.dumps(samples_data, indent=2))
|
||||
|
||||
zip_buffer.seek(0)
|
||||
return zip_buffer.read()
|
||||
|
||||
|
||||
async def import_profile_from_zip(file_bytes: bytes, db: Session) -> VoiceProfileResponse:
|
||||
"""
|
||||
Import a voice profile from a ZIP archive.
|
||||
|
||||
Args:
|
||||
file_bytes: ZIP file contents
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
Created profile
|
||||
|
||||
Raises:
|
||||
ValueError: If ZIP is invalid or missing required files
|
||||
"""
|
||||
zip_buffer = io.BytesIO(file_bytes)
|
||||
|
||||
try:
|
||||
with zipfile.ZipFile(zip_buffer, 'r') as zip_file:
|
||||
# Validate ZIP structure
|
||||
namelist = zip_file.namelist()
|
||||
|
||||
if "manifest.json" not in namelist:
|
||||
raise ValueError("ZIP archive missing manifest.json")
|
||||
|
||||
if "samples.json" not in namelist:
|
||||
raise ValueError("ZIP archive missing samples.json")
|
||||
|
||||
# Read manifest
|
||||
manifest_data = json.loads(zip_file.read("manifest.json"))
|
||||
|
||||
if "version" not in manifest_data:
|
||||
raise ValueError("Invalid manifest.json: missing version")
|
||||
|
||||
if "profile" not in manifest_data:
|
||||
raise ValueError("Invalid manifest.json: missing profile")
|
||||
|
||||
profile_data = manifest_data["profile"]
|
||||
|
||||
# Read samples mapping
|
||||
samples_data = json.loads(zip_file.read("samples.json"))
|
||||
|
||||
if not isinstance(samples_data, dict):
|
||||
raise ValueError("Invalid samples.json: must be a dictionary")
|
||||
|
||||
# Get unique profile name
|
||||
original_name = profile_data.get("name", "Imported Profile")
|
||||
unique_name = _get_unique_profile_name(original_name, db)
|
||||
|
||||
# Create profile
|
||||
profile_create = VoiceProfileCreate(
|
||||
name=unique_name,
|
||||
description=profile_data.get("description"),
|
||||
language=profile_data.get("language", "en"),
|
||||
)
|
||||
|
||||
profile = await create_profile(profile_create, db)
|
||||
|
||||
# Extract and add samples
|
||||
profile_dir = _get_profiles_dir() / profile.id
|
||||
profile_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
for filename, reference_text in samples_data.items():
|
||||
# Validate filename
|
||||
if not filename.endswith('.wav'):
|
||||
raise ValueError(f"Invalid sample filename: {filename} (must be .wav)")
|
||||
|
||||
# Extract audio file to temp location
|
||||
zip_path = f"samples/{filename}"
|
||||
|
||||
if zip_path not in namelist:
|
||||
raise ValueError(f"Sample file not found in ZIP: {zip_path}")
|
||||
|
||||
# Extract to temporary file
|
||||
import tempfile
|
||||
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
|
||||
tmp.write(zip_file.read(zip_path))
|
||||
tmp_path = tmp.name
|
||||
|
||||
try:
|
||||
# Add sample to profile
|
||||
await add_profile_sample(
|
||||
profile.id,
|
||||
tmp_path,
|
||||
reference_text,
|
||||
db,
|
||||
)
|
||||
finally:
|
||||
# Clean up temp file
|
||||
Path(tmp_path).unlink(missing_ok=True)
|
||||
|
||||
return profile
|
||||
|
||||
except zipfile.BadZipFile:
|
||||
raise ValueError("Invalid ZIP file")
|
||||
except json.JSONDecodeError as e:
|
||||
raise ValueError(f"Invalid JSON in archive: {e}")
|
||||
except Exception as e:
|
||||
if isinstance(e, ValueError):
|
||||
raise
|
||||
raise ValueError(f"Error importing profile: {str(e)}")
|
||||
+65
-2
@@ -6,7 +6,7 @@ Handles voice cloning, generation history, and server mode.
|
||||
|
||||
from fastapi import FastAPI, Depends, UploadFile, File, Form, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.responses import FileResponse
|
||||
from fastapi.responses import FileResponse, StreamingResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import List, Optional
|
||||
@@ -14,10 +14,11 @@ import uvicorn
|
||||
import argparse
|
||||
import torch
|
||||
import tempfile
|
||||
import io
|
||||
from pathlib import Path
|
||||
import uuid
|
||||
|
||||
from . import database, models, profiles, history, tts, transcribe, config
|
||||
from . import database, models, profiles, history, tts, transcribe, config, export_import
|
||||
from .database import get_db, Generation as DBGeneration, VoiceProfile as DBVoiceProfile
|
||||
from .utils.progress import get_progress_manager
|
||||
|
||||
@@ -227,6 +228,68 @@ async def delete_profile_sample(
|
||||
return {"message": "Sample deleted successfully"}
|
||||
|
||||
|
||||
@app.get("/profiles/{profile_id}/export")
|
||||
async def export_profile(
|
||||
profile_id: str,
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Export a voice profile as a ZIP archive."""
|
||||
try:
|
||||
# Get profile to get name for filename
|
||||
profile = await profiles.get_profile(profile_id, db)
|
||||
if not profile:
|
||||
raise HTTPException(status_code=404, detail="Profile not found")
|
||||
|
||||
# Export to ZIP
|
||||
zip_bytes = export_import.export_profile_to_zip(profile_id, db)
|
||||
|
||||
# Create safe filename
|
||||
safe_name = "".join(c for c in profile.name if c.isalnum() or c in (' ', '-', '_')).strip()
|
||||
if not safe_name:
|
||||
safe_name = "profile"
|
||||
filename = f"profile-{safe_name}.voicebox.zip"
|
||||
|
||||
# Return as streaming response
|
||||
return StreamingResponse(
|
||||
io.BytesIO(zip_bytes),
|
||||
media_type="application/zip",
|
||||
headers={
|
||||
"Content-Disposition": f'attachment; filename="{filename}"'
|
||||
}
|
||||
)
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@app.post("/profiles/import", response_model=models.VoiceProfileResponse)
|
||||
async def import_profile(
|
||||
file: UploadFile = File(...),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""Import a voice profile from a ZIP archive."""
|
||||
# Validate file size (max 100MB)
|
||||
MAX_FILE_SIZE = 100 * 1024 * 1024 # 100MB
|
||||
|
||||
# Read file content
|
||||
content = await file.read()
|
||||
|
||||
if len(content) > MAX_FILE_SIZE:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"File too large. Maximum size is {MAX_FILE_SIZE / (1024 * 1024)}MB"
|
||||
)
|
||||
|
||||
try:
|
||||
profile = await export_import.import_profile_from_zip(content, db)
|
||||
return profile
|
||||
except ValueError as e:
|
||||
raise HTTPException(status_code=400, detail=str(e))
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
# ============================================
|
||||
# GENERATION ENDPOINTS
|
||||
# ============================================
|
||||
|
||||
Reference in New Issue
Block a user