Add automatic PII removal during file extraction

Dockerfile

@@ -8,6 +8,9 @@ COPY requirements.txt .
 # Install dependencies
 RUN pip install --no-cache-dir -r requirements.txt
 
+# Download spaCy model for Presidio (required for PII detection)
+RUN python -m spacy download en_core_web_lg
+
 # Copy the entire backend
 COPY . .
 

requirements.txt

@@ -1,7 +1,9 @@
 fastapi==0.104.1
 uvicorn==0.24.0
 python-dotenv==1.0.0
-groq==0.9.0
-httpx==0.27.0
+groq==0.4.1
 pydantic==2.5.0
-python-multipart==0.0.6
+python-multipart==0.0.6
+presidio-analyzer==2.2.354
+presidio-anonymizer==2.2.354
+spacy==3.7.2

src/api/routes.py

@@ -29,7 +29,7 @@ async def analyze_provider_notes(request: ProviderNotesRequest):
                 detail="Provider notes must be at least 10 characters long"
             )
         
-        # Process through Groq service - CORRECT METHOD NAME
+        # Process through Groq service
         result = await groq_service.analyze_provider_notes(provider_notes)
         
         logger.info("Successfully processed coding request")
@@ -52,43 +52,55 @@ async def analyze_provider_notes(request: ProviderNotesRequest):
         )
 
 
-# NEW ENDPOINT - File Upload
+# UPDATED ENDPOINT - File Upload with PII Removal
 @router.post("/upload-file", response_model=FileUploadResponse)
 async def upload_provider_notes_file(file: UploadFile = File(...)):
     """
     Upload a TXT file containing provider notes and extract ICD-10 and CPT codes
     
-    This endpoint accepts a TXT file, extracts the text, and processes it through the LLM.
+    This endpoint:
+    1. Extracts text from uploaded TXT file
+    2. Automatically detects and removes patient personal information (PII)
+    3. Processes sanitized text through LLM
+    4. Returns ICD-10 and CPT codes
     
     Args:
         file: TXT file containing provider notes
         
     Returns:
-        FileUploadResponse with extracted codes and explanations
+        FileUploadResponse with codes, explanations, and PII removal info
     """
     try:
-        logger.info(f"Received file upload request: {file.filename}")
+        logger.info(f"📁 Received file upload request: {file.filename}")
         
-        # Step 1: Extract text from uploaded file
-        extraction_result = await file_service.extract_text_from_file(file)
+        # Step 1: Extract text from file with automatic PII removal
+        extraction_result = await file_service.extract_text_from_file(
+            file=file,
+            remove_pii=True  # Always remove PII for safety
+        )
         
         extracted_text = extraction_result["text"]
         filename = extraction_result["filename"]
         text_length = extraction_result["text_length"]
+        pii_info = extraction_result["pii_info"]
+        
+        logger.info(f"✅ Extracted {text_length} characters from {filename}")
         
-        logger.info(f"Extracted {text_length} characters from {filename}")
+        if pii_info["pii_removed"]:
+            logger.info(f"🔒 Removed {pii_info['pii_count']} PII entities before processing")
         
-        # Step 2: Process extracted text through Groq LLM
-        # FIXED: Use the correct method name 'analyze_provider_notes'
+        # Step 2: Process sanitized text through Groq LLM
         coding_result = await groq_service.analyze_provider_notes(extracted_text)
         
-        logger.info(f"Successfully processed file: {filename}")
+        logger.info(f"✅ Successfully processed file: {filename}")
         
-        # Step 3: Return combined response
+        # Step 3: Return combined response with PII info
         return FileUploadResponse(
             success=True,
             filename=filename,
             extracted_text_length=text_length,
+            pii_removed=pii_info["pii_removed"],
+            pii_count=pii_info["pii_count"],
             cpt_codes=coding_result.get("CPT", []),
             cpt_explanation=coding_result.get("CPT_explanation", ""),
             icd_codes=coding_result.get("ICD", []),
@@ -98,7 +110,7 @@ async def upload_provider_notes_file(file: UploadFile = File(...)):
     except HTTPException:
         raise
     except Exception as e:
-        logger.error(f"Error in upload_provider_notes_file: {str(e)}")
+        logger.error(f"❌ Error in upload_provider_notes_file: {str(e)}")
         raise HTTPException(
             status_code=500,
             detail=f"Error processing uploaded file: {str(e)}"

src/models/response_models.py

@@ -1,38 +1,64 @@
 from pydantic import BaseModel, Field
-from typing import List
+from typing import List, Optional
 
-class ICDCode(BaseModel):
-    code: str = Field(..., description="ICD-10 diagnosis code")
-    description: str = Field(..., description="Description of the diagnosis")
-    explanation: str = Field(..., description="Explanation for why this code was selected")
+class ProviderNotesRequest(BaseModel):
+    provider_notes: str = Field(
+        ..., 
+        description="The medical provider notes to analyze",
+        min_length=10,
+        example="Patient presents with acute bronchitis. Performed comprehensive examination and prescribed antibiotics."
+    )
+    
+    class Config:
+        json_schema_extra = {
+            "example": {
+                "provider_notes": "Patient presents with acute bronchitis. Cough for 5 days, productive with yellow sputum. Lung exam reveals diffuse wheezing. Prescribed azithromycin 500mg."
+            }
+        }
 
-class CPTCode(BaseModel):
-    code: str = Field(..., description="CPT procedure code")
-    description: str = Field(..., description="Description of the procedure/service")
-    explanation: str = Field(..., description="Explanation for why this code was selected")
+class ProviderNote(BaseModel):
+    note: str
 
 class CodingResponse(BaseModel):
-    icd_codes: List[ICDCode] = Field(default_factory=list, description="List of ICD-10 codes")
-    cpt_codes: List[CPTCode] = Field(default_factory=list, description="List of CPT codes")
-    overall_summary: str = Field(..., description="Overall summary of coding decisions")
+    cpt_codes: list
+    cpt_explanation: str
+    icd_codes: list
+    icd_explanation: str
+
+
+# PII Detection Detail Model
+class PIIDetail(BaseModel):
+    """Details of detected PII entity"""
+    entity_type: str = Field(description="Type of PII detected (e.g., PERSON, PHONE_NUMBER)")
+    start: int = Field(description="Start position in original text")
+    end: int = Field(description="End position in original text")
+    score: float = Field(description="Confidence score")
+
+
+# Updated File Upload Response with PII info
+class FileUploadResponse(BaseModel):
+    """Response model for file upload endpoint with PII removal info"""
+    success: bool = Field(description="Whether file processing was successful")
+    filename: str = Field(description="Name of uploaded file")
+    extracted_text_length: int = Field(description="Length of extracted text (after PII removal)")
+    pii_removed: bool = Field(description="Whether PII was detected and removed")
+    pii_count: int = Field(description="Number of PII entities removed")
+    cpt_codes: list = Field(description="List of CPT codes")
+    cpt_explanation: str = Field(description="Explanation of CPT codes")
+    icd_codes: list = Field(description="List of ICD codes")
+    icd_explanation: str = Field(description="Explanation of ICD codes")
     
     class Config:
         json_schema_extra = {
             "example": {
-                "icd_codes": [
-                    {
-                        "code": "J20.9",
-                        "description": "Acute bronchitis, unspecified",
-                        "explanation": "Patient presents with acute bronchitis as documented in provider notes"
-                    }
-                ],
-                "cpt_codes": [
-                    {
-                        "code": "99213",
-                        "description": "Office visit, established patient",
-                        "explanation": "Comprehensive examination performed as documented"
-                    }
-                ],
-                "overall_summary": "Patient encounter for acute bronchitis with examination and treatment"
+                "success": True,
+                "filename": "provider_notes.txt",
+                "extracted_text_length": 450,
+                "pii_removed": True,
+                "pii_count": 3,
+                "cpt_codes": ["99213", "93000"],
+                "cpt_explanation": "Office visit and EKG",
+                "icd_codes": ["I20.0", "R07.9"],
+                "icd_explanation": "Unstable angina and chest pain"
             }
         }

src/services/file_service.py

@@ -2,6 +2,7 @@ from fastapi import UploadFile, HTTPException
 import os
 from typing import Dict
 import logging
+from services.pii_detector import pii_detector
 
 logger = logging.getLogger(__name__)
 
@@ -36,15 +37,16 @@ class FileService:
             )
     
     @staticmethod
-    async def extract_text_from_file(file: UploadFile) -> Dict[str, any]:
+    async def extract_text_from_file(file: UploadFile, remove_pii: bool = True) -> Dict[str, any]:
         """
-        Extract text content from uploaded file
+        Extract text content from uploaded file and optionally remove PII
         
         Args:
             file: Uploaded file object
+            remove_pii: Whether to remove PII from extracted text (default: True)
             
         Returns:
-            Dictionary containing extracted text and metadata
+            Dictionary containing extracted text, PII removal info, and metadata
         """
         try:
             # Validate file
@@ -86,19 +88,43 @@ class FileService:
                     detail="Extracted text is too short. Please provide more detailed provider notes"
                 )
             
-            logger.info(f"Successfully extracted {len(text)} characters from {file.filename}")
+            logger.info(f"✅ Successfully extracted {len(text)} characters from {file.filename}")
+            
+            # Remove PII if requested
+            pii_info = {
+                "pii_removed": False,
+                "pii_count": 0,
+                "pii_details": []
+            }
+            
+            if remove_pii:
+                logger.info("🔒 Removing PII from extracted text...")
+                pii_result = pii_detector.remove_pii(text)
+                
+                text = pii_result["sanitized_text"]
+                pii_info = {
+                    "pii_removed": pii_result["was_pii_removed"],
+                    "pii_count": pii_result["pii_count"],
+                    "pii_details": pii_result["pii_detected"]
+                }
+                
+                if pii_result["was_pii_removed"]:
+                    logger.info(f"✅ Removed {pii_result['pii_count']} PII entities")
+                else:
+                    logger.info("✅ No PII detected in text")
             
             return {
                 "text": text,
                 "filename": file.filename,
                 "file_size": file_size,
-                "text_length": len(text)
+                "text_length": len(text),
+                "pii_info": pii_info
             }
             
         except HTTPException:
             raise
         except Exception as e:
-            logger.error(f"Error extracting text from file: {str(e)}")
+            logger.error(f"❌ Error extracting text from file: {str(e)}")
             raise HTTPException(
                 status_code=500,
                 detail=f"Error processing file: {str(e)}"

src/services/pii_detector.py

@@ -0,0 +1,197 @@
+from presidio_analyzer import AnalyzerEngine
+from presidio_anonymizer import AnonymizerEngine
+from typing import Dict, List
+import re
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class PIIDetector:
+    """Service to detect and remove Personal Identifiable Information from medical notes"""
+    
+    def __init__(self):
+        """Initialize PII detection engines"""
+        try:
+            self.analyzer = AnalyzerEngine()
+            self.anonymizer = AnonymizerEngine()
+            
+            # Entities to detect (common in medical notes)
+            self.entities_to_detect = [
+                "PERSON",              # Names
+                "EMAIL_ADDRESS",       # Email
+                "PHONE_NUMBER",        # Phone numbers
+                "US_SSN",             # Social Security Number
+                "CREDIT_CARD",        # Credit card numbers
+                "US_DRIVER_LICENSE",  # Driver's license
+                "LOCATION",           # Addresses, cities
+                "DATE_TIME",          # Birth dates, appointment dates
+                "US_PASSPORT",        # Passport numbers
+                "MEDICAL_LICENSE",    # Medical license numbers
+                "IP_ADDRESS",         # IP addresses
+                "URL"                 # URLs
+            ]
+            
+            logger.info("✅ PII Detector initialized successfully")
+        except Exception as e:
+            logger.error(f"❌ Failed to initialize PII Detector: {str(e)}")
+            raise
+    
+    def detect_pii(self, text: str) -> List[Dict]:
+        """
+        Detect PII entities in text
+        
+        Args:
+            text: Input text to analyze
+            
+        Returns:
+            List of detected PII entities with details
+        """
+        try:
+            results = self.analyzer.analyze(
+                text=text,
+                entities=self.entities_to_detect,
+                language='en'
+            )
+            
+            pii_findings = []
+            for result in results:
+                pii_findings.append({
+                    "entity_type": result.entity_type,
+                    "start": result.start,
+                    "end": result.end,
+                    "score": result.score,
+                    "text": text[result.start:result.end]
+                })
+            
+            logger.info(f"🔍 Detected {len(pii_findings)} PII entities")
+            return pii_findings
+            
+        except Exception as e:
+            logger.error(f"❌ Error detecting PII: {str(e)}")
+            return []
+    
+    def remove_pii(self, text: str) -> Dict[str, any]:
+        """
+        Remove PII from text while preserving medical information
+        
+        Args:
+            text: Input text containing potential PII
+            
+        Returns:
+            Dictionary with sanitized text and PII removal report
+        """
+        try:
+            # Step 1: Detect PII
+            analyzer_results = self.analyzer.analyze(
+                text=text,
+                entities=self.entities_to_detect,
+                language='en'
+            )
+            
+            if not analyzer_results:
+                logger.info("✅ No PII detected in text")
+                return {
+                    "sanitized_text": text,
+                    "pii_detected": [],
+                    "pii_count": 0,
+                    "was_pii_removed": False
+                }
+            
+            # Step 2: Anonymize detected PII
+            anonymized_result = self.anonymizer.anonymize(
+                text=text,
+                analyzer_results=analyzer_results
+            )
+            
+            sanitized_text = anonymized_result.text
+            
+            # Step 3: Additional pattern-based cleaning for medical notes
+            # Replace common medical note PII patterns
+            sanitized_text = self._clean_medical_patterns(sanitized_text)
+            
+            # Step 4: Collect PII detection details
+            pii_detected = []
+            for result in analyzer_results:
+                pii_detected.append({
+                    "entity_type": result.entity_type,
+                    "start": result.start,
+                    "end": result.end,
+                    "score": result.score
+                })
+            
+            logger.info(f"✅ Removed {len(pii_detected)} PII entities from text")
+            
+            return {
+                "sanitized_text": sanitized_text,
+                "pii_detected": pii_detected,
+                "pii_count": len(pii_detected),
+                "was_pii_removed": True
+            }
+            
+        except Exception as e:
+            logger.error(f"❌ Error removing PII: {str(e)}")
+            # Return original text if PII removal fails
+            return {
+                "sanitized_text": text,
+                "pii_detected": [],
+                "pii_count": 0,
+                "was_pii_removed": False,
+                "error": str(e)
+            }
+    
+    def _clean_medical_patterns(self, text: str) -> str:
+        """
+        Clean common medical note PII patterns that might be missed
+        
+        Args:
+            text: Text to clean
+            
+        Returns:
+            Cleaned text
+        """
+        # Pattern 1: "Patient: <NAME>" or "Pt: <NAME>"
+        text = re.sub(
+            r'(Patient|Pt|Patient Name):\s*<[A-Z_]+>',
+            r'\1: [REDACTED]',
+            text,
+            flags=re.IGNORECASE
+        )
+        
+        # Pattern 2: "DOB: <DATE>"
+        text = re.sub(
+            r'(DOB|Date of Birth|Birth Date):\s*<[A-Z_]+>',
+            r'\1: [REDACTED]',
+            text,
+            flags=re.IGNORECASE
+        )
+        
+        # Pattern 3: "Address: <LOCATION>"
+        text = re.sub(
+            r'(Address|Addr|Home Address):\s*<[A-Z_]+>',
+            r'\1: [REDACTED]',
+            text,
+            flags=re.IGNORECASE
+        )
+        
+        # Pattern 4: "Phone: <PHONE_NUMBER>"
+        text = re.sub(
+            r'(Phone|Tel|Telephone|Cell|Mobile):\s*<[A-Z_]+>',
+            r'\1: [REDACTED]',
+            text,
+            flags=re.IGNORECASE
+        )
+        
+        # Pattern 5: "MRN: <NUMBER>" (Medical Record Number)
+        text = re.sub(
+            r'(MRN|Medical Record Number|Record #):\s*<[A-Z_]+>',
+            r'\1: [REDACTED]',
+            text,
+            flags=re.IGNORECASE
+        )
+        
+        return text
+
+
+# Singleton instance
+pii_detector = PIIDetector()

tests/test_api.py

@@ -23,7 +23,6 @@ def test_coding_endpoint():
 
 def test_file_upload_endpoint():
     """Test new file upload endpoint"""
-    # Create a sample TXT file
     file_content = b"Patient John Doe presents with acute bronchitis. Cough for 5 days, productive with yellow sputum. Lung exam reveals diffuse wheezing."
     
     files = {
@@ -38,10 +37,36 @@ def test_file_upload_endpoint():
     assert data["success"] is True
     assert data["filename"] == "provider_notes.txt"
     assert data["extracted_text_length"] > 0
+    assert "pii_removed" in data
+    assert "pii_count" in data
     assert "cpt_codes" in data
     assert "icd_codes" in data
-    assert isinstance(data["cpt_codes"], list)
-    assert isinstance(data["icd_codes"], list)
+
+
+def test_file_upload_with_pii():
+    """Test file upload with PII - should be removed"""
+    file_content = b"""
+    Patient: John Doe
+    DOB: 01/15/1980
+    Phone: 555-123-4567
+    Address: 123 Main St, New York, NY
+    
+    Chief Complaint: Chest pain
+    History: Patient presents with acute chest pain...
+    """
+    
+    files = {
+        "file": ("notes_with_pii.txt", BytesIO(file_content), "text/plain")
+    }
+    
+    response = client.post("/api/upload-file", files=files)
+    
+    assert response.status_code == 200
+    data = response.json()
+    
+    # PII should be detected and removed
+    assert data["pii_removed"] is True
+    assert data["pii_count"] > 0
 
 
 def test_file_upload_invalid_extension():