| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| import os |
| import json |
| import logging |
| import asyncpg |
| import time |
| from datetime import datetime, timedelta, timezone |
| from typing import Optional, Dict, Any, List |
|
|
| logger = logging.getLogger("neon_client") |
|
|
| |
| _pool: Optional[asyncpg.Pool] = None |
|
|
| |
| _config_cache: Dict[tuple, tuple] = {} |
|
|
| async def init_db_pool(): |
| """Initializes the asyncpg connection pool.""" |
| global _pool |
| if _pool is not None: |
| return |
| database_url = os.getenv("DATABASE_URL") |
| if not database_url: |
| logger.error("DATABASE_URL environment variable is missing. Database operations will fail.") |
| return |
| try: |
| |
| _pool = await asyncpg.create_pool(dsn=database_url, min_size=1, max_size=5) |
| logger.info("Neon database connection pool established.") |
| except Exception as e: |
| logger.exception(f"Failed to create Neon connection pool: {e}") |
|
|
| async def close_db_pool(): |
| """Closes the connection pool.""" |
| global _pool |
| if _pool is not None: |
| await _pool.close() |
| _pool = None |
| logger.info("Neon database connection pool closed.") |
|
|
| async def fetch_tenant_config(company_id: str, line_token: Optional[str] = None) -> Optional[Dict[str, Any]]: |
| """ |
| Fetches the tenant configurations for a company, verified by company_id and optional line_token. |
| Normalizes the caller details list structure on the fly. |
| """ |
| global _pool, _config_cache |
| |
| |
| cache_key = (company_id, line_token) |
| if cache_key in _config_cache: |
| cached_time, cached_val = _config_cache[cache_key] |
| if time.time() - cached_time < 60.0: |
| logger.info(f"Returning cached tenant config for company_id={company_id}") |
| return cached_val |
|
|
| if _pool is None: |
| await init_db_pool() |
| if _pool is None: |
| return None |
|
|
| query = """ |
| SELECT |
| c.id AS company_id, |
| c.name AS company_name, |
| c.name_ur AS company_name_ur, |
| c.overview, |
| c.overview_ur, |
| c.is_active AS company_active, |
| ac.agent_name, |
| ac.agent_name_ur, |
| ac.agent_role, |
| ac.agent_role_ur, |
| ac.gender AS agent_gender, |
| ac.system_instruction_rules, |
| ac.caller_details_to_collect, |
| ac.rag_enabled, |
| ac.rag_similarity_threshold, |
| ac.id AS agent_config_id, |
| pl.id AS phone_line_id, |
| pl.line_token, |
| pl.is_enabled AS line_active |
| FROM companies c |
| LEFT JOIN phone_lines pl ON pl.company_id = c.id AND pl.is_enabled = true |
| LEFT JOIN agent_configs ac ON ac.id = pl.agent_config_id |
| WHERE c.id = $1 AND c.is_active = true |
| AND ($2::text IS NULL OR pl.line_token = $2) |
| LIMIT 1 |
| """ |
|
|
| try: |
| async with _pool.acquire() as conn: |
| row = await conn.fetchrow(query, company_id, line_token) |
| if not row: |
| return None |
|
|
| config = dict(row) |
|
|
| |
| raw_details = config.get("caller_details_to_collect") |
| normalized_details = [] |
| if raw_details: |
| if isinstance(raw_details, str): |
| try: |
| raw_details = json.loads(raw_details) |
| except Exception: |
| pass |
|
|
| if isinstance(raw_details, list): |
| for item in raw_details: |
| if isinstance(item, str): |
| label = item.replace("_", " ").title() |
| normalized_details.append({ |
| "key": item, |
| "label": label, |
| "label_ur": label, |
| "type": "text" |
| }) |
| elif isinstance(item, dict) and "key" in item: |
| normalized_details.append({ |
| "key": item["key"], |
| "label": item.get("label", item["key"].replace("_", " ").title()), |
| "label_ur": item.get("label_ur", item.get("label", item["key"].replace("_", " ").title())), |
| "type": item.get("type", "text") |
| }) |
|
|
| config["caller_details_to_collect"] = normalized_details |
| _config_cache[cache_key] = (time.time(), config) |
| return config |
| except Exception as e: |
| logger.error(f"Error querying tenant config: {e}") |
| return None |
|
|
| async def insert_call_log( |
| call_id: str, |
| company_id: str, |
| phone_line_id: Optional[int], |
| full_transcript: str, |
| extracted_caller_data: Dict[str, Any], |
| duration_seconds: int, |
| status: str = "completed" |
| ) -> bool: |
| """ |
| Inserts or updates a call record inside Neon database. |
| |
| FIX F: started_at is now computed in Python as a plain TIMESTAMPTZ |
| parameter ($8), rather than derived in SQL via |
| `NOW() - INTERVAL '1 second' * $7`. This avoids asyncpg/Postgres |
| deducing two conflicting types (integer vs double precision) for the |
| same placeholder when duration_seconds is also bound to the |
| INT column total_duration_seconds ($7). |
| """ |
| global _pool |
| if _pool is None: |
| await init_db_pool() |
| if _pool is None: |
| return False |
|
|
| query = """ |
| INSERT INTO call_logs ( |
| id, company_id, phone_line_id, direction, status, |
| full_transcript_history, extracted_caller_data, |
| total_duration_seconds, started_at, ended_at |
| ) VALUES ( |
| $1, $2, $3, 'inbound', $4, $5, $6, $7, $8, NOW() |
| ) |
| ON CONFLICT (id) DO UPDATE SET |
| status = EXCLUDED.status, |
| full_transcript_history = EXCLUDED.full_transcript_history, |
| extracted_caller_data = EXCLUDED.extracted_caller_data, |
| total_duration_seconds = EXCLUDED.total_duration_seconds, |
| ended_at = EXCLUDED.ended_at |
| """ |
|
|
| try: |
| extracted_json = json.dumps(extracted_caller_data) |
| duration_int = int(duration_seconds) |
| started_at = datetime.now(timezone.utc) - timedelta(seconds=duration_int) |
|
|
| async with _pool.acquire() as conn: |
| await conn.execute( |
| query, |
| call_id, |
| company_id, |
| phone_line_id, |
| status, |
| full_transcript, |
| extracted_json, |
| duration_int, |
| started_at, |
| ) |
| logger.info(f"Successfully saved call log to database: {call_id}") |
| return True |
| except Exception as e: |
| logger.error(f"Failed to insert call log to database: {e}") |
| return False |
