import gradio as gr import logging import os import numpy as np from simple_salesforce import Salesforce from dotenv import load_dotenv # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load environment variables from .env file load_dotenv() # Load the .env file # Get the Salesforce credentials from environment variables sf_username = os.getenv("SF_USERNAME") sf_password = os.getenv("SF_PASSWORD") sf_security_token = os.getenv("SF_SECURITY_TOKEN") sf_instance_url = os.getenv("SF_INSTANCE_URL") # Check if the environment variables are correctly set if not sf_username or not sf_password or not sf_security_token or not sf_instance_url: logger.error("❌ Salesforce credentials are missing from environment variables!") raise ValueError("Salesforce credentials are not properly set.") # Salesforce connection try: sf = Salesforce( username=sf_username, password=sf_password, security_token=sf_security_token, instance_url=sf_instance_url ) logger.info("✅ Connected to Salesforce") except Exception as e: logger.error(f"❌ Salesforce connection failed: {str(e)}") raise # --- AI functions (These should be implemented in your project) --- def get_lead_score(stage, emails, meetings, close_gap, amount): # Example logic for lead score calculation return 0.8 * amount + 0.1 * emails + 0.1 * meetings def calculate_score(lead_score, emails, meetings, close_gap, amount): # Example AI score calculation return lead_score * 0.5 + 0.3 * emails + 0.2 * meetings def calculate_confidence(ai_score): # Ensure confidence is between 0 and 100 confidence = ai_score * 100 return min(max(confidence, 0), 100) # Clamp the value between 0 and 100 def calculate_risk(ai_score, confidence, emails, meetings): # Make sure to map the value to the Salesforce picklist options if ai_score > 0.75 and confidence > 75: return "Low" # Update to match Salesforce picklist value elif ai_score > 0.5: return "Medium" # Assuming "Medium" is another valid value else: return "High" # Update to match Salesforce picklist value def generate_recommendation(stage, emails, meetings, risk): # Example recommendation generation return "Proceed with caution" if risk == "High" else "Proceed" def explain_score(lead_score, ai_score, confidence, risk, stage, close_gap, emails, meetings): # Example explanation generation return f"Lead score based on {emails} emails and {meetings} meetings is {lead_score}. AI score is {ai_score}, confidence is {confidence}%. Risk: {risk}." # --- Push to Salesforce --- def push_to_salesforce(data: dict) -> str: try: response = sf.qualification_engine__c.create({ "Deal_Amount__c": data.get("amount"), "Stage__c": data.get("stage"), "Industry__c": data.get("industry"), "Emails_7_Days__c": data.get("emails"), "Meetings_30_Days__c": data.get("meetings"), "Days_Until_Close__c": data.get("gap"), "Rep_Feedback__c": data.get("feedback"), "Lead_Score__c": data.get("lead_score"), "AI_Score__c": data.get("score"), "Confidence__c": data.get("confidence"), "Risk_Level__c": data.get("risk"), "AI_Recommendation__c": data.get("recommendation"), "Explanation__c": data.get("explanation") }) return f"✅ Pushed to Salesforce with ID: {response['id']}" except Exception as e: return f"❌ Salesforce Error: {str(e)}" # --- Run Engine (Calculate and Push to Salesforce) --- def run_engine(amount, stage, industry, emails, meetings, close_gap, feedback=""): try: lead_score = get_lead_score(stage, emails, meetings, close_gap, amount) ai_score = calculate_score(lead_score, emails, meetings, close_gap, amount) confidence = calculate_confidence(ai_score) risk = calculate_risk(ai_score, confidence, emails, meetings) recommendation = generate_recommendation(stage, emails, meetings, risk) explanation = explain_score(lead_score, ai_score, confidence, risk, stage, close_gap, emails, meetings) sf_status = push_to_salesforce({ "amount": amount, "stage": stage, "industry": industry, "emails": emails, "meetings": meetings, "gap": close_gap, "feedback": feedback, "lead_score": lead_score, "score": ai_score, "confidence": confidence, "risk": risk, "recommendation": recommendation, "explanation": explanation }) return lead_score, ai_score, confidence, risk, recommendation, explanation, sf_status except Exception as e: return 0, 0, 0.0, "Error", "N/A", f"Error occurred: {str(e)}", f"❌ Error: {str(e)}" # Gradio UI with gr.Blocks(title="AI Deal Qualification Engine") as app: gr.Markdown("""

🤖 AI-Powered Deal Qualification Engine

Intelligently qualify sales deals using engagement and pipeline signals.

""", elem_id="header") with gr.Tab("📥 Input", elem_id="input-tab"): with gr.Row(): amount = gr.Number(label="💰 Deal Amount (USD)", value=50000, elem_id="deal-amount") stage = gr.Dropdown( ["Prospecting", "Proposal/Price Quote", "Negotiation", "Closed Won", "Closed Lost"], label="📊 Stage", elem_id="deal-stage" ) industry = gr.Textbox(label="🏭 Industry", value="Software", elem_id="industry") with gr.Row(): emails = gr.Number(label="✉️ Emails (Last 7 Days)", value=3, elem_id="emails") meetings = gr.Number(label="📅 Meetings (Last 30 Days)", value=2, elem_id="meetings") close_gap = gr.Number(label="📆 Days Until Close", value=14, elem_id="days-until-close") feedback = gr.Textbox(label="💬 Optional: Rep Feedback", placeholder="Add any qualitative insights...", elem_id="feedback") submit = gr.Button("🚀 Run AI Scoring", elem_id="submit-btn") with gr.Tab("📈 Results", elem_id="result-tab"): with gr.Accordion("AI Scoring Output", open=True): lead_score_out = gr.Number(label="🔢 Lead Score", interactive=False, elem_id="lead-score") ai_score_out = gr.Number(label="🌟 AI Score (0–100)", interactive=False, elem_id="ai-score") confidence_out = gr.Number(label="📐 Confidence", interactive=False, elem_id="confidence") risk_out = gr.Textbox(label="⚠️ Risk Level", lines=1, interactive=False, elem_id="risk-level") reco_out = gr.Textbox(label="💡 AI Recommendation", lines=2, interactive=False, elem_id="ai-recommendation") explain_out = gr.Textbox(label="🧠 Explanation", lines=5, interactive=False, elem_id="explanation") status = gr.Markdown("", elem_id="status") submit.click( fn=run_engine, inputs=[amount, stage, industry, emails, meetings, close_gap, feedback], outputs=[lead_score_out, ai_score_out, confidence_out, risk_out, reco_out, explain_out, status] ) app.launch(share=True)