Create utils.py

utils.py

@@ -0,0 +1,64 @@
+import re
+import hashlib
+import io
+import json
+from datetime import datetime
+from typing import Dict, List, Tuple
+from bson import ObjectId
+import logging
+from config import logger
+
+def clean_text_response(text: str) -> str:
+    text = re.sub(r'\n\s*\n', '\n\n', text)
+    text = re.sub(r'[ ]+', ' ', text)
+    return text.replace("**", "").replace("__", "").strip()
+
+def extract_section(text: str, heading: str) -> str:
+    try:
+        pattern = rf"{re.escape(heading)}:\s*\n(.*?)(?=\n[A-Z][^\n]*:|\Z)"
+        match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
+        return match.group(1).strip() if match else ""
+    except Exception as e:
+        logger.error(f"Section extraction failed for heading '{heading}': {e}")
+        return ""
+
+def structure_medical_response(text: str) -> Dict:
+    def extract_improved(text: str, heading: str) -> str:
+        patterns = [
+            rf"{re.escape(heading)}:\s*\n(.*?)(?=\n\s*\n|\Z)",
+            rf"\*\*{re.escape(heading)}\*\*:\s*\n(.*?)(?=\n\s*\n|\Z)",
+            rf"{re.escape(heading)}[\s\-]+(.*?)(?=\n\s*\n|\Z)",
+            rf"\n{re.escape(heading)}\s*\n(.*?)(?=\n\s*\n|\Z)"
+        ]
+        for pattern in patterns:
+            match = re.search(pattern, text, re.DOTALL | re.IGNORECASE)
+            if match:
+                content = match.group(1).strip()
+                content = re.sub(r'^\s*[\-\*]\s*', '', content, flags=re.MULTILINE)
+                return content
+        return ""
+    
+    text = text.replace('**', '').replace('__', '')
+    return {
+        "summary": extract_improved(text, "Summary of Patient's Medical History") or 
+                  extract_improved(text, "Summarize the patient's medical history"),
+        "risks": extract_improved(text, "Identify Risks or Red Flags") or 
+                extract_improved(text, "Risks or Red Flags"),
+        "missed_issues": extract_improved(text, "Missed Diagnoses or Treatments") or 
+                       extract_improved(text, "What the doctor might have missed"),
+        "recommendations": extract_improved(text, "Suggest Next Clinical Steps") or 
+                         extract_improved(text, "Suggested Clinical Actions")
+    }
+
+def serialize_patient(patient: dict) -> dict:
+    patient_copy = patient.copy()
+    if "_id" in patient_copy:
+        patient_copy["_id"] = str(patient_copy["_id"])
+    return patient_copy
+
+def compute_patient_data_hash(data: dict) -> str:
+    serialized = json.dumps(data, sort_keys=True)
+    return hashlib.sha256(serialized.encode()).hexdigest()
+
+def compute_file_content_hash(file_content: bytes) -> str:
+    return hashlib.sha256(file_content).hexdigest()