Spaces:
Runtime error
Runtime error
| import re | |
| import uuid | |
| from datetime import UTC, datetime | |
| from fastapi import APIRouter, Depends, HTTPException, Request | |
| from fastapi.concurrency import run_in_threadpool | |
| from pydantic import BaseModel, Field | |
| from sqlalchemy import func, select | |
| from app.auth import Ctx, get_ctx | |
| from app.config import settings | |
| from app.models import Document, Job, Upload, VoiceConsent, VoiceProfile | |
| from app.services import tts | |
| from app.services.jobs import create_job, record_usage, serialize_job | |
| from app.services.notestack_voices import forget_video_voice, save_notestack_voice | |
| from app.services.plans import effective_plan, plan_limit_error | |
| from app.services.storage import storage | |
| router = APIRouter(prefix="/api/voice", tags=["voice"]) | |
| CONSENT_TEXT = ( | |
| "I confirm this recording is my own voice. I allow Notestack to create a synthetic copy of it with " | |
| "ElevenLabs, only for audio and video made in my workspace. I can revoke this at any time, which deletes " | |
| "the synthetic voice." | |
| ) | |
| PREVIEW_LINE = "Here is how I sound reading your archive. Every line I say comes from something you wrote." | |
| class BuildIn(BaseModel): | |
| document_ids: list[uuid.UUID] = Field(default_factory=list, max_length=10) | |
| class DeliveryIn(BaseModel): | |
| stability: float = Field(0.45, ge=0, le=1) | |
| similarity_boost: float = Field(0.8, ge=0, le=1) | |
| style: float = Field(0.2, ge=0, le=1) | |
| speed: float = Field(1.0, ge=0.7, le=1.2) | |
| class ProfileIn(BaseModel): | |
| profile: dict | None = None | |
| host_voices: dict[str, str] | None = None | |
| delivery: dict[str, DeliveryIn] | None = None # per host: {"host_a": {...}, "host_b": {...}} | |
| class ConsentIn(BaseModel): | |
| upload_ids: list[uuid.UUID] = Field(default_factory=list, max_length=10) | |
| upload_id: uuid.UUID | None = None # single sample (older clients) | |
| agreed: bool | |
| consent_text: str | |
| remove_background_noise: bool = True | |
| class PreviewIn(BaseModel): | |
| voice_id: str = Field(min_length=5, max_length=60) | |
| text: str | None = Field(None, max_length=400) | |
| host: str | None = None # use that host's saved delivery settings | |
| delivery: DeliveryIn | None = None | |
| class DesignIn(BaseModel): | |
| """Either picked options (Build from options) or a free description (Describe in your own words).""" | |
| prompt: str | None = Field(None, max_length=1000) | |
| gender: str = Field("", max_length=20) | |
| age: str = Field("", max_length=20) | |
| persona: str = Field("", max_length=30) | |
| pace: str = Field("", max_length=20) | |
| accent: str = Field("", max_length=40) | |
| class DesignSaveIn(BaseModel): | |
| generated_voice_id: str = Field(min_length=5, max_length=100) | |
| name: str = Field(min_length=1, max_length=60) | |
| description: str = Field("", max_length=1000) | |
| use_as: str | None = None # "host_a" | "host_b" | |
| class LibraryAddIn(BaseModel): | |
| public_owner_id: str = Field(min_length=5, max_length=100) | |
| voice_id: str = Field(min_length=5, max_length=60) | |
| name: str = Field(min_length=1, max_length=100) | |
| use_as: str | None = None # "host_a" | "host_b" | |
| def _profile(ctx: Ctx) -> VoiceProfile | None: | |
| return ctx.db.scalar(select(VoiceProfile).where(VoiceProfile.workspace_id == ctx.workspace.id)) | |
| def _ensure_profile(ctx: Ctx) -> VoiceProfile: | |
| vp = _profile(ctx) | |
| if not vp: | |
| vp = VoiceProfile(workspace_id=ctx.workspace.id) | |
| ctx.db.add(vp) | |
| return vp | |
| def _consent(ctx: Ctx) -> VoiceConsent | None: | |
| return ctx.db.scalar( | |
| select(VoiceConsent).where(VoiceConsent.workspace_id == ctx.workspace.id, VoiceConsent.revoked_at.is_(None)) | |
| .order_by(VoiceConsent.created_at.desc()) | |
| ) | |
| def _clone_job(ctx: Ctx, consent: VoiceConsent) -> Job | None: | |
| return ctx.db.scalar( | |
| select(Job).where(Job.workspace_id == ctx.workspace.id, Job.kind == "voice_clone") | |
| .order_by(Job.created_at.desc()).limit(1) | |
| ) | |
| def _serialize(ctx: Ctx) -> dict: | |
| vp = _profile(ctx) | |
| consent = _consent(ctx) | |
| plan = effective_plan(ctx.db, ctx.workspace) | |
| voices = (vp.host_voices if vp else None) or {} | |
| delivery = voices.get("settings") or {} | |
| job = _clone_job(ctx, consent) if consent else None | |
| preview_key = (job.result or {}).get("preview_key") if job and job.status == "done" else None | |
| return { | |
| "profile": vp.profile_json if vp else None, | |
| "sample_doc_ids": vp.sample_doc_ids if vp else [], | |
| "updated_at": vp.updated_at.isoformat() if vp and vp.updated_at else None, | |
| "host_voices": {"host_a": voices.get("host_a") or settings.elevenlabs_voice_a, | |
| "host_b": voices.get("host_b") or settings.elevenlabs_voice_b}, | |
| "delivery": {h: tts.VoiceSettings.from_dict(delivery.get(h)).body() for h in ("host_a", "host_b")}, | |
| "clone": { | |
| "allowed": plan.voice_cloning, | |
| "consent_text": CONSENT_TEXT, | |
| "status": ("ready" if consent and consent.elevenlabs_voice_id | |
| else "failed" if consent and job and job.status == "failed" | |
| else "processing" if consent else "none"), | |
| "voice_id": consent.elevenlabs_voice_id if consent else None, | |
| "created_at": consent.created_at.isoformat() if consent else None, | |
| "error": job.error if job and job.status == "failed" else None, | |
| "preview_url": storage.presign_get(preview_key) if preview_key else None, | |
| }, | |
| # Voices this workspace generated with Voice Design (the ElevenLabs account is shared across workspaces). | |
| "custom_voices": voices.get("custom") or [], | |
| "tts_configured": bool(settings.elevenlabs_api_key), | |
| "model": tts.model_id(), | |
| "models": tts.MODELS, | |
| } | |
| def get_voice(ctx: Ctx = Depends(get_ctx)): | |
| return _serialize(ctx) | |
| def update_voice(body: ProfileIn, ctx: Ctx = Depends(get_ctx)): | |
| vp = _ensure_profile(ctx) | |
| if body.profile is not None: | |
| vp.profile_json = body.profile | |
| voices = dict(vp.host_voices or {}) | |
| if body.host_voices is not None: | |
| voices.update({k: v for k, v in body.host_voices.items() if k in ("host_a", "host_b") and v}) | |
| if body.delivery is not None: | |
| saved = dict(voices.get("settings") or {}) | |
| saved.update({h: d.model_dump() for h, d in body.delivery.items() if h in ("host_a", "host_b")}) | |
| voices["settings"] = saved | |
| vp.host_voices = voices | |
| ctx.db.commit() | |
| return _serialize(ctx) | |
| def build(body: BuildIn, ctx: Ctx = Depends(get_ctx)): | |
| if not body.document_ids: # the automatic build after a sync may already be running; follow that one | |
| pending = ctx.db.scalar(select(Job).where(Job.workspace_id == ctx.workspace.id, Job.kind == "voice_profile", | |
| Job.status.in_(("queued", "running"))).limit(1)) | |
| if pending: | |
| return serialize_job(pending) | |
| job = create_job(ctx.db, ctx.workspace.id, "voice_profile", | |
| {"document_ids": [str(i) for i in body.document_ids]}, max_attempts=2) | |
| return serialize_job(job) | |
| async def voices(ctx: Ctx = Depends(get_ctx)): | |
| items = await run_in_threadpool(tts.list_voices) | |
| consent = _consent(ctx) | |
| if consent and consent.elevenlabs_voice_id: | |
| items = [{**v, "name": f"{v['name']} (my voice)", "category": "my voice"} | |
| if v["voice_id"] == consent.elevenlabs_voice_id else v for v in items] | |
| if not any(v["voice_id"] == consent.elevenlabs_voice_id for v in items): | |
| items = [{"voice_id": consent.elevenlabs_voice_id, "name": "My voice", "category": "my voice"}, *items] | |
| return items | |
| async def library(ctx: Ctx = Depends(get_ctx), search: str = "", gender: str = "", age: str = "", | |
| accent: str = "", language: str = "", use_case: str = "", page: int = 0): | |
| if not settings.elevenlabs_api_key: | |
| raise HTTPException(503, "Set ELEVENLABS_API_KEY in .env to browse the voice library.") | |
| try: | |
| return await run_in_threadpool(lambda: tts.search_library( | |
| search=search, gender=gender, age=age, accent=accent, language=language, use_case=use_case, | |
| page=max(page, 0))) | |
| except tts.TTSError as exc: | |
| raise HTTPException(502, str(exc)) from exc | |
| async def add_from_library(body: LibraryAddIn, ctx: Ctx = Depends(get_ctx)): | |
| try: | |
| voice_id = await run_in_threadpool(tts.add_library_voice, body.public_owner_id, body.voice_id, body.name) | |
| except tts.TTSError as exc: | |
| raise HTTPException(502, str(exc)) from exc | |
| vp = _ensure_profile(ctx) | |
| voices = dict(vp.host_voices or {}) | |
| # Listed, so it can be added to the workspace's video voices too. | |
| added = [v for v in voices.get("added") or [] if v.get("voice_id") != voice_id] | |
| voices["added"] = [*added, {"voice_id": voice_id, "name": body.name[:100]}] | |
| if body.use_as in ("host_a", "host_b"): | |
| voices[body.use_as] = voice_id | |
| vp.host_voices = voices | |
| save_notestack_voice(ctx.db, ctx.workspace.id, voice_id, body.name) # a voice added is a voice saved | |
| ctx.db.commit() | |
| return {"voice_id": voice_id, **_serialize(ctx)} | |
| async def preview(body: PreviewIn, ctx: Ctx = Depends(get_ctx)): | |
| """Speak a short line in any voice with the given (or a host's saved) delivery. Cached, metered.""" | |
| if not settings.elevenlabs_api_key: | |
| raise HTTPException(503, "Set ELEVENLABS_API_KEY in .env to preview voices.") | |
| if body.delivery: | |
| vs = tts.VoiceSettings.from_dict(body.delivery.model_dump()) | |
| else: | |
| saved = (((_profile(ctx) or VoiceProfile()).host_voices or {}).get("settings") or {}).get(body.host or "") | |
| vs = tts.VoiceSettings.from_dict(saved) | |
| text = (body.text or PREVIEW_LINE).strip()[:400] | |
| try: | |
| clip = await run_in_threadpool(tts.synthesize, ctx.workspace.id, text, body.voice_id, vs) | |
| except tts.TTSError as exc: | |
| raise HTTPException(502, str(exc)) from exc | |
| key = f"ws/{ctx.workspace.id}/voice-previews/{uuid.uuid4().hex}.mp3" | |
| storage.put_bytes(key, clip, "audio/mpeg") | |
| record_usage(ctx.db, workspace_id=ctx.workspace.id, kind="tts", provider="elevenlabs", | |
| quantity=round(tts.mp3_seconds(clip), 1), unit="seconds") | |
| return {"url": storage.presign_get(key), "seconds": round(tts.mp3_seconds(clip), 1)} | |
| def _design_description(body: DesignIn) -> str: | |
| if body.prompt and body.prompt.strip(): | |
| text = body.prompt.strip() | |
| if len(text) < 20: | |
| raise HTTPException(400, "Describe the voice in at least 20 characters.") | |
| return text | |
| parts = [f"A {body.age} {body.gender} voice".replace(" ", " ") if body.gender or body.age else "A voice"] | |
| if body.accent: | |
| parts.append(f"with a {body.accent} accent") | |
| sentence = " ".join(parts) + "." | |
| if body.persona: | |
| sentence += f" Sounds {body.persona}." | |
| if body.pace: | |
| sentence += f" Speaks at a {body.pace} pace." | |
| return sentence + " Clear studio quality, suited to podcasts and narration." | |
| async def design(body: DesignIn, ctx: Ctx = Depends(get_ctx)): | |
| """Three previews of a new voice. Nothing is saved until the writer picks one.""" | |
| if not settings.elevenlabs_api_key: | |
| raise HTTPException(503, "Set ELEVENLABS_API_KEY in .env to generate voices.") | |
| description = _design_description(body) | |
| try: | |
| previews = await run_in_threadpool(tts.design_voice, description) | |
| except tts.TTSError as exc: | |
| raise HTTPException(502, str(exc)) from exc | |
| out = [] | |
| for p in previews: | |
| key = f"ws/{ctx.workspace.id}/voice-previews/{uuid.uuid4().hex}.mp3" | |
| storage.put_bytes(key, p["audio"], "audio/mpeg") | |
| out.append({"generated_voice_id": p["generated_voice_id"], "url": storage.presign_get(key), | |
| "seconds": p["seconds"]}) | |
| record_usage(ctx.db, workspace_id=ctx.workspace.id, kind="voice_design", provider="elevenlabs", | |
| quantity=len(out), unit="previews") | |
| return {"description": description, "previews": out} | |
| async def save_design(body: DesignSaveIn, ctx: Ctx = Depends(get_ctx)): | |
| try: | |
| voice_id = await run_in_threadpool(tts.create_designed_voice, body.name, body.description, | |
| body.generated_voice_id) | |
| except tts.TTSError as exc: | |
| raise HTTPException(502, str(exc)) from exc | |
| vp = _ensure_profile(ctx) | |
| voices = dict(vp.host_voices or {}) | |
| voices["custom"] = [*(voices.get("custom") or []), | |
| {"voice_id": voice_id, "name": body.name, "description": body.description[:300]}] | |
| if body.use_as in ("host_a", "host_b"): | |
| voices[body.use_as] = voice_id | |
| vp.host_voices = voices | |
| save_notestack_voice(ctx.db, ctx.workspace.id, voice_id, body.name) # a voice made is a voice saved | |
| ctx.db.commit() | |
| return {"voice_id": voice_id, **_serialize(ctx)} | |
| def reading_script(ctx: Ctx = Depends(get_ctx)): | |
| """About two minutes of the writer's own prose to read aloud for the clone (their words, their rhythm).""" | |
| doc = ctx.db.scalar( | |
| select(Document).where(Document.workspace_id == ctx.workspace.id, Document.clean_text != "") | |
| .order_by(func.length(Document.clean_text).desc()).limit(1) | |
| ) | |
| if not doc: | |
| text = ("Writing is how I think out loud. I start with a question I cannot answer yet, then I follow it " | |
| "until the answer surprises me. Some weeks that takes a paragraph; some weeks it takes a month. " | |
| "What matters is that every sentence earns its place, and that the reader leaves with something " | |
| "they did not have when they arrived.") | |
| return {"title": "Sample passage", "text": text, "words": len(text.split())} | |
| paragraphs = [p for p in re.split(r"\n\s*\n", doc.clean_text) if len(p.split()) > 12 and not p.startswith("## ")] | |
| picked, words = [], 0 | |
| for p in paragraphs: | |
| picked.append(p.strip()) | |
| words += len(p.split()) | |
| if words >= 280: # roughly two minutes read aloud | |
| break | |
| if not picked: # short or list heavy posts: read the text as is | |
| plain = re.sub(r"^## .*$", "", doc.clean_text, flags=re.M).split() | |
| picked, words = [" ".join(plain[:280])], min(len(plain), 280) | |
| return {"title": doc.title, "text": "\n\n".join(picked), "words": words} | |
| async def quota(ctx: Ctx = Depends(get_ctx)): | |
| if not settings.elevenlabs_api_key: | |
| return {} | |
| return await run_in_threadpool(tts.subscription) | |
| def give_consent(body: ConsentIn, request: Request, ctx: Ctx = Depends(get_ctx)): | |
| plan = effective_plan(ctx.db, ctx.workspace) | |
| if not plan.voice_cloning: | |
| raise plan_limit_error(plan, "voice_cloning", "Voice cloning is on the Writer and Studio plans.") | |
| if not body.agreed or body.consent_text.strip() != CONSENT_TEXT: | |
| raise HTTPException(400, "Please read and accept the consent statement.") | |
| if not settings.elevenlabs_api_key: | |
| raise HTTPException(503, "Voice cloning needs ELEVENLABS_API_KEY in .env") | |
| ids = body.upload_ids or ([body.upload_id] if body.upload_id else []) | |
| uploads = ctx.db.scalars(select(Upload).where(Upload.id.in_(ids), Upload.workspace_id == ctx.workspace.id)).all() | |
| uploads = [u for u in uploads if u.status == "complete" and u.content_type.startswith("audio/")] | |
| if not uploads: | |
| raise HTTPException(400, "Record or upload at least one audio sample first.") | |
| # A new clone replaces the previous one. | |
| old = _consent(ctx) | |
| if old: | |
| old.revoked_at = datetime.now(UTC) | |
| consent = VoiceConsent(workspace_id=ctx.workspace.id, user_id=ctx.user.id, sample_key=uploads[0].key, | |
| consent_text=CONSENT_TEXT, ip_address=request.client.host if request.client else None) | |
| ctx.db.add(consent) | |
| ctx.db.commit() | |
| job = create_job(ctx.db, ctx.workspace.id, "voice_clone", { | |
| "consent_id": str(consent.id), "sample_keys": [u.key for u in uploads], | |
| "remove_background_noise": body.remove_background_noise, | |
| "replaces_voice_id": old.elevenlabs_voice_id if old else None, | |
| }, max_attempts=1) | |
| return {"job": serialize_job(job), **_serialize(ctx)} | |
| async def revoke_consent(ctx: Ctx = Depends(get_ctx)): | |
| consent = _consent(ctx) | |
| if not consent: | |
| return _serialize(ctx) | |
| if consent.elevenlabs_voice_id and settings.elevenlabs_api_key: | |
| try: | |
| await run_in_threadpool(tts.delete_voice, consent.elevenlabs_voice_id) | |
| except tts.TTSError: | |
| pass | |
| consent.revoked_at = datetime.now(UTC) | |
| storage.delete(consent.sample_key) | |
| vp = _profile(ctx) | |
| if vp and vp.host_voices: | |
| vp.host_voices = {k: v for k, v in vp.host_voices.items() if v != consent.elevenlabs_voice_id} | |
| if consent.elevenlabs_voice_id: | |
| forget_video_voice(ctx.db, ctx.workspace.id, consent.elevenlabs_voice_id) | |
| ctx.db.commit() | |
| return _serialize(ctx) | |