encoach_backend_v4/custom_addons/encoach_ai/services/elai_service.py

"""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"),
        }