from flask import Flask, request, jsonify from flask_cors import CORS import pymysql from datetime import datetime, timedelta import json import logging from functools import wraps from config import Config from db_handler import DatabaseHandler from analytics import AnalyticsService # 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", "DELETE", "OPTIONS"], "allow_headers": ["Content-Type", "Authorization"] } }) # Initialize database handler db_handler = DatabaseHandler(app.config) # Initialize analytics service analytics_service = AnalyticsService(db_handler) # 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 # ============================================================================ # 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 # ============================================================================ # 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(): """Get list of all groupnames for Jubilant Foods (bid 7987) with their call counts""" groupnames = db_handler.get_all_groupnames() return jsonify(groupnames), 200 @app.route('/location-stats', methods=['GET']) @handle_errors def get_location_stats(): """ Get call statistics for a specific location (groupname) for Jubilant Foods Query params: - groupname: Filter by location/groupname (optional) """ groupname = request.args.get('groupname') stats = db_handler.get_location_stats(groupname) return jsonify(stats), 200 @app.route('/location-calls', methods=['GET']) @handle_errors def get_location_calls(): """ 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) - limit: Number of records (default: 100) """ groupname = request.args.get('groupname') direction = request.args.get('direction') call_status = request.args.get('call_status') limit = request.args.get('limit', 100, type=int) calls = db_handler.get_filtered_raw_calls(groupname, direction, call_status, limit) return jsonify(calls), 200 @app.route('/raw-calls/', methods=['GET']) @handle_errors def get_raw_call_details_route(callid): """Get call details from 7987_raw_calls with transcript from 7987_sarvamresponse""" call = db_handler.get_raw_call_details(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(callid): """Get analytics for a specific call""" analytics = db_handler.get_call_analytics(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(): """Get calls that need analytics""" limit = request.args.get('limit', 10, type=int) calls = db_handler.get_calls_for_analysis(limit) return jsonify({ 'count': len(calls), 'calls': calls }), 200 @app.route('/analytics/dashboard', methods=['GET']) @handle_errors def get_analytics_dashboard(): """Get comprehensive analytics dashboard data""" groupname = request.args.get('groupname') # Get analytics overview overview = db_handler.get_analytics_overview(groupname) # Get sentiment by location sentiment_by_location = db_handler.get_sentiment_by_location(groupname) # Get quality score by location quality_by_location = db_handler.get_quality_by_location(groupname) # Get call purpose frequency call_purposes = db_handler.get_call_purpose_frequency(groupname) # Get objections/concerns frequency concerns_frequency = db_handler.get_concerns_frequency(groupname) # Get busy locations busy_locations = db_handler.get_busy_locations(groupname) return jsonify({ 'overview': overview, 'sentiment_by_location': sentiment_by_location, 'quality_by_location': quality_by_location, '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(): """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(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', '') call['speaker_segments'] = transcript_data.get('speaker_segments') call['num_speakers'] = transcript_data.get('num_speakers') call['duration'] = transcript_data.get('duration') return jsonify(call), 200 @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 # ============================================================================ # 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 # ============================================================================ # 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: 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 (From source DB to 109 server) # ============================================================================ @app.route('/sync/check-count/', methods=['GET']) @handle_errors def check_call_count(bid): """ Check the number of calls available for syncing from source database 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') if not date_from or not date_to: return jsonify({'error': 'date_from and date_to are required'}), 400 # Source database configuration source_config = { 'host': '10.40.180.35', 'user': 'root', 'password': '4Tq73tXMcUbEJ5Q3t3', 'database': 'mcube_cl1', 'charset': 'utf8mb4' } try: source_conn = pymysql.connect(**source_config) source_cursor = source_conn.cursor(pymysql.cursors.DictCursor) # Count from callhistory table with status breakdown callhistory_query = f""" SELECT COUNT(*) as total, SUM(CASE WHEN dialstatus = 'ANSWER' THEN 1 ELSE 0 END) as answered, SUM(CASE WHEN dialstatus = 'CANCEL' THEN 1 ELSE 0 END) as cancelled, SUM(CASE WHEN dialstatus = 'BUSY' THEN 1 ELSE 0 END) as busy, SUM(CASE WHEN dialstatus = 'No Agent Available' THEN 1 ELSE 0 END) as no_agent FROM {bid}_callhistory WHERE DATE(starttime) BETWEEN %s AND %s AND dialstatus IN ('ANSWER', 'CANCEL', 'BUSY', 'No Agent Available') """ source_cursor.execute(callhistory_query, (date_from, date_to)) callhistory_data = source_cursor.fetchone() # Count from callarchive table with status breakdown callarchive_query = f""" SELECT COUNT(*) as total, SUM(CASE WHEN dialstatus = 'ANSWER' THEN 1 ELSE 0 END) as answered, SUM(CASE WHEN dialstatus = 'CANCEL' THEN 1 ELSE 0 END) as cancelled, SUM(CASE WHEN dialstatus = 'BUSY' THEN 1 ELSE 0 END) as busy, SUM(CASE WHEN dialstatus = 'No Agent Available' THEN 1 ELSE 0 END) as no_agent FROM {bid}_callarchive WHERE DATE(starttime) BETWEEN %s AND %s AND dialstatus IN ('ANSWER', 'CANCEL', 'BUSY', 'No Agent Available') """ source_cursor.execute(callarchive_query, (date_from, date_to)) callarchive_data = source_cursor.fetchone() source_cursor.close() source_conn.close() return jsonify({ 'callhistory': { 'total': int(callhistory_data['total'] or 0), 'answered': int(callhistory_data['answered'] or 0), 'cancelled': int(callhistory_data['cancelled'] or 0), 'busy': int(callhistory_data['busy'] or 0), 'no_agent': int(callhistory_data['no_agent'] or 0) }, 'callarchive': { 'total': int(callarchive_data['total'] or 0), 'answered': int(callarchive_data['answered'] or 0), 'cancelled': int(callarchive_data['cancelled'] or 0), 'busy': int(callarchive_data['busy'] or 0), 'no_agent': int(callarchive_data['no_agent'] or 0) }, 'total_count': int((callhistory_data['total'] or 0) + (callarchive_data['total'] or 0)), '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): """ Sync calls from source database (callhistory and callarchive) to 109 server Body: { "date_from": "YYYY-MM-DD", "date_to": "YYYY-MM-DD" } """ data = request.get_json() date_from = data.get('date_from') date_to = data.get('date_to') if not date_from or not date_to: return jsonify({'error': 'date_from and date_to are required'}), 400 # Source database configuration source_config = { 'host': '10.40.180.35', 'user': 'root', 'password': '4Tq73tXMcUbEJ5Q3t3', 'database': 'mcube_cl1', 'charset': 'utf8mb4' } # Destination database configuration (109 server) dest_config = { 'host': '10.0.0.109', 'user': 'admin', 'password': 'mcube@admin123', 'database': 'voicebot_cluster', 'charset': 'utf8mb4' } try: source_conn = pymysql.connect(**source_config) source_cursor = source_conn.cursor(pymysql.cursors.DictCursor) dest_conn = pymysql.connect(**dest_config) dest_cursor = dest_conn.cursor() total_inserted = 0 total_updated = 0 total_calls = [] # Fetch calls from both tables for table_name in [f'{bid}_callhistory', f'{bid}_callarchive']: try: query = f""" SELECT callid, bid, agentname, groupname, starttime, endtime, dialstatus, direction, filename, emp_phone, clicktocalldid FROM {table_name} WHERE DATE(starttime) BETWEEN %s AND %s AND dialstatus IN ('ANSWER', 'CANCEL', 'No Agent Available') """ source_cursor.execute(query, (date_from, date_to)) calls = source_cursor.fetchall() total_calls.extend(calls) logger.info(f"Found {len(calls)} calls in {table_name}") except pymysql.err.ProgrammingError as e: # Table might not exist logger.warning(f"Table {table_name} might not exist: {str(e)}") continue logger.info(f"Total calls to sync: {len(total_calls)}") # Insert into destination insert_query = f""" INSERT INTO {bid}_raw_calls (bid, callid, fileurl, status, agentname, groupname, call_starttime, call_endtime, call_status, agent_callinfo, customer_callinfo, direction, transcription_requested, transcription_status, selected_for_processing) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE fileurl = VALUES(fileurl), agentname = VALUES(agentname), groupname = VALUES(groupname), call_starttime = VALUES(call_starttime), call_endtime = VALUES(call_endtime), call_status = VALUES(call_status), agent_callinfo = VALUES(agent_callinfo), customer_callinfo = VALUES(customer_callinfo), direction = VALUES(direction) """ for call in total_calls: dest_cursor.execute(insert_query, ( call['bid'], call['callid'], call['filename'] or '', 0, # status call['agentname'] or '', call['groupname'] or '', call['starttime'], call['endtime'], call['dialstatus'] or '', call['emp_phone'] or '', call['clicktocalldid'] or '', call['direction'] or 'inbound', 0, # transcription_requested 'not_requested', # transcription_status 0 # selected_for_processing )) if dest_cursor.rowcount == 1: total_inserted += 1 elif dest_cursor.rowcount == 2: total_updated += 1 dest_conn.commit() source_cursor.close() source_conn.close() dest_cursor.close() dest_conn.close() return jsonify({ 'message': f'Successfully synced {len(total_calls)} calls', 'total_synced': len(total_calls), 'inserted': total_inserted, 'updated': total_updated, 'skipped': 0, 'date_from': date_from, 'date_to': date_to }), 200 except Exception as e: logger.error(f"Error syncing calls: {str(e)}") return jsonify({'error': f'Failed to sync calls: {str(e)}'}), 500 @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 = { 'host': '10.0.0.109', 'user': 'admin', 'password': 'mcube@admin123', 'database': 'voicebot_cluster', 'charset': 'utf8mb4' } 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 = { 'host': '10.0.0.109', 'user': 'admin', 'password': 'mcube@admin123', 'database': 'voicebot_cluster', 'charset': 'utf8mb4' } 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 # ============================================================================ # 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 # ============================================================================ 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) )