Upload 3 files

app.py

@@ -0,0 +1,241 @@
+import os
+import random
+import gradio as gr
+from datetime import datetime
+from transformers import pipeline
+from simple_salesforce import Salesforce, SalesforceLogin
+from dotenv import load_dotenv
+import xml.etree.ElementTree as ET
+
+# ---------- 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"
+}
+
+# ---------- Load QnA Model ----------
+qa_pipeline = pipeline("text2text-generation", model="google/flan-t5-large")
+
+# ---------- 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}")
+
+# ---------- Salesforce Connection ----------
+try:
+    session_id, instance = SalesforceLogin(
+        username=SF_USERNAME,
+        password=SF_PASSWORD,
+        security_token=SF_SECURITY_TOKEN
+    )
+    sf = Salesforce(instance=instance, session_id=session_id)
+    print("✅ Connected to Salesforce successfully")
+except Exception as e:
+    sf = None
+    print(f"❌ Failed to connect to Salesforce: {e}")
+
+# ---------- 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]
+
+    review_data = {
+        "Name": f"Review_{issue_type}",
+        "CodeSnippet__c": code,
+        "IssueType__c": issue_type,
+        "Suggestion__c": suggestion,
+        "Severity__c": severity
+    }
+
+    log_to_console(review_data, "Code Review")
+
+    if sf:
+        try:
+            result = sf.CodeReviewResult__c.create(review_data)
+            if result.get("success"):
+                log_to_console({"Salesforce Record ID": result["id"]}, "Salesforce Create")
+            else:
+                log_to_console(result, "Salesforce Error")
+        except Exception as e:
+            log_to_console({"Salesforce Exception": str(e)}, "Salesforce Error")
+    else:
+        log_to_console("Salesforce not connected.", "Salesforce Error")
+
+    return issue_type, suggestion, severity
+
+# ---------- Metadata Validator ----------
+def validate_metadata(metadata, admin_id=None):
+    if not metadata.strip():
+        return "No metadata provided.", "", ""
+
+    mtype = "Field"
+    issue = "Unknown"
+    recommendation = "No recommendation found."
+
+    try:
+        root = ET.fromstring(metadata)
+        description_found = any(elem.tag.endswith('description') for elem in root)
+
+        if not description_found:
+            issue = "Missing description"
+            recommendation = "Add a meaningful <description> to improve maintainability and clarity."
+        else:
+            issue = "Unused field detected"
+            recommendation = "Remove it to improve performance or document its purpose."
+    except Exception as e:
+        issue = "Invalid XML"
+        recommendation = f"Could not parse metadata XML. Error: {str(e)}"
+
+    log_data = {
+        "Name": f"MetadataLog_{mtype}",
+        "MetadataType__c": mtype,
+        "IssueDescription__c": issue,
+        "Recommendation__c": recommendation,
+        "Status__c": "Open"
+    }
+
+    if admin_id:
+        log_data["Admin__c"] = admin_id
+
+    log_to_console(log_data, "Metadata Validation")
+
+    if sf:
+        try:
+            result = sf.MetadataAuditLog__c.create(log_data)
+            if result.get("success"):
+                log_to_console({"Salesforce MetadataAuditLog Record ID": result["id"]}, "Salesforce Create")
+            else:
+                log_to_console(result, "Salesforce Metadata Error")
+        except Exception as e:
+            log_to_console({"Salesforce Exception": str(e)}, "Salesforce Error")
+    else:
+        log_to_console("Salesforce not connected.", "Salesforce Error")
+
+    return mtype, issue, recommendation
+
+# ---------- Salesforce Chatbot (Final Improved Prompt) ----------
+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", "aura", "visualforce", "salesforce", "governor limits",
+        "dml", "metadata", "batch apex", "queueable", "future method", "api", "sfdc", "heap", "limits"
+    ]
+
+    if not any(keyword.lower() in query.lower() for keyword in salesforce_keywords):
+        return "Please ask a Salesforce-related question."
+
+    history_summary = "\n".join([f"User: {q}\nAssistant: {a}" for q, a in conversation_history[-4:]])
+
+    prompt = f"""
+You are a certified Salesforce developer and architect. You answer Salesforce questions with 100% accuracy and clarity, using official platform knowledge.
+
+Guidelines:
+- Always give answers that are at least 2 full lines long.
+- When asked about Apex limits (e.g., DML, SOQL, heap), always:
+  - Mention the numeric limit (e.g., 150 DML statements)
+  - Include the transaction context (sync/async)
+  - Suggest using monitoring methods like Limits.getDMLStatements()
+- Avoid making up numbers — if unsure, suggest Trailhead or official docs.
+- Format clearly, using bullet points or code if helpful.
+
+Conversation History:
+{history_summary}
+
+User: {query.strip()}
+Assistant:
+"""
+
+    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("Assistant:"):
+            output = output.replace("Assistant:", "").strip()
+
+        if len(output.split()) < 20:
+            output += "\n\nPlease refer to: https://developer.salesforce.com/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)}"
+
+# ---------- Gradio UI ----------
+with gr.Blocks(theme=gr.themes.Soft()) as demo:
+    gr.Markdown("# 🤖 Advanced Salesforce AI Code Review & Chatbot")
+
+    with gr.Tab("Code Review"):
+        code_input = gr.Textbox(label="Apex / LWC Code", lines=8, placeholder="Enter your Apex or LWC 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])
+
+    with gr.Tab("Metadata Validation"):
+        metadata_input = gr.Textbox(label="Metadata XML", lines=8, placeholder="Enter your metadata XML here")
+        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])
+
+    with gr.Tab("Salesforce Chatbot"):
+        chatbot_output = gr.Chatbot(label="Conversation History", height=400)
+        query_input = gr.Textbox(label="Your Question", placeholder="e.g., How many DML operations are allowed in Apex?")
+        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."
+            response = salesforce_chatbot(query, chat_history)
+            chat_history.append((query, response))
+            return chat_history, ""
+
+        def clear_chat():
+            global conversation_history
+            conversation_history = []
+            return [], ""
+
+        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])
+
+if __name__ == "__main__":
+    demo.launch()
+how can i call an external api from apex?

env

@@ -0,0 +1,5 @@
+SF_USERNAME= aicodereview@sathkrutha.com
+SF_PASSWORD=aicode@123
+SF_SECURITY_TOKEN=AkaHhoPeAZFNNSvw8zuDPxAaF
+SF_DOMAIN=LOGIN
+DEVELOPER_ID=005NS00000Sn2q9

requirements.txt

@@ -0,0 +1,4 @@
+gradio>=4.0.0
+transformers>=4.35.0
+torch>=2.0.0
+simple-salesforce>=1.12.3