src/db/seed.py

import sqlite3
from typing import Dict, List, Tuple
from passlib.context import CryptContext

from src.utils.time_utils import utc_now_iso
from src.utils.sqlite_retry import run_tx

PWD_CTX = CryptContext(schemes=["bcrypt"], deprecated="auto")

ROOMS: Dict[str, List[int]] = {
    "Suite": [101, 102, 103, 104],
    "Standard King": [
        201, 202, 203, 206, 208, 209, 210,
        301, 302, 303, 306, 308, 309, 310,
        401, 402, 403, 406, 408, 409, 410,
        501, 502, 503, 508, 509, 510,
        601, 602, 603, 606, 608, 609,
        701, 706, 708, 709, 710
    ],
    "Standard Twin": [207, 307, 407, 507, 610, 707],
    "Deluxe King (Road Side)": [205, 212, 305, 312, 405, 412, 505, 512, 605, 612, 705, 712],
    "Deluxe Twin (Road Side)": [204, 211, 304, 311, 404, 411, 504, 511, 604, 611, 704, 711],
    "Unknown": [506, 607, 702, 703] + list(range(801, 811)),
}

DR_ROOMS = {102, 104}
STAFF_CLOSED = {506, 607, 702, 703}
DEFAULT_OOS = set(range(401, 413)) | {502, 503, 508, 509, 602, 608} | set(range(801, 811))

KNOWN_PROBLEMS: List[Tuple[int, str, str]] = [
    (202, "bed", "bed"),
    (205, "heater", "heater"),
    (207, "heater", "heater"),
    (208, "heater", "heater"),
    (209, "heater", "heater"),
    (301, "heater", "heater"),
    (302, "heater", "heater"),
    (303, "heater", "heater"),
    (306, "heater", "heater"),
    (308, "heater", "heater"),
    (309, "heater", "heater"),
    (504, "heater", "heater"),
    (505, "heater", "heater"),
    (507, "heater", "heater"),
    (510, "heater", "heater"),
    (605, "heater", "heater"),
    (609, "heater", "heater"),
    (610, "heater", "heater"),
    (612, "heater", "heater"),
    (705, "heater", "heater"),
    (708, "heater", "heater"),
    (712, "heater", "heater"),
    (212, "ac", "ac"),
    (311, "ac", "ac"),
    (312, "ac", "ac"),
    (611, "bed", "bed"),
]

SEED_USERS = [
    ("Amir", "staff", "Amir-2468"),
    ("Shahin", "staff", "Shahin-739"),
    ("Kazem", "staff", "Kazem-513"),
    ("Brian", "staff", "Brian-842"),
    ("Mahfuz", "staff", "Mahfuz-617"),
    ("Dr", "admin", "Dr-9051"),
]

def seed_if_empty(conn: sqlite3.Connection) -> None:
    def _seed(c: sqlite3.Connection):
        now = utc_now_iso()

        rooms_count = c.execute("SELECT COUNT(*) AS c FROM rooms").fetchone()["c"]
        if rooms_count == 0:
            room_type_map = {}
            for rtype, nums in ROOMS.items():
                for n in nums:
                    room_type_map[n] = rtype

            for room_no in sorted(room_type_map.keys()):
                sellable = 1
                if room_no in DR_ROOMS or room_no in STAFF_CLOSED:
                    sellable = 0

                status = "free"
                if room_no in DEFAULT_OOS:
                    status = "out_of_service"

                c.execute(
                    """
                    INSERT INTO rooms(room_no, room_type, sellable, status, created_at_utc, updated_at_utc)
                    VALUES(?,?,?,?,?,?)
                    """,
                    (room_no, room_type_map[room_no], sellable, status, now, now),
                )
                c.execute(
                    """
                    INSERT INTO room_status_history(room_no, from_status, to_status, note, changed_by, changed_at_utc)
                    VALUES(?,?,?,?,?,?)
                    """,
                    (room_no, None, status, "Initial seed", "system", now),
                )
                c.execute(
                    "INSERT INTO audit_log(entity, entity_id, action, note, changed_by, changed_at_utc) VALUES(?,?,?,?,?,?)",
                    ("room", str(room_no), "seed", "Initial seed", "system", now),
                )

        prob_count = c.execute("SELECT COUNT(*) AS c FROM room_problems").fetchone()["c"]
        if prob_count == 0:
            for room_no, ptype, desc in KNOWN_PROBLEMS:
                c.execute(
                    """
                    INSERT INTO room_problems(room_no, problem_type, description, is_active, created_at_utc, updated_at_utc)
                    VALUES(?,?,?,?,?,?)
                    """,
                    (room_no, ptype, desc, 1, now, now),
                )
                c.execute(
                    """
                    INSERT INTO room_problem_history(room_no, problem_type, action, description, note, changed_by, changed_at_utc)
                    VALUES(?,?,?,?,?,?,?)
                    """,
                    (room_no, ptype, "add", desc, "Initial seed", "system", now),
                )
                c.execute(
                    "INSERT INTO audit_log(entity, entity_id, action, note, changed_by, changed_at_utc) VALUES(?,?,?,?,?,?)",
                    ("problem", f"{room_no}:{ptype}", "seed", "Initial seed", "system", now),
                )

        user_count = c.execute("SELECT COUNT(*) AS c FROM users").fetchone()["c"]
        if user_count == 0:
            for username, role, plain in SEED_USERS:
                c.execute(
                    """
                    INSERT INTO users(username, role, password_hash, is_active, created_at_utc)
                    VALUES(?,?,?,?,?)
                    """,
                    (username, role, PWD_CTX.hash(plain), 1, now),
                )
            c.execute(
                "INSERT INTO audit_log(entity, entity_id, action, note, changed_by, changed_at_utc) VALUES(?,?,?,?,?,?)",
                ("user", "*", "seed", "Initial users created", "system", now),
            )

    run_tx(conn, _seed)