Create app.py

app.py

@@ -0,0 +1,404 @@
+import os
+import json
+from datetime import datetime
+import gradio as gr
+from twilio.rest import Client
+from twilio.twiml.voice_response import VoiceResponse, Gather, Dial
+
+# Initialize Twilio client - you'll need to set these environment variables
+account_sid = os.environ.get('TWILIO_ACCOUNT_SID')
+auth_token = os.environ.get('TWILIO_AUTH_TOKEN')
+twilio_client = Client(account_sid, auth_token)
+
+# Database simulation (in a real app, use a proper database)
+call_routes = {}
+call_logs = []
+tracking_numbers = {}
+
+# Call routing configuration
+def save_call_route(name, description, tracking_number, destinations, routing_type, schedules=None):
+    """Save a call routing configuration"""
+    if tracking_number in tracking_numbers:
+        route_id = tracking_numbers[tracking_number]
+    else:
+        route_id = f"route_{len(call_routes) + 1}"
+        tracking_numbers[tracking_number] = route_id
+    
+    call_routes[route_id] = {
+        "name": name,
+        "description": description,
+        "tracking_number": tracking_number,
+        "destinations": destinations,
+        "routing_type": routing_type,  # "sequential", "simultaneous", "percentage", "time_based"
+        "schedules": schedules if schedules else [],
+        "created_at": datetime.now().isoformat()
+    }
+    return route_id
+
+# Twilio TwiML generation for call routing
+def generate_call_routing_twiml(tracking_number, caller_id, call_sid):
+    """Generate TwiML for routing calls based on the configuration"""
+    resp = VoiceResponse()
+    
+    # Log the call
+    log_call(call_sid, caller_id, tracking_number, "inbound", "initiated")
+    
+    # Find the appropriate route
+    if tracking_number not in tracking_numbers:
+        resp.say("We're sorry, but this number is not configured for routing.")
+        return str(resp)
+    
+    route_id = tracking_numbers[tracking_number]
+    route = call_routes[route_id]
+    
+    # Check if there's a greeting message
+    if "greeting" in route:
+        resp.say(route["greeting"])
+    
+    # Check routing type and implement accordingly
+    if route["routing_type"] == "simultaneous":
+        # Ring all destinations at once
+        dial = Dial(
+            caller_id=caller_id,
+            action=f"/call-status?call_sid={call_sid}&tracking_number={tracking_number}",
+            timeout=30
+        )
+        for dest in route["destinations"]:
+            dial.number(dest["number"])
+        resp.append(dial)
+    
+    elif route["routing_type"] == "sequential":
+        # Use Twilio's redirect to chain through destinations
+        resp.redirect(f"/sequential-route?call_sid={call_sid}&tracking_number={tracking_number}&index=0")
+    
+    elif route["routing_type"] == "percentage":
+        # Implement percentage-based routing (simplified version)
+        import random
+        total = sum(dest.get("percentage", 0) for dest in route["destinations"])
+        r = random.randint(1, total)
+        cumulative = 0
+        
+        for dest in route["destinations"]:
+            cumulative += dest.get("percentage", 0)
+            if r <= cumulative:
+                dial = Dial(
+                    caller_id=caller_id,
+                    action=f"/call-status?call_sid={call_sid}&tracking_number={tracking_number}",
+                    timeout=30
+                )
+                dial.number(dest["number"])
+                resp.append(dial)
+                break
+    
+    elif route["routing_type"] == "time_based":
+        # Time-based routing implementation
+        now = datetime.now()
+        current_day = now.strftime("%A").lower()
+        current_hour = now.hour
+        
+        # Find the appropriate schedule
+        for schedule in route.get("schedules", []):
+            if current_day in schedule["days"]:
+                start_hour = int(schedule["start_time"].split(":")[0])
+                end_hour = int(schedule["end_time"].split(":")[0])
+                
+                if start_hour <= current_hour < end_hour:
+                    # Route to the destination for this schedule
+                    dial = Dial(
+                        caller_id=caller_id,
+                        action=f"/call-status?call_sid={call_sid}&tracking_number={tracking_number}",
+                        timeout=30
+                    )
+                    dial.number(schedule["destination"])
+                    resp.append(dial)
+                    return str(resp)
+        
+        # If no schedule matches, use the default destination or voicemail
+        if "default_destination" in route:
+            dial = Dial(
+                caller_id=caller_id,
+                action=f"/call-status?call_sid={call_sid}&tracking_number={tracking_number}",
+                timeout=30
+            )
+            dial.number(route["default_destination"])
+            resp.append(dial)
+        else:
+            resp.say("I'm sorry, we're currently closed. Please leave a message after the tone.")
+            resp.record(
+                action=f"/voicemail?call_sid={call_sid}&tracking_number={tracking_number}",
+                max_length=120
+            )
+    
+    return str(resp)
+
+# Call logging
+def log_call(call_sid, caller_id, tracking_number, call_type, status, duration=0, recording_url=None):
+    """Log a call in the system"""
+    call_logs.append({
+        "call_sid": call_sid,
+        "caller_id": caller_id,
+        "tracking_number": tracking_number,
+        "timestamp": datetime.now().isoformat(),
+        "type": call_type,
+        "status": status,
+        "duration": duration,
+        "recording_url": recording_url
+    })
+    return len(call_logs) - 1
+
+# Purchase a new Twilio number
+def purchase_tracking_number(area_code, friendly_name):
+    """Purchase a new Twilio phone number"""
+    try:
+        available_numbers = twilio_client.available_phone_numbers('US').local.list(area_code=area_code, limit=1)
+        
+        if not available_numbers:
+            return {"success": False, "message": f"No available numbers found with area code {area_code}"}
+        
+        number = available_numbers[0]
+        purchased_number = twilio_client.incoming_phone_numbers.create(
+            phone_number=number.phone_number,
+            friendly_name=friendly_name,
+            voice_url="https://your-gradio-app-url.com/voice" # You'll need to update this with your actual URL
+        )
+        
+        return {
+            "success": True, 
+            "number": purchased_number.phone_number,
+            "sid": purchased_number.sid
+        }
+    except Exception as e:
+        return {"success": False, "message": str(e)}
+
+# Gradio Interface
+
+# Function to display the call logs
+def view_call_logs():
+    if not call_logs:
+        return "No calls logged yet."
+    
+    output = "Call Logs:\n\n"
+    for i, log in enumerate(call_logs):
+        output += f"Call {i+1}:\n"
+        output += f"  SID: {log['call_sid']}\n"
+        output += f"  From: {log['caller_id']}\n"
+        output += f"  To: {log['tracking_number']}\n"
+        output += f"  Time: {log['timestamp']}\n"
+        output += f"  Status: {log['status']}\n"
+        output += f"  Duration: {log['duration']} seconds\n"
+        if log.get('recording_url'):
+            output += f"  Recording: {log['recording_url']}\n"
+        output += "\n"
+    
+    return output
+
+# Function to create a new call route
+def create_call_route(name, description, tracking_number, destinations_json, routing_type, schedules_json=None):
+    try:
+        destinations = json.loads(destinations_json)
+        schedules = json.loads(schedules_json) if schedules_json else None
+        
+        route_id = save_call_route(name, description, tracking_number, destinations, routing_type, schedules)
+        
+        return f"Call route created successfully with ID: {route_id}"
+    except Exception as e:
+        return f"Error creating call route: {str(e)}"
+
+# Function to list all call routes
+def list_call_routes():
+    if not call_routes:
+        return "No call routes configured yet."
+    
+    output = "Call Routes:\n\n"
+    for route_id, route in call_routes.items():
+        output += f"Route ID: {route_id}\n"
+        output += f"  Name: {route['name']}\n"
+        output += f"  Tracking Number: {route['tracking_number']}\n"
+        output += f"  Routing Type: {route['routing_type']}\n"
+        output += f"  Destinations: {len(route['destinations'])}\n"
+        output += "\n"
+    
+    return output
+
+# Function to purchase a new tracking number
+def buy_tracking_number(area_code, name):
+    result = purchase_tracking_number(area_code, name)
+    
+    if result["success"]:
+        return f"Successfully purchased number: {result['number']}"
+    else:
+        return f"Failed to purchase number: {result['message']}"
+
+# Create the Gradio interface
+with gr.Blocks(title="Twilio Call Router (CallRail Alternative)") as app:
+    gr.Markdown("# Twilio Call Router\nA CallRail alternative built with Twilio and Gradio")
+    
+    with gr.Tab("Dashboard"):
+        gr.Markdown("## Dashboard")
+        dashboard_btn = gr.Button("Refresh Dashboard")
+        dashboard_output = gr.Textbox(label="Call Statistics", lines=10)
+        dashboard_btn.click(view_call_logs, inputs=[], outputs=dashboard_output)
+    
+    with gr.Tab("Call Routes"):
+        gr.Markdown("## Call Routes")
+        
+        with gr.Accordion("Create New Route"):
+            route_name = gr.Textbox(label="Route Name")
+            route_desc = gr.Textbox(label="Description")
+            tracking_num = gr.Textbox(label="Tracking Number (E.164 format)")
+            destinations = gr.Textbox(
+                label="Destinations (JSON format)", 
+                value="""[
+                    {"number": "+15551234567", "name": "Main Office"},
+                    {"number": "+15557654321", "name": "Sales Team"}
+                ]"""
+            )
+            routing_type = gr.Dropdown(
+                label="Routing Type", 
+                choices=["sequential", "simultaneous", "percentage", "time_based"],
+                value="simultaneous"
+            )
+            schedules = gr.Textbox(
+                label="Schedules (JSON format, for time_based routing)", 
+                value="""[
+                    {"days": ["monday", "tuesday", "wednesday", "thursday", "friday"], 
+                     "start_time": "09:00", 
+                     "end_time": "17:00", 
+                     "destination": "+15551234567"}
+                ]"""
+            )
+            create_route_btn = gr.Button("Create Route")
+            create_route_output = gr.Textbox(label="Result")
+            
+            create_route_btn.click(
+                create_call_route, 
+                inputs=[route_name, route_desc, tracking_num, destinations, routing_type, schedules], 
+                outputs=create_route_output
+            )
+        
+        list_routes_btn = gr.Button("List Routes")
+        routes_output = gr.Textbox(label="Routes", lines=10)
+        list_routes_btn.click(list_call_routes, inputs=[], outputs=routes_output)
+    
+    with gr.Tab("Numbers"):
+        gr.Markdown("## Phone Numbers")
+        
+        with gr.Row():
+            area_code = gr.Textbox(label="Area Code (e.g., 415)")
+            number_name = gr.Textbox(label="Friendly Name")
+        
+        buy_number_btn = gr.Button("Purchase Number")
+        number_output = gr.Textbox(label="Result")
+        
+        buy_number_btn.click(buy_tracking_number, inputs=[area_code, number_name], outputs=number_output)
+    
+    with gr.Tab("Call Logs"):
+        gr.Markdown("## Call Logs")
+        view_logs_btn = gr.Button("View Call Logs")
+        logs_output = gr.Textbox(label="Logs", lines=15)
+        view_logs_btn.click(view_call_logs, inputs=[], outputs=logs_output)
+
+# Add Flask routes for Twilio webhooks
+# Note: In a production environment, you'll need to expose these routes to the internet
+# For local development, you can use ngrok
+# These routes would be registered with your Flask app
+"""
+@app.route('/voice', methods=['POST'])
+def voice():
+    # Extract call details from the request
+    caller_id = request.values.get('From')
+    tracking_number = request.values.get('To')
+    call_sid = request.values.get('CallSid')
+    
+    # Generate and return TwiML
+    return generate_call_routing_twiml(tracking_number, caller_id, call_sid)
+
+@app.route('/sequential-route', methods=['POST'])
+def sequential_route():
+    # Handle sequential routing logic
+    call_sid = request.values.get('call_sid')
+    tracking_number = request.values.get('tracking_number')
+    index = int(request.values.get('index', 0))
+    
+    route_id = tracking_numbers[tracking_number]
+    route = call_routes[route_id]
+    
+    resp = VoiceResponse()
+    
+    if index >= len(route["destinations"]):
+        resp.say("We're sorry, but we were unable to connect your call. Please try again later.")
+        return str(resp)
+    
+    destination = route["destinations"][index]
+    dial = Dial(
+        caller_id=request.values.get('From'),
+        action=f"/sequential-next?call_sid={call_sid}&tracking_number={tracking_number}&index={index}",
+        timeout=20
+    )
+    dial.number(destination["number"])
+    resp.append(dial)
+    
+    return str(resp)
+
+@app.route('/sequential-next', methods=['POST'])
+def sequential_next():
+    # Handle the next step in sequential routing
+    call_sid = request.values.get('call_sid')
+    tracking_number = request.values.get('tracking_number')
+    index = int(request.values.get('index', 0))
+    dial_status = request.values.get('DialCallStatus')
+    
+    resp = VoiceResponse()
+    
+    if dial_status == 'completed':
+        # Call was answered and completed
+        return str(resp)
+    else:
+        # Call failed or wasn't answered, try the next destination
+        resp.redirect(f"/sequential-route?call_sid={call_sid}&tracking_number={tracking_number}&index={index + 1}")
+        return str(resp)
+
+@app.route('/call-status', methods=['POST'])
+def call_status():
+    # Update call status
+    call_sid = request.values.get('call_sid')
+    tracking_number = request.values.get('tracking_number')
+    dial_status = request.values.get('DialCallStatus')
+    duration = int(request.values.get('DialCallDuration', 0))
+    
+    # Find the call in logs and update its status
+    for log in call_logs:
+        if log['call_sid'] == call_sid:
+            log['status'] = dial_status
+            log['duration'] = duration
+            break
+    
+    resp = VoiceResponse()
+    if dial_status != 'completed':
+        resp.say("We're sorry, but we were unable to connect your call. Please try again later.")
+    
+    return str(resp)
+
+@app.route('/voicemail', methods=['POST'])
+def voicemail():
+    # Handle voicemail recording
+    call_sid = request.values.get('call_sid')
+    tracking_number = request.values.get('tracking_number')
+    recording_url = request.values.get('RecordingUrl')
+    
+    # Update the call log with the recording URL
+    for log in call_logs:
+        if log['call_sid'] == call_sid:
+            log['recording_url'] = recording_url
+            break
+    
+    resp = VoiceResponse()
+    resp.say("Thank you for your message. We'll get back to you as soon as possible.")
+    
+    return str(resp)
+"""
+
+# Launch the Gradio app
+if __name__ == "__main__":
+    app.launch()
+