"""
SMS management routes — list messages, send SMS, view timeline, metrics.
Management routes for authenticated business users.
Webhook routes (inbound SMS, delivery status, call events) live in sms_webhooks.py.
"""
import logging
from datetime import datetime, timezone, timedelta
from functools import wraps
from flask import Blueprint, request, jsonify, current_app
from flask_login import login_required, current_user
from sqlalchemy import select, func, and_, or_
from app import db
from app.models import SmsMessage, Call, LeadTouchpoint, OptOut, AngiLead, Company, User
from app.connectors import build_connector, ConnectorError
from app.utils.csrf import require_csrf
logger = logging.getLogger(__name__)
def _get_sms_connector(company_id):
"""Look up and build the active SMS connector for a company."""
from app.models import Connector
connector = Connector.query.filter_by(
company_id=company_id,
service='sms',
active=True,
).first()
if not connector:
raise ConnectorError('No active SMS connector configured for this company')
return build_connector('sms', company_id, connector.config, connector.id)
sms_bp = Blueprint('sms', __name__, url_prefix='/api/sms')
def _current_company_id():
"""Resolve the current user's company_id.
User has NO ``company_id`` column — membership goes through the
UserCompany association table. Honors an optional ``?company_id=``
query param (membership-checked via tenancy helpers), otherwise
falls back to the user's first company membership. Aborts 403 if
the user has no company access.
"""
from flask import abort
from app.models import UserCompany
from app.utils.tenancy import resolve_company_id
membership = UserCompany.query.filter_by(user_id=current_user.id).first()
default_id = membership.company_id if membership else None
company_id = resolve_company_id(default=default_id)
if not company_id:
abort(403, description='No company access')
return company_id
# ---------------------------------------------------------------------------
# Auth helper
# ---------------------------------------------------------------------------
def company_required(f):
"""Require current_user to be associated with the requested company."""
@wraps(f)
def decorated(*args, **kwargs):
company_id = kwargs.get('company_id') or request.args.get('company_id')
if not company_id:
return jsonify({'error': 'company_id required'}), 400
# Check user membership via users_companies (tenancy helper)
from app.utils.tenancy import user_in_company
if getattr(current_user, 'role', None) != 'super_admin' and \
not user_in_company(current_user.id, company_id):
return jsonify({'error': 'Access denied'}), 403
return f(*args, **kwargs)
return decorated
# ---------------------------------------------------------------------------
# Outbound SMS
# ---------------------------------------------------------------------------
@sms_bp.route('/send', methods=['POST'])
@login_required
@require_csrf
def send_sms():
"""Send outbound SMS to a lead's phone number."""
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
company_id = _current_company_id()
to_phone = data.get('to_phone', '').strip()
body = data.get('body', '').strip()
lead_id = data.get('lead_id')
if not to_phone or not body:
return jsonify({'error': 'to_phone and body are required'}), 400
if len(body) > 1600:
return jsonify({'error': 'Message body exceeds 1600 characters'}), 400
try:
connector = _get_sms_connector(company_id)
result = connector.send_sms(
company_id=company_id,
to_phone=to_phone,
body=body,
lead_id=lead_id,
)
return jsonify({'success': True, **result}), 200
except ConnectorError as e:
logger.error(f'SMS send connector error: {e}')
return jsonify({'error': 'SMS provider error'}), 400
except Exception as e:
logger.error(f'SMS send failed: {e}')
return jsonify({'error': 'Failed to send SMS'}), 500
# ---------------------------------------------------------------------------
# Message list
# ---------------------------------------------------------------------------
@sms_bp.route('/messages', methods=['GET'])
@login_required
def list_messages():
"""List SMS messages with optional filters."""
company_id = _current_company_id()
lead_id = request.args.get('lead_id')
direction = request.args.get('direction') # inbound, outbound
status = request.args.get('status') # queued, sent, delivered, failed
phone = request.args.get('phone')
limit = min(int(request.args.get('limit', 50)), 200)
offset = int(request.args.get('offset', 0))
query = select(SmsMessage).where(
SmsMessage.company_id == company_id
)
if lead_id:
query = query.where(SmsMessage.lead_id == lead_id)
if direction:
query = query.where(SmsMessage.direction == direction)
if status:
query = query.where(SmsMessage.status == status)
if phone:
query = query.where(
or_(
SmsMessage.from_phone == phone,
SmsMessage.to_phone == phone,
)
)
query = query.order_by(SmsMessage.created_at.desc()).offset(offset).limit(limit)
results = db.session.execute(query).scalars().all()
# Get count for pagination
count_query = select(func.count()).select_from(SmsMessage).where(
SmsMessage.company_id == company_id
)
if lead_id:
count_query = count_query.where(SmsMessage.lead_id == lead_id)
total = db.session.execute(count_query).scalar_one()
return jsonify({
'messages': [m.to_dict() for m in results],
'total': total,
'limit': limit,
'offset': offset,
})
# ---------------------------------------------------------------------------
# Message detail
# ---------------------------------------------------------------------------
@sms_bp.route('/messages/<message_id>', methods=['GET'])
@login_required
def get_message(message_id):
"""Get single SMS message with full details."""
company_id = _current_company_id()
message = db.session.execute(
select(SmsMessage).where(
and_(SmsMessage.id == message_id, SmsMessage.company_id == company_id)
)
).scalar_one_or_none()
if not message:
return jsonify({'error': 'Message not found'}), 404
return jsonify(message.to_dict())
# ---------------------------------------------------------------------------
# Outbound Call
# ---------------------------------------------------------------------------
@sms_bp.route('/call', methods=['POST'])
@login_required
@require_csrf
def trigger_call():
"""Trigger outbound call to a phone number."""
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
company_id = _current_company_id()
to_phone = data.get('to_phone', '').strip()
lead_id = data.get('lead_id')
twiml_url = data.get('twiml_url')
if not to_phone:
return jsonify({'error': 'to_phone is required'}), 400
try:
connector = _get_sms_connector(company_id)
result = connector.make_call(
company_id=company_id,
to_phone=to_phone,
lead_id=lead_id,
twiml_url=twiml_url,
)
return jsonify({'success': True, **result}), 200
except ConnectorError as e:
logger.error(f'Call trigger connector error: {e}')
return jsonify({'error': 'Call provider error'}), 400
except Exception as e:
logger.error(f'Call trigger failed: {e}')
return jsonify({'error': 'Failed to trigger call'}), 500
# ---------------------------------------------------------------------------
# Call list
# ---------------------------------------------------------------------------
@sms_bp.route('/calls', methods=['GET'])
@login_required
def list_calls():
"""List calls with optional filters."""
company_id = _current_company_id()
lead_id = request.args.get('lead_id')
disposition = request.args.get('disposition')
limit = min(int(request.args.get('limit', 50)), 200)
offset = int(request.args.get('offset', 0))
query = select(Call).where(Call.company_id == company_id)
if lead_id:
query = query.where(Call.lead_id == lead_id)
if disposition:
query = query.where(Call.disposition == disposition)
query = query.order_by(Call.created_at.desc()).offset(offset).limit(limit)
results = db.session.execute(query).scalars().all()
count_query = select(func.count()).select_from(Call).where(
Call.company_id == company_id
)
if lead_id:
count_query = count_query.where(Call.lead_id == lead_id)
total = db.session.execute(count_query).scalar_one()
return jsonify({
'calls': [c.to_dict() for c in results],
'total': total,
'limit': limit,
'offset': offset,
})
# ---------------------------------------------------------------------------
# Call disposition update
# ---------------------------------------------------------------------------
@sms_bp.route('/calls/<call_id>/disposition', methods=['PUT'])
@login_required
@require_csrf
def update_call_disposition(call_id):
"""Update call disposition after agent review."""
company_id = _current_company_id()
call = db.session.execute(
select(Call).where(
and_(Call.id == call_id, Call.company_id == company_id)
)
).scalar_one_or_none()
if not call:
return jsonify({'error': 'Call not found'}), 404
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
disposition = data.get('disposition')
valid_dispositions = ('connected', 'voicemail', 'no_answer', 'busy', 'appointment_set', 'unknown')
if disposition and disposition not in valid_dispositions:
return jsonify({
'error': f'Invalid disposition. Must be one of: {", ".join(valid_dispositions)}'
}), 400
call.disposition = disposition or call.disposition
db.session.commit()
# Update lead status if appointment_set
if disposition == 'appointment_set' and call.lead_id:
lead = db.session.get(AngiLead, call.lead_id)
if lead:
lead.status = 'appointment_set'
lead.updated_at = datetime.now(timezone.utc)
db.session.commit()
# Create touchpoint for disposition change
if disposition:
touchpoint_map = {
'connected': 'call_connected',
'voicemail': 'call_voicemail',
'no_answer': 'call_no_answer',
'appointment_set': 'call_connected',
'busy': 'call_no_answer',
'unknown': 'call_outbound',
}
tp_type = touchpoint_map.get(disposition, 'call_outbound')
tp = LeadTouchpoint(
company_id=company_id,
lead_id=call.lead_id,
touchpoint_type=tp_type,
direction='system',
content=f'Call disposition: {disposition}',
reference_id=call.id,
reference_type='call',
)
db.session.add(tp)
db.session.commit()
return jsonify({'success': True, **call.to_dict()})
# ---------------------------------------------------------------------------
# Lead Touchpoint Timeline
# ---------------------------------------------------------------------------
@sms_bp.route('/leads/<lead_id>/timeline', methods=['GET'])
@login_required
def lead_timeline(lead_id):
"""Get unified chronological timeline for a lead (SMS + calls + status changes)."""
company_id = _current_company_id()
# Verify lead belongs to company
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
touchpoints = db.session.execute(
select(LeadTouchpoint).where(
and_(
LeadTouchpoint.company_id == company_id,
LeadTouchpoint.lead_id == lead_id,
)
).order_by(LeadTouchpoint.created_at)
).scalars().all()
return jsonify({
'lead_id': lead_id,
'lead_status': lead.status,
'touchpoints': [tp.to_dict() for tp in touchpoints],
'total': len(touchpoints),
})
# ---------------------------------------------------------------------------
# Opt-Out List
# ---------------------------------------------------------------------------
@sms_bp.route('/opt-outs', methods=['GET'])
@login_required
def list_opt_outs():
"""List opt-out suppression entries."""
company_id = _current_company_id()
limit = min(int(request.args.get('limit', 100)), 500)
offset = int(request.args.get('offset', 0))
query = select(OptOut).where(OptOut.company_id == company_id)
query = query.order_by(OptOut.created_at.desc()).offset(offset).limit(limit)
results = db.session.execute(query).scalars().all()
total = db.session.execute(
select(func.count()).select_from(OptOut).where(OptOut.company_id == company_id)
).scalar_one()
return jsonify({
'opt_outs': [o.to_dict() for o in results],
'total': total,
'limit': limit,
'offset': offset,
})
@sms_bp.route('/opt-outs/<opt_out_id>', methods=['DELETE'])
@login_required
@require_csrf
def remove_opt_out(opt_out_id):
"""Remove a phone from opt-out list (manual override)."""
company_id = _current_company_id()
opt_out = db.session.execute(
select(OptOut).where(
and_(OptOut.id == opt_out_id, OptOut.company_id == company_id)
)
).scalar_one_or_none()
if not opt_out:
return jsonify({'error': 'Opt-out not found'}), 404
phone = opt_out.phone
db.session.delete(opt_out)
db.session.commit()
return jsonify({'success': True, 'removed_phone': phone})
@sms_bp.route('/opt-outs', methods=['POST'])
@login_required
@require_csrf
def add_opt_out():
"""Manually add a phone to the opt-out suppression list."""
data = request.get_json()
if not data:
return jsonify({'error': 'JSON body required'}), 400
company_id = _current_company_id()
phone = data.get('phone', '').strip()
reason = data.get('reason', 'manual')
if not phone:
return jsonify({'error': 'phone is required'}), 400
if reason not in ('stop', 'manual', 'bounced', 'complaint'):
return jsonify({
'error': f'Invalid reason. Must be one of: stop, manual, bounced, complaint'
}), 400
# Check if already opted out
existing = db.session.execute(
select(OptOut).where(
and_(OptOut.company_id == company_id, OptOut.phone == phone)
)
).scalar_one_or_none()
if existing:
return jsonify({'error': f'Phone {phone} is already on the suppression list'}), 409
opt_out = OptOut(
company_id=company_id,
phone=phone,
reason=reason,
)
db.session.add(opt_out)
db.session.commit()
return jsonify({'success': True, **opt_out.to_dict()}), 201
# ---------------------------------------------------------------------------
# Speed-to-Lead Metrics
# ---------------------------------------------------------------------------
@sms_bp.route('/leads/<lead_id>/metrics', methods=['GET'])
@login_required
def lead_metrics(lead_id):
"""Get time-to-first-contact metrics for a lead."""
company_id = _current_company_id()
lead = db.session.get(AngiLead, lead_id)
if not lead or lead.company_id != company_id:
return jsonify({'error': 'Lead not found'}), 404
# Time from lead creation to first outbound touchpoint (SMS or call)
first_contact = db.session.execute(
select(LeadTouchpoint).where(
and_(
LeadTouchpoint.lead_id == lead_id,
LeadTouchpoint.direction == 'outbound',
LeadTouchpoint.touchpoint_type.in_(['sms_sent', 'call_outbound']),
)
).order_by(LeadTouchpoint.created_at).limit(1)
).scalar_one_or_none()
time_to_first_sms = None
time_to_first_call = None
# First SMS
first_sms = db.session.execute(
select(SmsMessage).where(
and_(
SmsMessage.lead_id == lead_id,
SmsMessage.direction == 'outbound',
)
).order_by(SmsMessage.created_at).limit(1)
).scalar_one_or_none()
# First call
first_call = db.session.execute(
select(Call).where(
and_(
Call.lead_id == lead_id,
Call.direction == 'outbound',
)
).order_by(Call.created_at).limit(1)
).scalar_one_or_none()
if first_sms and lead.created_at:
time_to_first_sms = (first_sms.created_at - lead.created_at).total_seconds() / 60 # minutes
if first_call and lead.created_at:
time_to_first_call = (first_call.created_at - lead.created_at).total_seconds() / 60 # minutes
time_to_first_contact = None
if first_contact and lead.created_at:
time_to_first_contact = (first_contact.created_at - lead.created_at).total_seconds() / 60 # minutes
return jsonify({
'lead_id': lead_id,
'lead_created_at': lead.created_at.isoformat() if lead.created_at else None,
'time_to_first_contact_minutes': round(time_to_first_contact, 1) if time_to_first_contact is not None else None,
'time_to_first_sms_minutes': round(time_to_first_sms, 1) if time_to_first_sms is not None else None,
'time_to_first_call_minutes': round(time_to_first_call, 1) if time_to_first_call is not None else None,
'first_contact_type': first_contact.touchpoint_type if first_contact else None,
'first_contact_at': first_contact.created_at.isoformat() if first_contact and first_contact.created_at else None,
})
@sms_bp.route('/metrics/summary', methods=['GET'])
@login_required
def sms_metrics_summary():
"""Get company-wide SMS/call metrics for a time period."""
company_id = _current_company_id()
days = int(request.args.get('days', 7))
since = datetime.now(timezone.utc) - timedelta(days=days)
# SMS counts by status
sms_counts = db.session.execute(
select(
SmsMessage.status,
func.count().label('count'),
).where(
and_(
SmsMessage.company_id == company_id,
SmsMessage.direction == 'outbound',
SmsMessage.created_at >= since,
)
).group_by(SmsMessage.status)
).all()
# Call counts by disposition
call_counts = db.session.execute(
select(
Call.disposition,
func.count().label('count'),
).where(
and_(
Call.company_id == company_id,
Call.created_at >= since,
)
).group_by(Call.disposition)
).all()
# Average time to first contact
avg_response = db.session.execute(
select(func.avg(AngiLead.response_time_minutes)).where(
and_(
AngiLead.company_id == company_id,
AngiLead.created_at >= since,
AngiLead.response_time_minutes.isnot(None),
)
)
).scalar_one_or_none()
# Keyword stats
keyword_counts = db.session.execute(
select(
SmsMessage.keyword_flag,
func.count().label('count'),
).where(
and_(
SmsMessage.company_id == company_id,
SmsMessage.direction == 'inbound',
SmsMessage.keyword_flag != '',
SmsMessage.created_at >= since,
)
).group_by(SmsMessage.keyword_flag)
).all()
# Opt-out count
opt_out_count = db.session.execute(
select(func.count()).select_from(OptOut).where(
OptOut.company_id == company_id
)
).scalar_one()
return jsonify({
'period_days': days,
'sms_by_status': {row.status: row.count for row in sms_counts},
'calls_by_disposition': {row.disposition: row.count for row in call_counts},
'avg_response_time_minutes': round(avg_response, 1) if avg_response else None,
'keyword_counts': {row.keyword_flag: row.count for row in keyword_counts},
'opt_out_count': opt_out_count,
})
# ---------------------------------------------------------------------------
# Connector config
# ---------------------------------------------------------------------------
@sms_bp.route('/connector/status', methods=['GET'])
@login_required
def connector_status():
"""Check SMS connector status and configuration."""
company_id = _current_company_id()
try:
connector = _get_sms_connector(company_id)
return jsonify({
'connected': True,
'type': connector.connector_type,
'name': connector.connector_name,
'from_phone': connector.get_from_phone(),
})
except Exception as e:
logger.error(f'SMS connector status check failed: {e}')
return jsonify({
'connected': False,
'error': 'An error occurred',
})