from transformers import BlipProcessor, BlipForConditionalGeneration from PIL import Image import gradio as gr import torch from datetime import datetime from fpdf import FPDF from simple_salesforce import Salesforce import requests import json import os from dotenv import load_dotenv # Load environment variables from .env file load_dotenv() # Salesforce credentials SF_USERNAME = os.getenv('SF_USERNAME') SF_PASSWORD = os.getenv('SF_PASSWORD') SF_SECURITY_TOKEN = os.getenv('SF_SECURITY_TOKEN') # Initialize Salesforce connection sf = Salesforce(username=SF_USERNAME, password=SF_PASSWORD, security_token=SF_SECURITY_TOKEN) print(f"SF_USERNAME: {SF_USERNAME}, SF_PASSWORD: {SF_PASSWORD}, SF_SECURITY_TOKEN: {SF_SECURITY_TOKEN}") print("Salesforce Base URL:", sf.base_url) # Load BLIP model and processor processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base") model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base") model.eval() device = "cuda" if torch.cuda.is_available() else "cpu" model.to(device) # Inference function to generate captions dynamically based on image content def generate_captions_from_image(image): if image.mode != "RGB": image = image.convert("RGB") # Preprocess the image and generate a caption inputs = processor(image, return_tensors="pt").to(device, torch.float16) output = model.generate(**inputs, max_new_tokens=50) caption = processor.decode(output[0], skip_special_tokens=True) return caption # Function to create PDF and upload to Salesforce def create_and_upload_pdf(dpr_content): # Create PDF instance using FPDF pdf = FPDF() pdf.set_auto_page_break(auto=True, margin=15) pdf.add_page() # Set title and content for the PDF pdf.set_font("Arial", size=12) pdf.cell(200, 10, txt="Daily Progress Report", ln=True, align='C') pdf.ln(10) # Add space between lines # Add the content of the DPR text to the PDF pdf.multi_cell(0, 10, dpr_content) # Save PDF to a file (temporary storage) pdf_output_path = "/tmp/dpr_report.pdf" pdf.output(pdf_output_path) # Read the generated PDF as binary data with open(pdf_output_path, 'rb') as pdf_file: pdf_data = pdf_file.read() # Prepare request to Salesforce ContentVersion API content_version_payload = { "Title": "Daily Progress Report", "PathOnClient": "DPR_Report.pdf", } content_version_url = f"https://sathkruthatechsolutionspvt6-dev-ed.develop.my.salesforce.com/services/data/v59.0/sobjects/ContentVersion/" headers = { "Authorization": f"Bearer {sf.session_id}", "Content-Type": "application/json" # For metadata } # Debugging step: Log the payload being sent to Salesforce print("Content Version Payload:", json.dumps(content_version_payload, indent=4)) # Pass the binary data in the files parameter (do not include it in the JSON payload) files = { "VersionData": ("DPR_Report.pdf", pdf_data) # Tuple (filename, binary data) } # Make the POST request to upload the file response = requests.post(content_version_url, headers=headers, files=files) # Log the full response from Salesforce print("Response Status Code:", response.status_code) print("Response Headers:", response.headers) print("Response Body:", response.text) # Check the response status if response.status_code == 201: content_version = response.json() pdf_url = f"/sfc/servlet.shepherd/version/download/{content_version['Id']}" return pdf_url else: # If response is not 201, log the response text for further debugging print("Error details from Salesforce response:", response.text) raise Exception("Error uploading PDF to Salesforce: " + response.text) # Function to generate the daily progress report (DPR) text def generate_dpr(files): # Query the latest Daily Progress Report records to get the highest number last_report = sf.query("SELECT Name FROM Daily_Progress_Reports__c ORDER BY CreatedDate DESC LIMIT 1") # Generate the new report name if last_report['totalSize'] > 0: last_name = last_report['records'][0]['Name'] # Extract the number from the last report name, assuming format "Daily Progress Report X" report_number = int(last_name.split()[-1]) + 1 # Increment the number else: # If no records exist, start with "1" report_number = 1 # Generate the new report name dynamically new_report_name = f"Daily Progress Report {report_number}" # Prepare the current time and start the DPR text current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") dpr_text = [f"Daily Progress Report\nGenerated on: {current_time}\n"] # Process each uploaded file (image) for file in files: # Open the image from the file path image = Image.open(file.name) # Using file.name for filepath if image.mode != "RGB": image = image.convert("RGB") # Generate a caption for the image caption = generate_captions_from_image(image) # Generate a section of the DPR for this image dpr_section = f"\nImage: {file.name}\nDescription: {caption}\n" dpr_text.append(dpr_section) # Join the DPR sections into one complete string dpr_content = "\n".join(dpr_text) # Create and upload the PDF for this report to Salesforce pdf_url = create_and_upload_pdf(dpr_content) # Create the new Daily Progress Report record in Salesforce new_report_data = { "Name": new_report_name, # Set the dynamic name "Report_Date__c": current_time, # Set the report date to the current time "Detected_Activities__c": dpr_content, # Set the activities field to the generated text "Photo_Uploads__c": ", ".join([file.name for file in files]), # Upload the image names as Photo Uploads "PDF_URL__c": pdf_url # Set the PDF URL field with the generated PDF link } # Insert the new report into Salesforce new_report = sf.Daily_Progress_Reports__c.create(new_report_data) return f"New report created in Salesforce with PDF URL: {pdf_url}" # Gradio interface for uploading multiple files and displaying the text-based DPR iface = gr.Interface( fn=generate_dpr, inputs=gr.Files(type="filepath", label="Upload Site Photos"), # Handle batch upload of images outputs="text", # Display the DPR as text in the output section title="Daily Progress Report Generator", description="Upload up to 10 site photos. The AI model will dynamically detect construction activities, materials, and progress and generate a text-based Daily Progress Report (DPR).", allow_flagging="never" # Optional: Disable flagging ) iface.launch()