import json import os from uuid import uuid4 from collections import defaultdict from typing import Any, Dict, List, Optional from django.db import connections from rest_framework import status, viewsets from rest_framework.decorators import action from rest_framework.response import Response from .models import AgentConfig from .serializers import AgentConfigCreateSerializer, AgentConfigSerializer from apps.cluster.dynamic_tables import _q_ident, ensure_business_cluster_tables from apps.livekit.cli import list_cloud_agent_deployments, prepare_workdir_for_bot, run_lk_agent import httpx def _truthy_col(v: Any) -> bool: return v in (1, True, "1", "true", "True") def _safe_int(v: Any, default: int) -> int: try: return int(v) except (TypeError, ValueError): return default def _parse_transfer_destinations(val: Any) -> list: if val is None: return [] if isinstance(val, list): return val if isinstance(val, str): try: parsed = json.loads(val) return parsed if isinstance(parsed, list) else [] except Exception: return [] return [] def _enrich_cluster_bot_response(row: dict) -> dict: """ Cluster rows store features as flat columns; clients expect nested objects matching POST shape. Build `end_conversation`, `skip_turn`, etc. without putting them back into `platform_settings`. """ out = dict(row) voice_raw = out.get("voice") if isinstance(voice_raw, str): try: out["voice"] = json.loads(voice_raw) except Exception: pass ps = out.get("platform_settings") if isinstance(ps, str): try: out["platform_settings"] = json.loads(ps) except Exception: pass # --- end_conversation --- if not isinstance(out.get("end_conversation"), dict): out["end_conversation"] = { "enabled": _truthy_col(out.get("end_conversation")), "description": (out.get("end_conversation_description") or "") or "", "prompt": (out.get("end_conversation_prompt") or "") or "", "timeout": _safe_int(out.get("end_conversation_timeout"), 10), "disable_interruptions": _truthy_col(out.get("end_conversation_disable_interruptions")), } # --- skip_turn --- if not isinstance(out.get("skip_turn"), dict): out["skip_turn"] = { "enabled": _truthy_col(out.get("skip_turn")), "description": (out.get("skip_turn_description") or "") or "", "prompt": (out.get("skip_turn_prompt") or "") or "", "disable_interruptions": _truthy_col(out.get("skip_turn_disable_interruptions")), } # --- voicemail_detection --- if not isinstance(out.get("voicemail_detection"), dict): out["voicemail_detection"] = { "enabled": _truthy_col(out.get("voicemail_detection")), "description": (out.get("voicemail_description") or "") or "", "prompt": (out.get("voicemail_detection_prompt") or "") or "", "disable_interruptions": _truthy_col(out.get("voicemail_detection_disable_interruptions")), } # --- detect_language (column is tinyint; nested object replaces it) --- if not isinstance(out.get("detect_language"), dict): out["detect_language"] = { "enabled": _truthy_col(out.get("detect_language")), "description": (out.get("detect_language_description") or "") or "", "prompt": (out.get("detect_language_prompt") or "") or "", "disable_interruptions": _truthy_col(out.get("detect_language_disable_interruptions")), } # --- transfer_to_number (column `transfer_to_number` holds phone string) --- phone_col = out.get("transfer_to_number") if isinstance(phone_col, dict): phone_str = "" else: phone_str = str(phone_col).strip() if phone_col is not None else "" destinations = _parse_transfer_destinations(out.get("transfer_destinations")) if not phone_str: phone_str = _transfer_phone_for_column({"destinations": destinations}, {}) if not isinstance(out.get("transfer_to_number"), dict): out["transfer_to_number"] = { "enabled": _truthy_col(out.get("transfer_enabled")), "description": (out.get("transfer_number_description") or "") or "", "prompt": (out.get("transfer_to_number_prompt") or "") or "", "destination_type": (out.get("transfer_destination_type") or "phone") or "phone", "phone_number": phone_str, "destinations": destinations, "disable_interruptions": _truthy_col(out.get("transfer_to_number_disable_interruptions")), } adv = out.get("advanced_settings") if isinstance(adv, str): try: adv = json.loads(adv) except Exception: adv = {} elif adv is None: adv = {} elif not isinstance(adv, dict): adv = {} fk = out.get("filler_keywords") if fk is not None: adv["keywords"] = str(fk).strip() elif not adv.get("keywords"): adv["keywords"] = "" out["advanced_settings"] = adv raw_cc = out.get("concurrent_calls") if not isinstance(raw_cc, dict): lim = raw_cc if lim is None: lim = -1 out["concurrent_calls"] = {"limit": lim} return out def _normalize_cluster_bot(row: dict) -> dict: """Align cluster `{bid}_bots` rows with AgentConfig-shaped API (`id`, `name`).""" out = _enrich_cluster_bot_response(dict(row)) bid = out.get("bot_id") if bid is not None: out["id"] = bid bot_name = out.get("bot_name") or out.get("name") if bot_name is not None: out["name"] = bot_name return out def _voice_config_from_request(request_data: dict, validated: Optional[Dict[str, Any]] = None) -> dict: voice_cfg = (request_data.get("voice_config") if request_data else None) or ( (validated or {}).get("voice_config") if validated else None ) if isinstance(voice_cfg, str): try: voice_cfg = json.loads(voice_cfg) except Exception: voice_cfg = {} if not isinstance(voice_cfg, dict): voice_cfg = {} return voice_cfg VOICE_TAB_PLATFORM_KEYS = ( "text_normalisation_type", "optimize_streaming_latency", "streaming_latency_optimization", "pronunciation_dictionary_path", "stability", "speed", "similarity", ) def _voice_tab_settings_for_platform_settings(voice_cfg: dict, bs: dict) -> Dict[str, Any]: """Voice tab controls stored under `platform_settings.voice` (subset of full `voice` column).""" merged: Dict[str, Any] = dict(bs.get("voice") or {}) for k, v in (voice_cfg or {}).items(): merged[k] = v out: Dict[str, Any] = {} for k in VOICE_TAB_PLATFORM_KEYS: if k not in merged: continue v = merged.get(k) if v is None: continue if isinstance(v, str) and v.strip() == "": continue out[k] = v return out def _normalize_transfer_destination_type(val: Any) -> str: if val is None: return "" s = str(val).lower().strip().replace(" ", "_") if s in ("phone", "phone_number"): return "phone" if s in ("phone_dynamic_variable", "phone_number_dynamic", "phone_as_dynamic_variable"): return "phone_dynamic_variable" if s in ("sip_uri",): return "sip_uri" if s in ("sip_uri_dynamic_variable", "sip_uri_dynamic"): return "sip_uri_dynamic_variable" return s def _parse_destinations_list(raw: Any) -> list: if raw is None: return [] if isinstance(raw, list): return raw if isinstance(raw, str): try: parsed = json.loads(raw) return parsed if isinstance(parsed, list) else [] except Exception: return [] return [] def _transfer_phone_for_column(tn: dict, features: dict) -> str: """ Resolve `transfer_to_number` VARCHAR from explicit fields or first phone-type destination row. UI label \"Phone Number\" maps to destination type `phone`. """ phone = str( tn.get("phone_number") or features.get("transfer_phone_number") or "" ).strip() if phone: return phone[:64] destinations = _parse_destinations_list(tn.get("destinations")) if not destinations: destinations = _parse_destinations_list(features.get("transfer_destinations")) for dest in destinations: if not isinstance(dest, dict): continue dt = _normalize_transfer_destination_type( dest.get("transfer_destination_type") or dest.get("type") ) if dt != "phone": continue p = dest.get("transfer_phone_number") or dest.get("phone_number") or "" ps = str(p).strip() if ps: return ps[:64] if destinations: first = destinations[0] if isinstance(first, dict): dt = _normalize_transfer_destination_type( first.get("transfer_destination_type") or first.get("type") ) if dt == "phone": p = first.get("transfer_phone_number") or first.get("phone_number") or "" ps = str(p).strip() if ps: return ps[:64] return "" def _blankish(v: Any) -> bool: return v is None or v == "" or (isinstance(v, str) and v.strip() == "") def _parse_bot_settings(request_data: dict) -> dict: raw = request_data.get("bot_settings") if raw is None: return {} if isinstance(raw, str): try: parsed = json.loads(raw) return parsed if isinstance(parsed, dict) else {} except Exception: return {} return raw if isinstance(raw, dict) else {} def _parse_conversation_behavior(bs: dict, request_data: dict) -> Optional[dict]: raw = request_data.get("conversation_behavior") if raw is None and bs: raw = bs.get("conversation_behavior") if isinstance(raw, str): try: raw = json.loads(raw) except Exception: return None if isinstance(raw, dict) and raw: return raw return None def _build_platform_settings_column_value( bs: dict, voice_cfg: dict, vd: dict, request_data: dict ) -> str: """ `platform_settings` keeps a compact `voice` object for Voice tab fields only. Full merged voice (incl. voice_id) stays in the `voice` column; language in `default_language`; features/prompts in dedicated columns. """ voice_tab = _voice_tab_settings_for_platform_settings(voice_cfg, bs) payload: Dict[str, Any] = {} if voice_tab: payload["voice"] = voice_tab return json.dumps(payload) def _truthy(v: Any) -> bool: return v in (True, 1, "1", "true", "True") def _apply_features_to_columns(features: dict, table_cols: set, row: dict) -> None: """Map `platform_settings.features` into dedicated `{bid}_bots` columns.""" if not features: return bool_keys = ( ("skip_turn", "skip_turn"), ("detect_language", "detect_language"), ("end_conversation", "end_conversation"), ("voicemail_detection", "voicemail_detection"), ) for feat_key, col in bool_keys: if feat_key not in features or col not in table_cols: continue row[col] = 1 if _truthy(features.get(feat_key)) else 0 if "transfer_enabled" in table_cols and "transfer_to_number" in features: row["transfer_enabled"] = 1 if _truthy(features.get("transfer_to_number")) else 0 long_text = ( ("skip_turn_prompt", "skip_turn_prompt"), ("detect_language_prompt", "detect_language_prompt"), ("end_conversation_prompt", "end_conversation_prompt"), ("transfer_to_number_prompt", "transfer_to_number_prompt"), ("voicemail_detection_prompt", "voicemail_detection_prompt"), ) for fk, col in long_text: if col not in table_cols or fk not in features: continue v = features.get(fk) row[col] = str(v) if v is not None and str(v).strip() != "" else None short_text = ( ("skip_turn_description", "skip_turn_description"), ("detect_language_description", "detect_language_description"), ("end_conversation_description", "end_conversation_description"), ("transfer_number_description", "transfer_number_description"), ("voicemail_description", "voicemail_description"), ) for fk, col in short_text: if col not in table_cols or fk not in features: continue v = features.get(fk) row[col] = str(v) if v is not None and str(v).strip() != "" else None if "end_conversation_timeout" in table_cols and "end_conversation_timeout" in features: try: row["end_conversation_timeout"] = max(0, int(features.get("end_conversation_timeout"))) except (TypeError, ValueError): row["end_conversation_timeout"] = None if "transfer_destination_type" in table_cols and "transfer_destination_type" in features: v = features.get("transfer_destination_type") row["transfer_destination_type"] = str(v)[:32] if v is not None and str(v).strip() != "" else None if "transfer_destinations" in table_cols and "transfer_destinations" in features: td = features.get("transfer_destinations") if td is not None: row["transfer_destinations"] = json.dumps(td) if not isinstance(td, str) else td interrupt_cols = ( ("skip_turn_disable_interruptions", "skip_turn_disable_interruptions"), ("detect_language_disable_interruptions", "detect_language_disable_interruptions"), ("end_conversation_disable_interruptions", "end_conversation_disable_interruptions"), ("transfer_to_number_disable_interruptions", "transfer_to_number_disable_interruptions"), ("voicemail_detection_disable_interruptions", "voicemail_detection_disable_interruptions"), ) for fk, col in interrupt_cols: if col not in table_cols or fk not in features: continue row[col] = 1 if _truthy(features.get(fk)) else 0 def _apply_flat_bot_settings_detail_columns(bs: dict, features: dict, table_cols: set, row: dict) -> None: """ Fill prompt/description/timeout/interrupt columns from top-level `bot_settings` sections when those keys were not supplied via `platform_settings.features`. """ ec = bs.get("end_conversation") or {} if "end_conversation_prompt" not in features and "end_conversation_prompt" in table_cols: v = ec.get("prompt") row["end_conversation_prompt"] = str(v) if v is not None and str(v).strip() != "" else None if "end_conversation_description" not in features and "end_conversation_description" in table_cols: v = ec.get("description") row["end_conversation_description"] = str(v) if v is not None and str(v).strip() != "" else None if "end_conversation_timeout" not in features and "end_conversation_timeout" in table_cols: try: row["end_conversation_timeout"] = max(0, int(ec.get("timeout"))) except (TypeError, ValueError): row["end_conversation_timeout"] = None if ( "end_conversation_disable_interruptions" not in features and "end_conversation_disable_interruptions" in table_cols ): row["end_conversation_disable_interruptions"] = 1 if _truthy(ec.get("disable_interruptions")) else 0 tn = bs.get("transfer_to_number") or {} if "transfer_to_number_prompt" not in features and "transfer_to_number_prompt" in table_cols: v = tn.get("prompt") row["transfer_to_number_prompt"] = str(v) if v is not None and str(v).strip() != "" else None if "transfer_number_description" not in features and "transfer_number_description" in table_cols: v = tn.get("description") row["transfer_number_description"] = str(v) if v is not None and str(v).strip() != "" else None if "transfer_destination_type" not in features and "transfer_destination_type" in table_cols: v = tn.get("destination_type") row["transfer_destination_type"] = str(v)[:32] if v is not None and str(v).strip() != "" else None if "transfer_destinations" not in features and "transfer_destinations" in table_cols: td = tn.get("destinations") if td is not None: row["transfer_destinations"] = json.dumps(td) if not isinstance(td, str) else td if ( "transfer_to_number_disable_interruptions" not in features and "transfer_to_number_disable_interruptions" in table_cols ): row["transfer_to_number_disable_interruptions"] = 1 if _truthy(tn.get("disable_interruptions")) else 0 vm = bs.get("voicemail_detection") or {} if "voicemail_detection_prompt" not in features and "voicemail_detection_prompt" in table_cols: v = vm.get("prompt") row["voicemail_detection_prompt"] = str(v) if v is not None and str(v).strip() != "" else None if "voicemail_description" not in features and "voicemail_description" in table_cols: v = vm.get("description") row["voicemail_description"] = str(v) if v is not None and str(v).strip() != "" else None if ( "voicemail_detection_disable_interruptions" not in features and "voicemail_detection_disable_interruptions" in table_cols ): row["voicemail_detection_disable_interruptions"] = 1 if _truthy(vm.get("disable_interruptions")) else 0 st = bs.get("skip_turn") or {} if "skip_turn_prompt" not in features and "skip_turn_prompt" in table_cols: v = st.get("prompt") row["skip_turn_prompt"] = str(v) if v is not None and str(v).strip() != "" else None if "skip_turn_description" not in features and "skip_turn_description" in table_cols: v = st.get("description") row["skip_turn_description"] = str(v) if v is not None and str(v).strip() != "" else None if "skip_turn_disable_interruptions" not in features and "skip_turn_disable_interruptions" in table_cols: row["skip_turn_disable_interruptions"] = 1 if _truthy(st.get("disable_interruptions")) else 0 dl = bs.get("detect_language") or {} if "detect_language_prompt" not in features and "detect_language_prompt" in table_cols: v = dl.get("prompt") row["detect_language_prompt"] = str(v) if v is not None and str(v).strip() != "" else None if "detect_language_description" not in features and "detect_language_description" in table_cols: v = dl.get("description") row["detect_language_description"] = str(v) if v is not None and str(v).strip() != "" else None if ( "detect_language_disable_interruptions" not in features and "detect_language_disable_interruptions" in table_cols ): row["detect_language_disable_interruptions"] = 1 if _truthy(dl.get("disable_interruptions")) else 0 def _apply_bot_settings_to_cluster_row( bs: dict, table_cols: set, row: dict, voice_cfg: dict ) -> None: """Dedicated columns for features + toggles; platform_settings JSON stays minimal.""" features: dict = {} if bs: ps = bs.get("platform_settings") if isinstance(ps, dict): rawf = ps.get("features") if isinstance(rawf, dict): features = rawf _apply_features_to_columns(features, table_cols, row) _apply_flat_bot_settings_detail_columns(bs, features, table_cols, row) if not bs: if "detect_language" in table_cols and "detect_language" not in row: row["detect_language"] = 1 return ec = bs.get("end_conversation") or {} if "end_conversation" in table_cols and "end_conversation" not in features: row["end_conversation"] = 1 if ec.get("enabled") else 0 st = bs.get("skip_turn") or {} if "skip_turn" in table_cols and "skip_turn" not in features: row["skip_turn"] = 1 if st.get("enabled") else 0 vm = bs.get("voicemail_detection") or {} if "voicemail_detection" in table_cols and "voicemail_detection" not in features: row["voicemail_detection"] = 1 if vm.get("enabled") else 0 dl = bs.get("detect_language") or {} if "detect_language" in table_cols and "detect_language" not in features: row["detect_language"] = 1 if dl.get("enabled") else 0 tn = bs.get("transfer_to_number") or {} if "transfer_enabled" in table_cols and "transfer_to_number" not in features: row["transfer_enabled"] = 1 if tn.get("enabled") else 0 if "transfer_to_number" in table_cols: phone = _transfer_phone_for_column(tn, features) row["transfer_to_number"] = phone if phone else None cc = bs.get("concurrent_calls") or {} lim = cc.get("limit") if "concurrent_calls" in table_cols: if lim is None or lim == "" or lim == -1: row["concurrent_calls"] = None else: try: row["concurrent_calls"] = max(0, min(300, int(lim))) except (TypeError, ValueError): row["concurrent_calls"] = None adv = bs.get("advanced_settings") or {} keywords = adv.get("keywords") if "filler_keywords" in table_cols: if keywords is not None and str(keywords).strip() != "": row["filler_keywords"] = str(keywords).strip() else: row["filler_keywords"] = None if "advanced_settings" in table_cols: rest = {} if adv.get("turn_timeout") is not None: try: rest["turn_timeout"] = int(adv.get("turn_timeout")) except (TypeError, ValueError): rest["turn_timeout"] = adv.get("turn_timeout") if rest: row["advanced_settings"] = json.dumps(rest) else: row["advanced_settings"] = None vo = dict(bs.get("voice") or {}) if voice_cfg.get("voice_id"): vo.setdefault("voice_id", voice_cfg.get("voice_id")) for k, v in voice_cfg.items(): vo.setdefault(k, v) if "voice" in table_cols and vo: row["voice"] = json.dumps(vo) def _sanitize_cluster_bots_row_for_insert(row: dict) -> dict: """ Legacy `{bid}_bots` tables often use INT for mcube_gid; empty strings from JSON must not be sent. """ out = dict(row) if "mcube_gid" in out: v = out["mcube_gid"] if _blankish(v): out["mcube_gid"] = 1 else: s = str(v).strip() try: out["mcube_gid"] = int(s) except ValueError: out["mcube_gid"] = s for col in ("concurrent_calls",): if col not in out: continue v = out[col] if _blankish(v): out[col] = None else: try: out[col] = int(v) except (TypeError, ValueError): out[col] = None # `agent_id` may be a LiveKit Cloud agent id like "sb_..."; keep as string. if "agent_id" in out: v = out.get("agent_id") if _blankish(v): out["agent_id"] = None else: out["agent_id"] = str(v).strip()[:128] return out def _build_cluster_bot_row( request_data: dict, bid: int, table_cols: set, validated: Optional[Dict[str, Any]] = None ) -> dict: vd = validated or {} voice_cfg = _voice_config_from_request(request_data, vd) bs = _parse_bot_settings(request_data) row = { "business_id": bid, "bot_name": (vd.get("name") or request_data.get("name") or "").strip() or "Unnamed Bot", # Optional: LiveKit Cloud agent id (e.g. "sb_...") assigned to this bot. "agent_id": request_data.get("agent_id") or request_data.get("agentId"), # Explicit mapping for dashboards/filters: store the bot name as the agent_name we use in-app. "agent_name": (vd.get("name") or request_data.get("name") or "").strip() or "Unnamed Bot", "llm_model": vd.get("llm_model") if vd.get("llm_model") is not None else (request_data.get("llm_model") or ""), "prompt": vd.get("system_prompt") if vd.get("system_prompt") is not None else (request_data.get("system_prompt") or ""), "platform_settings": _build_platform_settings_column_value(bs, voice_cfg, vd, request_data), "voice_name": "", "is_active": 1, } cb = _parse_conversation_behavior(bs, request_data) if cb is not None and "conversation_behavior" in table_cols: row["conversation_behavior"] = json.dumps(cb) optional_map = { "stt_provider": vd.get("stt_provider") if vd.get("stt_provider") is not None else request_data.get("stt_provider"), "tts_provider": vd.get("tts_provider") if vd.get("tts_provider") is not None else request_data.get("tts_provider"), "llm_provider": vd.get("llm_provider") if vd.get("llm_provider") is not None else request_data.get("llm_provider"), "mcube_exenumber": vd.get("mcube_exenumber") if vd.get("mcube_exenumber") is not None else request_data.get("mcube_exenumber"), "mcube_gid": vd.get("mcube_gid") if vd.get("mcube_gid") is not None else request_data.get("mcube_gid"), "message_inbound": vd.get("first_message_inbound") if vd.get("first_message_inbound") is not None else request_data.get("first_message_inbound"), "message_outbound": vd.get("first_message_outbound") if vd.get("first_message_outbound") is not None else request_data.get("first_message_outbound"), "guardrails": json.dumps(request_data.get("guardrails") or vd.get("guardrails") or []), "system_tools": json.dumps(request_data.get("system_tools") or vd.get("system_tools") or []), "platform_settings": row["platform_settings"], } for k, v in optional_map.items(): if k in table_cols and v is not None: row[k] = v if "default_language" in table_cols: dlang = ( vd.get("default_language") if vd.get("default_language") is not None else request_data.get("default_language") ) if dlang is not None and str(dlang).strip() != "": row["default_language"] = str(dlang).strip()[:32] else: row["default_language"] = None _apply_bot_settings_to_cluster_row(bs, table_cols, row, voice_cfg) filtered = {k: v for k, v in row.items() if k in table_cols} return _sanitize_cluster_bots_row_for_insert(filtered) def _cluster_update_values(request_data: dict, table_cols: set) -> dict: """Map JSON/API fields to cluster columns for PATCH.""" mapping = { "name": "bot_name", "agent_id": "agent_id", "agent_name": "agent_name", "system_prompt": "prompt", "llm_model": "llm_model", "stt_provider": "stt_provider", "tts_provider": "tts_provider", "llm_provider": "llm_provider", "mcube_exenumber": "mcube_exenumber", "mcube_gid": "mcube_gid", "first_message_inbound": "message_inbound", "first_message_outbound": "message_outbound", "is_active": "is_active", } updates: dict = {} for api_key, col in mapping.items(): if api_key in request_data and col in table_cols: val = request_data[api_key] if api_key == "is_active": updates[col] = 1 if val in (True, 1, "1", "true") else 0 elif api_key == "agent_id": updates[col] = str(val).strip()[:128] if not _blankish(val) else None elif api_key == "agent_name": updates[col] = str(val).strip()[:255] if not _blankish(val) else None elif col == "mcube_gid": if _blankish(val): updates[col] = 1 else: s = str(val).strip() try: updates[col] = int(s) except ValueError: updates[col] = s else: updates[col] = val voice_cfg = _voice_config_from_request(request_data) if "bot_settings" in request_data: bs = _parse_bot_settings(request_data) tmp: dict = {} _apply_bot_settings_to_cluster_row(bs, table_cols, tmp, voice_cfg) for k, v in tmp.items(): if k in table_cols: updates[k] = v cb = _parse_conversation_behavior(bs, request_data) if cb is not None and "conversation_behavior" in table_cols: updates["conversation_behavior"] = json.dumps(cb) vd_patch = {"default_language": request_data.get("default_language")} if "platform_settings" in table_cols: updates["platform_settings"] = _build_platform_settings_column_value( bs, voice_cfg, vd_patch, request_data ) if "default_language" in table_cols and "default_language" in request_data: dl = request_data.get("default_language") updates["default_language"] = str(dl).strip()[:32] if dl not in (None, "") else None elif "voice_config" in request_data or "default_language" in request_data: if "platform_settings" in table_cols: updates["platform_settings"] = _build_platform_settings_column_value( {}, voice_cfg, {"default_language": request_data.get("default_language")}, request_data ) if "default_language" in table_cols and "default_language" in request_data: dl = request_data.get("default_language") updates["default_language"] = str(dl).strip()[:32] if dl not in (None, "") else None return updates def _shape_cluster_kb_document(row: dict) -> dict: """Match `DocumentViewSet` cluster rows so `doc_name` appears as `name` for the UI.""" out = dict(row) out["name"] = out.get("doc_name") or "" out["file_path"] = out.get("doc_url") or "" du = out.get("doc_url") if du and str(du).startswith(("http://", "https://")): out["web_url"] = du return out def _knowledge_documents_by_bot_id(business_id: int, bot_ids: List[int]) -> Dict[int, List[dict]]: """Load `{business_id}_knowledgebase` rows grouped by `bot_id` (newest first per bot).""" if not bot_ids: return {} bid = int(business_id) ensure_business_cluster_tables(bid) table = f"{bid}_knowledgebase" uniq: List[int] = [] seen = set() for x in bot_ids: try: ix = int(x) except (TypeError, ValueError): continue if ix not in seen: seen.add(ix) uniq.append(ix) if not uniq: return {} placeholders = ", ".join(["%s"] * len(uniq)) sql = ( f"SELECT * FROM {_q_ident(table)} WHERE bot_id IN ({placeholders}) " "ORDER BY bot_id ASC, id DESC" ) with connections["cluster"].cursor() as cur: cur.execute(sql, uniq) cols = [c[0] for c in cur.description] raw_rows = cur.fetchall() by_bot: Dict[int, List[dict]] = defaultdict(list) for r in raw_rows: d = _shape_cluster_kb_document(dict(zip(cols, r))) b_id = d.get("bot_id") if b_id is None: continue try: b_id = int(b_id) except (TypeError, ValueError): continue by_bot[b_id].append(d) return dict(by_bot) def _attach_knowledge_documents(business_id: int, bots: List[dict]) -> None: """Mutate bot dicts in place: set `knowledge_documents` from cluster `{bid}_knowledgebase`.""" ids: List[int] = [] for b in bots: raw = b.get("bot_id") if raw is None: raw = b.get("id") if raw is not None: try: ids.append(int(raw)) except (TypeError, ValueError): pass grouped = _knowledge_documents_by_bot_id(business_id, ids) for b in bots: raw = b.get("bot_id") if raw is None: raw = b.get("id") try: ik = int(raw) if raw is not None else None except (TypeError, ValueError): ik = None b["knowledge_documents"] = grouped.get(ik, []) if ik is not None else [] class BotViewSet(viewsets.ModelViewSet): """ Bot CRUD. With `business_id`, stores rows only in cluster DB `{business_id}_bots`. Without `business_id`, uses legacy master `AgentConfig` (agents_agent_config). """ queryset = AgentConfig.objects.all().order_by("name") def get_serializer_class(self): if self.action == "create": return AgentConfigCreateSerializer return AgentConfigSerializer def list(self, request, *args, **kwargs): """ If business_id is provided, read bots from cluster `{business_id}_bots`. Otherwise fall back to master DB AgentConfig list. """ business_id = request.query_params.get("business_id") if business_id: try: bid = int(business_id) ensure_business_cluster_tables(bid) table = f"{bid}_bots" with connections["cluster"].cursor() as cur: cur.execute(f"SELECT * FROM {_q_ident(table)} ORDER BY bot_id DESC") cols = [c[0] for c in cur.description] rows = [_normalize_cluster_bot(dict(zip(cols, r))) for r in cur.fetchall()] _attach_knowledge_documents(bid, rows) return Response(rows) except Exception as e: return Response( {"message": "Failed to load bots from cluster.", "error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) return super().list(request, *args, **kwargs) def retrieve(self, request, *args, **kwargs): bid_raw = request.query_params.get("business_id") pk = kwargs.get("pk") if bid_raw and pk is not None: try: bid = int(bid_raw) 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", [pk]) row = cur.fetchone() if not row: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) cols = [c[0] for c in cur.description] payload = _normalize_cluster_bot(dict(zip(cols, row))) _attach_knowledge_documents(bid, [payload]) return Response(payload) except Exception as e: return Response( {"message": "Failed to load bot from cluster.", "error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) return super().retrieve(request, *args, **kwargs) def partial_update(self, request, *args, **kwargs): return self._update_cluster_or_master(request, partial=True, *args, **kwargs) def update(self, request, *args, **kwargs): return self._update_cluster_or_master(request, partial=False, *args, **kwargs) def _update_cluster_or_master(self, request, partial: bool, *args, **kwargs): bid_raw = request.data.get("business_id") or request.query_params.get("business_id") pk = kwargs.get("pk") if bid_raw and pk is not None: try: bid = int(bid_raw) ensure_business_cluster_tables(bid) table = f"{bid}_bots" with connections["cluster"].cursor() as cur: cur.execute(f"SELECT * FROM {_q_ident(table)} LIMIT 0") table_cols = {c[0] for c in cur.description} updates = _cluster_update_values(request.data, table_cols) if not updates: cur.execute(f"SELECT * FROM {_q_ident(table)} WHERE bot_id = %s", [pk]) row = cur.fetchone() if not row: return Response({"detail": "Not found."}, status=status.HTTP_404_NOT_FOUND) cols = [c[0] for c in cur.description] payload = _normalize_cluster_bot(dict(zip(cols, row))) _attach_knowledge_documents(bid, [payload]) return Response(payload) set_parts = ", ".join(f"{_q_ident(k)} = %s" for k in updates) vals = list(updates.values()) vals.append(pk) cur.execute( f"UPDATE {_q_ident(table)} SET {set_parts} WHERE bot_id = %s", vals, ) cur.execute(f"SELECT * FROM {_q_ident(table)} WHERE bot_id = %s", [pk]) row = cur.fetchone() cols = [c[0] for c in cur.description] payload = _normalize_cluster_bot(dict(zip(cols, row))) _attach_knowledge_documents(bid, [payload]) return Response(payload) except Exception as e: return Response( {"message": "Failed to update bot in cluster.", "error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) if partial: return super().partial_update(request, *args, **kwargs) return super().update(request, *args, **kwargs) def destroy(self, request, *args, **kwargs): bid_raw = request.query_params.get("business_id") or request.data.get("business_id") pk = kwargs.get("pk") if bid_raw and pk is not None: try: bid = int(bid_raw) ensure_business_cluster_tables(bid) table = f"{bid}_bots" with connections["cluster"].cursor() as cur: cur.execute(f"DELETE FROM {_q_ident(table)} WHERE bot_id = %s", [pk]) return Response(status=status.HTTP_204_NO_CONTENT) except Exception as e: return Response( {"message": "Failed to delete bot from cluster.", "error": str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) return super().destroy(request, *args, **kwargs) def create(self, request, *args, **kwargs): """ With business_id: insert only into cluster `{business_id}_bots` (livekitvoicebot_cluster). Without business_id: legacy create on master AgentConfig. """ serializer = self.get_serializer(data=request.data) serializer.is_valid(raise_exception=True) business_id_raw = request.data.get("business_id") if business_id_raw not in (None, ""): try: bid = int(business_id_raw) except (TypeError, ValueError): return Response( {"business_id": ["Invalid business_id."]}, status=status.HTTP_400_BAD_REQUEST, ) ensure_business_cluster_tables(bid) table = f"{bid}_bots" with connections["cluster"].cursor() as cur: cur.execute(f"SELECT * FROM {_q_ident(table)} LIMIT 0") table_cols = {c[0] for c in cur.description} # Optional: auto-deploy a LiveKit Cloud Agent BEFORE inserting the bot row. # This guarantees: create agent in LiveKit -> get CA_... -> store same id in DB. auto_deploy = str(os.getenv("LIVEKIT_AUTO_DEPLOY_ON_BOT_CREATE", "false")).lower() in ( "1", "true", "yes", "on", ) deploy_info: dict | None = None livekit_agent_id: str | None = None bot_name = ( str(serializer.validated_data.get("name") or request.data.get("name") or "").strip() or "Unnamed Bot" ) if auto_deploy: try: workdir = prepare_workdir_for_bot(bot_name) secrets_file = os.getenv("LIVEKIT_AGENT_SECRETS_FILE") region = os.getenv("LIVEKIT_AGENT_REGION") or "us-east" deploy_info = run_lk_agent( "create", workdir, region=region, secrets_file=secrets_file if secrets_file and str(secrets_file).strip() else None, timeout_s=int(os.getenv("LIVEKIT_AGENT_CREATE_TIMEOUT_S", "1800")), silent=True, ) livekit_agent_id = deploy_info.get("agent_id") if not livekit_agent_id: stderr = str(deploy_info.get("stderr") or "") if "maximum number of agents reached" in stderr.lower(): existing = list_cloud_agent_deployments() livekit_agent_id = (existing[0] if existing else None) or os.getenv( "LIVEKIT_FALLBACK_AGENT_ID" ) except Exception as e: deploy_info = {"ok": False, "message": str(e)} # Inject the LiveKit mapping into the insert row (if available). request_for_row = dict(request.data) if livekit_agent_id: request_for_row["agent_id"] = livekit_agent_id request_for_row["agent_name"] = bot_name row = _build_cluster_bot_row(request_for_row, bid, table_cols, serializer.validated_data) if not row: return Response( {"detail": "Cluster bots table has no writable columns."}, status=status.HTTP_500_INTERNAL_SERVER_ERROR, ) cols = list(row.keys()) placeholders = ", ".join(["%s"] * len(cols)) col_sql = ", ".join(_q_ident(c) for c in cols) sql = f"INSERT INTO {_q_ident(table)} ({col_sql}) VALUES ({placeholders})" with connections["cluster"].cursor() as cur: cur.execute(sql, [row[c] for c in cols]) bot_pk = cur.lastrowid cur.execute(f"SELECT * FROM {_q_ident(table)} WHERE bot_id = %s", [bot_pk]) fetched = cur.fetchone() out_cols = [c[0] for c in cur.description] payload = _normalize_cluster_bot(dict(zip(out_cols, fetched))) _attach_knowledge_documents(bid, [payload]) # Attach deploy debug info (non-breaking). if deploy_info is not None: payload["livekit_deploy"] = { "ok": bool(deploy_info.get("ok")), "exit_code": deploy_info.get("exit_code"), "command": deploy_info.get("command"), "agent_id": deploy_info.get("agent_id"), "message": deploy_info.get("message"), "stdout": deploy_info.get("stdout"), "stderr": deploy_info.get("stderr"), } headers = self.get_success_headers(payload) return Response(payload, status=status.HTTP_201_CREATED, headers=headers) # Legacy: master DB only (no cluster row). bot = serializer.save() payload = AgentConfigSerializer(bot, context={"request": request}).data headers = self.get_success_headers(payload) return Response(payload, status=status.HTTP_201_CREATED, headers=headers) @action(detail=True, methods=["post"], url_path="initiate-call") def initiate_call(self, request, pk=None): """ Click-to-call entrypoint from the UI. Expected body: { "customer_number": "...", "bot_id": "" } Also accepts aliases used by other integrations: `to`, `custnumber`. """ data = request.data or {} customer_number = ( str(data.get("customer_number") or data.get("custnumber") or data.get("to") or "").strip() ) if not customer_number: return Response( {"success": False, "message": "customer_number is required"}, status=status.HTTP_400_BAD_REQUEST, ) # Optional: route cluster-bot config resolution in the MCube webhook runtime. business_id = data.get("business_id") or request.query_params.get("business_id") try: business_id_int = int(business_id) if business_id not in (None, "") else None except Exception: business_id_int = None # Best-effort: enrich from master AgentConfig when pk exists there. bot: AgentConfig | None = None try: bot = AgentConfig.objects.filter(id=pk).first() if pk is not None else None except Exception: bot = None call_id = str(data.get("call_id") or data.get("refid") or uuid4()) public_base = str(os.getenv("MCUBE_PUBLIC_BASE_URL", "") or "").rstrip("/") public_ws_base = str(os.getenv("MCUBE_PUBLIC_WS_URL_BASE", "") or "").rstrip("/") webhook_path = str(os.getenv("MCUBE_WEBHOOK_PATH", "/webhooks/mcube") or "/webhooks/mcube") ws_path_prefix = str(os.getenv("MCUBE_WS_PATH_PREFIX", "/ws") or "/ws").strip() callback_url = str(data.get("callback_url") or "").strip() if not callback_url and public_base: callback_url = f"{public_base}{webhook_path}" refurl = str(data.get("refurl") or "").strip() if not refurl and public_ws_base: # Matches the working example: wss:///ws/ refurl = f"{public_ws_base}{ws_path_prefix.rstrip('/')}/{call_id}" agent_name = str(data.get("agent_name") or "").strip() if not agent_name: # Prefer DB bot name; fall back to "default" (matches your cURL example). agent_name = str(getattr(bot, "name", "") or "").strip() or "default" payload: dict[str, Any] = { "to": customer_number, "agent_name": agent_name, "call_id": call_id, } if callback_url: payload["callback_url"] = callback_url if refurl: payload["refurl"] = refurl # Help the webhook runtime resolve per-bot/per-business settings dynamically. if pk is not None: try: payload["bot_id"] = int(pk) except Exception: payload["bot_id"] = pk if business_id_int is not None: payload["business_id"] = business_id_int # Include telephony defaults (if present) so outbound-call works even without runtime DB fetch. if bot: if getattr(bot, "mcube_exenumber", ""): payload["exenumber"] = str(bot.mcube_exenumber).strip() if getattr(bot, "mcube_gid", ""): payload["gid"] = str(bot.mcube_gid).strip() or "1" if getattr(bot, "mcube_http_authorization", ""): payload["HTTP_AUTHORIZATION"] = str(bot.mcube_http_authorization).strip() mcube_url = os.getenv( "MCUBE_OUTBOUND_CALL_URL", "https://app3.syntheon.in/api/mcube/outbound-call", ) timeout_s = float(os.getenv("MCUBE_OUTBOUND_CALL_TIMEOUT_S", "20")) try: with httpx.Client(timeout=timeout_s) as client: resp = client.post(mcube_url, json=payload) resp_json = resp.json() if resp.content else {} except Exception as e: return Response( {"success": False, "message": "Failed to call outbound-call service", "error": str(e)}, status=status.HTTP_502_BAD_GATEWAY, ) if resp.status_code not in (200, 201): return Response( { "success": False, "message": "Outbound-call service returned an error", "http_status": resp.status_code, "response": resp_json, }, status=status.HTTP_502_BAD_GATEWAY, ) return Response( {"success": True, "message": "Call initiated", "call_id": call_id, "response": resp_json}, status=status.HTTP_200_OK, )