import traceback, io, json, re, requests, fitz, base64, os, pickle, faiss from rest_framework.views import APIView from rest_framework.response import Response from rest_framework import status from mimetypes import guess_type from rest_framework.parsers import MultiPartParser, FormParser from .utils import ( extract_text_from_pdf, parse_prescription_data, parse_bill_data, calculate_basic_validation_results, calculate_intelligent_validation_results, extract_text_from_image_with_gemini ) from rest_framework.permissions import AllowAny import asyncio import concurrent.futures from asgiref.sync import sync_to_async import google.generativeai as genai from django.conf import settings from django.http import JsonResponse from PIL import Image import numpy as np from sentence_transformers import SentenceTransformer class PDFAnalysisAPIView(APIView): # parser_classes = (MultiPartParser, FormParser) permission_classes = [AllowAny] def post(self, request): # Check if this is a chat request is_chat = request.query_params.get('is_chat', '').lower() == 'true' if is_chat: return self.handle_chat_request(request) # Get uploaded files prescription_file = request.FILES.get('prescription_pdf') bill_file = request.FILES.get('bill_pdf') if not prescription_file: return Response({ "success": False, "status": 400, "message": "Prescription file is required" }, status=status.HTTP_400_BAD_REQUEST) if not bill_file: return Response({ "success": False, "status": 400, "message": "Bill file is required" }, status=status.HTTP_400_BAD_REQUEST) try: # Check file type for prescription mime_type_pres, _ = guess_type(prescription_file.name) if mime_type_pres and mime_type_pres.startswith("image"): prescription_text = extract_text_from_image_with_gemini(prescription_file) else: prescription_text = extract_text_from_pdf(prescription_file) # Check file type for bill mime_type_bill, _ = guess_type(bill_file.name) if mime_type_bill and mime_type_bill.startswith("image"): bill_text = extract_text_from_image_with_gemini(bill_file) else: bill_text = extract_text_from_pdf(bill_file) if not prescription_text: return Response({"success": False, "status": 400, "message": "Could not extract text from prescription"}, status=status.HTTP_400_BAD_REQUEST) if not bill_text: return Response({"success": False, "status": 400, "message": "Could not extract text from bill"}, status=status.HTTP_400_BAD_REQUEST) # With this async version: with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor: # Submit all AI tasks concurrently future_prescription_ai = executor.submit(self.get_ai_analysis, prescription_text, "prescription") future_bill_ai = executor.submit(self.get_ai_analysis, bill_text, "bill") # Get AI analysis results prescription_ai_analysis = future_prescription_ai.result() bill_ai_analysis = future_bill_ai.result() # Submit parsing tasks concurrently future_prescription_data = executor.submit(parse_prescription_data, prescription_text, prescription_ai_analysis) future_bill_data = executor.submit(parse_bill_data, bill_text, bill_ai_analysis) # Get parsing results prescription_data = future_prescription_data.result() bill_data = future_bill_data.result() # Submit validation tasks concurrently # basic_validation_results future_basic_validation = executor.submit(calculate_basic_validation_results, prescription_data, bill_data, prescription_text, bill_text) # basic_validation_results future_intelligent_validation = executor.submit(calculate_intelligent_validation_results, prescription_data, bill_data, prescription_text, bill_text) # Get validation results basic_validation_results = future_basic_validation.result() intelligent_validation_results = future_intelligent_validation.result() # Prepare response data response_data = { # Add ocrResults for frontend compatibility 'ocrResults': { 'prescription': { 'patientName': prescription_data.get('patientName', 'Not found'), 'doctorName': prescription_data.get('doctorName', 'Not found'), 'date': prescription_data.get('date', 'Not found'), 'medications': prescription_data.get('medications', []), 'tests': prescription_data.get('tests', []) }, 'bill': { 'hospitalName': bill_data.get('hospitalName', 'Not found'), 'billDate': bill_data.get('billDate', 'Not found'), 'totalAmount': bill_data.get('totalAmount', '$0.00'), 'medications': bill_data.get('medications', []), 'tests': bill_data.get('tests', []) } }, 'basic_validation': { # Original fields for frontend compatibility 'prescriptionValid': basic_validation_results.get('prescriptionValid', False), 'testsMatch': basic_validation_results.get('testsMatch', False), 'medicationsMatch': basic_validation_results.get('medicationsMatch', False), 'amountReasonable': basic_validation_results.get('amountReasonable', False), 'overallScore': basic_validation_results.get('overallScore', 0), # Additional detailed fields 'prescription_score': basic_validation_results.get('prescriptionScore', 0), 'tests_score': basic_validation_results.get('testsScore', 0), 'medications_score': basic_validation_results.get('medicationsScore', 0), 'amount_score': basic_validation_results.get('amountScore', 0), 'matched_medications': basic_validation_results.get('matchedMedications', []), 'matched_tests': basic_validation_results.get('matchedTests', []), 'unmatched_prescription_meds': basic_validation_results.get('unmatchedPrescriptionMeds', []), 'unmatched_bill_meds': basic_validation_results.get('unmatchedBillMeds', []), 'unmatched_prescription_tests': basic_validation_results.get('unmatchedPrescriptionTests', []), 'unmatched_bill_tests': basic_validation_results.get('unmatchedBillTests', []) }, 'intelligent_validation': { 'medical_accuracy': intelligent_validation_results.get('medicalAccuracy', {}), 'bill_accuracy': intelligent_validation_results.get('billAccuracy', {}), 'cross_validation': intelligent_validation_results.get('crossValidation', {}), 'fraud_detection': intelligent_validation_results.get('fraudDetection', {}), 'recommendations': intelligent_validation_results.get('recommendations', {}), 'overall_intelligent_score': intelligent_validation_results.get('overallIntelligentScore', 0), 'summary': intelligent_validation_results.get('summary', 'No summary available') }, 'ai_analysis': { 'prescription_analysis': prescription_ai_analysis, 'bill_analysis': bill_ai_analysis } } return Response({ "success": True, "status": 200, "message": "PDF analysis completed successfully", "data": response_data }, status=status.HTTP_200_OK) except Exception as e: return Response({ "success": False, "status": 500, "message": str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) def get_ai_analysis(self, text, document_type): """Get AI analysis for the document""" try: model = genai.GenerativeModel('gemini-1.5-flash') if document_type == "prescription": prompt = f""" Analyze this prescription document and provide a detailed analysis: {text} Please provide: 1. Patient information 2. Doctor information 3. Prescribed medications with dosages 4. Prescribed tests/procedures 5. Patient's ailment/condition 6. Any concerns or observations """ else: # bill prompt = f""" Analyze this medical bill document and provide a detailed analysis: {text} Please provide: 1. Hospital/clinic information 2. Billed medications with costs 3. Billed tests/procedures with costs 4. Total amount breakdown 5. Any concerns or observations about pricing """ response = model.generate_content(prompt) return response.text except Exception as e: return f"AI analysis failed: {str(e)}" def handle_chat_request(self, request): """Handle chat requests with direct AI interaction""" try: # Get the prompt from request data prompt = request.data.get('prompt', '').strip() if not prompt: return Response({ "success": False, "status": 400, "message": "Prompt is required for chat requests" }, status=status.HTTP_400_BAD_REQUEST) # Get AI response model = genai.GenerativeModel('gemini-1.5-flash') response = model.generate_content(prompt) return Response({ "success": True, "status": 200, "message": "Chat response generated successfully", "data": { "prompt": prompt, "response": response.text, "timestamp": "2025-06-17T09:25:20Z" # You can use datetime.now().isoformat() } }, status=status.HTTP_200_OK) except Exception as e: traceback_str = traceback.format_exc() # Captures full traceback as a string # print(traceback_str) # Logs to console; remove this in production return Response({ "success": False, "status": 500, "message": f"Chat request failed: {str(e)}" }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) # API to get analysis history (if you want to add this later) class PDFAnalysisHistoryView(APIView): def get(self, request): try: # You can implement history functionality here if needed return Response({ "success": True, "status": 200, "message": "Analysis history fetched successfully", "data": [] }, status=status.HTTP_200_OK) except Exception as e: return Response({ "success": False, "status": 500, "message": str(e) }, status=status.HTTP_500_INTERNAL_SERVER_ERROR) # ----------------- Gemini Vision API Helper ----------------- GEMINI_API_KEY = settings.GOOGLE_API_KEY GEMINI_VISION_URL = "https://generativelanguage.googleapis.com/v1/models/gemini-1.5-flash:generateContent" def extract_text_from_image_with_gemini(image_file): img_bytes = image_file.read() img_base64 = base64.b64encode(img_bytes).decode("utf-8") payload = { "contents": [{ "parts": [ {"text": "Extract all text clearly and completely from this prescription image."}, { "inline_data": { "mime_type": "image/jpeg", "data": img_base64 } } ] }] } resp = requests.post( f"{GEMINI_VISION_URL}?key={GEMINI_API_KEY}", json=payload ) resp.raise_for_status() data = resp.json() return data['candidates'][0]['content']['parts'][0]['text'] def extract_text_from_pdf(pdf_file): pdf_bytes = pdf_file.read() pdf_doc = fitz.open(stream=pdf_bytes, filetype="pdf") full_text = "" for page in pdf_doc: pix = page.get_pixmap() img = Image.open(io.BytesIO(pix.tobytes("png"))) buf = io.BytesIO() img.save(buf, format="JPEG") buf.seek(0) full_text += extract_text_from_image_with_gemini(buf) + "\n" return full_text.strip() def format_prescription_text_to_json(text): lines = [line.strip() for line in text.split("\n") if line.strip()] data = { "hospital_name": None, "address": None, "phone": None, "patient_name": None, "age": None, "gender": None, "diagnosis": None, "date": None, "medicines": [], "doctor_advice": [], "doctor_name": None, "specialization": None } # Basic extraction if lines: data["hospital_name"] = lines[0] for i, line in enumerate(lines): if re.search(r"street|road|india", line, re.IGNORECASE): data["address"] = line if re.search(r"phone", line, re.IGNORECASE): data["phone"] = line.split(":")[-1].strip() if re.search(r"Patient Name", line, re.IGNORECASE): data["patient_name"] = line.split(":")[-1].strip() if re.search(r"Age/Gender", line, re.IGNORECASE): ag = line.split(":")[-1].strip() if "/" in ag: age, gender = ag.split("/") data["age"] = age.strip() data["gender"] = gender.strip() if re.search(r"Diagnosis", line, re.IGNORECASE): data["diagnosis"] = line.split(":")[-1].strip() if re.search(r"Date", line, re.IGNORECASE): data["date"] = line.split(":")[-1].strip() if re.match(r"\d+\.", line): # medicine parsing med_parts = line.split(" - ") med_name_dose = med_parts[0].split(".", 1)[1].strip() med_freq = med_parts[1].strip() if len(med_parts) > 1 else None dose_match = re.search(r"(\d+mg.*)", med_name_dose, re.IGNORECASE) dose = dose_match.group(1) if dose_match else None name = med_name_dose.replace(dose, "").strip() if dose else med_name_dose med_obj = {"name": name} if dose: med_obj["dose"] = dose if med_freq: med_obj["frequency"] = med_freq data["medicines"].append(med_obj) if re.search(r"^•", line): data["doctor_advice"].append(line.replace("•", "").strip()) if re.search(r"Dr\.", line): data["doctor_name"] = line.strip() if "(" in line and ")" in line: data["specialization"] = re.search(r"\((.*?)\)", line).group(1) return data def gemini_text_analysis(text): prompt = f""" You are a professional medical AI assistant. Analyze the following prescription and return the result ONLY in valid JSON format. Prescription: {text} JSON Structure to return: {{ "is_safe": true/false, "doctor_analysis": "Detailed technical analysis for medical professionals.", "safety_check": "Summary of dosage correctness, drug interactions, and monitoring requirements.", "patient_summary": "Simple, patient-friendly explanation of treatment plan.", "issues": ["List of potential problems."], "suggestions": ["List of recommendations or precautions."] }} Rules: - Respond ONLY with raw JSON (no extra commentary, no markdown, no triple backticks). - Use proper boolean values for is_safe. - If data is missing (e.g., patient weight, renal function), mention it under issues. - Do not include any formatting like ```json or ``` in your response. """ payload = { "contents": [{ "parts": [{"text": prompt}] }] } resp = requests.post( f"https://generativelanguage.googleapis.com/v1/models/gemini-1.5-flash:generateContent?key={GEMINI_API_KEY}", json=payload ) resp.raise_for_status() data = resp.json() return data['candidates'][0]['content']['parts'][0]['text'] # ----------------- FAISS + SentenceTransformer Helper ----------------- def get_similar_chunks(query, top_k=5): faiss_index_path = os.path.join(settings.EMBEDDING_DIR, "faiss_index.bin") texts_path = os.path.join(settings.EMBEDDING_DIR, "texts.pkl") model = SentenceTransformer("all-MiniLM-L6-v2") index = faiss.read_index(faiss_index_path) with open(texts_path, "rb") as f: texts = pickle.load(f) query_embedding = model.encode([query]) distances, indices = index.search(np.array(query_embedding), top_k) return [texts[i] for i in indices[0]] def analyze_prescription_with_faiss(structured_data): # Convert structured prescription data to a single query string query = f"Patient: {structured_data.get('patient_name')}, Age: {structured_data.get('age')}, Gender: {structured_data.get('gender')}, Diagnosis: {structured_data.get('diagnosis')}, Medicines: {', '.join([m['name'] for m in structured_data.get('medicines', [])])}" chunks = get_similar_chunks(query=query, top_k=5) # Extract known medicines from FAISS chunks faiss_meds = set() for chunk in chunks: for word in chunk.split(","): faiss_meds.add(word.strip().lower()) # Compare AI-prescribed medicines with FAISS medicines ai_meds = set([m['name'].strip().lower() for m in structured_data.get('medicines', [])]) mismatches = { "not_in_faiss": list(ai_meds - faiss_meds), "extra_in_faiss": list(faiss_meds - ai_meds) } return { "faiss_context": chunks, "medicine_comparison": mismatches } class PrescriptionAnalysisAPIView(APIView): permission_classes = [AllowAny] parser_classes = (MultiPartParser, FormParser) def post(self, request): file = request.FILES.get('file') if not file: return JsonResponse({"error": "No file uploaded"}, status=400) mime_type = file.content_type if mime_type in ["image/jpeg", "image/png", "image/jpg"]: extracted_text = extract_text_from_image_with_gemini(file) elif mime_type == "application/pdf": extracted_text = extract_text_from_pdf(file) else: return JsonResponse({"error": "Unsupported file type"}, status=400) structured_ocr = format_prescription_text_to_json(extracted_text) ai_raw = gemini_text_analysis(extracted_text) # Clean JSON if fenced cleaned = ai_raw.strip() if cleaned.startswith("```"): cleaned = cleaned.strip("`") # remove "json" if exists after ``` cleaned = cleaned.replace("json\n", "").replace("json\r\n", "") try: ai_json = json.loads(cleaned) except Exception: return JsonResponse({ "ocrResults": extracted_text, "ai_analysis_raw": ai_raw, "error": "AI did not return valid JSON" }, status=500) # Integrate FAISS-based knowledge lookup + medicine comparison faiss_data = analyze_prescription_with_faiss(structured_ocr) return JsonResponse({ "ocrResults": structured_ocr, "analysis": ai_json, "faiss_analysis": faiss_data })