Update utils.py

utils.py

@@ -1,166 +1,124 @@
+import gradio as gr
 import os
-import fitz  # PyMuPDF
-import json
-import re
-import difflib
-from simple_salesforce import Salesforce
-from dotenv import load_dotenv
-
-# Load .env variables
-load_dotenv()
-
-def get_salesforce_client():
-    try:
-        sf = Salesforce(
-            username=os.getenv("SF_USERNAME"),
-            password=os.getenv("SF_PASSWORD"),
-            security_token=os.getenv("SF_SECURITY_TOKEN"),
-            domain=os.getenv("SF_DOMAIN").replace("https://", "").replace(".salesforce.com", "")
-        )
-        return sf, None
-    except Exception as e:
-        return None, str(e)
-
-def get_salesforce_objects(sf):
-    try:
-        desc = sf.describe()
-        object_names = []
-        for obj in desc['sobjects']:
-            if not obj['deprecatedAndHidden']:  # Skip hidden/deprecated
-                object_names.append(obj['name'])
-        object_names.sort()
-        return object_names, None
-    except Exception as e:
-        return [], str(e)
-
-def get_object_fields(sf, object_name):
-    try:
-        metadata = sf.__getattr__(object_name).describe()
-        fields = [field['name'] for field in metadata['fields']]
-        return fields, None
-    except Exception as e:
-        return [], str(e)
-
-def extract_text_from_pdf(pdf_path):
-    try:
-        doc = fitz.open(pdf_path)
-        text = ""
-        for page in doc:
-            text += page.get_text()
-        return text.strip(), None
-    except Exception as e:
-        return None, str(e)
-
-# ✅ Extract key-value pairs smartly from the PDF text
-def extract_key_value_pairs(pdf_path):
-    try:
-        doc = fitz.open(pdf_path)
-        text = ""
-        for page in doc:
-            text += page.get_text()
-
-        keys = []
-        values = []
-
-        # Pattern 1: Direct "Key: Value"
-        lines = text.splitlines()
-        for line in lines:
-            match = re.match(r"^([A-Z][A-Za-z0-9 ()/_\-]{3,50})\s*[:\-]\s*(.+)$", line.strip())
-            if match:
-                key = match.group(1).strip()
-                value = match.group(2).strip()
-                if len(value) > 1:
-                    keys.append(key)
-                    values.append(value)
-
-        # Pattern 2: Special extractions
-
-        # Total Agreement Value
-        match_val = re.search(r"Total Agreement Value[^\$]*\$\s?([\d,]+(?:\.\d{2})?)", text, re.IGNORECASE)
-        if match_val:
-            keys.append("Total Agreement Value")
-            numeric_value = match_val.group(1).replace(",", "")
-            values.append(numeric_value)
-
-        # Agreement Name
-        match_name = re.search(r"Agreement\s+(MSA\s+[A-Za-z0-9 _\-]+)", text, re.IGNORECASE)
-        if match_name:
-            keys.append("Agreement Name")
-            values.append(match_name.group(1).strip())
-
-        # Agreement Start and End Dates
-        match_dates = re.search(
-            r"effective as of\s*([0-9]{1,2}/[0-9]{1,2}/[0-9]{2,4}).*?until\s*[<\(]?([0-9]{1,2}/[0-9]{1,2}/[0-9]{2,4})",
-            text, re.IGNORECASE | re.DOTALL
-        )
-        if match_dates:
-            keys.append("Agreement Start Date")
-            values.append(match_dates.group(1).strip())
-            keys.append("Agreement End Date")
-            values.append(match_dates.group(2).strip())
-
-        return [{"keys": keys, "values": values}], None
-    except Exception as e:
-        return None, str(e)
-
-# ✅ FIXED: Use original key as mapping key (not value)
-def map_fields(extracted_data, object_fields):
-    try:
-        mappings = {}
-        confidence_scores = {}
-        keys = extracted_data[0]["keys"]
-        values = extracted_data[0]["values"]
-
-        for key, value in zip(keys, values):
-            if key.lower() in ["name", "email"]:
-                continue  # Skip these
-
-            best_match = difflib.get_close_matches(key, object_fields, n=1, cutoff=0.0)
-            if best_match:
-                matched_field = best_match[0]
-                confidence = difflib.SequenceMatcher(None, key.lower(), matched_field.lower()).ratio()
-                mappings[key] = matched_field
-                confidence_scores[key] = round(confidence, 2)
-            else:
-                mappings[key] = object_fields[0]
-                confidence_scores[key] = 0.0
-
-        return mappings, confidence_scores, None
-    except Exception as e:
-        return None, None, str(e)
-
-def create_record(sf, object_name, data):
-    try:
-        result = sf.__getattr__(object_name).create(data)
-        return result.get("id", "Unknown ID"), None
-    except Exception as e:
-        return None, str(e)
-
-def attach_pdf(sf, record_id, pdf_path):
-    try:
-        with open(pdf_path, "rb") as f:
-            body = f.read()
-        content_version = sf.ContentVersion.create({
-            "Title": os.path.basename(pdf_path),
-            "PathOnClient": os.path.basename(pdf_path),
-            "VersionData": body.encode("base64") if isinstance(body, str) else body
-        })
-
-        content_document_id = sf.query(
-            f"SELECT ContentDocumentId FROM ContentVersion WHERE Id = '{content_version['id']}'"
-        )["records"][0]["ContentDocumentId"]
-
-        sf.ContentDocumentLink.create({
-            "ContentDocumentId": content_document_id,
-            "LinkedEntityId": record_id,
-            "ShareType": "V"
-        })
-
-        return "PDF attached successfully", None
-    except Exception as e:
-        return None, str(e)
-
-def log_failure(pdf_path, object_name, error):
-    with open("failures.json", "a") as f:
-        json.dump({"pdf": pdf_path, "object": object_name, "error": error}, f)
-        f.write("\n")
+from utils import (
+    get_salesforce_client,
+    get_salesforce_objects,
+    get_object_fields,
+    extract_key_value_pairs,
+    map_fields,
+    create_record,
+    attach_pdf,
+    log_failure
+)
+
+# Set up environment
+os.environ["GRADIO_TEMP_DIR"] = "temp"
+os.makedirs("temp", exist_ok=True)
+
+# Global variable
+object_fields_cache = {}
+
+# Get all Salesforce object names
+def get_salesforce_object_names():
+    sf, error = get_salesforce_client()
+    if error:
+        return ["Error connecting to Salesforce"]
+    object_names, _ = get_salesforce_objects(sf)
+    return object_names
+
+# Extract key-value pairs
+def extract_contract_fields(pdf_paths):
+    if not pdf_paths or not isinstance(pdf_paths, list) or not pdf_paths[0]:
+        return {}, "No PDF uploaded"
+    extracted_data, error = extract_key_value_pairs(pdf_paths[0])
+    if error:
+        return {}, f"Extraction error: {error}"
+    return extracted_data[0], "Extraction successful"
+
+# Map fields between extracted data and Salesforce object fields
+def map_contract_fields(pdf_paths, object_name):
+    if not pdf_paths or not isinstance(pdf_paths, list) or not pdf_paths[0]:
+        return "No PDF uploaded"
+    sf, error = get_salesforce_client()
+    if error:
+        return f"Error: {error}"
+    extracted_data, error = extract_key_value_pairs(pdf_paths[0])
+    if error:
+        return f"Error: {error}"
+    if object_name in object_fields_cache:
+        fields = object_fields_cache[object_name]
+    else:
+        fields, error = get_object_fields(sf, object_name)
+        object_fields_cache[object_name] = fields
+    mappings, scores, _ = map_fields(extracted_data, fields)
+    mapping_display = ""
+    for key, sf_field in mappings.items():
+        mapping_display += f"{key} -> {sf_field} (Confidence: {scores.get(key, 0)})\n"
+    return mapping_display
+
+# Migrate to Salesforce
+def migrate_to_salesforce(pdf_paths, object_name):
+    if not pdf_paths or not isinstance(pdf_paths, list) or not pdf_paths[0]:
+        return "Error: No valid PDF file provided"
+    pdf_path = pdf_paths[0]
+    sf, error = get_salesforce_client()
+    if error:
+        log_failure(pdf_path, object_name, error)
+        return f"Error: {error}"
+    extracted_data, error = extract_key_value_pairs(pdf_path)
+    if error:
+        log_failure(pdf_path, object_name, error)
+        return f"Error: {error}"
+    fields, error = get_object_fields(sf, object_name)
+    if error:
+        log_failure(pdf_path, object_name, error)
+        return f"Error: {error}"
+    mappings, _, error = map_fields(extracted_data, fields)
+    if error:
+        log_failure(pdf_path, object_name, error)
+        return f"Error: {error}"
+
+    data = {}
+    for key, value in zip(extracted_data[0]["keys"], extracted_data[0]["values"]):
+        sf_field = mappings.get(key)
+        if not sf_field:
+            continue
+        if "value" in key.lower() and isinstance(value, str):
+            value = float(value.replace(",", "").replace("$", ""))
+        data[sf_field] = value
+
+    data["AccountId"] = "001XXXXXXXXXXXXXXX"  # Replace with actual AccountId
+
+    record_id, error = create_record(sf, object_name, data)
+    if error:
+        log_failure(pdf_path, object_name, error)
+        return f"Error: {error}"
+    attach_status, error = attach_pdf(sf, record_id, pdf_path)
+    if error:
+        log_failure(pdf_path, object_name, error)
+        return f"Error: {error}"
+    return f"✅ Record Created: {record_id}\n📎 Attachment: {attach_status}"
+
+# Gradio UI
+with gr.Blocks(title="Smart Contract Migrator") as demo:
+    gr.Markdown("# 📄 Smart Contract Migrator to Salesforce")
+
+    with gr.Row():
+        pdf_input = gr.File(label="Upload Contract PDF", file_types=[".pdf"])
+        object_dropdown = gr.Dropdown(label="Salesforce Object", choices=get_salesforce_object_names())
+
+    extract_btn = gr.Button("Extract Fields")
+    extract_output = gr.JSON(label="Extracted Fields")
+
+    map_btn = gr.Button("Map Fields")
+    map_output = gr.Textbox(label="Field Mappings")
+
+    migrate_btn = gr.Button("Migrate to Salesforce")
+    migrate_output = gr.Textbox(label="Migration Result")
+
+    extract_btn.click(fn=extract_contract_fields, inputs=[pdf_input], outputs=[extract_output, map_output])
+    map_btn.click(fn=map_contract_fields, inputs=[pdf_input, object_dropdown], outputs=map_output)
+    migrate_btn.click(fn=migrate_to_salesforce, inputs=[pdf_input, object_dropdown], outputs=migrate_output)
+
+if __name__ == "__main__":
+    demo.launch()