zkxw/backend/scripts/seed_clean_rooms.py

#!/usr/bin/env python3
"""批量初始化洁净室及子表测试数据（biz_clean_room + access_log + uv_disinfect_log），已存在编码则跳过。"""
from __future__ import annotations

import sys
from datetime import datetime
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from app.core.database import SessionLocal
from app.crud import clean_room as clean_room_svc

ADMIN_ID = "d3000000000000000000000000000001"
TEST_USER_ID = "b13c2be24e7111f19f4700163e3bdd9601"

# (room_code, room_name, room_grade, room_status, area_sqm, order_num, remark)
CLEAN_ROOMS = [
    ("CR-A01", "细胞制备A级洁净室", "A", "zc", 45.50, 1, "主制备区，日常细胞培养"),
    ("CR-B01", "样本接收B级洁净室", "B", "zc", 32.00, 2, "样本接收与初检"),
    ("CR-B02", "质检B级洁净室", "B", "zc", 28.80, 3, "质检与留样"),
    ("CR-C01", "物料暂存C级洁净室", "C", "zc", 18.60, 4, "物料暂存与缓冲"),
    ("CR-D01", "更衣D级洁净室", "D", "yc", 12.00, 5, "压差异常，待检修"),
]

# room_code -> list of access log dicts
ACCESS_LOGS = {
    "CR-A01": [
        {
            "user_id": ADMIN_ID,
            "person_name": "管理员",
            "dept_name": "综合部",
            "enter_time": datetime(2026, 6, 10, 8, 15),
            "purpose": "日常巡检",
            "leave_time": datetime(2026, 6, 10, 8, 45),
            "body_temp": "36.5",
        },
        {
            "user_id": TEST_USER_ID,
            "person_name": "中科兴皖测试1",
            "dept_name": "技术研发部",
            "enter_time": datetime(2026, 6, 10, 9, 0),
            "purpose": "细胞培养操作",
            "leave_time": datetime(2026, 6, 10, 12, 30),
            "body_temp": "36.4",
        },
    ],
    "CR-B01": [
        {
            "user_id": TEST_USER_ID,
            "person_name": "中科兴皖测试1",
            "dept_name": "生产部",
            "enter_time": datetime(2026, 6, 11, 7, 50),
            "purpose": "样本接收登记",
            "leave_time": datetime(2026, 6, 11, 8, 40),
            "body_temp": "36.6",
        },
        {
            "person_name": "张质检",
            "dept_name": "质量安全部",
            "enter_time": datetime(2026, 6, 11, 14, 0),
            "purpose": "样本初检",
            "leave_time": datetime(2026, 6, 11, 15, 20),
            "body_temp": "36.3",
            "remark": "外部访客",
        },
    ],
    "CR-B02": [
        {
            "user_id": ADMIN_ID,
            "person_name": "管理员",
            "dept_name": "质量安全部",
            "enter_time": datetime(2026, 6, 12, 10, 0),
            "purpose": "留样复核",
            "leave_time": datetime(2026, 6, 12, 11, 0),
            "body_temp": "36.5",
        },
    ],
    "CR-C01": [
        {
            "person_name": "李物料",
            "dept_name": "生产部",
            "enter_time": datetime(2026, 6, 12, 15, 30),
            "purpose": "物料入库",
            "leave_time": datetime(2026, 6, 12, 16, 0),
            "body_temp": "36.7",
        },
    ],
    "CR-D01": [
        {
            "user_id": ADMIN_ID,
            "person_name": "管理员",
            "dept_name": "综合部",
            "enter_time": datetime(2026, 6, 10, 16, 0),
            "purpose": "压差异常排查",
            "leave_time": None,
            "body_temp": "36.5",
            "remark": "尚未离开",
        },
    ],
}

# room_code -> list of uv disinfect log dicts
UV_LOGS = {
    "CR-A01": [
        {
            "user_id": ADMIN_ID,
            "operator_name": "管理员",
            "start_time": datetime(2026, 6, 9, 18, 0),
            "end_time": datetime(2026, 6, 9, 19, 0),
            "remark": "班后消毒",
        },
        {
            "user_id": ADMIN_ID,
            "operator_name": "管理员",
            "start_time": datetime(2026, 6, 10, 18, 0),
            "end_time": datetime(2026, 6, 10, 19, 0),
            "remark": "班后消毒",
        },
    ],
    "CR-B01": [
        {
            "user_id": ADMIN_ID,
            "operator_name": "管理员",
            "start_time": datetime(2026, 6, 10, 17, 30),
            "end_time": datetime(2026, 6, 10, 18, 30),
        },
    ],
    "CR-B02": [
        {
            "user_id": ADMIN_ID,
            "operator_name": "管理员",
            "start_time": datetime(2026, 6, 11, 18, 0),
            "end_time": datetime(2026, 6, 11, 19, 0),
        },
    ],
    "CR-C01": [
        {
            "user_id": ADMIN_ID,
            "operator_name": "管理员",
            "start_time": datetime(2026, 6, 11, 17, 0),
            "end_time": datetime(2026, 6, 11, 17, 45),
            "remark": "C级区缩短时长",
        },
    ],
    "CR-D01": [
        {
            "user_id": ADMIN_ID,
            "operator_name": "管理员",
            "start_time": datetime(2026, 6, 10, 14, 30),
            "end_time": None,
            "remark": "异常区暂停消毒",
        },
    ],
}


def _room_exists(db, code: str, tenant_no: str) -> bool:
    return clean_room_svc.room_code_exists(db, code, tenant_no)


def _access_log_exists(db, room_id: str, enter_time: datetime, tenant_no: str) -> bool:
    rows, _ = clean_room_svc.list_access_logs(
        db, page=1, page_size=100, clean_room_id=room_id, tenant_no=tenant_no
    )
    ts = enter_time.strftime("%Y-%m-%d %H:%M:%S")
    return any(r.get("enterTime") == ts for r in rows)


def _uv_log_exists(db, room_id: str, start_time: datetime, tenant_no: str) -> bool:
    rows, _ = clean_room_svc.list_uv_disinfect_logs(
        db, page=1, page_size=100, clean_room_id=room_id, tenant_no=tenant_no
    )
    ts = start_time.strftime("%Y-%m-%d %H:%M:%S")
    return any(r.get("startTime") == ts for r in rows)


def seed(tenant_no: str = "zkxw", user_id: str = ADMIN_ID) -> None:
    db = SessionLocal()
    rooms_created = 0
    rooms_skipped = 0
    access_created = 0
    access_skipped = 0
    uv_created = 0
    uv_skipped = 0
    try:
        room_ids: dict[str, str] = {}
        for code, name, grade, status, area, order_num, remark in CLEAN_ROOMS:
            if _room_exists(db, code, tenant_no):
                rooms_skipped += 1
                existing = clean_room_svc.list_clean_rooms(db, page=1, page_size=1, keyword=code, tenant_no=tenant_no)
                if existing[0]:
                    room_ids[code] = existing[0][0]["id"]
                continue
            obj = clean_room_svc.create_clean_room(
                db,
                room_code=code,
                room_name=name,
                room_grade=grade,
                room_status=status,
                area_sqm=area,
                tenant_no=tenant_no,
                order_num=order_num,
                remark=remark,
                user_id=user_id,
            )
            room_ids[code] = obj.id
            rooms_created += 1

        for code, logs in ACCESS_LOGS.items():
            room_id = room_ids.get(code)
            if not room_id:
                continue
            for idx, log in enumerate(logs, start=1):
                enter_time = log["enter_time"]
                if _access_log_exists(db, room_id, enter_time, tenant_no):
                    access_skipped += 1
                    continue
                clean_room_svc.create_access_log(
                    db,
                    clean_room_id=room_id,
                    user_id=log.get("user_id"),
                    person_name=log.get("person_name"),
                    dept_name=log.get("dept_name"),
                    enter_time=enter_time,
                    purpose=log.get("purpose"),
                    leave_time=log.get("leave_time"),
                    body_temp=log.get("body_temp"),
                    tenant_no=tenant_no,
                    order_num=idx,
                    remark=log.get("remark"),
                    user_id_param=user_id,
                )
                access_created += 1

        for code, logs in UV_LOGS.items():
            room_id = room_ids.get(code)
            if not room_id:
                continue
            for idx, log in enumerate(logs, start=1):
                start_time = log["start_time"]
                if _uv_log_exists(db, room_id, start_time, tenant_no):
                    uv_skipped += 1
                    continue
                clean_room_svc.create_uv_disinfect_log(
                    db,
                    clean_room_id=room_id,
                    user_id=log.get("user_id"),
                    operator_name=log.get("operator_name"),
                    start_time=start_time,
                    end_time=log.get("end_time"),
                    tenant_no=tenant_no,
                    order_num=idx,
                    remark=log.get("remark"),
                    user_id_param=user_id,
                )
                uv_created += 1

        print(
            f"洁净室测试数据初始化完成："
            f"洁净室 新增 {rooms_created} / 跳过 {rooms_skipped}；"
            f"进出记录 新增 {access_created} / 跳过 {access_skipped}；"
            f"消毒记录 新增 {uv_created} / 跳过 {uv_skipped}。"
        )
    finally:
        db.close()


if __name__ == "__main__":
    seed()