Update app.py

app.py

@@ -1,18 +1,135 @@
-# [All your existing imports]
+import os
 import random
 import gradio as gr
 from datetime import datetime
 from transformers import pipeline
 from simple_salesforce import Salesforce, SalesforceLogin, SFType
+from dotenv import load_dotenv
+
+# ---------- Load Environment Variables ----------
+load_dotenv()
+SF_USERNAME = os.getenv("SF_USERNAME")
+SF_PASSWORD = os.getenv("SF_PASSWORD")
+SF_SECURITY_TOKEN = os.getenv("SF_SECURITY_TOKEN")
+
+# ---------- Label Mapping ----------
+label_to_issue_type = {
+    "LABEL_0": "Performance",
+    "LABEL_1": "Error",
+    "LABEL_2": "Security",
+    "LABEL_3": "Best Practice"
+}
+
+suggestions = {
+    "Performance": "Consider optimizing loops and database access. Use collections to reduce SOQL queries.",
+    "Error": "Add proper error handling and null checks. Use try-catch blocks effectively.",
+    "Security": "Avoid dynamic SOQL. Use binding variables to prevent SOQL injection.",
+    "Best Practice": "Refactor for readability and use bulk-safe patterns, such as processing records in batches."
+}
+
+severities = {
+    "Performance": "Medium",
+    "Error": "High",
+    "Security": "High",
+    "Best Practice": "Low"
+}
+
+# ---------- Mock Salesforce Knowledge Base ----------
+salesforce_knowledge_base = {
+    "governor limits soql": "...",  # Truncated for brevity
+    "governor limits dml": "...",
+    "bulkify apex trigger": "...",
+    "soql injection": "...",
+    "lwc best practices": "...",
+    "batch apex": "..."
+}
+
+# ---------- Load QnA Model ----------
+try:
+    qa_pipeline = pipeline("text2text-generation", model="google/flan-t5-large")
+except Exception as e:
+    print(f"Model loading error: {e}. Falling back to flan-t5-base.")
+    qa_pipeline = pipeline("text2text-generation", model="google/flan-t5-base")
+
+# ---------- Local Logging ----------
+def log_to_console(data, log_type):
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+    print(f"[{timestamp}] {log_type} Log: {data}")
+
+# ---------- Code Analyzer ----------
+def analyze_code(code):
+    if not code.strip():
+        return "No code provided.", "", ""
+    label = random.choice(list(label_to_issue_type.keys()))
+    issue_type = label_to_issue_type[label]
+    suggestion = suggestions[issue_type]
+    severity = severities[issue_type]
+    log_data = {
+        "Name": f"Review_{issue_type}",
+        "CodeSnippet": code,
+        "IssueType": issue_type,
+        "Suggestion": suggestion,
+        "Severity": severity
+    }
+    log_to_console(log_data, "Code Review")
+    return issue_type, suggestion, severity
+
+# ---------- Metadata Validator ----------
+def validate_metadata(metadata):
+    if not metadata.strip():
+        return "No metadata provided.", "", ""
+    mtype = "Field"
+    issue = "Unused field detected"
+    recommendation = "Remove it to improve performance or document its purpose."
+    log_data = {
+        "Name": f"MetadataLog_{mtype}",
+        "MetadataType": mtype,
+        "IssueDescription": issue,
+        "Recommendation": recommendation,
+        "Status": "Open"
+    }
+    log_to_console(log_data, "Metadata Validation")
+    return mtype, issue, recommendation
+
+# ---------- Salesforce Chatbot ----------
+conversation_history = []
+def salesforce_chatbot(query, history=[]):
+    global conversation_history
+    if not query.strip():
+        return "Please provide a valid Salesforce-related question."
+    salesforce_keywords = [ "apex", "soql", "trigger", "lwc", "governor limits", "dml", "metadata", "batch apex", "api", "profile", "workflow", "custom object", "record type", "sfdc", "force.com" ]
+    if not any(keyword in query.lower() for keyword in salesforce_keywords):
+        return "Please ask a Salesforce-related question (e.g., about Apex, SOQL, LWC, or Salesforce platform features)."
+    query_key = query.lower().strip()
+    for kb_key, kb_answer in salesforce_knowledge_base.items():
+        if kb_key in query_key:
+            conversation_history.append((query, kb_answer))
+            conversation_history = conversation_history[-6:]
+            log_to_console({"Question": query, "Answer": kb_answer}, "Chatbot Query")
+            return kb_answer
+    history_summary = "\n".join([f"User: {q}\nAssistant: {a}" for q, a in conversation_history[-4:]])
+    prompt = f"""
+You are an expert Salesforce developer...
+Conversation History:
+{history_summary}
+Question: {query.strip()}
+Answer:
+"""
+    try:
+        result = qa_pipeline(prompt, max_new_tokens=1024, do_sample=False, temperature=0.1, top_k=50)
+        output = result[0]["generated_text"].strip()
+        if output.startswith("Answer:"):
+            output = output[7:].strip()
+        if len(output) < 20:
+            output = f"I'm sorry, I couldn't find a precise answer for '{query}'. Please check Salesforce docs."
+        conversation_history.append((query, output))
+        conversation_history = conversation_history[-6:]
+        log_to_console({"Question": query, "Answer": output}, "Chatbot Query")
+        return output
+    except Exception as e:
+        return f"⚠️ Error generating response: {str(e)}"
 
-# [Your existing mappings, knowledge base, and model loading code remain unchanged]
-
-# ---------- Salesforce Connection Setup ----------
-# Replace these with your actual Salesforce credentials or use environment variables for security
-SF_USERNAME = "your_username@example.com"
-SF_PASSWORD = "your_password"
-SF_SECURITY_TOKEN = "your_security_token"
-
+# ---------- Salesforce Connection ----------
 try:
     session_id, instance = SalesforceLogin(
         username=SF_USERNAME,
@@ -25,59 +142,51 @@ except Exception as e:
     sf = None
     print(f"❌ Failed to connect to Salesforce: {e}")
 
-# ---------- Function to Create Salesforce Record ----------
+# ---------- Create Salesforce Record ----------
 def create_salesforce_record(object_type, field_name, field_value):
     if not sf:
         return "Salesforce connection not established."
-
     try:
         sobject = SFType(object_type, sf.session_id, sf.sf_instance)
         result = sobject.create({field_name: field_value})
         if result.get("success"):
-            return f"✅ Record created successfully in {object_type} with ID: {result['id']}"
+            return f"✅ Record created in {object_type} with ID: {result['id']}"
         else:
             return f"❌ Failed to create record: {result}"
     except Exception as e:
         return f"⚠️ Error: {str(e)}"
 
-# [Your existing functions: analyze_code, validate_metadata, salesforce_chatbot, etc. remain unchanged]
-
 # ---------- Gradio UI ----------
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
     gr.Markdown("# 🤖 Advanced Salesforce AI Code Review & Chatbot")
 
-    # Tab 1: Code Review
     with gr.Tab("Code Review"):
-        code_input = gr.Textbox(label="Apex / LWC Code", lines=8, placeholder="Enter your Apex or LWC code here")
+        code_input = gr.Textbox(label="Apex / LWC Code", lines=8, placeholder="Enter code here")
         issue_type = gr.Textbox(label="Issue Type")
         suggestion = gr.Textbox(label="AI Suggestion")
         severity = gr.Textbox(label="Severity")
         code_button = gr.Button("Analyze Code")
         code_button.click(analyze_code, inputs=code_input, outputs=[issue_type, suggestion, severity])
 
-    # Tab 2: Metadata Validation
     with gr.Tab("Metadata Validation"):
-        metadata_input = gr.Textbox(label="Metadata XML", lines=8, placeholder="Enter your metadata XML here")
+        metadata_input = gr.Textbox(label="Metadata XML", lines=8)
         mtype = gr.Textbox(label="Type")
         issue = gr.Textbox(label="Issue")
         recommendation = gr.Textbox(label="Recommendation")
         metadata_button = gr.Button("Validate Metadata")
         metadata_button.click(validate_metadata, inputs=metadata_input, outputs=[mtype, issue, recommendation])
 
-    # Tab 3: Salesforce Chatbot
     with gr.Tab("Salesforce Chatbot"):
-        gr.Markdown("### Ask a Salesforce Question\nGet expert answers on Apex, SOQL, LWC, and more!")
         chatbot_output = gr.Chatbot(label="Conversation History", height=400)
-        query_input = gr.Textbox(label="Your Question", placeholder="e.g., How do I bulkify an Apex trigger?")
+        query_input = gr.Textbox(label="Your Question")
         with gr.Row():
             chatbot_button = gr.Button("Ask")
             clear_button = gr.Button("Clear Chat")
-
         chat_state = gr.State(value=[])
 
         def update_chatbot(query, chat_history):
             if not query.strip():
-                return chat_history, "Please enter a valid question."
+                return chat_history, "Please enter a question."
             response = salesforce_chatbot(query, chat_history)
             chat_history.append((query, response))
             return chat_history, ""
@@ -90,17 +199,14 @@ with gr.Blocks(theme=gr.themes.Soft()) as demo:
         chatbot_button.click(fn=update_chatbot, inputs=[query_input, chat_state], outputs=[chatbot_output, query_input])
         clear_button.click(fn=clear_chat, inputs=None, outputs=[chatbot_output, query_input])
 
-    # Tab 4: Create Salesforce Record
     with gr.Tab("Create Salesforce Record"):
         gr.Markdown("### 📇 Create a New Salesforce Record")
-        object_type = gr.Textbox(label="Salesforce Object API Name", placeholder="e.g., Account")
+        object_type = gr.Textbox(label="Object API Name", placeholder="e.g., Account")
         field_name = gr.Textbox(label="Field API Name", placeholder="e.g., Name")
         field_value = gr.Textbox(label="Field Value", placeholder="e.g., Test Company")
         create_button = gr.Button("Create Record")
         record_result = gr.Textbox(label="Result")
-
         create_button.click(create_salesforce_record, inputs=[object_type, field_name, field_value], outputs=record_result)
 
-# ---------- Start UI ----------
 if __name__ == "__main__":
     demo.launch()