Spaces:
Sleeping
Sleeping
File size: 7,208 Bytes
f871fed | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 | from typing import Any, Dict, List
from fastapi import APIRouter, HTTPException
from loguru import logger
from pydantic import BaseModel, Field
from open_notebook.domain.podcast import SpeakerProfile
router = APIRouter()
class SpeakerProfileResponse(BaseModel):
id: str
name: str
description: str
tts_provider: str
tts_model: str
speakers: List[Dict[str, Any]]
@router.get("/speaker-profiles", response_model=List[SpeakerProfileResponse])
async def list_speaker_profiles():
"""List all available speaker profiles"""
try:
profiles = await SpeakerProfile.get_all(order_by="name asc")
return [
SpeakerProfileResponse(
id=str(profile.id),
name=profile.name,
description=profile.description or "",
tts_provider=profile.tts_provider,
tts_model=profile.tts_model,
speakers=profile.speakers
)
for profile in profiles
]
except Exception as e:
logger.error(f"Failed to fetch speaker profiles: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to fetch speaker profiles: {str(e)}"
)
@router.get("/speaker-profiles/{profile_name}", response_model=SpeakerProfileResponse)
async def get_speaker_profile(profile_name: str):
"""Get a specific speaker profile by name"""
try:
profile = await SpeakerProfile.get_by_name(profile_name)
if not profile:
raise HTTPException(
status_code=404,
detail=f"Speaker profile '{profile_name}' not found"
)
return SpeakerProfileResponse(
id=str(profile.id),
name=profile.name,
description=profile.description or "",
tts_provider=profile.tts_provider,
tts_model=profile.tts_model,
speakers=profile.speakers
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to fetch speaker profile '{profile_name}': {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to fetch speaker profile: {str(e)}"
)
class SpeakerProfileCreate(BaseModel):
name: str = Field(..., description="Unique profile name")
description: str = Field("", description="Profile description")
tts_provider: str = Field(..., description="TTS provider")
tts_model: str = Field(..., description="TTS model name")
speakers: List[Dict[str, Any]] = Field(..., description="Array of speaker configurations")
@router.post("/speaker-profiles", response_model=SpeakerProfileResponse)
async def create_speaker_profile(profile_data: SpeakerProfileCreate):
"""Create a new speaker profile"""
try:
profile = SpeakerProfile(
name=profile_data.name,
description=profile_data.description,
tts_provider=profile_data.tts_provider,
tts_model=profile_data.tts_model,
speakers=profile_data.speakers
)
await profile.save()
return SpeakerProfileResponse(
id=str(profile.id),
name=profile.name,
description=profile.description or "",
tts_provider=profile.tts_provider,
tts_model=profile.tts_model,
speakers=profile.speakers
)
except Exception as e:
logger.error(f"Failed to create speaker profile: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to create speaker profile: {str(e)}"
)
@router.put("/speaker-profiles/{profile_id}", response_model=SpeakerProfileResponse)
async def update_speaker_profile(profile_id: str, profile_data: SpeakerProfileCreate):
"""Update an existing speaker profile"""
try:
profile = await SpeakerProfile.get(profile_id)
if not profile:
raise HTTPException(
status_code=404,
detail=f"Speaker profile '{profile_id}' not found"
)
# Update fields
profile.name = profile_data.name
profile.description = profile_data.description
profile.tts_provider = profile_data.tts_provider
profile.tts_model = profile_data.tts_model
profile.speakers = profile_data.speakers
await profile.save()
return SpeakerProfileResponse(
id=str(profile.id),
name=profile.name,
description=profile.description or "",
tts_provider=profile.tts_provider,
tts_model=profile.tts_model,
speakers=profile.speakers
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to update speaker profile: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to update speaker profile: {str(e)}"
)
@router.delete("/speaker-profiles/{profile_id}")
async def delete_speaker_profile(profile_id: str):
"""Delete a speaker profile"""
try:
profile = await SpeakerProfile.get(profile_id)
if not profile:
raise HTTPException(
status_code=404,
detail=f"Speaker profile '{profile_id}' not found"
)
await profile.delete()
return {"message": "Speaker profile deleted successfully"}
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to delete speaker profile: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to delete speaker profile: {str(e)}"
)
@router.post("/speaker-profiles/{profile_id}/duplicate", response_model=SpeakerProfileResponse)
async def duplicate_speaker_profile(profile_id: str):
"""Duplicate a speaker profile"""
try:
original = await SpeakerProfile.get(profile_id)
if not original:
raise HTTPException(
status_code=404,
detail=f"Speaker profile '{profile_id}' not found"
)
# Create duplicate with modified name
duplicate = SpeakerProfile(
name=f"{original.name} - Copy",
description=original.description,
tts_provider=original.tts_provider,
tts_model=original.tts_model,
speakers=original.speakers
)
await duplicate.save()
return SpeakerProfileResponse(
id=str(duplicate.id),
name=duplicate.name,
description=duplicate.description or "",
tts_provider=duplicate.tts_provider,
tts_model=duplicate.tts_model,
speakers=duplicate.speakers
)
except HTTPException:
raise
except Exception as e:
logger.error(f"Failed to duplicate speaker profile: {e}")
raise HTTPException(
status_code=500,
detail=f"Failed to duplicate speaker profile: {str(e)}"
) |