import json import os import urllib.request import urllib.error from rest_framework import status, viewsets from rest_framework.decorators import action from rest_framework.response import Response from apps.agents.models import AgentConfig from django.db import connections from apps.cluster.dynamic_tables import _q_ident, ensure_business_cluster_tables from .models import Call from .serializers import CallCreateSerializer, CallSerializer def _post_json(url: str, payload: dict, *, timeout_s: float = 20.0) -> tuple[int, dict | None, str]: data = json.dumps(payload).encode("utf-8") req = urllib.request.Request( url, data=data, headers={"content-type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=timeout_s) as resp: raw = resp.read() try: return resp.status, json.loads(raw.decode("utf-8")), "" except Exception: return resp.status, None, raw.decode("utf-8", errors="replace")[:500] except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") if hasattr(e, "read") else str(e) return int(getattr(e, "code", 500) or 500), None, body[:500] except Exception as e: return 0, None, str(e) class CallViewSet(viewsets.ModelViewSet): queryset = Call.objects.select_related("bot").all() serializer_class = CallSerializer def get_serializer_class(self): if self.action == "create": return CallCreateSerializer return CallSerializer def create(self, request, *args, **kwargs): ser = CallCreateSerializer(data=request.data) ser.is_valid(raise_exception=True) data = ser.validated_data bot = None cluster_bot_row: dict | None = None business_id = data.get("business_id") bot_id = data.get("bot_id") # If business_id is provided, treat bot_id as a cluster `{bid}_bots.bot_id` and # resolve config from cluster DB (dynamic tables). This makes calls dynamic per # logged-in business. if business_id not in (None, "") and bot_id is not None: try: bid = int(business_id) bpk = int(bot_id) ensure_business_cluster_tables(bid) table = f"{bid}_bots" with connections["cluster"].cursor() as cur: cur.execute(f"SELECT * FROM {_q_ident(table)} WHERE bot_id = %s LIMIT 1", [bpk]) row = cur.fetchone() if row: cols = [c[0] for c in cur.description] cluster_bot_row = dict(zip(cols, row)) except Exception: cluster_bot_row = None if bot_id is not None: bot = AgentConfig.objects.filter(id=bot_id).first() custnumber = str(data["custnumber"]).strip() exenumber = (data.get("exenumber") or (bot.mcube_exenumber if bot else "") or "").strip() gid = (data.get("gid") or (bot.mcube_gid if bot else "") or "1").strip() if cluster_bot_row: exenumber = (data.get("exenumber") or str(cluster_bot_row.get("mcube_exenumber") or "")).strip() or exenumber gid = (data.get("gid") or str(cluster_bot_row.get("mcube_gid") or "")).strip() or gid payload = { "custnumber": custnumber, "exenumber": exenumber, "gid": gid, } if cluster_bot_row and business_id not in (None, "") and bot_id is not None: # Let the MCube webhook runtime fetch full bot config dynamically. payload["business_id"] = int(business_id) payload["bot_id"] = int(bot_id) # Best-effort direct overrides so it works even if the runtime can't call Django. prompt = cluster_bot_row.get("prompt") if prompt is not None and str(prompt).strip() != "": payload["system_prompt"] = str(prompt) msg_out = cluster_bot_row.get("message_outbound") if msg_out is not None and str(msg_out).strip() != "": payload["first_message"] = str(msg_out).strip() llm_model = cluster_bot_row.get("llm_model") if llm_model is not None and str(llm_model).strip() != "": payload["llm_model"] = str(llm_model).strip() stt_provider = cluster_bot_row.get("stt_provider") if stt_provider is not None and str(stt_provider).strip() != "": payload["stt_provider"] = str(stt_provider).strip() tts_provider = cluster_bot_row.get("tts_provider") if tts_provider is not None and str(tts_provider).strip() != "": payload["tts_provider"] = str(tts_provider).strip() llm_provider = cluster_bot_row.get("llm_provider") if llm_provider is not None and str(llm_provider).strip() != "": payload["llm_provider"] = str(llm_provider).strip() # voice JSON may contain voice_id used by ElevenLabs/cartesia plugins voice_raw = cluster_bot_row.get("voice") if isinstance(voice_raw, (bytes, bytearray)): voice_raw = voice_raw.decode("utf-8", errors="replace") if isinstance(voice_raw, str) and voice_raw.strip(): try: vobj = json.loads(voice_raw) if isinstance(vobj, dict): vid = (vobj.get("voice_id") or vobj.get("voiceId") or "") if vid and str(vid).strip(): payload["tts_voice_id"] = str(vid).strip() except Exception: pass # Apply bot defaults as call overrides (best-effort). if bot: if bot.system_prompt: payload["system_prompt"] = bot.system_prompt if bot.llm_model: payload["llm_model"] = bot.llm_model if bot.stt_provider: payload["stt_provider"] = bot.stt_provider if bot.tts_provider: payload["tts_provider"] = bot.tts_provider if getattr(bot, "voice_config", None) and bot.voice_config.voice_id: payload["tts_voice_id"] = bot.voice_config.voice_id if bot.first_message_outbound: payload["first_message"] = bot.first_message_outbound if bot.mcube_http_authorization: payload["HTTP_AUTHORIZATION"] = bot.mcube_http_authorization # Per-request overrides (highest priority) for k in [ "first_message", "system_prompt", "llm_model", "stt_provider", "tts_provider", "tts_model", "tts_voice_id", "tts_encoding", ]: if k in data and str(data.get(k) or "").strip() != "": payload[k] = str(data.get(k)).strip() mcube_url = os.getenv("MCUBE_OUTBOUND_CALL_URL", "http://127.0.0.1:8002/api/mcube/outbound-call") http_status, resp_json, err = _post_json(mcube_url, payload) # Never store secrets in DB metadata. payload_for_storage = dict(payload) if "HTTP_AUTHORIZATION" in payload_for_storage: payload_for_storage["HTTP_AUTHORIZATION"] = "***redacted***" call = Call.objects.create( provider=Call.Provider.MCUBE, direction=Call.Direction.OUTBOUND, bot=bot, custnumber=custnumber, exenumber=exenumber, gid=gid, status=str((resp_json or {}).get("status") or ""), provider_call_id=str((resp_json or {}).get("mcube_call_sid") or ""), metadata={"request": payload_for_storage, "response": resp_json, "http_status": http_status}, last_error=err, ) out = CallSerializer(call).data out["telephony_response"] = resp_json out["telephony_http_status"] = http_status if http_status not in (200, 201): return Response(out, status=status.HTTP_502_BAD_GATEWAY) return Response(out, status=status.HTTP_201_CREATED) @action(detail=False, methods=["get"]) def health(self, request): return Response({"ok": True})