import gradio as gr import pandas as pd from simple_salesforce import Salesforce import io from datetime import datetime import logging # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Global variables to store connection sf_connection = None available_objects = [] def connect_to_salesforce(username, password, security_token, sandbox): """Connect to Salesforce""" global sf_connection, available_objects try: domain = 'test' if sandbox else None sf_connection = Salesforce( username=username, password=password, security_token=security_token, domain=domain ) # Test connection and get objects common_objects = ['Account', 'Contact', 'Lead', 'Opportunity', 'Case'] available_objects = [] for obj_name in common_objects: try: obj = getattr(sf_connection, obj_name) obj.describe() available_objects.append(obj_name) except: continue if not available_objects: available_objects = ['Account', 'Contact', 'Lead'] return f"✅ Successfully connected to Salesforce as {username}\nAvailable objects: {', '.join(available_objects)}" except Exception as e: sf_connection = None available_objects = [] error_msg = str(e) if "INVALID_LOGIN" in error_msg: return "❌ Invalid credentials. Please check your username, password, and security token." elif "API_DISABLED_FOR_ORG" in error_msg: return "❌ API access is disabled for your organization. Please contact your Salesforce admin." elif "LOGIN_MUST_USE_SECURITY_TOKEN" in error_msg: return "❌ Security token required. Please append your security token to your password." else: return f"❌ Connection failed: {error_msg}" def upload_data_to_salesforce(file, object_name, operation): """Upload data to Salesforce""" global sf_connection if not sf_connection: return "❌ Please connect to Salesforce first", None if not file or not object_name: return "❌ Please select a file and object", None try: # Read the uploaded file if file.name.endswith('.csv'): df = pd.read_csv(file.name) elif file.name.endswith(('.xlsx', '.xls')): df = pd.read_excel(file.name) else: return "❌ Please upload a CSV or Excel file", None if df.empty: return "❌ The uploaded file is empty", None # Get Salesforce object sf_object = getattr(sf_connection, object_name) # Prepare data for upload records = df.to_dict('records') # Clean data - remove NaN values cleaned_records = [] for record in records: cleaned_record = {k: v for k, v in record.items() if pd.notna(v)} cleaned_records.append(cleaned_record) # Perform operation if operation == "Insert": result = sf_object.bulk.insert(cleaned_records) elif operation == "Update": result = sf_object.bulk.update(cleaned_records) else: # Upsert return "❌ Upsert operation requires additional configuration", None # Process results success_count = sum(1 for r in result if r.get('success')) error_count = len(result) - success_count summary = f"✅ Operation completed!\n" summary += f"📊 Total records: {len(records)}\n" summary += f"✅ Successful: {success_count}\n" summary += f"❌ Failed: {error_count}\n" # Create results file results_df = pd.DataFrame(result) results_file = f"salesforce_upload_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" results_df.to_csv(results_file, index=False) return summary, results_file except Exception as e: logger.error(f"Upload error: {str(e)}") return f"❌ Error: {str(e)}", None def export_data_from_salesforce(object_name, record_limit): """Export data from Salesforce""" global sf_connection if not sf_connection: return "❌ Please connect to Salesforce first", None if not object_name: return "❌ Please select an object", None try: # Get object metadata to find some fields obj = getattr(sf_connection, object_name) metadata = obj.describe() # Get first 10 fields fields = [field['name'] for field in metadata['fields'][:10]] fields_str = ', '.join(fields) # Build and execute query query = f"SELECT {fields_str} FROM {object_name} LIMIT {record_limit}" result = sf_connection.query_all(query) records = result['records'] if not records: return "❌ No records found", None # Convert to DataFrame and clean df = pd.DataFrame(records) if 'attributes' in df.columns: df = df.drop('attributes', axis=1) # Create export file export_file = f"salesforce_export_{object_name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv" df.to_csv(export_file, index=False) summary = f"✅ Export completed!\n" summary += f"📊 Records exported: {len(records)}\n" summary += f"📋 Fields: {', '.join(fields)}\n" return summary, export_file except Exception as e: logger.error(f"Export error: {str(e)}") return f"❌ Error: {str(e)}", None # Create the Gradio interface with gr.Blocks(title="Salesforce Data Loader", theme=gr.themes.Default()) as demo: gr.Markdown(""" # 🚀 Salesforce Data Loader A simple tool to upload and download data from Salesforce. """) with gr.Tab("🔐 Connect"): gr.Markdown("### Connect to Salesforce") username = gr.Textbox(label="Username", placeholder="your.email@company.com") password = gr.Textbox(label="Password", type="password") security_token = gr.Textbox(label="Security Token", type="password") sandbox = gr.Checkbox(label="Sandbox Environment") connect_btn = gr.Button("🔗 Connect", variant="primary") connection_status = gr.Textbox(label="Connection Status", interactive=False) connect_btn.click( fn=connect_to_salesforce, inputs=[username, password, security_token, sandbox], outputs=[connection_status] ) with gr.Tab("📤 Upload"): gr.Markdown("### Upload CSV/Excel data to Salesforce") file_upload = gr.File(label="Upload CSV or Excel file") upload_object = gr.Dropdown( label="Salesforce Object", choices=["Account", "Contact", "Lead", "Opportunity", "Case"], value="Contact" ) upload_operation = gr.Dropdown( label="Operation", choices=["Insert", "Update"], value="Insert" ) upload_btn = gr.Button("📤 Upload Data", variant="primary") upload_results = gr.Textbox(label="Upload Results", interactive=False) download_results = gr.File(label="Download Results") upload_btn.click( fn=upload_data_to_salesforce, inputs=[file_upload, upload_object, upload_operation], outputs=[upload_results, download_results] ) with gr.Tab("📥 Export"): gr.Markdown("### Export data from Salesforce") export_object = gr.Dropdown( label="Salesforce Object", choices=["Account", "Contact", "Lead", "Opportunity", "Case"], value="Account" ) export_limit = gr.Slider( label="Record Limit", minimum=100, maximum=10000, value=1000, step=100 ) export_btn = gr.Button("📥 Export Data", variant="primary") export_results = gr.Textbox(label="Export Results", interactive=False) download_export = gr.File(label="Download Export") export_btn.click( fn=export_data_from_salesforce, inputs=[export_object, export_limit], outputs=[export_results, download_export] ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860)