Spaces:
Runtime error
Runtime error
| """ | |
| PostgreSQL client for Pipecat Server. | |
| Provides direct database access for: | |
| - Loading agent configuration at call start | |
| - Writing call logs after call completion | |
| """ | |
| import os | |
| import json | |
| from typing import Optional | |
| import asyncpg | |
| from utils.logging import log_info, log_error, log_warning | |
| class PostgresClient: | |
| """Async PostgreSQL client using asyncpg connection pool.""" | |
| _pool: Optional[asyncpg.Pool] = None | |
| async def get_pool(cls) -> asyncpg.Pool: | |
| """Get or create connection pool.""" | |
| if cls._pool is None: | |
| database_url = os.getenv("DATABASE_URL") | |
| if not database_url: | |
| raise ValueError("DATABASE_URL environment variable not set") | |
| cls._pool = await asyncpg.create_pool( | |
| database_url, | |
| min_size=2, | |
| max_size=10, | |
| command_timeout=30, | |
| ) | |
| log_info("postgres", "Connection pool created") | |
| return cls._pool | |
| async def close(cls) -> None: | |
| """Close connection pool.""" | |
| if cls._pool: | |
| await cls._pool.close() | |
| cls._pool = None | |
| log_info("postgres", "Connection pool closed") | |
| async def get_agent(cls, agent_id: str) -> Optional[dict]: | |
| """ | |
| Fetch ServiceConfiguration with services and botData by ID. | |
| Returns dict with: | |
| - id, companyName, voiceId, model, maxDuration, language, botData | |
| - services: list of {id, name, description, keywords, order} | |
| """ | |
| pool = await cls.get_pool() | |
| async with pool.acquire() as conn: | |
| # Get configuration including botData | |
| config_row = await conn.fetchrow( | |
| ''' | |
| SELECT id, "companyName", "voiceId", model, "maxDuration", language, "botData" | |
| FROM "ServiceConfiguration" | |
| WHERE id = $1 | |
| ''', | |
| agent_id, | |
| ) | |
| if not config_row: | |
| log_error("postgres", f"Agent not found: {agent_id}") | |
| return None | |
| # Get services | |
| service_rows = await conn.fetch( | |
| ''' | |
| SELECT id, name, description, keywords, "order" | |
| FROM "Service" | |
| WHERE "configurationId" = $1 | |
| ORDER BY "order" ASC | |
| ''', | |
| agent_id, | |
| ) | |
| config = dict(config_row) | |
| config["services"] = [dict(row) for row in service_rows] | |
| # Parse botData if it's a string | |
| if config.get("botData") and isinstance(config["botData"], str): | |
| config["botData"] = json.loads(config["botData"]) | |
| log_info("postgres", f"Loaded agent: {config['companyName']} ({len(config['services'])} services, botData={'yes' if config.get('botData') else 'no'})") | |
| return config | |
| async def get_bot_data(cls, agent_id: str) -> Optional[dict]: | |
| """ | |
| Load botData JSON from ServiceConfiguration. | |
| Returns the botData dict if found, None otherwise. | |
| Falls back to legacy generation if botData is not set. | |
| """ | |
| agent_config = await cls.get_agent(agent_id) | |
| if not agent_config: | |
| return None | |
| bot_data = agent_config.get("botData") | |
| if bot_data: | |
| log_info("postgres", f"Loaded botData for agent: {agent_id}") | |
| return bot_data | |
| # Fallback: generate from agent config (legacy mode) | |
| log_warning("postgres", f"No botData in database for {agent_id}, generating legacy config") | |
| return cls._generate_bot_data_legacy(agent_config) | |
| def _generate_bot_data_legacy(agent_config: dict) -> dict: | |
| """ | |
| Legacy BotData generation from agent config. | |
| DEPRECATED: Ensure botData is generated by Next.js API. | |
| """ | |
| company_name = agent_config.get("companyName", "Компанія") | |
| services = agent_config.get("services", []) | |
| services_list = "\n".join( | |
| f"{i + 1}. {s['name']} - {s['description']}" | |
| for i, s in enumerate(sorted(services, key=lambda x: x.get("order", 0))) | |
| ) | |
| service_ids = [s["id"] for s in services] | |
| return { | |
| "flow": { | |
| "initial_node": "greeting", | |
| "nodes": { | |
| "greeting": { | |
| "role_messages": [{"role": "system", "content": f'Ти - голосовий асистент компанії "{company_name}". Говори українською мовою.'}], | |
| "task_messages": [{"role": "system", "content": f'Привітай клієнта від імені компанії "{company_name}" і запропонуй обрати послугу:\n{services_list}'}], | |
| "functions": [ | |
| {"type": "function", "function": {"name": "select_service", "description": "Клієнт обрав послугу", "parameters": {"type": "object", "properties": {"service_id": {"type": "string", "enum": service_ids}}, "required": ["service_id"]}, "transition_to": "service_details"}}, | |
| {"type": "function", "function": {"name": "end_call", "description": "Завершити розмову", "parameters": {"type": "object", "properties": {}}, "transition_to": "goodbye"}}, | |
| ], | |
| }, | |
| "service_details": { | |
| "role_messages": [{"role": "system", "content": f'Ти - голосовий асистент компанії "{company_name}". Надай інформацію про послугу.'}], | |
| "task_messages": [{"role": "system", "content": "Надай інформацію про обрану послугу."}], | |
| "functions": [ | |
| {"type": "function", "function": {"name": "back_to_menu", "description": "Повернутися до меню", "parameters": {"type": "object", "properties": {}}, "transition_to": "greeting"}}, | |
| {"type": "function", "function": {"name": "end_call", "description": "Завершити розмову", "parameters": {"type": "object", "properties": {}}, "transition_to": "goodbye"}}, | |
| ], | |
| }, | |
| "goodbye": { | |
| "role_messages": [{"role": "system", "content": f'Попрощайся з клієнтом від імені "{company_name}".'}], | |
| "task_messages": [{"role": "system", "content": "Подякуй і побажай гарного дня."}], | |
| }, | |
| }, | |
| }, | |
| "config": { | |
| "voice_id": agent_config.get("voiceId") or "79a125e8-cd45-4c13-8a67-188112f4dd22", | |
| "model": agent_config.get("model") or "gpt-4.1-mini", | |
| }, | |
| "metadata": { | |
| "max_duration": agent_config.get("maxDuration") or 3600, | |
| "environment": "production", | |
| }, | |
| } | |
| async def create_call_log( | |
| cls, | |
| configuration_id: str, | |
| caller_number: str, | |
| call_sid: Optional[str] = None, | |
| stream_sid: Optional[str] = None, | |
| pipecat_session_id: Optional[str] = None, | |
| duration: int = 0, | |
| status: str = "COMPLETED", | |
| transcript: Optional[dict] = None, | |
| ) -> str: | |
| """ | |
| Create a new call log entry. | |
| Args: | |
| pipecat_session_id: Pipecat Cloud session ID for fetching logs via API | |
| Returns the created log ID. | |
| """ | |
| pool = await cls.get_pool() | |
| # Generate cuid-like ID | |
| import time | |
| import random | |
| import string | |
| timestamp = hex(int(time.time() * 1000))[2:] | |
| random_part = "".join(random.choices(string.ascii_lowercase + string.digits, k=8)) | |
| log_id = f"cl{timestamp}{random_part}" | |
| async with pool.acquire() as conn: | |
| await conn.execute( | |
| ''' | |
| INSERT INTO "CallLog" ( | |
| id, "configurationId", "callerNumber", "callSid", "streamSid", | |
| "pipecatSessionId", duration, status, transcript, "startedAt", "endedAt", "createdAt" | |
| ) | |
| VALUES ($1, $2, $3, $4, $5, $6, $7, $8::\"CallStatus\", $9, NOW(), NOW(), NOW()) | |
| ''', | |
| log_id, | |
| configuration_id, | |
| caller_number, | |
| call_sid, | |
| stream_sid, | |
| pipecat_session_id, | |
| duration, | |
| status, | |
| json.dumps(transcript) if transcript else None, | |
| ) | |
| log_info("postgres", f"Call log created: {log_id} [status={status}, session={pipecat_session_id}]") | |
| return log_id | |
| async def update_call_log( | |
| cls, | |
| log_id: str, | |
| duration: Optional[int] = None, | |
| status: Optional[str] = None, | |
| transcript: Optional[dict] = None, | |
| ) -> None: | |
| """Update an existing call log.""" | |
| pool = await cls.get_pool() | |
| updates = [] | |
| params = [log_id] | |
| param_idx = 2 | |
| if duration is not None: | |
| updates.append(f"duration = ${param_idx}") | |
| params.append(duration) | |
| param_idx += 1 | |
| if status is not None: | |
| updates.append(f'status = ${param_idx}::"CallStatus"') | |
| params.append(status) | |
| param_idx += 1 | |
| if transcript is not None: | |
| updates.append(f"transcript = ${param_idx}") | |
| params.append(json.dumps(transcript)) | |
| param_idx += 1 | |
| updates.append('"endedAt" = NOW()') | |
| if not updates: | |
| return | |
| query = f'UPDATE "CallLog" SET {", ".join(updates)} WHERE id = $1' | |
| async with pool.acquire() as conn: | |
| await conn.execute(query, *params) | |
| log_info("postgres", f"Call log updated: {log_id}") | |
| # Module-level instance for convenience | |
| postgres = PostgresClient() | |