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"Authentication successful! Access Token: {self.access_token[:10]}... (truncated)" # Neon Green Success
)
return "Authentication failed. Please check the code." # 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 "Not authenticated. Please provide a valid authorization code first." # 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 "No patient data returned." # Neon Yellow
)
return "Failed to retrieve patient data." # Crimson Error
except Exception as e:
error_msg = f"Error in get_patient_data: {str(e)}"
logger.error(error_msg)
return f"Error retrieving patient data: {str(e)} {str(e)}" # Tomato Error
def get_patient_documents(self, patient_id: str = None):
"""Fetch patient documents from MeldRx"""
if not self.access_token:
return "Not authenticated. Please provide a valid authorization code first." # 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"Error retrieving patient documents: {str(e)}: {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