import json import os import sys from collections import defaultdict from argparse import ArgumentParser CURRENT_DIR = os.path.dirname(__file__) PROJECT_ROOT = os.path.abspath(os.path.join(CURRENT_DIR, "..")) if PROJECT_ROOT not in sys.path: sys.path.insert(0, PROJECT_ROOT) from elevenlabs import ElevenLabs from app.database import SessionLocal from app.services import elevenlabs_keys from app.constants import FREE_PREMADE_VOICE_IDS as KNOWN_PREMADE_VOICE_IDS from app.constants import FREE_PREMADE_FALLBACK from app.models.prebuilt_voice import PrebuiltVoice from app.models.saved_voice import SavedVoice from app.models.user import User from app.services.elevenlabs_voice_design import get_voice_metadata TARGET_PREBUILT_CATALOG_SIZE = 30 def _fetch_live_premade_catalog() -> list[dict]: api_key = elevenlabs_keys.get_voice_design_api_key() elevenlabs_keys.log_key_usage(api_key, "sync_prebuilt_voices catalog fetch") client = ElevenLabs(api_key=api_key) try: voices_response = client.voices.get_all(show_legacy=True) except TypeError: voices_response = client.voices.get_all() rows: list[dict] = [] for v in voices_response.voices: voice_id = getattr(v, "voice_id", None) or getattr(v, "id", None) if not voice_id: continue category = getattr(v, "category", None) if category != "premade" and voice_id not in KNOWN_PREMADE_VOICE_IDS: continue labels = getattr(v, "labels", None) or {} labels_json = json.dumps(labels) if isinstance(labels, dict) else "{}" rows.append( { "voice_id": voice_id, "name": getattr(v, "name", None) or "", "preview_url": getattr(v, "preview_url", None), "labels_json": labels_json, "description": (getattr(v, "description", None) or "") or None, "plan": "free" if voice_id in KNOWN_PREMADE_VOICE_IDS else "paid", } ) return rows def _build_row_from_metadata(voice_id: str, fallback_name: str | None = None) -> dict: meta = get_voice_metadata(voice_id) or {} labels = meta.get("labels", {}) if isinstance(meta.get("labels"), dict) else {} return { "voice_id": voice_id, "name": (meta.get("name") or fallback_name or voice_id), "preview_url": (meta.get("preview_url") or None), "labels_json": json.dumps(labels) if isinstance(labels, dict) else "{}", "description": (meta.get("description") or "") or None, "plan": "free" if voice_id in KNOWN_PREMADE_VOICE_IDS else "paid", } def _choose_free_voice_ids(rows: list[PrebuiltVoice], target_free_count: int = 4) -> set[str]: """Pick exactly target_free_count voice IDs to mark as free.""" preferred = [vid for vid in KNOWN_PREMADE_VOICE_IDS if any(r.voice_id == vid for r in rows)] chosen = list(preferred[:target_free_count]) if len(chosen) < target_free_count: remaining = sorted( [r.voice_id for r in rows if r.voice_id not in chosen], key=lambda v: v, ) chosen.extend(remaining[: target_free_count - len(chosen)]) return set(chosen) def sync_prebuilt_voices(dry_run: bool = False, rebuild_catalog: bool = False) -> dict[str, int]: db = SessionLocal() updated_prebuilt = 0 scanned_prebuilt = 0 updated_saved = 0 scanned_saved = 0 failed = 0 deduped_prebuilt = 0 deduped_saved = 0 inserted_prebuilt = 0 deleted_prebuilt = 0 collapsed_alias_prebuilt = 0 plan_adjusted_prebuilt = 0 plan_adjusted_saved = 0 replaced_saved_prebuilt_users = 0 replaced_saved_prebuilt_deleted = 0 replaced_saved_prebuilt_inserted = 0 try: if rebuild_catalog: live_rows = _fetch_live_premade_catalog() # Collapse exact metadata duplicates (aliases): same visible payload, different voice_id. # Keep the first deterministic voice_id (lexicographically) for stable catalogs. grouped: dict[tuple[str, str, str, str], list[dict]] = defaultdict(list) for row in live_rows: key = ( row["name"] or "", row["preview_url"] or "", row["labels_json"] or "{}", row["description"] or "", ) grouped[key].append(row) canonical_rows: list[dict] = [] alias_pool: list[dict] = [] for group in grouped.values(): # Never drop configured free IDs when aliases share the same metadata. preferred = [r for r in group if r["voice_id"] in KNOWN_PREMADE_VOICE_IDS] if preferred: keep = sorted(preferred, key=lambda r: str(r["voice_id"]))[0] else: keep = sorted(group, key=lambda r: str(r["voice_id"]))[0] canonical_rows.append(keep) for item in group: if item["voice_id"] != keep["voice_id"]: alias_pool.append(item) collapsed_alias_prebuilt += max(0, len(group) - 1) # Ensure 4 configured free IDs always exist in catalog. canonical_by_id = {r["voice_id"]: r for r in canonical_rows} for item in FREE_PREMADE_FALLBACK: vid = item["voice_id"] if vid in canonical_by_id: continue try: row = _build_row_from_metadata(vid, fallback_name=item["name"]) except Exception: row = { "voice_id": vid, "name": item["name"], "preview_url": None, "labels_json": "{}", "description": None, "plan": "free", } canonical_rows.append(row) canonical_by_id[vid] = row # Ensure rebuild catalog has at least TARGET_PREBUILT_CATALOG_SIZE rows. if len(canonical_rows) < TARGET_PREBUILT_CATALOG_SIZE: for item in sorted(alias_pool, key=lambda r: str(r["voice_id"])): if item["voice_id"] in canonical_by_id: continue canonical_rows.append(item) canonical_by_id[item["voice_id"]] = item if len(canonical_rows) >= TARGET_PREBUILT_CATALOG_SIZE: break # If more than target, trim deterministically but keep the 4 free IDs. if len(canonical_rows) > TARGET_PREBUILT_CATALOG_SIZE: must_keep = {i["voice_id"] for i in FREE_PREMADE_FALLBACK} kept = [r for r in canonical_rows if r["voice_id"] in must_keep] rest = sorted( [r for r in canonical_rows if r["voice_id"] not in must_keep], key=lambda r: str(r["voice_id"]), ) remaining_slots = max(0, TARGET_PREBUILT_CATALOG_SIZE - len(kept)) canonical_rows = kept + rest[:remaining_slots] existing = db.query(PrebuiltVoice).all() existing_by_voice_id = {r.voice_id: r for r in existing} live_voice_ids = {r["voice_id"] for r in canonical_rows} # Delete rows not present in live canonical catalog. for row in existing: if row.voice_id not in live_voice_ids: db.delete(row) deleted_prebuilt += 1 # Upsert live rows. for row in canonical_rows: cur = existing_by_voice_id.get(row["voice_id"]) if cur is None: db.add( PrebuiltVoice( voice_id=row["voice_id"], name=row["name"], preview_url=row["preview_url"], labels=row["labels_json"], description=row["description"], plan=row["plan"], ) ) inserted_prebuilt += 1 continue changed = False if cur.name != row["name"]: cur.name = row["name"] changed = True if cur.preview_url != row["preview_url"]: cur.preview_url = row["preview_url"] changed = True if cur.labels != row["labels_json"]: cur.labels = row["labels_json"] changed = True if cur.description != row["description"]: cur.description = row["description"] changed = True if cur.plan != row["plan"]: cur.plan = row["plan"] changed = True if changed: updated_prebuilt += 1 # Deduplicate prebuilt_voices by voice_id (keep most recently updated, fallback highest id) pb_rows = db.query(PrebuiltVoice).all() pb_groups: dict[str, list[PrebuiltVoice]] = defaultdict(list) for row in pb_rows: pb_groups[row.voice_id].append(row) for voice_id, group in pb_groups.items(): if len(group) <= 1: continue keep = sorted( group, key=lambda r: ( r.updated_at.isoformat() if getattr(r, "updated_at", None) else "", r.id, ), reverse=True, )[0] for row in group: if row.id == keep.id: continue db.delete(row) deduped_prebuilt += 1 print(f"[sync-prebuilt] dedup prebuilt voice_id={voice_id}: removed={len(group)-1}, kept_id={keep.id}") # Enforce exactly 4 free voices in prebuilt_voices. pb_rows = db.query(PrebuiltVoice).all() free_ids = _choose_free_voice_ids(pb_rows, target_free_count=4) for row in pb_rows: desired_plan = "free" if row.voice_id in free_ids else "paid" if row.plan != desired_plan: row.plan = desired_plan plan_adjusted_prebuilt += 1 # Deduplicate saved prebuilt voices by (user_id, voice_id) (keep newest id) sv_rows = ( db.query(SavedVoice) .filter(SavedVoice.source == "prebuilt") .all() ) sv_groups: dict[tuple[int, str], list[SavedVoice]] = defaultdict(list) for row in sv_rows: sv_groups[(row.user_id, row.voice_id)].append(row) for key, group in sv_groups.items(): if len(group) <= 1: continue keep = sorted( group, key=lambda r: ( r.created_at.isoformat() if getattr(r, "created_at", None) else "", r.id, ), reverse=True, )[0] for row in group: if row.id == keep.id: continue db.delete(row) deduped_saved += 1 print( f"[sync-prebuilt] dedup saved user_id={key[0]} voice_id={key[1]}: removed={len(group)-1}, kept_id={keep.id}" ) # Keep saved prebuilt rows aligned with the 4-free policy. # Replace each user's prebuilt saved voices with the current 4 free voices. free_rows = ( db.query(PrebuiltVoice) .filter(PrebuiltVoice.voice_id.in_(free_ids)) .all() ) free_rows_by_id = {r.voice_id: r for r in free_rows} free_ids_in_seed_order = [item["voice_id"] for item in FREE_PREMADE_FALLBACK if item["voice_id"] in free_ids] # If any selected free id is not in fallback ordering, append deterministically. for vid in sorted(list(free_ids)): if vid not in free_ids_in_seed_order: free_ids_in_seed_order.append(vid) users = db.query(User.id).all() for (uid,) in users: existing_prebuilt = ( db.query(SavedVoice) .filter(SavedVoice.user_id == uid, SavedVoice.source == "prebuilt") .all() ) if existing_prebuilt: for row in existing_prebuilt: db.delete(row) replaced_saved_prebuilt_deleted += len(existing_prebuilt) for vid in free_ids_in_seed_order: row = free_rows_by_id.get(vid) fallback = next((f for f in FREE_PREMADE_FALLBACK if f["voice_id"] == vid), None) db.add( SavedVoice( user_id=uid, voice_id=vid, name=(row.name if row else (fallback["name"] if fallback else vid)), preview_url=(row.preview_url if row else None), source="prebuilt", plan="free", gender=None, accent=None, description=(row.description if row else None), custom_voice_id=None, ) ) replaced_saved_prebuilt_inserted += 1 replaced_saved_prebuilt_users += 1 rows = db.query(PrebuiltVoice).order_by(PrebuiltVoice.name).all() metadata_by_voice_id: dict[str, dict] = {} for row in rows: scanned_prebuilt += 1 try: meta = get_voice_metadata(row.voice_id) or {} metadata_by_voice_id[row.voice_id] = meta labels = meta.get("labels", {}) if isinstance(meta.get("labels"), dict) else {} next_preview = (meta.get("preview_url") or "").strip() or None next_description = (meta.get("description") or "").strip() if isinstance(meta.get("description"), str) else None next_name = (meta.get("name") or "").strip() if isinstance(meta.get("name"), str) else None changed = False if next_preview and row.preview_url != next_preview: row.preview_url = next_preview changed = True if labels: labels_json = json.dumps(labels) if row.labels != labels_json: row.labels = labels_json changed = True if next_description is not None and row.description != next_description: row.description = next_description changed = True if next_name and row.name != next_name: row.name = next_name changed = True if changed: updated_prebuilt += 1 except Exception as exc: failed += 1 print(f"[sync-prebuilt] failed voice_id={row.voice_id}: {exc}") # Also sync saved voices that are prebuilt copies. saved_rows = ( db.query(SavedVoice) .filter(SavedVoice.source == "prebuilt") .all() ) for row in saved_rows: scanned_saved += 1 try: meta = metadata_by_voice_id.get(row.voice_id) if meta is None: meta = get_voice_metadata(row.voice_id) or {} metadata_by_voice_id[row.voice_id] = meta labels = meta.get("labels", {}) if isinstance(meta.get("labels"), dict) else {} next_preview = (meta.get("preview_url") or "").strip() or None next_gender = (labels.get("gender") or "").strip() if isinstance(labels.get("gender"), str) else None next_accent = (labels.get("accent") or "").strip() if isinstance(labels.get("accent"), str) else None next_description = (meta.get("description") or "").strip() if isinstance(meta.get("description"), str) else None changed = False if next_preview and row.preview_url != next_preview: row.preview_url = next_preview changed = True if next_gender and row.gender != next_gender: row.gender = next_gender changed = True if next_accent and row.accent != next_accent: row.accent = next_accent changed = True if next_description is not None and row.description != next_description: row.description = next_description changed = True if changed: updated_saved += 1 except Exception as exc: failed += 1 print(f"[sync-prebuilt] failed saved voice id={row.id} voice_id={row.voice_id}: {exc}") if dry_run: db.rollback() else: db.commit() finally: db.close() return { "scanned_prebuilt": scanned_prebuilt, "updated_prebuilt": updated_prebuilt, "inserted_prebuilt": inserted_prebuilt, "deleted_prebuilt": deleted_prebuilt, "collapsed_alias_prebuilt": collapsed_alias_prebuilt, "plan_adjusted_prebuilt": plan_adjusted_prebuilt, "scanned_saved": scanned_saved, "updated_saved": updated_saved, "plan_adjusted_saved": plan_adjusted_saved, "replaced_saved_prebuilt_users": replaced_saved_prebuilt_users, "replaced_saved_prebuilt_deleted": replaced_saved_prebuilt_deleted, "replaced_saved_prebuilt_inserted": replaced_saved_prebuilt_inserted, "deduped_prebuilt": deduped_prebuilt, "deduped_saved": deduped_saved, "failed": failed, "dry_run": int(dry_run), } if __name__ == "__main__": parser = ArgumentParser(description="Refresh prebuilt metadata in prebuilt_voices and saved_voices from ElevenLabs.") parser.add_argument("--dry-run", action="store_true", help="Compute updates without writing to DB.") parser.add_argument( "--rebuild-catalog", action="store_true", help="Rebuild prebuilt_voices from live ElevenLabs premade catalog and remove stale rows.", ) args = parser.parse_args() result = sync_prebuilt_voices(dry_run=args.dry_run, rebuild_catalog=args.rebuild_catalog) print( "[sync-prebuilt] prebuilt: scanned={scanned_prebuilt} inserted={inserted_prebuilt} updated={updated_prebuilt} deleted={deleted_prebuilt} aliases_collapsed={collapsed_alias_prebuilt} deduped={deduped_prebuilt} plan_adjusted={plan_adjusted_prebuilt} | saved(prebuilt): scanned={scanned_saved} updated={updated_saved} deduped={deduped_saved} plan_adjusted={plan_adjusted_saved} replaced_users={replaced_saved_prebuilt_users} replaced_deleted={replaced_saved_prebuilt_deleted} replaced_inserted={replaced_saved_prebuilt_inserted} | failed={failed} dry_run={dry_run}".format( **result ) )