Open raw ↗"""HAD Digital MVP - Main HTTP Server.
A complete Hospitalisation a Domicile (HAD) digital platform for managing
chemotherapy toxicity monitoring in home care settings.
Usage:
python app.py [--host HOST] [--port PORT] [--debug]
"""
import http.server
import json
import os
import secrets
import sys
import uuid
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlparse, parse_qs
sys.path.insert(0, str(Path(__file__).resolve().parent))
from config_manager import config
from database import get_db
from user_store import authenticate, get_user, list_users, list_users_by_role
from ctcae_engine import get_ctcae_engine
from alert_engine import get_alert_engine
from audit_logger import log_action, get_audit_log
from seed_demo import seed_demo_data
from adapters.stub import StubAdapter
from adapters.glm import GLMAdapter
SESSIONS = {}
AI_ADAPTERS = {"stub": StubAdapter, "glm": GLMAdapter}
class HADRequestHandler(http.server.BaseHTTPRequestHandler):
server_version = "HAD-Digital/1.0"
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/") or "/"
params = parse_qs(parsed.query)
if path == "/" or path == "/index.html":
return self._serve_static("index.html")
if path.startswith("/static/"):
return self._serve_static(path[8:])
routes = {
"/api/whoami": lambda: self._api_whoami(),
"/api/patients": lambda: self._api_list_patients(params),
"/api/reports": lambda: self._api_list_reports(params),
"/api/grades": lambda: self._api_list_grades(params),
"/api/alerts": lambda: self._api_list_alerts(params),
"/api/timeline": lambda: self._api_get_timeline(params),
"/api/treatment-plan": lambda: self._api_get_treatment_plan(params),
"/api/messages": lambda: self._api_list_messages(params),
"/api/export/summary": lambda: self._api_export_summary(params),
"/api/audit-log": lambda: self._api_audit_log(params),
"/api/symptoms": lambda: self._api_list_symptoms(),
"/api/guidance": lambda: self._api_get_guidance(params),
}
if path in routes:
return routes[path]()
if path.startswith("/api/patients/"):
return self._api_get_patient(path.split("/")[3])
self._json_response({"error": "Not found"}, 404)
def do_POST(self):
parsed = urlparse(self.path)
path = parsed.path.rstrip("/")
routes = {
"/api/login": lambda: self._api_login(),
"/api/logout": lambda: self._api_logout(),
"/api/reports": lambda: self._api_submit_report(),
"/api/messages": lambda: self._api_send_message(),
"/api/chat": lambda: self._api_chat(),
}
if path in routes:
return routes[path]()
if path.startswith("/api/alerts/") and path.endswith("/acknowledge"):
return self._api_acknowledge_alert(path.split("/")[3])
self._json_response({"error": "Not found"}, 404)
# Security headers
def _add_security_headers(self):
security = config.get("security", default={})
self.send_header("Content-Security-Policy", security.get("csp", "default-src 'self'"))
self.send_header("X-Content-Type-Options", security.get("x_content_type_options", "nosniff"))
self.send_header("X-Frame-Options", security.get("x_frame_options", "DENY"))
self.send_header("Referrer-Policy", security.get("referrer_policy", "strict-origin-when-cross-origin"))
self.send_header("Cache-Control", "no-store, no-cache, must-revalidate")
# Session management
def _get_session(self):
cookie_header = self.headers.get("Cookie", "")
session_cookie = config.get("session", "cookie_name", default="HAD_SESSION")
for cookie in cookie_header.split(";"):
cookie = cookie.strip()
if cookie.startswith(f"{session_cookie}="):
sid = cookie.split("=", 1)[1]
return SESSIONS.get(sid)
return None
def _set_session(self, user):
sid = str(uuid.uuid4())
SESSIONS[sid] = {"user": user, "created_at": datetime.now(timezone.utc).isoformat()}
self._pending_cookies = getattr(self, "_pending_cookies", [])
session_cookie = config.get("session", "cookie_name", default="HAD_SESSION")
max_age = config.get("session", "max_age_seconds", default=86400)
self._pending_cookies.append(f"{session_cookie}={sid}; Path=/; HttpOnly; SameSite=Strict; Max-Age={max_age}")
def _clear_session(self):
cookie_header = self.headers.get("Cookie", "")
session_cookie = config.get("session", "cookie_name", default="HAD_SESSION")
for cookie in cookie_header.split(";"):
cookie = cookie.strip()
if cookie.startswith(f"{session_cookie}="):
sid = cookie.split("=", 1)[1]
SESSIONS.pop(sid, None)
self._pending_cookies = getattr(self, "_pending_cookies", [])
self._pending_cookies.append(f"{session_cookie}=; Path=/; HttpOnly; SameSite=Strict; Max-Age=0")
def _get_current_user(self):
session = self._get_session()
return session.get("user") if session else None
# JSON helpers
def _json_response(self, data, status=200):
body = json.dumps(data, ensure_ascii=False, default=str).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self._add_security_headers()
for cookie in getattr(self, "_pending_cookies", []):
self.send_header("Set-Cookie", cookie)
self._pending_cookies = []
self.end_headers()
self.wfile.write(body)
def _read_json_body(self):
cl = int(self.headers.get("Content-Length", 0))
if cl == 0:
return {}
try:
return json.loads(self.rfile.read(cl).decode("utf-8"))
except json.JSONDecodeError:
return {}
# Static files
def _serve_static(self, file_path):
static_dir = Path(__file__).resolve().parent / "static"
try:
resolved = (static_dir / file_path).resolve()
if not str(resolved).startswith(str(static_dir.resolve())):
self._json_response({"error": "Path traversal blocked"}, 403)
return
except (ValueError, OSError):
self._json_response({"error": "Invalid path"}, 400)
return
if not resolved.exists() or not resolved.is_file():
self._json_response({"error": "File not found"}, 404)
return
ct = {".html": "text/html; charset=utf-8", ".css": "text/css; charset=utf-8",
".js": "application/javascript; charset=utf-8", ".json": "application/json; charset=utf-8",
".png": "image/png", ".jpg": "image/jpeg", ".svg": "image/svg+xml", ".ico": "image/x-icon"}
try:
content = resolved.read_bytes()
self.send_response(200)
self.send_header("Content-Type", ct.get(resolved.suffix.lower(), "application/octet-stream"))
self.send_header("Content-Length", str(len(content)))
self._add_security_headers()
self.end_headers()
self.wfile.write(content)
except OSError:
self._json_response({"error": "File read error"}, 500)
# Auth API
def _api_login(self):
data = self._read_json_body()
username = data.get("username", "").strip()
password = data.get("password", "")
if not username or not password:
self._json_response({"error": "Username and password required"}, 400)
return
ip = self.client_address[0] if self.client_address else None
user = authenticate(username, password, ip)
if user is None:
log_action(action="login_failed", details=f"Failed login for '{username}'", ip_address=ip)
self._json_response({"error": "Invalid credentials or account locked"}, 401)
return
self._set_session(user)
log_action(user_id=user["id"], username=user["username"], action="login", resource_type="session", ip_address=ip)
self._json_response({"message": "Login successful", "user": {k: user[k] for k in ("id", "username", "role", "display_name")}})
def _api_logout(self):
user = self._get_current_user()
if user:
log_action(user_id=user["id"], username=user["username"], action="logout", resource_type="session")
self._clear_session()
self._json_response({"message": "Logged out"})
def _api_whoami(self):
user = self._get_current_user()
if user is None:
self._json_response({"authenticated": False}, 401)
return
self._json_response({"authenticated": True, "user": user})
# Patient API
def _api_list_patients(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
db = get_db()
role = user["role"]
if role in ("patient", "caregiver"):
p = db.fetchone("SELECT * FROM patients WHERE id = ?", (user.get("patient_id"),))
patients = [dict(p)] if p else []
elif role in ("oncologist", "had_nurse"):
patients = [dict(r) for r in db.fetchall("SELECT * FROM patients WHERE primary_oncologist_id = ? OR status = 'active'", (user["id"],))]
else:
patients = [dict(r) for r in db.fetchall("SELECT * FROM patients WHERE status = 'active'")]
log_action(user_id=user["id"], action="list_patients", resource_type="patient")
self._json_response({"patients": patients})
def _api_get_patient(self, pid):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
db = get_db()
patient = db.fetchone("SELECT * FROM patients WHERE id = ?", (pid,))
if patient is None:
self._json_response({"error": "Patient not found"}, 404)
return
d = dict(patient)
plan = db.fetchone("SELECT * FROM treatment_plans WHERE patient_id = ? AND status = 'active' ORDER BY created_at DESC LIMIT 1", (pid,))
if plan:
d["active_treatment_plan"] = dict(plan)
d["recent_grades"] = [dict(g) for g in db.fetchall("SELECT * FROM toxicity_grades WHERE patient_id = ? ORDER BY created_at DESC LIMIT 10", (pid,))]
d["pending_alerts"] = [dict(a) for a in db.fetchall("SELECT * FROM alerts WHERE patient_id = ? AND status = 'pending' ORDER BY created_at DESC", (pid,))]
log_action(user_id=user["id"], action="view_patient", resource_type="patient", resource_id=int(pid))
self._json_response(d)
# Report API
def _api_submit_report(self):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
data = self._read_json_body()
for f in ("patient_id", "symptom_id", "symptom_category"):
if f not in data:
self._json_response({"error": f"Missing required field: {f}"}, 400)
return
db = get_db()
pid = data["patient_id"]
plan = db.fetchone("SELECT * FROM treatment_plans WHERE patient_id = ? AND status = 'active' LIMIT 1", (pid,))
cur = db.execute(
"INSERT INTO toxicity_reports (patient_id, episode_id, treatment_plan_id, reporter_id, symptom_id, symptom_category, severity_score, lab_values, notes, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending')",
(pid, plan["episode_id"] if plan else None, plan["id"] if plan else None, user["id"], data["symptom_id"], data["symptom_category"], data.get("severity_score"), json.dumps(data.get("lab_values", {})) if data.get("lab_values") else None, data.get("notes", ""))
)
db.commit()
rid = cur.lastrowid
engine = get_ctcae_engine()
gi = data.get("grading_inputs", {})
if data.get("severity_score") is not None:
gi["severity_score"] = data["severity_score"]
if data.get("lab_values"):
gi.update(data["lab_values"])
gr = engine.grade_and_save(rid, pid, data["symptom_id"], gi)
alert = None
if gr.get("grade") and gr["grade"] >= 2:
alert = get_alert_engine().create_alert_for_grade(
grade_id=gr.get("grade_id"), patient_id=pid, report_id=rid, grade=gr["grade"],
symptom_id=data["symptom_id"], display_name=gr.get("display_name", data["symptom_id"]), reporter_id=user["id"])
db.execute("INSERT INTO timeline_events (patient_id, episode_id, event_type, title, description, created_by) VALUES (?, ?, 'report', ?, ?, ?)",
(pid, plan["episode_id"] if plan else None, f"Toxicity report: {data['symptom_id']}", f"Grade {gr.get('grade', 'N/A')} - {gr.get('criteria', '')}", user["id"]))
db.commit()
log_action(user_id=user["id"], action="submit_report", resource_type="toxicity_report", resource_id=rid, details=f"Report for {data['symptom_id']} on patient {pid}, grade {gr.get('grade')}")
self._json_response({"message": "Report submitted and graded", "report_id": rid, "grading": gr, "alert": alert}, 201)
def _api_list_reports(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
pid = params.get("patient_id", [None])[0]
if not pid:
self._json_response({"error": "patient_id parameter required"}, 400)
return
reports = get_db().fetchall("SELECT * FROM toxicity_reports WHERE patient_id = ? ORDER BY report_date DESC", (pid,))
log_action(user_id=user["id"], action="list_reports", resource_type="toxicity_report", details=f"Patient {pid}")
self._json_response({"reports": [dict(r) for r in reports]})
def _api_list_grades(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
pid = params.get("patient_id", [None])[0]
if not pid:
self._json_response({"error": "patient_id parameter required"}, 400)
return
grades = get_db().fetchall("SELECT * FROM toxicity_grades WHERE patient_id = ? ORDER BY created_at DESC", (pid,))
self._json_response({"grades": [dict(g) for g in grades]})
# Alert API
def _api_list_alerts(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
ae = get_alert_engine()
pid = params.get("patient_id", [None])[0]
status = params.get("status", [None])[0]
if user["role"] in ("patient", "caregiver"):
pid = user.get("patient_id")
alerts = ae.get_alerts(patient_id=int(pid) if pid else None, status=status,
assigned_to=user["id"] if user["role"] in ("oncologist", "had_nurse") else None)
self._json_response({"alerts": alerts})
def _api_acknowledge_alert(self, aid):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
if get_alert_engine().acknowledge_alert(int(aid), user["id"]):
self._json_response({"message": "Alert acknowledged"})
else:
self._json_response({"error": "Alert not found"}, 404)
# Timeline API
def _api_get_timeline(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
pid = params.get("patient_id", [None])[0]
eid = params.get("episode_id", [None])[0]
limit = int(params.get("limit", [50])[0])
conds, qparams = [], []
if pid:
conds.append("patient_id = ?"); qparams.append(int(pid))
if eid:
conds.append("episode_id = ?"); qparams.append(int(eid))
wc = " AND ".join(conds) if conds else "1=1"
events = get_db().fetchall(f"SELECT * FROM timeline_events WHERE {wc} ORDER BY event_date DESC LIMIT ?", tuple(qparams) + (limit,))
self._json_response({"events": [dict(e) for e in events]})
# Treatment Plan API
def _api_get_treatment_plan(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
pid = params.get("patient_id", [None])[0]
if not pid:
self._json_response({"error": "patient_id parameter required"}, 400)
return
plans = get_db().fetchall("SELECT * FROM treatment_plans WHERE patient_id = ? ORDER BY created_at DESC", (pid,))
self._json_response({"treatment_plans": [dict(p) for p in plans]})
# Message API
def _api_send_message(self):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
data = self._read_json_body()
body = data.get("body", "").strip()
if not body:
self._json_response({"error": "Message body required"}, 400)
return
db = get_db()
cur = db.execute("INSERT INTO messages (sender_id, recipient_id, recipient_role, patient_id, subject, body, message_type) VALUES (?, ?, ?, ?, ?, ?, ?)",
(user["id"], data.get("recipient_id"), data.get("recipient_role"), data.get("patient_id"), data.get("subject", ""), body, data.get("message_type", "direct")))
db.commit()
log_action(user_id=user["id"], action="send_message", resource_type="message", resource_id=cur.lastrowid)
self._json_response({"message": "Message sent", "message_id": cur.lastrowid}, 201)
def _api_list_messages(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
db = get_db()
pid = params.get("patient_id", [None])[0]
limit = int(params.get("limit", [50])[0])
if pid:
msgs = db.fetchall("SELECT * FROM messages WHERE patient_id = ? AND (recipient_id = ? OR sender_id = ? OR recipient_role = ?) ORDER BY created_at DESC LIMIT ?",
(int(pid), user["id"], user["id"], user["role"], limit))
else:
msgs = db.fetchall("SELECT * FROM messages WHERE recipient_id = ? OR sender_id = ? OR recipient_role = ? ORDER BY created_at DESC LIMIT ?",
(user["id"], user["id"], user["role"], limit))
self._json_response({"messages": [dict(m) for m in msgs]})
# Export API
def _api_export_summary(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
pid = params.get("patient_id", [None])[0]
if not pid:
self._json_response({"error": "patient_id parameter required"}, 400)
return
db = get_db()
patient = db.fetchone("SELECT * FROM patients WHERE id = ?", (pid,))
if patient is None:
self._json_response({"error": "Patient not found"}, 404)
return
grades = db.fetchall("SELECT * FROM toxicity_grades WHERE patient_id = ? ORDER BY created_at DESC", (pid,))
plans = db.fetchall("SELECT * FROM treatment_plans WHERE patient_id = ? ORDER BY created_at DESC", (pid,))
summary = {
"patient": {"mrn": patient["mrn"], "name": f"{patient['first_name']} {patient['last_name']}",
"date_of_birth": patient["date_of_birth"], "cancer_type": patient["cancer_type"], "cancer_stage": patient["cancer_stage"]},
"treatment_plans": [{"protocol": p["protocol_name"], "regimen": p["regimen"], "current_cycle": p["current_cycle"], "total_cycles": p["cycle_count"], "status": p["status"]} for p in plans],
"toxicity_summary": {},
"generated_at": datetime.now(timezone.utc).isoformat(),
"generated_by": user["username"],
}
for g in grades:
s = g["symptom_id"]
if s not in summary["toxicity_summary"]:
summary["toxicity_summary"][s] = {"latest_grade": g["grade"], "latest_criteria": g["criteria"], "provisional": bool(g["provisional"]), "history": []}
summary["toxicity_summary"][s]["history"].append({"grade": g["grade"], "criteria": g["criteria"], "date": g["created_at"]})
log_action(user_id=user["id"], action="export_summary", resource_type="patient", resource_id=int(pid))
self._json_response(summary)
# Audit API
def _api_audit_log(self, params):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
if user["role"] != "admin":
self._json_response({"error": "Admin access required"}, 403)
return
limit = int(params.get("limit", [100])[0])
offset = int(params.get("offset", [0])[0])
action_f = params.get("action", [None])[0]
self._json_response({"audit_log": get_audit_log(action=action_f, limit=limit, offset=offset)})
# Symptom & Guidance API
def _api_list_symptoms(self):
self._json_response({"symptoms": get_ctcae_engine().list_symptoms()})
def _api_get_guidance(self, params):
sid = params.get("symptom_id", [None])[0]
grade = params.get("grade", [None])[0]
if not sid:
self._json_response({"error": "symptom_id parameter required"}, 400)
return
gp = config.get("guidance", "path", default="guidance/guidance.json")
try:
with open(gp, "r", encoding="utf-8") as f:
gd = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
self._json_response({"error": "Guidance data not available"}, 500)
return
sg = gd.get("symptoms", {}).get(sid)
if sg is None:
self._json_response({"error": f"No guidance found for symptom: {sid}"}, 404)
return
if grade:
gg = sg.get("grades", {}).get(str(grade))
if gg:
self._json_response({"symptom_id": sid, "display_name": sg.get("display_name"), "grade": int(grade), "guidance": gg})
else:
self._json_response({"error": f"No guidance for grade {grade}"}, 404)
else:
self._json_response({"symptom_id": sid, "display_name": sg.get("display_name"), "grades": sg.get("grades", {})})
# Chat API
def _api_chat(self):
user = self._get_current_user()
if user is None:
self._json_response({"error": "Authentication required"}, 401)
return
data = self._read_json_body()
msg = data.get("message", "").strip()
if not msg:
self._json_response({"error": "Message required"}, 400)
return
aname = config.get("ai", "adapter", default="stub")
adapter = AI_ADAPTERS.get(aname, StubAdapter)()
ctx = {"patient_id": data.get("patient_id"), "user_role": user["role"], "history": data.get("history", []), "symptoms": data.get("symptoms", [])}
result = adapter.chat(msg, ctx)
log_action(user_id=user["id"], action="chat", resource_type="ai_chat", details=f"Adapter: {aname}, len: {len(msg)}")
self._json_response(result)
def log_message(self, format, *args):
if config.get("debug", default=False):
super().log_message(format, *args)
def run_server(host=None, port=None, debug=None):
if host is None:
host = config.get("host", default="127.0.0.1")
if port is None:
port = config.get("port", default=8080)
if debug is None:
debug = config.get("debug", default=False)
db = get_db()
print(f"[app] Database initialized at {config.get('database', 'path')}")
if config.get("demo", "auto_seed", default=True):
seed_demo_data()
server = http.server.HTTPServer((host, port), HADRequestHandler)
print(f"""
+------------------------------------------------------------------+
| HAD Digital MVP |
| Hospitalisation a Domicile Platform |
+------------------------------------------------------------------+
| Server: http://{host}:{port}
| Debug: {debug}
| DB: SQLite (WAL mode)
+------------------------------------------------------------------+
| Demo Users:
| admin / admin123 (Admin)
| dr.martin / demo123 (Oncologist)
| inf.moret / demo123 (HAD Nurse)
| patient.durand / demo123 (Patient)
+------------------------------------------------------------------+
""")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n[app] Shutting down...")
server.shutdown()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="HAD Digital MVP Server")
parser.add_argument("--host", default=None, help="Host to bind to")
parser.add_argument("--port", type=int, default=None, help="Port to bind to")
parser.add_argument("--debug", action="store_true", help="Enable debug mode")
args = parser.parse_args()
if args.debug:
os.environ["HAD_DEBUG"] = "true"
run_server(host=args.host, port=args.port, debug=args.debug)