"""
Cancer Patient Management System - Database Module
SQLite database with WAL mode for concurrent access
"""
import sqlite3
import hashlib
import os
import json
from datetime import datetime, timedelta
from pathlib import Path
DB_PATH = Path(__file__).parent / "cancer_patients.db"
def get_connection():
conn = sqlite3.connect(str(DB_PATH))
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
conn.row_factory = sqlite3.Row
return conn
def init_db():
conn = get_connection()
c = conn.cursor()
c.execute("""CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
role TEXT NOT NULL CHECK(role IN ('admin','doctor','nurse','patient')),
full_name TEXT NOT NULL,
email TEXT,
phone TEXT,
department TEXT,
specialization TEXT,
is_active INTEGER DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_login TIMESTAMP
)""")
c.execute("""CREATE TABLE IF NOT EXISTS patients (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_code TEXT UNIQUE NOT NULL,
first_name TEXT NOT NULL,
last_name TEXT NOT NULL,
date_of_birth DATE NOT NULL,
gender TEXT CHECK(gender IN ('M','F')),
cancer_type TEXT NOT NULL,
cancer_stage TEXT NOT NULL,
diagnosis_date DATE NOT NULL,
diagnosis_notes TEXT,
allergies TEXT,
comorbidities TEXT,
emergency_contact_name TEXT,
emergency_contact_phone TEXT,
address TEXT,
insurance_id TEXT,
assigned_doctor_id INTEGER REFERENCES users(id),
assigned_nurse_id INTEGER REFERENCES users(id),
status TEXT DEFAULT 'active' CHECK(status IN ('active','in_treatment','remission','discharged','deceased')),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)""")
c.execute("""CREATE TABLE IF NOT EXISTS medications (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER REFERENCES patients(id),
drug_name TEXT NOT NULL,
drug_class TEXT NOT NULL,
dosage TEXT NOT NULL,
unit TEXT NOT NULL,
route TEXT NOT NULL,
frequency TEXT NOT NULL,
start_date DATE NOT NULL,
end_date DATE,
prescribing_doctor_id INTEGER REFERENCES users(id),
indication TEXT,
side_effects TEXT,
contraindications TEXT,
monitoring_required TEXT,
status TEXT DEFAULT 'active' CHECK(status IN ('active','completed','discontinued','on_hold')),
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)""")
c.execute("""CREATE TABLE IF NOT EXISTS treatment_cycles (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER REFERENCES patients(id),
protocol_name TEXT NOT NULL,
cycle_number INTEGER NOT NULL,
total_cycles INTEGER,
start_date DATE NOT NULL,
end_date DATE,
medications_used TEXT,
response_assessment TEXT,
toxicity_grade INTEGER CHECK(toxicity_grade BETWEEN 0 AND 5),
side_effects_reported TEXT,
dose_modifications TEXT,
status TEXT DEFAULT 'planned' CHECK(status IN ('planned','in_progress','completed','delayed','cancelled')),
administered_by INTEGER REFERENCES users(id),
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)""")
c.execute("""CREATE TABLE IF NOT EXISTS vital_signs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER REFERENCES patients(id),
recorded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
recorded_by INTEGER REFERENCES users(id),
temperature REAL,
heart_rate INTEGER,
blood_pressure_systolic INTEGER,
blood_pressure_diastolic INTEGER,
respiratory_rate INTEGER,
oxygen_saturation REAL,
weight_kg REAL,
height_cm REAL,
bmi REAL,
pain_score INTEGER CHECK(pain_score BETWEEN 0 AND 10),
notes TEXT
)""")
c.execute("""CREATE TABLE IF NOT EXISTS lab_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER REFERENCES patients(id),
test_name TEXT NOT NULL,
test_date DATE NOT NULL,
result_value REAL,
unit TEXT,
reference_range TEXT,
is_abnormal INTEGER DEFAULT 0,
lab_name TEXT,
ordered_by INTEGER REFERENCES users(id),
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)""")
c.execute("""CREATE TABLE IF NOT EXISTS symptom_reports (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER REFERENCES patients(id),
reported_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
symptom_type TEXT NOT NULL,
severity INTEGER CHECK(severity BETWEEN 1 AND 10),
description TEXT,
duration TEXT,
triggers TEXT,
interventions TEXT,
resolved INTEGER DEFAULT 0,
resolved_at TIMESTAMP,
reviewed_by INTEGER REFERENCES users(id),
review_notes TEXT
)""")
c.execute("""CREATE TABLE IF NOT EXISTS alerts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
patient_id INTEGER REFERENCES patients(id),
alert_type TEXT NOT NULL,
severity TEXT NOT NULL CHECK(severity IN ('low','medium','high','critical')),
message TEXT NOT NULL,
trigger_condition TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
acknowledged_by INTEGER REFERENCES users(id),
acknowledged_at TIMESTAMP,
resolved INTEGER DEFAULT 0,
resolved_at TIMESTAMP,
action_taken TEXT
)""")
c.execute("""CREATE TABLE IF NOT EXISTS audit_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER REFERENCES users(id),
action TEXT NOT NULL,
table_name TEXT,
record_id INTEGER,
old_value TEXT,
new_value TEXT,
ip_address TEXT,
user_agent TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)""")
c.execute("""CREATE TABLE IF NOT EXISTS nurse_patient_assignments (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nurse_id INTEGER REFERENCES users(id),
patient_id INTEGER REFERENCES patients(id),
assigned_date DATE NOT NULL,
unassigned_date DATE,
shift TEXT CHECK(shift IN ('morning','afternoon','night','full_day')),
is_primary INTEGER DEFAULT 0,
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(nurse_id, patient_id, assigned_date)
)""")
c.execute("""CREATE TABLE IF NOT EXISTS test_results (
id INTEGER PRIMARY KEY AUTOINCREMENT,
test_name TEXT NOT NULL,
test_suite TEXT,
status TEXT NOT NULL CHECK(status IN ('passed','failed','skipped','error')),
execution_time REAL,
error_message TEXT,
executed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
executed_by TEXT,
environment TEXT DEFAULT 'test',
details TEXT
)""")
for idx in [
"CREATE INDEX IF NOT EXISTS idx_patients_cancer_type ON patients(cancer_type)",
"CREATE INDEX IF NOT EXISTS idx_patients_status ON patients(status)",
"CREATE INDEX IF NOT EXISTS idx_patients_doctor ON patients(assigned_doctor_id)",
"CREATE INDEX IF NOT EXISTS idx_patients_nurse ON patients(assigned_nurse_id)",
"CREATE INDEX IF NOT EXISTS idx_medications_patient ON medications(patient_id)",
"CREATE INDEX IF NOT EXISTS idx_vital_signs_patient ON vital_signs(patient_id)",
"CREATE INDEX IF NOT EXISTS idx_alerts_patient ON alerts(patient_id)",
"CREATE INDEX IF NOT EXISTS idx_alerts_severity ON alerts(severity)",
"CREATE INDEX IF NOT EXISTS idx_audit_log_user ON audit_log(user_id)",
"CREATE INDEX IF NOT EXISTS idx_nurse_assignments ON nurse_patient_assignments(nurse_id, patient_id)",
]:
c.execute(idx)
conn.commit()
conn.close()
print("[OK] Database initialized")
return True
def hash_password(password, salt=None):
if salt is None:
salt = os.urandom(32).hex()
pw_hash = hashlib.sha256((salt + password).encode()).hexdigest()
return pw_hash, salt
def verify_password(password, stored_hash, salt):
return hashlib.sha256((salt + password).encode()).hexdigest() == stored_hash
if __name__ == "__main__":
init_db()