"""ELAI avatar video generation service.""" import logging import time _logger = logging.getLogger(__name__) try: import requests as _requests except ImportError: _requests = None ELAI_BASE = "https://apis.elai.io/api/v1" AVATAR_PRESETS = { "vadim": {"code": "vadim.casual", "gender": "male", "voice": "en-GB-RyanNeural"}, "gia": {"code": "gia.casual", "gender": "female", "voice": "en-US-AriaNeural"}, "zara": {"code": "zara.regular", "gender": "female", "voice": "en-US-JennyNeural"}, "mason": {"code": "mason.casual", "gender": "male", "voice": "en-US-GuyNeural"}, "default": {"code": "gia.casual", "gender": "female", "voice": "en-US-AriaNeural"}, } class ElaiService: """Generate avatar videos for listening exercises and instructional content.""" def __init__(self, env): self.env = env self._get_param = env["ir.config_parameter"].sudo().get_param def _get_token(self): token = self._get_param("encoach_ai.elai_token", "") if not token: import os token = os.environ.get("ELAI_TOKEN", "") if not token: raise RuntimeError("ELAI token not configured — set in AI Settings") return token def _headers(self): return { "Authorization": f"Bearer {self._get_token()}", "Content-Type": "application/json", "Accept": "application/json", } def _log(self, action, latency, status="success", error=None): try: self.env["encoach.ai.log"].sudo().create({ "service": "elai", "action": action, "latency_ms": latency, "status": status, "error_message": error, }) except Exception: pass def _resolve_avatar(self, avatar_id): if not avatar_id or avatar_id == "default": return AVATAR_PRESETS["default"] key = avatar_id.split(".")[0].lower() if "." in avatar_id else avatar_id.lower() if key in AVATAR_PRESETS: return AVATAR_PRESETS[key] return {"code": avatar_id, "gender": "female", "voice": "en-US-AriaNeural"} def list_avatars(self): """List available ELAI avatars.""" if not _requests: raise RuntimeError("requests package not installed") resp = _requests.get(f"{ELAI_BASE}/avatars", headers=self._headers(), timeout=15) resp.raise_for_status() return resp.json() @staticmethod def _split_script(script, max_chars=500): """Split a long script into chunks that stay under ELAI's 60s/slide limit. Uses ~10 chars/second heuristic, so 500 chars ~ 50s (with safety margin). Splits on sentence boundaries ('. ', '? ', '! ', newlines). """ if len(script) <= max_chars: return [script] import re sentences = re.split(r'(?<=[.!?])\s+|\n+', script) chunks, current = [], "" for sent in sentences: if not sent.strip(): continue if current and len(current) + len(sent) + 1 > max_chars: chunks.append(current.strip()) current = sent else: current = f"{current} {sent}" if current else sent if current.strip(): chunks.append(current.strip()) return chunks or [script] def create_video(self, script, *, avatar_id=None, title="EnCoach Video", language="en"): """Create an avatar video from a script. Automatically splits long scripts into multiple slides to stay under ELAI's 60-second-per-slide limit. Returns: dict with 'video_id', 'status' """ if not _requests: raise RuntimeError("requests package not installed") avatar_preset = self._resolve_avatar(avatar_id) avatar_code = avatar_preset["code"] avatar_src = f"https://elai-avatars.s3.us-east-2.amazonaws.com/common/{avatar_code.replace('.', '/')}/{avatar_code.replace('.', '_')}.png" chunks = self._split_script(script) slides = [] for idx, chunk in enumerate(chunks, 1): slides.append({ "id": idx, "speech": chunk, "language": "English", "voice": avatar_preset["voice"], "voiceType": "text", "voiceProvider": "azure", "avatar": { "code": avatar_code, "gender": avatar_preset["gender"], }, "canvas": { "version": "4.4.0", "background": "#ffffff", "objects": [ { "type": "avatar", "left": 151.5, "top": 36, "fill": "#4868FF", "scaleX": 0.3, "scaleY": 0.3, "height": 1080, "width": 1080, "src": avatar_src, } ], }, }) _logger.info("ELAI: creating video with %d slide(s) from %d chars", len(slides), len(script)) payload = { "name": title, "slides": slides, "tags": ["encoach"], } t0 = time.time() try: resp = _requests.post( f"{ELAI_BASE}/videos", json=payload, headers=self._headers(), timeout=30, ) resp.raise_for_status() data = resp.json() video_id = data.get("_id", data.get("id")) # Step 2: trigger rendering -- ELAI requires a separate render call try: render_resp = _requests.post( f"{ELAI_BASE}/videos/render/{video_id}", headers=self._headers(), timeout=30, ) render_resp.raise_for_status() _logger.info("ELAI render triggered for video %s", video_id) except Exception as render_err: _logger.warning("ELAI render trigger failed for %s: %s (video created but not rendered)", video_id, render_err) self._log("create_video", int((time.time() - t0) * 1000)) return {"video_id": video_id, "status": "rendering"} except _requests.exceptions.HTTPError as exc: body = "" try: body = exc.response.text[:500] except Exception: pass _logger.error("ELAI create_video failed: %s — %s", exc, body) self._log("create_video", int((time.time() - t0) * 1000), "error", f"{exc} | {body}") raise RuntimeError(f"ELAI API error: {exc.response.status_code} — {body}") from exc except Exception as exc: self._log("create_video", int((time.time() - t0) * 1000), "error", str(exc)) raise def get_video_status(self, video_id): """Check video generation status.""" if not _requests: raise RuntimeError("requests package not installed") resp = _requests.get( f"{ELAI_BASE}/videos/{video_id}", headers=self._headers(), timeout=15, ) resp.raise_for_status() data = resp.json() video_url = data.get("url", "") or data.get("video_url", "") return { "video_id": video_id, "status": data.get("status", "unknown"), "url": video_url, "video_url": video_url, "duration": data.get("duration"), }