import PyPDF2 import re from collections import Counter import nltk from nltk.tokenize import sent_tokenize, word_tokenize from nltk.corpus import stopwords from nltk.stem import WordNetLemmatizer import string from datetime import datetime, timedelta import json import torch from transformers import T5ForConditionalGeneration, T5Tokenizer, pipeline, BlipProcessor, BlipForConditionalGeneration import warnings import fitz # PyMuPDF from PIL import Image, ImageEnhance, ImageFilter import io import base64 import os import pytesseract import hashlib import logging import getpass import tempfile import shutil from fastapi import FastAPI from fastapi.staticfiles import StaticFiles from pydantic import BaseModel from typing import List, Dict, Any, Optional from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC try: import cv2 import numpy as np OPENCV_AVAILABLE = True except ImportError: print("OpenCV not available. Using PIL for image preprocessing.") OPENCV_AVAILABLE = False import numpy as np warnings.filterwarnings('ignore') app = FastAPI(title='AI (PDF→Summary+QnA+Scores)', version='0.2.1') app.mount("/static", StaticFiles(directory="static"), name="static") class HIPAALogger: """HIPAA-compliant audit logging system""" def __init__(self, log_file="hipaa_audit.log"): self.log_file = log_file self.setup_logging() def setup_logging(self): """Setup secure audit logging""" logging.basicConfig( filename=self.log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) self.logger = logging.getLogger('HIPAA_AUDIT') def log_access(self, user_id, action, resource, success=True): """Log access attempts and actions""" status = "SUCCESS" if success else "FAILURE" message = f"USER:{user_id} ACTION:{action} RESOURCE:{resource} STATUS:{status}" self.logger.info(message) def log_phi_processing(self, user_id, document_hash, action): """Log PHI processing events""" message = f"PHI_PROCESSING USER:{user_id} DOC_HASH:{document_hash} ACTION:{action}" self.logger.info(message) class SecureFileHandler: """Secure file handling with encryption and secure deletion""" def __init__(self, password=None): self.password = password self.key = self._derive_key(password) if password else None self.fernet = Fernet(self.key) if self.key else None def _derive_key(self, password): """Derive encryption key from password""" password_bytes = password.encode() kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=b'hipaa_thesis_analyzer_salt', iterations=100000, ) key = base64.urlsafe_b64encode(kdf.derive(password_bytes)) return key def encrypt_data(self, data): """Encrypt sensitive data""" if not self.fernet: return data if isinstance(data, str): data = data.encode() return self.fernet.encrypt(data) def decrypt_data(self, encrypted_data): """Decrypt sensitive data""" if not self.fernet: return encrypted_data decrypted = self.fernet.decrypt(encrypted_data) return decrypted.decode() def secure_save(self, data, filepath): """Save data with encryption""" if self.fernet: encrypted_data = self.encrypt_data(json.dumps(data)) with open(filepath + '.enc', 'wb') as f: f.write(encrypted_data) else: with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2) def secure_load(self, filepath): """Load encrypted data""" if self.fernet and os.path.exists(filepath + '.enc'): with open(filepath + '.enc', 'rb') as f: encrypted_data = f.read() decrypted_data = self.decrypt_data(encrypted_data) return json.loads(decrypted_data) elif os.path.exists(filepath): with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) return None def secure_delete(self, filepath): """Securely delete files by overwriting""" if os.path.exists(filepath): # Overwrite file multiple times before deletion file_size = os.path.getsize(filepath) with open(filepath, 'rb+') as f: for _ in range(3): # DoD 5220.22-M standard f.seek(0) f.write(os.urandom(file_size)) f.flush() os.remove(filepath) # Also check for encrypted version if os.path.exists(filepath + '.enc'): file_size = os.path.getsize(filepath + '.enc') with open(filepath + '.enc', 'rb+') as f: for _ in range(3): f.seek(0) f.write(os.urandom(file_size)) f.flush() os.remove(filepath + '.enc') class HIPAACompliantThesisAnalyzer: """HIPAA-compliant version of the thesis analyzer""" def __init__(self, user_id=None, password=None, session_timeout=30): self.user_id = user_id or getpass.getuser() self.session_timeout = session_timeout # minutes self.session_start = datetime.now() self.last_activity = datetime.now() # Initialize HIPAA compliance components self.hipaa_logger = HIPAALogger() self.secure_handler = SecureFileHandler(password) # Log session start self.hipaa_logger.log_access(self.user_id, "SESSION_START", "THESIS_ANALYZER") # Initialize base analyzer components self._initialize_analyzer() print(f"HIPAA-Compliant Thesis Analyzer initialized for user: {self.user_id}") print(f"Session timeout: {session_timeout} minutes") print(f"Encryption enabled: {'Yes' if password else 'No'}") def _initialize_analyzer(self): """Initialize the core analyzer components""" try: self.lemmatizer = WordNetLemmatizer() self.stop_words = set(stopwords.words('english')) except LookupError as e: print(f"NLTK resource error: {e}") self._download_nltk_resources() self.lemmatizer = WordNetLemmatizer() self.stop_words = set(stopwords.words('english')) self.thesis_text = "" self.sentences = [] self.key_terms = [] self.extracted_images = [] self.image_descriptions = [] self.ocr_results = [] self.use_ocr = True self.use_blip = True # Initialize T5 model print("Loading T5-small model (HIPAA-compliant local processing)...") self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.model_name = "t5-small" self.tokenizer = T5Tokenizer.from_pretrained(self.model_name) self.model = T5ForConditionalGeneration.from_pretrained(self.model_name) self.model.to(self.device) # Initialize pipelines self.summarizer = pipeline( "summarization", model=self.model_name, tokenizer=self.model_name, device=0 if torch.cuda.is_available() else -1, max_length=200, min_length=150, do_sample=True, temperature=0.7 ) self.qa_pipeline = pipeline( "text2text-generation", model=self.model_name, tokenizer=self.model_name, device=0 if torch.cuda.is_available() else -1, max_length=512, do_sample=True, temperature=0.7 ) # Initialize BLIP if enabled if self.use_blip: try: self.blip_processor = BlipProcessor.from_pretrained("Salesforce/blip-image-captioning-base") self.blip_model = BlipForConditionalGeneration.from_pretrained("Salesforce/blip-image-captioning-base") self.blip_model.to(self.device) print("BLIP model loaded for local image analysis") except Exception as e: print(f"BLIP model loading failed: {e}") self.use_blip = False # Check OCR availability if self.use_ocr: try: pytesseract.get_tesseract_version() print("Tesseract OCR available for local processing") except Exception as e: print(f"Tesseract OCR not available: {e}") self.use_ocr = False def _download_nltk_resources(self): """Download required NLTK resources""" resources = [ ('tokenizers/punkt', 'punkt'), ('tokenizers/punkt_tab', 'punkt_tab'), ('corpora/stopwords', 'stopwords'), ('corpora/wordnet', 'wordnet'), ('corpora/omw-1.4', 'omw-1.4') ] for resource_path, resource_name in resources: try: nltk.data.find(resource_path) except LookupError: try: nltk.download(resource_name, quiet=True) except Exception as e: print(f"Warning: Failed to download {resource_name}: {e}") def check_session_timeout(self): """Check if session has timed out""" time_since_start = datetime.now() - self.session_start time_since_activity = datetime.now() - self.last_activity if time_since_activity.total_seconds() > (self.session_timeout * 60): self.hipaa_logger.log_access(self.user_id, "SESSION_TIMEOUT", "THESIS_ANALYZER") raise Exception("Session timed out due to inactivity. Please restart for security.") self.last_activity = datetime.now() def calculate_document_hash(self, content): """Calculate secure hash of document content""" return hashlib.sha256(content.encode()).hexdigest() def process_document_securely(self, pdf_path, questions, output_file=None): """Process document with full HIPAA compliance""" self.check_session_timeout() # Calculate document hash for audit trail with open(pdf_path, 'rb') as f: doc_content = f.read() doc_hash = hashlib.sha256(doc_content).hexdigest()[:16] self.hipaa_logger.log_phi_processing(self.user_id, doc_hash, "DOCUMENT_LOAD") try: # Extract text and images text, images = self._extract_text_and_images(pdf_path) self.hipaa_logger.log_phi_processing(self.user_id, doc_hash, "TEXT_EXTRACTION") # Perform OCR if enabled ocr_results = [] if self.use_ocr and images: ocr_results = self._perform_secure_ocr(images) self.hipaa_logger.log_phi_processing(self.user_id, doc_hash, "OCR_PROCESSING") # Analyze images if BLIP enabled image_descriptions = [] if self.use_blip and images: image_descriptions = self._analyze_images_securely(images) self.hipaa_logger.log_phi_processing(self.user_id, doc_hash, "IMAGE_ANALYSIS") # Combine all text ocr_text = " ".join([result['ocr_text'] for result in ocr_results if result.get('ocr_text')]) combined_text = text + " " + ocr_text # Generate analysis sections = self._extract_key_sections(combined_text) key_terms = self._extract_key_terms(combined_text) summary = self._generate_summary_secure(combined_text) question_answers = self._answer_questions_secure(questions, combined_text) self.hipaa_logger.log_phi_processing(self.user_id, doc_hash, "ANALYSIS_COMPLETE") # Compile HIPAA-compliant report report = { "hipaa_compliance": { "processed_locally": True, "encrypted_storage": bool(self.secure_handler.fernet), "audit_logged": True, "user_id": self.user_id, "session_id": hashlib.md5(f"{self.user_id}{self.session_start}".encode()).hexdigest()[:8], "document_hash": doc_hash, "processing_timestamp": datetime.now().isoformat(), "no_external_apis": True, "local_processing_only": True }, "document_info": { "file_path": os.path.basename(pdf_path), # Only filename for privacy "analysis_timestamp": datetime.now().isoformat(), "total_characters": len(text), "total_images": len(images), "device_used": str(self.device) }, "text_analysis": { "summary": summary, "key_terms": key_terms[:15], "sections_found": list(sections.keys()) }, "image_analysis": { "total_images_extracted": len(images), "images_with_text": len([r for r in ocr_results if r.get('has_text', False)]), "ocr_available": self.use_ocr, "blip_available": self.use_blip }, "question_responses": question_answers, "statistics": { "total_text_characters": len(text), "ocr_text_characters": len(ocr_text), "questions_processed": len(questions), "sections_identified": len(sections), "key_terms_extracted": len(key_terms) } } # Save securely if output file specified if output_file: self.secure_handler.secure_save(report, output_file) self.hipaa_logger.log_access(self.user_id, "REPORT_SAVE", output_file) return report except Exception as e: self.hipaa_logger.log_access(self.user_id, "PROCESSING_ERROR", pdf_path, success=False) raise e def _extract_text_and_images(self, pdf_path): """Securely extract text and images from PDF""" text = "" images = [] try: # Use PyMuPDF for comprehensive extraction doc = fitz.open(pdf_path) for page_num in range(len(doc)): page = doc.load_page(page_num) # Extract text page_text = page.get_text() if page_text.strip(): text += page_text + "\n" # Extract images image_list = page.get_images() for img_index, img in enumerate(image_list): try: xref = img[0] pix = fitz.Pixmap(doc, xref) if pix.n - pix.alpha < 4: img_data = pix.tobytes("ppm") img_pil = Image.open(io.BytesIO(img_data)) image_info = { 'page': page_num + 1, 'index': img_index, 'image': img_pil, 'size': img_pil.size, 'format': img_pil.format or 'Unknown' } images.append(image_info) pix = None except Exception as e: print(f"Error extracting image {img_index} from page {page_num + 1}: {e}") continue doc.close() except Exception as e: print(f"Error in secure extraction: {e}") return text, images def _perform_secure_ocr(self, images): """Perform OCR with audit logging""" ocr_results = [] for i, img_info in enumerate(images): try: img = img_info['image'] if img.mode != 'RGB': img = img.convert('RGB') # Preprocess for OCR if OPENCV_AVAILABLE: img_array = np.array(img) gray = cv2.cvtColor(img_array, cv2.COLOR_RGB2GRAY) denoised = cv2.medianBlur(gray, 3) clahe = cv2.createCLAHE(clipLimit=2.0, tileGridSize=(8,8)) enhanced = clahe.apply(denoised) _, thresh = cv2.threshold(enhanced, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) processed_img = Image.fromarray(thresh) else: gray = img.convert('L') enhancer = ImageEnhance.Contrast(gray) enhanced = enhancer.enhance(2.0) processed_img = enhanced.filter(ImageFilter.SHARPEN) # Perform OCR locally ocr_text = pytesseract.image_to_string(processed_img, config='--psm 6') ocr_result = { 'page': img_info['page'], 'image_index': img_info['index'], 'ocr_text': ocr_text.strip(), 'has_text': bool(ocr_text.strip()), 'processing_method': 'Local_OCR' } ocr_results.append(ocr_result) except Exception as e: ocr_results.append({ 'page': img_info['page'], 'image_index': img_info['index'], 'ocr_text': '', 'has_text': False, 'error': str(e) }) return ocr_results def _analyze_images_securely(self, images): """Analyze images locally with BLIP""" if not self.use_blip: return [] descriptions = [] for img_info in images: try: image = img_info['image'] if image.mode != 'RGB': image = image.convert('RGB') inputs = self.blip_processor(image, return_tensors="pt").to(self.device) with torch.no_grad(): out = self.blip_model.generate(**inputs, max_length=100, num_beams=5) caption = self.blip_processor.decode(out[0], skip_special_tokens=True) description = { 'page': img_info['page'], 'image_index': img_info['index'], 'caption': caption, 'processing_method': 'Local_BLIP' } descriptions.append(description) except Exception as e: descriptions.append({ 'page': img_info['page'], 'image_index': img_info['index'], 'caption': 'Analysis failed', 'error': str(e) }) return descriptions def _extract_key_sections(self, text): """Extract key sections from text""" sections = {} section_patterns = { 'abstract': r'abstract\s*:?\s*(.*?)(?=\n\s*(?:introduction|chapter|acknowledgment|table of contents))', 'introduction': r'introduction\s*:?\s*(.*?)(?=\n\s*(?:literature review|methodology|chapter|background))', 'methodology': r'(?:methodology|methods)\s*:?\s*(.*?)(?=\n\s*(?:results|findings|analysis|chapter))', 'results': r'(?:results|findings)\s*:?\s*(.*?)(?=\n\s*(?:discussion|conclusion|chapter))', 'conclusion': r'conclusion\s*:?\s*(.*?)(?=\n\s*(?:references|bibliography|appendix))' } for section_name, pattern in section_patterns.items(): match = re.search(pattern, text.lower(), re.DOTALL | re.IGNORECASE) if match: sections[section_name] = match.group(1).strip()[:1000] # Truncate for privacy return sections def _extract_key_terms(self, text): """Extract key terms securely""" try: words = re.findall(r'\b[a-zA-Z]+\b', text.lower()) words = [ self.lemmatizer.lemmatize(word) for word in words if word not in self.stop_words and len(word) > 3 and word.isalpha() ] word_freq = Counter(words) return [term for term, freq in word_freq.most_common(20)] except Exception as e: print(f"Error in key term extraction: {e}") return [] def _generate_summary_secure(self, text): """Generate summary using local T5 model""" try: clean_text = re.sub(r'\s+', ' ', text).strip() # Chunk text for processing max_length = 1000 if len(clean_text) > max_length: clean_text = clean_text[:max_length] summary = self.summarizer( clean_text, max_length=200, min_length=150, do_sample=True, temperature=0.7 ) return summary[0]['summary_text'] except Exception as e: print(f"Error in T5 summarization: {e}") # Fallback to extractive summary sentences = re.split(r'[.!?]+', text) return " ".join(sentences[:3]) + "..." def _answer_questions_secure(self, questions, text): """Answer questions using local T5 model""" answers = {} for question in questions: try: prompt = f"question: {question} context: {text[:1000]}" answer_result = self.qa_pipeline( prompt, max_length=200, min_length=30, do_sample=True, temperature=0.7, num_return_sequences=1 ) answer = answer_result[0]['generated_text'] answer = re.sub(r'^(answer:|Answer:)', '', answer).strip() answers[question] = { 'answer': answer, 'method': 'Local_T5', 'processed_securely': True } except Exception as e: answers[question] = { 'answer': 'Unable to process question securely', 'error': str(e), 'method': 'Error' } return answers def cleanup_session(self): """Clean up session data securely""" self.hipaa_logger.log_access(self.user_id, "SESSION_END", "THESIS_ANALYZER") # Clear sensitive data from memory self.thesis_text = "" self.extracted_images = [] self.ocr_results = [] self.image_descriptions = [] # Clear model cache if needed if hasattr(torch.cuda, 'empty_cache'): torch.cuda.empty_cache() print("Session cleaned up securely") class AnalyzeReq(BaseModel): storageKey: str # path to PDF on disk (or adjust to your storage scheme) projectId: Optional[str] = None documentId: Optional[str] = None ocr: bool = False blip: bool = False userId:str password:str useEncryption: bool =False @app.post('/analyze') def analyze(req: AnalyzeReq): """Main function with HIPAA compliance demonstration""" print("HIPAA-COMPLIANT THESIS ANALYZER") print("=" * 50) try: # Initialize HIPAA-compliant analyzer analyzer = HIPAACompliantThesisAnalyzer( user_id=req.userId, password=req.password, session_timeout=30 ) pdf_path = req.storageKey # Sample questions questions = [ "What is the main objective of the research?", "What methodology was used in the study?", "What are the key findings or results?", "What conclusions did the authors draw?", "What are the limitations of the study?", "What motivated the researchers to conduct this study?", "How does this research relate to existing literature?", "What are the practical implications of the findings?", "What assumptions underlie the research?", "What statistical methods were used to analyze the data?", "How robust are the study’s findings?", "Are there any potential biases in the study design or data collection?", "How do the results compare with previous studies on the same topic?", "What are the potential future applications of this research?", "How could this research be expanded or built upon in future studies?", "What new questions have emerged as a result of this study?" ] # Process document securely print("\nProcessing document with HIPAA compliance...") report = analyzer.process_document_securely( pdf_path=pdf_path, questions=questions, output_file="hipaa_compliant_analysis" ) print("\n" + "="*60) print("HIPAA-COMPLIANT ANALYSIS COMPLETE") print("="*60) print(f"✓ Processed locally: {report['hipaa_compliance']['processed_locally']}") print(f"✓ Encrypted storage: {report['hipaa_compliance']['encrypted_storage']}") print(f"✓ Audit logged: {report['hipaa_compliance']['audit_logged']}") print(f"✓ No external APIs: {report['hipaa_compliance']['no_external_apis']}") print(f"✓ Session ID: {report['hipaa_compliance']['session_id']}") # Cleanup analyzer.cleanup_session() return report except Exception as e: print(f"Error: {e}") print("Ensure all requirements are installed and Tesseract is available.") #if __name__ == "__main__": print(""" HIPAA-COMPLIANT THESIS ANALYZER =============================== HIPAA COMPLIANCE FEATURES: ✓ Local processing only - no external API calls ✓ Encryption at rest with password protection ✓ Comprehensive audit logging ✓ Session timeout and access controls ✓ Secure file deletion ✓ PHI processing audit trail ✓ User authentication ✓ Data integrity verification INSTALLATION: pip install torch transformers PyPDF2 nltk PyMuPDF pillow pytesseract cryptography SECURITY FEATURES: - All processing happens locally - Optional file encryption - Secure memory cleanup - Audit trail for all operations - Session management with timeouts - Secure file overwriting for deletion COMPLIANCE NOTES: - This tool provides technical safeguards - You must implement administrative and physical safeguards - Ensure your workstation meets HIPAA requirements - Regular security assessments recommended """) #main()