Spaces:
Sleeping
Sleeping
| import os | |
| os.environ["OMP_NUM_THREADS"] = "1" | |
| import re | |
| import json | |
| import datetime as dt | |
| import gradio as gr | |
| from utils import extract_kyc_fields | |
| # ------------------ HARD-CODED SALESFORCE CREDS (as requested) ------------------ | |
| SF_USERNAME = "licproject@2025.com" | |
| SF_PASSWORD = "Lic@2025" | |
| SF_SECURITY_TOKEN = "AmmfRcd6IiYaRtSGntBnzNMQU" | |
| SF_DOMAIN = "login" # "login" for prod, "test" for sandbox | |
| # ------------------------------------------------------------------------------- | |
| # simple-salesforce + exceptions | |
| try: | |
| from simple_salesforce import Salesforce | |
| from simple_salesforce.exceptions import ( | |
| SalesforceAuthenticationFailed, | |
| SalesforceGeneralError, | |
| SalesforceMalformedRequest, | |
| SalesforceExpiredSession, | |
| SalesforceRefusedRequest, | |
| SalesforceMoreThanOneRecord, | |
| SalesforceResourceNotFound | |
| ) | |
| SF_AVAILABLE = True | |
| except Exception: | |
| SF_AVAILABLE = False | |
| # ---------- helpers ---------- | |
| def _parse_birthdate(dob_text: str): | |
| """ | |
| Normalize DOB to YYYY-MM-DD (Salesforce Date fields). | |
| Supports dd/mm/yyyy, dd-mm-yyyy, dd.mm.yyyy, yyyy-mm-dd, or just YYYY (mapped to mid-year). | |
| """ | |
| if not dob_text or dob_text == "Not found": | |
| return None | |
| s = dob_text.strip() | |
| m = re.fullmatch(r"(\d{4})-(\d{2})-(\d{2})", s) | |
| if m: | |
| y, mo, d = map(int, m.groups()) | |
| try: | |
| return dt.date(y, mo, d).isoformat() | |
| except ValueError: | |
| return None | |
| m = re.fullmatch(r"(\d{2})[./-](\d{2})[./-](\d{4})", s) | |
| if m: | |
| d, mo, y = map(int, m.groups()) | |
| try: | |
| return dt.date(y, mo, d).isoformat() | |
| except ValueError: | |
| return None | |
| m = re.fullmatch(r"(19|20)\d{2}", s) | |
| if m: | |
| y = int(s) | |
| try: | |
| return dt.date(y, 6, 15).isoformat() | |
| except ValueError: | |
| return None | |
| return None | |
| def _fmt_sf_error(err: Exception): | |
| """Produce a clear, JSON-safe dict from any simple-salesforce error.""" | |
| info = {"type": err.__class__.__name__, "message": str(err)} | |
| if isinstance(err, SalesforceAuthenticationFailed): | |
| content = getattr(err, "content", None) or getattr(err, "response", None) | |
| info.update({ | |
| "category": "AUTHENTICATION", | |
| "status": getattr(err, "status", None), | |
| "content": getattr(content, "content", None) if hasattr(content, "content") else content, | |
| }) | |
| elif isinstance(err, (SalesforceMalformedRequest, SalesforceGeneralError, SalesforceRefusedRequest)): | |
| info.update({ | |
| "category": "REQUEST", | |
| "status": getattr(err, "status", None), | |
| "resource": getattr(err, "resource_name", None), | |
| "url": getattr(err, "url", None), | |
| "content": getattr(err, "content", None), | |
| }) | |
| elif isinstance(err, SalesforceResourceNotFound): | |
| info.update({"category": "NOT_FOUND"}) | |
| elif isinstance(err, SalesforceExpiredSession): | |
| info.update({"category": "AUTH_EXPIRED"}) | |
| return info | |
| def sf_connect(): | |
| """ | |
| Connect to Salesforce using the hardcoded credentials. | |
| Also runs a quick auth query to surface any auth issues clearly. | |
| """ | |
| if not SF_AVAILABLE: | |
| raise RuntimeError("simple-salesforce is not installed. Add `simple-salesforce` to requirements.txt.") | |
| sf = Salesforce( | |
| username=SF_USERNAME, | |
| password=SF_PASSWORD, | |
| security_token=SF_SECURITY_TOKEN, | |
| domain=SF_DOMAIN, | |
| ) | |
| # quick auth sanity check (will raise on bad auth) | |
| sf.query("SELECT Id FROM User LIMIT 1") | |
| return sf | |
| def _raw_create_with_fallback(sf, object_api_name: str, payload: dict): | |
| """ | |
| Fallback path that bypasses simple-salesforce's error wrappers and calls REST directly. | |
| Always returns a dict: | |
| - on success: {"success": True, "id": "...", "url": "...", "status_code": 201} | |
| - on error: {"success": False, "status_code": <int>, "url": "...", "response_json": <json or None>, "response_text": <str>} | |
| """ | |
| url = f"{sf.base_url}sobjects/{object_api_name}" | |
| try: | |
| resp = sf.session.post(url, json=payload, headers=sf.headers, timeout=30) | |
| status = resp.status_code | |
| try: | |
| body = resp.json() | |
| except Exception: | |
| body = None | |
| if 200 <= status < 300: | |
| # Salesforce returns {"id": "...", "success": true, "errors": []} | |
| rec_id = (body or {}).get("id") if isinstance(body, dict) else None | |
| return {"success": True, "id": rec_id, "status_code": status, "url": url, "response_json": body} | |
| else: | |
| # Return raw details so you see the exact field/object error | |
| return { | |
| "success": False, | |
| "status_code": status, | |
| "url": url, | |
| "response_json": body, | |
| "response_text": resp.text, | |
| } | |
| except Exception as e: | |
| # Network or session issues | |
| return {"success": False, "url": url, "exception": _fmt_sf_error(e)} | |
| def sf_push_kyc_record(sf, ocr_results): | |
| """ | |
| Create one KYC_Record__c combining Aadhaar + PAN. | |
| Salesforce custom object fields: | |
| Aadhaar_Number__c, Aadhaar_Name__c, Aadhaar_DOB__c (Date) | |
| PAN_Number__c, Pan_Name__c, Pan_DOB__c (Date) | |
| """ | |
| a = ocr_results.get("aadhaar") or {} | |
| p = ocr_results.get("pan") or {} | |
| aadhaar_number = a.get("aadhaar_number") if (a.get("card_type") == "AADHAAR") else None | |
| aadhaar_name = a.get("name") if (a.get("card_type") == "AADHAAR") else None | |
| aadhaar_dob = _parse_birthdate(a.get("dob")) if (a.get("card_type") == "AADHAAR") else None | |
| pan_number = p.get("pan_number") if (p.get("card_type") == "PAN") else None | |
| pan_name = p.get("name") if (p.get("card_type") == "PAN") else None | |
| pan_dob = _parse_birthdate(p.get("dob")) if (p.get("card_type") == "PAN") else None | |
| payload = { | |
| "Aadhaar_Number__c": aadhaar_number, | |
| "Aadhaar_Name__c": aadhaar_name, | |
| "Aadhaar_DOB__c": aadhaar_dob, # Date field in SF | |
| "PAN_Number__c": pan_number, | |
| "Pan_Name__c": pan_name, | |
| "Pan_DOB__c": pan_dob, # Date field in SF | |
| } | |
| # Remove None keys to avoid nulling non-nullable fields | |
| payload = {k: v for k, v in payload.items() if v is not None} | |
| # First try the nice SDK method | |
| try: | |
| result = sf.KYC_Record__c.create(payload) | |
| return {"success": True, "id": result.get("id"), "payload": payload, "via": "sdk"} | |
| except Exception as e: | |
| # If simple-salesforce throws its unhelpful TypeError, fall back to a raw REST POST | |
| raw = _raw_create_with_fallback(sf, "KYC_Record__c", payload) | |
| if raw.get("success"): | |
| return {"success": True, "id": raw.get("id"), "payload": payload, "via": "raw", "raw": raw} | |
| else: | |
| return {"success": False, "error": _fmt_sf_error(e), "payload": payload, "raw": raw} | |
| # ---------- gradio callback ---------- | |
| def process_documents(aadhaar_file, pan_file, push_to_sf): | |
| """ | |
| - Runs OCR on Aadhaar and PAN separately. | |
| - Optionally pushes a single KYC_Record__c to Salesforce with robust fallback. | |
| """ | |
| results = {"aadhaar": None, "pan": None} | |
| if not aadhaar_file and not pan_file: | |
| return {"error": "Please upload at least one file (Aadhaar and/or PAN)."} | |
| # OCR Aadhaar | |
| if aadhaar_file: | |
| try: | |
| res = extract_kyc_fields(aadhaar_file.name) | |
| res["source_file"] = os.path.basename(aadhaar_file.name) | |
| results["aadhaar"] = res | |
| except Exception as e: | |
| results["aadhaar"] = {"error": f"Aadhaar OCR failed: {str(e)}", "card_type": "UNKNOWN"} | |
| # OCR PAN | |
| if pan_file: | |
| try: | |
| res = extract_kyc_fields(pan_file.name) | |
| res["source_file"] = os.path.basename(pan_file.name) | |
| results["pan"] = res | |
| except Exception as e: | |
| results["pan"] = {"error": f"PAN OCR failed: {str(e)}", "card_type": "UNKNOWN"} | |
| output = {"ocr": results} | |
| if push_to_sf: | |
| try: | |
| sf = sf_connect() | |
| created = sf_push_kyc_record(sf, results) | |
| output["salesforce"] = {"pushed": created.get("success", False), **created} | |
| except Exception as e: | |
| # Even connection/auth errors will be formatted | |
| output["salesforce"] = {"pushed": False, "error": _fmt_sf_error(e)} | |
| return output | |
| # ---------- UI ---------- | |
| with gr.Blocks(title="Smart KYC OCR → Salesforce (KYC_Record__c)") as demo: | |
| gr.Markdown( | |
| """ | |
| # 🧾 Smart KYC OCR → Salesforce | |
| Upload **Aadhaar** and **PAN** in separate boxes, then (optional) push one **KYC_Record__c**. | |
| """ | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| aadhaar_uploader = gr.File( | |
| label="📤 Aadhaar Upload", | |
| file_types=[".jpg", ".jpeg", ".png"] | |
| ) | |
| with gr.Column(scale=1): | |
| pan_uploader = gr.File( | |
| label="📤 PAN Upload", | |
| file_types=[".jpg", ".jpeg", ".png"] | |
| ) | |
| push_to_sf = gr.Checkbox(label="Push to Salesforce (create KYC_Record__c)", value=False) | |
| submit_btn = gr.Button("🔍 Extract KYC Info", variant="primary") | |
| output_json = gr.JSON(label="📋 Output (OCR + Salesforce)") | |
| submit_btn.click( | |
| fn=process_documents, | |
| inputs=[aadhaar_uploader, pan_uploader, push_to_sf], | |
| outputs=output_json, | |
| ) | |
| gr.Markdown("---") | |
| gr.Markdown( | |
| """ | |
| If an error occurs, you'll now see **status_code**, **url**, and Salesforce’s **response_json** for fast debugging. | |
| """ | |
| ) | |
| # Important for Spaces: keep `demo` at module level | |
| if __name__ == "__main__": | |
| demo.launch() | |