345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""洁净室及进出记录 CRUD（biz_clean_room + biz_clean_room_access_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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 洁净室
|
||
# ---------------------------------------------------------------------------
|
||
|
||
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,
|
||
"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,
|
||
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,
|
||
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,
|
||
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 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)))
|