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 from quality_parameters_handler import QualityParametersHandler from objection_handler import ObjectionClassificationsHandler from auth_handler import AuthHandler # 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 # 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 # ============================================================================ # 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 result, status_code = auth_handler.login(data['username'], data['password']) 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""" 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') 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') stats = db_handler.get_location_stats(bid, groupname) 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) - 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(bid, groupname, direction, call_status, limit) 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') # Get analytics overview overview = db_handler.get_analytics_overview(bid, groupname) # Get sentiment by location sentiment_by_location = db_handler.get_sentiment_by_location(bid, groupname) # Get quality score by location quality_by_location = db_handler.get_quality_by_location(bid, groupname) # Get call purpose frequency call_purposes = db_handler.get_call_purpose_frequency(bid, groupname) # Get objections/concerns frequency concerns_frequency = db_handler.get_concerns_frequency(bid, groupname) # Get busy locations busy_locations = db_handler.get_busy_locations(bid, 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(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', '') call['speaker_segments'] = transcript_data.get('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') return jsonify(call), 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 # ============================================================================ # 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 @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 # ============================================================================ # 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 # ============================================================================ 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) ) # ============================================================================ # 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 and data.get('businesses'): user_id = result['id'] for business in data['businesses']: bid = business.get('bid') if isinstance(business, dict) else business role = business.get('role', 'user') if isinstance(business, dict) else 'user' auth_handler.assign_business_access(user_id, bid, role) 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 data = request.json 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') ) 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