zkxw/backend/app/crud/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 func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.material import BizMaterial, BizMaterialStockBatch, BizMaterialUsage


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:
    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: BizMaterial) -> Dict[str, Any]:
    return {
        "id": r.id,
        "materialCode": r.material_code,
        "materialName": r.material_name,
        "category": r.category or "",
        "specModel": r.spec_model or "",
        "unit": r.unit or "",
        "stockQty": _dec(r.stock_qty),
        "warningQty": _dec(r.warning_qty),
        "status": _norm_status(r.status),
        "imageUrls": r.image_urls or "",
        "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(BizMaterial).filter(
        BizMaterial.material_code == c,
        BizMaterial.tenant_no == tenant_no,
        BizMaterial.del_flag == 0,
    )
    if exclude_id:
        q = q.filter(BizMaterial.id != exclude_id)
    return q.first() is not None


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

    total = _filtered().order_by(None).with_entities(func.count(BizMaterial.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizMaterial.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, material_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    obj = (
        db.query(BizMaterial)
        .filter(
            BizMaterial.id == material_id,
            BizMaterial.tenant_no == tenant_no,
            BizMaterial.del_flag == 0,
        )
        .first()
    )
    if not obj:
        return None
    return _row_to_dict(obj)


def create_material(
    db: Session,
    data: Dict[str, Any],
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    dept_id: str = "",
) -> Dict[str, Any]:
    code = (data.get("materialCode") or "").strip()
    if not code:
        raise ValueError("物料编码不能为空")
    if _code_exists(db, code, tenant_no):
        raise ValueError("物料编码已存在")
    name = (data.get("materialName") or "").strip()
    if not name:
        raise ValueError("物料名称不能为空")
    status = (data.get("status") or "1").strip()
    if status not in ("0", "1"):
        raise ValueError("状态无效")
    obj = BizMaterial(
        id=generate_uuid(),
        material_code=code,
        material_name=name,
        category=(data.get("category") or "").strip() or None,
        spec_model=(data.get("specModel") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        stock_qty=0,
        warning_qty=Decimal(str(data.get("warningQty") or 0)),
        status=status,
        image_urls=(data.get("imageUrls") or "").strip() or None,
        tenant_no=tenant_no,
        order_num=data.get("orderNum") if data.get("orderNum") is not None else 0,
        remark=(data.get("remark") or "").strip() or None,
        create_by=user_id,
        create_dept_id=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 _row_to_dict(obj)


def update_material(
    db: Session,
    material_id: str,
    data: Dict[str, Any],
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> Dict[str, Any]:
    obj = (
        db.query(BizMaterial)
        .filter(
            BizMaterial.id == material_id,
            BizMaterial.tenant_no == tenant_no,
            BizMaterial.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("物料不存在")

    if "materialCode" in data and data["materialCode"] is not None:
        code = data["materialCode"].strip()
        if not code:
            raise ValueError("物料编码不能为空")
        if code != obj.material_code and _code_exists(db, code, tenant_no, exclude_id=material_id):
            raise ValueError("物料编码已存在")
        obj.material_code = code
    if "materialName" in data and data["materialName"] is not None:
        name = data["materialName"].strip()
        if not name:
            raise ValueError("物料名称不能为空")
        obj.material_name = name
    if "category" in data:
        obj.category = (data["category"] or "").strip() or None
    if "specModel" in data:
        obj.spec_model = (data["specModel"] or "").strip() or None
    if "unit" in data:
        obj.unit = (data["unit"] or "").strip() or None
    if "warningQty" in data and data["warningQty"] is not None:
        obj.warning_qty = Decimal(str(data["warningQty"]))
    if "status" in data and data["status"] is not None:
        st = data["status"].strip()
        if st not in ("0", "1"):
            raise ValueError("状态无效")
        obj.status = st
    if "imageUrls" in data:
        obj.image_urls = (data["imageUrls"] or "").strip() or None
    if "orderNum" in data:
        obj.order_num = data["orderNum"]
    if "remark" in data:
        obj.remark = (data["remark"] or "").strip() or None

    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return _row_to_dict(obj)


def soft_delete_material(
    db: Session,
    material_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    obj = (
        db.query(BizMaterial)
        .filter(
            BizMaterial.id == material_id,
            BizMaterial.tenant_no == tenant_no,
            BizMaterial.del_flag == 0,
        )
        .first()
    )
    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_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def get_material_entity(
    db: Session, material_id: str, tenant_no: str
) -> Optional[BizMaterial]:
    return (
        db.query(BizMaterial)
        .filter(
            BizMaterial.id == material_id,
            BizMaterial.tenant_no == tenant_no,
            BizMaterial.del_flag == 0,
        )
        .first()
    )


def update_material_fields(
    db: Session,
    material_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    material_code: Optional[str] = None,
    material_name: Optional[str] = None,
    category: 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,
) -> BizMaterial:
    obj = get_material_entity(db, material_id, 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, tenant_no, exclude_id=material_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 category is not None:
        obj.category = (category or "").strip() or None
    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 options_list(db: Session, tenant_no: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizMaterial)
        .filter(
            BizMaterial.tenant_no == tenant_no,
            BizMaterial.del_flag == 0,
            BizMaterial.status == "1",
        )
        .order_by(BizMaterial.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
    ]


def _my_usage_row_to_dict(
    r: BizMaterialUsage,
    *,
    create_by_name: str = "",
    batch: Optional[BizMaterialStockBatch] = None,
    receiver_name: str = "",
) -> Dict[str, Any]:
    """将 BizMaterialUsage 行转换为字典（用于我的领用记录）。"""
    create_by = r.create_by or ""
    receiver = (getattr(r, "receiver", None) or "").strip()
    purpose = (getattr(r, "purpose", None) or "").strip()
    create_time = r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else ""
    stock_after = float(r.stock_after) if r.stock_after is not None else 0
    change_qty = float(r.change_qty) if r.change_qty is not None else 0
    stock_before = (
        float(r.stock_before)
        if r.stock_before is not None
        else stock_after - change_qty
    )
    unit = (getattr(r, "usage_unit", None) or "").strip()
    if not unit and batch:
        unit = (batch.unit or "").strip()
    return {
        "id": r.id,
        "materialId": r.material_id,
        "materialCode": r.material_code or "",
        "materialName": r.material_name or "",
        "batchId": r.batch_id or "",
        "stockInNo": r.stock_in_no or "",
        "catalogNo": r.catalog_no or "",
        "batchNo": r.batch_no or "",
        "manufacturer": r.manufacturer or "",
        "unitPrice": float(r.unit_price) if r.unit_price is not None else None,
        "amount": float(r.amount) if r.amount is not None else None,
        "changeType": r.change_type,
        "changeQty": change_qty,
        "usageUnit": unit,
        "stockBefore": stock_before,
        "stockAfter": stock_after,
        "receiver": receiver,
        "receiverName": receiver_name or receiver,
        "purpose": purpose,
        "sourceType": r.source_type or "",
        "requisitionStatus": (getattr(r, "requisition_status", None) or "open").strip() or "open",
        "useTime": create_time,
        "createBy": create_by,
        "createByName": create_by_name or create_by,
        "createTime": create_time,
        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
    }


def _my_usage_batch_map(db: Session, batch_ids: List[str]) -> Dict[str, BizMaterialStockBatch]:
    ids = [x for x in batch_ids if x]
    if not ids:
        return {}
    rows = (
        db.query(BizMaterialStockBatch)
        .filter(BizMaterialStockBatch.id.in_(ids), BizMaterialStockBatch.del_flag == 0)
        .all()
    )
    return {r.id: r for r in rows}


def _my_usage_user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
    ids = list({x for x in user_ids if x})
    if not ids:
        return {}
    from app.models.models import SysUser
    rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
    return {u.id: (u.nick_name or "").strip() for u in rows}


def _mine_usage_receiver_filter(user_id: str, nick_name: str):
    """当前登录人的手动出库领用记录：领用人字段存昵称或用户 ID（与生产/质检一致）。"""
    uid = (user_id or "").strip()
    nn = (nick_name or "").strip()
    receiver_match: List[Any] = []
    if nn:
        receiver_match.append(BizMaterialUsage.receiver == nn)
        receiver_match.append(BizMaterialUsage.remark.like(f"%领用人：{nn}%"))
    if uid:
        receiver_match.append(BizMaterialUsage.receiver == uid)
    if receiver_match:
        return or_(*receiver_match)
    return BizMaterialUsage.id == ""


def get_my_usages(
    db: Session,
    current_user_id: str,
    tenant_no: str,
    *,
    current_nick_name: str = "",
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    material_id: Optional[str] = None,
    material_type: Optional[str] = None,
    change_type: Optional[str] = None,
    start_time: Optional[str] = None,
    end_time: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """
    获取当前用户的物料领用记录（手动出库记录）

    筛选条件:
    - tenant_no = 当前用户租户
    - source_type = 'manual_out'（手动出库）
    - receiver 匹配当前用户昵称或用户 ID
    - del_flag = 0

    按 create_time DESC 排序
    """
    q0 = db.query(BizMaterialUsage).filter(
        BizMaterialUsage.tenant_no == tenant_no,
        BizMaterialUsage.del_flag == 0,
        BizMaterialUsage.source_type == "manual_out",
        _mine_usage_receiver_filter(current_user_id, current_nick_name),
    )

    if material_id:
        q0 = q0.filter(BizMaterialUsage.material_id == material_id)

    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(
                BizMaterialUsage.material_code.like(like),
                BizMaterialUsage.material_name.like(like),
                BizMaterialUsage.stock_in_no.like(like),
                BizMaterialUsage.batch_no.like(like),
                BizMaterialUsage.catalog_no.like(like),
                BizMaterialUsage.purpose.like(like),
            )
        )

    if start_time:
        q0 = q0.filter(BizMaterialUsage.create_time >= start_time)
    if end_time:
        q0 = q0.filter(BizMaterialUsage.create_time <= end_time + " 23:59:59")

    total = q0.order_by(None).with_entities(func.count(BizMaterialUsage.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(BizMaterialUsage.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    bmap = _my_usage_batch_map(db, [x.batch_id for x in rows if x.batch_id])
    name_map = _my_usage_user_nick_name_map(db, [x.create_by for x in rows if x.create_by])

    return [
        _my_usage_row_to_dict(
            x,
            create_by_name=name_map.get(x.create_by or "", ""),
            batch=bmap.get(x.batch_id or ""),
        )
        for x in rows
    ], int(total)