feat: used redacted version of the bank statment for the llm analysis

Benji918 · Benji918 · commit 11c43b5480c1 · 2025-09-22T01:03:58.000+01:00
diff --git a/app/api/v1/endpoints/statements.py b/app/api/v1/endpoints/statements.py
@@ -18,7 +18,7 @@
 import os
 import tempfile
 from app.services.pdf_service import PDFExcelService
-
+import io, re, fitz
 
 router = APIRouter()
 logger = get_logger(__name__)
@@ -310,7 +310,8 @@ async def analyze_statement(file: UploadFile = File(...)):
         "phone": r'\b(?:\+234|0)([789]\d{9})\b',
         "account_number": r'\b\d{10,20}\b',
         "address": r'\d+\s+\w+(?:\s+\w+)*\s+(Street|St|Avenue|Ave|Close|Rd|Road|Lane|Ln|Crescent|Cres)\b',
-        "name": r'\b[A-Z][a-z]+ [A-Z][a-z]+\b',
+         "name": r'\b(?:[A-Z][A-Za-z\'\.-]+(?:\s+[A-Z][A-Za-z\'\.-]+){1,3}'
+            r'|[A-Z]{2,}(?:\s+[A-Z]{2,}){1,3})\b',
         "bvn": r'\b\d{11}\b',
         "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
         "credit_card": r'\b(?:\d{4}[- ]?){3}\d{4}\b',
diff --git a/app/services/ai_service.py b/app/services/ai_service.py
@@ -14,201 +14,13 @@
     Insight, Recommendation, RiskAssessment
 )
 import json
-
+from app.services.pdf_service import PDFExcelService
 import re
 import hashlib
 from typing import Dict, Any, List
 from dataclasses import dataclass
 
-@dataclass
-class SanitizationConfig:
-    """Configuration for what data to sanitize"""
-    redact_account_numbers: bool = True
-    redact_phone_numbers: bool = True
-    redact_emails: bool = True
-    redact_addresses: bool = True
-    redact_names: bool = True
-    redact_ssn: bool = True
-    preserve_transaction_amounts: bool = True
-    preserve_dates: bool = True
-    preserve_merchant_names: bool = True
-
-class BankStatementSanitizer:
-    def __init__(self, config: SanitizationConfig = None):
-        self.config = config or SanitizationConfig()
-        self.replacement_map = {}  # Store original -> sanitized mappings
-        self.logger = LoggerMixin
-
-    def _generate_consistent_replacement(self, original_value: str, prefix: str) -> str:
-        """Generate consistent replacement for same values"""
-        self.logger.log_operation("generating_replacement", original_value=original_value, prefix=prefix)
-        if original_value in self.replacement_map:
-            return self.replacement_map[original_value]
-
-        # Create hash-based consistent replacement
-        hash_obj = hashlib.md5(original_value.encode())
-        hash_hex = hash_obj.hexdigest()[:8]
-        replacement = f"{prefix}_{hash_hex}"
-
-        self.replacement_map[original_value] = replacement
-        return replacement
-
-    def _sanitize_account_numbers(self, text: str) -> str:
-        """Replace account numbers with sanitized versions"""
-        self.logger.log_operation("sanitizing_account_numbers")
-        if not self.config.redact_account_numbers:
-            return text
-
-        # Pattern for account numbers (8-17 digits, possibly with dashes/spaces)
-        account_pattern = r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4,9}\b'
-
-        def replace_account(match):
-            account = match.group(0)
-            return self._generate_consistent_replacement(account, "ACCT")
-
-        return re.sub(account_pattern, replace_account, text)
-
-    def _sanitize_phone_numbers(self, text: str) -> str:
-        """Replace phone numbers"""
-        self.logger.log_operation("sanitizing_phone_numbers")
-        if not self.config.redact_phone_numbers:
-            return text
-
-        phone_patterns = [
-            r'\b\d{3}[-.\s]?\d{3}[-.\s]?\d{4}\b',  # US format
-            r'\(\d{3}\)\s?\d{3}[-.\s]?\d{4}',      # (123) 456-7890
-            r'\+\d{1,3}[-.\s]?\d{3,14}\b'          # International
-        ]
-
-        for pattern in phone_patterns:
-            def replace_phone(match):
-                phone = match.group(0)
-                return self._generate_consistent_replacement(phone, "PHONE")
-            text = re.sub(pattern, replace_phone, text)
-
-        return text
-
-    def _sanitize_emails(self, text: str) -> str:
-        """Replace email addresses"""
-        self.logger.log_operation("sanitizing_emails")
-        if not self.config.redact_emails:
-            return text
-
-        email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
-
-        def replace_email(match):
-            email = match.group(0)
-            return self._generate_consistent_replacement(email, "EMAIL")
-
-        return re.sub(email_pattern, replace_email, text)
-
-    def _sanitize_addresses(self, text: str) -> str:
-        """Replace street addresses (basic implementation)"""
-        self.logger.log_operation("sanitizing_addresses")
-        if not self.config.redact_addresses:
-            return text
-
-        # Simple address patterns - you may need more sophisticated NER
-        address_patterns = [
-            r'\d+\s+[A-Za-z\s]+(?:Street|St|Avenue|Ave|Road|Rd|Drive|Dr|Lane|Ln|Boulevard|Blvd|Way|Place|Pl)\b',
-            r'P\.?O\.?\s+Box\s+\d+',
-        ]
-
-        for pattern in address_patterns:
-            def replace_address(match):
-                address = match.group(0)
-                return self._generate_consistent_replacement(address, "ADDRESS")
-            text = re.sub(pattern, replace_address, text, flags=re.IGNORECASE)
-
-        return text
-
-    def _sanitize_ssn(self, text: str) -> str:
-        """Replace Social Security Numbers"""
-        if not self.config.redact_ssn:
-            return text
-
-        ssn_pattern = r'\b\d{3}[-\s]?\d{2}[-\s]?\d{4}\b'
 
-        def replace_ssn(match):
-            ssn = match.group(0)
-            return self._generate_consistent_replacement(ssn, "SSN")
-
-        return re.sub(ssn_pattern, replace_ssn, text)
-
-    def _sanitize_names(self, text: str) -> str:
-        """Replace common name patterns (basic implementation)"""
-        if not self.config.redact_names:
-            return text
-
-        # This is a basic implementation - consider using NER libraries like spaCy
-        # Pattern for capitalized words that might be names
-        name_indicators = [
-            r'\b(?:Mr\.?|Mrs\.?|Ms\.?|Dr\.?|Prof\.?)\s+[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*',
-            # Add more sophisticated name detection as needed
-        ]
-
-        for pattern in name_indicators:
-            def replace_name(match):
-                name = match.group(0)
-                return self._generate_consistent_replacement(name, "NAME")
-            text = re.sub(pattern, replace_name, text)
-
-        return text
-
-    def sanitize_text(self, text: str) -> str:
-        """Apply all sanitization rules to text"""
-        sanitized = text
-
-        # Apply sanitization in order
-        sanitized = self._sanitize_account_numbers(sanitized)
-        sanitized = self._sanitize_phone_numbers(sanitized)
-        sanitized = self._sanitize_emails(sanitized)
-        sanitized = self._sanitize_ssn(sanitized)
-        sanitized = self._sanitize_addresses(sanitized)
-        sanitized = self._sanitize_names(sanitized)
-
-        return sanitized
-
-    def sanitize_pdf(self, pdf_bytes: bytes) -> bytes:
-        """Sanitize a PDF document and return sanitized PDF bytes"""
-        self.logger.log_operation("sanitizing_pdf")
-        # Open PDF from bytes
-        doc = fitz.open(stream=pdf_bytes, filetype="pdf")
-
-        for page_num in range(len(doc)):
-            page = doc[page_num]
-
-            # Extract text blocks
-            text_dict = page.get_text("dict")
-
-            # Process each text block
-            for block in text_dict["blocks"]:
-                if "lines" in block:
-                    for line in block["lines"]:
-                        for span in line["spans"]:
-                            original_text = span["text"]
-                            sanitized_text = self.sanitize_text(original_text)
-
-                            if original_text != sanitized_text:
-                                # Replace text in PDF
-                                rect = fitz.Rect(span["bbox"])
-                                page.add_redact_annot(rect)
-                                page.apply_redactions()
-
-                                # Add sanitized text
-                                page.insert_text(
-                                    rect.tl,
-                                    sanitized_text,
-                                    fontsize=span["size"],
-                                    color=(0, 0, 0)
-                                )
-
-        # Return sanitized PDF as bytes
-        return doc.write()
-
-    def get_replacement_map(self) -> Dict[str, str]:
-        """Get the mapping of original -> sanitized values for audit purposes"""
-        return self.replacement_map.copy()
 
 class AIAnalysisService(LoggerMixin):
     """Service for AI-powered financial analysis using Google Gemini with direct file upload"""
@@ -422,45 +234,64 @@ async def analyze_financial_document(
         file_content: bytes, 
         filename: str,
         analysis_type: str = "comprehensive",
-        sanitization_config: SanitizationConfig = None
     ) -> Dict[str, Any]:
         """Perform comprehensive financial analysis using direct file upload to Gemini"""
+        patterns = {
+            "email": r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
+            "phone": r'\b(?:\+234|0)([789]\d{9})\b',
+            "account_number": r'\b\d{10,20}\b',
+            "address": r'\d+\s+\w+(?:\s+\w+)*\s+(Street|St|Avenue|Ave|Close|Rd|Road|Lane|Ln|Crescent|Cres)\b',
+            "name": r'\b(?:[A-Z][A-Za-z\'\.-]+(?:\s+[A-Z][A-Za-z\'\.-]+){1,3}'
+                    r'|[A-Z]{2,}(?:\s+[A-Z]{2,}){1,3})\b',
+            "bvn": r'\b\d{11}\b',
+            "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
+            "credit_card": r'\b(?:\d{4}[- ]?){3}\d{4}\b',
+            "routing_number": r'\b\d{9}\b',
+            "iban": r'\b[A-Za-z]{2}\d{2}[A-Za-z0-9]{11,30}\b',
+            # "swift_code": r'\b[A-Z]{4}[A-Z]{2}[A-Za-z0-9]{2}([A-Za-z0-9]{3})?\b',
+            # "currency": r'\b[\$€¥₹£]?\d{1,3}(,\d{3})*(\.\d{1,2})?\b',
+            "tin": r'\b\d{2}-\d{7}\b',
+            "pan": r'\b[A-Z]{5}\d{4}[A-Z]\b',
+            "gstin": r'\b\d{2}[A-Z]{5}\d{4}[A-Z][1-9A-Z][Z][A-Z0-9]\b',
+            # "cheque_number": r'\b\d{6,9}\b',
+            "reference_number": r'\b[A-Z]{3,4}-\d{6,10}\b',
+            # "business_reg": r'\b[A-Z0-9]{7,15}\b',
+            # "sort_code": r'\b\d{9}\b',
+            "ip_address": r'\b\d{1,3}(?:\.\d{1,3}){3}\b',
+            "ipv4": r'\b\d{1,3}(?:\.\d{1,3}){3}\b',
+            "url": r'\bhttps?://[^\s]+\b',
+            "merchant_id": r'\b(M|C)-[A-Za-z0-9]{6,12}\b',
+            "nin": r'\b\d{11}\b',
+        }
+
         try:
             self.log_operation("ai_analysis_start", filename=filename, analysis_type=analysis_type)
 
-            # sanitizer = BankStatementSanitizer(sanitization_config)
-            #
-            # if filename.lower().endswith('.pdf'):
-            #     sanitized_content = sanitizer.sanitize_pdf(file_content)
-            #
-            # else:
-            #     text_content = file_content.decode('utf-8', errors='ignore')
-            #     sanitized_text = sanitizer.sanitize_text(text_content)
-            #     sanitized_content = sanitized_text.encode('utf-8')
-            #
-            # replacement_count = len(sanitizer.get_replacement_map())
-            # self.log_operation("sanitization_complete",
-            #                    replacements_made=replacement_count)
 
             with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as temp_file:
                 temp_file.write(file_content)
                 temp_file.flush()
-                
+                tmp_path = temp_file.name
+
+                text = PDFExcelService.extract_text_from_pdf(tmp_path)
+                anonymized_text = PDFExcelService.anonymize_text(text, patterns)
+
+
                 try:
                     self.log_operation("uploading_file_to_gemini", filename=filename)
-                    uploaded_file = self.client.files.upload(
-                        file=temp_file.name,
-
-                    )
+                    # uploaded_file = self.client.files.upload(
+                    #     file=temp_file.name,
+                    #
+                    # )
 
                     import time
-                    while uploaded_file.state.name == "PROCESSING":
-                        self.log_operation("waiting_for_file_processing")
-                        time.sleep(2)
-                        uploaded_file = self.client.files.get(uploaded_file.name)
-                    
-                    if uploaded_file.state.name == "FAILED":
-                        raise ExternalServiceError("File processing failed in Gemini")
+                    # while uploaded_file.state.name == "PROCESSING":
+                    #     self.log_operation("waiting_for_file_processing")
+                    #     time.sleep(2)
+                    #     # uploaded_file = self.client.files.get(uploaded_file.name)
+                    #
+                    # if uploaded_file.state.name == "FAILED":
+                    #     raise ExternalServiceError("File processing failed in Gemini")
 
                     prompt = self._create_analysis_prompt(analysis_type)
                     
@@ -469,11 +300,11 @@ async def analyze_financial_document(
                     response = self.client.models.generate_content(
                         model=settings.GEMINI_MODEL,
                         contents=[
-                        uploaded_file,
+                        anonymized_text,
                         prompt
                     ])
 
-                    if not response or len(response.strip()) == 0:
+                    if not response:
                         raise ExternalServiceError("Gemini returned empty content.")
 
                     if hasattr(response, 'text'):
@@ -510,7 +341,7 @@ async def analyze_financial_document(
 
 
                     try:
-                        self.client.files.delete(name=uploaded_file.name)
+                        # self.client.files.delete(name=uploaded_file.name)
                         self.log_operation("gemini_file_cleanup_successful")
                     except Exception as e:
                         self.log_error(e, "gemini_file_cleanup_failed")
diff --git a/app/services/analysis_service.py b/app/services/analysis_service.py
@@ -40,20 +40,23 @@ async def create_analysis(
             if not statement:
                 raise ValidationError("Statement not found")
             
-            if statement.status != StatementStatus.UPLOADED:
-                raise ValidationError("Statement must be in uploaded status for analysis")
+            # if statement.status != StatementStatus.UPLOADED:
+            #     raise ValidationError("Statement must be in uploaded status for analysis")
             
 
-            from app.services.statement_service import statement_service
-            statement_service.update_processing_status(
-                db, statement_id, StatementStatus.PROCESSING
-            )
+            # from app.services.statement_service import statement_service
+            # statement_service.update_processing_status(
+            #     db, statement_id, StatementStatus.PROCESSING
+            # )
             
 
             pdf_content = await file_service.download_from_cloudinary(
                 statement.cloudinary_public_id
             )
 
+            self.log_operation("downloaded file from cloudinary", statement_id=statement_id, user_id=user_id)
+
+
 
             start_time = datetime.utcnow()
             self.log_operation("create_analysis", statement_id=statement_id, user_id=user_id)
@@ -120,9 +123,9 @@ async def create_analysis(
             db.add(statement)
             
 
-            statement_service.update_processing_status(
-                db, statement_id, StatementStatus.COMPLETED
-            )
+            # statement_service.update_processing_status(
+            #     db, statement_id, StatementStatus.COMPLETED
+            # )
             
             self.log_operation(
                 "create_analysis_process_time",
diff --git a/app/services/pdf_service.py b/app/services/pdf_service.py
