Implement avatar upload and management for voice profiles

- Added functionality to upload, delete, and retrieve avatar images for voice profiles.
- Introduced new API endpoints for avatar management, including upload and delete operations.
- Enhanced profile forms and components to support avatar image handling, including previews and error handling.
- Updated database schema to include avatar_path for profiles and added necessary migrations.
- Implemented image validation and processing utilities to ensure proper avatar uploads.
This commit is contained in:
Jamie Pine
2026-01-29 19:28:42 -08:00
parent ef3c3a7f8c
commit 89f3127c37
15 changed files with 594 additions and 35 deletions
+12 -1
View File
@@ -17,11 +17,12 @@ Base = declarative_base()
class VoiceProfile(Base):
"""Voice profile database model."""
__tablename__ = "profiles"
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
name = Column(String, unique=True, nullable=False)
description = Column(Text)
language = Column(String, default="en")
avatar_path = Column(String, nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@@ -277,6 +278,16 @@ def _run_migrations(engine):
conn.commit()
print("Added trim_end_ms column to story_items")
# Migration: Add avatar_path to profiles table
if 'profiles' in inspector.get_table_names():
columns = {col['name'] for col in inspector.get_columns('profiles')}
if 'avatar_path' not in columns:
print("Migrating profiles: adding avatar_path column")
with engine.connect() as conn:
conn.execute(text("ALTER TABLE profiles ADD COLUMN avatar_path VARCHAR"))
conn.commit()
print("Added avatar_path column to profiles")
def get_db():
"""Get database session (generator for dependency injection)."""
+40 -9
View File
@@ -75,6 +75,16 @@ def export_profile_to_zip(profile_id: str, db: Session) -> bytes:
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, 'w', zipfile.ZIP_DEFLATED) as zip_file:
# Check if profile has avatar
has_avatar = False
if profile.avatar_path:
avatar_path = Path(profile.avatar_path)
if avatar_path.exists():
has_avatar = True
# Add avatar to ZIP root with original extension
avatar_ext = avatar_path.suffix
zip_file.write(avatar_path, f"avatar{avatar_ext}")
# Create manifest.json
manifest = {
"version": "1.0",
@@ -82,30 +92,31 @@ def export_profile_to_zip(profile_id: str, db: Session) -> bytes:
"name": profile.name,
"description": profile.description,
"language": profile.language,
}
},
"has_avatar": has_avatar,
}
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)
@@ -168,11 +179,31 @@ async def import_profile_from_zip(file_bytes: bytes, db: Session) -> VoiceProfil
)
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)
# Handle avatar if present
avatar_files = [f for f in namelist if f.startswith("avatar.")]
if avatar_files:
try:
avatar_file = avatar_files[0]
# Extract to temporary file
import tempfile
with tempfile.NamedTemporaryFile(suffix=Path(avatar_file).suffix, delete=False) as tmp:
tmp.write(zip_file.read(avatar_file))
tmp_path = tmp.name
try:
from .profiles import upload_avatar
await upload_avatar(profile.id, tmp_path, db)
finally:
Path(tmp_path).unlink(missing_ok=True)
except Exception as e:
# Avatar import is optional - continue even if it fails
pass
for filename, reference_text in samples_data.items():
# Validate filename
if not filename.endswith('.wav'):
+55
View File
@@ -296,6 +296,61 @@ async def update_profile_sample(
return sample
@app.post("/profiles/{profile_id}/avatar", response_model=models.VoiceProfileResponse)
async def upload_profile_avatar(
profile_id: str,
file: UploadFile = File(...),
db: Session = Depends(get_db),
):
"""Upload or update avatar image for a profile."""
# Save uploaded file to temp location
with tempfile.NamedTemporaryFile(delete=False, suffix=Path(file.filename).suffix) as tmp:
content = await file.read()
tmp.write(content)
tmp_path = tmp.name
try:
profile = await profiles.upload_avatar(profile_id, tmp_path, db)
return profile
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
finally:
# Clean up temp file
Path(tmp_path).unlink(missing_ok=True)
@app.get("/profiles/{profile_id}/avatar")
async def get_profile_avatar(
profile_id: str,
db: Session = Depends(get_db),
):
"""Get avatar image for a profile."""
profile = await profiles.get_profile(profile_id, db)
if not profile:
raise HTTPException(status_code=404, detail="Profile not found")
if not profile.avatar_path:
raise HTTPException(status_code=404, detail="No avatar found for this profile")
avatar_path = Path(profile.avatar_path)
if not avatar_path.exists():
raise HTTPException(status_code=404, detail="Avatar file not found")
return FileResponse(avatar_path)
@app.delete("/profiles/{profile_id}/avatar")
async def delete_profile_avatar(
profile_id: str,
db: Session = Depends(get_db),
):
"""Delete avatar image for a profile."""
success = await profiles.delete_avatar(profile_id, db)
if not success:
raise HTTPException(status_code=404, detail="Profile not found or no avatar to delete")
return {"message": "Avatar deleted successfully"}
@app.get("/profiles/{profile_id}/export")
async def export_profile(
profile_id: str,
+1
View File
@@ -20,6 +20,7 @@ class VoiceProfileResponse(BaseModel):
name: str
description: Optional[str]
language: str
avatar_path: Optional[str] = None
created_at: datetime
updated_at: datetime
+105 -8
View File
@@ -21,6 +21,7 @@ from .database import (
ProfileSample as DBProfileSample,
)
from .utils.audio import validate_reference_audio, load_audio, save_audio
from .utils.images import validate_image, process_avatar
from .tts import get_tts_model
from . import config
@@ -307,23 +308,23 @@ async def create_voice_prompt_for_profile(
) -> dict:
"""
Create a combined voice prompt from all samples in a profile.
Args:
profile_id: Profile ID
db: Database session
use_cache: Whether to use cached prompts
Returns:
Voice prompt dictionary
"""
# Get all samples for profile
samples = db.query(DBProfileSample).filter_by(profile_id=profile_id).all()
if not samples:
raise ValueError(f"No samples found for profile {profile_id}")
tts_model = get_tts_model()
if len(samples) == 1:
# Single sample - use directly
sample = samples[0]
@@ -337,19 +338,19 @@ async def create_voice_prompt_for_profile(
# Multiple samples - combine them
audio_paths = [s.audio_path for s in samples]
reference_texts = [s.reference_text for s in samples]
# Combine audio
combined_audio, combined_text = await tts_model.combine_voice_prompts(
audio_paths,
reference_texts,
)
# Save combined audio temporarily
import tempfile
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
save_audio(combined_audio, tmp.name, 24000)
tmp_path = tmp.name
try:
# Create prompt from combined audio
voice_prompt, _ = await tts_model.create_voice_prompt(
@@ -361,3 +362,99 @@ async def create_voice_prompt_for_profile(
finally:
# Clean up temp file
Path(tmp_path).unlink(missing_ok=True)
async def upload_avatar(
profile_id: str,
image_path: str,
db: Session,
) -> VoiceProfileResponse:
"""
Upload and process avatar image for a profile.
Args:
profile_id: Profile ID
image_path: Path to uploaded image file
db: Database session
Returns:
Updated profile
"""
# Validate profile exists
profile = db.query(DBVoiceProfile).filter_by(id=profile_id).first()
if not profile:
raise ValueError(f"Profile {profile_id} not found")
# Validate image
is_valid, error_msg = validate_image(image_path)
if not is_valid:
raise ValueError(error_msg)
# Delete existing avatar if present
if profile.avatar_path:
old_avatar = Path(profile.avatar_path)
if old_avatar.exists():
old_avatar.unlink()
# Determine file extension from uploaded file
from PIL import Image
with Image.open(image_path) as img:
# Normalize JPEG variants (MPO is multi-picture format from some cameras)
img_format = img.format
if img_format in ('MPO', 'JPG'):
img_format = 'JPEG'
ext_map = {
'PNG': '.png',
'JPEG': '.jpg',
'WEBP': '.webp'
}
ext = ext_map.get(img_format, '.png')
# Save processed image to profile directory
profile_dir = _get_profiles_dir() / profile_id
profile_dir.mkdir(parents=True, exist_ok=True)
output_path = profile_dir / f"avatar{ext}"
process_avatar(image_path, str(output_path))
# Update database
profile.avatar_path = str(output_path)
profile.updated_at = datetime.utcnow()
db.commit()
db.refresh(profile)
return VoiceProfileResponse.model_validate(profile)
async def delete_avatar(
profile_id: str,
db: Session,
) -> bool:
"""
Delete avatar image for a profile.
Args:
profile_id: Profile ID
db: Database session
Returns:
True if deleted, False if not found or no avatar
"""
profile = db.query(DBVoiceProfile).filter_by(id=profile_id).first()
if not profile or not profile.avatar_path:
return False
# Delete avatar file
avatar_path = Path(profile.avatar_path)
if avatar_path.exists():
avatar_path.unlink()
# Update database
profile.avatar_path = None
profile.updated_at = datetime.utcnow()
db.commit()
return True
+1
View File
@@ -21,3 +21,4 @@ numpy>=1.24.0
# Utilities
python-multipart>=0.0.6
Pillow>=10.0.0
+114
View File
@@ -0,0 +1,114 @@
"""Image processing utilities for avatar uploads."""
from pathlib import Path
from typing import Optional, Tuple
from PIL import Image
# JPEG can be reported as 'JPEG' or 'MPO' (for multi-picture format from some cameras)
ALLOWED_FORMATS = {'PNG', 'JPEG', 'WEBP', 'MPO', 'JPG'}
MAX_SIZE = 512
MAX_FILE_SIZE = 5 * 1024 * 1024 # 5MB
def validate_image(file_path: str) -> Tuple[bool, Optional[str]]:
"""
Validate image format and file size.
Args:
file_path: Path to image file
Returns:
Tuple of (is_valid, error_message)
"""
path = Path(file_path)
# Check file size
if path.stat().st_size > MAX_FILE_SIZE:
return False, f"File size exceeds maximum of {MAX_FILE_SIZE // (1024 * 1024)}MB"
try:
with Image.open(file_path) as img:
# Verify the image can be loaded
img.load()
# Check format (normalize JPEG variants)
img_format = img.format
if img_format in ('MPO', 'JPG'):
img_format = 'JPEG'
if img_format not in {'PNG', 'JPEG', 'WEBP'}:
return False, f"Invalid format '{img_format}'. Allowed formats: PNG, JPEG, WEBP"
return True, None
except Exception as e:
return False, f"Invalid image file: {str(e)}"
def process_avatar(input_path: str, output_path: str, max_size: int = MAX_SIZE) -> None:
"""
Process avatar image: resize and optimize.
Resizes image to fit within max_size x max_size while maintaining aspect ratio.
Args:
input_path: Path to input image
output_path: Path to save processed image
max_size: Maximum width or height in pixels
"""
with Image.open(input_path) as img:
# Handle EXIF orientation for JPEG images
try:
from PIL import ExifTags
for orientation in ExifTags.TAGS.keys():
if ExifTags.TAGS[orientation] == 'Orientation':
break
exif = img._getexif()
if exif is not None:
orientation_value = exif.get(orientation)
if orientation_value == 3:
img = img.rotate(180, expand=True)
elif orientation_value == 6:
img = img.rotate(270, expand=True)
elif orientation_value == 8:
img = img.rotate(90, expand=True)
except (AttributeError, KeyError, IndexError, TypeError):
# No EXIF data or orientation tag
pass
# Convert to RGB if necessary (handles RGBA, P, CMYK, etc.)
if img.mode not in ('RGB', 'L'):
if img.mode == 'RGBA':
# Create white background for RGBA images
background = Image.new('RGB', img.size, (255, 255, 255))
background.paste(img, mask=img.split()[3]) # Use alpha channel as mask
img = background
elif img.mode == 'CMYK':
# Convert CMYK to RGB
img = img.convert('RGB')
elif img.mode == 'P':
# Convert palette mode to RGB
img = img.convert('RGB')
else:
img = img.convert('RGB')
# Calculate new size maintaining aspect ratio
img.thumbnail((max_size, max_size), Image.Resampling.LANCZOS)
# Determine output format from extension
output_ext = Path(output_path).suffix.lower()
format_map = {
'.png': 'PNG',
'.jpeg': 'JPEG',
'.jpg': 'JPEG',
'.webp': 'WEBP'
}
output_format = format_map.get(output_ext, 'PNG')
# Save with optimization
save_kwargs = {'optimize': True}
if output_format == 'JPEG':
save_kwargs['quality'] = 90
img.save(output_path, format=output_format, **save_kwargs)