hospital-management/tools/create_insurance_tables.py
Marwan Alwali 263292f6be update
2025-11-04 00:50:06 +03:00

214 lines
9.9 KiB
Python

"""
Directly create insurance_approvals tables using SQL.
"""
import sqlite3
import os
db_path = 'db.sqlite3'
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
print("Creating insurance_approvals tables directly...")
# Disable foreign key checks
cursor.execute("PRAGMA foreign_keys = OFF;")
# Create tables from migration SQL
tables_sql = [
# InsuranceApprovalRequest table
"""
CREATE TABLE IF NOT EXISTS "insurance_approvals_request" (
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
"approval_id" char(32) NOT NULL UNIQUE,
"approval_number" varchar(30) NOT NULL UNIQUE,
"request_type" varchar(20) NOT NULL,
"service_description" varchar(500) NOT NULL,
"procedure_codes" text NOT NULL,
"diagnosis_codes" text NOT NULL,
"clinical_justification" text NOT NULL,
"medical_necessity" text NULL,
"alternative_treatments_tried" text NULL,
"requested_quantity" integer unsigned NOT NULL CHECK ("requested_quantity" >= 0),
"requested_visits" integer unsigned NULL CHECK ("requested_visits" >= 0),
"requested_units" integer unsigned NULL CHECK ("requested_units" >= 0),
"service_start_date" date NOT NULL,
"service_end_date" date NULL,
"status" varchar(30) NOT NULL,
"priority" varchar(20) NOT NULL,
"submission_method" varchar(20) NULL,
"submitted_date" datetime NULL,
"decision_date" datetime NULL,
"authorization_number" varchar(100) NULL,
"reference_number" varchar(100) NULL,
"approved_quantity" integer unsigned NULL CHECK ("approved_quantity" >= 0),
"approved_visits" integer unsigned NULL CHECK ("approved_visits" >= 0),
"approved_units" integer unsigned NULL CHECK ("approved_units" >= 0),
"approved_amount" decimal NULL,
"effective_date" date NULL,
"expiration_date" date NULL,
"denial_reason" text NULL,
"denial_code" varchar(50) NULL,
"appeal_date" datetime NULL,
"appeal_reason" text NULL,
"appeal_deadline" date NULL,
"last_contact_date" datetime NULL,
"last_contact_method" varchar(20) NULL,
"last_contact_notes" text NULL,
"is_urgent" bool NOT NULL,
"is_expedited" bool NOT NULL,
"requires_peer_review" bool NOT NULL,
"internal_notes" text NULL,
"insurance_notes" text NULL,
"created_at" datetime NOT NULL,
"updated_at" datetime NOT NULL,
"tenant_id" bigint NOT NULL REFERENCES "core_tenant" ("id") DEFERRABLE INITIALLY DEFERRED,
"patient_id" bigint NOT NULL REFERENCES "patients_patient_profile" ("id") DEFERRABLE INITIALLY DEFERRED,
"insurance_info_id" bigint NOT NULL REFERENCES "patients_insurance_info" ("id") DEFERRABLE INITIALLY DEFERRED,
"content_type_id" integer NOT NULL REFERENCES "django_content_type" ("id") DEFERRABLE INITIALLY DEFERRED,
"object_id" integer unsigned NOT NULL CHECK ("object_id" >= 0),
"submitted_by_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED,
"assigned_to_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED,
"requesting_provider_id" integer NOT NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED,
"created_by_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED
);
""",
# ApprovalDocument table
"""
CREATE TABLE IF NOT EXISTS "insurance_approvals_document" (
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
"document_id" char(32) NOT NULL UNIQUE,
"document_type" varchar(30) NOT NULL,
"title" varchar(200) NOT NULL,
"description" text NULL,
"file" varchar(100) NOT NULL,
"file_size" integer unsigned NOT NULL CHECK ("file_size" >= 0),
"mime_type" varchar(100) NOT NULL,
"uploaded_at" datetime NOT NULL,
"approval_request_id" bigint NOT NULL REFERENCES "insurance_approvals_request" ("id") DEFERRABLE INITIALLY DEFERRED,
"uploaded_by_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED
);
""",
# ApprovalStatusHistory table
"""
CREATE TABLE IF NOT EXISTS "insurance_approvals_status_history" (
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
"from_status" varchar(30) NULL,
"to_status" varchar(30) NOT NULL,
"reason" text NULL,
"notes" text NULL,
"changed_at" datetime NOT NULL,
"approval_request_id" bigint NOT NULL REFERENCES "insurance_approvals_request" ("id") DEFERRABLE INITIALLY DEFERRED,
"changed_by_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED
);
""",
# ApprovalCommunicationLog table
"""
CREATE TABLE IF NOT EXISTS "insurance_approvals_communication_log" (
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
"communication_id" char(32) NOT NULL UNIQUE,
"communication_type" varchar(20) NOT NULL,
"contact_person" varchar(200) NULL,
"contact_number" varchar(50) NULL,
"subject" varchar(200) NOT NULL,
"message" text NOT NULL,
"response" text NULL,
"outcome" varchar(200) NULL,
"follow_up_required" bool NOT NULL,
"follow_up_date" date NULL,
"communicated_at" datetime NOT NULL,
"approval_request_id" bigint NOT NULL REFERENCES "insurance_approvals_request" ("id") DEFERRABLE INITIALLY DEFERRED,
"communicated_by_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED
);
""",
# ApprovalTemplate table
"""
CREATE TABLE IF NOT EXISTS "insurance_approvals_template" (
"id" integer NOT NULL PRIMARY KEY AUTOINCREMENT,
"template_id" char(32) NOT NULL UNIQUE,
"name" varchar(200) NOT NULL,
"description" text NULL,
"request_type" varchar(20) NOT NULL,
"insurance_company" varchar(200) NULL,
"clinical_justification_template" text NOT NULL,
"medical_necessity_template" text NULL,
"required_documents" text NOT NULL,
"required_codes" text NOT NULL,
"is_active" bool NOT NULL,
"usage_count" integer unsigned NOT NULL CHECK ("usage_count" >= 0),
"created_at" datetime NOT NULL,
"updated_at" datetime NOT NULL,
"tenant_id" bigint NOT NULL REFERENCES "core_tenant" ("id") DEFERRABLE INITIALLY DEFERRED,
"created_by_id" integer NULL REFERENCES "accounts_user" ("id") DEFERRABLE INITIALLY DEFERRED,
CONSTRAINT "insurance_approvals_template_tenant_id_name_uniq" UNIQUE ("tenant_id", "name")
);
"""
]
# Create tables
for i, sql in enumerate(tables_sql, 1):
try:
cursor.execute(sql)
print(f"✓ Created table {i}/5")
except sqlite3.OperationalError as e:
if "already exists" in str(e):
print(f"✓ Table {i}/5 already exists")
else:
print(f"✗ Error creating table {i}/5: {e}")
# Create indexes
indexes_sql = [
'CREATE INDEX IF NOT EXISTS "insurance_a_tenant__d6763e_idx" ON "insurance_approvals_request" ("tenant_id", "status");',
'CREATE INDEX IF NOT EXISTS "insurance_a_patient_00ddbd_idx" ON "insurance_approvals_request" ("patient_id", "status");',
'CREATE INDEX IF NOT EXISTS "insurance_a_insuran_438196_idx" ON "insurance_approvals_request" ("insurance_info_id", "status");',
'CREATE INDEX IF NOT EXISTS "insurance_a_approva_cefd6a_idx" ON "insurance_approvals_request" ("approval_number");',
'CREATE INDEX IF NOT EXISTS "insurance_a_authori_f05618_idx" ON "insurance_approvals_request" ("authorization_number");',
'CREATE INDEX IF NOT EXISTS "insurance_a_expirat_ef068e_idx" ON "insurance_approvals_request" ("expiration_date");',
'CREATE INDEX IF NOT EXISTS "insurance_a_assigne_19b007_idx" ON "insurance_approvals_request" ("assigned_to_id", "status");',
'CREATE INDEX IF NOT EXISTS "insurance_a_content_de7404_idx" ON "insurance_approvals_request" ("content_type_id", "object_id");',
'CREATE INDEX IF NOT EXISTS "insurance_a_priorit_0abcbd_idx" ON "insurance_approvals_request" ("priority", "status");',
'CREATE INDEX IF NOT EXISTS "insurance_a_approva_a5298b_idx" ON "insurance_approvals_status_history" ("approval_request_id", "changed_at");',
'CREATE INDEX IF NOT EXISTS "insurance_a_approva_eae9e0_idx" ON "insurance_approvals_document" ("approval_request_id", "document_type");',
'CREATE INDEX IF NOT EXISTS "insurance_a_approva_fa779d_idx" ON "insurance_approvals_communication_log" ("approval_request_id", "communicated_at");',
'CREATE INDEX IF NOT EXISTS "insurance_a_follow__2a3b0a_idx" ON "insurance_approvals_communication_log" ("follow_up_required", "follow_up_date");',
'CREATE INDEX IF NOT EXISTS "insurance_a_tenant__8f9c3a_idx" ON "insurance_approvals_template" ("tenant_id", "is_active");',
'CREATE INDEX IF NOT EXISTS "insurance_a_request_4b3e2f_idx" ON "insurance_approvals_template" ("request_type");',
]
print("\nCreating indexes...")
for i, sql in enumerate(indexes_sql, 1):
try:
cursor.execute(sql)
except sqlite3.OperationalError as e:
print(f" Index {i}: {e}")
print(f"✓ Created {len(indexes_sql)} indexes")
# Mark migration as applied
cursor.execute("""
INSERT OR IGNORE INTO django_migrations (app, name, applied)
VALUES ('insurance_approvals', '0001_initial', datetime('now'))
""")
conn.commit()
# Re-enable foreign key checks
cursor.execute("PRAGMA foreign_keys = ON;")
conn.close()
print("\n" + "="*60)
print("✓ Insurance Approvals tables created successfully!")
print("="*60)
print("\nTables created:")
print(" 1. insurance_approvals_request")
print(" 2. insurance_approvals_document")
print(" 3. insurance_approvals_status_history")
print(" 4. insurance_approvals_communication_log")
print(" 5. insurance_approvals_template")
print("\nModule is ready to use!")
print("Access admin at: http://127.0.0.1:8000/admin/insurance_approvals/")