zkxw/backend/app/crud/qc_material.py

"""物料库 CRUD"""
from __future__ import annotations

from datetime import 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 BizQcMaterial, BizQcMaterialUsage


def _dec(v: Any) -> float:
    if v is None:
        return 0.0
    return float(v)


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _norm_status(v: Any) -> str:
    """统一状态为 '1'（启用）或 '0'（停用），避免 0 / '0' / or 判断歧义。"""
    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 _is_enabled_status(v: Any) -> bool:
    return _norm_status(v) == "1"


def _row_to_dict(r: BizQcMaterial) -> Dict[str, Any]:
    return {
        "id": r.id,
        "materialCode": r.material_code,
        "materialName": r.material_name,
        "specModel": r.spec_model or "",
        "unit": r.unit or "",
        "stockQty": _dec(r.stock_qty),
        "warningQty": _dec(r.warning_qty),
        "status": _norm_status(r.status),
        "orderNum": r.order_num if r.order_num is not None else 0,
        "remark": r.remark or "",
        "createTime": _fmt_dt(r.create_time),
        "updateTime": _fmt_dt(r.update_time),
    }


def code_exists(
    db: Session, code: str, tenant_no: str, exclude_id: Optional[str] = None
) -> bool:
    c = (code or "").strip()
    if not c:
        return False
    q = db.query(BizQcMaterial).filter(
        BizQcMaterial.material_code == c,
        BizQcMaterial.tenant_no == tenant_no,
        BizQcMaterial.del_flag == 0,
    )
    if exclude_id:
        q = q.filter(BizQcMaterial.id != exclude_id)
    return q.first() is not None


def total_pages(db: Session, tenant_no: str, page_size: int = 10) -> int:
    cnt = (
        db.query(func.count(BizQcMaterial.id))
        .filter(
            BizQcMaterial.tenant_no == tenant_no,
            BizQcMaterial.del_flag == 0,
        )
        .scalar()
        or 0
    )
    ps = max(1, page_size)
    return (cnt + ps - 1) // ps


def list_materials(
    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(BizQcMaterial).filter(BizQcMaterial.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizQcMaterial.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizQcMaterial.material_code.like(like),
                    BizQcMaterial.material_name.like(like),
                )
            )
        st = (status or "").strip()
        if st in ("0", "1"):
            q0 = q0.filter(BizQcMaterial.status == st)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizQcMaterial, user_info, db, tenant_no)
        return q0

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


def create_material(
    db: Session,
    *,
    material_code: str,
    material_name: str,
    spec_model: Optional[str] = None,
    unit: Optional[str] = None,
    stock_qty: Optional[Decimal] = None,
    warning_qty: Optional[Decimal] = None,
    status: str = "1",
    tenant_no: str,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
) -> BizQcMaterial:
    code = (material_code or "").strip()
    if not code:
        raise ValueError("物料编码不能为空")
    if code_exists(db, code, tenant_no):
        raise ValueError("物料编码已存在")
    name = (material_name or "").strip()
    if not name:
        raise ValueError("物料名称不能为空")
    st = _norm_status(status)
    obj = BizQcMaterial(
        id=generate_uuid(),
        material_code=code,
        material_name=name,
        spec_model=(spec_model or "").strip() or None,
        unit=(unit or "").strip() or None,
        stock_qty=stock_qty if stock_qty is not None else 0,
        warning_qty=warning_qty if warning_qty is not None else 0,
        status=st,
        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_dept_id or "",
        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_material(
    db: Session,
    row_id: str,
    *,
    material_code: Optional[str] = None,
    material_name: Optional[str] = None,
    spec_model: Optional[str] = None,
    unit: Optional[str] = None,
    warning_qty: Optional[Decimal] = None,
    status: Optional[str] = None,
    order_num: Optional[int] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
) -> BizQcMaterial:
    obj = get_material(db, row_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("物料不存在")
    if material_code is not None:
        code = material_code.strip()
        if not code:
            raise ValueError("物料编码不能为空")
        if code != obj.material_code and code_exists(db, code, obj.tenant_no, exclude_id=row_id):
            raise ValueError("物料编码已存在")
        obj.material_code = code
    if material_name is not None:
        name = (material_name or "").strip()
        if not name:
            raise ValueError("物料名称不能为空")
        obj.material_name = name
    if spec_model is not None:
        obj.spec_model = (spec_model or "").strip() or None
    if unit is not None:
        obj.unit = (unit or "").strip() or None
    if warning_qty is not None:
        obj.warning_qty = warning_qty
    if status is not None:
        obj.status = _norm_status(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)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_material(db: Session, row_id: str, tenant_no: Optional[str] = None) -> None:
    obj = get_material(db, row_id, tenant_no=tenant_no)
    if not obj:
        raise ValueError("物料不存在")
    if obj.stock_qty is not None and _dec(obj.stock_qty) > 0:
        raise ValueError("该物料仍有库存，无法删除")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def _build_manual_out_remark(
    *,
    receiver: Optional[str] = None,
    purpose: Optional[str] = None,
    remark: Optional[str] = None,
    line_remark: Optional[str] = None,
) -> Optional[str]:
    parts: List[str] = []
    r = (receiver or "").strip()
    if r:
        parts.append(f"领用人：{r}")
    p = (purpose or "").strip()
    if p:
        parts.append(f"用途：{p}")
    extra = (line_remark or remark or "").strip()
    if extra:
        parts.append(extra)
    return "；".join(parts) if parts else None


def stock_in(
    db: Session,
    row_id: str,
    *,
    qty: Decimal,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
    operate_time: Optional[datetime] = None,
) -> BizQcMaterial:
    if qty is None or float(qty) <= 0:
        raise ValueError("入库数量必须大于0")
    # 行锁锁定物料行
    obj = (
        db.query(BizQcMaterial)
        .filter(BizQcMaterial.id == row_id, BizQcMaterial.del_flag == 0)
        .with_for_update()
        .first()
    )
    if not obj:
        raise ValueError("物料不存在")
    if tenant_no and obj.tenant_no != tenant_no:
        raise ValueError("物料不存在")
    if not _is_enabled_status(obj.status):
        raise ValueError(f"物料「{obj.material_name}」已停用，无法入库")
    op_time = operate_time or datetime.now()
    old_stock = obj.stock_qty or 0
    new_stock = old_stock + qty
    obj.stock_qty = new_stock
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    usage = BizQcMaterialUsage(
        id=generate_uuid(),
        material_id=obj.id,
        material_code=obj.material_code,
        material_name=obj.material_name,
        change_type="in",
        change_qty=qty,
        stock_after=new_stock,
        source_type="manual_in",
        tenant_no=obj.tenant_no,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=obj.create_dept_id or "",
        create_time=op_time,
        update_by=user_id,
        update_time=op_time,
        del_flag=0,
    )
    db.add(usage)
    db.commit()
    db.refresh(obj)
    return obj


def batch_stock_in(
    db: Session,
    *,
    items: List[Dict[str, Any]],
    remark: Optional[str] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
    operate_time: Optional[datetime] = None,
) -> int:
    """批量入库，同一事务；任一条失败则全部回滚。"""
    if not items:
        raise ValueError("请至少选择一条物料")
    op_time = operate_time or datetime.now()
    shared_remark = (remark or "").strip() or None

    try:
        for it in items:
            material_id = (it.get("material_id") or it.get("materialId") or "").strip()
            qty = it.get("qty")
            if not material_id:
                raise ValueError("物料ID不能为空")
            if qty is None or float(qty) <= 0:
                raise ValueError("入库数量必须大于0")
            qty_dec = Decimal(str(qty))
            line_remark = (it.get("remark") or "").strip()
            log_remark = line_remark or shared_remark or None

            obj = (
                db.query(BizQcMaterial)
                .filter(
                    BizQcMaterial.id == material_id,
                    BizQcMaterial.tenant_no == tenant_no,
                    BizQcMaterial.del_flag == 0,
                )
                .with_for_update()
                .first()
            )
            if not obj:
                raise ValueError("物料不存在")
            if not _is_enabled_status(obj.status):
                raise ValueError(f"物料「{obj.material_name}」已停用，无法入库")

            old_stock = obj.stock_qty or 0
            new_stock = old_stock + qty_dec
            obj.stock_qty = new_stock
            obj.update_by = user_id
            obj.update_time = datetime.now()
            db.add(obj)

            usage = BizQcMaterialUsage(
                id=generate_uuid(),
                material_id=obj.id,
                material_code=obj.material_code,
                material_name=obj.material_name,
                change_type="in",
                change_qty=qty_dec,
                stock_after=new_stock,
                source_type="manual_in",
                tenant_no=obj.tenant_no,
                remark=log_remark,
                create_by=user_id,
                create_dept_id=obj.create_dept_id or "",
                create_time=op_time,
                update_by=user_id,
                update_time=op_time,
                del_flag=0,
            )
            db.add(usage)
        db.commit()
        return len(items)
    except Exception:
        db.rollback()
        raise


def stock_out(
    db: Session,
    row_id: str,
    *,
    qty: Decimal,
    receiver: Optional[str] = None,
    purpose: Optional[str] = None,
    remark: Optional[str] = None,
    user_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
    operate_time: Optional[datetime] = None,
) -> BizQcMaterial:
    if qty is None or float(qty) <= 0:
        raise ValueError("出库数量必须大于0")
    qty_dec = Decimal(str(qty))
    if not (receiver or "").strip():
        raise ValueError("请选择领用人")

    obj = (
        db.query(BizQcMaterial)
        .filter(BizQcMaterial.id == row_id, BizQcMaterial.del_flag == 0)
        .with_for_update()
        .first()
    )
    if not obj:
        raise ValueError("物料不存在")
    if tenant_no and obj.tenant_no != tenant_no:
        raise ValueError("物料不存在")
    if not _is_enabled_status(obj.status):
        raise ValueError(f"物料「{obj.material_name}」已停用，无法出库")

    stock_before = obj.stock_qty if obj.stock_qty is not None else Decimal("0")
    if stock_before < qty_dec:
        raise ValueError(
            f"物料「{obj.material_name}」库存不足，当前库存 {stock_before}，需出库 {qty_dec}"
        )

    op_time = operate_time or datetime.now()
    new_stock = stock_before - qty_dec
    obj.stock_qty = new_stock
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)

    usage = BizQcMaterialUsage(
        id=generate_uuid(),
        material_id=obj.id,
        material_code=obj.material_code,
        material_name=obj.material_name,
        change_type="use",
        change_qty=-qty_dec,
        stock_after=new_stock,
        source_type="manual_out",
        tenant_no=obj.tenant_no,
        receiver=(receiver or "").strip() or None,
        purpose=(purpose or "").strip() or None,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=obj.create_dept_id or "",
        create_time=op_time,
        update_by=user_id,
        update_time=op_time,
        del_flag=0,
    )
    db.add(usage)
    db.commit()
    db.refresh(obj)
    return obj


def batch_stock_out(
    db: Session,
    *,
    items: List[Dict[str, Any]],
    receiver: Optional[str] = None,
    purpose: Optional[str] = None,
    remark: Optional[str] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
    operate_time: Optional[datetime] = None,
) -> int:
    """批量出库，同一事务；任一条失败则全部回滚。"""
    if not items:
        raise ValueError("请至少选择一条物料")
    if not (receiver or "").strip():
        raise ValueError("请选择领用人")

    op_time = operate_time or datetime.now()
    recv = (receiver or "").strip() or None
    purp = (purpose or "").strip() or None
    shared_remark = (remark or "").strip() or None

    try:
        for it in items:
            material_id = (it.get("material_id") or it.get("materialId") or "").strip()
            qty = it.get("qty")
            if not material_id:
                raise ValueError("物料ID不能为空")
            if qty is None or float(qty) <= 0:
                raise ValueError("出库数量必须大于0")
            qty_dec = Decimal(str(qty))
            line_remark = (it.get("remark") or "").strip() or shared_remark

            obj = (
                db.query(BizQcMaterial)
                .filter(
                    BizQcMaterial.id == material_id,
                    BizQcMaterial.tenant_no == tenant_no,
                    BizQcMaterial.del_flag == 0,
                )
                .with_for_update()
                .first()
            )
            if not obj:
                raise ValueError("物料不存在")
            if not _is_enabled_status(obj.status):
                raise ValueError(f"物料「{obj.material_name}」已停用，无法出库")

            stock_before = obj.stock_qty if obj.stock_qty is not None else Decimal("0")
            if stock_before < qty_dec:
                raise ValueError(
                    f"物料「{obj.material_name}」库存不足，当前库存 {stock_before}，需出库 {qty_dec}"
                )

            new_stock = stock_before - qty_dec
            obj.stock_qty = new_stock
            obj.update_by = user_id
            obj.update_time = datetime.now()
            db.add(obj)

            usage = BizQcMaterialUsage(
                id=generate_uuid(),
                material_id=obj.id,
                material_code=obj.material_code,
                material_name=obj.material_name,
                change_type="use",
                change_qty=-qty_dec,
                stock_after=new_stock,
                source_type="manual_out",
                tenant_no=obj.tenant_no,
                receiver=recv,
                purpose=purp,
                remark=line_remark,
                create_by=user_id,
                create_dept_id=obj.create_dept_id or "",
                create_time=op_time,
                update_by=user_id,
                update_time=op_time,
                del_flag=0,
            )
            db.add(usage)
        db.commit()
        return len(items)
    except Exception:
        db.rollback()
        raise


def options_list(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizQcMaterial)
        .filter(
            BizQcMaterial.tenant_no == tenant_no,
            BizQcMaterial.del_flag == 0,
            BizQcMaterial.status == "1",
        )
        .order_by(BizQcMaterial.create_time.desc())
        .all()
    )
    return [
        {
            "id": r.id,
            "label": r.material_name,
            "materialCode": r.material_code,
            "materialName": r.material_name,
            "specModel": r.spec_model or "",
            "unit": r.unit or "",
            "stockQty": _dec(r.stock_qty),
        }
        for r in rows
    ]