Labyricorn/voicebox
1
0
Fork 0
mirror of https://github.com/jamiepine/voicebox.git synced 2026-09-20 07:10:40 -07:00
Code Issues Packages Projects Releases Wiki Activity

chore(backend): repair test suite and bring ruff to green

Browse Source 
The suite hadn't run green since the routes refactor:
- test_profile_duplicate_names.py imported the pre-refactor module
  layout and broke collection; now imports backend.services.profiles
- tests/conftest.py puts the repo root and backend dir on sys.path so
  files collect standalone instead of depending on run order
- test_cors.py tested a hand-copied mirror of the origin list that had
  drifted from app.py (missing http://tauri.localhost); it now builds
  the app via the real create_app() factory
- test_progress.py simulated a 1KB download, below the tracker's 1MB
  reporting threshold; simulation raised to 5MB
- slow/timeout markers registered in pyproject

Ruff: ~900 violations auto-fixed (typing modernization, import
sorting, unused imports, whitespace). The remaining rules are baselined
in pyproject.toml with per-rule counts to burn down, plus per-file
carve-outs for deliberate env-before-import ordering. ruff check is
now clean; suite is 134 passed, 2 skipped.
This commit is contained in:
Jamie Pine
2026-07-26 23:16:09 -07:00
parent 766c51a8a1
commit b434db22f6
82 changed files with 970 additions and 999 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/services/export_import.py
+81 -76
View File
@@ -5,39 +5,43 @@ Handles exporting profiles to ZIP archives and importing them back.
Also handles exporting individual generations.
"""
import io
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, Generation as DBGeneration, GenerationVersion as DBGenerationVersion
from .profiles import create_profile, add_profile_sample
from ..models import VoiceProfileCreate
from .. import config
from ..database import (
Generation as DBGeneration,
GenerationVersion as DBGenerationVersion,
ProfileSample as DBProfileSample,
VoiceProfile as DBVoiceProfile,
)
from ..models import VoiceProfileCreate, VoiceProfileResponse
from .profiles import add_profile_sample, create_profile
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
@@ -45,14 +49,14 @@ def _get_unique_profile_name(name: str, db: Session) -> str:
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
"""
@@ -60,15 +64,15 @@ def export_profile_to_zip(profile_id: str, db: Session) -> bytes:
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:
# Check if profile has avatar
has_avatar = False
@@ -115,7 +119,7 @@ def export_profile_to_zip(profile_id: str, db: Session) -> bytes:
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()
@@ -123,58 +127,58 @@ def export_profile_to_zip(profile_id: str, db: Session) -> bytes:
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
@@ -197,7 +201,7 @@ async def import_profile_from_zip(file_bytes: bytes, db: Session) -> VoiceProfil
await upload_avatar(profile.id, tmp_path, db)
finally:
Path(tmp_path).unlink(missing_ok=True)
except Exception as e:
except Exception:
# Avatar import is optional - continue even if it fails
pass
@@ -205,19 +209,19 @@ async def import_profile_from_zip(file_bytes: bytes, db: Session) -> VoiceProfil
# 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(
@@ -229,9 +233,9 @@ async def import_profile_from_zip(file_bytes: bytes, db: Session) -> VoiceProfil
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:
@@ -239,20 +243,20 @@ async def import_profile_from_zip(file_bytes: bytes, db: Session) -> VoiceProfil
except Exception as e:
if isinstance(e, ValueError):
raise
raise ValueError(f"Error importing profile: {str(e)}")
raise ValueError(f"Error importing profile: {e!s}")
def export_generation_to_zip(generation_id: str, db: Session) -> bytes:
"""
Export a generation to a ZIP archive.
Args:
generation_id: Generation ID to export
db: Database session
Returns:
ZIP file contents as bytes
Raises:
ValueError: If generation not found
"""
@@ -260,12 +264,12 @@ def export_generation_to_zip(generation_id: str, db: Session) -> bytes:
generation = db.query(DBGeneration).filter_by(id=generation_id).first()
if not generation:
raise ValueError(f"Generation {generation_id} not found")
# Get profile info
profile = db.query(DBVoiceProfile).filter_by(id=generation.profile_id).first()
if not profile:
raise ValueError(f"Profile {generation.profile_id} not found")
# Get all versions for this generation
versions = (
db.query(DBGenerationVersion)
@@ -276,7 +280,7 @@ def export_generation_to_zip(generation_id: str, db: Session) -> bytes:
# Create ZIP in memory
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
# Build version manifest entries
version_entries = []
@@ -313,7 +317,7 @@ def export_generation_to_zip(generation_id: str, db: Session) -> bytes:
"versions": version_entries,
}
zip_file.writestr("manifest.json", json.dumps(manifest, indent=2))
# Add all version audio files
for v in versions:
v_path = config.resolve_storage_path(v.audio_path)
@@ -325,7 +329,7 @@ def export_generation_to_zip(generation_id: str, db: Session) -> bytes:
audio_path = config.resolve_storage_path(generation.audio_path)
if audio_path is not None and audio_path.exists():
zip_file.write(audio_path, f"audio/{audio_path.name}")
zip_buffer.seek(0)
return zip_buffer.read()
@@ -333,68 +337,69 @@ def export_generation_to_zip(generation_id: str, db: Session) -> bytes:
async def import_generation_from_zip(file_bytes: bytes, db: Session) -> dict:
"""
Import a generation from a ZIP archive.
Args:
file_bytes: ZIP file contents
db: Database session
Returns:
Dictionary with generation ID and profile info
Raises:
ValueError: If ZIP is invalid or missing required files
"""
from pathlib import Path
import tempfile
import shutil
import tempfile
from datetime import datetime
from pathlib import Path
from .. import config
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")
# 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 "generation" not in manifest_data:
raise ValueError("Invalid manifest.json: missing generation data")
generation_data = manifest_data["generation"]
profile_data = manifest_data.get("profile", {})
# Validate required fields
required_fields = ["text", "language", "duration"]
for field in required_fields:
if field not in generation_data:
raise ValueError(f"Invalid manifest.json: missing generation.{field}")
# Find audio file in archive
audio_files = [f for f in namelist if f.startswith("audio/") and f.endswith(".wav")]
if not audio_files:
raise ValueError("No audio file found in ZIP archive")
audio_file_path = audio_files[0]
# Check if we should match an existing profile or create metadata
profile_id = None
profile_name = profile_data.get("name", "Unknown Profile")
# Try to find matching profile by name
if profile_name and profile_name != "Unknown Profile":
existing_profile = db.query(DBVoiceProfile).filter_by(name=profile_name).first()
if existing_profile:
profile_id = existing_profile.id
# If no matching profile, use a placeholder or the first available profile
if not profile_id:
# Get any profile, or None if no profiles exist
@@ -404,24 +409,24 @@ async def import_generation_from_zip(file_bytes: bytes, db: Session) -> dict:
profile_name = any_profile.name
else:
raise ValueError("No voice profiles found. Please create a profile before importing generations.")
# Extract audio file to temporary location
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
tmp.write(zip_file.read(audio_file_path))
tmp_path = tmp.name
try:
# Create generations directory
generations_dir = config.get_generations_dir()
generations_dir.mkdir(parents=True, exist_ok=True)
# Generate new ID for this generation
new_generation_id = str(__import__('uuid').uuid4())
# Copy audio to generations directory
audio_dest = generations_dir / f"{new_generation_id}.wav"
shutil.copy(tmp_path, audio_dest)
# Create generation record
db_generation = DBGeneration(
id=new_generation_id,
@@ -434,11 +439,11 @@ async def import_generation_from_zip(file_bytes: bytes, db: Session) -> dict:
instruct=generation_data.get("instruct"),
created_at=datetime.utcnow(),
)
db.add(db_generation)
db.commit()
db.refresh(db_generation)
return {
"id": db_generation.id,
"profile_id": profile_id,
@@ -446,11 +451,11 @@ async def import_generation_from_zip(file_bytes: bytes, db: Session) -> dict:
"text": db_generation.text,
"message": f"Generation imported successfully (assigned to profile: {profile_name})"
}
finally:
# Clean up temp file
Path(tmp_path).unlink(missing_ok=True)
except zipfile.BadZipFile:
raise ValueError("Invalid ZIP file")
except json.JSONDecodeError as e:
@@ -458,4 +463,4 @@ async def import_generation_from_zip(file_bytes: bytes, db: Session) -> dict:
except Exception as e:
if isinstance(e, ValueError):
raise
raise ValueError(f"Error importing generation: {str(e)}")
raise ValueError(f"Error importing generation: {e!s}")