zkxw/backend/app/crud/consumable_pack.py

"""耗材包 CRUD（biz_consumable_pack + biz_consumable_pack_item）"""
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 BizConsumablePack, BizConsumablePackItem


def _item_row_to_dict(item: BizConsumablePackItem) -> Dict[str, Any]:
    return {
        "id": item.id,
        "packId": item.pack_id,
        "consumableCode": item.consumable_code,
        "consumableName": item.consumable_name,
        "specModel": item.spec_model or "",
        "standardUsage": item.standard_usage or "",
        "unit": item.unit or "",
        "remark": item.remark or "",
    }


def _row_to_dict(p: BizConsumablePack) -> Dict[str, Any]:
    return {
        "id": p.id,
        "packCode": p.pack_code,
        "packName": p.pack_name,
        "usageScene": p.usage_scene or "",
        "packType": p.pack_type or "",
        "status": int(p.status) if p.status not in (None, "") else 0,
        "tenantNo": p.tenant_no,
        "orderNum": p.order_num if p.order_num is not None else 0,
        "remark": p.remark or "",
        "createTime": p.create_time.strftime("%Y-%m-%d %H:%M:%S") if p.create_time else "",
        "updateTime": p.update_time.strftime("%Y-%m-%d %H:%M:%S") if p.update_time else "",
    }


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

    total = _filtered().order_by(None).with_entities(func.count(BizConsumablePack.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizConsumablePack.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_row_to_dict(x) for x in rows], int(total)


def list_all_packs(db: Session, tenant_no: str, user_info=None) -> List[Dict[str, Any]]:
    q0 = db.query(BizConsumablePack).filter(and_(BizConsumablePack.tenant_no == tenant_no, BizConsumablePack.del_flag == 0))
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, BizConsumablePack, user_info, db, tenant_no)
    rows = q0.order_by(BizConsumablePack.create_time.desc()).all()
    return [_row_to_dict(x) for x in rows]


def get_consumable_pack(db: Session, pack_id: str) -> Optional[BizConsumablePack]:
    return (
        db.query(BizConsumablePack)
        .filter(and_(BizConsumablePack.id == pack_id, BizConsumablePack.del_flag == 0))
        .first()
    )


def get_pack_detail(db: Session, pack_id: str) -> Optional[Dict[str, Any]]:
    pack = get_consumable_pack(db, pack_id)
    if not pack:
        return None
    result = _row_to_dict(pack)
    items = (
        db.query(BizConsumablePackItem)
        .filter(and_(BizConsumablePackItem.pack_id == pack_id, BizConsumablePackItem.del_flag == 0))
        .order_by(BizConsumablePackItem.create_time.asc())
        .all()
    )
    result["items"] = [_item_row_to_dict(i) for i in items]
    return result


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


def create_consumable_pack(
    db: Session,
    *,
    pack_code: str,
    pack_name: str,
    usage_scene: Optional[str] = None,
    pack_type: Optional[str] = None,
    status: Optional[int] = None,
    items: Optional[List[Dict[str, Any]]] = None,
    tenant_no: str = "zkxw",
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizConsumablePack:
    pc = (pack_code or "").strip()
    if not pc:
        raise ValueError("耗材包编码不能为空")
    if pack_code_exists(db, pc, tenant_no):
        raise ValueError("耗材包编码已存在")
    now = datetime.now()
    obj = BizConsumablePack(
        id=generate_uuid(),
        pack_code=pc,
        pack_name=(pack_name or "").strip() or pc,
        usage_scene=(usage_scene or "").strip() or None,
        pack_type=(pack_type or "").strip() or None,
        status=str(status) if status is not None else "1",
        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=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.flush()

    if items:
        for it in items:
            item_obj = BizConsumablePackItem(
                id=generate_uuid(),
                pack_id=obj.id,
                consumable_code=(it.get("consumable_code") or "").strip(),
                consumable_name=(it.get("consumable_name") or "").strip(),
                spec_model=(it.get("spec_model") or "").strip() or None,
                standard_usage=(it.get("standard_usage") or "").strip() or None,
                unit=(it.get("unit") or "").strip() or None,
                remark=(it.get("remark") or "").strip() or None,
                tenant_no=tenant_no,
                order_num=0,
                create_by=user_id,
                create_dept_id="",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
            db.add(item_obj)

    db.commit()
    db.refresh(obj)
    return obj


def update_consumable_pack(
    db: Session,
    pack_id: str,
    *,
    pack_code: Optional[str] = None,
    pack_name: Optional[str] = None,
    usage_scene: Optional[str] = None,
    pack_type: Optional[str] = None,
    status: Optional[int] = None,
    items: Optional[List[Dict[str, Any]]] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
) -> BizConsumablePack:
    obj = get_consumable_pack(db, pack_id)
    if not obj:
        raise ValueError("耗材包不存在")
    if pack_code is not None:
        new_code = pack_code.strip()
        if not new_code:
            raise ValueError("耗材包编码不能为空")
        if new_code != obj.pack_code and pack_code_exists(db, new_code, obj.tenant_no, exclude_id=pack_id):
            raise ValueError("耗材包编码已存在")
        obj.pack_code = new_code
    if pack_name is not None:
        obj.pack_name = (pack_name or "").strip() or obj.pack_name
    if usage_scene is not None:
        obj.usage_scene = (usage_scene or "").strip() or None
    if pack_type is not None:
        obj.pack_type = (pack_type or "").strip() or None
    if status is not None:
        obj.status = str(status)
    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)

    if items is not None:
        existing = (
            db.query(BizConsumablePackItem)
            .filter(and_(BizConsumablePackItem.pack_id == pack_id, BizConsumablePackItem.del_flag == 0))
            .all()
        )
        now = datetime.now()
        for ei in existing:
            ei.del_flag = 1
            ei.update_time = now
            db.add(ei)

        for it in items:
            item_obj = BizConsumablePackItem(
                id=generate_uuid(),
                pack_id=pack_id,
                consumable_code=(it.get("consumable_code") or "").strip(),
                consumable_name=(it.get("consumable_name") or "").strip(),
                spec_model=(it.get("spec_model") or "").strip() or None,
                standard_usage=(it.get("standard_usage") or "").strip() or None,
                unit=(it.get("unit") or "").strip() or None,
                remark=(it.get("remark") or "").strip() or None,
                tenant_no=obj.tenant_no,
                order_num=0,
                create_by=user_id,
                create_dept_id="",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
            db.add(item_obj)

    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_consumable_pack(db: Session, pack_id: str) -> None:
    obj = get_consumable_pack(db, pack_id)
    if not obj:
        raise ValueError("耗材包不存在")
    now = datetime.now()
    obj.del_flag = 1
    obj.update_time = now
    db.add(obj)

    items = (
        db.query(BizConsumablePackItem)
        .filter(and_(BizConsumablePackItem.pack_id == pack_id, BizConsumablePackItem.del_flag == 0))
        .all()
    )
    for it in items:
        it.del_flag = 1
        it.update_time = now
        db.add(it)

    db.commit()


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