Spaces:
Runtime error
Runtime error
| from utils.db.db import mongo | |
| from utils.logging.logger_setup import logger | |
| def get_patient_rag_data(patient_id): | |
| patient_data = [] | |
| patient = mongo.db.patient.find_one({"$or": [{"sql_id": patient_id}, {"id": patient_id}]}) | |
| if not patient: | |
| return [] | |
| actual_p_id = patient.get('sql_id', patient.get('id')) | |
| pid_query = {"$in": [actual_p_id, str(actual_p_id)]} | |
| patient_data.append({'content': f"Patient Name: {patient.get('name')}, Age: {patient.get('age')}, Gender: {patient.get('gender')}, Contact: {patient.get('contact')}"}) | |
| history_records = list(mongo.db.medicalhistory.find({"patient_id": pid_query})) | |
| for h in history_records: | |
| content = f"Medical History - Condition: {h.get('condition')}, Diagnosis Date: {h.get('diagnosis_date')}, Notes: {h.get('notes','N/A')}, Medications: {h.get('medications_history','N/A')}, Allergies: {h.get('allergies','N/A')}" | |
| patient_data.append({'content': content}) | |
| consultations = list(mongo.db.consultation.find({"patient_id": pid_query})) | |
| for c in consultations: | |
| content = f"Consultation - Date: {c.get('date')}, Diagnosis: {c.get('diagnosis')}, Doctor: {c.get('doctor_name','Unknown')}, Symptoms: {c.get('symptoms','N/A')}, Prescription: {c.get('prescription','N/A')}, Vitals BP: {c.get('vitals_bp','N/A')}, Vitals Pulse: {c.get('vitals_pulse','N/A')}" | |
| patient_data.append({'content': content}) | |
| reports = list(mongo.db.testreport.find({"patient_id": pid_query})) | |
| for r in reports: | |
| content = f"Test Report - Name: {r.get('test_name')}, Date: {r.get('date')}, Result: {r.get('result','N/A')}" | |
| patient_data.append({'content': content}) | |
| treatments = list(mongo.db.treatment.find({"patient_id": pid_query})) | |
| for t in treatments: | |
| content = f"Treatment - Name: {t.get('treatment_name')}, Start Date: {t.get('start_date')}, Status: {t.get('status','N/A')}, Medications: {t.get('medications_detail','N/A')}" | |
| patient_data.append({'content': content}) | |
| billing_records = list(mongo.db.billing.find({"patient_id": pid_query})) | |
| for b in billing_records: | |
| content = f"Billing - Amount: {b.get('amount')}, Date: {b.get('date')}, Status: {b.get('status','N/A')}" | |
| patient_data.append({'content': content}) | |
| insurance_records = list(mongo.db.insurance.find({"patient_id": pid_query})) | |
| for i in insurance_records: | |
| content = f"Insurance - Provider: {i.get('provider')}, Policy Number: {i.get('policy_number')}, Plan Type: {i.get('plan_type','N/A')}" | |
| patient_data.append({'content': content}) | |
| recordings = list(mongo.db.recordings.find({"patient_id": pid_query})) | |
| for rec in recordings: | |
| content = f"Voice Note - Timestamp: {rec.get('timestamp')}, Processed Note: {rec.get('processed_note','N/A')}" | |
| patient_data.append({'content': content}) | |
| return patient_data | |
| def generate_patient_record_markdown(patient_id): | |
| # Try finding by sql_id or id | |
| patient = mongo.db.patient.find_one({"$or": [{"sql_id": patient_id}, {"id": patient_id}]}) | |
| if not patient: | |
| return None, f"❌ No patient found with ID **{patient_id}**." | |
| # Consistent ID usage | |
| actual_p_id = patient.get('sql_id', patient.get('id')) | |
| # Helper to query by both int and str ID to match legacy data | |
| pid_query = {"$in": [actual_p_id, str(actual_p_id)]} | |
| # Fetch all records | |
| history_records = list(mongo.db.medicalhistory.find({"patient_id": pid_query})) | |
| consultations = list(mongo.db.consultation.find({"patient_id": pid_query})) | |
| reports = list(mongo.db.testreport.find({"patient_id": pid_query})) | |
| treatments = list(mongo.db.treatment.find({"patient_id": pid_query})) | |
| billing_records = list(mongo.db.billing.find({"patient_id": pid_query})) | |
| insurance_records = list(mongo.db.insurance.find({"patient_id": pid_query})) | |
| recordings = list(mongo.db.recordings.find({"patient_id": pid_query})) | |
| # Build Markdown | |
| md = f"### 👤 Patient Record: {patient.get('name')}\n" | |
| md += f"**ID:** {actual_p_id} | **Age:** {patient.get('age', 'N/A')} | **Gender:** {patient.get('gender', 'N/A')} | **Contact:** {patient.get('contact', 'N/A')}\n\n" | |
| md += "---\n" | |
| # 1. Medical History | |
| md += "#### 📜 Medical History\n" | |
| if history_records: | |
| for h in history_records: | |
| md += f"- **{h.get('condition')}** ({h.get('diagnosis_date')})\n" | |
| if h.get('notes'): md += f" > *Notes: {h.get('notes')}*\n" | |
| if h.get('medications_history'): md += f" - Meds: {h.get('medications_history')}\n" | |
| if h.get('allergies'): md += f" - Allergies: {h.get('allergies')}\n" | |
| else: | |
| md += "_No medical history recorded._\n" | |
| md += "\n" | |
| # 2. Consultations | |
| md += "#### 🩺 Consultations\n" | |
| if consultations: | |
| for c in consultations: | |
| # Format: Date - Diagnosis (Doctor) | |
| md += f"- **{c.get('date')}** — {c.get('diagnosis')} (Dr. {c.get('doctor_name') or 'Unknown'})\n" | |
| if c.get('symptoms'): md += f" - *Symptoms:* {c.get('symptoms')}\n" | |
| if c.get('prescription'): md += f" - *Rx:* {c.get('prescription')}\n" | |
| if c.get('vitals_bp') or c.get('vitals_pulse'): | |
| md += f" - *Vitals:* BP {c.get('vitals_bp', '-')} | Pulse {c.get('vitals_pulse', '-')}\n" | |
| else: | |
| md += "_No consultation history._\n" | |
| md += "\n" | |
| # 3. Test Reports | |
| md += "#### 🧪 Test Reports\n" | |
| if reports: | |
| for r in reports: | |
| md += f"- **{r.get('test_name')}** ({r.get('date')})\n" | |
| if r.get('result'): md += f" - Result: **{r.get('result')}**\n" | |
| if r.get('file_data'): md += f" - *[Attachment: {r.get('file_name')}]*\n" | |
| else: | |
| md += "_No test reports found._\n" | |
| md += "\n" | |
| # 4. Treatments | |
| md += "#### 💊 Active Treatments\n" | |
| if treatments: | |
| for t in treatments: | |
| status_icon = "🟢" if t.get('status') == 'Active' else "⚪" | |
| md += f"- {status_icon} **{t.get('treatment_name')}** (Started: {t.get('start_date')})\n" | |
| if t.get('medications_detail'): md += f" - Meds: {t.get('medications_detail')}\n" | |
| else: | |
| md += "_No active treatments._\n" | |
| md += "\n" | |
| # 5. Billing & Insurance | |
| md += "#### 💳 Insurance & Billing\n" | |
| if insurance_records: | |
| for i in insurance_records: | |
| md += f"- **{i.get('provider')}** (Policy: `{i.get('policy_number')}`)\n" | |
| if i.get('plan_type'): md += f" - Plan: {i.get('plan_type')}\n" | |
| else: | |
| md += "_No insurance details._\n" | |
| if billing_records: | |
| md += "\n**Recent Bills:**\n" | |
| for b in billing_records: | |
| params = f"Status: {b.get('status')}" | |
| md += f"- ${b.get('amount')} ({params}) - {b.get('date')}\n" | |
| else: | |
| md += "\n_No billing records._\n" | |
| # 6. Audio Recordings | |
| md += "\n#### 🎙️ Recent Voice Notes\n" | |
| if recordings: | |
| for rec in recordings: | |
| md += f"- **{rec.get('timestamp')}**\n" | |
| # SAFE CAST TO STR (Fix for slice error) | |
| note_str = str(rec.get('processed_note', '') or '') | |
| preview = note_str[:150] + "..." if len(note_str) > 150 else note_str | |
| md += f" > {preview}\n" | |
| else: | |
| md += "_No recordings found._\n" | |
| return patient, md | |