Update app.py

app.py

@@ -12,11 +12,11 @@ from io import BytesIO
 import uuid
 import logging
 import textwrap
+import tempfile
 from reportlab.lib.pagesizes import letter
 from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
 from reportlab.lib.styles import getSampleStyleSheet
-from reportlab.lib.units import inch
-from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type
+import time
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -353,14 +353,13 @@ footer, .gradio-footer, .hide, [data-testid="Use via API"], [data-testid="mmsett
 }
 """
 
-# Salesforce credentials
+# Salesforce credentials (kept for potential future use, but not used in this update)
 SF_USERNAME = "Kushalpavansekharm503@agentforce.com"
 SF_PASSWORD = "Kushal@123"
 SF_TOKEN = "WwUIFWBVUjeKn9VPKyWJmawY0"
 
-@retry(stop_max_attempt_number=3, wait_fixed=2000, retry=retry_if_exception_type(Exception))
 def authenticate_salesforce() -> Salesforce:
-    """Authenticate with Salesforce and return a Salesforce client"""
+    """Authenticate with Salesforce and return a Salesforce client (not used in this update)"""
     try:
         sf = Salesforce(
             username=SF_USERNAME,
@@ -371,7 +370,7 @@ def authenticate_salesforce() -> Salesforce:
         return sf
     except Exception as e:
         logger.error(f"Failed to authenticate with Salesforce: {str(e)}")
-        raise
+        raise Exception(f"Salesforce authentication failed: {str(e)}")
 
 def get_hugging_face_sentiment(text: str) -> float:
     """Get sentiment score using Hugging Face model"""
@@ -385,135 +384,37 @@ def get_hugging_face_sentiment(text: str) -> float:
         logger.error(f"Hugging Face sentiment analysis failed: {str(e)}. Using fallback score.")
         return 0.5
 
-def generate_contract_analysis_pdf(data: Dict, temp_dir: str) -> Tuple[str, str, BytesIO]:
-    """Generate a PDF with contract analysis results"""
+def generate_sentiment_pdf(sentiment_score: float, output_dir: str) -> Tuple[str, str, BytesIO]:
+    """Generate a PDF with sentiment analysis results and return path, URL, and file object"""
     try:
-        pdf_filename = f"contract_analysis_{data['record_id']}_{int(time.time())}.pdf"
-        pdf_path = os.path.join(temp_dir, pdf_filename)
+        pdf_filename = f"sentiment_analysis_{int(time.time())}.pdf"
+        pdf_path = os.path.join(output_dir, pdf_filename)
         pdf_file = BytesIO()
-        c = canvas.Canvas(pdf_file, pagesize=letter)
-        
-        c.setFont("Helvetica-Bold", 16)
-        c.drawString(1 * inch, 10 * inch, "Contract Analysis Report")
-        
-        c.setFont("Helvetica", 12)
-        c.drawString(1 * inch, 9.5 * inch, f"Date: {time.strftime('%Y-%m-%d')}")
-        c.drawString(1 * inch, 9.2 * inch, f"Time: {time.strftime('%H:%M:%S')} IST")
-        
-        c.setFont("Helvetica-Bold", 14)
-        c.drawString(1 * inch, 8.7 * inch, f"Risk Score: {data['risk_score']:.1f}/100 ({data['risk_level']})")
-        c.drawString(1 * inch, 8.4 * inch, f"Sentiment Score: {data['sentiment_score']:.2f} (0.0 Negative to 1.0 Positive)")
-        
-        y_position = 8.0 * inch
-        c.setFont("Helvetica-Bold", 12)
-        c.drawString(1 * inch, y_position, "Summary:")
-        y_position -= 0.3 * inch
-        
-        c.setFont("Helvetica", 10)
-        summary_data = {
-            "Total Penalties": sum(data['penalty_counts'].values()),
-            "Total Obligations": sum(data['obligation_counts'].values()),
-            "Total Delays": sum(data['delay_counts'].values()),
-            "Analysis Timestamp": time.strftime("%Y-%m-%d %H:%M:%S IST")
-        }
-        for key, value in summary_data.items():
-            c.drawString(1 * inch, y_position, f"{key}: {value}")
-            y_position -= 0.25 * inch
-        
-        y_position -= 0.5 * inch
-        c.setFont("Helvetica-Bold", 12)
-        c.drawString(1 * inch, y_position, "Detailed Analysis:")
-        y_position -= 0.3 * inch
-        
-        c.setFont("Helvetica", 10)
-        sections = [
-            ("Penalty Clauses", data['penalty_details']),
-            ("Obligation Clauses", data['obligation_details']),
-            ("Delay Clauses", data['delay_details']),
-            ("Penalty Amounts", data['penalty_amounts']),
-            ("Extracted Data", data['penalty_examples'])
-        ]
-        for title, content in sections:
-            c.drawString(1 * inch, y_position, f"{title}:")
-            y_position -= 0.2 * inch
-            lines = content.split("\n")
-            for line in lines:
-                if y_position < 1 * inch:
-                    c.showPage()
-                    c.setFont("Helvetica", 10)
-                    y_position = 10 * inch
-                c.drawString(1.2 * inch, y_position, line)
-                y_position -= 0.2 * inch
-        
-        c.save()
+        doc = SimpleDocTemplate(pdf_file, pagesize=letter)
+        styles = getSampleStyleSheet()
+        story = []
+
+        story.append(Paragraph("Sentiment Analysis Report", styles['Title']))
+        story.append(Spacer(1, 12))
+        sentiment_text = f"Sentiment Score: {sentiment_score:.2f} (Scale: 0.0 Negative to 1.0 Positive)"
+        story.append(Paragraph(sentiment_text, styles['Normal']))
+        story.append(Spacer(1, 12))
+        interpretation = "Positive" if sentiment_score > 0.6 else "Negative" if sentiment_score < 0.4 else "Neutral"
+        story.append(Paragraph(f"Interpretation: {interpretation}", styles['Normal']))
+
+        doc.build(story)
         pdf_file.seek(0)
-        
+
         with open(pdf_path, "wb") as f:
             f.write(pdf_file.getvalue())
-        
-        logger.info(f"PDF generated: {pdf_path}")
-        return pdf_path, f"file://{pdf_path}", pdf_file
+
+        public_url = f"http://localhost:7860/static/output/{pdf_filename}"  # Placeholder URL; adjust as needed
+        logger.info(f"PDF generated: {public_url}")
+        return pdf_path, public_url, pdf_file
     except Exception as e:
         logger.error(f"Error generating PDF: {e}")
         return "", "", None
 
-def upload_pdf_to_salesforce(sf: Salesforce, pdf_file: BytesIO, report_id: str) -> str:
-    """Upload PDF to Salesforce and return the download URL"""
-    try:
-        if not pdf_file:
-            logger.error("No PDF file provided for upload")
-            return ""
-        
-        encoded_pdf = base64.b64encode(pdf_file.getvalue()).decode('utf-8')
-        content_version_data = {
-            "Title": f"Contract_Analysis_{int(time.time())}",
-            "PathOnClient": f"contract_analysis_{int(time.time())}.pdf",
-            "VersionData": encoded_pdf,
-            "FirstPublishLocationId": report_id
-        }
-        content_version = sf.ContentVersion.create(content_version_data)
-        result = sf.query(f"SELECT Id, ContentDocumentId FROM ContentVersion WHERE Id = '{content_version['id']}'")
-        
-        if not result['records']:
-            logger.error("Failed to retrieve ContentVersion")
-            return ""
-        
-        file_url = f"https://{sf.sf_instance}/sfc/servlet.shepherd/version/download/{content_version['id']}"
-        logger.info(f"PDF uploaded to Salesforce: {file_url}")
-        return file_url
-    except Exception as e:
-        logger.error(f"Error uploading PDF to Salesforce: {e}")
-        return ""
-
-def save_to_salesforce(sf: Salesforce, data: Dict, pdf_file: BytesIO) -> Dict:
-    """Save analysis results and PDF to Salesforce, return record and file URLs"""
-    try:
-        # Save Custom_Risk_Analysis__c record
-        record = {
-            'Sentiment_Score__c': data['sentiment_score'],
-            'Risk_Score__c': data['risk_score'],
-            'Risk_Level__c': data['risk_level'],
-            'Record_Id__c': data['record_id'],
-            'Penalty_Examples__c': data['penalty_examples'][:131072],
-            'Penalty_Details__c': data['penalty_details'][:131072],
-            'Penalty_Amounts__c': data['penalty_amounts'][:131072],
-            'Obligation_Details__c': data['obligation_details'][:131072],
-            'Delay_Details__c': data['delay_details'][:131072]
-        }
-        result = sf.Custom_Risk_Analysis__c.create(record)
-        logger.info(f"Successfully created Salesforce record: {result['id']}")
-        
-        # Upload PDF if available
-        download_url = upload_pdf_to_salesforce(sf, pdf_file, result['id']) if pdf_file else ""
-        
-        return {
-            'record_id': result['id'],
-            'download_url': download_url
-        }
-    except Exception as e:
-        logger.error(f"Failed to save to Salesforce: {str(e)}")
-        raise Exception(f"Salesforce record or file creation failed: {str(e)}")
-
 def extract_text_from_pdf(pdf_path: str) -> str:
     """Extract text from PDF using pdfplumber"""
     try:
@@ -666,16 +567,11 @@ def format_clause_example(example: str, index: int) -> str:
 def analyze_pdf(file_obj) -> List:
     """Main analysis function for Gradio interface"""
     try:
-        import tempfile
-        temp_dir = tempfile.mkdtemp()
-        
         if not file_obj:
             raise Exception("No PDF file uploaded. Please upload a valid PDF file.")
 
-        try:
-            sf = authenticate_salesforce()
-        except Exception as e:
-            raise Exception(f"Salesforce authentication failed: {str(e)}")
+        temp_dir = tempfile.mkdtemp(prefix="contract_analysis_")
+        logger.info(f"Created temporary directory for PDF processing: {temp_dir}")
 
         try:
             text = extract_text_from_pdf(file_obj.name)
@@ -758,29 +654,12 @@ def analyze_pdf(file_obj) -> List:
         extracted_data = "\n".join([format_clause_example(sent, i+1) for i, sent in enumerate(penalty_sentences[:3])]) if penalty_sentences else "<div class='success-box'>✅ No penalty clauses found - Excellent contract terms!</div>"
         
         record_id = str(uuid.uuid4())
-        sf_data = {
-            'sentiment_score': sentiment_score,
-            'risk_score': risk_score,
-            'risk_level': risk_level,
-            'record_id': record_id,
-            'penalty_counts': penalty_counts,
-            'penalty_details': "\n".join([f"{kw}: {count}" for kw, count in penalty_counts.items()]),
-            'penalty_amounts': "\n".join([f"${amt:,.2f}" for amt in penalty_values[:5]]) if penalty_values else "No specific penalty amounts found",
-            'obligation_counts': obligation_counts,
-            'obligation_details': "\n".join([f"{kw}: {count}" for kw, count in obligation_counts.items()]),
-            'delay_counts': delay_counts,
-            'delay_details': "\n".join([f"{kw}: {count}" for kw, count in delay_counts.items()]),
-            'penalty_examples': extracted_data
-        }
-        
         try:
-            pdf_path, pdf_url, pdf_file = generate_contract_analysis_pdf(sf_data, temp_dir)
-            salesforce_result = save_to_salesforce(sf, sf_data, pdf_file)
-            download_url = salesforce_result['download_url'] or pdf_url
-            logger.info(f"Saved to Salesforce with Record ID: {salesforce_result['record_id']}, PDF URL: {download_url}")
+            pdf_path, pdf_url, pdf_file = generate_sentiment_pdf(sentiment_score, temp_dir)
+            logger.info(f"Generated PDF with URL: {pdf_url}")
         except Exception as e:
-            logger.error(f"Salesforce record creation failed: {str(e)}")
-            download_url = pdf_url if pdf_url else ""
+            logger.error(f"PDF generation failed: {e}")
+            pdf_url = "Failed to generate PDF download link"
 
         box_class = "success-box" if risk_level == "Low" else "warning-box" if risk_level == "Medium" else "danger-box"
         risk_icon = "✅" if risk_level == "Low" else "⚠" if risk_level == "Medium" else "🚨"
@@ -800,8 +679,8 @@ def analyze_pdf(file_obj) -> List:
             </div>
             {sentiment_meter}
             <div style='margin-top: 15px;'>
-                <strong>Download Full Report:</strong>
-                {'<a href="' + download_url + '" download="contract_analysis.pdf" style="color: var(--primary-color); text-decoration: underline;">Download PDF</a>' if download_url else 'Failed to generate PDF download link'}
+                <strong>Download Sentiment Report:</strong>
+                <a href="{pdf_url}" target="_blank" style="color: var(--primary-color); text-decoration: underline;">Download PDF</a>
             </div>
         </div>
         """
@@ -844,7 +723,7 @@ def analyze_pdf(file_obj) -> List:
         """
         return [error_message] * 8
     finally:
-        if os.path.exists(temp_dir):
+        if 'temp_dir' in locals() and os.path.exists(temp_dir):
             shutil.rmtree(temp_dir, ignore_errors=True)
             logger.info(f"Cleaned up temporary directory: {temp_dir}")