Spaces:
Build error
Build error
| 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() | |