""" Call Persistence — the data-access layer for calls and call flows. Holds the on-hangup persistence hook plus the query/write functions that both the REST handlers and the MCP tools call, so the two surfaces can't drift. Every function takes an AsyncSession; callers own the transaction (get_db for REST, session_scope for MCP/services). """ import logging import uuid from datetime import datetime from sqlalchemy import desc, select from sqlalchemy.ext.asyncio import AsyncSession from db.database import ( CallRecord, RecordingRecord, StoredCallFlow, TranscriptChunk, session_scope, ) from models.call import ActiveCall, CallStatus from models.call_flow import CallFlow, CallFlowStep logger = logging.getLogger(__name__) def flow_to_model(row: StoredCallFlow) -> CallFlow: """The one StoredCallFlow-row → CallFlow-model mapping.""" return CallFlow( id=row.id, name=row.name, phone_number=row.phone_number, description=row.description or "", steps=[CallFlowStep(**s) for s in (row.steps or [])], tags=row.tags or [], notes=row.notes, avg_hold_time=row.avg_hold_time, success_rate=row.success_rate, last_used=row.last_used, times_used=row.times_used or 0, ) # ================================================================ # Call flows # ================================================================ async def get_flow(session: AsyncSession, flow_id: str) -> StoredCallFlow | None: result = await session.execute( select(StoredCallFlow).where(StoredCallFlow.id == flow_id) ) return result.scalar_one_or_none() async def get_flow_by_number( session: AsyncSession, phone_number: str ) -> StoredCallFlow | None: result = await session.execute( select(StoredCallFlow).where(StoredCallFlow.phone_number == phone_number) ) return result.scalar_one_or_none() async def list_flows(session: AsyncSession) -> list[StoredCallFlow]: result = await session.execute(select(StoredCallFlow)) return list(result.scalars().all()) async def create_flow( session: AsyncSession, flow_id: str, name: str, phone_number: str, steps: list[dict], description: str | None = None, tags: list[str] | None = None, notes: str | None = None, ) -> StoredCallFlow: row = StoredCallFlow( id=flow_id, name=name, phone_number=phone_number, description=description, steps=steps, tags=tags, notes=notes, last_verified=datetime.now(), ) session.add(row) await session.flush() return row # ================================================================ # Call history / records # ================================================================ async def get_record(session: AsyncSession, call_id: str) -> CallRecord | None: result = await session.execute( select(CallRecord).where(CallRecord.id == call_id) ) return result.scalar_one_or_none() async def search_history( session: AsyncSession, number: str | None = None, number_contains: str | None = None, intent_contains: str | None = None, status: str | None = None, since: datetime | None = None, until: datetime | None = None, limit: int = 50, offset: int = 0, ) -> list[CallRecord]: stmt = select(CallRecord).order_by(desc(CallRecord.started_at)) if number: stmt = stmt.where(CallRecord.remote_number == number) if number_contains: stmt = stmt.where(CallRecord.remote_number.contains(number_contains)) if intent_contains: stmt = stmt.where(CallRecord.intent.icontains(intent_contains)) if status: stmt = stmt.where(CallRecord.status == status) if since: stmt = stmt.where(CallRecord.started_at >= since) if until: stmt = stmt.where(CallRecord.started_at <= until) result = await session.execute(stmt.offset(offset).limit(limit)) return list(result.scalars().all()) async def get_transcript_chunks( session: AsyncSession, call_id: str ) -> list[TranscriptChunk]: result = await session.execute( select(TranscriptChunk) .where(TranscriptChunk.call_id == call_id) .order_by(TranscriptChunk.seq) ) return list(result.scalars().all()) async def latest_recording( session: AsyncSession, call_id: str ) -> RecordingRecord | None: result = await session.execute( select(RecordingRecord) .where(RecordingRecord.call_id == call_id) .order_by(desc(RecordingRecord.started_at)) ) return result.scalars().first() async def persist_call_on_end(call: ActiveCall, final_status: CallStatus) -> None: """Insert a CallRecord and any transcript chunks for `call`. Wired into CallManager as its on_call_ended hook by the composition root in main.py. """ try: async with session_scope() as session: record = CallRecord( id=call.id, direction=call.direction, remote_number=call.remote_number, status=final_status.value, mode=call.mode.value, intent=call.intent, started_at=call.started_at, ended_at=datetime.now(), duration=int(call.duration), hold_time=int(call.hold_time), device_used=call.device, call_flow_id=call.call_flow_id, classification_timeline=[ { "timestamp": c.timestamp, "audio_type": c.audio_type.value, "confidence": c.confidence, } for c in call.classification_history ], metadata_={"services": list(call.services)}, ) session.add(record) # Each transcript chunk gets its own row with a sequence number # so the dashboard can render them in order with click-to-seek. for seq, text in enumerate(call.transcript_chunks): speaker = "unknown" payload = text if ":" in text: head, rest = text.split(":", 1) head = head.strip().lower() if head in {"caller", "agent", "receptionist", "caller_message"}: speaker = head if head != "caller_message" else "caller" payload = rest.strip() session.add(TranscriptChunk( id=f"tc_{uuid.uuid4().hex[:10]}", call_id=call.id, seq=seq, t_offset_ms=0, speaker=speaker, text=payload, )) except Exception as e: logger.warning(f"Could not persist call {call.id}: {e}")