#!/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()