from flask import Flask, request, jsonify from flask_cors import CORS import pymysql import os from datetime import datetime, timedelta import time import json import logging import redis import re from functools import wraps from urllib.parse import unquote from config import Config from db_handler import DatabaseHandler from analytics import AnalyticsService from quality_parameters_handler import QualityParametersHandler from objection_handler import ObjectionClassificationsHandler from auth_handler import AuthHandler, require_auth, require_business_access from rag_handler import RAGHandler from lead_insights_service import build_lead_insights from leadsquared_service import LeadSquaredService # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Initialize Flask app app = Flask(__name__) app.config.from_object(Config) # Enable CORS for React frontend CORS(app, resources={ r"/*": { "origins": "*", # Allow all origins for development "methods": ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"], "allow_headers": ["Content-Type", "Authorization"] } }) # Initialize database handler db_handler = DatabaseHandler(app.config) # Initialize analytics service analytics_service = AnalyticsService(db_handler) # Initialize quality parameters handler quality_params_handler = QualityParametersHandler(app.config) # Initialize objection classifications handler objection_handler = ObjectionClassificationsHandler(app.config) # Initialize authentication handler auth_handler = AuthHandler(app.config) app.auth_handler = auth_handler # Make it accessible to decorators # Initialize RAG handler rag_handler = RAGHandler(app.config) _PRESALES_MAP_CACHE = {} _PRESALES_MAP_CACHE_TTL_SECONDS = 300 # Error handler decorator def handle_errors(f): @wraps(f) def decorated_function(*args, **kwargs): try: return f(*args, **kwargs) except Exception as e: logger.error(f"Error in {f.__name__}: {str(e)}", exc_info=True) return jsonify({ 'error': str(e), 'message': 'An error occurred processing your request' }), 500 return decorated_function def user_has_business_admin(user, bid): from rbac import is_business_admin return is_business_admin(user, bid) def _require_business_admin_or_master(bid): user = getattr(request, 'current_user', None) or {} if not user_has_business_admin(user, bid): return jsonify({'error': 'Business admin access required'}), 403 return None def get_sync_source_db_config(): return { 'host': os.getenv('SYNC_SOURCE_DB_HOST', app.config.get('DB_HOST', '127.0.0.1')), 'port': int(os.getenv('SYNC_SOURCE_DB_PORT', app.config.get('DB_PORT', 3306))), 'user': os.getenv('SYNC_SOURCE_DB_USER', app.config.get('DB_USER', 'admin')), 'password': os.getenv('SYNC_SOURCE_DB_PASSWORD', app.config.get('DB_PASSWORD', '')), 'database': os.getenv('SYNC_SOURCE_DB_NAME', app.config.get('DB_NAME', 'voicebot_cluster')), 'charset': 'utf8mb4' } def get_primary_db_config(): return { 'host': app.config.get('DB_HOST', '127.0.0.1'), 'port': app.config.get('DB_PORT', 3306), 'user': app.config.get('DB_USER', 'admin'), 'password': app.config.get('DB_PASSWORD', ''), 'database': app.config.get('DB_NAME', 'voicebot_cluster'), 'charset': 'utf8mb4' } def _as_bool(value, default=False): if value is None: return default return str(value).strip().lower() in ('1', 'true', 'yes', 'y', 'on') def _normalize_phone_variants(phone): digits = ''.join(ch for ch in str(phone or '') if ch.isdigit()) if not digits: return [] core10 = digits[-10:] if len(digits) > 10 else digits variants = {digits, f"+{digits}", core10} if len(core10) == 10: variants.update({ f"91{core10}", f"+91{core10}", f"0{core10}", }) return [v for v in variants if v] def _extract_lsq_lead_list(payload): if payload is None: return [] if isinstance(payload, list): return payload if isinstance(payload, dict): for key in ('List', 'Data', 'list', 'data'): value = payload.get(key) if isinstance(value, list): return value return [] def _get_presales_mapping_from_leadsquared(bid, groupname=None, row_count=250, force_refresh=False): safe_row_count = max(10, min(int(row_count or 250), 500)) cache_key = f"{bid}:{groupname or ''}:{safe_row_count}" now = int(time.time()) cached = _PRESALES_MAP_CACHE.get(cache_key) if cached and not force_refresh and (now - cached.get('ts', 0) <= _PRESALES_MAP_CACHE_TTL_SECONDS): return cached.get('data') creds = db_handler.get_crm_credentials(bid, 'leadsquared') if not creds or not creds.get('access_key') or not creds.get('secret_key') or not creds.get('is_active'): return { 'success': False, 'connected': False, 'message': 'LeadSquared integration not configured or inactive', 'groups': [], 'agents': [], 'customers': [], 'stats': { 'lsq_leads_fetched': 0, 'matched_customers': 0, 'mapped_rows': 0, } } service = LeadSquaredService( access_key=creds['access_key'], secret_key=creds['secret_key'], api_host=creds.get('api_host'), ) lsq_search = service.search_leads({ 'Paging': { 'Offset': 0, 'RowCount': safe_row_count, } }) if not lsq_search.get('success'): return { 'success': False, 'connected': True, 'message': lsq_search.get('message') or 'Failed to fetch leads from LeadSquared', 'groups': [], 'agents': [], 'customers': [], 'stats': { 'lsq_leads_fetched': 0, 'matched_customers': 0, 'mapped_rows': 0, } } leads = _extract_lsq_lead_list(lsq_search.get('data')) phone_to_lead = {} all_phone_variants = set() for lead in leads: phone = _lsq_field(lead, 'Phone', 'Mobile', 'PhoneNumber') if not phone: continue variants = _normalize_phone_variants(phone) for variant in variants: all_phone_variants.add(variant) if variant not in phone_to_lead: phone_to_lead[variant] = lead rows = db_handler.get_group_agent_customer_rows( bid=bid, customer_numbers=list(all_phone_variants), groupname=groupname ) group_index = {} agent_index = {} customer_rows = [] for row in rows: group = row.get('groupname') or '-' agent = row.get('agentname') or '-' customer_phone = row.get('customer_callinfo') or '' variants = _normalize_phone_variants(customer_phone) matched_lead = None for variant in variants: matched_lead = phone_to_lead.get(variant) if matched_lead: break if group not in group_index: group_index[group] = { 'groupname': group, 'totalCalls': 0, 'matchedCustomers': set(), 'agents': set(), } if agent not in agent_index: agent_index[agent] = { 'agentname': agent, 'groupnames': set(), 'totalCalls': 0, 'matchedCustomers': set(), } total_calls = int(row.get('total_calls') or 0) group_index[group]['totalCalls'] += total_calls group_index[group]['agents'].add(agent) group_index[group]['matchedCustomers'].add(customer_phone) agent_index[agent]['totalCalls'] += total_calls agent_index[agent]['groupnames'].add(group) agent_index[agent]['matchedCustomers'].add(customer_phone) customer_rows.append({ 'customer_callinfo': customer_phone, 'groupname': group, 'agentname': agent, 'total_calls': total_calls, 'last_call': row.get('last_call'), 'lsq_name': _lsq_field(matched_lead, 'FirstName', 'Name', 'ProspectName'), 'lsq_owner_name': _lsq_field(matched_lead, 'OwnerName', 'Owner', 'owner_name'), 'lsq_status': _lsq_field(matched_lead, 'LeadStatus', 'Status', 'lead_status'), }) groups = [] for value in group_index.values(): groups.append({ 'groupname': value['groupname'], 'totalCalls': value['totalCalls'], 'matchedCustomers': len(value['matchedCustomers']), 'agentsCount': len(value['agents']), }) groups.sort(key=lambda x: x['totalCalls'], reverse=True) agents = [] for value in agent_index.values(): groups_for_agent = sorted(value['groupnames']) agents.append({ 'agentname': value['agentname'], 'groupnames': groups_for_agent, 'groupname': groups_for_agent[0] if groups_for_agent else None, 'totalCalls': value['totalCalls'], 'matchedCustomers': len(value['matchedCustomers']), }) agents.sort(key=lambda x: x['totalCalls'], reverse=True) response = { 'success': True, 'connected': True, 'message': 'Pre-sales mapping generated from LeadSquared leads', 'groups': groups, 'agents': agents, 'customers': customer_rows, 'stats': { 'lsq_leads_fetched': len(leads), 'matched_customers': len({row['customer_callinfo'] for row in customer_rows}), 'mapped_rows': len(customer_rows), } } _PRESALES_MAP_CACHE[cache_key] = { 'ts': now, 'data': response, } return response # ============================================================================ # HEALTH CHECK # ============================================================================ @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint""" return jsonify({ 'status': 'healthy', 'timestamp': datetime.now().isoformat(), 'service': 'Call Analytics Dashboard API' }), 200 # ============================================================================ # AUTHENTICATION # ============================================================================ @app.route('/auth/register', methods=['POST']) @handle_errors def register(): """Register a new user""" data = request.json required_fields = ['bid', 'username', 'email', 'password'] for field in required_fields: if not data.get(field): return jsonify({'error': f'{field} is required'}), 400 result, status_code = auth_handler.create_user( bid=data['bid'], username=data['username'], email=data['email'], password=data['password'], full_name=data.get('full_name'), role=data.get('role', 'user') ) return jsonify(result), status_code @app.route('/auth/login', methods=['POST']) @handle_errors def login(): """Login user""" data = request.json if not data.get('username') or not data.get('password'): return jsonify({'error': 'Username and password are required'}), 400 # Get IP address and user agent for activity logging ip_address = request.remote_addr user_agent = request.headers.get('User-Agent') result, status_code = auth_handler.login( data['username'], data['password'], ip_address=ip_address, user_agent=user_agent ) return jsonify(result), status_code @app.route('/auth/logout', methods=['POST']) @handle_errors def logout(): """Logout user""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') result, status_code = auth_handler.logout(token) return jsonify(result), status_code @app.route('/auth/me', methods=['GET']) @handle_errors def get_current_user(): """Get current logged-in user info""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_session(token) if not user: return jsonify({'error': 'Invalid or expired session'}), 401 return jsonify({'user': user}), 200 @app.route('/auth/users/', methods=['GET']) @handle_errors def get_business_users(bid): """Get all users for a business (admin only)""" users, status_code = auth_handler.get_users_by_business(bid) return jsonify({'users': users}), status_code # ============================================================================ # BUSINESS MANAGEMENT # ============================================================================ @app.route('/list-businesses', methods=['GET']) @handle_errors def list_businesses(): """Get list of all available businesses with their call counts""" businesses = db_handler.get_all_businesses() return jsonify(businesses), 200 @app.route('/businesses//info', methods=['GET']) @handle_errors def get_business_info(bid): """Get detailed information for a specific business""" info = db_handler.get_business_info(bid) if not info: return jsonify({'error': 'Business not found'}), 404 return jsonify(info), 200 @app.route('/groupnames/', methods=['GET']) @handle_errors def get_groupnames(bid): """Get list of all groupnames for Jubilant Foods (bid 7987) with their call counts""" presales_only = _as_bool(request.args.get('presales_only'), default=False) if presales_only: row_count = request.args.get('lsq_row_count', 250, type=int) force_refresh = _as_bool(request.args.get('force_refresh'), default=False) mapping = _get_presales_mapping_from_leadsquared( bid=bid, groupname=None, row_count=row_count, force_refresh=force_refresh ) groupnames = mapping.get('groups', []) if not groupnames: groupnames = db_handler.get_all_groupnames(bid) else: groupnames = db_handler.get_all_groupnames(bid) return jsonify(groupnames), 200 @app.route('/agentnames/', methods=['GET']) @handle_errors def get_agentnames(bid): """Get list of agent names filtered by location/groupname""" groupname = request.args.get('groupname') presales_only = _as_bool(request.args.get('presales_only'), default=False) if presales_only: row_count = request.args.get('lsq_row_count', 250, type=int) force_refresh = _as_bool(request.args.get('force_refresh'), default=False) mapping = _get_presales_mapping_from_leadsquared( bid=bid, groupname=groupname, row_count=row_count, force_refresh=force_refresh ) agents = [agent.get('agentname') for agent in mapping.get('agents', []) if agent.get('agentname')] if not agents: agents = db_handler.get_agent_names(bid, groupname) else: agents = db_handler.get_agent_names(bid, groupname) return jsonify(agents), 200 @app.route('/location-stats/', methods=['GET']) @handle_errors def get_location_stats(bid): """ Get call statistics for a specific location (groupname) for Jubilant Foods Query params: - groupname: Filter by location/groupname (optional) """ groupname = request.args.get('groupname') date_from = request.args.get('date_from') date_to = request.args.get('date_to') agent_names = request.args.get('agent_names') detailed_calls = request.args.get('detailed_calls') detailed_threshold_seconds = request.args.get('detailed_threshold_seconds', 120) direction = request.args.get('direction') stats = db_handler.get_location_stats( bid, groupname, date_from, date_to, agent_names=agent_names, detailed_calls=detailed_calls, detailed_threshold_seconds=detailed_threshold_seconds, direction=direction, ) return jsonify(stats), 200 @app.route('/location-calls/', methods=['GET']) @handle_errors def get_location_calls(bid): """ Get filtered raw calls for Jubilant Foods Query params: - groupname: Filter by location/groupname (optional) - direction: Filter by direction (inbound/outbound) (optional) - call_status: Filter by call_status (ANSWER/BUSY/CANCEL/NOANSWER) (optional) - date_from: Start date (YYYY-MM-DD) (optional) - date_to: End date (YYYY-MM-DD) (optional) - limit: Number of records (default: 100) - offset: Offset for pagination (default: 0) """ groupname = request.args.get('groupname') direction = request.args.get('direction') call_status = request.args.get('call_status') date_from = request.args.get('date_from') date_to = request.args.get('date_to') limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) calls = db_handler.get_filtered_raw_calls( bid, groupname, direction, call_status, limit, offset, date_from, date_to, ) return jsonify(calls), 200 @app.route('/raw-calls//', methods=['GET']) @handle_errors def get_raw_call_details_route(bid, callid): """Get call details from 7987_raw_calls with transcript from 7987_sarvamresponse""" call = db_handler.get_raw_call_details(bid, callid) if not call: return jsonify({'error': 'Call not found'}), 404 return jsonify(call), 200 # ============================================================================ # CALL ANALYTICS (AWS NOVA) # ============================================================================ @app.route('/analytics//', methods=['GET']) @handle_errors def get_call_analytics(bid, callid): """Get analytics for a specific call""" analytics = db_handler.get_call_analytics(bid, callid) if not analytics: return jsonify({'error': 'Analytics not found for this call'}), 404 return jsonify(analytics), 200 @app.route('/analytics//pending', methods=['GET']) @handle_errors def get_pending_analytics(bid): """Get calls that need analytics""" limit = request.args.get('limit', 10, type=int) calls = db_handler.get_calls_for_analysis(bid, limit) return jsonify({ 'count': len(calls), 'calls': calls }), 200 @app.route('/analytics//dashboard', methods=['GET']) @handle_errors def get_analytics_dashboard(bid): """Get comprehensive analytics dashboard data""" groupname = request.args.get('groupname') date_from = request.args.get('date_from') date_to = request.args.get('date_to') # Get analytics overview overview = db_handler.get_analytics_overview(bid, groupname, date_from, date_to) # Get sentiment by location sentiment_by_location = db_handler.get_sentiment_by_location(bid, groupname, date_from, date_to) # Get quality score by location quality_by_location = db_handler.get_quality_by_location(bid, groupname, date_from, date_to) # Get quality score by agent quality_by_agent = db_handler.get_quality_by_agent(bid, groupname, date_from, date_to) # Get call purpose frequency call_purposes = db_handler.get_call_purpose_frequency(bid, groupname, date_from, date_to) # Get objections/concerns frequency concerns_frequency = db_handler.get_concerns_frequency(bid, groupname, date_from, date_to) # Get busy locations busy_locations = db_handler.get_busy_locations(bid, groupname, date_from, date_to) return jsonify({ 'overview': overview, 'sentiment_by_location': sentiment_by_location, 'quality_by_location': quality_by_location, 'quality_by_agent': quality_by_agent, 'call_purposes': call_purposes, 'concerns_frequency': concerns_frequency, 'busy_locations': busy_locations }), 200 @app.route('/analytics//calls-by-objection', methods=['GET']) @handle_errors def get_calls_by_objection(bid): """Get calls filtered by specific objection/concern""" objection = request.args.get('objection') groupname = request.args.get('groupname') if not objection: return jsonify({'error': 'Objection parameter is required'}), 400 calls = db_handler.get_calls_by_objection(bid, objection, groupname) return jsonify(calls), 200 # ============================================================================ # CALL MANAGEMENT # ============================================================================ @app.route('/calls/', methods=['GET']) @handle_errors def get_calls(bid): """ Get all calls for a business with optional filtering Query params: - status: Filter by status (0-3) - sales_intent: Filter by intent (High/Medium/Low) - date_from: Start date (YYYY-MM-DD) - date_to: End date (YYYY-MM-DD) - limit: Number of records (default: 100) - offset: Offset for pagination (default: 0) """ # Get query parameters status = request.args.get('status', type=int) sales_intent = request.args.get('sales_intent') date_from = request.args.get('date_from') date_to = request.args.get('date_to') limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) # Build filters filters = {} if status is not None: filters['status'] = status if sales_intent: filters['sales_intent'] = sales_intent if date_from: filters['date_from'] = date_from if date_to: filters['date_to'] = date_to calls = db_handler.get_calls(bid, filters, limit, offset) total_count = db_handler.get_calls_count(bid, filters) return jsonify({ 'calls': calls, 'total': total_count, 'limit': limit, 'offset': offset }), 200 @app.route('/calls//', methods=['GET']) @handle_errors def get_call_details(bid, callid): """Get detailed information for a specific call""" call = db_handler.get_call_by_id(bid, callid) if not call: return jsonify({'error': 'Call not found'}), 404 # Get transcript and diarization data from sarvamresponse table transcript_data = db_handler.get_call_transcript(bid, callid) if transcript_data: call['transcripts'] = transcript_data.get('transcript', '') call['language'] = transcript_data.get('language', '') speaker_segments = transcript_data.get('speaker_segments') if speaker_segments and isinstance(speaker_segments, str): try: speaker_segments = json.loads(speaker_segments) except json.JSONDecodeError: speaker_segments = [] call['speaker_segments'] = speaker_segments call['num_speakers'] = transcript_data.get('num_speakers') call['duration'] = transcript_data.get('duration') # Get analytics data from callanalytics table analytics_data = db_handler.get_call_analytics(bid, callid) if analytics_data: call['summary'] = analytics_data.get('summary', '') call['call_purpose'] = analytics_data.get('call_purpose', '') call['objections_concerns'] = analytics_data.get('objections_concerns', '') call['objection_type'] = analytics_data.get('objection_type', '') call['quality_score'] = analytics_data.get('quality_score') call['sentiments'] = analytics_data.get('sentiment', '') call['parameter_scores'] = analytics_data.get('parameter_scores') call['talk_listen_ratio'] = analytics_data.get('talk_listen_ratio') call['agent_speak_percentage'] = analytics_data.get('agent_speak_percentage') call['customer_speak_percentage'] = analytics_data.get('customer_speak_percentage') call['dead_air_percentage'] = analytics_data.get('dead_air_percentage') bant_data = db_handler.get_bant_analysis(bid, callid) if bant_data: call['bant_profile'] = bant_data.get('profile') call['bant_summary'] = bant_data.get('summary') return jsonify(call), 200 @app.route('/bant//', methods=['GET']) @handle_errors def get_bant_details(bid, callid): """Get BANT profile for a specific call""" bant = db_handler.get_bant_analysis(bid, callid) if not bant: return jsonify({'error': 'BANT not found'}), 404 return jsonify(bant), 200 @app.route('/calls///transcript', methods=['DELETE']) @handle_errors def delete_transcript(bid, callid): """Delete transcript and mark call for re-transcription""" try: # Delete from sarvamresponse table db_handler.delete_transcript(bid, callid) # Update raw_calls transcription status to 0 db_handler.reset_transcription_status(bid, callid) return jsonify({'message': 'Transcript deleted successfully', 'callid': callid}), 200 except Exception as e: logger.error(f"Error deleting transcript for {callid}: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/calls///segment/', methods=['PATCH']) @handle_errors def update_segment_text(bid, callid, segment_index): """Update the text of a specific speaker segment""" try: data = request.get_json() new_text = data.get('text', '').strip() if not new_text: return jsonify({'error': 'Text cannot be empty'}), 400 # Update segment in database db_handler.update_speaker_segment_text(bid, callid, segment_index, new_text) return jsonify({ 'message': 'Segment updated successfully', 'callid': callid, 'segment_index': segment_index }), 200 except Exception as e: logger.error(f"Error updating segment {segment_index} for {callid}: {str(e)}") return jsonify({'error': str(e)}), 500 @app.route('/calls//recent', methods=['GET']) @handle_errors def get_recent_calls(bid): """Get recent calls for a business""" limit = request.args.get('limit', 10, type=int) calls = db_handler.get_recent_calls(bid, limit) return jsonify(calls), 200 @app.route('/calls/search', methods=['POST']) @handle_errors def search_calls(): """ Search calls across all fields Body: { "bid": "6840", "query": "search term", "limit": 50 } """ data = request.get_json() bid = data.get('bid') query = data.get('query', '') limit = data.get('limit', 50) if not bid: return jsonify({'error': 'Business ID is required'}), 400 results = db_handler.search_calls(bid, query, limit) return jsonify(results), 200 @app.route('/leads/', methods=['GET']) @handle_errors def get_leads(bid): """Get customer-level lead aggregates built from raw calls.""" groupname = request.args.get('groupname') limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) transcripts_only = request.args.get('transcripts_only', 'false').lower() == 'true' result = db_handler.get_leads_list( bid=bid, groupname=groupname, limit=limit, offset=offset, transcripts_only=transcripts_only ) return jsonify({ 'leads': result.get('leads', []), 'total': int(result.get('total', 0)), 'limit': limit, 'offset': offset }), 200 def _invoke_lead_insights_chat(prompt): return rag_handler._invoke_chat_model(prompt) def _enrich_lead_details_with_bant(bid, details): for call in details.get('calls') or []: callid = call.get('callid') if not callid or call.get('bant_profile'): continue bant = db_handler.get_bant_analysis(bid, callid) if bant: call['bant_profile'] = bant.get('profile') call['bant_summary'] = bant.get('summary') or '' return details @app.route('/leads///insights', methods=['GET']) @handle_errors def get_lead_insights(bid, lead_phone): decoded_phone = unquote(lead_phone) details = db_handler.get_lead_details(bid=bid, lead_phone=decoded_phone) if not details: return jsonify({'error': 'Lead not found'}), 404 details = _enrich_lead_details_with_bant(bid, details) data_capture_fields = db_handler.list_data_capture_fields(bid) return jsonify({ 'insights': build_lead_insights(details, data_capture_fields=data_capture_fields) }), 200 @app.route('/leads///insights/generate', methods=['POST']) @handle_errors def generate_lead_insights(bid, lead_phone): decoded_phone = unquote(lead_phone) details = db_handler.get_lead_details(bid=bid, lead_phone=decoded_phone) if not details: return jsonify({'error': 'Lead not found'}), 404 details = _enrich_lead_details_with_bant(bid, details) calls_with_transcripts = [call for call in details.get("calls", []) if call.get("transcript")] if not calls_with_transcripts: return jsonify({'error': 'No transcripts found for this lead.'}), 400 data_capture_fields = db_handler.list_data_capture_fields(bid) return jsonify({ 'insights': build_lead_insights( details, invoke_chat=_invoke_lead_insights_chat, use_llm=True, data_capture_fields=data_capture_fields, ) }), 200 @app.route('/leads//', methods=['GET']) @handle_errors def get_lead_detail(bid, lead_phone): """Get lead detail timeline and optional CRM enrichment for one customer.""" decoded_phone = unquote(lead_phone) groupname = request.args.get('groupname') include_crm = request.args.get('include_crm', 'true').lower() == 'true' details = db_handler.get_lead_details( bid=bid, lead_phone=decoded_phone, groupname=groupname ) if not details: return jsonify({'error': 'Lead not found'}), 404 crm = { 'connected': False, 'matched': False, 'provider': 'leadsquared', 'lead': None } if include_crm: creds = db_handler.get_crm_credentials(bid, 'leadsquared') if creds and creds.get('access_key') and creds.get('secret_key') and creds.get('is_active'): crm['connected'] = True # Try local cache first (fast, no external API call) cached = db_handler.get_cached_crm_lead_by_phone(bid, 'leadsquared', decoded_phone) if cached and cached.get('lead_payload'): lsq_record = cached['lead_payload'] crm['matched'] = True crm['source'] = 'cache' crm['lead'] = { 'raw': lsq_record, 'name': cached.get('lead_name') or _lsq_field(lsq_record, 'FirstName', 'first_name', 'Name', 'ProspectName'), 'email': cached.get('email') or _lsq_field(lsq_record, 'EmailAddress', 'email'), 'phone': cached.get('phone_primary') or _lsq_field(lsq_record, 'Phone', 'Mobile', 'PhoneNumber'), 'status': cached.get('lead_status') or _lsq_field(lsq_record, 'LeadStatus', 'Status', 'lead_status'), 'owner_name': cached.get('owner_name') or _lsq_field(lsq_record, 'OwnerName', 'Owner', 'owner_name'), 'next_task_due_date': cached.get('next_task_due_date') or _lsq_field(lsq_record, 'NextTaskDueDate', 'next_task_due_date'), } else: # Cache miss — fall back to live LSQ API lookup service = LeadSquaredService( access_key=creds['access_key'], secret_key=creds['secret_key'], api_host=creds.get('api_host'), ) search_result = service.search_lead_by_phone(decoded_phone) if search_result.get('success'): lsq_record = _extract_lsq_lead_record(search_result.get('data')) if lsq_record: crm['matched'] = True crm['source'] = 'live' crm['lead'] = { 'raw': lsq_record, 'name': _lsq_field(lsq_record, 'FirstName', 'first_name', 'Name', 'ProspectName'), 'email': _lsq_field(lsq_record, 'EmailAddress', 'email'), 'phone': _lsq_field(lsq_record, 'Phone', 'Mobile', 'PhoneNumber'), 'status': _lsq_field(lsq_record, 'LeadStatus', 'Status', 'lead_status'), 'owner_name': _lsq_field(lsq_record, 'OwnerName', 'Owner', 'owner_name'), 'next_task_due_date': _lsq_field(lsq_record, 'NextTaskDueDate', 'next_task_due_date'), } details['crm'] = crm return jsonify(details), 200 # ============================================================================ # ANALYTICS ENDPOINTS # ============================================================================ @app.route('/analytics//stats', methods=['GET']) @handle_errors def get_call_stats(bid): """ Get overall statistics for a business Query params: - date_from: Start date (YYYY-MM-DD) - date_to: End date (YYYY-MM-DD) """ date_from = request.args.get('date_from') date_to = request.args.get('date_to') stats = analytics_service.get_call_statistics(bid, date_from, date_to) return jsonify(stats), 200 @app.route('/analytics//sentiment', methods=['GET']) @handle_errors def get_sentiment_analysis(bid): """Get sentiment distribution for a business""" date_from = request.args.get('date_from') date_to = request.args.get('date_to') sentiment = analytics_service.get_sentiment_distribution(bid, date_from, date_to) return jsonify(sentiment), 200 @app.route('/analytics//intent', methods=['GET']) @handle_errors def get_intent_distribution(bid): """Get sales intent distribution for a business""" date_from = request.args.get('date_from') date_to = request.args.get('date_to') intent = analytics_service.get_intent_distribution(bid, date_from, date_to) return jsonify(intent), 200 @app.route('/analytics//trends', methods=['GET']) @handle_errors def get_trends(bid): """ Get trends over time Query params: - period: day, week, month (default: day) - days: Number of days to look back (default: 7) """ period = request.args.get('period', 'day') days = request.args.get('days', 7, type=int) trends = analytics_service.get_trends(bid, period, days) return jsonify(trends), 200 @app.route('/analytics//agents', methods=['GET']) @handle_errors def get_agent_performance(bid): """Get performance metrics for all agents""" date_from = request.args.get('date_from') date_to = request.args.get('date_to') performance = analytics_service.get_agent_performance(bid, date_from, date_to) return jsonify(performance), 200 @app.route('/analytics//keywords', methods=['GET']) @handle_errors def get_top_keywords(bid): """Get most common keywords across all calls""" limit = request.args.get('limit', 20, type=int) date_from = request.args.get('date_from') date_to = request.args.get('date_to') keywords = analytics_service.get_top_keywords(bid, limit, date_from, date_to) return jsonify(keywords), 200 # ============================================================================ # CALL PROCESSING (Integration with existing system) # ============================================================================ @app.route('/queue-calls/', methods=['POST']) @handle_errors def queue_calls(bid): """Queue unprocessed calls for processing""" # Get unprocessed calls calls = db_handler.get_calls(bid, {'status': 0}, limit=1000) # Here you would integrate with your RabbitMQ system # For now, return the count of calls to be queued return jsonify({ 'message': f'Queued {len(calls)} calls for processing', 'count': len(calls), 'business_id': bid }), 200 @app.route('/process-calls/', methods=['POST']) @handle_errors def process_calls(bid): """Directly process calls (bypass queue)""" # This would integrate with your sarvam_processor # Return processing status return jsonify({ 'message': 'Processing started', 'business_id': bid }), 202 # ============================================================================ # TRANSCRIPTS # ============================================================================ @app.route('/transcripts/', methods=['GET']) @handle_errors def get_transcripts(bid): """Get all transcripts for a business""" transcripts = db_handler.get_transcripts(bid) return jsonify(transcripts), 200 # ============================================================================ # DATA EXPORT # ============================================================================ @app.route('/export//calls', methods=['GET']) @handle_errors def export_calls(bid): """Export calls data as JSON""" format_type = request.args.get('format', 'json') filters = { 'status': request.args.get('status', type=int), 'date_from': request.args.get('date_from'), 'date_to': request.args.get('date_to') } # Remove None values filters = {k: v for k, v in filters.items() if v is not None} calls = db_handler.get_calls(bid, filters, limit=10000) if format_type == 'csv': # Convert to CSV format import csv from io import StringIO output = StringIO() if calls: writer = csv.DictWriter(output, fieldnames=calls[0].keys()) writer.writeheader() writer.writerows(calls) return output.getvalue(), 200, { 'Content-Type': 'text/csv', 'Content-Disposition': f'attachment; filename=calls_{bid}_{datetime.now().strftime("%Y%m%d")}.csv' } return jsonify(calls), 200 def _split_csv_filter(value): return [item.strip() for item in str(value or "").split(",") if item.strip()] def _export_filters( groupname, date_from, date_to, agent_names=None, customer_name=None, detailed_calls=None, quality_min=None, quality_max=None, direction=None, call_status=None, include_quality_filters=False, ): where_clauses = [] params = [] if groupname: where_clauses.append("r.groupname = %s") params.append(groupname) if date_from: where_clauses.append("DATE(r.call_starttime) >= %s") params.append(date_from) if date_to: where_clauses.append("DATE(r.call_starttime) <= %s") params.append(date_to) agents = _split_csv_filter(agent_names) if agents: placeholders = ", ".join(["%s"] * len(agents)) where_clauses.append(f"r.agentname IN ({placeholders})") params.extend(agents) if customer_name: where_clauses.append("CAST(r.customer_callinfo AS CHAR) LIKE %s") params.append(f"%{customer_name}%") if direction: where_clauses.append("LOWER(r.direction) = LOWER(%s)") params.append(direction) if call_status: where_clauses.append("UPPER(r.call_status) = UPPER(%s)") params.append(call_status) detailed = str(detailed_calls or "").strip().lower() if detailed == "yes": where_clauses.append("TIMESTAMPDIFF(SECOND, r.call_starttime, r.call_endtime) > 120") elif detailed == "no": where_clauses.append("(TIMESTAMPDIFF(SECOND, r.call_starttime, r.call_endtime) <= 120 OR r.call_endtime IS NULL)") if include_quality_filters: if quality_min is not None: where_clauses.append("a.quality_score >= %s") params.append(quality_min) if quality_max is not None: where_clauses.append("a.quality_score <= %s") params.append(quality_max) return (" WHERE " + " AND ".join(where_clauses)) if where_clauses else "", params EXPORT_HEADER_LABELS = { "customer_number": "CUSTOMER NUMBER", "customer_name": "CUSTOMER NAME", "callid": "CALL ID", "agentname": "AGENT NAME", "groupname": "GROUP NAME", "call_starttime": "CALL START TIME", "call_endtime": "CALL END TIME", "direction": "DIRECTION", "call_status": "CALL STATUS", "duration_seconds": "DURATION SECONDS", "transcription_status": "TRANSCRIPTION STATUS", "language": "LANGUAGE", "num_speakers": "NUM SPEAKERS", "transcript_duration": "TRANSCRIPT DURATION", "transcript": "TRANSCRIPT", "quality_score": "QUALITY SCORE", "sentiment": "SENTIMENT", "call_purpose": "CALL PURPOSE", "objections_concerns": "OBJECTIONS CONCERNS", "objection_type": "OBJECTION TYPE", "talk_listen_ratio": "TALK LISTEN RATIO", "agent_speak_percentage": "AGENT SPEAK PERCENTAGE", "customer_speak_percentage": "CUSTOMER SPEAK PERCENTAGE", "dead_air_percentage": "DEAD AIR PERCENTAGE", "summary": "SUMMARY", "parameter_scores": "PARAMETER SCORES", } EXPORT_FIRST_COLUMNS = ("customer_number", "customer_name", "callid") QUALITY_EXPORT_HEADER_LABELS = {**EXPORT_HEADER_LABELS, "duration_seconds": "DURATION"} TRANSCRIPTION_EXPORT_HEADER_LABELS = {**EXPORT_HEADER_LABELS, "duration_seconds": "DURATION"} def _format_seconds_as_hhmmss(seconds): try: total = int(round(float(seconds))) except (TypeError, ValueError): return "" if total < 0: total = 0 hours = total // 3600 minutes = (total % 3600) // 60 secs = total % 60 return f"{hours:02d}:{minutes:02d}:{secs:02d}" def _export_ratio_mm_ss(left, right): return f'="{int(left)}:{int(right)}"' def _parse_speak_percentage(value): if value is None or str(value).strip() == "": return None try: return int(round(float(value))) except (TypeError, ValueError): return None def _ratio_from_talk_times(agent_talk_time, customer_talk_time): try: agent_seconds = max(0.0, float(agent_talk_time or 0)) customer_seconds = max(0.0, float(customer_talk_time or 0)) except (TypeError, ValueError): return None total = agent_seconds + customer_seconds if total <= 0: return None return int(round(100 * agent_seconds / total)), int(round(100 * customer_seconds / total)) def _parse_talk_listen_ratio_parts(value): import re from datetime import time, timedelta if value is None: return None if isinstance(value, timedelta): total = max(0, int(value.total_seconds())) hours, rem = divmod(total, 3600) minutes, seconds = divmod(rem, 60) if hours == 0: return minutes, seconds return hours * 60 + minutes, seconds if isinstance(value, time): if value.hour == 0: return value.minute, value.second return value.hour * 60 + value.minute, value.second if isinstance(value, (int, float)) and not isinstance(value, bool): numeric = float(value) if 0 < numeric < 1: total_seconds = int(round(numeric * 86400)) minutes, seconds = divmod(total_seconds, 60) hours, minutes = divmod(minutes, 60) if hours == 0: return minutes, seconds return hours * 60 + minutes, seconds return None text = str(value).strip() if not text or text.upper() in ("N/A", "NA"): return None normalized = re.sub(r"\s*:\s*", ":", text) parts = [p for p in normalized.split(":") if p != ""] if len(parts) < 2: return None def _to_int(part): return int(round(float(part))) if len(parts) >= 3: third = parts[2] if third in ("0", "00", "0.0", "0.00") or (third.replace(".", "", 1).isdigit() and int(float(third)) == 0): return _to_int(parts[0]), _to_int(parts[1]) if parts[0] in ("0", "00") and parts[1].replace(".", "", 1).isdigit() and parts[2].replace(".", "", 1).isdigit(): return _to_int(parts[1]), _to_int(parts[2]) return _to_int(parts[0]), _to_int(parts[1]) if len(parts) == 2 and all(p.replace(".", "", 1).isdigit() for p in parts): return _to_int(parts[0]), _to_int(parts[1]) return None def _quality_export_talk_listen_ratio(row): agent_pct = _parse_speak_percentage(row.get("agent_speak_percentage")) customer_pct = _parse_speak_percentage(row.get("customer_speak_percentage")) if agent_pct is not None and customer_pct is not None and (agent_pct > 0 or customer_pct > 0): return _export_ratio_mm_ss(agent_pct, customer_pct) from_talk_times = _ratio_from_talk_times(row.get("agent_talk_time"), row.get("customer_talk_time")) if from_talk_times: return _export_ratio_mm_ss(from_talk_times[0], from_talk_times[1]) parsed = _parse_talk_listen_ratio_parts(row.get("talk_listen_ratio")) if parsed: return _export_ratio_mm_ss(parsed[0], parsed[1]) raw = str(row.get("talk_listen_ratio") or "").strip() if raw.upper() in ("N/A", "NA"): return "NA" return raw def _prepare_transcription_export_rows(rows): prepared = [] for row in rows or []: new_row = dict(row) if "duration_seconds" in new_row: new_row["duration_seconds"] = _format_seconds_as_hhmmss(new_row.get("duration_seconds")) prepared.append(new_row) return prepared def _prepare_quality_export_rows(rows): prepared = [] for row in rows or []: new_row = dict(row) if "duration_seconds" in new_row: new_row["duration_seconds"] = _format_seconds_as_hhmmss(new_row.get("duration_seconds")) new_row["talk_listen_ratio"] = _quality_export_talk_listen_ratio(new_row) prepared.append(new_row) return prepared def _add_customer_export_fields(bid, rows): if not rows: return rows phones = [ str(row.get("customer_number") or row.get("customer_callinfo") or "").strip() for row in rows ] phones = [phone for phone in phones if phone] crm_by_phone = {} if phones: try: crm_by_phone = db_handler.get_crm_enrichment_for_phones(bid, "leadsquared", phones) or {} except Exception as exc: logger.warning("Customer name enrichment skipped for export %s: %s", bid, exc) enriched = [] for row in rows: new_row = dict(row) customer_number = str(new_row.pop("customer_callinfo", None) or new_row.get("customer_number") or "").strip() crm_info = crm_by_phone.get(customer_number) or {} new_row["customer_number"] = customer_number new_row["customer_name"] = new_row.get("customer_name") or crm_info.get("lead_name") or "" enriched.append(new_row) return enriched def _format_export_csv_rows(rows, header_labels=None): labels = header_labels or EXPORT_HEADER_LABELS formatted = [] for row in rows or []: ordered_keys = list(EXPORT_FIRST_COLUMNS) ordered_keys.extend(key for key in row.keys() if key not in EXPORT_FIRST_COLUMNS) formatted.append({ labels.get(key, str(key).replace("_", " ").upper()): row.get(key, "") for key in ordered_keys }) return formatted def _csv_download(rows, filename, header_labels=None): import csv from io import StringIO rows = _format_export_csv_rows(rows, header_labels=header_labels) output = StringIO() if rows: writer = csv.DictWriter(output, fieldnames=list(rows[0].keys()), extrasaction="ignore") writer.writeheader() writer.writerows(rows) return output.getvalue(), 200, { 'Content-Type': 'text/csv', 'Content-Disposition': f'attachment; filename={filename}' } @app.route('/export//transcriptions', methods=['GET']) @handle_errors def export_transcriptions(bid): raw_table = f"{bid}_raw_calls" sarvam_table = f"{bid}_sarvamresponse" analytics_table = f"{bid}_callanalytics" groupname = request.args.get('groupname') date_from = request.args.get('date_from') date_to = request.args.get('date_to') agent_names = request.args.get('agent_names') customer_name = request.args.get('customer_name') detailed_calls = request.args.get('detailed_calls') quality_min = request.args.get('quality_min', type=float) quality_max = request.args.get('quality_max', type=float) direction = request.args.get('direction') call_status = request.args.get('call_status') with db_handler.get_connection() as conn: cursor = conn.cursor() if not db_handler._table_exists(cursor, raw_table): rows = [] else: has_sarvam = db_handler._table_exists(cursor, sarvam_table) needs_quality_join = quality_min is not None or quality_max is not None has_analytics = db_handler._table_exists(cursor, analytics_table) if needs_quality_join and not has_analytics: return _csv_download([], f"transcriptions_{bid}_{datetime.now().strftime('%Y%m%d')}.csv") where_sql, params = _export_filters( groupname, date_from, date_to, agent_names=agent_names, customer_name=customer_name, detailed_calls=detailed_calls, quality_min=quality_min, quality_max=quality_max, direction=direction, call_status=call_status, include_quality_filters=needs_quality_join, ) sarvam_cols = db_handler.sarvam_export_select_exprs(cursor, sarvam_table) quality_join_sql = f"INNER JOIN `{analytics_table}` a ON r.callid = a.callid" if needs_quality_join else "" cursor.execute( f""" SELECT r.customer_callinfo AS customer_number, r.callid, r.agentname, r.groupname, r.call_starttime, r.call_endtime, r.direction, r.call_status, TIMESTAMPDIFF(SECOND, r.call_starttime, r.call_endtime) AS duration_seconds, r.transcription_status, {sarvam_cols["num_speakers"]} AS num_speakers, {sarvam_cols["transcript_duration"]} AS transcript_duration, {sarvam_cols["transcript"]} AS transcript, {sarvam_cols["raw_response"]} AS raw_response FROM `{raw_table}` r {quality_join_sql} {where_sql} ORDER BY r.call_starttime DESC LIMIT 10000 """, params, ) rows = db_handler.prepare_transcription_csv_export_rows(cursor.fetchall() or []) rows = _add_customer_export_fields(bid, rows) rows = _prepare_transcription_export_rows(rows) filename = f"transcriptions_{bid}_{datetime.now().strftime('%Y%m%d')}.csv" return _csv_download(rows, filename, header_labels=TRANSCRIPTION_EXPORT_HEADER_LABELS) @app.route('/export//quality', methods=['GET']) @handle_errors def export_quality(bid): raw_table = f"{bid}_raw_calls" analytics_table = f"{bid}_callanalytics" groupname = request.args.get('groupname') date_from = request.args.get('date_from') date_to = request.args.get('date_to') agent_names = request.args.get('agent_names') customer_name = request.args.get('customer_name') detailed_calls = request.args.get('detailed_calls') quality_min = request.args.get('quality_min', type=float) quality_max = request.args.get('quality_max', type=float) direction = request.args.get('direction') call_status = request.args.get('call_status') with db_handler.get_connection() as conn: cursor = conn.cursor() if not db_handler._table_exists(cursor, raw_table) or not db_handler._table_exists(cursor, analytics_table): rows = [] else: where_sql, params = _export_filters( groupname, date_from, date_to, agent_names=agent_names, customer_name=customer_name, detailed_calls=detailed_calls, quality_min=quality_min, quality_max=quality_max, direction=direction, call_status=call_status, include_quality_filters=True, ) cursor.execute( f""" SELECT r.customer_callinfo AS customer_number, r.callid, r.agentname, r.groupname, r.call_starttime, r.call_endtime, r.direction, r.call_status, TIMESTAMPDIFF(SECOND, r.call_starttime, r.call_endtime) AS duration_seconds, a.quality_score, a.sentiment, a.call_purpose, a.objections_concerns, a.objection_type, a.talk_listen_ratio, a.agent_talk_time, a.customer_talk_time, a.agent_speak_percentage, a.customer_speak_percentage, a.dead_air_percentage, a.summary, a.parameter_scores FROM `{raw_table}` r INNER JOIN `{analytics_table}` a ON r.callid = a.callid {where_sql} ORDER BY r.call_starttime DESC LIMIT 10000 """, params, ) rows = cursor.fetchall() or [] rows = _add_customer_export_fields(bid, rows) rows = _prepare_quality_export_rows(rows) filename = f"quality_{bid}_{datetime.now().strftime('%Y%m%d')}.csv" return _csv_download(rows, filename, header_labels=QUALITY_EXPORT_HEADER_LABELS) # ============================================================================ # WEBHOOK ENDPOINTS (for receiving updates from processing system) # ============================================================================ @app.route('/webhook/call-update', methods=['POST']) @handle_errors def receive_call_update(): """ Receive call updates from processing system Body: { "bid": "6840", "callid": "CALL-001", "status": 2, "data": {...} } """ data = request.get_json() bid = data.get('bid') callid = data.get('callid') if not bid or not callid: return jsonify({'error': 'bid and callid are required'}), 400 # Update call in database success = db_handler.update_call(bid, callid, data.get('data', {})) if success: # Best-effort auto-ingestion into RAG when call updates arrive. try: rag_handler.backfill_transcripts( bid=bid, presales_only=False, limit=1, overwrite_existing=True, callids=[callid], ) except Exception as rag_exc: logger.warning("RAG auto-ingest skipped for %s/%s: %s", bid, callid, rag_exc) return jsonify({'message': 'Call updated successfully'}), 200 else: return jsonify({'error': 'Failed to update call'}), 500 @app.route('/webhook/conversation-summary', methods=['POST']) @handle_errors def receive_conversation_summary(): """ Receive conversation summary Body: { "business_id": "6840", "callid": "CALL-001", "transfer_reason": {...} } """ data = request.get_json() bid = data.get('business_id') callid = data.get('callid') transfer_reason = data.get('transfer_reason') if not bid or not callid: return jsonify({'error': 'business_id and callid are required'}), 400 success = db_handler.save_conversation_summary(bid, callid, transfer_reason) if success: return jsonify({'message': 'Summary saved successfully'}), 200 else: return jsonify({'error': 'Failed to save summary'}), 500 # ============================================================================ # CALL SYNC ENDPOINTS # ============================================================================ CALL_SYNC_CACHE_TTL_SECONDS = int(os.getenv('CALL_SYNC_CACHE_TTL_SECONDS', '3600')) CALL_SYNC_CACHE_SCHEMA_VERSION = "v5" def _get_call_sync_redis(): redis_url = os.getenv('REDIS_URL') or app.config.get('RAG_REDIS_URL') or 'redis://127.0.0.1:6379/0' try: client = redis.from_url( redis_url, decode_responses=True, socket_timeout=5, socket_connect_timeout=5 ) client.ping() return client except Exception as e: logger.warning("Call sync Redis unavailable: %s", str(e)) return None def _sync_auth_user(bid): auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return None, (jsonify({'error': 'Missing authorization header'}), 401) token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user: return None, (jsonify({'error': 'Invalid or expired token'}), 401) if not user_has_business_admin(user, bid): return None, (jsonify({'error': 'Business admin access required'}), 403) return user, None def _resolve_source_call_table(cursor, bid, table_kind): if table_kind == 'history': candidates = [f'{bid}_callhistory', f'{bid}_call_history'] else: candidates = [f'{bid}_callarchive', f'{bid}_call_archive'] for candidate in candidates: cursor.execute("SHOW TABLES LIKE %s", (candidate,)) if cursor.fetchone(): return candidate return None def _table_has_column(cursor, table_name, column_name): cursor.execute(f"SHOW COLUMNS FROM `{table_name}` LIKE %s", (column_name,)) return cursor.fetchone() is not None def _serialize_call_sync_row(row): serialized = {} for key, value in row.items(): if isinstance(value, datetime): serialized[key] = value.isoformat() else: serialized[key] = value return serialized def _fetch_source_calls(bid, table_kind, date_from=None, date_to=None, limit=500): source_config = get_sync_source_db_config() conn = pymysql.connect(**source_config) try: cursor = conn.cursor(pymysql.cursors.DictCursor) table_name = _resolve_source_call_table(cursor, bid, table_kind) if not table_name: return [], None has_direction = _table_has_column(cursor, table_name, 'direction') has_customer_callinfo = _table_has_column(cursor, table_name, 'customer_callinfo') has_callfrom = _table_has_column(cursor, table_name, 'callfrom') has_callto = _table_has_column(cursor, table_name, 'callto') has_clicktocall = _table_has_column(cursor, table_name, 'clicktocalldid') customer_candidates = [] if has_customer_callinfo: customer_candidates.append("NULLIF(TRIM(CAST(customer_callinfo AS CHAR)), '')") # Direction-aware lead extraction: # Outbound -> customer is usually callto # Inbound -> customer is usually callfrom if has_direction and has_callto and has_callfrom: customer_candidates.append( "CASE " "WHEN LOWER(TRIM(CAST(direction AS CHAR))) = 'outbound' THEN NULLIF(TRIM(CAST(callto AS CHAR)), '') " "WHEN LOWER(TRIM(CAST(direction AS CHAR))) = 'inbound' THEN NULLIF(TRIM(CAST(callfrom AS CHAR)), '') " "ELSE NULL END" ) # Keep non-directional fallbacks conservative to avoid mapping agent numbers # into customer_callinfo. `callfrom` is often the agent leg on outbound calls. if has_callto: customer_candidates.append("NULLIF(TRIM(CAST(callto AS CHAR)), '')") if has_clicktocall: customer_candidates.append("NULLIF(TRIM(CAST(clicktocalldid AS CHAR)), '')") customer_expr = f"COALESCE({', '.join(customer_candidates)})" if customer_candidates else "NULL" query = f""" SELECT callid, %s as bid, agentname, groupname, starttime, endtime, dialstatus, direction, filename, {customer_expr} as customer_callinfo, countrycode, emp_phone, clicktocalldid FROM `{table_name}` WHERE 1 = 1 """ params = [str(bid)] if date_from and date_to: query += " AND DATE(starttime) BETWEEN %s AND %s" params.extend([date_from, date_to]) query += " ORDER BY starttime DESC LIMIT %s" params.append(int(limit)) cursor.execute(query, tuple(params)) rows = cursor.fetchall() or [] return [_serialize_call_sync_row(row) for row in rows], table_name finally: conn.close() def _fetch_source_count(bid, table_kind, date_from=None, date_to=None): source_config = get_sync_source_db_config() conn = pymysql.connect(**source_config) try: cursor = conn.cursor(pymysql.cursors.DictCursor) table_name = _resolve_source_call_table(cursor, bid, table_kind) if not table_name: return 0 query = f""" SELECT COUNT(*) as total FROM `{table_name}` WHERE 1 = 1 """ params = [] if date_from and date_to: query += " AND DATE(starttime) BETWEEN %s AND %s" params.extend([date_from, date_to]) cursor.execute(query, tuple(params)) row = cursor.fetchone() or {} return int(row.get('total') or 0) finally: conn.close() @app.route('/sync/cache//history', methods=['GET']) @handle_errors def get_cached_history_calls(bid): _, auth_error = _sync_auth_user(bid) if auth_error: return auth_error limit = request.args.get('limit', 300, type=int) refresh = request.args.get('refresh', '0') in ['1', 'true', 'True'] cache_key = f"callsync:{CALL_SYNC_CACHE_SCHEMA_VERSION}:{bid}:history:limit:{limit}" redis_client = _get_call_sync_redis() if redis_client and not refresh: cached = redis_client.get(cache_key) if cached: payload = json.loads(cached) payload['cache_hit'] = True return jsonify(payload), 200 records, table_name = _fetch_source_calls(bid, 'history', limit=limit) if not table_name: return jsonify({ 'source': 'call_history', 'table': None, 'count': 0, 'records': [], 'cached_at': datetime.now().isoformat(), 'cache_hit': False, 'message': 'Source call history table not found for this business', 'expected_tables': [f'{bid}_call_history', f'{bid}_callhistory'], 'configured_source_db': { 'host': get_sync_source_db_config().get('host'), 'database': get_sync_source_db_config().get('database') } }), 200 payload = { 'source': 'call_history', 'table': table_name, 'count': len(records), 'records': records, 'cached_at': datetime.now().isoformat(), 'cache_hit': False } if redis_client: redis_client.setex(cache_key, CALL_SYNC_CACHE_TTL_SECONDS, json.dumps(payload, ensure_ascii=True)) return jsonify(payload), 200 @app.route('/sync/cache//archive', methods=['GET']) @handle_errors def get_cached_archive_calls(bid): _, auth_error = _sync_auth_user(bid) if auth_error: return auth_error date_from = request.args.get('date_from') date_to = request.args.get('date_to') limit = request.args.get('limit', 500, type=int) refresh = request.args.get('refresh', '0') in ['1', 'true', 'True'] range_key = f"{date_from}:{date_to}" if date_from and date_to else "recent" cache_key = f"callsync:{CALL_SYNC_CACHE_SCHEMA_VERSION}:{bid}:archive:{range_key}:limit:{limit}" redis_client = _get_call_sync_redis() if redis_client and not refresh: cached = redis_client.get(cache_key) if cached: payload = json.loads(cached) payload['cache_hit'] = True return jsonify(payload), 200 records, table_name = _fetch_source_calls( bid, 'archive', date_from=date_from, date_to=date_to, limit=limit ) if not table_name: return jsonify({ 'source': 'call_archive', 'table': None, 'count': 0, 'records': [], 'date_from': date_from, 'date_to': date_to, 'cached_at': datetime.now().isoformat(), 'cache_hit': False, 'message': 'Source call archive table not found for this business', 'expected_tables': [f'{bid}_callarchive', f'{bid}_call_archive'], 'configured_source_db': { 'host': get_sync_source_db_config().get('host'), 'database': get_sync_source_db_config().get('database') } }), 200 payload = { 'source': 'call_archive', 'table': table_name, 'count': len(records), 'records': records, 'date_from': date_from, 'date_to': date_to, 'cached_at': datetime.now().isoformat(), 'cache_hit': False } if redis_client: redis_client.setex(cache_key, CALL_SYNC_CACHE_TTL_SECONDS, json.dumps(payload, ensure_ascii=True)) return jsonify(payload), 200 @app.route('/sync/check-count/', methods=['GET']) @handle_errors def check_call_count(bid): _, auth_error = _sync_auth_user(bid) if auth_error: return auth_error date_from = request.args.get('date_from') date_to = request.args.get('date_to') try: history_total = _fetch_source_count(bid, 'history') archive_total = _fetch_source_count(bid, 'archive', date_from=date_from, date_to=date_to) if date_from and date_to else 0 return jsonify({ 'callhistory': {'total': history_total}, 'callarchive': {'total': archive_total}, 'total_count': int(history_total + archive_total), 'date_from': date_from, 'date_to': date_to }), 200 except Exception as e: logger.error(f"Error checking call count: {str(e)}") return jsonify({'error': f'Failed to check call count: {str(e)}'}), 500 @app.route('/sync/calls/', methods=['POST']) @handle_errors def sync_calls(bid): _, auth_error = _sync_auth_user(bid) if auth_error: return auth_error data = request.get_json() or {} date_from = data.get('date_from') date_to = data.get('date_to') try: redis_client = _get_call_sync_redis() limit = int(data.get('limit', 500)) if date_from and date_to: records, table_name = _fetch_source_calls( bid, 'archive', date_from=date_from, date_to=date_to, limit=limit ) source = 'call_archive' cache_key = f"callsync:{CALL_SYNC_CACHE_SCHEMA_VERSION}:{bid}:archive:{date_from}:{date_to}:limit:{limit}" else: records, table_name = _fetch_source_calls( bid, 'history', limit=limit ) source = 'call_history' cache_key = f"callsync:{CALL_SYNC_CACHE_SCHEMA_VERSION}:{bid}:history:limit:{limit}" payload = { 'source': source, 'table': table_name, 'count': len(records), 'records': records, 'date_from': date_from, 'date_to': date_to, 'cached_at': datetime.now().isoformat() } if not table_name: return jsonify({ 'message': f'Source table not found for {source}', 'cached_count': 0, 'source': source, 'table': None, 'cache_key': cache_key, 'stored_in': 'redis' }), 200 if redis_client: redis_client.setex(cache_key, CALL_SYNC_CACHE_TTL_SECONDS, json.dumps(payload, ensure_ascii=True)) return jsonify({ 'message': f'Successfully cached {len(records)} calls from {source}', 'cached_count': len(records), 'source': source, 'table': table_name, 'cache_key': cache_key, 'stored_in': 'redis' }), 200 except Exception as e: logger.error(f"Error caching calls: {str(e)}") return jsonify({'error': f'Failed to cache calls: {str(e)}'}), 500 @app.route('/sync/upload/', methods=['POST']) @handle_errors def upload_raw_calls(bid): """Upload raw calls Excel file into {bid}_raw_calls (admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user: return jsonify({'error': 'Invalid or expired token'}), 401 if not user_has_business_admin(user, bid): return jsonify({'error': 'Business admin access required'}), 403 upload_file = request.files.get('file') if not upload_file: return jsonify({'error': 'Excel file is required'}), 400 result = db_handler.import_raw_calls_from_excel(bid, upload_file) return jsonify(result), 200 @app.route('/transcription/calls/', methods=['GET']) @handle_errors def get_transcription_calls(bid): """ Get calls from raw_calls table filtered by status for transcription management Query params: - status: Filter by status (0, 1, 2, or 'all') """ status = request.args.get('status', '0') dest_config = get_primary_db_config() try: conn = pymysql.connect(**dest_config) cursor = conn.cursor(pymysql.cursors.DictCursor) if status == 'all': query = f""" SELECT callid, bid, agentname, groupname, call_starttime, call_endtime, call_status, status, transcription_status, transcription_requested, selected_for_processing FROM {bid}_raw_calls WHERE call_status = 'ANSWER' ORDER BY call_starttime DESC LIMIT 500 """ cursor.execute(query) else: query = f""" SELECT callid, bid, agentname, groupname, call_starttime, call_endtime, call_status, status, transcription_status, transcription_requested, selected_for_processing FROM {bid}_raw_calls WHERE call_status = 'ANSWER' AND status = %s ORDER BY call_starttime DESC LIMIT 500 """ cursor.execute(query, (status,)) calls = cursor.fetchall() cursor.close() conn.close() return jsonify({ 'calls': calls, 'count': len(calls) }), 200 except Exception as e: logger.error(f"Error getting transcription calls: {str(e)}") return jsonify({'error': f'Failed to get calls: {str(e)}'}), 500 @app.route('/transcription/trigger/', methods=['POST']) @handle_errors def trigger_batch_transcription(bid): """ Trigger transcription for selected calls Body: { "callids": ["call1", "call2", ...] } """ data = request.get_json() callids = data.get('callids', []) if not callids: return jsonify({'error': 'No call IDs provided'}), 400 dest_config = get_primary_db_config() try: conn = pymysql.connect(**dest_config) cursor = conn.cursor() # Update calls to mark them for transcription placeholders = ','.join(['%s'] * len(callids)) query = f""" UPDATE {bid}_raw_calls SET transcription_requested = 1, transcription_status = 'pending', selected_for_processing = 1 WHERE callid IN ({placeholders}) """ cursor.execute(query, callids) conn.commit() affected_rows = cursor.rowcount cursor.close() conn.close() return jsonify({ 'message': f'Successfully queued {affected_rows} calls for transcription', 'queued_count': affected_rows }), 200 except Exception as e: logger.error(f"Error triggering transcription: {str(e)}") return jsonify({'error': f'Failed to trigger transcription: {str(e)}'}), 500 @app.route('/analysis/trigger/', methods=['POST']) @handle_errors def trigger_batch_analysis(bid): """ Trigger AI analysis for selected answered calls with transcripts Body: { "callids": ["call1", "call2", ...] } """ data = request.get_json() callids = data.get('callids', []) if not callids: return jsonify({'error': 'No call IDs provided'}), 400 try: # Import analyzer (lazy import to avoid circular dependencies) from analyze_calls_with_parameters import CallAnalyzer analyzer = CallAnalyzer(app.config) success_count = 0 error_count = 0 errors = [] for callid in callids: try: # Get call data with transcript and speaker segments call_data = db_handler.get_raw_call_details(bid, callid) if not call_data: errors.append(f"{callid}: Call not found") error_count += 1 continue if call_data.get('call_status') != 'ANSWER': errors.append(f"{callid}: Call not answered") error_count += 1 continue transcript = call_data.get('transcripts') if not transcript: errors.append(f"{callid}: No transcript available") error_count += 1 continue # Parse speaker segments if they're JSON string speaker_segments = call_data.get('speaker_segments') if speaker_segments and isinstance(speaker_segments, str): speaker_segments = json.loads(speaker_segments) # Get actual duration actual_duration = call_data.get('duration') or call_data.get('duration_seconds') # Run analysis logger.info(f"Analyzing call {callid} for BID {bid}") analysis_result = analyzer.analyze_call( bid=bid, callid=callid, transcript=transcript, speaker_segments=speaker_segments or [], actual_duration=actual_duration ) success_count += 1 logger.info(f"Successfully analyzed call {callid}") except Exception as e: logger.error(f"Error analyzing call {callid}: {str(e)}") errors.append(f"{callid}: {str(e)}") error_count += 1 continue response_data = { 'message': f'Analysis completed: {success_count} successful, {error_count} failed', 'success_count': success_count, 'error_count': error_count } if errors: response_data['errors'] = errors[:10] # Limit to first 10 errors return jsonify(response_data), 200 except Exception as e: logger.error(f"Error in batch analysis: {str(e)}") return jsonify({'error': f'Failed to trigger analysis: {str(e)}'}), 500 # ============================================================================ # QUALITY PARAMETERS # ============================================================================ @app.route('/quality-parameters/', methods=['GET']) @handle_errors def get_quality_parameters(bid): """Get all quality parameters for a business""" parameters = quality_params_handler.get_parameters(bid) return jsonify(parameters), 200 @app.route('/quality-parameters/', methods=['POST']) @handle_errors def save_quality_parameter(bid): """Save or update a quality parameter""" data = request.get_json() if not data: return jsonify({'error': 'No data provided'}), 400 parameter_id = quality_params_handler.save_parameter(bid, data) return jsonify({ 'message': 'Parameter saved successfully', 'parameter_id': parameter_id }), 200 @app.route('/quality-parameters//', methods=['GET']) @handle_errors def get_quality_parameter(bid, param_id): """Get a specific quality parameter""" parameter = quality_params_handler.get_parameter_by_id(bid, param_id) if not parameter: return jsonify({'error': 'Parameter not found'}), 404 return jsonify(parameter), 200 @app.route('/quality-parameters//', methods=['DELETE']) @handle_errors def delete_quality_parameter(bid, param_id): """Delete a quality parameter""" success = quality_params_handler.delete_parameter(bid, param_id) if not success: return jsonify({'error': 'Parameter not found or could not be deleted'}), 404 return jsonify({'message': 'Parameter deleted successfully'}), 200 @app.route('/quality-parameters//groups', methods=['GET']) @handle_errors def get_parameter_groups(bid): """Get list of all parameter groups for a business""" groups = quality_params_handler.get_parameter_groups(bid) return jsonify(groups), 200 @app.route('/quality-parameters//total-score', methods=['GET']) @handle_errors def get_total_possible_score(bid): """Get total possible score for all parameters""" total_score = quality_params_handler.calculate_total_possible_score(bid) return jsonify({'total_score': total_score}), 200 # ============================================================================ # OBJECTION CLASSIFICATIONS ENDPOINTS # ============================================================================ @app.route('/objection-classifications/', methods=['GET']) @handle_errors def get_objection_classifications(bid): """ Get all objection classifications for a business Query params: - business_type: Filter by business type (optional) - is_active: Filter by active status (optional) """ business_type = request.args.get('business_type') is_active = request.args.get('is_active') if is_active is not None: is_active = is_active.lower() == 'true' classifications = objection_handler.get_all_classifications(bid, business_type, is_active) return jsonify(classifications), 200 @app.route('/objection-classifications//', methods=['GET']) @handle_errors def get_objection_classification_by_id(bid, classification_id): """Get a specific objection classification by ID""" classification = objection_handler.get_classification_by_id(bid, classification_id) if not classification: return jsonify({'error': 'Classification not found'}), 404 return jsonify(classification), 200 @app.route('/objection-classifications/', methods=['POST']) @handle_errors def create_objection_classification(bid): """Create a new objection classification""" data = request.get_json() if not data: return jsonify({'error': 'No data provided'}), 400 if not data.get('category_name'): return jsonify({'error': 'category_name is required'}), 400 created_by = data.get('created_by', 'admin') classification_id = objection_handler.create_classification(bid, data, created_by) return jsonify({ 'id': classification_id, 'message': 'Classification created successfully' }), 201 @app.route('/objection-classifications//', methods=['PUT']) @handle_errors def update_objection_classification(bid, classification_id): """Update an existing objection classification""" data = request.get_json() if not data: return jsonify({'error': 'No data provided'}), 400 updated_by = data.get('updated_by', 'admin') success = objection_handler.update_classification(bid, classification_id, data, updated_by) if not success: return jsonify({'error': 'Classification not found or no changes made'}), 404 return jsonify({ 'message': 'Classification updated successfully' }), 200 @app.route('/objection-classifications//', methods=['DELETE']) @handle_errors def delete_objection_classification(bid, classification_id): """Delete an objection classification""" success = objection_handler.delete_classification(bid, classification_id) if not success: return jsonify({'error': 'Classification not found'}), 404 return jsonify({ 'message': 'Classification deleted successfully' }), 200 @app.route('/objection-classifications///toggle', methods=['POST']) @handle_errors def toggle_objection_classification(bid, classification_id): """Toggle the active status of a classification""" success = objection_handler.toggle_active_status(bid, classification_id) if not success: return jsonify({'error': 'Classification not found'}), 404 return jsonify({ 'message': 'Classification status toggled successfully' }), 200 @app.route('/objection-classifications//search', methods=['GET']) @handle_errors def search_objection_classifications(bid): """ Search objection classifications Query params: - q: Search term (required) """ search_term = request.args.get('q') if not search_term: return jsonify({'error': 'Search term (q) is required'}), 400 classifications = objection_handler.search_classifications(bid, search_term) return jsonify(classifications), 200 @app.route('/objection-classifications//by-severity/', methods=['GET']) @handle_errors def get_classifications_by_severity(bid, severity): """Get all classifications of a specific severity level""" if severity not in ['low', 'medium', 'high', 'critical']: return jsonify({'error': 'Invalid severity level'}), 400 classifications = objection_handler.get_classifications_by_severity(bid, severity) return jsonify(classifications), 200 @app.route('/objection-classifications//classify', methods=['POST']) @handle_errors def classify_objection_text(bid): """ Auto-classify an objection text based on keywords Body: { "objection_text": "The text to classify" } """ data = request.get_json() if not data or not data.get('objection_text'): return jsonify({'error': 'objection_text is required'}), 400 result = objection_handler.classify_objection(bid, data['objection_text']) return jsonify(result), 200 @app.route('/objection-classifications//statistics', methods=['GET']) @handle_errors def get_objection_statistics(bid): """Get statistics about objection classifications for a business""" stats = objection_handler.get_statistics(bid) return jsonify(stats), 200 # ============================================================================ # RAG API # ============================================================================ @app.route('/rag//documents', methods=['POST']) @handle_errors @require_auth @require_business_access def rag_ingest_documents(bid): """ Upsert RAG documents and chunks for a business. Body: { "documents": [ { "source_id": "doc-1", "title": "FAQ", "source_type": "kb", "source_uri": "https://...", "metadata": {}, "chunks": [ {"chunk_id": "doc-1-1", "text": "...", "embedding": [0.1, ...], "metadata": {}} ] } ] } """ data = request.get_json(silent=True) or {} documents = data.get('documents', []) if not documents: return jsonify({'error': 'documents array is required'}), 400 result = rag_handler.ingest_documents(bid, documents) return jsonify({ 'status': 'success', 'ingestion': result }), 200 @app.route('/rag//ingest-transcripts', methods=['POST']) @handle_errors @require_auth @require_business_access def rag_ingest_transcripts(bid): """ Backfill transcripts from DB into RAG chunks/documents. Body: { "presales_only": true, "limit": 1000, "overwrite_existing": false, "callids": ["optional", "callid-list"] } """ data = request.get_json(silent=True) or {} result = rag_handler.backfill_transcripts( bid=bid, presales_only=bool(data.get('presales_only', True)), limit=int(data.get('limit', 1000)), overwrite_existing=bool(data.get('overwrite_existing', False)), callids=data.get('callids'), ) return jsonify(result), 200 @app.route('/rag//query', methods=['POST']) @handle_errors @require_auth @require_business_access def rag_query(bid): """ Query RAG with memory and customer profile updates. Body: { "user_id": "cust-123", "message": "user question", "query_embedding": [optional vector], "conversation_id": "optional", "top_k": 8, "min_similarity": 0.2, "metadata": {}, "profile_updates": {"traits": {...}} } """ data = request.get_json(silent=True) or {} user_id = data.get('user_id') message = data.get('message') if not user_id: return jsonify({'error': 'user_id is required'}), 400 if not message: return jsonify({'error': 'message is required'}), 400 result = rag_handler.query( bid=bid, user_id=user_id, message=message, query_embedding=data.get('query_embedding'), conversation_id=data.get('conversation_id'), top_k=data.get('top_k'), min_similarity=data.get('min_similarity'), metadata=data.get('metadata'), profile_updates=data.get('profile_updates'), ) return jsonify(result), 200 @app.route('/rag//conversations/', methods=['GET']) @handle_errors @require_auth @require_business_access def rag_get_conversation(bid, conversation_id): """Get conversation history.""" limit = request.args.get('limit', default=200, type=int) messages = rag_handler.get_conversation_messages(bid, conversation_id, limit=limit) return jsonify({ 'conversation_id': conversation_id, 'messages': messages }), 200 @app.route('/rag//profiles/', methods=['GET']) @handle_errors @require_auth @require_business_access def rag_get_profile(bid, user_id): """Get current user profile learned from prior chats.""" profile = rag_handler.get_user_profile(bid, user_id) return jsonify(profile), 200 # ============================================================================ # CRM INTEGRATIONS (LeadSquared) # ============================================================================ @app.route('/crm//leadsquared/presales-mapping', methods=['GET']) @handle_errors @require_auth @require_business_access def leadsquared_presales_mapping(bid): row_count = request.args.get('lsq_row_count', 250, type=int) groupname = request.args.get('groupname') force_refresh = _as_bool(request.args.get('force_refresh'), default=False) mapping = _get_presales_mapping_from_leadsquared( bid=bid, groupname=groupname, row_count=row_count, force_refresh=force_refresh ) status = 200 if mapping.get('success') else 400 return jsonify(mapping), status @app.route('/crm//leadsquared/integration', methods=['GET']) @handle_errors @require_auth @require_business_access def get_leadsquared_integration(bid): integration = db_handler.get_crm_integration(bid, 'leadsquared') if not integration: return jsonify({ 'success': False, 'message': 'LeadSquared integration not found', 'data': None }), 404 return jsonify({'success': True, 'data': integration}), 200 @app.route('/crm//leadsquared/integration', methods=['POST']) @handle_errors @require_auth @require_business_access def save_leadsquared_integration(bid): admin_err = _require_business_admin_or_master(bid) if admin_err: return admin_err data = request.get_json(silent=True) or {} access_key = str(data.get('lsq_access_key', '')).strip() secret_key = str(data.get('lsq_secret_key', '')).strip() api_host = str(data.get('lsq_api_host', '')).strip() is_active = bool(data.get('is_active', True)) existing_integration = db_handler.get_crm_integration(bid, 'leadsquared') existing_creds = db_handler.get_crm_credentials(bid, 'leadsquared') if existing_integration else {} if not access_key: access_key = existing_creds.get('access_key', '') if not secret_key: secret_key = existing_creds.get('secret_key', '') if not access_key or not secret_key: return jsonify({'success': False, 'message': 'lsq_access_key and lsq_secret_key are required'}), 400 if not api_host: api_host = (existing_integration or {}).get('api_host') or 'https://api-in21.leadsquared.com/v2/' db_handler.upsert_crm_integration( bid=bid, provider='leadsquared', access_key=access_key, secret_key=secret_key, api_host=api_host, is_active=is_active, config=data.get('config') or {}, ) return jsonify({'success': True, 'message': 'LeadSquared integration saved successfully'}), 200 @app.route('/crm//leadsquared/integration/push-activity', methods=['POST']) @handle_errors @require_auth @require_business_access def push_leadsquared_activity(bid): if not _crm_write_allowed(): return jsonify({'success': False, 'message': 'CRM write operations are disabled'}), 403 payload = request.get_json(silent=True) or {} if not payload: return jsonify({'success': False, 'message': 'Activity payload is required'}), 400 creds = db_handler.get_crm_credentials(bid, 'leadsquared') if not creds or not creds.get('access_key') or not creds.get('secret_key'): return jsonify({'success': False, 'message': 'LeadSquared integration not configured or inactive'}), 400 service = LeadSquaredService( access_key=creds['access_key'], secret_key=creds['secret_key'], api_host=creds.get('api_host'), ) result = service.create_activity(payload) status_code = 200 if result.get('success') else 400 return jsonify(result), status_code @app.route('/crm//leadsquared/integration/test', methods=['POST']) @handle_errors @require_auth @require_business_access def test_leadsquared_integration(bid): creds = db_handler.get_crm_credentials(bid, 'leadsquared') if not creds or not creds.get('access_key') or not creds.get('secret_key'): return jsonify({'success': False, 'message': 'LeadSquared integration not configured'}), 404 service = LeadSquaredService( access_key=creds['access_key'], secret_key=creds['secret_key'], api_host=creds.get('api_host'), ) result = service.test_connection() if result.get('success'): db_handler.mark_crm_integration_tested(bid, 'leadsquared') return jsonify(result), 200 return jsonify(result), 400 @app.route('/crm//leadsquared/integration', methods=['DELETE']) @handle_errors @require_auth @require_business_access def delete_leadsquared_integration(bid): admin_err = _require_business_admin_or_master(bid) if admin_err: return admin_err removed = db_handler.delete_crm_integration(bid, 'leadsquared') return jsonify({ 'success': True, 'message': 'LeadSquared integration deleted successfully' if removed else 'LeadSquared integration did not exist' }), 200 def _get_lsq_service_for_bid_or_error(bid): creds = db_handler.get_crm_credentials(bid, 'leadsquared') if not creds or not creds.get('access_key') or not creds.get('secret_key') or not creds.get('is_active'): return None, (jsonify({'success': False, 'message': 'LeadSquared integration not configured or inactive'}), 400) service = LeadSquaredService( access_key=creds['access_key'], secret_key=creds['secret_key'], api_host=creds.get('api_host'), ) return service, None def _extract_lsq_lead_record(payload): if payload is None: return None if isinstance(payload, list): return payload[0] if payload else None if isinstance(payload, dict): for key in ('List', 'Data', 'list', 'data'): value = payload.get(key) if isinstance(value, list) and value: return value[0] return payload return None def _lsq_field(record, *keys): if not isinstance(record, dict): return None for key in keys: if key in record and record.get(key) not in (None, ''): return record.get(key) for key in keys: for record_key, value in record.items(): if str(record_key).lower() == str(key).lower() and value not in (None, ''): return value return None def _crm_write_allowed(): return str(os.getenv('ALLOW_CRM_WRITE', 'false')).strip().lower() == 'true' @app.route('/crm//leadsquared/leads/search', methods=['POST']) @handle_errors @require_auth @require_business_access def leadsquared_search_leads(bid): service, err = _get_lsq_service_for_bid_or_error(bid) if err: return err payload = request.get_json(silent=True) or {} return jsonify(service.search_leads(payload)), 200 @app.route('/crm//leadsquared/leads/', methods=['GET']) @handle_errors @require_auth @require_business_access def leadsquared_get_lead(bid, lead_id): service, err = _get_lsq_service_for_bid_or_error(bid) if err: return err result = service.get_lead(lead_id) return jsonify(result), (200 if result.get('success') else 400) @app.route('/crm//leadsquared/leads', methods=['POST']) @handle_errors @require_auth @require_business_access def leadsquared_create_lead(bid): if not _crm_write_allowed(): return jsonify({'success': False, 'message': 'CRM write operations are disabled'}), 403 service, err = _get_lsq_service_for_bid_or_error(bid) if err: return err payload = request.get_json(silent=True) or {} result = service.create_lead(payload) return jsonify(result), (200 if result.get('success') else 400) @app.route('/crm//leadsquared/leads/', methods=['PUT']) @handle_errors @require_auth @require_business_access def leadsquared_update_lead(bid, lead_id): if not _crm_write_allowed(): return jsonify({'success': False, 'message': 'CRM write operations are disabled'}), 403 service, err = _get_lsq_service_for_bid_or_error(bid) if err: return err payload = request.get_json(silent=True) or {} result = service.update_lead(lead_id, payload) return jsonify(result), (200 if result.get('success') else 400) @app.route('/crm//leadsquared/leads/', methods=['DELETE']) @handle_errors @require_auth @require_business_access def leadsquared_delete_lead(bid, lead_id): if not _crm_write_allowed(): return jsonify({'success': False, 'message': 'CRM write operations are disabled'}), 403 service, err = _get_lsq_service_for_bid_or_error(bid) if err: return err result = service.delete_lead(lead_id) return jsonify(result), (200 if result.get('success') else 400) # ============================================================================ # ERROR HANDLERS # ============================================================================ @app.errorhandler(404) def not_found(error): return jsonify({'error': 'Endpoint not found'}), 404 @app.errorhandler(405) def method_not_allowed(error): return jsonify({'error': 'Method not allowed'}), 405 @app.errorhandler(500) def internal_error(error): return jsonify({'error': 'Internal server error'}), 500 # ============================================================================ # MAIN # ============================================================================ # Audio Proxy Endpoint # ============================================================================ @app.route('/audio/proxy//', methods=['GET']) @handle_errors def proxy_audio(bid, callid): """Proxy audio file from Google Drive to bypass CORS restrictions""" import requests from flask import Response, stream_with_context try: # Get call details to retrieve file URL call = db_handler.get_call_by_id(bid, callid) if not call or not call.get('fileUrl'): return jsonify({'error': 'Audio file not found'}), 404 file_url = call['fileUrl'] # Stream the file from Google Drive response = requests.get(file_url, stream=True, timeout=60) if response.status_code != 200: logger.error(f"Failed to fetch audio from {file_url}: {response.status_code}") return jsonify({'error': 'Failed to fetch audio file'}), 502 # Determine content type content_type = response.headers.get('Content-Type', 'audio/mpeg') # Stream the response back to the client def generate(): for chunk in response.iter_content(chunk_size=8192): if chunk: yield chunk return Response( stream_with_context(generate()), content_type=content_type, headers={ 'Accept-Ranges': 'bytes', 'Cache-Control': 'public, max-age=3600' } ) except Exception as e: logger.error(f"Error proxying audio for {callid}: {str(e)}") return jsonify({'error': 'Internal server error'}), 500 # ============================================================================ # MASTER ADMIN ENDPOINTS # ============================================================================ @app.route('/admin/users', methods=['GET']) @handle_errors def admin_get_all_users(): """Get all users (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 users, status_code = auth_handler.get_all_users() return jsonify({'users': users}), status_code @app.route('/admin/users/create', methods=['POST']) @handle_errors def admin_create_user(): """Create a new user (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 data = request.json required_fields = ['username', 'email', 'password'] for field in required_fields: if not data.get(field): return jsonify({'error': f'{field} is required'}), 400 result, status_code = auth_handler.create_user( username=data['username'], email=data['email'], password=data['password'], full_name=data.get('full_name'), role=data.get('role', 'user'), is_master=data.get('is_master', False) ) # If user created successfully, assign business access if provided if status_code == 201: user_id = result['id'] # Assign business access if data.get('businesses'): default_role = data.get('role', 'user') for business in data['businesses']: bid = business.get('bid') if isinstance(business, dict) else business if isinstance(business, dict): role = business.get('role') if role is None: role = default_role else: role = default_role auth_handler.assign_business_access(user_id, bid, role) # Log activity auth_handler.log_activity( user_id=user.get('id'), username=user.get('username'), activity_type='create_user', description=f"Created new user: {data['username']}", ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) return jsonify(result), status_code @app.route('/admin/businesses', methods=['GET']) @handle_errors def admin_get_all_businesses(): """Get all businesses""" businesses, status_code = auth_handler.get_all_businesses() return jsonify({'businesses': businesses}), status_code @app.route('/admin/businesses/create', methods=['POST']) @handle_errors def admin_create_business(): """Create a new business (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 upload_file = None if request.content_type and request.content_type.startswith('multipart/form-data'): data = request.form upload_file = request.files.get('raw_calls_file') else: data = request.get_json() data = data or {} if not data.get('bid') or not data.get('name'): return jsonify({'error': 'bid and name are required'}), 400 result, status_code = auth_handler.create_business( bid=data['bid'], name=data['name'], description=data.get('description') ) # Log activity if successful if status_code == 201: auth_handler.log_activity( user_id=user.get('id'), username=user.get('username'), activity_type='create_business', description=f"Created new business: {data['name']} (ID: {data['bid']})", ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) if upload_file and upload_file.filename: try: upload_info = db_handler.import_raw_calls_from_excel( data['bid'], upload_file ) result['upload'] = upload_info except Exception as e: logger.error(f"Error importing raw calls: {str(e)}") return jsonify({ 'error': str(e), 'business_created': True }), 500 return jsonify(result), status_code @app.route('/admin/users//businesses', methods=['POST']) @handle_errors def admin_assign_business_access(user_id): """Assign business access to a user (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 data = request.json if not data.get('bid'): return jsonify({'error': 'bid is required'}), 400 result, status_code = auth_handler.assign_business_access( user_id=user_id, bid=data['bid'], role=data.get('role', 'user') ) return jsonify(result), status_code @app.route('/admin/activity-log', methods=['GET']) @handle_errors def admin_get_activity_log(): """Get activity log (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 # Get query parameters limit = request.args.get('limit', 100, type=int) offset = request.args.get('offset', 0, type=int) user_id = request.args.get('user_id', type=int) activity_type = request.args.get('activity_type') action_code = request.args.get('action_code') bid = request.args.get('bid') username_contains = request.args.get('username_contains') description_contains = request.args.get('description_contains') date_from = request.args.get('date_from') date_to = request.args.get('date_to') result, status_code = auth_handler.get_activity_log( limit=limit, offset=offset, user_id=user_id, activity_type=activity_type, action_code=action_code, bid=bid, username_contains=username_contains, description_contains=description_contains, date_from=date_from, date_to=date_to ) return jsonify(result), status_code # ============================================================================ # EMBED / IFRAME INTEGRATION ENDPOINTS # ============================================================================ @app.route('/api/embed/token', methods=['POST']) @handle_errors def get_embed_token(): """Generate a short-lived embed token for iframe integration. Called by external app backends with their API key.""" data = request.json if not data: return jsonify({'error': 'Request body is required'}), 400 api_key = data.get('api_key') bid = data.get('bid') if not api_key or not bid: return jsonify({'error': 'api_key and bid are required'}), 400 # Validate the API key key_record = auth_handler.validate_api_key(api_key, str(bid)) if not key_record: return jsonify({'error': 'Invalid API key or unauthorized for this business'}), 401 # Check origin if allowed_origins is set origin = request.headers.get('Origin') if key_record.get('allowed_origins') and origin: allowed = [o.strip() for o in key_record['allowed_origins'].split(',')] if origin not in allowed: return jsonify({'error': 'Origin not allowed'}), 403 # Generate short-lived embed token token = auth_handler.generate_embed_token( bid=str(bid), partner_name=key_record['partner_name'], api_key_id=key_record['id'] ) # Build embed URL embed_base_url = app.config.get('EMBED_BASE_URL', 'http://localhost:6174') embed_url = f"{embed_base_url}/#/embed?token={token}" return jsonify({ 'token': token, 'embed_url': embed_url, 'expires_in': 3600, 'bid': str(bid), 'partner_name': key_record['partner_name'] }), 200 @app.route('/api/embed/validate', methods=['GET']) @handle_errors def validate_embed_token_route(): """Validate an embed token. Called by the frontend embed page on load.""" token = request.args.get('token') if not token: return jsonify({'error': 'Token is required'}), 400 result = auth_handler.validate_embed_token(token) if not result: return jsonify({'error': 'Invalid or expired embed token'}), 401 return jsonify({ 'valid': True, 'bid': result['bid'], 'partner_name': result['partner_name'] }), 200 @app.route('/admin/embed-keys', methods=['POST']) @handle_errors def admin_create_embed_key(): """Create a new embed API key (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 data = request.json if not data: return jsonify({'error': 'Request body is required'}), 400 bid = data.get('bid') partner_name = data.get('partner_name') if not bid or not partner_name: return jsonify({'error': 'bid and partner_name are required'}), 400 allowed_origins = data.get('allowed_origins') expires_at = data.get('expires_at') result, status_code = auth_handler.create_embed_api_key( bid=str(bid), partner_name=partner_name, allowed_origins=allowed_origins, expires_at=expires_at ) # Log activity auth_handler.log_activity( user_id=user['id'], username=user['username'], activity_type='create_embed_key', description=f"Created embed API key for business {bid} (partner: {partner_name})", ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) return jsonify(result), status_code @app.route('/admin/embed-keys', methods=['GET']) @handle_errors def admin_list_embed_keys(): """List all embed API keys (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 bid = request.args.get('bid') keys, status_code = auth_handler.list_embed_api_keys(bid=bid) return jsonify({'keys': keys}), status_code @app.route('/admin/embed-keys/', methods=['DELETE']) @handle_errors def admin_revoke_embed_key(key_id): """Revoke an embed API key (master admin only)""" auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith('Bearer '): return jsonify({'error': 'Missing authorization header'}), 401 token = auth_header.replace('Bearer ', '') user = auth_handler.validate_token(token) if not user or not user.get('is_master'): return jsonify({'error': 'Master admin access required'}), 403 result, status_code = auth_handler.revoke_embed_api_key(key_id) # Log activity auth_handler.log_activity( user_id=user['id'], username=user['username'], activity_type='revoke_embed_key', description=f"Revoked embed API key {key_id}", ip_address=request.remote_addr, user_agent=request.headers.get('User-Agent') ) return jsonify(result), status_code if __name__ == '__main__': logger.info("Starting Call Analytics Dashboard API") logger.info(f"Environment: {app.config.get('ENV', 'development')}") logger.info(f"Debug mode: {app.config.get('DEBUG', False)}") # Run the app app.run( host=app.config.get('HOST', '0.0.0.0'), port=app.config.get('PORT', 5000), debug=app.config.get('DEBUG', False) ) # ============================================================================