"""洁净室及进出记录、紫外线消毒记录 CRUD（biz_clean_room + biz_clean_room_access_log + biz_clean_room_uv_disinfect_log）""" from __future__ import annotations import math from datetime import datetime from typing import Any, Dict, List, Optional, Tuple from sqlalchemy import and_, func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.models.models import BizCleanRoom, BizCleanRoomAccessLog, BizCleanRoomUvDisinfectLog # --------------------------------------------------------------------------- # 洁净室 # --------------------------------------------------------------------------- def _row_to_dict(c: BizCleanRoom) -> Dict[str, Any]: return { "id": c.id, "roomCode": c.room_code, "roomName": c.room_name, "roomGrade": c.room_grade, "roomStatus": c.room_status, "areaSqm": float(c.area_sqm) if c.area_sqm is not None else 0.0, "roomImage": c.room_image or "", "tenantNo": c.tenant_no, "orderNum": c.order_num if c.order_num is not None else 0, "remark": c.remark or "", "createTime": c.create_time.strftime("%Y-%m-%d %H:%M:%S") if c.create_time else "", "updateTime": c.update_time.strftime("%Y-%m-%d %H:%M:%S") if c.update_time else "", } def list_clean_rooms( db: Session, *, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(BizCleanRoom).filter(BizCleanRoom.del_flag == 0) if tenant_no: q0 = q0.filter(BizCleanRoom.tenant_no == tenant_no) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter(or_(BizCleanRoom.room_code.like(like), BizCleanRoom.room_name.like(like))) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCleanRoom.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizCleanRoom.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return [_row_to_dict(x) for x in rows], int(total) def get_clean_room(db: Session, room_id: str, tenant_no: Optional[str] = None) -> Optional[BizCleanRoom]: q = db.query(BizCleanRoom).filter(and_(BizCleanRoom.id == room_id, BizCleanRoom.del_flag == 0)) if tenant_no: q = q.filter(BizCleanRoom.tenant_no == tenant_no) return q.first() def room_code_exists(db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool: q = db.query(BizCleanRoom).filter( and_(BizCleanRoom.room_code == code.strip(), BizCleanRoom.tenant_no == tenant_no, BizCleanRoom.del_flag == 0) ) if exclude_id: q = q.filter(BizCleanRoom.id != exclude_id) return q.first() is not None def create_clean_room( db: Session, *, room_code: str, room_name: str, room_grade: Optional[str] = None, room_status: Optional[str] = None, area_sqm: Optional[Any] = None, room_image: Optional[str] = None, tenant_no: str = "zkxw", order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, ) -> BizCleanRoom: rc = (room_code or "").strip() if not rc: raise ValueError("洁净室编码不能为空") if room_code_exists(db, rc, tenant_no): raise ValueError("洁净室编码已存在") obj = BizCleanRoom( id=generate_uuid(), room_code=rc, room_name=(room_name or "").strip() or rc, room_grade=(room_grade or "").strip() or None, room_status=(room_status or "").strip() or None, area_sqm=area_sqm if area_sqm is not None else 0, room_image=(room_image or "").strip() or None, tenant_no=tenant_no, order_num=order_num if order_num is not None else 0, remark=(remark or "").strip() or None, create_by=user_id, create_dept_id="", create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_clean_room( db: Session, room_id: str, *, room_code: Optional[str] = None, room_name: Optional[str] = None, room_grade: Optional[str] = None, room_status: Optional[str] = None, area_sqm: Optional[Any] = None, room_image: Optional[str] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id: Optional[str] = None, tenant_no: Optional[str] = None, ) -> BizCleanRoom: obj = get_clean_room(db, room_id, tenant_no=tenant_no) if not obj: raise ValueError("洁净室不存在") if room_code is not None: new_code = room_code.strip() if not new_code: raise ValueError("洁净室编码不能为空") if new_code != obj.room_code and room_code_exists(db, new_code, obj.tenant_no, exclude_id=room_id): raise ValueError("洁净室编码已存在") obj.room_code = new_code if room_name is not None: obj.room_name = (room_name or "").strip() or obj.room_name if room_grade is not None: obj.room_grade = (room_grade or "").strip() or None if room_status is not None: obj.room_status = (room_status or "").strip() or None if area_sqm is not None: obj.area_sqm = area_sqm if room_image is not None: obj.room_image = (room_image or "").strip() or None if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = (remark or "").strip() or None obj.update_by = user_id obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_clean_room(db: Session, room_id: str, tenant_no: Optional[str] = None) -> None: obj = get_clean_room(db, room_id, tenant_no=tenant_no) if not obj: raise ValueError("洁净室不存在") obj.del_flag = 1 obj.update_time = datetime.now() db.add(obj) db.commit() # --------------------------------------------------------------------------- # 进出记录 # --------------------------------------------------------------------------- def _access_log_to_dict(log: BizCleanRoomAccessLog) -> Dict[str, Any]: return { "id": log.id, "cleanRoomId": log.clean_room_id, "userId": log.user_id, "personName": log.person_name, "deptName": log.dept_name, "enterTime": log.enter_time.strftime("%Y-%m-%d %H:%M:%S") if log.enter_time else "", "purpose": log.purpose or "", "leaveTime": log.leave_time.strftime("%Y-%m-%d %H:%M:%S") if log.leave_time else "", "bodyTemp": log.body_temp or "", "tenantNo": log.tenant_no, "orderNum": log.order_num if log.order_num is not None else 0, "remark": log.remark or "", "createTime": log.create_time.strftime("%Y-%m-%d %H:%M:%S") if log.create_time else "", "updateTime": log.update_time.strftime("%Y-%m-%d %H:%M:%S") if log.update_time else "", } def list_access_logs( db: Session, *, page: int = 1, page_size: int = 10, clean_room_id: Optional[str] = None, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(BizCleanRoomAccessLog).filter(BizCleanRoomAccessLog.del_flag == 0) if clean_room_id: q0 = q0.filter(BizCleanRoomAccessLog.clean_room_id == clean_room_id) if tenant_no: q0 = q0.filter(BizCleanRoomAccessLog.tenant_no == tenant_no) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter(or_( BizCleanRoomAccessLog.person_name.like(like), BizCleanRoomAccessLog.dept_name.like(like), )) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCleanRoomAccessLog.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizCleanRoomAccessLog.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return [_access_log_to_dict(x) for x in rows], int(total) def get_access_log(db: Session, log_id: str, tenant_no: Optional[str] = None) -> Optional[BizCleanRoomAccessLog]: q = db.query(BizCleanRoomAccessLog).filter(and_(BizCleanRoomAccessLog.id == log_id, BizCleanRoomAccessLog.del_flag == 0)) if tenant_no: q = q.filter(BizCleanRoomAccessLog.tenant_no == tenant_no) return q.first() def create_access_log( db: Session, *, clean_room_id: str, user_id: Optional[str] = None, person_name: Optional[str] = None, dept_name: Optional[str] = None, enter_time: Optional[datetime] = None, purpose: Optional[str] = None, leave_time: Optional[datetime] = None, body_temp: Optional[str] = None, tenant_no: str = "zkxw", order_num: Optional[int] = None, remark: Optional[str] = None, user_id_param: Optional[str] = None, ) -> BizCleanRoomAccessLog: cr_id = (clean_room_id or "").strip() if not cr_id: raise ValueError("洁净室ID不能为空") obj = BizCleanRoomAccessLog( id=generate_uuid(), clean_room_id=cr_id, user_id=(user_id or "").strip() or None, person_name=(person_name or "").strip() or None, dept_name=(dept_name or "").strip() or None, enter_time=enter_time or datetime.now(), purpose=(purpose or "").strip() or None, leave_time=leave_time, body_temp=(body_temp or "").strip() or None, tenant_no=tenant_no, order_num=order_num if order_num is not None else 0, remark=(remark or "").strip() or None, create_by=user_id_param, create_dept_id="", create_time=datetime.now(), update_by=user_id_param, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_access_log( db: Session, log_id: str, *, person_name: Optional[str] = None, dept_name: Optional[str] = None, enter_time: Optional[datetime] = None, purpose: Optional[str] = None, leave_time: Optional[datetime] = None, body_temp: Optional[str] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id_param: Optional[str] = None, tenant_no: Optional[str] = None, ) -> BizCleanRoomAccessLog: obj = get_access_log(db, log_id, tenant_no=tenant_no) if not obj: raise ValueError("进出记录不存在") if person_name is not None: obj.person_name = (person_name or "").strip() or None if dept_name is not None: obj.dept_name = (dept_name or "").strip() or None if enter_time is not None: obj.enter_time = enter_time if purpose is not None: obj.purpose = (purpose or "").strip() or None if leave_time is not None: obj.leave_time = leave_time if body_temp is not None: obj.body_temp = (body_temp or "").strip() or None if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = (remark or "").strip() or None obj.update_by = user_id_param obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_access_log(db: Session, log_id: str, tenant_no: Optional[str] = None) -> None: obj = get_access_log(db, log_id, tenant_no=tenant_no) if not obj: raise ValueError("进出记录不存在") obj.del_flag = 1 obj.update_time = datetime.now() db.add(obj) db.commit() def total_pages(total: int, page_size: int) -> int: if total <= 0: return 0 return int(math.ceil(total / float(page_size))) # --------------------------------------------------------------------------- # 紫外线消毒记录 # --------------------------------------------------------------------------- def _calc_duration_minutes(start_time: Optional[datetime], end_time: Optional[datetime]) -> Optional[int]: if not start_time or not end_time: return None delta = end_time - start_time minutes = int(delta.total_seconds() // 60) return max(minutes, 0) def _uv_disinfect_log_to_dict(log: BizCleanRoomUvDisinfectLog) -> Dict[str, Any]: duration = log.duration_minutes if duration is None: duration = _calc_duration_minutes(log.start_time, log.end_time) return { "id": log.id, "cleanRoomId": log.clean_room_id, "userId": log.user_id, "operatorName": log.operator_name, "startTime": log.start_time.strftime("%Y-%m-%d %H:%M:%S") if log.start_time else "", "endTime": log.end_time.strftime("%Y-%m-%d %H:%M:%S") if log.end_time else "", "durationMinutes": duration if duration is not None else None, "tenantNo": log.tenant_no, "orderNum": log.order_num if log.order_num is not None else 0, "remark": log.remark or "", "createTime": log.create_time.strftime("%Y-%m-%d %H:%M:%S") if log.create_time else "", "updateTime": log.update_time.strftime("%Y-%m-%d %H:%M:%S") if log.update_time else "", } def list_uv_disinfect_logs( db: Session, *, page: int = 1, page_size: int = 10, clean_room_id: Optional[str] = None, keyword: Optional[str] = None, tenant_no: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: def _filtered(): q0 = db.query(BizCleanRoomUvDisinfectLog).filter(BizCleanRoomUvDisinfectLog.del_flag == 0) if clean_room_id: q0 = q0.filter(BizCleanRoomUvDisinfectLog.clean_room_id == clean_room_id) if tenant_no: q0 = q0.filter(BizCleanRoomUvDisinfectLog.tenant_no == tenant_no) kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter(BizCleanRoomUvDisinfectLog.operator_name.like(like)) return q0 total = _filtered().order_by(None).with_entities(func.count(BizCleanRoomUvDisinfectLog.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizCleanRoomUvDisinfectLog.create_time.desc()) .offset((page - 1) * page_size) .limit(page_size) .all() ) return [_uv_disinfect_log_to_dict(x) for x in rows], int(total) def get_uv_disinfect_log( db: Session, log_id: str, tenant_no: Optional[str] = None ) -> Optional[BizCleanRoomUvDisinfectLog]: q = db.query(BizCleanRoomUvDisinfectLog).filter( and_(BizCleanRoomUvDisinfectLog.id == log_id, BizCleanRoomUvDisinfectLog.del_flag == 0) ) if tenant_no: q = q.filter(BizCleanRoomUvDisinfectLog.tenant_no == tenant_no) return q.first() def create_uv_disinfect_log( db: Session, *, clean_room_id: str, user_id: Optional[str] = None, operator_name: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, duration_minutes: Optional[int] = None, tenant_no: str = "zkxw", order_num: Optional[int] = None, remark: Optional[str] = None, user_id_param: Optional[str] = None, ) -> BizCleanRoomUvDisinfectLog: cr_id = (clean_room_id or "").strip() if not cr_id: raise ValueError("洁净室ID不能为空") op_name = (operator_name or "").strip() if not op_name: raise ValueError("操作人不能为空") st = start_time or datetime.now() if end_time and end_time < st: raise ValueError("结束时间不能早于开始时间") dur = duration_minutes if dur is None: dur = _calc_duration_minutes(st, end_time) obj = BizCleanRoomUvDisinfectLog( id=generate_uuid(), clean_room_id=cr_id, user_id=(user_id or "").strip() or None, operator_name=op_name, start_time=st, end_time=end_time, duration_minutes=dur, tenant_no=tenant_no, order_num=order_num if order_num is not None else 0, remark=(remark or "").strip() or None, create_by=user_id_param, create_dept_id="", create_time=datetime.now(), update_by=user_id_param, update_time=datetime.now(), del_flag=0, ) db.add(obj) db.commit() db.refresh(obj) return obj def update_uv_disinfect_log( db: Session, log_id: str, *, user_id: Optional[str] = None, operator_name: Optional[str] = None, start_time: Optional[datetime] = None, end_time: Optional[datetime] = None, duration_minutes: Optional[int] = None, order_num: Optional[int] = None, remark: Optional[str] = None, user_id_param: Optional[str] = None, tenant_no: Optional[str] = None, clear_end_time: bool = False, ) -> BizCleanRoomUvDisinfectLog: obj = get_uv_disinfect_log(db, log_id, tenant_no=tenant_no) if not obj: raise ValueError("紫外线消毒记录不存在") if operator_name is not None: op_name = (operator_name or "").strip() if not op_name: raise ValueError("操作人不能为空") obj.operator_name = op_name if user_id is not None: obj.user_id = (user_id or "").strip() or None if start_time is not None: obj.start_time = start_time if clear_end_time: obj.end_time = None elif end_time is not None: obj.end_time = end_time st = obj.start_time et = obj.end_time if et and st and et < st: raise ValueError("结束时间不能早于开始时间") if duration_minutes is not None: obj.duration_minutes = duration_minutes else: obj.duration_minutes = _calc_duration_minutes(st, et) if order_num is not None: obj.order_num = order_num if remark is not None: obj.remark = (remark or "").strip() or None obj.update_by = user_id_param obj.update_time = datetime.now() db.add(obj) db.commit() db.refresh(obj) return obj def soft_delete_uv_disinfect_log(db: Session, log_id: str, tenant_no: Optional[str] = None) -> None: obj = get_uv_disinfect_log(db, log_id, tenant_no=tenant_no) if not obj: raise ValueError("紫外线消毒记录不存在") obj.del_flag = 1 obj.update_time = datetime.now() db.add(obj) db.commit()