Spaces:
Running
Running
File size: 3,817 Bytes
c586aa6 e4889dc c586aa6 1ae8f9f e4889dc c8d472e a4d0a87 c586aa6 281bbb2 c586aa6 a4d0a87 e4889dc a4d0a87 806c938 a4d0a87 806c938 a4d0a87 e4889dc 806c938 a4d0a87 e4889dc a4d0a87 e4889dc c586aa6 e4889dc a4d0a87 c586aa6 a4d0a87 e4889dc c586aa6 e4889dc c586aa6 57d3fac e4889dc c586aa6 e4889dc c586aa6 8dcb382 e4889dc 7fb29ae a4d0a87 c586aa6 a4d0a87 c586aa6 e4889dc c586aa6 e4889dc c586aa6 52f3a2e c586aa6 e4889dc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 |
import os
import gradio as gr
from datetime import datetime
from utils import extract_kyc_fields
from simple_salesforce import Salesforce
from paddleocr import PaddleOCR
# ------------------------
# CONFIGURATION
# ------------------------
SF_USERNAME = os.getenv("SF_USERNAME", "licproject@2025.com")
SF_PASSWORD = os.getenv("SF_PASSWORD", "Lic@2025@")
SF_TOKEN = os.getenv("SF_TOKEN", "AmmfRcd6IiYaRtSGntBnzNMQU")
# Limit Paddle threads to avoid container crashes
os.environ["OMP_NUM_THREADS"] = "1"
# Initialize OCR
ocr = PaddleOCR(use_angle_cls=True, lang='en')
# ------------------------
# CONNECT TO SALESFORCE
# ------------------------
def connect_salesforce():
try:
sf = Salesforce(
username=SF_USERNAME,
password=SF_PASSWORD,
security_token=SF_TOKEN,
domain="login" # Production
)
print("✅ Connected to Salesforce (Production)")
return sf
except Exception as e:
print("❌ Production login failed:", e)
try:
sf = Salesforce(
username=SF_USERNAME,
password=SF_PASSWORD,
security_token=SF_TOKEN,
domain="test" # Sandbox
)
print("✅ Connected to Salesforce (Sandbox)")
return sf
except Exception as e2:
print("❌ Sandbox login failed:", e2)
return None
sf = connect_salesforce()
# ------------------------
# CREATE KYC RECORD IN SALESFORCE
# ------------------------
def create_kyc_record(sf, agent_id, kyc_data, file_name=None):
try:
if not sf:
return {"status": "error", "message": "Salesforce connection failed"}
record = {
"Agent__c": agent_id,
"Name": kyc_data.get("name", "Unknown KYC"),
"Document_Number__c": kyc_data.get("pan_number") or kyc_data.get("aadhaar_number"),
"Document_Type__c": kyc_data.get("card_type", "UNKNOWN"),
"Upload_Date__c": datetime.now().isoformat()
}
# Optional: store file name / URL if needed
if file_name:
record["KYC_Document_URL__c"] = f"Uploaded File: {file_name}"
response = sf.KYC_Document__c.create(record)
return {"status": "success", "record_id": response.get("id")}
except Exception as e:
return {"status": "error", "message": str(e)}
# ------------------------
# PROCESS IMAGE FUNCTION
# ------------------------
def process_image(image_file, agent_id):
if not sf:
return {"status": "error", "message": "Salesforce connection failed"}
if not image_file:
return {"status": "error", "message": "No image uploaded"}
if not agent_id:
return {"status": "error", "message": "Agent ID is required"}
try:
# Extract KYC data using utils.py
kyc_data = extract_kyc_fields(image_file)
# Create Salesforce record
sf_response = create_kyc_record(sf, agent_id, kyc_data, os.path.basename(image_file))
return {
"ocr_result": kyc_data,
"salesforce_result": sf_response
}
except Exception as e:
return {"status": "error", "message": f"OCR/Salesforce error: {str(e)}"}
# ------------------------
# GRADIO INTERFACE
# ------------------------
iface = gr.Interface(
fn=process_image,
inputs=[
gr.Image(type="filepath", label="Upload PAN / Aadhaar"),
gr.Textbox(label="Agent Salesforce ID", placeholder="Enter Agent record ID")
],
outputs=gr.JSON(),
title="AI KYC OCR & Salesforce Integration",
description="Upload PAN or Aadhaar card images to extract text and automatically create KYC records in Salesforce."
)
if __name__ == "__main__":
iface.launch(server_name="0.0.0.0", server_port=7860)
|