zkxw/backend/app/crud/clean_room.py

"""洁净室及进出记录、紫外线消毒记录 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()