File size: 7,877 Bytes
e3d8805 2103c03 e3d8805 f84395e 2103c03 e3d8805 f84395e e3d8805 f84395e e3d8805 59431d4 e3d8805 59431d4 e3d8805 f84395e e3d8805 2103c03 e3d8805 2103c03 e3d8805 2103c03 e3d8805 2103c03 e3d8805 2103c03 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 | # neon_client.py
# Database client for Neon Postgres using asyncpg connection pool
#
# v2 (FIX F):
# insert_call_log() previously reused the same placeholder ($7,
# duration_seconds) both as an INT column value (total_duration_seconds)
# AND inside `NOW() - INTERVAL '1 second' * $7`, where Postgres infers
# the multiplication operand should be `double precision`. asyncpg's
# prepared-statement planner then sees two conflicting inferred types
# for the same parameter ("inconsistent types deduced for parameter $7
# ... double precision versus integer"), and every insert failed.
#
# Fix: compute `started_at` as a plain Python datetime
# (utcnow() - timedelta(seconds=duration_seconds)) and pass it as its
# own TIMESTAMPTZ parameter. duration_seconds is now used only once,
# as a plain integer, with no SQL-side arithmetic/type inference.
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")
# Connection pool singleton
_pool: Optional[asyncpg.Pool] = None
# Cache dictionary: (company_id, line_token) -> (timestamp, config_dict)
_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:
# Create pool with minimum 1 and maximum 5 connections
_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
# Check cache hit (60s TTL)
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)
# Normalize caller details configuration list
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 |