""" GOLS - D-ID hosted talking-avatar integration ============================================== Turns a still photo + our generated voice into a realistic talking video (lips, subtle head/eye motion) via the D-ID API — no local GPU needed. Flow: 1. Upload the source image -> POST /images -> image url 2. Upload our voice mp3 -> POST /audios -> audio url 3. Create a talk (image+audio) -> POST /talks -> talk id 4. Poll until status == "done" -> GET /talks/id -> result_url (mp4) 5. Download the result mp4 Auth: D-ID uses HTTP Basic. Put your key in backend/.env as DID_KEY. Accepts either the ready-made token from the dashboard ("Basic ") or a raw "email:api_key" pair (auto-encoded here). Requires the `requests` package (falls back to httpx). Network calls are made by YOUR backend to api.d-id.com. """ from __future__ import annotations import base64 import os import time from typing import Optional from app.config import cfg from app.utils.logger import log BASE_URL = "https://api.d-id.com" class DIDError(Exception): pass # ── HTTP helper (requests preferred, httpx fallback, urllib last) ───────────── def _http(): try: import requests return ("requests", requests) except Exception: pass try: import httpx return ("httpx", httpx) except Exception: pass return (None, None) def is_enabled() -> bool: return bool((cfg.DID_KEY or "").strip()) and _http()[0] is not None def _auth_header() -> str: key = (cfg.DID_KEY or "").strip() # If it's a raw "email:api_key" pair, base64-encode it; otherwise use as-is. if ":" in key and "=" not in key[-3:]: key = base64.b64encode(key.encode()).decode() return f"Basic {key}" def _headers(json_ct: bool = False) -> dict: h = {"Authorization": _auth_header(), "accept": "application/json"} if json_ct: h["content-type"] = "application/json" return h # ── Low-level requests/httpx wrappers ───────────────────────────────────────── def _post_json(path: str, payload: dict) -> dict: name, lib = _http() url = f"{BASE_URL}{path}" if name == "requests": r = lib.post(url, json=payload, headers=_headers(True), timeout=60) if r.status_code >= 300: raise DIDError(f"POST {path} -> {r.status_code}: {r.text[:300]}") return r.json() r = lib.post(url, json=payload, headers=_headers(True), timeout=60) if r.status_code >= 300: raise DIDError(f"POST {path} -> {r.status_code}: {r.text[:300]}") return r.json() def _get_json(path: str) -> dict: name, lib = _http() url = f"{BASE_URL}{path}" r = lib.get(url, headers=_headers(), timeout=60) if r.status_code >= 300: raise DIDError(f"GET {path} -> {r.status_code}: {r.text[:300]}") return r.json() def _upload(path: str, field: str, endpoint: str) -> str: """Multipart upload a file, return the hosted url D-ID gives back.""" name, lib = _http() url = f"{BASE_URL}{endpoint}" fname = os.path.basename(path) with open(path, "rb") as fh: files = {field: (fname, fh)} r = lib.post(url, files=files, headers=_headers(), timeout=120) if r.status_code >= 300: raise DIDError(f"upload {endpoint} -> {r.status_code}: {r.text[:300]}") data = r.json() u = data.get("url") if not u: raise DIDError(f"upload {endpoint}: no url in response {data}") return u def _download(url: str, out_path: str) -> None: name, lib = _http() if name == "requests": with lib.get(url, stream=True, timeout=180) as r: r.raise_for_status() with open(out_path, "wb") as f: for chunk in r.iter_content(8192): f.write(chunk) else: # httpx with lib.stream("GET", url, timeout=180) as r: r.raise_for_status() with open(out_path, "wb") as f: for chunk in r.iter_bytes(): f.write(chunk) # D-ID built-in (Microsoft) voices for the text-script fallback _DID_VOICES = { "en": ("en-US-JennyNeural", "en-US-GuyNeural"), "hi": ("hi-IN-SwaraNeural", "hi-IN-MadhurNeural"), "mr": ("mr-IN-AarohiNeural", "mr-IN-ManoharNeural"), "ta": ("ta-IN-PallaviNeural", "ta-IN-ValluvarNeural"), "te": ("te-IN-ShrutiNeural", "te-IN-MohanNeural"), "gu": ("gu-IN-DhwaniNeural", "gu-IN-NiranjanNeural"), "bn": ("bn-IN-TanishaaNeural", "bn-IN-BashkarNeural"), "kn": ("kn-IN-SapnaNeural", "kn-IN-GaganNeural"), "ml": ("ml-IN-SobhanaNeural", "ml-IN-MidhunNeural"), "pa": ("pa-IN-OjasNeural", "pa-IN-OjasNeural"), "ur": ("ur-IN-GulNeural", "ur-IN-SalmanNeural"), "es": ("es-ES-ElviraNeural", "es-ES-AlvaroNeural"), "fr": ("fr-FR-DeniseNeural", "fr-FR-HenriNeural"), "de": ("de-DE-KatjaNeural", "de-DE-ConradNeural"), "ja": ("ja-JP-NanamiNeural", "ja-JP-KeitaNeural"), "zh": ("zh-CN-XiaoxiaoNeural", "zh-CN-YunxiNeural"), } def _did_voice(language: str, gender: str) -> str: fem, male = _DID_VOICES.get((language or "en").lower(), _DID_VOICES["en"]) return fem if (gender or "").lower() == "female" else male def check() -> dict: """Live diagnostic: verify the key works and report remaining credits.""" if _http()[0] is None: return {"enabled": False, "ok": False, "detail": "'requests' not installed — run: pip install requests"} if not (cfg.DID_KEY or "").strip(): return {"enabled": False, "ok": False, "detail": "DID_KEY not set in backend/.env"} try: data = _get_json("/credits") remaining = data.get("remaining") return {"enabled": True, "ok": True, "detail": f"connected — {remaining} credits remaining", "credits": remaining} except Exception as e: return {"enabled": True, "ok": False, "detail": str(e)} def generate_talk(image_path: str, audio_path: str, out_path: str, *, script_text: Optional[str] = None, gender: str = "female", language: str = "en", poll_timeout: int = 300, poll_interval: float = 2.0): """ Create a talking video from `image_path`. Tries our uploaded audio first; if the account/plan rejects audio scripts, falls back to a D-ID text script using a matching Microsoft neural voice. Returns (ok: bool, detail: str). On success out_path is written. """ if _http()[0] is None: return False, "'requests' not installed — run: pip install requests" if not (cfg.DID_KEY or "").strip(): return False, "DID_KEY not set in backend/.env" if not (image_path and os.path.isfile(image_path)): return False, f"source image missing: {image_path}" try: log.info("[d-id] uploading image …") image_url = _upload(image_path, "image", "/images") # Build the script: prefer our own audio; fall back to D-ID TTS text. script = None if audio_path and os.path.isfile(audio_path): try: audio_url = _upload(audio_path, "audio", "/audios") script = {"type": "audio", "audio_url": audio_url} except Exception as e: log.warning("[d-id] audio upload failed (%s) — trying text script", e) if script is None: if not script_text: return False, "audio rejected and no script text for text fallback" script = {"type": "text", "input": script_text[:3000], "provider": {"type": "microsoft", "voice_id": _did_voice(language, gender)}} log.info("[d-id] creating talk (%s) …", script["type"]) # NOTE on watermarks: D-ID burns its own logo into output ONLY on the # free/trial plan. There is no API/config flag that removes it on the # trial — it is a plan limitation. To get clean (un-watermarked) MP4s # the account behind DID_KEY must be on a paid API plan (Build/Launch or # higher). `result_format` is set explicitly below for a predictable mp4; # it does not control the trial watermark. created = _post_json("/talks", {"source_url": image_url, "script": script, "config": {"stitch": True, "result_format": "mp4"}}) talk_id = created.get("id") if not talk_id: return False, f"no talk id in response: {created}" log.info("[d-id] talk %s queued — polling …", talk_id) deadline = time.time() + poll_timeout while time.time() < deadline: info = _get_json(f"/talks/{talk_id}") status = info.get("status") if status == "done": result_url = info.get("result_url") if not result_url: return False, "talk done but no result_url" _download(result_url, out_path) if os.path.isfile(out_path) and os.path.getsize(out_path) > 1000: log.info("[d-id] done -> %s", out_path) return True, "ok" return False, "downloaded file empty" if status in ("error", "rejected"): return False, f"talk failed: {info.get('error') or info}" time.sleep(poll_interval) return False, f"talk {talk_id} timed out after {poll_timeout}s" except Exception as e: log.error("[d-id] %s", e) return False, str(e)