"""
Cancer Patient Management System - API Server
HTTP server with REST-like API endpoints using Python standard library
"""
import json
import os
import sys
import webbrowser
import threading
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
from pathlib import Path
from urllib.parse import urlparse, parse_qs
from datetime import datetime
sys.path.insert(0, str(Path(__file__).parent))
from database import get_connection, verify_password, init_db
from seed_data import run_seed
# Check data/ subfolder first (for packaged distribution), then script directory
_script_dir = Path(__file__).parent
DB_PATH = _script_dir / "data" / "cancer_patients.db"
if not DB_PATH.exists():
DB_PATH = _script_dir / "cancer_patients.db"
STATIC_DIR = Path(__file__).parent / "static"
# Session store
sessions = {}
def get_user_by_session(session_id):
if session_id in sessions:
return sessions[session_id]
return None
class CancerAPIHandler(BaseHTTPRequestHandler):
def log_message(self, format, *args):
pass # Suppress default logging
def send_json(self, data, status=200):
self.send_response(status)
self.send_header("Content-Type", "application/json")
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(json.dumps(data, default=str).encode())
def send_html(self, html):
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.end_headers()
self.wfile.write(html.encode())
def do_OPTIONS(self):
self.send_response(200)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type, Authorization")
self.end_headers()
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path
params = parse_qs(parsed.query)
# Serve static files
if path == "/" or path == "/index.html":
html_file = STATIC_DIR / "index.html"
if html_file.exists():
self.send_html(html_file.read_text(encoding="utf-8"))
else:
self.send_json({"error": "index.html not found"}, 404)
return
# API routes
session_id = self.headers.get("Authorization", "").replace("Bearer ", "")
user = get_user_by_session(session_id)
if path == "/api/patients":
if not user:
self.send_json({"error": "Unauthorized"}, 401)
return
conn = get_connection()
if user["role"] == "doctor":
rows = conn.execute("SELECT * FROM patients WHERE assigned_doctor_id=?", (user["id"],)).fetchall()
elif user["role"] == "nurse":
rows = conn.execute("""
SELECT p.* FROM patients p
LEFT JOIN nurse_patient_assignments n ON p.id = n.patient_id
WHERE p.assigned_nurse_id=? OR n.nurse_id=?
""", (user["id"], user["id"])).fetchall()
elif user["role"] == "patient":
rows = conn.execute("SELECT * FROM patients WHERE patient_code LIKE ?", (f"%{user['username'].split('.')[1]}%",)).fetchall()
else:
rows = conn.execute("SELECT * FROM patients").fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/") and "/medications" in path:
pid = path.split("/")[3]
conn = get_connection()
rows = conn.execute("SELECT * FROM medications WHERE patient_id=?", (pid,)).fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/") and "/vitals" in path:
pid = path.split("/")[3]
conn = get_connection()
rows = conn.execute("SELECT * FROM vital_signs WHERE patient_id=? ORDER BY recorded_at DESC", (pid,)).fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/") and "/labs" in path:
pid = path.split("/")[3]
conn = get_connection()
rows = conn.execute("SELECT * FROM lab_results WHERE patient_id=? ORDER BY test_date DESC", (pid,)).fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/") and "/symptoms" in path:
pid = path.split("/")[3]
conn = get_connection()
rows = conn.execute("SELECT * FROM symptom_reports WHERE patient_id=? ORDER BY reported_at DESC", (pid,)).fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/") and "/cycles" in path:
pid = path.split("/")[3]
conn = get_connection()
rows = conn.execute("SELECT * FROM treatment_cycles WHERE patient_id=? ORDER BY cycle_number", (pid,)).fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/") and "/alerts" in path:
pid = path.split("/")[3]
conn = get_connection()
rows = conn.execute("SELECT * FROM alerts WHERE patient_id=? ORDER BY created_at DESC", (pid,)).fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path.startswith("/api/patients/"):
pid = path.split("/")[3]
conn = get_connection()
row = conn.execute("SELECT * FROM patients WHERE id=?", (pid,)).fetchone()
conn.close()
if row:
self.send_json(dict(row))
else:
self.send_json({"error": "Patient not found"}, 404)
elif path == "/api/alerts":
conn = get_connection()
severity = params.get("severity", [None])[0]
if severity:
rows = conn.execute("SELECT a.*, p.first_name, p.last_name FROM alerts a JOIN patients p ON a.patient_id=p.id WHERE a.severity=? ORDER BY a.created_at DESC", (severity,)).fetchall()
else:
rows = conn.execute("SELECT a.*, p.first_name, p.last_name FROM alerts a JOIN patients p ON a.patient_id=p.id ORDER BY a.created_at DESC").fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path == "/api/users":
if not user or user["role"] not in ("admin", "doctor"):
self.send_json({"error": "Forbidden"}, 403)
return
conn = get_connection()
rows = conn.execute("SELECT id, username, role, full_name, email, department, specialization, is_active FROM users").fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path == "/api/nurse-assignments":
conn = get_connection()
rows = conn.execute("""
SELECT n.*, u.full_name as nurse_name, p.first_name||' '||p.last_name as patient_name, p.patient_code
FROM nurse_patient_assignments n
JOIN users u ON n.nurse_id=u.id
JOIN patients p ON n.patient_id=p.id
ORDER BY n.assigned_date DESC
""").fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path == "/api/audit-log":
if not user or user["role"] != "admin":
self.send_json({"error": "Forbidden"}, 403)
return
conn = get_connection()
rows = conn.execute("""
SELECT a.*, u.full_name as user_name
FROM audit_log a JOIN users u ON a.user_id=u.id
ORDER BY a.created_at DESC LIMIT 100
""").fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path == "/api/test-results":
conn = get_connection()
rows = conn.execute("SELECT * FROM test_results ORDER BY executed_at DESC").fetchall()
conn.close()
self.send_json([dict(r) for r in rows])
elif path == "/api/dashboard/stats":
conn = get_connection()
stats = {
"total_patients": conn.execute("SELECT COUNT(*) FROM patients").fetchone()[0],
"active_patients": conn.execute("SELECT COUNT(*) FROM patients WHERE status='in_treatment'").fetchone()[0],
"total_medications": conn.execute("SELECT COUNT(*) FROM medications WHERE status='active'").fetchone()[0],
"critical_alerts": conn.execute("SELECT COUNT(*) FROM alerts WHERE severity='critical' AND resolved=0").fetchone()[0],
"high_alerts": conn.execute("SELECT COUNT(*) FROM alerts WHERE severity='high' AND resolved=0").fetchone()[0],
"total_users": conn.execute("SELECT COUNT(*) FROM users WHERE is_active=1").fetchone()[0],
"cancer_types": [dict(r) for r in conn.execute("SELECT cancer_type, COUNT(*) as count FROM patients GROUP BY cancer_type").fetchall()],
"alerts_by_severity": [dict(r) for r in conn.execute("SELECT severity, COUNT(*) as count FROM alerts GROUP BY severity").fetchall()],
}
conn.close()
self.send_json(stats)
elif path == "/api/me":
if not user:
self.send_json({"error": "Unauthorized"}, 401)
return
self.send_json({k: v for k, v in user.items() if k not in ("password_hash", "salt")})
else:
self.send_json({"error": "Not found"}, 404)
def do_POST(self):
parsed = urlparse(self.path)
path = parsed.path
content_length = int(self.headers.get("Content-Length", 0))
body = json.loads(self.rfile.read(content_length)) if content_length > 0 else {}
if path == "/api/login":
username = body.get("username", "")
password = body.get("password", "")
conn = get_connection()
user = conn.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
conn.close()
if user and verify_password(password, user["password_hash"], user["salt"]):
session_id = os.urandom(32).hex()
sessions[session_id] = dict(user)
self.send_json({"token": session_id, "user": {k: v for k, v in dict(user).items() if k not in ("password_hash", "salt")}})
else:
self.send_json({"error": "Invalid credentials"}, 401)
return
session_id = self.headers.get("Authorization", "").replace("Bearer ", "")
user = get_user_by_session(session_id)
if path == "/api/logout":
if session_id in sessions:
del sessions[session_id]
self.send_json({"ok": True})
elif path == "/api/vitals":
if not user:
self.send_json({"error": "Unauthorized"}, 401)
return
conn = get_connection()
conn.execute("""INSERT INTO vital_signs (patient_id,recorded_by,temperature,heart_rate,blood_pressure_systolic,blood_pressure_diastolic,respiratory_rate,oxygen_saturation,weight_kg,pain_score,notes)
VALUES (?,?,?,?,?,?,?,?,?,?,?)""",
(body["patient_id"], user["id"], body.get("temperature"), body.get("heart_rate"),
body.get("bp_systolic"), body.get("bp_diastolic"), body.get("respiratory_rate"),
body.get("o2_sat"), body.get("weight"), body.get("pain_score"), body.get("notes")))
conn.commit()
conn.close()
self.send_json({"ok": True})
elif path == "/api/symptoms":
if not user:
self.send_json({"error": "Unauthorized"}, 401)
return
conn = get_connection()
conn.execute("""INSERT INTO symptom_reports (patient_id,symptom_type,severity,description,duration,triggers,interventions)
VALUES (?,?,?,?,?,?,?)""",
(body["patient_id"], body["symptom_type"], body["severity"],
body.get("description"), body.get("duration"), body.get("triggers"), body.get("interventions")))
conn.commit()
conn.close()
self.send_json({"ok": True})
elif path == "/api/alerts/acknowledge":
if not user:
self.send_json({"error": "Unauthorized"}, 401)
return
conn = get_connection()
conn.execute("UPDATE alerts SET acknowledged_by=?, acknowledged_at=CURRENT_TIMESTAMP WHERE id=?",
(user["id"], body["alert_id"]))
conn.commit()
conn.close()
self.send_json({"ok": True})
else:
self.send_json({"error": "Not found"}, 404)
def main():
# Initialize database and seed if needed
if not DB_PATH.exists():
print("First run - seeding demo data...")
run_seed()
else:
print(f"Database exists: {DB_PATH}")
# Ensure static directory
STATIC_DIR.mkdir(exist_ok=True)
port = int(os.environ.get("PORT", "8080"))
class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
server = ThreadedHTTPServer(("127.0.0.1", port), CancerAPIHandler)
print(f"\n{'='*60}")
print(f" CANCER PATIENT MANAGEMENT SYSTEM")
print(f" Server running at http://127.0.0.1:{port}")
print(f"{'='*60}")
print(f"\n Login credentials:")
print(f" Admin: admin / admin123")
print(f" Doctor: dr.martin / doctor123")
print(f" Nurse: inf.moret / nurse123")
print(f" Patient: patient.durand / patient123")
print(f"\n Press Ctrl+C to stop\n")
# Auto-open browser
threading.Timer(1.5, lambda: webbrowser.open(f"http://127.0.0.1:{port}")).start()
try:
server.serve_forever()
except KeyboardInterrupt:
print("\nServer stopped.")
server.server_close()
if __name__ == "__main__":
main()