import os from paddleocr import PaddleOCR import gradio as gr import re import pandas as pd from fuzzywuzzy import process from simple_salesforce import Salesforce # Attribute mappings: readable names to Salesforce API names ATTRIBUTE_MAPPING = { "Product name": "Productname__c", "Colour": "Colour__c", "Motortype": "Motortype__c", "Frequency": "Frequency__c", "Grossweight": "Grossweight__c", "Ratio": "Ratio__c", "MotorFrame": "Motorframe__c", "Model": "Model__c", "Speed": "Speed__c", "Quantity": "Quantity__c", "Voltage": "Voltage__c", "Material": "Material__c", "Type": "Type__c", "Horsepower": "Horsepower__c", "Consignee": "Consignee__c", "LOT": "LOT__c", "Stage": "Stage__c", "Outlet": "Outlet__c", "Serialnumber": "Serialnumber__c", "HeadSize": "Headsize__c", "Deliverysize": "Deliverysize__c", "Phase": "Phase__c", "Size": "Size__c", "MRP": "MRP__c", "Usebefore": "Usebefore__c", "Height": "Height__c", "MaximumDischarge Flow": "Maximumdischargeflow__c", "DischargeRange": "Dischargeflow__c", "Assembledby": "Manufacturer__c", "Manufacturedate": "Manufacturedate__c", "Companyname": "Companyname__c", "Customercarenumber": "Customercarenumber__c", "SellerAddress": "Selleraddress__c", "Selleremail": "Selleremail__c", "GSTIN": "GSTIN__c", "Totalamount": "Totalamount__c", "Paymentstatus": "Paymentstatus__c", "Paymentmethod": "Paymentstatus__c", "Invoicedate": "Manufacturedate__c", "Warranty": "Warranty__c", "Brand": "Brand__c", "Motorhorsepower": "Motorhorsepower__c", "Power": "Power__c", "Motorphase": "Motorphase__c", "Enginetype": "Enginetype__c", "Tankcapacity": "Tankcapacity__c", "Head": "Head__c", "Usage/Application": "Usage_Application__c", "Volts": "volts__c", "Hertz": "Hertz__c", "Frame": "frame__c", "Mounting": "Mounting__c", "Tollfreenumber": "Tollfreenumber__c", "Pipesize": "Pipesize__c", "Manufacturer": "Manufacturer__c", "Office": "Office__c", "SRnumber": "SRnumber__c", "TypeOfEndUse": "TypeOfEndUse__c", "Model Name": "Model_Name_Number__c", "coolingmethod": "coolingmethod__c", "H.P.": "H_p__c" } # List of product names to match PRODUCT_NAMES = [ "Fusion", "Agroking", "CG commercial motors", "Jaguar", "Gaurav" ] # Salesforce credentials SALESFORCE_USERNAME = "venkataramanamotors@sales.com" SALESFORCE_PASSWORD = "Seta12345@" SALESFORCE_SECURITY_TOKEN = "zQjGfQsrFa5dqc4xEgunt8PwR" # Initialize PaddleOCR ocr = PaddleOCR(use_angle_cls=True, lang='en') # Function to extract text using PaddleOCR def extract_text(image): result = ocr.ocr(image) extracted_text = [] for line in result[0]: extracted_text.append(line[1][0]) return "\n".join(extracted_text) # Function to match product name using fuzzy matching def match_product_name(extracted_text): best_match = None best_score = 0 for line in extracted_text.split("\n"): match, score = process.extractOne(line, PRODUCT_NAMES) if score > best_score: best_match = match best_score = score return best_match if best_score >= 70 else None # Function to extract attributes and their values def extract_attributes(extracted_text): attributes = {} for readable_attr, sf_attr in ATTRIBUTE_MAPPING.items(): pattern = rf"{re.escape(readable_attr)}[:\-]?\s*(.+)" match = re.search(pattern, extracted_text, re.IGNORECASE) if match: attributes[readable_attr] = match.group(1).strip() return attributes # Function to filter attributes for valid Salesforce fields def filter_valid_attributes(attributes, valid_fields): return {ATTRIBUTE_MAPPING[key]: value for key, value in attributes.items() if ATTRIBUTE_MAPPING[key] in valid_fields} # Function to interact with Salesforce based on mode and type def interact_with_salesforce(mode, entry_type, quantity, attributes): try: sf = Salesforce( username=SALESFORCE_USERNAME, password=SALESFORCE_PASSWORD, security_token=SALESFORCE_SECURITY_TOKEN ) object_name = None field_name = None field_names = [] product_field_name = "Productname__c" model_field_name = "Model__c" stage_field_name = "Stage__c" hp_field_name = "H_p__c" if mode == "Entry": if entry_type == "Sales": object_name = "VENKATA_RAMANA_MOTORS__c" field_name = "Quantity__c" elif entry_type == "Non-Sales": object_name = "UNBILLING_DATA__c" field_name = "TotalQuantity__c" elif mode == "Exit": if entry_type == "Sales": object_name = "Inventory_Management__c" field_names = ["Quantity_Sold__c", "soldstock__c"] elif entry_type == "Non-Sales": object_name = "Un_Billable__c" field_names = ["Sold_Out__c", "soldstock__c"] if not object_name or (not field_name and not field_names): return "Invalid mode or entry type." sf_object = sf.__getattr__(object_name) schema = sf_object.describe() valid_fields = {field["name"] for field in schema["fields"]} filtered_attributes = filter_valid_attributes(attributes, valid_fields) if mode == "Exit": query_conditions = [f"{product_field_name} = '{attributes['Product name']}'"] if "Model Name" in attributes and attributes["Model Name"]: query_conditions.append(f"{model_field_name} = '{attributes['Model Name']}'") if "Stage" in attributes and attributes["Stage"] != "": query_conditions.append(f"{stage_field_name} = '{attributes['Stage']}'") if "H.P." in attributes and attributes["H.P."]: query_conditions.append(f"{hp_field_name} = '{attributes['H.P.']}'") query = f"SELECT Id, {', '.join(field_names)} FROM {object_name} WHERE {' AND '.join(query_conditions)} LIMIT 1" response = sf.query(query) if response["records"]: record_id = response["records"][0]["Id"] updated_fields = {field: quantity for field in field_names} sf_object.update(record_id, updated_fields) return f"✅ Updated record for product '{attributes['Product name']}' in {object_name}. Updated fields: {updated_fields}." else: return f"❌ No matching record found for product '{attributes['Product name']}' in {object_name}." else: filtered_attributes[field_name] = quantity sf_object.create(filtered_attributes) return f"✅ Data successfully exported to Salesforce object {object_name}." except Exception as e: return f"❌ Error interacting with Salesforce: {str(e)}" # Function to process image, extract attributes, and allow editing def process_image(image, mode, entry_type, quantity): extracted_text = extract_text(image) if not extracted_text: return "No text detected in the image.", None, None product_name = match_product_name(extracted_text) attributes = extract_attributes(extracted_text) if product_name: attributes["Product name"] = product_name # Ensure fixed attributes are present for fixed_attr in ["Stage", "H.P.", "Product name", "Model"]: if fixed_attr not in attributes: attributes[fixed_attr] = "" # Convert attributes to DataFrame for editing df = pd.DataFrame(list(attributes.items()), columns=["Attribute", "Value"]) return f"Extracted Text:\n{extracted_text}", df, None # Function to handle edited attributes and export to Salesforce def export_to_salesforce(mode, entry_type, quantity, edited_df): try: # Convert edited DataFrame back to dictionary edited_attributes = dict(zip(edited_df["Attribute"], edited_df["Value"])) # Export to Salesforce message = interact_with_salesforce(mode, entry_type, quantity, edited_attributes) return message except Exception as e: return f"❌ Error exporting to Salesforce: {str(e)}" # Gradio Interface def app(): with gr.Blocks() as demo: with gr.Tab("📥 OCR Processing"): with gr.Row(): image_input = gr.Image(type="numpy", label="📄 Upload Image") mode_input = gr.Dropdown(label="📌 Mode", choices=["Entry", "Exit"], value="Entry") entry_type_input = gr.Radio(label="📦 Entry Type", choices=["Sales", "Non-Sales"], value="Sales") quantity_input = gr.Number(label="🔢 Quantity", value=1, interactive=True) extract_button = gr.Button("Extract Text and Attributes") extracted_text_output = gr.Text(label="📝 Extracted Image Data") editable_df_output = gr.Dataframe(label="✏️ Edit Attributes (Key-Value Pairs)", interactive=True) ok_button = gr.Button("OK") result_output = gr.Text(label="🚀 Result") # Define button actions extract_button.click( fn=process_image, inputs=[image_input, mode_input, entry_type_input, quantity_input], outputs=[extracted_text_output, editable_df_output, result_output] ) ok_button.click( fn=export_to_salesforce, inputs=[mode_input, entry_type_input, quantity_input, editable_df_output], outputs=[result_output] ) return demo if __name__ == "__main__": app().launch(share=True)