File size: 8,113 Bytes
09daffd d452590 a2a0a3e 09daffd a2a0a3e ff4408c 09daffd e714a62 09daffd a2a0a3e e714a62 a2a0a3e cefcb8b a2a0a3e 09daffd |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 |
import json
import os
import tempfile
from datetime import datetime
import traceback
import logging
from huggingface_hub import InferenceClient # Import InferenceClient
from urllib.parse import urlparse, parse_qs # Import URL parsing utilities
from utils.meldrx import MeldRxAPI # Import the MeldRxAPI class
import logging
from old.extractcode import extract_code_from_url
# ... (CallbackManager, display_form, generate_pdf_from_form, generate_pdf_from_meldrx, generate_discharge_paper_one_click, client initialization remain the same) ...
# Set up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CallbackManager:
def __init__(self, redirect_uri: str, client_secret: str = None):
client_id = os.getenv("APPID")
if not client_id:
raise ValueError("APPID environment variable not set.")
workspace_id = os.getenv("WORKSPACE_URL")
if not workspace_id:
raise ValueError("WORKSPACE_URL environment variable not set.")
self.api = MeldRxAPI(client_id, client_secret, workspace_id, redirect_uri)
self.auth_code = None
self.access_token = None
# Load token from file if it exists
token_path = '/tmp/access_token.txt'
if os.path.exists(token_path):
with open(token_path, 'r') as f:
self.access_token = f.read().strip()
self.api.access_token = self.access_token
def handle_callback(self, callback_url: str) -> str:
self.auth_code = extract_code_from_url(callback_url)
if not self.auth_code:
return "No authentication code found in URL."
if self.api.authenticate_with_code(self.auth_code):
self.access_token = self.api.access_token
with open('/tmp/access_token.txt', 'w') as f:
f.write(self.access_token)
return f"Authentication successful! Access Token: {self.access_token[:10]}... (truncated)"
return "Authentication failed. Please check the authorization code."
def get_auth_url(self) -> str:
return self.api.get_authorization_url()
def set_auth_code(self, code: str) -> str:
self.auth_code = code
if self.api.authenticate_with_code(code):
self.access_token = self.api.access_token
return (
f"<span style='color:#00FF7F;'>Authentication successful!</span> Access Token: {self.access_token[:10]}... (truncated)" # Neon Green Success
)
return "<span style='color:#FF4500;'>Authentication failed. Please check the code.</span>" # Neon Orange Error
def get_patient_data(self) -> str:
"""Fetch patient data from MeldRx"""
try:
if not self.access_token:
logger.warning("Not authenticated when getting patient data")
return "<span style='color:#FF8C00;'>Not authenticated. Please provide a valid authorization code first.</span>" # Neon Dark Orange
# For demo purposes, if there's no actual API connected, return mock data
# Remove this in production and use the real API call
if not hasattr(self.api, "get_patients") or self.api.get_patients is None:
logger.info("Using mock patient data (no API connection)")
# Return mock FHIR bundle with patient data
mock_data = {
"resourceType": "Bundle",
"type": "searchset",
"total": 2,
"link": [],
"entry": [
{
"resource": {
"resourceType": "Patient",
"id": "patient1",
"name": [
{
"use": "official",
"family": "Smith",
"given": ["John"],
}
],
"gender": "male",
"birthDate": "1970-01-01",
"address": [
{"city": "Boston", "state": "MA", "postalCode": "02108"}
],
}
},
{
"resource": {
"resourceType": "Patient",
"id": "patient2",
"name": [
{
"use": "official",
"family": "Johnson",
"given": ["Jane"],
}
],
"gender": "female",
"birthDate": "1985-05-15",
"address": [
{
"city": "Cambridge",
"state": "MA",
"postalCode": "02139",
}
],
}
},
],
}
return json.dumps(mock_data, indent=2)
# Real implementation with API call
logger.info("Calling Meldrx API to get patients")
patients = self.api.get_patients()
if patients is not None:
return (
json.dumps(patients, indent=2)
if patients
else "<span style='color:#FFFF00;'>No patient data returned.</span>" # Neon Yellow
)
return "<span style='color:#DC143C;'>Failed to retrieve patient data.</span>" # Crimson Error
except Exception as e:
error_msg = f"Error in get_patient_data: {str(e)}"
logger.error(error_msg)
return f"<span style='color:#FF6347;'>Error retrieving patient data: {str(e)}</span> {str(e)}" # Tomato Error
def get_patient_documents(self, patient_id: str = None):
"""Fetch patient documents from MeldRx"""
if not self.access_token:
return "<span style='color:#FF8C00;'>Not authenticated. Please provide a valid authorization code first.</span>" # Neon Dark Orange
try:
# This would call the actual MeldRx API to get documents for a specific patient
# For demonstration, we'll return mock document data
return [
{
"doc_id": "doc123",
"type": "clinical_note",
"date": "2023-01-16",
"author": "Dr. Sample Doctor",
"content": "Patient presented with symptoms of respiratory distress...",
},
{
"doc_id": "doc124",
"type": "lab_result",
"date": "2023-01-17",
"author": "Lab System",
"content": "CBC results: WBC 7.5, RBC 4.2, Hgb 14.1...",
},
]
except Exception as e:
return f"<span style='color:#FF6347;'>Error retrieving patient documents: {str(e)}</span>: {str(e)}" # Tomato Error
def extract_auth_code_from_url(redirected_url):
"""Extracts the authorization code from the redirected URL."""
try:
parsed_url = urlparse(redirected_url)
query_params = parse_qs(parsed_url.query)
if "code" in query_params:
return query_params["code"][0], None # Return code and no error
else:
return None, "Authorization code not found in URL." # Return None and error message
except Exception as e:
return None, f"Error parsing URL: {e}" # Return None and error message
|