from flask import Flask, request, jsonify import json import logging from rabbit import get_unprocessed_calls, send_to_rabbitmq from sarvam_processor import send_to_sarvam import mysql.connector from mysql.connector import Error from datetime import datetime # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) # Database configuration DB_CONFIG = { 'host': '10.0.0.204', 'port': 3306, 'database': 'voicebot_cluster', 'user': 'root', # Change this to your MySQL username 'password': 'mcube@admin123', # Change this to your MySQL password 'charset': 'utf8mb4', 'collation': 'utf8mb4_unicode_ci' } @app.route('/health', methods=['GET']) def health_check(): """Health check endpoint""" return jsonify({ "status": "healthy", "message": "RabbitMQ Sarvam API is running", "version": "1.0.0" }) @app.route('/queue-calls', methods=['POST']) def queue_calls(): """ Queue unprocessed calls for a specific business ID Expected JSON payload: { "bid": "6840" // Business ID (optional, defaults to 7417) } """ try: # Get JSON data from request data = request.get_json() if not data: return jsonify({ "success": False, "error": "No JSON data provided", "message": "Please provide JSON with 'bid' field" }), 400 # Extract bid from request bid = data.get('bid') if not bid: bid = None # Will use default table logger.info("No bid provided, using default table") logger.info(f"Processing request for business ID: {bid or 'default'}") # Get unprocessed calls calls = get_unprocessed_calls(bid) if not calls: return jsonify({ "success": True, "message": f"No unprocessed calls found for business: {bid or 'default'}", "business_id": bid, "queued_count": 0 }) # Queue the calls send_to_rabbitmq(calls) return jsonify({ "success": True, "message": f"Successfully queued {len(calls)} calls for business: {bid or 'default'}", "business_id": bid, "queued_count": len(calls), "calls": [{"callid": call[0], "bid": call[1]} for call in calls] }) except Exception as e: logger.error(f"Error in queue_calls: {str(e)}") return jsonify({ "success": False, "error": str(e), "message": "Failed to queue calls" }), 500 @app.route('/process-calls', methods=['POST']) def process_calls(): """ Directly process calls for a specific business ID (bypasses RabbitMQ) Expected JSON payload: { "bid": "6840" // Business ID (optional, defaults to 7417) } """ try: # Get JSON data from request data = request.get_json() if not data: return jsonify({ "success": False, "error": "No JSON data provided", "message": "Please provide JSON with 'bid' field" }), 400 # Extract bid from request bid = data.get('bid') if not bid: bid = None # Will use default table logger.info("No bid provided, using default table") logger.info(f"Direct processing request for business ID: {bid or 'default'}") # Process calls directly result = send_to_sarvam(bid) return jsonify({ "success": True, "message": "Calls processed successfully", "business_id": bid, "result": result }) except Exception as e: logger.error(f"Error in process_calls: {str(e)}") return jsonify({ "success": False, "error": str(e), "message": "Failed to process calls" }), 500 @app.route('/queue-calls/', methods=['POST']) def queue_calls_by_bid(bid): """ Queue calls for a specific business ID via URL parameter URL: POST /queue-calls/6840 """ try: logger.info(f"Processing request for business ID: {bid}") # Get unprocessed calls calls = get_unprocessed_calls(bid) if not calls: return jsonify({ "success": True, "message": f"No unprocessed calls found for business: {bid}", "business_id": bid, "queued_count": 0 }) # Queue the calls send_to_rabbitmq(calls) return jsonify({ "success": True, "message": f"Successfully queued {len(calls)} calls for business: {bid}", "business_id": bid, "queued_count": len(calls), "calls": [{"callid": call[0], "bid": call[1]} for call in calls] }) except Exception as e: logger.error(f"Error in queue_calls_by_bid: {str(e)}") return jsonify({ "success": False, "error": str(e), "message": "Failed to queue calls" }), 500 @app.route('/process-calls/', methods=['POST']) def process_calls_by_bid(bid): """ Directly process calls for a specific business ID via URL parameter URL: POST /process-calls/6840 """ try: logger.info(f"Direct processing request for business ID: {bid}") # Process calls directly result = send_to_sarvam(bid) return jsonify({ "success": True, "message": "Calls processed successfully", "business_id": bid, "result": result }) except Exception as e: logger.error(f"Error in process_calls_by_bid: {str(e)}") return jsonify({ "success": False, "error": str(e), "message": "Failed to process calls" }), 500 @app.route('/list-businesses', methods=['GET']) def list_businesses(): """ List all available business tables """ try: from db_config import get_db_connection conn = get_db_connection() cursor = conn.cursor() # Get all business tables cursor.execute("SHOW TABLES LIKE '%_calls'") tables = cursor.fetchall() businesses = [] for table in tables: bid = table[0].replace('_calls', '') businesses.append({ "bid": bid, "calls_table": f"{bid}_calls", "response_table": f"{bid}_sarvamresponse" }) cursor.close() conn.close() return jsonify({ "success": True, "businesses": businesses, "count": len(businesses) }) except Exception as e: logger.error(f"Error in list_businesses: {str(e)}") return jsonify({ "success": False, "error": str(e), "message": "Failed to list businesses" }), 500 def get_db_connection(): """Create database connection""" try: connection = mysql.connector.connect(**DB_CONFIG) return connection except Error as e: print(f"Error connecting to MySQL: {e}") return None def validate_call_data(data): """Validate the incoming call data""" required_fields = ['bid', 'callid'] for field in required_fields: if field not in data: return False, f"Missing required field: {field}" return True, "Valid" # @app.route('/update-call', methods=['POST']) # def update_call(): # """Update or insert call data into dynamic bid_calls table""" # try: # data = request.get_json() # if not data: # return jsonify({ # "success": False, # "error": "No JSON data provided" # }), 400 # # Validate required fields # is_valid, message = validate_call_data(data) # if not is_valid: # return jsonify({ # "success": False, # "error": message # }), 400 # # Extract bid and validate it # bid = data.get('bid') # if not bid: # return jsonify({ # "success": False, # "error": "Business ID (bid) is required" # }), 400 # # Convert bid to string and validate format # bid_str = str(bid).strip() # if not bid_str.isdigit(): # return jsonify({ # "success": False, # "error": "Business ID must be numeric" # }), 400 # # Create dynamic table name # calls_table = f"{bid_str}_calls" # # Validate allowed fields to prevent SQL injection # allowed_fields = { # 'fileurl', 'fileUrl', 'status', 'agentname', 'agent_name', # 'duration', 'call_duration', 'customer_id', 'customer_name', # 'call_type', 'call_date', 'call_time', 'notes', 'priority', # 'department', 'resolution', 'tags', 'metadata' # } # # Get database connection # connection = get_db_connection() # if not connection: # return jsonify({ # "success": False, # "error": "Database connection failed" # }), 500 # cursor = connection.cursor() # # Check if table exists # cursor.execute("SHOW TABLES LIKE %s", (calls_table,)) # if not cursor.fetchone(): # cursor.close() # connection.close() # return jsonify({ # "success": False, # "error": f"Table {calls_table} does not exist" # }), 400 # # Check if call already exists # check_query = f"SELECT id FROM {calls_table} WHERE bid = %s AND callid = %s" # cursor.execute(check_query, (bid_str, data['callid'])) # existing_call = cursor.fetchone() # current_time = datetime.now() # if existing_call: # # Update existing call # update_fields = [] # update_values = [] # # Build dynamic update query with field validation # for field, value in data.items(): # if field not in ['bid', 'callid'] and field.lower() in [f.lower() for f in allowed_fields]: # update_fields.append(f"`{field}` = %s") # update_values.append(value) # if update_fields: # update_fields.append("`updated_at` = %s") # update_values.append(current_time) # update_values.extend([bid_str, data['callid']]) # update_query = f"UPDATE {calls_table} SET {', '.join(update_fields)} WHERE bid = %s AND callid = %s" # cursor.execute(update_query, update_values) # message = f"Call updated successfully in {calls_table}" # else: # message = "No valid fields to update" # else: # # Insert new call # insert_fields = ['bid', 'callid', 'created_at', 'updated_at'] # insert_values = [bid_str, data['callid'], current_time, current_time] # # Add other fields if provided and valid # for field, value in data.items(): # if field not in ['bid', 'callid'] and field.lower() in [f.lower() for f in allowed_fields]: # insert_fields.append(f"`{field}`") # insert_values.append(value) # placeholders = ', '.join(['%s'] * len(insert_fields)) # insert_query = f"INSERT INTO {calls_table} ({', '.join(insert_fields)}) VALUES ({placeholders})" # cursor.execute(insert_query, insert_values) # message = f"New call inserted successfully into {calls_table}" # connection.commit() # cursor.close() # connection.close() # return jsonify({ # "success": True, # "message": message, # "bid": bid_str, # "callid": data['callid'], # "table": calls_table, # "timestamp": current_time.isoformat() # }) # except mysql.connector.Error as e: # return jsonify({ # "success": False, # "error": f"Database error: {str(e)}" # }), 500 # except Exception as e: # logger.error(f"Error in update_call: {str(e)}") # return jsonify({ # "success": False, # "error": f"Server error: {str(e)}" # }), 500 @app.route('/update-call', methods=['POST']) def update_call(): """Update or insert call data into dynamic bid_calls table""" try: data = request.get_json() if not data: return jsonify({ "success": False, "error": "No JSON data provided" }), 400 # Validate required fields is_valid, message = validate_call_data(data) if not is_valid: return jsonify({ "success": False, "error": message }), 400 # Extract bid and validate it bid = data.get('bid') if not bid: return jsonify({ "success": False, "error": "Business ID (bid) is required" }), 400 # Convert bid to string and validate format bid_str = str(bid).strip() if not bid_str.isdigit(): return jsonify({ "success": False, "error": "Business ID must be numeric" }), 400 # Create dynamic table name calls_table = f"{bid_str}_calls" # Validate allowed fields to prevent SQL injection allowed_fields = { 'fileurl', 'fileUrl', 'status', 'agentname', 'agent_name', 'duration', 'call_duration', 'customer_id', 'customer_name', 'call_type', 'call_date', 'call_time', 'notes', 'priority', 'department', 'resolution', 'tags', 'metadata' } # Get database connection connection = get_db_connection() if not connection: return jsonify({ "success": False, "error": "Database connection failed" }), 500 cursor = connection.cursor() # Check if table exists cursor.execute("SHOW TABLES LIKE %s", (calls_table,)) if not cursor.fetchone(): cursor.close() connection.close() return jsonify({ "success": False, "error": f"Table {calls_table} does not exist" }), 400 # Check if call already exists check_query = f"SELECT id FROM {calls_table} WHERE bid = %s AND callid = %s" cursor.execute(check_query, (bid_str, data['callid'])) existing_call = cursor.fetchone() current_time = datetime.now() if existing_call: # Skip update if callid already exists cursor.close() connection.close() return jsonify({ "success": True, "message": f"Call with callid {data['callid']} already exists in {calls_table}, skipping update", "bid": bid_str, "callid": data['callid'], "table": calls_table, "timestamp": current_time.isoformat(), "skipped": True }) else: # Insert new call insert_fields = ['bid', 'callid', 'created_at', 'updated_at'] insert_values = [bid_str, data['callid'], current_time, current_time] # Add other fields if provided and valid for field, value in data.items(): if field not in ['bid', 'callid'] and field.lower() in [f.lower() for f in allowed_fields]: insert_fields.append(f"`{field}`") insert_values.append(value) placeholders = ', '.join(['%s'] * len(insert_fields)) insert_query = f"INSERT INTO {calls_table} ({', '.join(insert_fields)}) VALUES ({placeholders})" cursor.execute(insert_query, insert_values) message = f"New call inserted successfully into {calls_table}" connection.commit() cursor.close() connection.close() return jsonify({ "success": True, "message": message, "bid": bid_str, "callid": data['callid'], "table": calls_table, "timestamp": current_time.isoformat() }) except mysql.connector.Error as e: return jsonify({ "success": False, "error": f"Database error: {str(e)}" }), 500 except Exception as e: logger.error(f"Error in update_call: {str(e)}") return jsonify({ "success": False, "error": f"Server error: {str(e)}" }), 500 @app.route('/send-conversation-summary', methods=['POST']) def receive_conversation_summary(): try: data = request.get_json() if not data: return jsonify({ 'success': False, 'error': 'No JSON data provided' }), 400 bid = data.get('bid') callid = data.get('callid') conversation_summary = data.get('conversation_summary') timestamp = data.get('timestamp') transfer_reason = data.get('transfer_reason') print(f"Received conversation summary for call {callid}") print(f"Business ID: {bid}") print(f"Summary: {conversation_summary}") # Validate required fields if not bid or not callid: return jsonify({ 'success': False, 'error': 'bid and callid are required fields' }), 400 # Get database connection connection = get_db_connection() if not connection: return jsonify({ 'success': False, 'error': 'Database connection failed' }), 500 cursor = connection.cursor() # Create the complete conversation summary data structure business_id = bid # Define business_id variable summary_data = { 'business_id': business_id, 'callid': callid, 'conversation_summary': conversation_summary, 'timestamp': timestamp, 'transfer_reason': transfer_reason, 'received_at': datetime.now().isoformat() } # Convert to JSON string for storage transfer_reason_json = json.dumps(summary_data, indent=2) call_history_table = f"{business_id}_call_history" # Update the bid_call_history table with the conversation summary update_query = f""" UPDATE `{call_history_table}` SET transfer_reason = %s, updated_at = %s WHERE business_id = %s AND callid = %s """ current_time = datetime.now() cursor.execute(update_query, (transfer_reason_json, current_time, business_id, callid)) # Check if any rows were affected if cursor.rowcount == 0: # If no rows were updated, try to insert a new record insert_query = f""" INSERT INTO `{call_history_table}` (business_id, callid, transfer_reason, created_at, updated_at) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(insert_query, (business_id, callid, transfer_reason_json, current_time, current_time)) message = f"New conversation summary inserted for call {callid}" else: message = f"Conversation summary updated for call {callid}" connection.commit() cursor.close() connection.close() logger.info(f"Successfully stored conversation summary for call {callid} in bid_call_history table") return jsonify({ 'success': True, 'message': message, 'callid': callid, 'bid': bid, 'timestamp': current_time.isoformat() }), 200 except mysql.connector.Error as e: logger.error(f"Database error in receive_conversation_summary: {str(e)}") return jsonify({ 'success': False, 'error': f'Database error: {str(e)}' }), 500 except Exception as e: logger.error(f"Error in receive_conversation_summary: {str(e)}") return jsonify({ 'success': False, 'error': f'Server error: {str(e)}' }), 500 @app.errorhandler(404) def not_found(error): return jsonify({ "success": False, "error": "Endpoint not found", "message": "Please check the API documentation" }), 404 @app.errorhandler(500) def internal_error(error): return jsonify({ "success": False, "error": "Internal server error", "message": "Something went wrong on our end" }), 500 if __name__ == '__main__': print("🚀 Starting RabbitMQ Sarvam API Server...") print("📋 Available endpoints:") print(" GET /health - Health check") print(" POST /queue-calls - Queue calls (JSON body)") print(" POST /queue-calls/ - Queue calls (URL param)") print(" POST /process-calls - Process calls directly (JSON body)") print(" POST /process-calls/ - Process calls directly (URL param)") print(" GET /list-businesses - List all businesses") print("🌐 Server will start on http://localhost:5000") app.run(debug=True, host='0.0.0.0', port=4567)