import gradio as gr from simple_salesforce import Salesforce import os import logging from datetime import datetime import pytz # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Salesforce credentials (using environment variables for security, with defaults) SF_USERNAME = os.getenv('SF_USERNAME', 'Ai@Coach.com') SF_PASSWORD = os.getenv('SF_PASSWORD', 'Teja90325@') SF_SECURITY_TOKEN = os.getenv('SF_SECURITY_TOKEN', 'clceSdBgQ30Rx9BSC66gAcRx') SF_DOMAIN = os.getenv('SF_DOMAIN', 'login') # Initialize Salesforce connection try: sf = Salesforce( username=SF_USERNAME, password=SF_PASSWORD, security_token=SF_SECURITY_TOKEN, domain=SF_DOMAIN ) logger.info("Successfully connected to Salesforce") except Exception as e: logger.error(f"Failed to connect to Salesforce: {e}") sf = None def store_in_salesforce(project_id, role, checklist, suggestions, reflection): """Store the generated data in Salesforce.""" if sf is None: logger.error("Salesforce connection not initialized") return "Error: Salesforce connection not initialized" try: # Map Hugging Face role "Supervisor" to Salesforce role "SiteSupervisor" salesforce_role = "SiteSupervisor" if role == "Supervisor" else role # Step 1: Find the Project__c record by Name (escape special characters in project_id) project_id_safe = project_id.replace("'", "''") # Escape single quotes for SOQL project_query = f""" SELECT Id FROM Project__c WHERE Name = '{project_id_safe}' LIMIT 1 """ project_result = sf.query(project_query) if not project_result['records']: logger.error(f"No Project__c record found with Name {project_id}") return f"Error: No Project__c record found with Name {project_id}" project_sf_id = project_result['records'][0]['Id'] logger.info(f"Found Project__c record with Id {project_sf_id} for Name {project_id}") # Step 2: Find the Supervisor__c record by Role__c (mapped to SiteSupervisor) supervisor_query = f""" SELECT Id, Name FROM Supervisor__c WHERE Role__c = '{salesforce_role}' LIMIT 1 """ supervisor_result = sf.query(supervisor_query) if not supervisor_result['records']: logger.error(f"No Supervisor__c record found with Role__c {salesforce_role}") return f"Error: No Supervisor__c record found with Role__c {salesforce_role}" supervisor_sf_id = supervisor_result['records'][0]['Id'] supervisor_name = supervisor_result['records'][0]['Name'] logger.info(f"Found Supervisor__c record with Id {supervisor_sf_id} and Name {supervisor_name} for Role__c {salesforce_role}") # Step 3: Check for existing Supervisor_AI_Coaching__c record coaching_query = f""" SELECT Id FROM Supervisor_AI_Coaching__c WHERE Project_ID__c = '{project_sf_id}' AND Supervisor_ID__c = '{supervisor_sf_id}' LIMIT 1 """ coaching_result = sf.query(coaching_query) logger.info(f"Found {len(coaching_result['records'])} existing Supervisor_AI_Coaching__c records") # Step 4: Calculate Engagement Score and KPI Flag engagement_score = 50 if checklist and suggestions: engagement_score = 75 elif checklist or suggestions: engagement_score = 60 kpi_flag = 'delay' in suggestions.lower() or 'issue' in suggestions.lower() # Step 5: Prepare the data to store # Format the DateTime field for Salesforce (ISO 8601 with 'Z' for UTC) last_refresh_date = datetime.utcnow().replace(tzinfo=pytz.UTC).isoformat() coaching_data = { 'Project_ID__c': project_sf_id, 'Supervisor_ID__c': supervisor_sf_id, 'Daily_Checklist__c': checklist, 'Suggested_Tips__c': suggestions, 'Reflection_Log__c': reflection, 'Engagement_Score__c': engagement_score, 'KPI_Flag__c': kpi_flag, 'Last_Refresh_Date__c': last_refresh_date } # Log the data being stored for verification logger.info(f"Data to be stored in Supervisor_AI_Coaching__c: {coaching_data}") # Step 6: Update or create the record if coaching_result['records']: record_id = coaching_result['records'][0]['Id'] sf.Supervisor_AI_Coaching__c.update(record_id, coaching_data) logger.info(f"Updated Supervisor_AI_Coaching__c record with Id {record_id}") else: result = sf.Supervisor_AI_Coaching__c.create(coaching_data) if not result['success']: logger.error(f"Failed to create Supervisor_AI_Coaching__c record: {result['errors']}") return f"Error creating Supervisor_AI_Coaching__c record: {result['errors']}" logger.info(f"Created new Supervisor_AI_Coaching__c record with Id {result['id']}") return "Successfully stored in Salesforce" except Exception as e: logger.error(f"Error storing in Salesforce: {e}") return f"Error storing in Salesforce: {e}" def generate_outputs(role, project_id, reflection, milestone): """Generate checklist and suggestions using rule-based logic for speed.""" # Input validation if not all([role, project_id, reflection, milestone]): logger.error("All fields are required") return "", "", "Error: All fields are required." # Checklist: Generate from milestone input milestones_list = "\n- ".join([m.strip() for m in milestone.split(",") if m.strip()]) if not milestones_list: logger.error("At least one valid milestone is required") return "", "", "Error: At least one valid milestone is required." checklist = f"- {milestones_list}" # Generate suggestions based on reflection (rule-based) suggestions_list = [] reflection_lower = reflection.lower() if "delays" in reflection_lower: suggestions_list.extend(["Adjust timelines for delays.", "Communicate with stakeholders."]) if "weather" in reflection_lower: suggestions_list.extend(["Ensure rain gear availability.", "Monitor weather updates."]) if "equipment" in reflection_lower: suggestions_list.extend(["Inspect equipment.", "Schedule maintenance."]) if any(keyword in reflection_lower for keyword in ["information", "plan", "specification", "management"]): suggestions_list.extend(["Review the project information plan.", "Ensure all team members are updated."]) if any(keyword in reflection_lower for keyword in ["personal", "experience", "learning", "feeling", "thinking"]): suggestions_list.extend(["Schedule a one-on-one meeting to discuss concerns.", "Provide additional training if needed."]) suggestions = "\n- ".join(suggestions_list) if suggestions_list else "No specific suggestions." # Store the generated outputs in Salesforce storage_result = store_in_salesforce(project_id, role, checklist, suggestions, reflection) return checklist, suggestions, storage_result def create_interface(): """Create Gradio interface for manual testing.""" with gr.Blocks() as demo: gr.Markdown("### AI Coach for Site Supervisors") with gr.Row(): role = gr.Dropdown(choices=["Site Supervisor", "Foreman", "Project Manager"], label="Role", value="Supervisor") project_id = gr.Textbox(label="Project ID", placeholder="e.g., PROJ-003") reflection = gr.Textbox( label="Reflection Log", lines=3, placeholder="Reflect on your personal response to what you have been experiencing, learning, doing, feeling, and thinking..." ) milestone = gr.Textbox( label="Milestone", lines=2, placeholder="Enter milestones (comma-separated), e.g., Foundation complete, Framing started..." ) submit = gr.Button("Generate") checklist_output = gr.Textbox(label="Daily Checklist", lines=4) suggestions_output = gr.Textbox(label="Suggested Tips", lines=4) submit.click( fn=generate_outputs, inputs=[role, project_id, reflection, milestone], outputs=[checklist_output, suggestions_output] ) return demo if __name__ == "__main__": try: demo = create_interface() demo.launch(server_name="0.0.0.0", server_port=7860, share=False) except Exception as e: logger.error(f"Error launching Gradio interface: {e}")