try to fix the identifier again

utils/oneclick.py

@@ -3,7 +3,7 @@ from typing import Tuple, Optional, Dict
 from .meldrx import MeldRxAPI
 from .responseparser import PatientDataExtractor
 from .pdfutils import PDFGenerator
-from .verifier import DischargeVerifier  # Import the verifier
+from .verifier import DischargeVerifier
 import logging
 import json
 from huggingface_hub import InferenceClient
@@ -16,7 +16,7 @@ if not HF_TOKEN:
     raise ValueError("HF_TOKEN environment variable not set.")
 client = InferenceClient(api_key=HF_TOKEN)
 MODEL_NAME = "meta-llama/Llama-3.3-70B-Instruct"
-verifier = DischargeVerifier()  # Initialize the verifier
+verifier = DischargeVerifier()
 
 def generate_ai_discharge_summary(patient_dict: Dict[str, str], client) -> Tuple[Optional[str], Optional[str]]:
     """Generate a discharge summary using AI and verify it for hallucinations."""
@@ -113,14 +113,16 @@ def generate_discharge_paper_one_click(
             logger.debug(f"Patient {i}: ID={patient_id_from_data}, Name={first_name_from_data} {last_name_from_data}")
             logger.debug(f"Comparing - Input: ID={patient_id_input}, First={first_name_input}, Last={last_name_input}")
             
-            matches = True
-            if patient_id_input and patient_id_from_data and patient_id_input != patient_id_from_data:
-                matches = False
-            if first_name_input and first_name_input != first_name_from_data:
-                matches = False
-            if last_name_input and last_name_input != last_name_from_data:
-                matches = False
-                
+            # Match logic: ID takes precedence, then first/last name
+            matches = False
+            if patient_id_input and patient_id_from_data == patient_id_input:
+                matches = True
+            elif not patient_id_input and first_name_input and last_name_input:
+                if first_name_input == first_name_from_data and last_name_input == last_name_from_data:
+                    matches = True
+            elif not patient_id_input and not first_name_input and not last_name_input:
+                continue  # Skip if no criteria provided
+            
             if matches:
                 matching_patients.append(patient_data)
                 logger.info(f"Found matching patient: ID={patient_id_from_data}, "
@@ -134,8 +136,8 @@ def generate_discharge_paper_one_click(
             return None, (f"No patients found matching criteria: {search_criteria}\n"
                          f"Available IDs: {', '.join(all_patient_ids)}\n"
                          f"Available Names: {', '.join(all_patient_names)}"), None, None, None
-        logger.debug(f"Raw patient data from API: {json.dumps(patients_data, indent=2)}")
-        patient_data = matching_patients[0]
+        
+        patient_data = matching_patients[0]  # Take the first match
         logger.info(f"Selected patient data: {patient_data}")
         
         basic_summary = format_discharge_summary(patient_data)
@@ -158,13 +160,12 @@ def generate_discharge_paper_one_click(
 
 def format_discharge_summary(patient_data: dict) -> str:
     """Format patient data into a discharge summary text."""
-    patient_data.setdefault('name_prefix', '')
+    patient_data.setdefault('id', 'Unknown')
     patient_data.setdefault('first_name', '')
     patient_data.setdefault('last_name', '')
     patient_data.setdefault('dob', 'Unknown')
     patient_data.setdefault('age', 'Unknown')
     patient_data.setdefault('sex', 'Unknown')
-    patient_data.setdefault('id', 'Unknown')
     patient_data.setdefault('address', 'Unknown')
     patient_data.setdefault('city', 'Unknown')
     patient_data.setdefault('state', 'Unknown')
@@ -186,11 +187,11 @@ def format_discharge_summary(patient_data: dict) -> str:
         "DISCHARGE SUMMARY",
         "",
         "PATIENT INFORMATION",
-        f"Name: {patient_data['name_prefix']} {patient_data['first_name']} {patient_data['last_name']}".strip(),
+        f"Name: {patient_data['first_name']} {patient_data['last_name']}".strip(),
+        f"Patient ID: {patient_data['id']}",
         f"Date of Birth: {patient_data['dob']}",
         f"Age: {patient_data['age']}",
         f"Gender: {patient_data['sex']}",
-        f"Patient ID: {patient_data['id']}",
         "",
         "CONTACT INFORMATION",
         f"Address: {patient_data['address']}",

utils/responseparser.py

@@ -1,3 +1,4 @@
+# utils/responseparser.py
 import json
 import lxml.etree as etree
 from datetime import datetime
@@ -5,7 +6,6 @@ from typing import List, Dict, Optional, Union
 import base64
 import logging
 
-# Set up logging
 logging.basicConfig(
     level=logging.INFO,
     format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
@@ -42,7 +42,7 @@ class PatientDataExtractor:
     def _extract_patients(self) -> List:
         """Extract all patient entries based on format."""
         if self.format == "xml":
-            return [self.data]  # C-CDA has one patient per document
+            return [self.data]
         elif self.format == "json":
             if self.data.get("resourceType") != "Bundle" or "entry" not in self.data:
                 raise ValueError("Invalid FHIR Bundle format")
@@ -65,33 +65,15 @@ class PatientDataExtractor:
             id_list = patient.xpath("//hl7:recordTarget/hl7:patientRole/hl7:id/@extension", namespaces=self.ns)
             return id_list[0] if id_list else ""
         elif self.format == "json":
-            # Check top-level 'id' first
             patient_id = patient.get("id", "")
             if patient_id:
                 return patient_id
-            # Fallback to 'identifier' field
             identifiers = patient.get("identifier", [])
             for identifier in identifiers:
-                if identifier.get("value"):  # Return the first non-empty identifier value
+                if identifier.get("value"):
                     return identifier["value"]
-            return ""  # Default to empty string if no ID found
-
-    def get_resource_type(self) -> str:
-        patient = self._get_current_patient()
-        if self.format == "xml":
-            return "ClinicalDocument"
-        elif self.format == "json":
-            return patient.get("resourceType", "")
-
-    def get_meta_last_updated(self) -> str:
-        patient = self._get_current_patient()
-        if self.format == "xml":
-            time_list = patient.xpath("//hl7:effectiveTime/@value", namespaces=self.ns)
-            return time_list[0] if time_list else ""
-        elif self.format == "json":
-            return patient.get("meta", {}).get("lastUpdated", "")
+            return ""
 
-    # Name Fields
     def get_first_name(self) -> str:
         patient = self._get_current_patient()
         if self.format == "xml":
@@ -114,18 +96,6 @@ class PatientDataExtractor:
                     return name["family"]
         return ""
 
-    def get_name_prefix(self) -> str:
-        patient = self._get_current_patient()
-        if self.format == "xml":
-            prefix = patient.xpath("//hl7:recordTarget/hl7:patientRole/hl7:patient/hl7:name/hl7:prefix/text()", namespaces=self.ns)
-            return prefix[0] if prefix else ""
-        elif self.format == "json":
-            for name in patient.get("name", []):
-                if name.get("use") == "official" and "prefix" in name:
-                    return name["prefix"][0]
-        return ""
-
-    # Demographic Fields
     def get_dob(self) -> str:
         patient = self._get_current_patient()
         if self.format == "xml":
@@ -139,7 +109,7 @@ class PatientDataExtractor:
         if not dob:
             return ""
         try:
-            birth_date = datetime.strptime(dob[:8], "%Y%m%d")
+            birth_date = datetime.strptime(dob[:8], "%Y%m%d") if len(dob) >= 8 else datetime.strptime(dob, "%Y-%m-%d")
             today = datetime.now()
             age = today.year - birth_date.year - ((today.month, today.day) < (birth_date.month, birth_date.day))
             return str(age)
@@ -154,7 +124,6 @@ class PatientDataExtractor:
         elif self.format == "json":
             return patient.get("gender", "").capitalize()
 
-    # Address Fields
     def get_address_line(self) -> str:
         patient = self._get_current_patient()
         if self.format == "xml":
@@ -191,7 +160,6 @@ class PatientDataExtractor:
             addresses = patient.get("address", [])
             return addresses[0]["postalCode"] if addresses and "postalCode" in addresses[0] else ""
 
-    # Contact Fields
     def get_phone(self) -> str:
         patient = self._get_current_patient()
         if self.format == "xml":
@@ -203,50 +171,6 @@ class PatientDataExtractor:
                     return telecom.get("value", "")
         return ""
 
-    # Extensions and Additional Fields
-    def get_race(self) -> str:
-        patient = self._get_current_patient()
-        if self.format == "xml":
-            race = patient.xpath("//hl7:recordTarget/hl7:patientRole/hl7:patient/hl7:raceCode/@displayName", namespaces=self.ns)
-            return race[0] if race else ""
-        elif self.format == "json":
-            for ext in patient.get("extension", []):
-                if ext.get("url") == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-race":
-                    for sub_ext in ext.get("extension", []):
-                        if sub_ext.get("url") == "text":
-                            return sub_ext.get("valueString", "")
-        return ""
-
-    def get_ethnicity(self) -> str:
-        patient = self._get_current_patient()
-        if self.format == "xml":
-            ethnicity = patient.xpath("//hl7:recordTarget/hl7:patientRole/hl7:patient/hl7:ethnicGroupCode/@displayName", namespaces=self.ns)
-            return ethnicity[0] if ethnicity else ""
-        elif self.format == "json":
-            for ext in patient.get("extension", []):
-                if ext.get("url") == "http://hl7.org/fhir/us/core/StructureDefinition/us-core-ethnicity":
-                    for sub_ext in ext.get("extension", []):
-                        if sub_ext.get("url") == "text":
-                            return sub_ext.get("valueString", "")
-        return ""
-
-    def get_language(self) -> str:
-        patient = self._get_current_patient()
-        if self.format == "xml":
-            lang = patient.xpath("//hl7:recordTarget/hl7:patientRole/hl7:patient/hl7:languageCommunication/hl7:languageCode/@code", namespaces=self.ns)
-            return lang[0] if lang else ""
-        elif self.format == "json":
-            comms = patient.get("communication", [])
-            if comms and "language" in comms[0]:
-                lang = comms[0]["language"]
-                # Try 'text' first, then fall back to 'coding' if available
-                if "text" in lang:
-                    return lang["text"]
-                elif "coding" in lang and lang["coding"]:
-                    return lang["coding"][0].get("display", lang["coding"][0].get("code", ""))
-            return ""
-
-    # Medications
     def get_medications(self) -> List[Dict[str, str]]:
         if self.format == "xml":
             section = self.data.xpath("//hl7:section[hl7:code/@code='10160-0']", namespaces=self.ns)
@@ -278,7 +202,6 @@ class PatientDataExtractor:
                     result.append({"start": start, "stop": stop, "description": desc, "code": code})
             return result
 
-    # Encounters
     def get_encounters(self) -> List[Dict[str, str]]:
         if self.format == "xml":
             service = self.data.xpath("//hl7:documentationOf/hl7:serviceEvent", namespaces=self.ns)
@@ -302,7 +225,6 @@ class PatientDataExtractor:
                     result.append({"start": start, "end": end, "description": desc, "code": code})
             return result
 
-    # Conditions/Diagnoses
     def get_conditions(self) -> List[Dict[str, str]]:
         if self.format == "xml":
             section = self.data.xpath("//hl7:section[hl7:code/@code='11450-4']", namespaces=self.ns)
@@ -331,107 +253,16 @@ class PatientDataExtractor:
                     result.append({"onset": onset, "description": desc, "code": code})
             return result
 
-    # Immunizations
-    def get_immunizations(self) -> List[Dict[str, str]]:
-        if self.format == "xml":
-            section = self.data.xpath("//hl7:section[hl7:code/@code='11369-6']", namespaces=self.ns)
-            if not section:
-                return []
-            immunizations = section[0].xpath(".//hl7:substanceAdministration", namespaces=self.ns)
-            result = []
-            for imm in immunizations:
-                date_list = imm.xpath(".//hl7:effectiveTime/@value", namespaces=self.ns)
-                date = date_list[0] if date_list else ""
-                desc_list = imm.xpath(".//hl7:consumable/hl7:manufacturedProduct/hl7:manufacturedMaterial/hl7:code/@displayName", namespaces=self.ns)
-                desc = desc_list[0] if desc_list else ""
-                code_list = imm.xpath(".//hl7:consumable/hl7:manufacturedProduct/hl7:manufacturedMaterial/hl7:code/@code", namespaces=self.ns)
-                code = code_list[0] if code_list else ""
-                result.append({"date": date, "description": desc, "code": code})
-            return result
-        elif self.format == "json":
-            entries = self.data.get("entry", [])
-            result = []
-            for entry in entries:
-                if entry["resource"]["resourceType"] == "Immunization":
-                    imm = entry["resource"]
-                    date = imm.get("occurrenceDateTime", "")
-                    desc = imm.get("vaccineCode", {}).get("text", "")
-                    code = imm.get("vaccineCode", {}).get("coding", [{}])[0].get("code", "")
-                    result.append({"date": date, "description": desc, "code": code})
-            return result
-
-    # Diagnostic Reports
-    def get_diagnostic_reports(self) -> List[Dict[str, str]]:
-        if self.format == "xml":
-            section = self.data.xpath("//hl7:section[hl7:code/@code='30954-2']", namespaces=self.ns)
-            if not section:
-                return []
-            reports = section[0].xpath(".//hl7:organizer", namespaces=self.ns)
-            result = []
-            for report in reports:
-                start_list = report.xpath(".//hl7:effectiveTime/hl7:low/@value", namespaces=self.ns)
-                start = start_list[0] if start_list else ""
-                desc_list = report.xpath(".//hl7:code/@displayName", namespaces=self.ns)
-                desc = desc_list[0] if desc_list else ""
-                code_list = report.xpath(".//hl7:code/@code", namespaces=self.ns)
-                code = code_list[0] if code_list else ""
-                result.append({"start": start, "description": desc, "code": code})
-            return result
-        elif self.format == "json":
-            entries = self.data.get("entry", [])
-            result = []
-            for entry in entries:
-                if entry["resource"]["resourceType"] == "DiagnosticReport":
-                    report = entry["resource"]
-                    start = report.get("effectiveDateTime", "")
-                    desc = report.get("code", {}).get("text", "")
-                    code = report.get("code", {}).get("coding", [{}])[0].get("code", "")
-                    data = report.get("presentedForm", [{}])[0].get("data", "")
-                    if data:
-                        decoded = base64.b64decode(data).decode('utf-8')
-                        result.append({"start": start, "description": desc, "code": code, "content": decoded})
-                    else:
-                        result.append({"start": start, "description": desc, "code": code})
-            return result
-
-    # Comprehensive Extraction
-    def get_all_patient_data(self) -> Dict[str, Union[str, List, Dict]]:
-        """Extract all available data for the current patient."""
-        return {
-            "id": self.get_id(),
-            "resource_type": self.get_resource_type(),
-            "meta_last_updated": self.get_meta_last_updated(),
-            "first_name": self.get_first_name(),
-            "last_name": self.get_last_name(),
-            "name_prefix": self.get_name_prefix(),
-            "dob": self.get_dob(),
-            "age": self.get_age(),
-            "gender": self.get_gender(),
-            "address_line": self.get_address_line(),
-            "city": self.get_city(),
-            "state": self.get_state(),
-            "zip_code": self.get_zip_code(),
-            "phone": self.get_phone(),
-            "race": self.get_race(),
-            "ethnicity": self.get_ethnicity(),
-            "language": self.get_language(),
-            "medications": self.get_medications(),
-            "encounters": self.get_encounters(),
-            "conditions": self.get_conditions(),
-            "immunizations": self.get_immunizations(),
-            "diagnostic_reports": self.get_diagnostic_reports()
-        }
-
     def get_patient_dict(self) -> Dict[str, str]:
         """Return a dictionary of patient data mapped to discharge form fields."""
         data = self.get_all_patient_data()
         latest_encounter = data["encounters"][-1] if data["encounters"] else {}
         latest_condition = data["conditions"][-1] if data["conditions"] else {}
-        medications_str = "; ".join([m["description"] for m in data["medications"]])
+        medications_str = "; ".join([m["description"] for m in data["medications"] if m["description"]])
         return {
+            "id": data["id"],
             "first_name": data["first_name"],
             "last_name": data["last_name"],
-            "middle_initial": "",
             "dob": data["dob"],
             "age": data["age"],
             "sex": data["gender"],
@@ -439,25 +270,37 @@ class PatientDataExtractor:
             "city": data["city"],
             "state": data["state"],
             "zip_code": data["zip_code"],
-            "doctor_first_name": "",
+            "phone": data["phone"],
+            "doctor_first_name": "",  # Could be extracted from Practitioner resource if available
             "doctor_last_name": "",
-            "doctor_middle_initial": "",
             "hospital_name": "",
             "doctor_address": "",
             "doctor_city": "",
             "doctor_state": "",
             "doctor_zip": "",
             "admission_date": latest_encounter.get("start", ""),
-            "referral_source": "",
-            "admission_method": "",
             "discharge_date": latest_encounter.get("end", ""),
-            "discharge_reason": "",
-            "date_of_death": "",
             "diagnosis": latest_condition.get("description", ""),
-            "procedures": "",
-            "medications": medications_str,
-            "preparer_name": "",
-            "preparer_job_title": ""
+            "medications": medications_str if medications_str else "None specified",
+        }
+
+    def get_all_patient_data(self) -> Dict[str, Union[str, List, Dict]]:
+        """Extract all available data for the current patient."""
+        return {
+            "id": self.get_id(),
+            "first_name": self.get_first_name(),
+            "last_name": self.get_last_name(),
+            "dob": self.get_dob(),
+            "age": self.get_age(),
+            "gender": self.get_gender(),
+            "address_line": self.get_address_line(),
+            "city": self.get_city(),
+            "state": self.get_state(),
+            "zip_code": self.get_zip_code(),
+            "phone": self.get_phone(),
+            "medications": self.get_medications(),
+            "encounters": self.get_encounters(),
+            "conditions": self.get_conditions(),
         }
 
     def get_all_patients(self) -> List[Dict[str, str]]:
@@ -468,8 +311,4 @@ class PatientDataExtractor:
             self.set_patient_by_index(i)
             all_patients.append(self.get_patient_dict())
         self.set_patient_by_index(original_idx)
-        return all_patients
-
-    def get_patient_ids(self) -> List[str]:
-        """Return a list of all patient IDs."""
-        return [self.get_id() for _ in self.patients]
+        return all_patients