Spaces:

larxius
/

larmedix

Running on CPU Upgrade

larmedix / init_db.py

larxius

revert: superadmin DB-owned, seed-once

04928f4 about 2 months ago

7.47 kB

from app import create_app
from models import db, User, Hospital
from database import get_database_uri, is_postgres
from werkzeug.security import generate_password_hash
import os

app = create_app()

with app.app_context():
    uri = get_database_uri()
    # SAFETY: never drop production PostgreSQL (larmedix-db) by accident.
    # First init on GCP only needs create_all (auto-creates tables).
    # Set ALLOW_DROP_TABLES=true explicitly for a fresh local reset.
    allow_drop = os.environ.get("ALLOW_DROP_TABLES", "false").lower() == "true"
    if is_postgres(uri) and not allow_drop:
        print(f"Postgres detected ({uri.split('@')[-1]}). Running create_all only (no drop).")
        db.create_all()
    else:
        if is_postgres(uri):
            print("WARNING: ALLOW_DROP_TABLES=true — dropping ALL Postgres tables!")
        db.drop_all()
        db.create_all()
    
    
    # Create Default Hospital
    default_hospital = Hospital.query.first()
    if not default_hospital:
        default_hospital = Hospital(name="First Health Hospital", email="admin@firsthealth.com")
        db.session.add(default_hospital)
        db.session.commit()
        print("Default Hospital created")

    # Seed Superadmin once (dev script). Prod: database.init_database does the same.
    # DB is source of truth afterwards — edit via SQL/app, not env.
    superadmin = User.query.filter_by(role='superadmin').first()
    if not superadmin:
        hashed_password = generate_password_hash('superadmin123')
        new_superadmin = User(email='superadmin@system.com', username='superadmin', password=hashed_password, role='superadmin', hospital_id=None)
        db.session.add(new_superadmin)
        print("Superadmin user created (email: superadmin@system.com)")

    # Seed Admin
    admin = User.query.filter_by(email='admin@firsthealth.com').first()
    if not admin:
        hashed_password = generate_password_hash('admin123')
        new_admin = User(email='admin@firsthealth.com', password=hashed_password, plaintext_password='admin123', role='admin', hospital_id=default_hospital.id)
        db.session.add(new_admin)
        print("Admin user created (email: admin@firsthealth.com, password: admin123)")

    # Seed Doctor
    doctor = User.query.filter_by(email='doctor1@firsthealth.com').first()
    if not doctor:
        hashed_password = generate_password_hash('doctor123')
        new_doctor = User(email='doctor1@firsthealth.com', password=hashed_password, plaintext_password='doctor123', role='doctor', hospital_id=default_hospital.id)
        db.session.add(new_doctor)
        print("Doctor user created (email: doctor1@firsthealth.com, password: doctor123)")

    # Seed Reception
    reception = User.query.filter_by(email='reception1@firsthealth.com').first()
    if not reception:
        hashed_password = generate_password_hash('reception123')
        new_reception = User(email='reception1@firsthealth.com', password=hashed_password, plaintext_password='reception123', role='reception', hospital_id=default_hospital.id)
        db.session.add(new_reception)
        print("Receptionist user created (email: reception1@firsthealth.com, password: reception123)")

    # Seed Patient
    patient = User.query.filter_by(email='patient1@firsthealth.com').first()
    if not patient:
        hashed_password = generate_password_hash('patient123')
        new_patient = User(email='patient1@firsthealth.com', password=hashed_password, plaintext_password='patient123', role='patient', hospital_id=default_hospital.id)
        db.session.add(new_patient)
        db.session.commit() # Commit to get patient user id
        # Link Patient Profile
        from models import PatientProfile
        import uuid
        from datetime import datetime
        new_profile = PatientProfile(
            user_id=new_patient.id,
            hospital_id=default_hospital.id,
            name="John Doe",
            dob="1985-05-15",
            age=41,
            gender="M",
            contact="9876543210",
            mobile="9876543210",
            address="123 Care Street, Medical Zone",
            blood_group="O+",
            occupation="Engineer",
            emergency_contact="Jane Doe (Wife): 9876543211",
            patient_uuid="P-" + datetime.today().strftime('%Y%m%d') + "-1001"
        )
        db.session.add(new_profile)
        print("Patient user and John Doe profile created (username: patient1, password: patient123)")
        
    # Seed AI Models (Phase 5)
    from models import DiseaseDefinition, DynamicQuestion, ClinicalRule, Branch, Department
    import json

    if not Branch.query.first():
        b1 = Branch(name="KIMS Main", hospital_id=default_hospital.id)
        b2 = Branch(name="KIMS North", hospital_id=default_hospital.id)
        db.session.add_all([b1, b2])
        db.session.commit()
        
        d1 = Department(name="Gastroenterology", branch_id=b1.id, hospital_id=default_hospital.id)
        d2 = Department(name="Urology", branch_id=b1.id, hospital_id=default_hospital.id)
        d3 = Department(name="General Medicine", branch_id=b2.id, hospital_id=default_hospital.id)
        db.session.add_all([d1, d2, d3])
        db.session.commit()
    
    if not DiseaseDefinition.query.first():
        gastritis = DiseaseDefinition(
            name="Gastritis",
            symptoms=json.dumps(["Epigastric Pain", "Nausea", "Bloating", "Loss of appetite"]),
            risk_factors=json.dumps(["NSAID use", "H. pylori infection", "Alcohol consumption", "Smoking"]),
            investigations=json.dumps(["Upper GI Endoscopy", "H. pylori stool antigen", "CBC"]),
            severity_scale=json.dumps({"Mild": "Occasional pain", "Severe": "Bleeding, intense vomiting, unintended weight loss"}),
            treatments=json.dumps(["PPIs (Omeprazole)", "Antacids", "H. pylori eradication therapy"]),
            guidelines="Standard Gastroenterology Guidelines 2026"
        )
        bph = DiseaseDefinition(
            name="BPH",
            symptoms=json.dumps(["Frequent urination", "Nocturia", "Weak stream", "Incomplete emptying"]),
            risk_factors=json.dumps(["Age > 50", "Family history", "Obesity"]),
            investigations=json.dumps(["PSA Blood Test", "Urinalysis", "Ultrasound"]),
            severity_scale=json.dumps({"Mild": "IPSS 1-7", "Moderate": "IPSS 8-19", "Severe": "IPSS 20-35"}),
            treatments=json.dumps(["Alpha-blockers", "5-ARIs", "TURP surgery"]),
            guidelines="AUA Guidelines for Benign Prostatic Hyperplasia"
        )
        db.session.add_all([gastritis, bph])

    if not DynamicQuestion.query.first():
        q1 = DynamicQuestion(question_text="Do you experience burning pain in your upper abdomen?", answer_type="Boolean", disease_mapping="Gastritis", weight=1.5)
        q2 = DynamicQuestion(question_text="Are you waking up multiple times at night to urinate?", answer_type="Boolean", disease_mapping="BPH", weight=1.8)
        db.session.add_all([q1, q2])

    if not ClinicalRule.query.first():
        rule1 = ClinicalRule(
            name="Elderly BPH Risk",
            condition_logic=json.dumps({"AND": [{"field": "Age", "op": ">", "val": 60}, {"field": "PSA", "op": ">", "val": 4}]}),
            action_logic=json.dumps({"adjust_score": {"disease": "BPH", "value": 0.25}}),
            is_active=True
        )
        db.session.add(rule1)

    db.session.commit()
    print("Database seeding completed.")

	from app import create_app
	from models import db, User, Hospital
	from database import get_database_uri, is_postgres
	from werkzeug.security import generate_password_hash
	import os

	app = create_app()

	with app.app_context():
	uri = get_database_uri()
	# SAFETY: never drop production PostgreSQL (larmedix-db) by accident.
	# First init on GCP only needs create_all (auto-creates tables).
	# Set ALLOW_DROP_TABLES=true explicitly for a fresh local reset.
	allow_drop = os.environ.get("ALLOW_DROP_TABLES", "false").lower() == "true"
	if is_postgres(uri) and not allow_drop:
	print(f"Postgres detected ({uri.split('@')[-1]}). Running create_all only (no drop).")
	db.create_all()
	else:
	if is_postgres(uri):
	print("WARNING: ALLOW_DROP_TABLES=true — dropping ALL Postgres tables!")
	db.drop_all()
	db.create_all()


	# Create Default Hospital
	default_hospital = Hospital.query.first()
	if not default_hospital:
	default_hospital = Hospital(name="First Health Hospital", email="admin@firsthealth.com")
	db.session.add(default_hospital)
	db.session.commit()
	print("Default Hospital created")

	# Seed Superadmin once (dev script). Prod: database.init_database does the same.
	# DB is source of truth afterwards — edit via SQL/app, not env.
	superadmin = User.query.filter_by(role='superadmin').first()
	if not superadmin:
	hashed_password = generate_password_hash('superadmin123')
	new_superadmin = User(email='superadmin@system.com', username='superadmin', password=hashed_password, role='superadmin', hospital_id=None)
	db.session.add(new_superadmin)
	print("Superadmin user created (email: superadmin@system.com)")

	# Seed Admin
	admin = User.query.filter_by(email='admin@firsthealth.com').first()
	if not admin:
	hashed_password = generate_password_hash('admin123')
	new_admin = User(email='admin@firsthealth.com', password=hashed_password, plaintext_password='admin123', role='admin', hospital_id=default_hospital.id)
	db.session.add(new_admin)
	print("Admin user created (email: admin@firsthealth.com, password: admin123)")

	# Seed Doctor
	doctor = User.query.filter_by(email='doctor1@firsthealth.com').first()
	if not doctor:
	hashed_password = generate_password_hash('doctor123')
	new_doctor = User(email='doctor1@firsthealth.com', password=hashed_password, plaintext_password='doctor123', role='doctor', hospital_id=default_hospital.id)
	db.session.add(new_doctor)
	print("Doctor user created (email: doctor1@firsthealth.com, password: doctor123)")

	# Seed Reception
	reception = User.query.filter_by(email='reception1@firsthealth.com').first()
	if not reception:
	hashed_password = generate_password_hash('reception123')
	new_reception = User(email='reception1@firsthealth.com', password=hashed_password, plaintext_password='reception123', role='reception', hospital_id=default_hospital.id)
	db.session.add(new_reception)
	print("Receptionist user created (email: reception1@firsthealth.com, password: reception123)")

	# Seed Patient
	patient = User.query.filter_by(email='patient1@firsthealth.com').first()
	if not patient:
	hashed_password = generate_password_hash('patient123')
	new_patient = User(email='patient1@firsthealth.com', password=hashed_password, plaintext_password='patient123', role='patient', hospital_id=default_hospital.id)
	db.session.add(new_patient)
	db.session.commit() # Commit to get patient user id
	# Link Patient Profile
	from models import PatientProfile
	import uuid
	from datetime import datetime
	new_profile = PatientProfile(
	user_id=new_patient.id,
	hospital_id=default_hospital.id,
	name="John Doe",
	dob="1985-05-15",
	age=41,
	gender="M",
	contact="9876543210",
	mobile="9876543210",
	address="123 Care Street, Medical Zone",
	blood_group="O+",
	occupation="Engineer",
	emergency_contact="Jane Doe (Wife): 9876543211",
	patient_uuid="P-" + datetime.today().strftime('%Y%m%d') + "-1001"
	)
	db.session.add(new_profile)
	print("Patient user and John Doe profile created (username: patient1, password: patient123)")

	# Seed AI Models (Phase 5)
	from models import DiseaseDefinition, DynamicQuestion, ClinicalRule, Branch, Department
	import json

	if not Branch.query.first():
	b1 = Branch(name="KIMS Main", hospital_id=default_hospital.id)
	b2 = Branch(name="KIMS North", hospital_id=default_hospital.id)
	db.session.add_all([b1, b2])
	db.session.commit()

	d1 = Department(name="Gastroenterology", branch_id=b1.id, hospital_id=default_hospital.id)
	d2 = Department(name="Urology", branch_id=b1.id, hospital_id=default_hospital.id)
	d3 = Department(name="General Medicine", branch_id=b2.id, hospital_id=default_hospital.id)
	db.session.add_all([d1, d2, d3])
	db.session.commit()

	if not DiseaseDefinition.query.first():
	gastritis = DiseaseDefinition(
	name="Gastritis",
	symptoms=json.dumps(["Epigastric Pain", "Nausea", "Bloating", "Loss of appetite"]),
	risk_factors=json.dumps(["NSAID use", "H. pylori infection", "Alcohol consumption", "Smoking"]),
	investigations=json.dumps(["Upper GI Endoscopy", "H. pylori stool antigen", "CBC"]),
	severity_scale=json.dumps({"Mild": "Occasional pain", "Severe": "Bleeding, intense vomiting, unintended weight loss"}),
	treatments=json.dumps(["PPIs (Omeprazole)", "Antacids", "H. pylori eradication therapy"]),
	guidelines="Standard Gastroenterology Guidelines 2026"
	)
	bph = DiseaseDefinition(
	name="BPH",
	symptoms=json.dumps(["Frequent urination", "Nocturia", "Weak stream", "Incomplete emptying"]),
	risk_factors=json.dumps(["Age > 50", "Family history", "Obesity"]),
	investigations=json.dumps(["PSA Blood Test", "Urinalysis", "Ultrasound"]),
	severity_scale=json.dumps({"Mild": "IPSS 1-7", "Moderate": "IPSS 8-19", "Severe": "IPSS 20-35"}),
	treatments=json.dumps(["Alpha-blockers", "5-ARIs", "TURP surgery"]),
	guidelines="AUA Guidelines for Benign Prostatic Hyperplasia"
	)
	db.session.add_all([gastritis, bph])

	if not DynamicQuestion.query.first():
	q1 = DynamicQuestion(question_text="Do you experience burning pain in your upper abdomen?", answer_type="Boolean", disease_mapping="Gastritis", weight=1.5)
	q2 = DynamicQuestion(question_text="Are you waking up multiple times at night to urinate?", answer_type="Boolean", disease_mapping="BPH", weight=1.8)
	db.session.add_all([q1, q2])

	if not ClinicalRule.query.first():
	rule1 = ClinicalRule(
	name="Elderly BPH Risk",
	condition_logic=json.dumps({"AND": [{"field": "Age", "op": ">", "val": 60}, {"field": "PSA", "op": ">", "val": 4}]}),
	action_logic=json.dumps({"adjust_score": {"disease": "BPH", "value": 0.25}}),
	is_active=True
	)
	db.session.add(rule1)

	db.session.commit()
	print("Database seeding completed.")