Spaces:
Sleeping
Sleeping
| 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}") |