""" SMS + Call connector — Twilio or Plivo, provider-abstracted. Handles outbound SMS, inbound webhooks, delivery status tracking, opt-out suppression, call triggering, and disposition logging. """ import json import time import hmac import hashlib import logging from datetime import datetime, timezone from typing import Optional import requests import requests.adapters from flask import request as flask_request from sqlalchemy import select, and_ from app.models import SmsMessage, Call, LeadTouchpoint, OptOut, AngiLead, Connector from app.connectors import BaseConnector, ConnectorError logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Provider abstraction layer # --------------------------------------------------------------------------- class SmsProvider: """Abstract SMS provider interface. Implement for each gateway.""" def send_sms(self, from_phone: str, to_phone: str, body: str) -> dict: raise NotImplementedError def make_call(self, from_phone: str, to_phone: str, url: str = '') -> dict: raise NotImplementedError def verify_webhook_signature(self, signature: str, body: str, url: str) -> bool: raise NotImplementedError class TwilioProvider(SmsProvider): """Twilio SMS + Voice provider.""" def __init__(self, account_sid: str, auth_token: str, from_phone: str): self.account_sid = account_sid self.auth_token = auth_token self.from_phone = from_phone self.base_url = 'https://api.twilio.com/2010-04-01' self.session = requests.Session() self.session.auth = (account_sid, auth_token) # Retry strategy: 3 retries with exponential backoff adapter = requests.adapters.HTTPAdapter(max_retries=requests.adapters.Retry( total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], )) self.session.mount('https://', adapter) self.session.headers.update({'Content-Type': 'application/x-www-form-urlencoded'}) def send_sms(self, from_phone: str, to_phone: str, body: str) -> dict: """Send outbound SMS via Twilio. Returns provider response dict.""" data = { 'From': from_phone or self.from_phone, 'To': to_phone, 'Body': body, } resp = self.session.post(f'{self.base_url}/Accounts/{self.account_sid}/Messages.json', data=data) resp.raise_for_status() result = resp.json() return { 'sid': result.get('sid', ''), 'status': result.get('status', 'queued'), 'direction': result.get('direction', 'outbound'), 'from_phone': result.get('from', ''), 'to_phone': result.get('to', ''), 'body': result.get('body', ''), 'price': result.get('price'), } def make_call(self, from_phone: str, to_phone: str, url: str = '') -> dict: """Trigger outbound call via Twilio. Returns provider response dict.""" data = { 'From': from_phone or self.from_phone, 'To': to_phone, } if url: data['Url'] = url # TwiML URL for call flow else: # Connect directly (conference/bridge style) data['Url'] = f'https://api.twilio.com/2010-04-01/Accounts/{self.account_sid}/Calls' resp = self.session.post(f'{self.base_url}/Accounts/{self.account_sid}/Calls.json', data=data) resp.raise_for_status() result = resp.json() return { 'sid': result.get('sid', ''), 'status': result.get('status', 'queued'), 'from_phone': result.get('from', {}).get('phoneNumber', from_phone or self.from_phone) if isinstance(result.get('from'), dict) else from_phone or self.from_phone, 'to_phone': result.get('to', {}).get('phoneNumber', to_phone) if isinstance(result.get('to'), dict) else to_phone, 'duration': result.get('duration'), } def verify_webhook_signature(self, signature: str, body: str, url: str) -> bool: """Verify X-Twilio-Signature HMAC-SHA1 header.""" # Fail closed: without an auth token we cannot verify anything, and an # empty HMAC key would let attackers forge signatures trivially. if not self.auth_token or not signature: return False # Compute expected signature: HMAC-SHA1 of (url + body) using auth_token as key # Note: Twilio uses the auth token as the secret, concatenated with the full request URL # and body. sig_to_check = url + body expected = hmac.new( self.auth_token.encode(), sig_to_check.encode(), hashlib.sha1 ).hexdigest() return hmac.compare_digest(expected, signature) class PlivoProvider(SmsProvider): """Plivo SMS + Voice provider.""" def __init__(self, auth_id: str, auth_token: str, from_phone: str): self.auth_id = auth_id self.auth_token = auth_token self.from_phone = from_phone self.base_url = 'https://api.plivo.com/v1' self.session = requests.Session() self.session.auth = (auth_id, auth_token) # Retry strategy: 3 retries with exponential backoff adapter = requests.adapters.HTTPAdapter(max_retries=requests.adapters.Retry( total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], )) self.session.mount('https://', adapter) self.session.headers.update({'Content-Type': 'application/json'}) def send_sms(self, from_phone: str, to_phone: str, body: str) -> dict: """Send outbound SMS via Plivo. Returns provider response dict.""" data = { 'src': from_phone or self.from_phone, 'dst': to_phone, 'text': body, 'type': 'text', } resp = self.session.post(f'{self.base_url}/Account/{self.auth_id}/Message/', json=data) resp.raise_for_status() result = resp.json() return { 'sid': result.get('message_uuid', ''), 'status': result.get('status', 'queued'), 'direction': 'outbound', 'from_phone': result.get('src', ''), 'to_phone': result.get('dst', ''), 'body': result.get('text', ''), 'price': result.get('unit_price', result.get('price')), } def make_call(self, from_phone: str, to_phone: str, url: str = '') -> dict: """Trigger outbound call via Plivo. Returns provider response dict.""" data = { 'src': from_phone or self.from_phone, 'dst': to_phone, 'answer_url': url, } resp = self.session.post(f'{self.base_url}/Account/{self.auth_id}/Call/', json=data) resp.raise_for_status() result = resp.json() return { 'sid': result.get('api_id', ''), 'status': result.get('message', 'queued'), 'from_phone': result.get('src', from_phone or self.from_phone), 'to_phone': result.get('dst', to_phone), 'duration': result.get('duration'), } def verify_webhook_signature(self, signature: str, body: str, url: str) -> bool: """Verify Plivo webhook HMAC-SHA256 signature.""" # Plivo signs the request body with the auth_token as the HMAC secret expected = hmac.new( self.auth_token.encode(), body.encode(), hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected, signature) # --------------------------------------------------------------------------- # SMSConnector — registered with the connector registry # --------------------------------------------------------------------------- class SmsConnector(BaseConnector): """ SMS + Call connector. - Sends outbound SMS with opt-out suppression - Records inbound SMS replies with keyword detection - Tracks delivery status via webhooks - Triggers calls from CALL ME keywords - Logs call disposition - Creates unified touchpoint timeline entries - Syncs lead status across channels """ connector_type = 'sms' connector_name = 'SMS + Call (Twilio)' _SERVICE = 'sms' # Twilio webhook routes are registered in app/routes/sms_webhooks.py webhook_routes = [ '/connectors/sms/webhook/sms', '/connectors/sms/webhook/call', ] def __init__(self, connector_id: str = None, settings: dict = None): # Extract company_id from settings or use a placeholder company_id = settings.get('company_id', '') if settings else '' super().__init__(company_id=company_id, config=settings or {}, connector_id=connector_id) account_sid = self.config.get('account_sid', '') auth_token = self.config.get('auth_token', '') from_phone = self.config.get('from_phone', '') provider = self.config.get('provider', 'twilio') if provider == 'twilio': if not account_sid or not auth_token: raise ConnectorError('Twilio account_sid and auth_token are required') self.provider = TwilioProvider(account_sid, auth_token, from_phone) elif provider == 'plivo': auth_id = self.config.get('auth_id', '') if not auth_id or not auth_token: raise ConnectorError('Plivo auth_id and auth_token are required') self.provider = PlivoProvider(auth_id, auth_token, from_phone) else: raise ConnectorError(f'Unsupported SMS provider: {provider}') self.from_phone = from_phone self.call_twiml_url = self.config.get('call_twiml_url', '') # ---- Abstract method implementations (BaseConnector contract) ---- def connect(self) -> dict: """Validate provider credentials by making a lightweight API call.""" from app import db provider_name = self.config.get('provider', 'twilio') try: if provider_name == 'twilio': # Verify by listing phone numbers (lightweight) resp = self.provider.session.get( f'{self.provider.base_url}/Accounts/{self.provider.account_sid}/IncomingPhoneNumbers.json', params={'PageSize': 1}, ) resp.raise_for_status() elif provider_name == 'plivo': # Verify by fetching account info (lightweight) resp = self.provider.session.get( f'{self.provider.base_url}/Account/{self.provider.auth_id}/', ) resp.raise_for_status() status = 'connected' error = None except Exception as e: status = 'error' error = str(e) # Update connector record connector = db.session.execute( select(Connector).where(Connector.id == self.connector_id) ).scalar_one_or_none() if connector: connector.status = status connector.last_sync_at = datetime.now(timezone.utc) if error: connector.error_message = error db.session.commit() return { 'status': status, 'provider': provider_name, 'from_phone': self.from_phone, 'error': error, } def disconnect(self) -> dict: """Clear connector credentials and mark as disconnected.""" from app import db connector = db.session.execute( select(Connector).where(Connector.id == self.connector_id) ).scalar_one_or_none() if connector: connector.status = 'disconnected' connector.last_sync_at = datetime.now(timezone.utc) db.session.commit() return {'status': 'disconnected'} def sync(self) -> dict: """SMS connectors are event-driven (webhook), not polling-based. Return last activity summary instead.""" return { 'status': 'up_to_date', 'mode': 'webhook', 'message': 'SMS connector is event-driven; no pull sync needed.', } def status(self) -> dict: """Return current connector health and recent activity.""" from app import db connector = db.session.execute( select(Connector).where(Connector.id == self.connector_id) ).scalar_one_or_none() result = { 'connected': connector.status == 'connected' if connector else False, 'provider': self.config.get('provider', 'twilio'), 'from_phone': self.from_phone, } if connector: result['status'] = connector.status result['last_sync_at'] = connector.last_sync_at.isoformat() if connector.last_sync_at else None result['error_message'] = connector.error_message # Recent SMS activity try: stats = self.get_sms_stats(self.company_id) result['sms_stats'] = stats result['call_stats'] = self.get_call_stats(self.company_id) except Exception: pass return result # ---- Provider access ---- def get_from_phone(self) -> str: """Get the configured business phone number.""" return self.from_phone or self.provider.from_phone if self.provider else '' def get_provider(self) -> 'SmsProvider': """Get the underlying provider instance for direct access.""" return self.provider # ---- Suppression check (MANDATORY — TCPA) ---- def is_opted_out(self, company_id: str, phone: str) -> bool: """Check if a phone number is on the suppression list.""" from app import db opt_out = db.session.execute( select(OptOut).where( and_(OptOut.company_id == company_id, OptOut.phone == phone) ) ).scalar_one_or_none() return opt_out is not None # ---- Outbound SMS ---- def send_sms(self, company_id: str, to_phone: str, body: str, lead_id: str = None) -> dict: """ Send outbound SMS with mandatory opt-out suppression check. Returns: dict with 'sms_id', 'status', 'provider_sid' Raises: ConnectorError if suppressed or provider fails """ # MANDATORY: check suppression list before every outbound SMS if self.is_opted_out(company_id, to_phone): raise ConnectorError(f'Phone {to_phone} is on the suppression list (TCPA opt-out)') from app import db # Send via provider provider_result = self.provider.send_sms( from_phone=self.from_phone, to_phone=to_phone, body=body, ) # Record the message sms = SmsMessage( company_id=company_id, connector_id=self.connector_id, lead_id=lead_id, from_phone=self.from_phone, to_phone=to_phone, direction='outbound', body=body, status=provider_result.get('status', 'queued'), provider_message_sid=provider_result.get('sid', ''), ) db.session.add(sms) db.session.flush() # Populate sms.id before creating touchpoint # Create touchpoint self._add_touchpoint( company_id=company_id, lead_id=lead_id, touchpoint_type='sms_sent', direction='outbound', content=body, reference_id=sms.id, reference_type='sms_message', ) # Update lead status: new/accepted -> sms_sent if lead_id: self._update_lead_status(lead_id, 'sms_sent') self._set_first_contact(lead_id) db.session.commit() return { 'sms_id': sms.id, 'status': sms.status, 'provider_sid': sms.provider_message_sid, 'to_phone': to_phone, 'body': body, } # ---- Outbound Call ---- def make_call(self, company_id: str, to_phone: str, lead_id: str = None, triggered_sms_id: str = None, twiml_url: str = None) -> dict: """ Trigger outbound call via Twilio. Returns: dict with 'call_id', 'status', 'provider_sid' Raises: ConnectorError if provider fails """ from app import db url = twiml_url or self.call_twiml_url # Trigger call via provider provider_result = self.provider.make_call( from_phone=self.from_phone, to_phone=to_phone, url=url, ) # Record the call call = Call( company_id=company_id, connector_id=self.connector_id, lead_id=lead_id, triggered_sms_id=triggered_sms_id, from_phone=self.from_phone, to_phone=to_phone, direction='outbound', provider_call_sid=provider_result.get('sid', ''), started_at=datetime.now(timezone.utc), ) db.session.add(call) # Create touchpoint self._add_touchpoint( company_id=company_id, lead_id=lead_id, touchpoint_type='call_outbound', direction='outbound', content=f'Outbound call to {to_phone}', reference_id=call.id, reference_type='call', ) # Update lead status if lead_id: self._update_lead_status(lead_id, 'call_attempted') self._set_first_contact(lead_id) db.session.commit() return { 'call_id': call.id, 'status': provider_result.get('status', 'queued'), 'provider_sid': call.provider_call_sid, 'to_phone': to_phone, } # ---- SMS Webhook handler ---- def handle_sms_webhook(self, company_id: str, data: dict) -> dict: """ Process inbound SMS webhook. Handles: - Inbound replies with keyword detection - Delivery status updates (sent/delivered/failed) - Opt-out processing (STOP keyword) Returns: dict with processing result """ from app import db # Normalize provider-specific webhook fields into a common format normalized = self._normalize_webhook_data(data) message_sid = normalized['message_sid'] from_phone = normalized['from_phone'] to_phone = normalized['to_phone'] body = normalized['body'] message_status = normalized['message_status'] error_code = normalized['error_code'] error_text = normalized['error_text'] # Determine if this is an inbound reply or a delivery status update sms_direction = normalized['direction'] if sms_direction == 'inbound' or not message_status or message_status in ('received',): # Inbound reply result = self._handle_inbound_reply(company_id, from_phone, to_phone, body, message_sid) else: # Delivery status update result = self._handle_delivery_status( company_id, message_sid, message_status, from_phone, to_phone, error_code, error_text ) return result def _normalize_webhook_data(self, data: dict) -> dict: """Map provider-specific webhook fields to a common internal format. Returns a dict with keys: message_sid, from_phone, to_phone, body, message_status, direction, error_code, error_text """ provider = self.config.get('provider', 'twilio') if provider == 'plivo': from_phone = data.get('from', '') to_phone = data.get('to', '') direction = data.get('direction', '') # Plivo does not send a Direction field; infer from phone comparison if not direction: if from_phone and to_phone: direction = 'inbound' if from_phone != to_phone else 'outbound' else: direction = '' return { 'message_sid': data.get('message_id', ''), 'from_phone': from_phone, 'to_phone': to_phone, 'body': data.get('text', ''), 'message_status': data.get('type', ''), 'direction': direction, 'error_code': data.get('status_code', ''), 'error_text': data.get('error_text', ''), } else: # Twilio (default) return { 'message_sid': data.get('MessageSid', ''), 'from_phone': data.get('From', ''), 'to_phone': data.get('To', ''), 'body': data.get('Body', ''), 'message_status': data.get('MessageStatus', ''), 'direction': data.get('Direction', ''), 'error_code': data.get('ErrorCode', ''), 'error_text': data.get('ErrorText', ''), } def _handle_inbound_reply(self, company_id: str, from_phone: str, to_phone: str, body: str, message_sid: str) -> dict: """Process inbound SMS reply with keyword detection.""" from app import db from app.utils.sms_keywords import detect_keyword, normalize_phone normalized_from = normalize_phone(from_phone) # Detect keywords keyword = detect_keyword(body) # Record inbound message sms = SmsMessage( company_id=company_id, connector_id=self.connector_id, from_phone=from_phone, to_phone=to_phone, direction='inbound', body=body, status='received', provider_message_sid=message_sid, keyword_flag=keyword, ) db.session.add(sms) # Try to find associated lead by normalized phone (normalize both sides) leads = db.session.execute( select(AngiLead).where(AngiLead.company_id == company_id) ).scalars().all() lead = None for l in leads: if normalize_phone(l.phone) == normalized_from: lead = l break if lead: sms.lead_id = lead.id # Create touchpoint self._add_touchpoint( company_id=company_id, lead_id=lead.id if lead else None, touchpoint_type='sms_received', direction='inbound', content=body, reference_id=sms.id, reference_type='sms_message', ) # Process keyword action if keyword == 'STOP': self._process_opt_out(company_id, normalized_from, sms.id) if lead: self._update_lead_status(lead.id, 'opted_out') db.session.commit() return {'action': 'opt_out', 'keyword': keyword, 'phone': from_phone} elif keyword == 'YES': if lead: self._update_lead_status(lead.id, 'replied') db.session.commit() return {'action': 'affirmative', 'keyword': keyword, 'phone': from_phone} elif keyword == 'CALL_ME': if lead: self._update_lead_status(lead.id, 'call_attempted') db.session.commit() # Trigger the call try: call_result = self.make_call( company_id=company_id, to_phone=from_phone, lead_id=lead.id if lead else None, triggered_sms_id=sms.id, ) return {'action': 'call_triggered', 'keyword': keyword, 'call': call_result} except Exception as e: logger.error(f'Failed to trigger call for CALL_ME from {from_phone}: {e}') return {'action': 'call_failed', 'keyword': keyword, 'error': str(e)} else: # Regular reply, no keyword if lead: self._update_lead_status(lead.id, 'replied') db.session.commit() return {'action': 'reply', 'phone': from_phone, 'body': body} def _handle_delivery_status(self, company_id: str, message_sid: str, status: str, from_phone: str, to_phone: str, error_code: str = '', error_text: str = '') -> dict: """Process delivery status webhook update.""" from app import db # Find the existing SMS message sms = db.session.execute( select(SmsMessage).where( and_( SmsMessage.company_id == company_id, SmsMessage.provider_message_sid == message_sid, ) ) ).scalar_one_or_none() if not sms: # Create a record if not found (webhook came before our record) sms = SmsMessage( company_id=company_id, connector_id=self.connector_id, from_phone=from_phone, to_phone=to_phone, direction='outbound', body='', status=status, provider_message_sid=message_sid, error_code=error_code, error_message=error_text, ) db.session.add(sms) old_status = sms.status sms.status = status sms.error_code = error_code sms.error_message = error_text sms.updated_at = datetime.now(timezone.utc) # Map Twilio status to our touchpoint types touchpoint_map = { 'sent': 'sms_sent', 'delivered': 'sms_delivered', 'failed': 'sms_failed', 'rejected': 'sms_failed', } tp_type = touchpoint_map.get(status) if tp_type: self._add_touchpoint( company_id=company_id, lead_id=sms.lead_id, touchpoint_type=tp_type, direction='system', content=f'SMS {status}: {error_text}' if status in ('failed', 'rejected') else f'SMS {status}', reference_id=sms.id, reference_type='sms_message', ) db.session.commit() return { 'action': 'delivery_status', 'message_sid': message_sid, 'old_status': old_status, 'new_status': status, } # ---- Call Webhook handler ---- def handle_call_webhook(self, company_id: str, data: dict) -> dict: """ Process call status webhook from Twilio. Handles: - Call connected/completed - Voicemail - No answer / busy - Duration & recording URL Returns: dict with disposition update """ from app import db call_sid = data.get('CallSid', '') call_status = data.get('CallStatus', '') duration = data.get('Duration', '0') from_phone = data.get('From', '') to_phone = data.get('To', '') recording_url = data.get('RecordingUrl', '') call_duration = data.get('CallDuration', '0') # Try to find existing call record call = db.session.execute( select(Call).where( and_( Call.company_id == company_id, Call.provider_call_sid == call_sid, ) ) ).scalar_one_or_none() if not call: # Create record from webhook data call = Call( company_id=company_id, connector_id=self.connector_id, from_phone=from_phone, to_phone=to_phone, direction='outbound', provider_call_sid=call_sid, started_at=datetime.now(timezone.utc), ) db.session.add(call) # Map Twilio call status to our disposition disposition_map = { 'completed': 'connected', 'answered': 'connected', 'no-answer': 'no_answer', 'busy': 'busy', 'failed': 'unknown', 'canceled': 'unknown', } disposition = disposition_map.get(call_status, 'unknown') call.disposition = disposition call.duration_seconds = int(duration) if duration else None if call_duration: call.duration_seconds = int(call_duration) call.recording_url = recording_url call.ended_at = datetime.now(timezone.utc) if call_status in ('completed', 'no-answer', 'busy', 'failed') else None # Create disposition touchpoint tp_type_map = { 'connected': 'call_connected', 'voicemail': 'call_voicemail', 'no_answer': 'call_no_answer', } tp_type = tp_type_map.get(disposition, 'call_outbound') self._add_touchpoint( company_id=company_id, lead_id=call.lead_id, touchpoint_type=tp_type, direction='system', content=f'Call {disposition}' + (f' ({call.duration_seconds}s)' if call.duration_seconds else ''), reference_id=call.id, reference_type='call', ) db.session.commit() return { 'action': 'call_status', 'call_sid': call_sid, 'disposition': disposition, 'duration': call.duration_seconds, } # ---- Opt-out processing ---- def _process_opt_out(self, company_id: str, phone: str, source_message_id: str = None): """Add phone to suppression list (TCPA compliance).""" from app import db existing = db.session.execute( select(OptOut).where( and_(OptOut.company_id == company_id, OptOut.phone == phone) ) ).scalar_one_or_none() if not existing: opt_out = OptOut( company_id=company_id, phone=phone, reason='stop', source_message_id=source_message_id, ) db.session.add(opt_out) logger.info(f'Opt-out recorded: {phone} (company={company_id})') # Update lead sms_opted_out flag leads = db.session.execute( select(AngiLead).where( and_( AngiLead.company_id == company_id, AngiLead.phone == phone, ) ) ).all() for (lead,) in leads: lead.sms_opted_out = True # ---- Lead status helpers ---- def _update_lead_status(self, lead_id: str, new_status: str): """Update lead status with transition rules.""" from app import db lead = db.session.execute( select(AngiLead).where(AngiLead.id == lead_id) ).scalar_one_or_none() if lead and lead.status != new_status: old_status = lead.status lead.status = new_status # Record status change touchpoint self._add_touchpoint( company_id=lead.company_id, lead_id=lead_id, touchpoint_type='status_change', direction='system', content=f'Status: {old_status} → {new_status}', reference_id=lead_id, reference_type='angi_lead_action', ) logger.info(f'Lead status updated: {lead_id} {old_status} → {new_status}') def _set_first_contact(self, lead_id: str): """Set first_contact_at on the lead if not already set.""" from app import db lead = db.session.execute( select(AngiLead).where(AngiLead.id == lead_id) ).scalar_one_or_none() if lead and not lead.first_contact_at: lead.first_contact_at = datetime.now(timezone.utc) # ---- Touchpoint helper ---- def _add_touchpoint(self, company_id: str, lead_id: str, touchpoint_type: str, direction: str, content: str, reference_id: str, reference_type: str): """Create a unified timeline touchpoint entry.""" from app import db tp = LeadTouchpoint( company_id=company_id, lead_id=lead_id, touchpoint_type=touchpoint_type, direction=direction, content=content, reference_id=reference_id, reference_type=reference_type, ) db.session.add(tp) # ---- Metrics ---- def get_time_to_first_contact(self, lead_id: str) -> dict: """ Calculate time-to-first-contact for a lead. Returns: dict with 'lead_id', 'received_at', 'first_contact_at', 'minutes_to_first_contact', 'first_contact_type' """ from app import db lead = db.session.execute( select(AngiLead).where(AngiLead.id == lead_id) ).scalar_one_or_none() if not lead: return {'error': 'Lead not found'} result = { 'lead_id': lead.id, 'received_at': lead.received_at.isoformat() if lead.received_at else None, 'first_contact_at': lead.first_contact_at.isoformat() if lead.first_contact_at else None, } if lead.received_at and lead.first_contact_at: delta = lead.first_contact_at - lead.received_at result['minutes_to_first_contact'] = delta.total_seconds() / 60 else: result['minutes_to_first_contact'] = None # Find first contact type from touchpoints first_tp = db.session.execute( select(LeadTouchpoint).where( and_( LeadTouchpoint.lead_id == lead_id, LeadTouchpoint.touchpoint_type.in_(['sms_sent', 'call_outbound']), ) ).order_by(LeadTouchpoint.created_at.asc()) ).scalar_one_or_none() result['first_contact_type'] = first_tp.touchpoint_type if first_tp else None return result def get_lead_touchpoint_timeline(self, lead_id: str) -> list: """Get chronological touchpoint timeline for a lead.""" from app import db touchpoints = db.session.execute( select(LeadTouchpoint).where( LeadTouchpoint.lead_id == lead_id, ).order_by(LeadTouchpoint.created_at.asc()) ).all() return [tp.to_dict() for (tp,) in touchpoints] # ---- Stats ---- def get_sms_stats(self, company_id: str) -> dict: """Get SMS statistics for a company.""" from app import db from sqlalchemy import func total = db.session.execute( select(func.count(SmsMessage.id)).where(SmsMessage.company_id == company_id) ).scalar_one() by_status = {} for (status, count) in db.session.execute( select(SmsMessage.status, func.count(SmsMessage.id)).where( SmsMessage.company_id == company_id ).group_by(SmsMessage.status) ).all(): by_status[status] = count inbound_count = db.session.execute( select(func.count(SmsMessage.id)).where( and_(SmsMessage.company_id == company_id, SmsMessage.direction == 'inbound') ) ).scalar_one() outbound_count = db.session.execute( select(func.count(SmsMessage.id)).where( and_(SmsMessage.company_id == company_id, SmsMessage.direction == 'outbound') ) ).scalar_one() return { 'total': total, 'inbound': inbound_count, 'outbound': outbound_count, 'by_status': by_status, } def get_call_stats(self, company_id: str) -> dict: """Get call statistics for a company.""" from app import db from sqlalchemy import func total = db.session.execute( select(func.count(Call.id)).where(Call.company_id == company_id) ).scalar_one() by_disposition = {} for (disposition, count) in db.session.execute( select(Call.disposition, func.count(Call.id)).where( Call.company_id == company_id ).group_by(Call.disposition) ).all(): by_disposition[disposition] = count avg_duration = db.session.execute( select(func.avg(Call.duration_seconds)).where( and_(Call.company_id == company_id, Call.duration_seconds.isnot(None)) ) ).scalar_one() return { 'total': total, 'by_disposition': by_disposition, 'avg_duration_seconds': float(avg_duration) if avg_duration else None, } # --------------------------------------------------------------------------- # Self-registration with the connector registry # --------------------------------------------------------------------------- from app.connectors import register_connector, _REGISTRY register_connector( 'sms', { 'service': 'sms', 'name': 'SMS + Call (Twilio/Plivo)', 'category': 'telephony', 'description': 'Send and receive SMS messages, trigger outbound calls, track delivery status, manage opt-outs.', 'requires_oauth': False, 'requires_webhook': True, 'webhook_routes': [ '/connectors/sms/webhook/sms', '/connectors/sms/webhook/call', ], 'config_fields': [ {'name': 'provider', 'label': 'Provider', 'type': 'select', 'options': ['twilio', 'plivo'], 'required': True}, {'name': 'account_sid', 'label': 'Twilio Account SID', 'type': 'text', 'required': False}, {'name': 'auth_id', 'label': 'Plivo Auth ID', 'type': 'text', 'required': False}, {'name': 'auth_token', 'label': 'Auth Token', 'type': 'password', 'required': True}, {'name': 'from_phone', 'label': 'Business Phone Number', 'type': 'text', 'required': True}, {'name': 'call_twiml_url', 'label': 'Call TwiML URL', 'type': 'text', 'required': False}, ], }, ) _REGISTRY['sms'] = SmsConnector