import pymysql from pymysql.cursors import DictCursor import logging import json from datetime import datetime from contextlib import contextmanager logger = logging.getLogger(__name__) class DatabaseHandler: """Handle all database operations for the dashboard""" def __init__(self, config): self.config = config self.db_config = { 'host': config.get('DB_HOST', '127.0.0.1'), 'port': config.get('DB_PORT', 3306), 'user': config.get('DB_USER', 'admin'), 'password': config.get('DB_PASSWORD', 'mcube@admin123'), 'database': config.get('DB_NAME', 'voicebot_cluster'), 'charset': 'utf8mb4', 'cursorclass': DictCursor, 'autocommit': True } @contextmanager def get_connection(self): """Context manager for database connections""" conn = None try: conn = pymysql.connect(**self.db_config) yield conn except pymysql.Error as e: logger.error(f"Database connection error: {e}") raise finally: if conn: conn.close() def _parse_json_field(self, value): """Parse JSON string fields""" if not value: return None if isinstance(value, str): try: return json.loads(value) except json.JSONDecodeError: return value return value def _format_call_record(self, record): """Format call record with proper JSON parsing""" if not record: return None # Parse JSON fields json_fields = ['keywords', 'sentiments', 'emotions', 'customer_details'] for field in json_fields: if field in record and record[field]: record[field] = self._parse_json_field(record[field]) return record # ======================================================================== # BUSINESS OPERATIONS # ======================================================================== def get_all_businesses(self): """Get list of all businesses with their call counts""" # This assumes you have a businesses table or can derive from call tables # If you only have dynamic tables, we'll scan for tables matching pattern {bid}_calls with self.get_connection() as conn: cursor = conn.cursor() # Get all tables matching pattern %_calls cursor.execute("SHOW TABLES LIKE '%_calls'") tables = cursor.fetchall() businesses = [] for table in tables: table_name = list(table.values())[0] bid = table_name.replace('_calls', '') # Get count of calls cursor.execute(f"SELECT COUNT(*) as count FROM `{table_name}`") count_result = cursor.fetchone() count = count_result['count'] if count_result else 0 # Try to get business name (if you have a businesses table) # For now, using BID as name businesses.append({ 'bid': bid, 'name': f'Business {bid}', 'totalCalls': count, 'callsTable': table_name }) return businesses def get_business_info(self, bid): """Get detailed information for a specific business""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_calls" # Check if table exists cursor.execute(f"SHOW TABLES LIKE '{table_name}'") if not cursor.fetchone(): return None # Get statistics query = f""" SELECT COUNT(*) as total_calls, COUNT(CASE WHEN status = 0 THEN 1 END) as unprocessed, COUNT(CASE WHEN status = 1 THEN 1 END) as transcribed, COUNT(CASE WHEN status = 2 THEN 1 END) as analyzed, COUNT(CASE WHEN status = 3 THEN 1 END) as message_sent, AVG(duration) as avg_duration, MIN(call_starttime) as first_call_starttime, MAX(call_starttime) as last_call_starttime FROM `{table_name}` """ cursor.execute(query) stats = cursor.fetchone() return { 'bid': bid, 'name': f'Business {bid}', 'statistics': stats } def get_all_groupnames(self): """Get list of all groupnames for Jubilant Foods (bid 7987) with their call counts""" with self.get_connection() as conn: cursor = conn.cursor() # Get distinct groupnames with counts from 7987_raw_calls query = """ SELECT groupname, COUNT(*) as totalCalls FROM `{bid}_raw_calls` WHERE groupname IS NOT NULL AND groupname != '' GROUP BY groupname ORDER BY totalCalls DESC """ cursor.execute(query) results = cursor.fetchall() groupnames = [] for row in results: groupnames.append({ 'groupname': row['groupname'], 'totalCalls': row['totalCalls'] }) return groupnames def get_location_stats(self, groupname=None): """Get call statistics for Jubilant Foods (7987) filtered by groupname""" with self.get_connection() as conn: cursor = conn.cursor() # Build WHERE clause where_clause = "WHERE 1=1" params = [] if groupname: where_clause += " AND groupname = %s" params.append(groupname) # Get comprehensive statistics query = f""" SELECT COUNT(*) as total_calls, -- Inbound statistics SUM(CASE WHEN direction = 'inbound' THEN 1 ELSE 0 END) as inbound_total, SUM(CASE WHEN direction = 'inbound' AND call_status = 'ANSWER' THEN 1 ELSE 0 END) as inbound_answered, SUM(CASE WHEN direction = 'inbound' AND call_status = 'BUSY' THEN 1 ELSE 0 END) as inbound_busy, SUM(CASE WHEN direction = 'inbound' AND call_status = 'CANCEL' THEN 1 ELSE 0 END) as inbound_cancel, SUM(CASE WHEN direction = 'inbound' AND call_status = 'NOANSWER' THEN 1 ELSE 0 END) as inbound_not_answered, -- Outbound statistics SUM(CASE WHEN direction = 'outbound' THEN 1 ELSE 0 END) as outbound_total, SUM(CASE WHEN direction = 'outbound' AND call_status = 'ANSWER' THEN 1 ELSE 0 END) as outbound_answered, SUM(CASE WHEN direction = 'outbound' AND call_status = 'BUSY' THEN 1 ELSE 0 END) as outbound_busy, SUM(CASE WHEN direction = 'outbound' AND call_status = 'CANCEL' THEN 1 ELSE 0 END) as outbound_cancel, SUM(CASE WHEN direction = 'outbound' AND call_status = 'NOANSWER' THEN 1 ELSE 0 END) as outbound_not_answered, -- Average duration for answered calls (in seconds) AVG( CASE WHEN call_status = 'ANSWER' AND call_starttime IS NOT NULL AND call_endtime IS NOT NULL THEN TIMESTAMPDIFF(SECOND, call_starttime, call_endtime) ELSE NULL END ) as avg_answered_duration FROM `{bid}_raw_calls` {where_clause} """ cursor.execute(query, params) result = cursor.fetchone() if not result: return { 'total_calls': 0, 'inbound_total': 0, 'inbound_answered': 0, 'inbound_busy': 0, 'inbound_cancel': 0, 'inbound_not_answered': 0, 'outbound_total': 0, 'outbound_answered': 0, 'outbound_busy': 0, 'outbound_cancel': 0, 'outbound_not_answered': 0, 'avg_answered_duration': 0 } return { 'total_calls': int(result['total_calls'] or 0), 'inbound_total': int(result['inbound_total'] or 0), 'inbound_answered': int(result['inbound_answered'] or 0), 'inbound_busy': int(result['inbound_busy'] or 0), 'inbound_cancel': int(result['inbound_cancel'] or 0), 'inbound_not_answered': int(result['inbound_not_answered'] or 0), 'outbound_total': int(result['outbound_total'] or 0), 'outbound_answered': int(result['outbound_answered'] or 0), 'outbound_busy': int(result['outbound_busy'] or 0), 'outbound_cancel': int(result['outbound_cancel'] or 0), 'outbound_not_answered': int(result['outbound_not_answered'] or 0), 'avg_answered_duration': int(result['avg_answered_duration'] or 0) } def get_filtered_raw_calls(self, groupname=None, direction=None, call_status=None, limit=100): """Get filtered raw calls from 7987_raw_calls table""" with self.get_connection() as conn: cursor = conn.cursor() # Build WHERE clause where_clauses = [] params = [] if groupname: where_clauses.append("groupname = %s") params.append(groupname) if direction: where_clauses.append("direction = %s") params.append(direction) if call_status: where_clauses.append("call_status = %s") params.append(call_status) where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" query = f""" SELECT callid, agentname, call_starttime, call_endtime, direction, call_status, groupname, TIMESTAMPDIFF(SECOND, call_starttime, call_endtime) as duration_seconds FROM `{bid}_raw_calls` {where_sql} ORDER BY call_starttime DESC LIMIT %s """ params.append(limit) cursor.execute(query, params) calls = cursor.fetchall() return calls def get_raw_call_details(self, callid): """Get call details from 7987_raw_calls joined with 7987_sarvamresponse""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT r.callid, r.bid, r.fileurl as fileUrl, r.agentname, r.groupname, r.call_starttime, r.call_endtime, r.call_status, r.agent_callinfo, r.customer_callinfo, r.direction, r.transcription_status, TIMESTAMPDIFF(SECOND, r.call_starttime, r.call_endtime) as duration_seconds, s.transcript as transcripts, s.speaker_segments, s.num_speakers, s.duration, s.language, s.request_id FROM `{bid}_raw_calls` r LEFT JOIN `{bid}_sarvamresponse` s ON r.callid = s.callid WHERE r.callid = %s LIMIT 1 """ cursor.execute(query, (callid,)) call = cursor.fetchone() if not call: return None # Format the call record call_dict = dict(call) # Set status for compatibility with CallDetail component call_dict['status'] = 2 if call_dict['transcripts'] else 0 return call_dict # ======================================================================== # CALL OPERATIONS # ======================================================================== def get_calls(self, bid, filters=None, limit=100, offset=0): """Get calls with optional filtering""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_calls" # Build WHERE clause where_clauses = [] params = [] if filters: if 'status' in filters and filters['status'] is not None: where_clauses.append("status = %s") params.append(filters['status']) if 'sales_intent' in filters: where_clauses.append("sales_intent = %s") params.append(filters['sales_intent']) if 'date_from' in filters: where_clauses.append("call_starttime >= %s") params.append(filters['date_from']) if 'date_to' in filters: where_clauses.append("call_starttime <= %s") params.append(filters['date_to']) where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" query = f""" SELECT * FROM `{table_name}` {where_sql} ORDER BY call_starttime DESC, call_endtime DESC LIMIT %s OFFSET %s """ params.extend([limit, offset]) cursor.execute(query, params) calls = cursor.fetchall() return [self._format_call_record(call) for call in calls] def get_calls_count(self, bid, filters=None): """Get total count of calls matching filters""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_calls" where_clauses = [] params = [] if filters: if 'status' in filters and filters['status'] is not None: where_clauses.append("status = %s") params.append(filters['status']) if 'sales_intent' in filters: where_clauses.append("sales_intent = %s") params.append(filters['sales_intent']) if 'date_from' in filters: where_clauses.append("call_starttime >= %s") params.append(filters['date_from']) if 'date_to' in filters: where_clauses.append("call_starttime <= %s") params.append(filters['date_to']) where_sql = " WHERE " + " AND ".join(where_clauses) if where_clauses else "" query = f"SELECT COUNT(*) as count FROM `{table_name}` {where_sql}" cursor.execute(query, params) result = cursor.fetchone() return result['count'] if result else 0 def get_call_by_id(self, bid, callid): """Get specific call by ID""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_calls" query = f"SELECT * FROM `{table_name}` WHERE callid = %s LIMIT 1" cursor.execute(query, (callid,)) call = cursor.fetchone() return self._format_call_record(call) def get_call_transcript(self, bid, callid): """Get transcript from sarvamresponse table""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_sarvamresponse" # Check if table exists cursor.execute(f"SHOW TABLES LIKE '{table_name}'") if not cursor.fetchone(): return None query = f""" SELECT transcript, language, raw_response, speaker_segments, num_speakers, duration FROM `{table_name}` WHERE callid = %s ORDER BY created_at DESC LIMIT 1 """ cursor.execute(query, (callid,)) return cursor.fetchone() def get_recent_calls(self, bid, limit=10): """Get most recent calls""" return self.get_calls(bid, filters=None, limit=limit, offset=0) def search_calls(self, bid, query, limit=50): """Search calls by query string""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_calls" search_query = f""" SELECT * FROM `{table_name}` WHERE callid LIKE %s OR customer_name LIKE %s OR agent_name LIKE %s OR summary LIKE %s OR transcripts LIKE %s ORDER BY call_starttime DESC, call_endtime DESC LIMIT %s """ search_term = f"%{query}%" cursor.execute(search_query, (search_term, search_term, search_term, search_term, search_term, limit)) calls = cursor.fetchall() return [self._format_call_record(call) for call in calls] def update_call(self, bid, callid, data): """Update call record""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_calls" # Build UPDATE query dynamically based on provided data set_clauses = [] params = [] for key, value in data.items(): if key in ['keywords', 'sentiments', 'emotions', 'customer_details']: # Convert dict/list to JSON string value = json.dumps(value) if value else None set_clauses.append(f"{key} = %s") params.append(value) if not set_clauses: return False set_clauses.append("updated_at = %s") params.append(datetime.now()) params.append(callid) query = f""" UPDATE `{table_name}` SET {', '.join(set_clauses)} WHERE callid = %s """ cursor.execute(query, params) return cursor.rowcount > 0 def save_conversation_summary(self, bid, callid, transfer_reason): """Save conversation summary to call_history table""" with self.get_connection() as conn: cursor = conn.cursor() table_name = f"{bid}_call_history" # Check if table exists cursor.execute(f"SHOW TABLES LIKE '{table_name}'") if not cursor.fetchone(): logger.warning(f"Table {table_name} does not exist") return False query = f""" INSERT INTO `{table_name}` (business_id, callid, transfer_reason, created_at) VALUES (%s, %s, %s, %s) """ cursor.execute(query, ( bid, callid, json.dumps(transfer_reason) if transfer_reason else None, datetime.now() )) return cursor.rowcount > 0 # ======================================================================== # TRANSCRIPT OPERATIONS # ======================================================================== def get_transcripts(self, bid): """Get all transcripts with metadata from sarvamresponse table""" with self.get_connection() as conn: cursor = conn.cursor() # Check if sarvamresponse table exists sarvam_table = f"{bid}_sarvamresponse" cursor.execute(f"SHOW TABLES LIKE '{sarvam_table}'") if not cursor.fetchone(): logger.warning(f"Table {sarvam_table} does not exist") return [] # Get all transcripts with their metadata query = f""" SELECT s.callid as transcript_id, s.callid, s.transcript as full_transcript, s.speaker_segments, s.num_speakers, s.duration, s.language, s.created_at, s.updated_at, c.call_starttime, c.call_endtime, c.customer_callinfo, c.agentname, TIMESTAMPDIFF(SECOND, c.call_starttime, c.call_endtime) as duration_seconds, CASE WHEN s.transcript IS NOT NULL AND s.transcript != '' THEN TRUE ELSE FALSE END as stored_in_vectordb FROM `{sarvam_table}` s LEFT JOIN `{bid}_calls` c ON s.callid = c.callid ORDER BY s.created_at DESC """ try: cursor.execute(query) transcripts = cursor.fetchall() # Process each transcript to add computed fields result = [] for t in transcripts: transcript_text = t.get('full_transcript', '') # Use actual speaker data from diarization if available num_speakers = t.get('num_speakers') if num_speakers is None: # Fallback estimation num_speakers = 2 if transcript_text else 0 # Calculate segments from speaker_segments JSON if available speaker_segments = t.get('speaker_segments') if speaker_segments: try: import json segments_data = json.loads(speaker_segments) if isinstance(speaker_segments, str) else speaker_segments num_segments = len(segments_data) if segments_data else 0 except: # Fallback: estimate segments by line breaks num_segments = len([line for line in transcript_text.split('\n') if line.strip()]) if transcript_text else 0 else: # Fallback: estimate segments by line breaks num_segments = len([line for line in transcript_text.split('\n') if line.strip()]) if transcript_text else 0 # Create filename from callid and timestamp customer_info = t.get('customer_callinfo', 'unknown') filename = f"Call_{t.get('callid', 'unknown')}_{str(customer_info).replace(' ', '_')}.txt" result.append({ 'transcript_id': t.get('transcript_id'), 'callid': t.get('callid'), 'filename': filename, 'full_transcript': transcript_text, 'language': t.get('language'), 'num_speakers': num_speakers, 'num_segments': num_segments, 'duration': t.get('duration') or t.get('duration_seconds', 0), # Use sarvam duration if available, else calculated 'created_at': t.get('created_at').isoformat() if t.get('created_at') else None, 'updated_at': t.get('updated_at').isoformat() if t.get('updated_at') else None, 'stored_in_vectordb': bool(t.get('stored_in_vectordb')), 'customer_name': customer_info, 'agent_name': t.get('agentname') }) return result except Exception as e: logger.error(f"Error fetching transcripts: {e}") return [] def save_call_analytics(self, callid, analytics_data): """Save call analytics to 7987_callanalytics table""" with self.get_connection() as conn: cursor = conn.cursor() query = """ INSERT INTO `{bid}_callanalytics` (callid, bid, summary, call_purpose, objections_concerns, quality_score, sentiment, analysis_model, raw_response) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE summary = VALUES(summary), call_purpose = VALUES(call_purpose), objections_concerns = VALUES(objections_concerns), quality_score = VALUES(quality_score), sentiment = VALUES(sentiment), analysis_model = VALUES(analysis_model), raw_response = VALUES(raw_response), updated_at = CURRENT_TIMESTAMP """ cursor.execute(query, ( callid, analytics_data.get('bid', '7987'), analytics_data.get('summary'), analytics_data.get('call_purpose'), analytics_data.get('objections_concerns'), analytics_data.get('quality_score'), analytics_data.get('sentiment'), analytics_data.get('analysis_model', 'aws-nova'), json.dumps(analytics_data.get('raw_response')) if analytics_data.get('raw_response') else None )) conn.commit() return cursor.lastrowid def get_call_analytics(self, callid): """Get analytics for a specific call""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT * FROM `{bid}_callanalytics` WHERE callid = %s LIMIT 1 """ cursor.execute(query, (callid,)) result = cursor.fetchone() if result and result.get('raw_response'): result['raw_response'] = self._parse_json_field(result['raw_response']) return result def get_calls_for_analysis(self, limit=10): """Get calls that have transcripts but no analytics yet""" with self.get_connection() as conn: cursor = conn.cursor() query = """ SELECT r.callid, r.bid, r.agentname, r.groupname, r.call_starttime, r.call_endtime, s.transcript, s.speaker_segments, s.num_speakers, s.duration FROM `{bid}_raw_calls` r INNER JOIN `{bid}_sarvamresponse` s ON r.callid = s.callid LEFT JOIN `{bid}_callanalytics` a ON r.callid = a.callid WHERE r.call_status = 'ANSWER' AND s.transcript IS NOT NULL AND s.transcript != '' AND a.callid IS NULL ORDER BY r.call_starttime DESC LIMIT %s """ cursor.execute(query, (limit,)) results = cursor.fetchall() # Parse JSON fields for result in results: if result.get('speaker_segments'): result['speaker_segments'] = self._parse_json_field(result['speaker_segments']) return results def get_analytics_overview(self, groupname=None): """Get overall analytics summary""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "" params = [] if groupname: where_clause = "WHERE r.groupname = %s" params.append(groupname) query = f""" SELECT COUNT(DISTINCT a.callid) as total_analyzed_calls, AVG(a.quality_score) as avg_quality_score, SUM(CASE WHEN a.sentiment = 'positive' THEN 1 ELSE 0 END) as positive_calls, SUM(CASE WHEN a.sentiment = 'neutral' THEN 1 ELSE 0 END) as neutral_calls, SUM(CASE WHEN a.sentiment = 'negative' THEN 1 ELSE 0 END) as negative_calls, COUNT(DISTINCT r.groupname) as total_locations FROM `{bid}_callanalytics` a INNER JOIN `{bid}_raw_calls` r ON a.callid = r.callid {where_clause} """ cursor.execute(query, params) result = cursor.fetchone() return result if result else {} def get_sentiment_by_location(self, groupname=None): """Get sentiment distribution by location""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "" params = [] if groupname: where_clause = "AND r.groupname = %s" params.append(groupname) query = f""" SELECT r.groupname as location, a.sentiment, COUNT(*) as count FROM `{bid}_callanalytics` a INNER JOIN `{bid}_raw_calls` r ON a.callid = r.callid WHERE a.sentiment IS NOT NULL {where_clause} GROUP BY r.groupname, a.sentiment ORDER BY r.groupname, a.sentiment """ cursor.execute(query, params) return cursor.fetchall() def get_quality_by_location(self, groupname=None): """Get average quality score by location""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "" params = [] if groupname: where_clause = "WHERE r.groupname = %s" params.append(groupname) query = f""" SELECT r.groupname as location, AVG(a.quality_score) as avg_quality_score, MIN(a.quality_score) as min_quality_score, MAX(a.quality_score) as max_quality_score, COUNT(*) as call_count FROM `{bid}_callanalytics` a INNER JOIN `{bid}_raw_calls` r ON a.callid = r.callid {where_clause} GROUP BY r.groupname ORDER BY avg_quality_score DESC """ cursor.execute(query, params) return cursor.fetchall() def get_call_purpose_frequency(self, groupname=None): """Get frequency of different call purposes""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "" params = [] if groupname: where_clause = "AND r.groupname = %s" params.append(groupname) query = f""" SELECT a.call_purpose, COUNT(*) as count FROM `{bid}_callanalytics` a INNER JOIN `{bid}_raw_calls` r ON a.callid = r.callid WHERE a.call_purpose IS NOT NULL AND a.call_purpose != '' {where_clause} GROUP BY a.call_purpose ORDER BY count DESC LIMIT 20 """ cursor.execute(query, params) return cursor.fetchall() def get_concerns_frequency(self, groupname=None): """Get frequency of different concerns/objections""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "" params = [] if groupname: where_clause = "AND r.groupname = %s" params.append(groupname) query = f""" SELECT a.objections_concerns, COUNT(*) as count FROM `{bid}_callanalytics` a INNER JOIN `{bid}_raw_calls` r ON a.callid = r.callid WHERE a.objections_concerns IS NOT NULL AND a.objections_concerns != '' AND a.objections_concerns != 'None identified' AND a.objections_concerns != 'None' {where_clause} GROUP BY a.objections_concerns ORDER BY count DESC LIMIT 20 """ cursor.execute(query, params) return cursor.fetchall() def get_busy_locations(self, groupname=None): """Get busiest locations by call volume""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "" params = [] if groupname: where_clause = "WHERE r.groupname = %s" params.append(groupname) query = f""" SELECT r.groupname as location, COUNT(DISTINCT a.callid) as analyzed_calls, COUNT(DISTINCT r.callid) as total_calls, ROUND(COUNT(DISTINCT a.callid) * 100.0 / COUNT(DISTINCT r.callid), 2) as analysis_percentage FROM `{bid}_raw_calls` r LEFT JOIN `{bid}_callanalytics` a ON r.callid = a.callid {where_clause} GROUP BY r.groupname ORDER BY analyzed_calls DESC LIMIT 20 """ cursor.execute(query, params) return cursor.fetchall() def get_calls_by_objection(self, objection, groupname=None): """Get all calls with a specific objection/concern""" with self.get_connection() as conn: cursor = conn.cursor() where_clause = "WHERE a.objections_concerns = %s" params = [objection] if groupname: where_clause += " AND r.groupname = %s" params.append(groupname) query = f""" SELECT a.callid, r.groupname as location, a.call_purpose, a.sentiment, a.quality_score, a.objections_concerns, a.created_at FROM `{bid}_callanalytics` a INNER JOIN `{bid}_raw_calls` r ON a.callid = r.callid {where_clause} ORDER BY a.created_at DESC LIMIT 100 """ cursor.execute(query, params) return cursor.fetchall()