zkxw/backend/app/crud/cleanroom_garment.py

"""洁净服及清洗/灭菌记录 CRUD"""
from __future__ import annotations

import math
from datetime import date, 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.cleanroom_garment import (
    BizCleanroomGarment,
    BizCleanroomGarmentSterilizeLog,
    BizCleanroomGarmentWashLog,
)

GARMENT_STATUS_NORMAL = "zc"
GARMENT_STATUS_DISCARDED = "yfq"


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / max(1, page_size)))


def _parse_date(v: Any) -> Optional[date]:
    if v is None or v == "":
        return None
    if isinstance(v, date) and not isinstance(v, datetime):
        return v
    if isinstance(v, datetime):
        return v.date()
    s = str(v).strip()
    if not s:
        return None
    for fmt in ("%Y-%m-%d", "%Y/%m/%d"):
        try:
            return datetime.strptime(s[:10], fmt).date()
        except ValueError:
            continue
    raise ValueError("日期格式无效，请使用 YYYY-MM-DD")


def _parse_datetime(v: Any) -> datetime:
    if isinstance(v, datetime):
        return v
    s = str(v or "").strip()
    if not s:
        raise ValueError("时间不能为空")
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            if fmt == "%Y-%m-%d":
                return datetime.strptime(s[:10], fmt)
            return datetime.strptime(s[:19] if len(s) >= 19 else s, fmt)
        except ValueError:
            continue
    raise ValueError("时间格式无效，请使用 YYYY-MM-DD HH:mm:ss")


def _parse_limit(v: Any, label: str) -> int:
    if v is None or v == "":
        raise ValueError(f"{label}不能为空")
    try:
        n = int(v)
    except (TypeError, ValueError):
        raise ValueError(f"{label}必须为整数")
    if n < 1:
        raise ValueError(f"{label}必须大于等于 1")
    return n


def _count_map(db: Session, model, garment_ids: List[str], tenant_no: Optional[str]) -> Dict[str, int]:
    if not garment_ids:
        return {}
    q = (
        db.query(model.garment_id, func.count(model.id))
        .filter(model.del_flag == 0, model.garment_id.in_(garment_ids))
    )
    if tenant_no:
        q = q.filter(model.tenant_no == tenant_no)
    return {gid: int(cnt) for gid, cnt in q.group_by(model.garment_id).all()}


def _garment_to_dict(
    g: BizCleanroomGarment,
    *,
    wash_count: int = 0,
    sterilize_count: int = 0,
) -> Dict[str, Any]:
    wash_limit_reached = wash_count >= int(g.wash_discard_limit or 0)
    sterilize_limit_reached = sterilize_count >= int(g.sterilize_discard_limit or 0)
    return {
        "id": g.id,
        "garmentCode": g.garment_code,
        "garmentName": g.garment_name,
        "washDiscardLimit": int(g.wash_discard_limit or 0),
        "sterilizeDiscardLimit": int(g.sterilize_discard_limit or 0),
        "inboundDate": g.inbound_date.strftime("%Y-%m-%d") if g.inbound_date else "",
        "discardDate": g.discard_date.strftime("%Y-%m-%d") if g.discard_date else "",
        "garmentStatus": g.garment_status or GARMENT_STATUS_NORMAL,
        "washCount": wash_count,
        "sterilizeCount": sterilize_count,
        "washLimitReached": wash_limit_reached,
        "sterilizeLimitReached": sterilize_limit_reached,
        "needAttention": wash_limit_reached or sterilize_limit_reached,
        "tenantNo": g.tenant_no,
        "orderNum": g.order_num if g.order_num is not None else 0,
        "remark": g.remark or "",
        "createTime": g.create_time.strftime("%Y-%m-%d %H:%M:%S") if g.create_time else "",
        "updateTime": g.update_time.strftime("%Y-%m-%d %H:%M:%S") if g.update_time else "",
    }


def list_garments(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    garment_status: Optional[str] = None,
    need_attention: Optional[bool] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    wash_cnt_sq = (
        db.query(
            BizCleanroomGarmentWashLog.garment_id.label("garment_id"),
            func.count(BizCleanroomGarmentWashLog.id).label("wash_count"),
        )
        .filter(BizCleanroomGarmentWashLog.del_flag == 0)
        .group_by(BizCleanroomGarmentWashLog.garment_id)
        .subquery()
    )
    ster_cnt_sq = (
        db.query(
            BizCleanroomGarmentSterilizeLog.garment_id.label("garment_id"),
            func.count(BizCleanroomGarmentSterilizeLog.id).label("sterilize_count"),
        )
        .filter(BizCleanroomGarmentSterilizeLog.del_flag == 0)
        .group_by(BizCleanroomGarmentSterilizeLog.garment_id)
        .subquery()
    )

    def _filtered():
        q0 = (
            db.query(
                BizCleanroomGarment,
                func.coalesce(wash_cnt_sq.c.wash_count, 0).label("wash_count"),
                func.coalesce(ster_cnt_sq.c.sterilize_count, 0).label("sterilize_count"),
            )
            .outerjoin(wash_cnt_sq, BizCleanroomGarment.id == wash_cnt_sq.c.garment_id)
            .outerjoin(ster_cnt_sq, BizCleanroomGarment.id == ster_cnt_sq.c.garment_id)
            .filter(BizCleanroomGarment.del_flag == 0)
        )
        if tenant_no:
            q0 = q0.filter(BizCleanroomGarment.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCleanroomGarment.garment_code.like(like),
                    BizCleanroomGarment.garment_name.like(like),
                )
            )
        st = (garment_status or "").strip()
        if st in (GARMENT_STATUS_NORMAL, GARMENT_STATUS_DISCARDED):
            q0 = q0.filter(BizCleanroomGarment.garment_status == st)
        if need_attention:
            q0 = q0.filter(
                BizCleanroomGarment.garment_status == GARMENT_STATUS_NORMAL,
                or_(
                    func.coalesce(wash_cnt_sq.c.wash_count, 0) >= BizCleanroomGarment.wash_discard_limit,
                    func.coalesce(ster_cnt_sq.c.sterilize_count, 0) >= BizCleanroomGarment.sterilize_discard_limit,
                ),
            )
        if user_info:
            from app.core.data_scope import apply_data_scope

            q0 = apply_data_scope(q0, BizCleanroomGarment, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(
            BizCleanroomGarment.create_time.desc(),
            BizCleanroomGarment.garment_code.asc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    result = []
    for g, wash_count, sterilize_count in rows:
        result.append(
            _garment_to_dict(
                g,
                wash_count=int(wash_count or 0),
                sterilize_count=int(sterilize_count or 0),
            )
        )
    return result, int(total)


def get_garment(
    db: Session, garment_id: str, tenant_no: Optional[str] = None
) -> Optional[BizCleanroomGarment]:
    q = db.query(BizCleanroomGarment).filter(
        and_(BizCleanroomGarment.id == garment_id, BizCleanroomGarment.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCleanroomGarment.tenant_no == tenant_no)
    return q.first()


def get_garment_detail(
    db: Session, garment_id: str, tenant_no: Optional[str] = None
) -> Optional[Dict[str, Any]]:
    obj = get_garment(db, garment_id, tenant_no=tenant_no)
    if not obj:
        return None
    wash_map = _count_map(db, BizCleanroomGarmentWashLog, [garment_id], tenant_no)
    ster_map = _count_map(db, BizCleanroomGarmentSterilizeLog, [garment_id], tenant_no)
    return _garment_to_dict(
        obj,
        wash_count=wash_map.get(garment_id, 0),
        sterilize_count=ster_map.get(garment_id, 0),
    )


def garment_code_exists(
    db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None
) -> bool:
    q = db.query(BizCleanroomGarment).filter(
        and_(
            BizCleanroomGarment.garment_code == code.strip(),
            BizCleanroomGarment.tenant_no == tenant_no,
            BizCleanroomGarment.del_flag == 0,
        )
    )
    if exclude_id:
        q = q.filter(BizCleanroomGarment.id != exclude_id)
    return q.first() is not None


def create_garment(
    db: Session,
    *,
    garment_code: str,
    garment_name: str,
    wash_discard_limit: Any,
    sterilize_discard_limit: Any,
    inbound_date: Any,
    tenant_no: str = "zkxw",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizCleanroomGarment:
    code = (garment_code or "").strip()
    if not code:
        raise ValueError("编号不能为空")
    if garment_code_exists(db, code, tenant_no):
        raise ValueError("编号已存在")
    name = (garment_name or "").strip()
    if not name:
        raise ValueError("名称不能为空")
    inbound = _parse_date(inbound_date)
    if not inbound:
        raise ValueError("入库日期不能为空")
    obj = BizCleanroomGarment(
        id=generate_uuid(),
        garment_code=code,
        garment_name=name,
        wash_discard_limit=_parse_limit(wash_discard_limit, "清洗报废次数"),
        sterilize_discard_limit=_parse_limit(sterilize_discard_limit, "灭菌报废次数"),
        inbound_date=inbound,
        discard_date=None,
        garment_status=GARMENT_STATUS_NORMAL,
        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_garment(
    db: Session,
    garment_id: str,
    *,
    garment_code: Optional[str] = None,
    garment_name: Optional[str] = None,
    wash_discard_limit: Any = None,
    sterilize_discard_limit: Any = None,
    inbound_date: Any = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> BizCleanroomGarment:
    obj = get_garment(db, garment_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("洁净服不存在")
    if garment_code is not None:
        new_code = garment_code.strip()
        if not new_code:
            raise ValueError("编号不能为空")
        if new_code != obj.garment_code and garment_code_exists(db, new_code, obj.tenant_no, exclude_id=garment_id):
            raise ValueError("编号已存在")
        obj.garment_code = new_code
    if garment_name is not None:
        name = (garment_name or "").strip()
        if not name:
            raise ValueError("名称不能为空")
        obj.garment_name = name
    if wash_discard_limit is not None:
        obj.wash_discard_limit = _parse_limit(wash_discard_limit, "清洗报废次数")
    if sterilize_discard_limit is not None:
        obj.sterilize_discard_limit = _parse_limit(sterilize_discard_limit, "灭菌报废次数")
    if inbound_date is not None:
        inbound = _parse_date(inbound_date)
        if not inbound:
            raise ValueError("入库日期不能为空")
        obj.inbound_date = inbound
    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_garment(db: Session, garment_id: str, tenant_no: Optional[str] = None) -> None:
    obj = get_garment(db, garment_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 batch_soft_delete_garments(
    db: Session, garment_ids: List[str], tenant_no: Optional[str] = None
) -> int:
    count = 0
    for gid in garment_ids:
        try:
            soft_delete_garment(db, gid, tenant_no=tenant_no)
            count += 1
        except ValueError:
            continue
    return count


def discard_garment(
    db: Session,
    garment_id: str,
    *,
    discard_date: Any = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> BizCleanroomGarment:
    obj = get_garment(db, garment_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("洁净服不存在")
    if obj.garment_status == GARMENT_STATUS_DISCARDED:
        raise ValueError("该洁净服已废弃")
    d = _parse_date(discard_date) if discard_date else date.today()
    obj.garment_status = GARMENT_STATUS_DISCARDED
    obj.discard_date = d
    if remark is not None:
        obj.remark = (remark or "").strip() or obj.remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def batch_discard_garments(
    db: Session,
    garment_ids: List[str],
    *,
    discard_date: Any = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> Dict[str, Any]:
    success_count = 0
    skipped_items: List[Dict[str, str]] = []
    for gid in garment_ids:
        obj = get_garment(db, gid, tenant_no=tenant_no)
        if not obj:
            skipped_items.append({"garmentId": gid, "reason": "不存在"})
            continue
        if obj.garment_status == GARMENT_STATUS_DISCARDED:
            skipped_items.append({"garmentId": gid, "garmentCode": obj.garment_code, "reason": "已废弃"})
            continue
        discard_garment(
            db,
            gid,
            discard_date=discard_date,
            remark=remark,
            user_id=user_id,
            tenant_no=tenant_no,
        )
        success_count += 1
    return {
        "successCount": success_count,
        "skippedCount": len(skipped_items),
        "skippedItems": skipped_items,
    }


def _ensure_garment_active(g: Optional[BizCleanroomGarment]) -> BizCleanroomGarment:
    if not g:
        raise ValueError("洁净服不存在")
    if g.garment_status == GARMENT_STATUS_DISCARDED:
        raise ValueError(f"洁净服「{g.garment_code}」已废弃，不可登记")
    return g


# ---------------------------------------------------------------------------
# 清洗记录
# ---------------------------------------------------------------------------


def _wash_log_to_dict(log: BizCleanroomGarmentWashLog) -> Dict[str, Any]:
    return {
        "id": log.id,
        "garmentId": log.garment_id,
        "washTime": log.wash_time.strftime("%Y-%m-%d %H:%M:%S") if log.wash_time else "",
        "operatorUserId": log.operator_user_id or "",
        "operatorName": log.operator_name 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_wash_logs(
    db: Session,
    *,
    garment_id: str,
    page: int = 1,
    page_size: int = 10,
    tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(BizCleanroomGarmentWashLog).filter(
        BizCleanroomGarmentWashLog.del_flag == 0,
        BizCleanroomGarmentWashLog.garment_id == garment_id,
    )
    if tenant_no:
        q = q.filter(BizCleanroomGarmentWashLog.tenant_no == tenant_no)
    total = q.count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizCleanroomGarmentWashLog.wash_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_wash_log_to_dict(x) for x in rows], int(total)


def get_wash_log(db: Session, log_id: str, tenant_no: Optional[str] = None) -> Optional[BizCleanroomGarmentWashLog]:
    q = db.query(BizCleanroomGarmentWashLog).filter(
        and_(BizCleanroomGarmentWashLog.id == log_id, BizCleanroomGarmentWashLog.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCleanroomGarmentWashLog.tenant_no == tenant_no)
    return q.first()


def create_wash_log(
    db: Session,
    *,
    garment_id: str,
    wash_time: Any,
    operator_user_id: Optional[str] = None,
    operator_name: Optional[str] = None,
    tenant_no: str = "zkxw",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizCleanroomGarmentWashLog:
    garment = _ensure_garment_active(get_garment(db, garment_id, tenant_no=tenant_no))
    wt = _parse_datetime(wash_time)
    obj = BizCleanroomGarmentWashLog(
        id=generate_uuid(),
        garment_id=garment.id,
        wash_time=wt,
        operator_user_id=(operator_user_id or "").strip() or None,
        operator_name=(operator_name 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_wash_log(
    db: Session,
    log_id: str,
    *,
    wash_time: Any = None,
    operator_user_id: Optional[str] = None,
    operator_name: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> BizCleanroomGarmentWashLog:
    obj = get_wash_log(db, log_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("清洗记录不存在")
    garment = get_garment(db, obj.garment_id, tenant_no=tenant_no)
    _ensure_garment_active(garment)
    if wash_time is not None:
        obj.wash_time = _parse_datetime(wash_time)
    if operator_user_id is not None:
        obj.operator_user_id = (operator_user_id or "").strip() or None
    if operator_name is not None:
        obj.operator_name = (operator_name 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_wash_log(db: Session, log_id: str, tenant_no: Optional[str] = None) -> None:
    obj = get_wash_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 batch_create_wash_logs(
    db: Session,
    *,
    garment_ids: List[str],
    wash_time: Any,
    operator_user_id: Optional[str] = None,
    operator_name: Optional[str] = None,
    remark: Optional[str] = None,
    tenant_no: str = "zkxw",
    user_id: Optional[str] = None,
) -> Dict[str, Any]:
    if not garment_ids:
        raise ValueError("请选择洁净服")
    discarded_codes: List[str] = []
    garments: List[BizCleanroomGarment] = []
    for gid in garment_ids:
        g = get_garment(db, gid, tenant_no=tenant_no)
        if not g:
            raise ValueError("存在无效的洁净服")
        if g.garment_status == GARMENT_STATUS_DISCARDED:
            discarded_codes.append(g.garment_code)
        else:
            garments.append(g)
    if discarded_codes:
        raise ValueError(f"以下洁净服已废弃，不可登记：{', '.join(discarded_codes)}")
    wt = _parse_datetime(wash_time)
    items: List[Dict[str, Any]] = []
    try:
        for g in garments:
            log = BizCleanroomGarmentWashLog(
                id=generate_uuid(),
                garment_id=g.id,
                wash_time=wt,
                operator_user_id=(operator_user_id or "").strip() or None,
                operator_name=(operator_name or "").strip() or None,
                tenant_no=tenant_no,
                order_num=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(log)
        db.commit()
    except Exception:
        db.rollback()
        raise
    wash_map = _count_map(db, BizCleanroomGarmentWashLog, [g.id for g in garments], tenant_no)
    for g in garments:
        wc = wash_map.get(g.id, 0)
        items.append(
            {
                "garmentId": g.id,
                "garmentCode": g.garment_code,
                "washCount": wc,
                "washLimitReached": wc >= int(g.wash_discard_limit or 0),
                "sterilizeCount": _count_map(db, BizCleanroomGarmentSterilizeLog, [g.id], tenant_no).get(g.id, 0),
                "sterilizeLimitReached": False,
            }
        )
    for item in items:
        g = next(x for x in garments if x.id == item["garmentId"])
        sc = item["sterilizeCount"]
        item["sterilizeLimitReached"] = sc >= int(g.sterilize_discard_limit or 0)
        item["needAttention"] = item["washLimitReached"] or item["sterilizeLimitReached"]
    return {"successCount": len(items), "items": items}


# ---------------------------------------------------------------------------
# 灭菌记录
# ---------------------------------------------------------------------------


def _sterilize_log_to_dict(log: BizCleanroomGarmentSterilizeLog) -> Dict[str, Any]:
    return {
        "id": log.id,
        "garmentId": log.garment_id,
        "sterilizeTime": log.sterilize_time.strftime("%Y-%m-%d %H:%M:%S") if log.sterilize_time else "",
        "operatorUserId": log.operator_user_id or "",
        "operatorName": log.operator_name 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_sterilize_logs(
    db: Session,
    *,
    garment_id: str,
    page: int = 1,
    page_size: int = 10,
    tenant_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(BizCleanroomGarmentSterilizeLog).filter(
        BizCleanroomGarmentSterilizeLog.del_flag == 0,
        BizCleanroomGarmentSterilizeLog.garment_id == garment_id,
    )
    if tenant_no:
        q = q.filter(BizCleanroomGarmentSterilizeLog.tenant_no == tenant_no)
    total = q.count()
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizCleanroomGarmentSterilizeLog.sterilize_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_sterilize_log_to_dict(x) for x in rows], int(total)


def get_sterilize_log(
    db: Session, log_id: str, tenant_no: Optional[str] = None
) -> Optional[BizCleanroomGarmentSterilizeLog]:
    q = db.query(BizCleanroomGarmentSterilizeLog).filter(
        and_(BizCleanroomGarmentSterilizeLog.id == log_id, BizCleanroomGarmentSterilizeLog.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCleanroomGarmentSterilizeLog.tenant_no == tenant_no)
    return q.first()


def create_sterilize_log(
    db: Session,
    *,
    garment_id: str,
    sterilize_time: Any,
    operator_user_id: Optional[str] = None,
    operator_name: Optional[str] = None,
    tenant_no: str = "zkxw",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizCleanroomGarmentSterilizeLog:
    garment = _ensure_garment_active(get_garment(db, garment_id, tenant_no=tenant_no))
    st = _parse_datetime(sterilize_time)
    obj = BizCleanroomGarmentSterilizeLog(
        id=generate_uuid(),
        garment_id=garment.id,
        sterilize_time=st,
        operator_user_id=(operator_user_id or "").strip() or None,
        operator_name=(operator_name 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_sterilize_log(
    db: Session,
    log_id: str,
    *,
    sterilize_time: Any = None,
    operator_user_id: Optional[str] = None,
    operator_name: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> BizCleanroomGarmentSterilizeLog:
    obj = get_sterilize_log(db, log_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("灭菌记录不存在")
    garment = get_garment(db, obj.garment_id, tenant_no=tenant_no)
    _ensure_garment_active(garment)
    if sterilize_time is not None:
        obj.sterilize_time = _parse_datetime(sterilize_time)
    if operator_user_id is not None:
        obj.operator_user_id = (operator_user_id or "").strip() or None
    if operator_name is not None:
        obj.operator_name = (operator_name 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_sterilize_log(db: Session, log_id: str, tenant_no: Optional[str] = None) -> None:
    obj = get_sterilize_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 batch_create_sterilize_logs(
    db: Session,
    *,
    garment_ids: List[str],
    sterilize_time: Any,
    operator_user_id: Optional[str] = None,
    operator_name: Optional[str] = None,
    remark: Optional[str] = None,
    tenant_no: str = "zkxw",
    user_id: Optional[str] = None,
) -> Dict[str, Any]:
    if not garment_ids:
        raise ValueError("请选择洁净服")
    discarded_codes: List[str] = []
    garments: List[BizCleanroomGarment] = []
    for gid in garment_ids:
        g = get_garment(db, gid, tenant_no=tenant_no)
        if not g:
            raise ValueError("存在无效的洁净服")
        if g.garment_status == GARMENT_STATUS_DISCARDED:
            discarded_codes.append(g.garment_code)
        else:
            garments.append(g)
    if discarded_codes:
        raise ValueError(f"以下洁净服已废弃，不可登记：{', '.join(discarded_codes)}")
    st = _parse_datetime(sterilize_time)
    items: List[Dict[str, Any]] = []
    try:
        for g in garments:
            log = BizCleanroomGarmentSterilizeLog(
                id=generate_uuid(),
                garment_id=g.id,
                sterilize_time=st,
                operator_user_id=(operator_user_id or "").strip() or None,
                operator_name=(operator_name or "").strip() or None,
                tenant_no=tenant_no,
                order_num=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(log)
        db.commit()
    except Exception:
        db.rollback()
        raise
    ster_map = _count_map(db, BizCleanroomGarmentSterilizeLog, [g.id for g in garments], tenant_no)
    wash_map = _count_map(db, BizCleanroomGarmentWashLog, [g.id for g in garments], tenant_no)
    for g in garments:
        sc = ster_map.get(g.id, 0)
        wc = wash_map.get(g.id, 0)
        items.append(
            {
                "garmentId": g.id,
                "garmentCode": g.garment_code,
                "washCount": wc,
                "washLimitReached": wc >= int(g.wash_discard_limit or 0),
                "sterilizeCount": sc,
                "sterilizeLimitReached": sc >= int(g.sterilize_discard_limit or 0),
                "needAttention": False,
            }
        )
    for item in items:
        item["needAttention"] = item["washLimitReached"] or item["sterilizeLimitReached"]
    return {"successCount": len(items), "items": items}