"""Hunar AI outbound-call provider. Places a real outbound call through the Hunar API and supports live status and call summary retrieval via Hunar's GET /external/v1/calls/{call_id}/ API. """ import logging import os import re import httpx from django.conf import settings from .base import BaseProvider logger = logging.getLogger(__name__) TIMEOUT_SECONDS = 30 class HunarAPIError(RuntimeError): """A Hunar API call failed, carrying enough detail to answer the client. `status_code` is Hunar's HTTP status (or 503 when we never reached it) and `details` is Hunar's own `details` array, so the UI can show per-field validation errors instead of a generic failure. """ def __init__(self, status_code: int, message: str, details=None): super().__init__(message) self.status_code = status_code self.message = message self.details = details if details is not None else [] class HunarProvider(BaseProvider): name = "hunar" def _get_api_key(self) -> str: return (os.getenv("HUNAR_API_KEY", "") or getattr(settings, "HUNAR_API_KEY", "") or "").strip().strip('"').strip("'") def _get_agent_id(self) -> str: return (os.getenv("HUNAR_AGENT_ID", "") or getattr(settings, "HUNAR_AGENT_ID", "d57196d8-9043-4472-92eb-4b41b38caeb5") or "").strip().strip('"').strip("'") def _get_base_url(self) -> str: return (os.getenv("HUNAR_BASE_URL", "") or getattr(settings, "HUNAR_BASE_URL", "https://api.voice.hunar.ai") or "").strip().rstrip("/") def _get_headers(self) -> dict: key = self._get_api_key() return { "X-API-Key": key, "Authorization": f"Bearer {key}", "Content-Type": "application/json", } def start_call(self, ai_call) -> str: """Place the outbound call. Returns Hunar's call id (UUID).""" api_key = self._get_api_key() agent_id = self._get_agent_id() if not api_key: raise RuntimeError("HUNAR_API_KEY is not configured in environment.") if not agent_id: raise RuntimeError("HUNAR_AGENT_ID is not configured in environment.") candidate = ai_call.candidate phone = (candidate.phone_number or "").strip() if not phone: raise RuntimeError("No phone number on this candidate — cannot place a call.") if not phone.startswith("+"): phone = f"+91{phone}" if not phone.startswith("91") else f"+{phone}" base_url = (getattr(settings, "PUBLIC_BASE_URL", "") or "").rstrip("/") callback_config = None if base_url and not (base_url.startswith("http://localhost") or base_url.startswith("http://127.")): callback_config = { "call_status_callback_url": f"{base_url}/api/v1/ai-calls/webhook", "call_recording_callback_url": f"{base_url}/api/v1/ai-calls/webhook", "call_result_callback_url": f"{base_url}/api/v1/ai-calls/completed", "call_summary_callback_url": f"{base_url}/api/v1/ai-calls/completed", } request_id = f"ats-{ai_call.id}-{ai_call.attempts}" payload = { "agent_id": agent_id, "callee_name": f"{candidate.first_name} {candidate.last_name}".strip(), "mobile_number": phone, "custom_data": { "date": ai_call.created_at.strftime("%Y-%m-%d") if ai_call.created_at else "", "email": candidate.email or "", "time": ai_call.created_at.strftime("%I:%M %p") if ai_call.created_at else "", "timezone": "Asia/Kolkata", "ai_call_id": str(ai_call.id), }, "request_id": request_id, } if callback_config: payload["callback_config"] = callback_config url = f"{self._get_base_url()}/external/v1/calls/" logger.info("[HUNAR] Outbound call request → URL: %s | agent: %s | phone: %s", url, agent_id, phone) try: response = httpx.post(url, json=payload, headers=self._get_headers(), timeout=TIMEOUT_SECONDS) except httpx.HTTPError as exc: logger.exception("[HUNAR] Request failed: %s", exc) raise RuntimeError(f"Could not reach Hunar API: {exc}") from exc if response.status_code >= 400: raise RuntimeError(f"Hunar API error {response.status_code}: {response.text[:300]}") try: data = response.json() except ValueError as exc: raise RuntimeError(f"Hunar API non-JSON response: {response.text[:200]}") from exc call_id = data.get("id") or data.get("call_id") or (data.get("data") or {}).get("call_id") or (data.get("data") or {}).get("id") if not call_id: call_id = request_id logger.info("[HUNAR] Call initiated successfully — call_id=%s", call_id) return str(call_id) #: Hunar's call id is a UUID. Anything else (notably our own #: `ats--` request_id fallback) is rejected by the detail #: endpoint with HTTP 422, so it must be resolved via the list endpoint first. _UUID_RE = re.compile(r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-" r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$") def get_call_details(self, provider_call_id: str, ai_call_id=None) -> dict: """Fetch real-time status/duration/summary for one call. `GET /external/v1/calls/{uuid}/` only accepts Hunar's own UUID. When the stored id is not a UUID — which happens whenever the create response was parsed without picking up its `id` — we fall back to scanning `GET /external/v1/calls/` for the record whose `custom_data.ai_call_id` matches ours. Hunar echoes that value back, so a call is still recoverable rather than being stuck on Dialing forever. """ api_key = self._get_api_key() if not api_key: return {} if provider_call_id and self._UUID_RE.match(provider_call_id): url = f"{self._get_base_url()}/external/v1/calls/{provider_call_id}/" try: response = httpx.get(url, headers=self._get_headers(), timeout=TIMEOUT_SECONDS) if response.status_code == 200: return response.json() logger.warning("[HUNAR] GET details %s returned HTTP %s: %s", provider_call_id, response.status_code, response.text[:200]) except httpx.HTTPError as exc: logger.warning("[HUNAR] GET details request failed for %s: %s", provider_call_id, exc) return {} if ai_call_id is None: logger.warning("[HUNAR] %r is not a Hunar UUID and no ai_call_id was given " "— cannot resolve this call.", provider_call_id) return {} return self.find_call_by_ai_call_id(ai_call_id) def find_call_by_ai_call_id(self, ai_call_id, max_pages: int = 10) -> dict: """Locate a Hunar call by the `custom_data.ai_call_id` we sent with it. Used to recover calls whose Hunar UUID was never stored. Pages through the list endpoint (newest first) and stops at the first match. """ api_key = self._get_api_key() if not api_key: return {} wanted = str(ai_call_id) url = f"{self._get_base_url()}/external/v1/calls/" headers = self._get_headers() for _page in range(max_pages): try: response = httpx.get(url, headers=headers, timeout=TIMEOUT_SECONDS) except httpx.HTTPError as exc: logger.warning("[HUNAR] list request failed while resolving ai_call_id=%s: %s", wanted, exc) return {} if response.status_code != 200: logger.warning("[HUNAR] list returned HTTP %s while resolving ai_call_id=%s", response.status_code, wanted) return {} try: page = response.json() except ValueError: return {} for row in (page.get("results") or []): custom = row.get("custom_data") if isinstance(row.get("custom_data"), dict) else {} if str(custom.get("ai_call_id") or "") == wanted: logger.info("[HUNAR] Resolved ai_call_id=%s to Hunar call %s", wanted, row.get("id")) return row url = page.get("next") if not url: break logger.info("[HUNAR] No Hunar call found for ai_call_id=%s", wanted) return {} def cancel_call(self, ai_call) -> bool: """Best-effort cancel.""" api_key = self._get_api_key() if not (api_key and ai_call.provider_call_id): return False url = f"{self._get_base_url()}/external/v1/calls/{ai_call.provider_call_id}/cancel/" try: response = httpx.post(url, headers=self._get_headers(), timeout=TIMEOUT_SECONDS) return response.status_code < 400 except httpx.HTTPError: return False # ------------------------------------------------------------------ # Agents API (Hunar Voice Agents API Docs — Agents section) # # Additive: the call methods above are untouched. Hunar owns agent state; we # proxy these four operations live and store nothing locally. # GET /external/v1/agents/ # GET /external/v1/agents/{agent_id}/ # POST /external/v1/agents/ # PUT /external/v1/agents/{agent_id}/ # ------------------------------------------------------------------ #: Every documented agent field. Used to filter outbound payloads so a stray #: key from the client is never forwarded to Hunar. AGENT_WRITABLE_FIELDS = ( "name", "language", "voice_persona", "persona_name", "agent_prompt", "objective", "introduction", "result_prompt", "result_schema", "status", ) def _agents_url(self, agent_id: str | None = None) -> str: base = f"{self._get_base_url()}/external/v1/agents/" return f"{base}{agent_id}/" if agent_id else base def _require_api_key(self) -> None: if not self._get_api_key(): raise HunarAPIError(503, "HUNAR_API_KEY is not configured in environment.") def _require_agent_uuid(self, agent_id) -> str: """Hunar's agent id is a UUID; anything else is rejected with HTTP 422. Fail before the network call so the client gets a clear 404 instead.""" value = str(agent_id or "").strip() if not self._UUID_RE.match(value): raise HunarAPIError(404, "Agent not found.") return value def _agent_request(self, method: str, url: str, *, params=None, json=None) -> dict: """One Hunar agents call. Returns parsed JSON or raises HunarAPIError. Never lets an httpx error or a non-JSON body escape as an unhandled exception — the view layer turns HunarAPIError into a clean response. """ self._require_api_key() try: response = httpx.request( method, url, headers=self._get_headers(), params=params, json=json, timeout=TIMEOUT_SECONDS, ) except httpx.HTTPError as exc: logger.warning("[HUNAR] agents %s %s failed: %s", method, url, exc) raise HunarAPIError(503, f"Could not reach the Hunar API: {exc}") from exc try: payload = response.json() if response.content else {"success": True} except ValueError: payload = {"success": True} if response.status_code < 400 else None if response.status_code >= 400: message, details = "", [] if isinstance(payload, dict): message = str(payload.get("message") or payload.get("detail") or "") raw_details = payload.get("details") details = raw_details if isinstance(raw_details, list) else [] if not message: message = f"Hunar API error {response.status_code}: {response.text[:300]}" logger.warning("[HUNAR] agents %s %s -> HTTP %s: %s", method, url, response.status_code, message) raise HunarAPIError(response.status_code, message, details) if payload is None: raise HunarAPIError(502, "Hunar returned a non-JSON response.") return payload def list_agents(self, language=None, voice_persona=None, status=None, page=1, page_size=20) -> dict: """Paginated agent list. Only the filters actually supplied are sent.""" params = {"page": page, "page_size": page_size} for key, value in (("language", language), ("voice_persona", voice_persona), ("status", status)): if value: params[key] = value return self._agent_request("GET", self._agents_url(), params=params) def get_agent(self, agent_id) -> dict: """Full detail for one agent.""" return self._agent_request("GET", self._agents_url(self._require_agent_uuid(agent_id))) def create_agent(self, payload: dict) -> dict: """Create an agent. `payload` must already be validated.""" body = {k: v for k, v in (payload or {}).items() if k in self.AGENT_WRITABLE_FIELDS} return self._agent_request("POST", self._agents_url(), json=body) def update_agent(self, agent_id, payload: dict) -> dict: """Update an agent — only the supplied fields are forwarded.""" body = {k: v for k, v in (payload or {}).items() if k in self.AGENT_WRITABLE_FIELDS} return self._agent_request( "PUT", self._agents_url(self._require_agent_uuid(agent_id)), json=body, ) def delete_agent(self, agent_id) -> dict: """Delete an agent on Hunar platform.""" return self._agent_request( "DELETE", self._agents_url(self._require_agent_uuid(agent_id)) )