zkxw/backend/app/crud/equipment.py

"""设备 CRUD（biz_equipment）"""
from __future__ import annotations

import math
from datetime import date, datetime
from decimal import Decimal
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 BizEquipment
_UNSET = object()


def _dec(v: Any) -> Optional[float]:
    if v is None:
        return None
    return float(v)


def _norm_status(v: Any) -> str:
    if v is None:
        return "1"
    if isinstance(v, bool):
        return "1" if v else "0"
    if isinstance(v, (int, float)):
        if int(v) == 0:
            return "0"
        if int(v) == 1:
            return "1"
    s = str(v).strip()
    if s in ("0", "1"):
        return s
    return "1"


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 _row_to_dict(r: BizEquipment) -> Dict[str, Any]:
    return {
        "id": r.id,
        "equipmentCode": r.equipment_code,
        "equipmentName": r.equipment_name,
        "equipmentModel": r.equipment_model or "",
        "manufacturer": r.manufacturer or "",
        "purchaseDate": r.purchase_date.strftime("%Y-%m-%d") if r.purchase_date else "",
        "purchaseAmount": _dec(r.purchase_amount),
        "location": r.location or "",
        "serialNo": r.serial_no or "",
        "status": _norm_status(r.status),
        "tenantNo": r.tenant_no,
        "orderNum": r.order_num if r.order_num is not None else 0,
        "remark": r.remark or "",
        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
    }


def list_equipments(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizEquipment).filter(BizEquipment.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizEquipment.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizEquipment.equipment_code.like(like),
                    BizEquipment.equipment_name.like(like),
                    BizEquipment.equipment_model.like(like),
                    BizEquipment.manufacturer.like(like),
                )
            )
        st = (status or "").strip()
        if st in ("0", "1"):
            q0 = q0.filter(BizEquipment.status == st)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizEquipment, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizEquipment.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizEquipment.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_equipment(db: Session, equipment_id: str) -> Optional[BizEquipment]:
    return (
        db.query(BizEquipment)
        .filter(and_(BizEquipment.id == equipment_id, BizEquipment.del_flag == 0))
        .first()
    )


def equipment_is_referenced(db: Session, equipment_id: str) -> bool:
    from app.models.models import BizProductionTask

    eid = (equipment_id or "").strip()
    if not eid:
        return False
    task_hit = (
        db.query(BizProductionTask.id)
        .filter(
            BizProductionTask.del_flag == 0,
            func.find_in_set(eid, BizProductionTask.equipment_ids) > 0,
        )
        .first()
    )
    return task_hit is not None


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


def create_equipment(
    db: Session,
    *,
    equipment_code: str,
    equipment_name: str,
    equipment_model: Optional[str] = None,
    manufacturer: Optional[str] = None,
    purchase_date: Optional[Any] = None,
    purchase_amount: Optional[float] = None,
    location: Optional[str] = None,
    serial_no: Optional[str] = None,
    status: Optional[str] = None,
    tenant_no: str = "zkxw",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizEquipment:
    ec = (equipment_code or "").strip()
    if not ec:
        raise ValueError("设备编码不能为空")
    if equipment_code_exists(db, ec, tenant_no):
        raise ValueError("设备编码已存在")
    obj = BizEquipment(
        id=generate_uuid(),
        equipment_code=ec,
        equipment_name=(equipment_name or "").strip() or ec,
        equipment_model=(equipment_model or "").strip() or None,
        manufacturer=(manufacturer or "").strip() or None,
        purchase_date=_parse_date(purchase_date),
        purchase_amount=Decimal(str(purchase_amount)) if purchase_amount is not None else None,
        location=(location or "").strip() or None,
        serial_no=(serial_no or "").strip() or None,
        status=_norm_status(status),
        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_equipment(
    db: Session,
    equipment_id: str,
    *,
    equipment_code: Any = _UNSET,
    equipment_name: Any = _UNSET,
    equipment_model: Any = _UNSET,
    manufacturer: Any = _UNSET,
    purchase_date: Any = _UNSET,
    purchase_amount: Any = _UNSET,
    location: Any = _UNSET,
    serial_no: Any = _UNSET,
    status: Any = _UNSET,
    order_num: Any = _UNSET,
    remark: Any = _UNSET,
    user_id: Optional[str] = None,
) -> BizEquipment:
    obj = get_equipment(db, equipment_id)
    if not obj:
        raise ValueError("设备不存在")
    if equipment_code is not _UNSET:
        new_code = equipment_code.strip()
        if not new_code:
            raise ValueError("设备编码不能为空")
        if new_code != obj.equipment_code and equipment_code_exists(
            db, new_code, obj.tenant_no, exclude_id=equipment_id
        ):
            raise ValueError("设备编码已存在")
        obj.equipment_code = new_code
    if equipment_name is not _UNSET:
        obj.equipment_name = (equipment_name or "").strip() or obj.equipment_name
    if equipment_model is not _UNSET:
        obj.equipment_model = (equipment_model or "").strip() or None
    if manufacturer is not _UNSET:
        obj.manufacturer = (manufacturer or "").strip() or None
    if purchase_date is not _UNSET:
        obj.purchase_date = _parse_date(purchase_date)
    if purchase_amount is not _UNSET:
        obj.purchase_amount = Decimal(str(purchase_amount)) if purchase_amount not in (None, "") else None
    if location is not _UNSET:
        obj.location = (location or "").strip() or None
    if serial_no is not _UNSET:
        obj.serial_no = (serial_no or "").strip() or None
    if status is not _UNSET:
        obj.status = _norm_status(status)
    if order_num is not _UNSET:
        obj.order_num = order_num
    if remark is not _UNSET:
        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_equipment(db: Session, equipment_id: str) -> None:
    obj = get_equipment(db, equipment_id)
    if not obj:
        raise ValueError("设备不存在")
    if equipment_is_referenced(db, equipment_id):
        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)))