zkxw/backend/app/crud/production_material_usage.py

"""物料变动记录 CRUD"""
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

from fastapi import HTTPException
from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.crud.production_material import _filter_material_type
from app.models.production_material import BizProductionMaterial, BizProductionMaterialPrep, BizProductionMaterialStockBatch, BizProductionMaterialUsage
from app.models.models import SysUser
from app.crud.production_material_stock_batch import _usage_batch_snapshots


def _parse_receiver_purpose_from_remark(remark: Optional[str]) -> Tuple[str, str]:
    """兼容旧数据：领用人/用途曾拼接在 remark 中。"""
    receiver, purpose = "", ""
    for part in (remark or "").split("；"):
        seg = part.strip()
        if seg.startswith("领用人："):
            receiver = seg[len("领用人：") :].strip()
        elif seg.startswith("用途："):
            purpose = seg[len("用途：") :].strip()
    return receiver, purpose


def _resolve_receiver_purpose(r: BizProductionMaterialUsage) -> Tuple[str, str]:
    receiver = (getattr(r, "receiver", None) or "").strip()
    purpose = (getattr(r, "purpose", None) or "").strip()
    if receiver or purpose:
        return receiver, purpose
    if (r.source_type or "") == "manual_out":
        return _parse_receiver_purpose_from_remark(r.remark)
    return "", ""


def _display_remark(r: BizProductionMaterialUsage) -> str:
    """手动出库时备注仅展示用户填写内容，不含领用人/用途拼接段。"""
    remark = (r.remark or "").strip()
    if (r.source_type or "") != "manual_out" or not remark:
        return remark
    parts: List[str] = []
    for part in remark.split("；"):
        seg = part.strip()
        if not seg or seg.startswith("领用人：") or seg.startswith("用途："):
            continue
        if seg.startswith("备注"):
            seg = seg[2:].strip() or seg
        parts.append(seg)
    return "；".join(parts) if parts else ""


def _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 {}
    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 _receiver_display_map(db: Session, receivers: List[str]) -> Dict[str, str]:
    """领用人字段可能是用户 ID 或昵称。"""
    keys = list({(x or "").strip() for x in receivers if (x or "").strip()})
    if not keys:
        return {}
    by_id = _user_nick_name_map(db, keys)
    out: Dict[str, str] = {}
    for k in keys:
        out[k] = by_id.get(k) or k
    return out


def _usage_unit_label(r: BizProductionMaterialUsage, batch: Optional[BizProductionMaterialStockBatch]) -> str:
    u = (getattr(r, "usage_unit", None) or "").strip()
    if u:
        return u
    if batch and (batch.unit or "").strip():
        return (batch.unit or "").strip()
    return ""


def _stock_in_date_for_row(r: BizProductionMaterialUsage, batch: Optional[BizProductionMaterialStockBatch]) -> str:
    if batch and batch.stock_in_date:
        return batch.stock_in_date.strftime("%Y-%m-%d")
    return ""


def _unit_price_for_row(r: BizProductionMaterialUsage, batch: Optional[BizProductionMaterialStockBatch]) -> Optional[float]:
    if r.unit_price is not None:
        return float(r.unit_price)
    if batch and batch.unit_price is not None:
        return float(batch.unit_price)
    return None


def _amount_for_row(r: BizProductionMaterialUsage, batch: Optional[BizProductionMaterialStockBatch]) -> Optional[float]:
    is_in = (r.change_type or "") == "in"
    if is_in:
        if r.amount is not None:
            return float(r.amount)
        if batch and batch.amount is not None:
            return float(batch.amount)
        return None
    unit_price = _unit_price_for_row(r, batch)
    change_qty = float(r.change_qty) if r.change_qty is not None else 0
    qty = abs(change_qty)
    if unit_price is None or qty <= 0:
        return None
    return round(unit_price * qty, 2)


def _row_to_dict(
    r: BizProductionMaterialUsage,
    *,
    create_by_name: str = "",
    batch: Optional[BizProductionMaterialStockBatch] = None,
    receiver_name: str = "",
    requisition_receiver: str = "",
    requisition_receiver_name: str = "",
) -> Dict[str, Any]:
    create_by = r.create_by or ""
    receiver, purpose = _resolve_receiver_purpose(r)
    create_time = r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else ""
    is_manual_out = (r.source_type or "") == "manual_out"
    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
    )
    usage_unit = _usage_unit_label(r, batch)
    req_receiver = requisition_receiver or receiver
    req_receiver_name = requisition_receiver_name or receiver_name or req_receiver
    return {
        "id": r.id,
        "materialId": r.material_id,
        "materialType": r.material_type or "",
        "materialCode": r.material_code or "",
        "materialName": r.material_name or "",
        "batchId": r.batch_id or "",
        "stockInNo": r.stock_in_no or "",
        "stockInDate": _stock_in_date_for_row(r, batch),
        "catalogNo": r.catalog_no or "",
        "batchNo": r.batch_no or "",
        "manufacturer": r.manufacturer or "",
        "storageLocation": (r.storage_location or "").strip() or (batch.storage_location if batch else "") or "",
        "unitPrice": _unit_price_for_row(r, batch),
        "amount": _amount_for_row(r, batch),
        "changeType": r.change_type,
        "changeQty": change_qty,
        "usageUnit": usage_unit,
        "stockBefore": stock_before,
        "stockAfter": stock_after,
        "stepId": r.step_id or "",
        "stepName": r.step_name or "",
        "taskId": r.task_id or "",
        "taskNo": r.task_no or "",
        "parentUsageId": (r.parent_usage_id or "").strip(),
        "requisitionUsageId": (r.parent_usage_id or "").strip(),
        "prepId": (r.prep_id or "").strip(),
        "receiver": receiver,
        "receiverName": receiver_name or receiver,
        "requisitionReceiver": req_receiver,
        "requisitionReceiverName": req_receiver_name,
        "purpose": purpose,
        "sourceType": r.source_type or "",
        "requisitionStatus": (getattr(r, "requisition_status", None) or "open").strip() or "open",
        "useTime": create_time if is_manual_out else "",
        "remark": _display_remark(r),
        "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 _batch_map(db: Session, batch_ids: List[str]) -> Dict[str, BizProductionMaterialStockBatch]:
    ids = [x for x in batch_ids if x]
    if not ids:
        return {}
    rows = (
        db.query(BizProductionMaterialStockBatch)
        .filter(BizProductionMaterialStockBatch.id.in_(ids), BizProductionMaterialStockBatch.del_flag == 0)
        .all()
    )
    return {r.id: r for r in rows}


def _mine_usage_filter(user_id: str, nick_name: str):
    """当前登录人的领用记录：手动出库（领用人）或任务使用（填报人）。"""
    uid = (user_id or "").strip()
    nn = (nick_name or "").strip()
    manual_conds = [BizProductionMaterialUsage.source_type == "manual_out"]
    receiver_match: List[Any] = []
    if nn:
        receiver_match.append(BizProductionMaterialUsage.receiver == nn)
        receiver_match.append(BizProductionMaterialUsage.remark.like(f"%领用人：{nn}%"))
    if uid:
        receiver_match.append(BizProductionMaterialUsage.receiver == uid)
    if receiver_match:
        manual_conds.append(or_(*receiver_match))
    else:
        manual_conds.append(BizProductionMaterialUsage.id == "")
    task_conds = [
        BizProductionMaterialUsage.source_type == "step_save",
        BizProductionMaterialUsage.create_by == uid,
    ]
    return and_(
        BizProductionMaterialUsage.change_type == "use",
        or_(and_(*manual_conds), and_(*task_conds)),
    )


def _list_usages_query(
    db: Session,
    tenant_no: str,
    *,
    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,
    keyword: Optional[str] = None,
    stock_in_no: Optional[str] = None,
    user_info=None,
    mine_user_id: Optional[str] = None,
    mine_nick_name: Optional[str] = None,
):
    q0 = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        or_(
            BizProductionMaterialUsage.source_type.is_(None),
            BizProductionMaterialUsage.source_type != "prep_consume",
        ),
    )
    if mine_user_id is not None:
        q0 = q0.filter(_mine_usage_filter(mine_user_id, mine_nick_name or ""))
    if material_id:
        q0 = q0.filter(BizProductionMaterialUsage.material_id == material_id)
    mt = _filter_material_type(material_type)
    if mt:
        q0 = q0.filter(BizProductionMaterialUsage.material_type == mt)
    ct = (change_type or "").strip()
    if ct == "out":
        q0 = q0.filter(
            BizProductionMaterialUsage.change_type == "use",
            BizProductionMaterialUsage.source_type == "manual_out",
        )
    elif ct == "use":
        q0 = q0.filter(
            BizProductionMaterialUsage.change_type == "use",
            BizProductionMaterialUsage.source_type == "step_save",
        )
    elif ct == "in":
        q0 = q0.filter(BizProductionMaterialUsage.change_type == "in")
    elif ct == "discard":
        q0 = q0.filter(BizProductionMaterialUsage.change_type == "discard")
    sno = (stock_in_no or "").strip()
    if sno:
        q0 = q0.filter(BizProductionMaterialUsage.stock_in_no.like(f"%{sno}%"))
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(
                BizProductionMaterialUsage.material_code.like(like),
                BizProductionMaterialUsage.material_name.like(like),
                BizProductionMaterialUsage.stock_in_no.like(like),
                BizProductionMaterialUsage.batch_no.like(like),
                BizProductionMaterialUsage.catalog_no.like(like),
            )
        )
    if start_time:
        q0 = q0.filter(BizProductionMaterialUsage.create_time >= start_time)
    if end_time:
        q0 = q0.filter(BizProductionMaterialUsage.create_time <= end_time + " 23:59:59")
    if user_info and mine_user_id is None:
        from app.core.data_scope import apply_data_scope

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


def _paginate_usages(db: Session, q0, *, page: int, page_size: int) -> Tuple[List[Dict[str, Any]], int]:
    total = q0.order_by(None).with_entities(func.count(BizProductionMaterialUsage.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(BizProductionMaterialUsage.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    bmap = _batch_map(db, [x.batch_id for x in rows if x.batch_id])
    name_map = _user_nick_name_map(db, [x.create_by for x in rows if x.create_by])
    receivers = []
    for x in rows:
        rcv, _ = _resolve_receiver_purpose(x)
        if rcv:
            receivers.append(rcv)
    receiver_map = _receiver_display_map(db, receivers)
    return [
        _row_to_dict(
            x,
            create_by_name=name_map.get(x.create_by or "", ""),
            batch=bmap.get(x.batch_id or ""),
            receiver_name=receiver_map.get(
                (_resolve_receiver_purpose(x)[0] or ""), _resolve_receiver_purpose(x)[0] or ""
            ),
        )
        for x in rows
    ], int(total)


def list_usages(
    db: Session,
    tenant_no: str,
    *,
    page: int = 1,
    page_size: int = 10,
    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,
    keyword: Optional[str] = None,
    stock_in_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q0 = _list_usages_query(
        db,
        tenant_no,
        material_id=material_id,
        material_type=material_type,
        change_type=change_type,
        start_time=start_time,
        end_time=end_time,
        keyword=keyword,
        stock_in_no=stock_in_no,
        user_info=user_info,
    )
    return _paginate_usages(db, q0, page=page, page_size=page_size)


def list_my_usages(
    db: Session,
    tenant_no: str,
    *,
    user_id: str,
    nick_name: str = "",
    page: int = 1,
    page_size: int = 10,
    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,
    keyword: Optional[str] = None,
    stock_in_no: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """当前登录人的物料领用记录（手动出库 + 任务使用）。"""
    q0 = _list_usages_query(
        db,
        tenant_no,
        material_id=material_id,
        material_type=material_type,
        change_type=change_type,
        start_time=start_time,
        end_time=end_time,
        keyword=keyword,
        stock_in_no=stock_in_no,
        mine_user_id=user_id,
        mine_nick_name=nick_name,
    )
    return _paginate_usages(db, q0, page=page, page_size=page_size)


def list_usages_by_step(
    db: Session,
    step_id: str,
    tenant_no: str,
    *,
    user_info=None,
) -> List[Dict[str, Any]]:
    """查询生产步骤关联的物料使用记录（step_save）。"""
    q = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.step_id == step_id,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type == "step_save",
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizProductionMaterialUsage, user_info, db, tenant_no)
    rows = q.order_by(BizProductionMaterialUsage.create_time.asc()).all()
    bmap = _batch_map(db, [x.batch_id for x in rows if x.batch_id])
    parent_ids = [x.parent_usage_id for x in rows if (x.parent_usage_id or "").strip()]
    parent_map: Dict[str, BizProductionMaterialUsage] = {}
    if parent_ids:
        parents = (
            db.query(BizProductionMaterialUsage)
            .filter(
                BizProductionMaterialUsage.id.in_(parent_ids),
                BizProductionMaterialUsage.del_flag == 0,
            )
            .all()
        )
        parent_map = {p.id: p for p in parents}
    parent_receivers = []
    for p in parent_map.values():
        rcv, _ = _resolve_receiver_purpose(p)
        if rcv:
            parent_receivers.append(rcv)
    parent_receiver_map = _receiver_display_map(db, parent_receivers)
    out: List[Dict[str, Any]] = []
    for x in rows:
        parent = parent_map.get((x.parent_usage_id or "").strip())
        req_rcv, _ = _resolve_receiver_purpose(parent) if parent else ("", "")
        req_rcv_name = parent_receiver_map.get(req_rcv, req_rcv) if req_rcv else ""
        rcv, _ = _resolve_receiver_purpose(x)
        out.append(
            _row_to_dict(
                x,
                batch=bmap.get(x.batch_id or ""),
                receiver_name=parent_receiver_map.get(rcv, rcv) if not parent else req_rcv_name,
                requisition_receiver=req_rcv,
                requisition_receiver_name=req_rcv_name,
            )
        )
    return out


def list_linked_step_usages(
    db: Session,
    parent_usage_id: str,
    tenant_no: str,
    *,
    user_info=None,
) -> List[Dict[str, Any]]:
    """查询领用记录关联的消耗明细（parent_usage_id → step_save / prep_consume）。"""
    pid = (parent_usage_id or "").strip()
    if not pid:
        return []
    parent = (
        db.query(BizProductionMaterialUsage)
        .filter(
            BizProductionMaterialUsage.id == pid,
            BizProductionMaterialUsage.tenant_no == tenant_no,
            BizProductionMaterialUsage.del_flag == 0,
        )
        .first()
    )
    if not parent:
        raise HTTPException(status_code=404, detail="领用记录不存在")
    if (parent.source_type or "") != "manual_out":
        return []

    q = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.parent_usage_id == pid,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type.in_(["step_save", "prep_consume"]),
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizProductionMaterialUsage, user_info, db, tenant_no)
    rows = q.order_by(BizProductionMaterialUsage.create_time.desc()).all()
    bmap = _batch_map(db, [x.batch_id for x in rows if x.batch_id])
    name_map = _user_nick_name_map(db, [x.create_by for x in rows if x.create_by])

    prep_ids = list({(x.prep_id or "").strip() for x in rows if (x.prep_id or "").strip()})
    prep_map: Dict[str, Dict[str, str]] = {}
    if prep_ids:
        preps = (
            db.query(BizProductionMaterialPrep)
            .filter(
                BizProductionMaterialPrep.id.in_(prep_ids),
                BizProductionMaterialPrep.tenant_no == tenant_no,
                BizProductionMaterialPrep.del_flag == 0,
            )
            .all()
        )
        prep_map = {
            p.id: {
                "prepNo": p.prep_no or "",
                "configuredMaterialName": p.configured_material_name or "",
            }
            for p in preps
        }

    out: List[Dict[str, Any]] = []
    for x in rows:
        item = _row_to_dict(
            x,
            create_by_name=name_map.get(x.create_by or "", ""),
            batch=bmap.get(x.batch_id or ""),
            requisition_receiver_name="",
        )
        item["productionTaskId"] = item.get("taskId") or ""
        item["productionTaskNo"] = item.get("taskNo") or ""
        prep_info = prep_map.get((x.prep_id or "").strip(), {})
        item["prepNo"] = prep_info.get("prepNo", "")
        item["configuredMaterialName"] = prep_info.get("configuredMaterialName", "")
        out.append(item)
    return out


def revert_step_material_usages(
    db: Session,
    step_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    """撤销步骤此前 step_save 扣减，便于重新填报物料使用。"""
    rows = (
        db.query(BizProductionMaterialUsage)
        .filter(
            BizProductionMaterialUsage.step_id == step_id,
            BizProductionMaterialUsage.tenant_no == tenant_no,
            BizProductionMaterialUsage.del_flag == 0,
            BizProductionMaterialUsage.source_type == "step_save",
            BizProductionMaterialUsage.change_type == "use",
        )
        .all()
    )
    if not rows:
        return
    now = datetime.now()
    for r in rows:
        if (r.parent_usage_id or "").strip():
            r.del_flag = 1
            r.update_by = user_id
            r.update_time = now
            continue
        restore_qty = abs(r.change_qty) if r.change_qty is not None else Decimal("0")
        if restore_qty <= 0:
            r.del_flag = 1
            r.update_by = user_id
            r.update_time = now
            continue
        batch_id = (r.batch_id or "").strip()
        if batch_id:
            batch = (
                db.query(BizProductionMaterialStockBatch)
                .with_for_update()
                .filter(
                    BizProductionMaterialStockBatch.id == batch_id,
                    BizProductionMaterialStockBatch.tenant_no == tenant_no,
                    BizProductionMaterialStockBatch.del_flag == 0,
                )
                .first()
            )
            if batch:
                batch.remaining_qty = (batch.remaining_qty or Decimal("0")) + restore_qty
                batch.update_by = user_id
                batch.update_time = now
        material = (
            db.query(BizProductionMaterial)
            .with_for_update()
            .filter(
                BizProductionMaterial.id == r.material_id,
                BizProductionMaterial.tenant_no == tenant_no,
                BizProductionMaterial.del_flag == 0,
            )
            .first()
        )
        if material:
            material.stock_qty = (material.stock_qty or Decimal("0")) + restore_qty
            material.update_by = user_id
            material.update_time = now
        r.del_flag = 1
        r.update_by = user_id
        r.update_time = now


def list_requisition_options(
    db: Session,
    tenant_no: str,
    *,
    keyword: Optional[str] = None,
    material_type: Optional[str] = None,
    limit: int = 50,
    user_info=None,
) -> List[Dict[str, Any]]:
    """领用记录选项（manual_out），供生产步骤关联物料使用。"""
    q0 = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type == "manual_out",
    )
    mt = _filter_material_type(material_type)
    if mt:
        q0 = q0.filter(BizProductionMaterialUsage.material_type == mt)
    if user_info:
        from app.core.data_scope import apply_data_scope

        q0 = apply_data_scope(q0, BizProductionMaterialUsage, user_info, db, tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(
                BizProductionMaterialUsage.material_code.like(like),
                BizProductionMaterialUsage.material_name.like(like),
                BizProductionMaterialUsage.stock_in_no.like(like),
                BizProductionMaterialUsage.batch_no.like(like),
                BizProductionMaterialUsage.catalog_no.like(like),
                BizProductionMaterialUsage.receiver.like(like),
            )
        )
    limit = min(max(1, limit), 100)
    rows = q0.order_by(BizProductionMaterialUsage.create_time.desc()).limit(limit).all()
    bmap = _batch_map(db, [x.batch_id for x in rows if x.batch_id])
    receivers = []
    for r in rows:
        rcv, _ = _resolve_receiver_purpose(r)
        if rcv:
            receivers.append(rcv)
    receiver_map = _receiver_display_map(db, receivers)
    options: List[Dict[str, Any]] = []
    for r in rows:
        batch = bmap.get(r.batch_id or "")
        rcv, _ = _resolve_receiver_purpose(r)
        rcv_name = receiver_map.get(rcv, rcv) if rcv else ""
        qty_abs = abs(float(r.change_qty or 0))
        unit = _usage_unit_label(r, batch)
        use_time = r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else ""
        unit_suffix = unit or ""
        label = (
            f"{use_time} {r.material_name or ''} / {r.stock_in_no or ''} "
            f"/ 批号:{r.batch_no or '—'} / 领用:{qty_abs}{unit_suffix} / 领用人:{rcv_name or '—'}"
        ).strip()
        options.append(
            {
                "id": r.id,
                "label": label,
                "materialId": r.material_id,
                "materialType": r.material_type or "",
                "materialName": r.material_name or "",
                "stockInNo": r.stock_in_no or "",
                "batchNo": r.batch_no or "",
                "batchId": r.batch_id or "",
                "receiver": rcv,
                "receiverName": rcv_name,
                "requisitionQty": qty_abs,
                "usageUnit": unit,
                "useTime": use_time,
            }
        )
    return options


def _record_step_usage_from_requisition(
    db: Session,
    step_id: str,
    step_name: str,
    task_id: str,
    task_no: str,
    item: Dict[str, Any],
    tenant_no: str,
    user_id: Optional[str] = None,
    dept_id: str = "",
) -> None:
    req_id = (
        item.get("requisitionUsageId")
        or item.get("requisition_usage_id")
        or item.get("parentUsageId")
        or item.get("parent_usage_id")
        or ""
    ).strip()
    if not req_id:
        raise HTTPException(status_code=400, detail="请选择领用记录")
    qty = item.get("qty")
    if qty is None or Decimal(str(qty)) <= 0:
        raise HTTPException(status_code=400, detail="物料使用数量必须大于0")
    qty_dec = Decimal(str(qty))

    req = (
        db.query(BizProductionMaterialUsage)
        .filter(
            BizProductionMaterialUsage.id == req_id,
            BizProductionMaterialUsage.tenant_no == tenant_no,
            BizProductionMaterialUsage.del_flag == 0,
            BizProductionMaterialUsage.source_type == "manual_out",
            BizProductionMaterialUsage.change_type == "use",
        )
        .first()
    )
    if not req:
        raise HTTPException(status_code=400, detail="领用记录不存在或已失效")

    material = (
        db.query(BizProductionMaterial)
        .filter(
            BizProductionMaterial.id == req.material_id,
            BizProductionMaterial.tenant_no == tenant_no,
            BizProductionMaterial.del_flag == 0,
        )
        .first()
    )
    if not material:
        raise HTTPException(status_code=400, detail="关联物料不存在")

    usage_unit = (
        str(item.get("usageUnit") or item.get("usage_unit") or "").strip()
        or _usage_unit_label(req, None)
        or ""
    )[:32]

    stock_qty = material.stock_qty if material.stock_qty is not None else Decimal("0")
    usage_remark = str(item.get("remark") or "").strip()[:512]
    now = datetime.now()

    usage = BizProductionMaterialUsage(
        id=generate_uuid(),
        material_id=req.material_id,
        material_type=req.material_type or material.material_type,
        material_code=req.material_code or material.material_code,
        material_name=req.material_name or material.material_name,
        batch_id=req.batch_id,
        stock_in_no=req.stock_in_no,
        catalog_no=req.catalog_no,
        batch_no=req.batch_no,
        manufacturer=req.manufacturer,
        storage_location=req.storage_location,
        unit_price=req.unit_price,
        amount=req.amount,
        change_type="use",
        change_qty=-qty_dec,
        usage_unit=usage_unit or None,
        stock_before=stock_qty,
        stock_after=stock_qty,
        step_id=step_id,
        step_name=step_name,
        task_id=task_id,
        task_no=task_no,
        parent_usage_id=req_id,
        source_type="step_save",
        tenant_no=tenant_no,
        remark=usage_remark,
        create_by=user_id,
        create_dept_id=dept_id,
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(usage)


def _deduct_step_usage_legacy_batch(
    db: Session,
    step_id: str,
    step_name: str,
    task_id: str,
    task_no: str,
    item: Dict[str, Any],
    tenant_no: str,
    user_id: Optional[str] = None,
    dept_id: str = "",
) -> None:
    """旧版：按入库批次扣减库存（兼容历史数据）。"""
    batch_id = (item.get("batchId") or item.get("batch_id") or "").strip()
    qty = item.get("qty")
    if not batch_id:
        raise HTTPException(status_code=400, detail="入库批次不能为空")
    if qty is None or Decimal(str(qty)) <= 0:
        raise HTTPException(status_code=400, detail="物料使用数量必须大于0")

    qty_dec = Decimal(str(qty))

    batch = (
        db.query(BizProductionMaterialStockBatch)
        .with_for_update()
        .filter(
            BizProductionMaterialStockBatch.id == batch_id,
            BizProductionMaterialStockBatch.tenant_no == tenant_no,
            BizProductionMaterialStockBatch.del_flag == 0,
        )
        .first()
    )
    if not batch:
        raise HTTPException(status_code=400, detail=f"入库批次不存在（ID: {batch_id}）")

    rem = batch.remaining_qty or Decimal("0")
    if rem < qty_dec:
        raise HTTPException(
            status_code=400,
            detail=f"批次「{batch.stock_in_no}」库存不足，当前剩余 {rem}，需使用 {qty_dec}",
        )

    material = (
        db.query(BizProductionMaterial)
        .with_for_update()
        .filter(
            BizProductionMaterial.id == batch.material_id,
            BizProductionMaterial.tenant_no == tenant_no,
            BizProductionMaterial.del_flag == 0,
            BizProductionMaterial.status == "1",
        )
        .first()
    )
    if not material:
        raise HTTPException(status_code=400, detail="物料不存在或已停用")

    batch.remaining_qty = rem - qty_dec
    batch.update_by = user_id
    batch.update_time = datetime.now()
    db.add(batch)

    stock_qty = material.stock_qty if material.stock_qty is not None else Decimal("0")
    if stock_qty < qty_dec:
        raise HTTPException(
            status_code=400,
            detail=f"物料「{material.material_name}」汇总库存不足",
        )
    new_stock = stock_qty - qty_dec
    material.stock_qty = new_stock
    material.update_by = user_id
    material.update_time = datetime.now()
    db.add(material)

    usage_remark = str(item.get("remark") or "").strip()[:512]
    usage_unit = str(item.get("usageUnit") or item.get("usage_unit") or "").strip()[:32] or None
    snaps = _usage_batch_snapshots(batch, out_qty=qty_dec)

    usage = BizProductionMaterialUsage(
        id=generate_uuid(),
        material_id=material.id,
        material_type=material.material_type,
        material_code=material.material_code,
        material_name=material.material_name,
        change_type="use",
        change_qty=-qty_dec,
        usage_unit=usage_unit,
        stock_before=stock_qty,
        stock_after=new_stock,
        step_id=step_id,
        step_name=step_name,
        task_id=task_id,
        task_no=task_no,
        source_type="step_save",
        tenant_no=tenant_no,
        remark=usage_remark,
        create_by=user_id,
        create_dept_id=dept_id,
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
        **snaps,
    )
    db.add(usage)


def register_step_usage_from_requisition(
    db: Session,
    step_id: str,
    step_name: str,
    task_id: str,
    task_no: str,
    material_usages: List[Dict[str, Any]],
    tenant_no: str,
    user_id: Optional[str] = None,
    dept_id: str = "",
) -> None:
    """登记步骤物料使用；关联领用记录时不扣减库存。调用方控制事务，本函数不 commit。"""
    revert_step_material_usages(db, step_id, tenant_no, user_id=user_id)
    seen_requisition: set = set()
    seen_batch: set = set()
    for item in material_usages:
        req_id = (
            item.get("requisitionUsageId")
            or item.get("requisition_usage_id")
            or item.get("parentUsageId")
            or ""
        ).strip()
        if req_id:
            if req_id in seen_requisition:
                raise HTTPException(status_code=400, detail="物料列表中存在重复领用记录，请检查")
            seen_requisition.add(req_id)
            _record_step_usage_from_requisition(
                db,
                step_id,
                step_name,
                task_id,
                task_no,
                item,
                tenant_no,
                user_id=user_id,
                dept_id=dept_id,
            )
            continue
        batch_id = (item.get("batchId") or item.get("batch_id") or "").strip()
        if not batch_id:
            raise HTTPException(status_code=400, detail="请选择领用记录")
        if batch_id in seen_batch:
            raise HTTPException(status_code=400, detail="物料列表中存在重复入库批次，请检查")
        seen_batch.add(batch_id)
        _deduct_step_usage_legacy_batch(
            db,
            step_id,
            step_name,
            task_id,
            task_no,
            item,
            tenant_no,
            user_id=user_id,
            dept_id=dept_id,
        )


def list_available_requisitions_for_prep(
    db: Session,
    tenant_no: str,
    *,
    page: int = 1,
    page_size: int = 50,
    keyword: Optional[str] = None,
    material_type: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    """配制子物料来源：全部未标记已用完的领用记录。"""
    q0 = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type == "manual_out",
        BizProductionMaterialUsage.material_type.in_(["reagent", "consumable"]),
        BizProductionMaterialUsage.requisition_status != "closed",
    )

    mt = _filter_material_type(material_type)
    if mt:
        q0 = q0.filter(BizProductionMaterialUsage.material_type == mt)

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

    if user_info:
        from app.core.data_scope import apply_data_scope

        q0 = apply_data_scope(q0, BizProductionMaterialUsage, user_info, db, tenant_no)

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

    bmap = _batch_map(db, [x.batch_id for x in rows if x.batch_id])
    receivers = []
    for r in rows:
        rcv, _ = _resolve_receiver_purpose(r)
        if rcv:
            receivers.append(rcv)
    receiver_map = _receiver_display_map(db, receivers)

    items: List[Dict[str, Any]] = []
    for r in rows:
        batch = bmap.get(r.batch_id or "")
        rcv, _ = _resolve_receiver_purpose(r)
        rcv_name = receiver_map.get(rcv, rcv) if rcv else ""
        req_qty = abs(float(r.change_qty or 0))
        unit = _usage_unit_label(r, batch)
        use_time = r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else ""
        items.append(
            {
                "requisitionUsageId": r.id,
                "batchId": r.batch_id or "",
                "materialId": r.material_id,
                "materialType": r.material_type or "",
                "materialCode": r.material_code or "",
                "materialName": r.material_name or "",
                "specModel": (batch.spec_model if batch else "") or "",
                "unit": unit,
                "stockInNo": r.stock_in_no or "",
                "stockInDate": _stock_in_date_for_row(r, batch),
                "expireDate": _fmt_date_val(batch.expire_date if batch else None),
                "catalogNo": r.catalog_no or "",
                "batchNo": r.batch_no or "",
                "manufacturer": r.manufacturer or "",
                "storageLocation": (r.storage_location or "").strip() or (batch.storage_location if batch else "") or "",
                "requisitionQty": req_qty,
                "unitPrice": _unit_price_for_row(r, batch),
                "receiver": rcv,
                "receiverName": rcv_name,
                "useTime": use_time,
            }
        )
    return items, int(total)


def _step_consumed_qty_map(
    db: Session,
    tenant_no: str,
    parent_usage_ids: List[str],
) -> Dict[str, float]:
    """统计领用记录关联的生产步骤消耗量。"""
    ids = [x for x in parent_usage_ids if x]
    if not ids:
        return {}
    rows = (
        db.query(
            BizProductionMaterialUsage.parent_usage_id,
            func.coalesce(func.sum(func.abs(BizProductionMaterialUsage.change_qty)), 0),
        )
        .filter(
            BizProductionMaterialUsage.tenant_no == tenant_no,
            BizProductionMaterialUsage.del_flag == 0,
            BizProductionMaterialUsage.parent_usage_id.in_(ids),
            BizProductionMaterialUsage.change_type == "use",
            BizProductionMaterialUsage.source_type == "step_save",
        )
        .group_by(BizProductionMaterialUsage.parent_usage_id)
        .all()
    )
    return {pid: float(qty or 0) for pid, qty in rows if pid}


def _fmt_date_val(v) -> str:
    if v is None:
        return ""
    if hasattr(v, "strftime"):
        return v.strftime("%Y-%m-%d")
    return str(v)[:10]


def list_stock_check_requisitions(
    db: Session,
    tenant_no: str,
    *,
    page: int = 1,
    page_size: int = 50,
    keyword: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    """盘点登记弹窗：待处理领用记录（未标记已使用完）。"""
    q0 = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type == "manual_out",
        BizProductionMaterialUsage.requisition_status == "open",
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q0 = apply_data_scope(q0, BizProductionMaterialUsage, user_info, db, tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(
                BizProductionMaterialUsage.material_code.like(like),
                BizProductionMaterialUsage.material_name.like(like),
                BizProductionMaterialUsage.stock_in_no.like(like),
                BizProductionMaterialUsage.batch_no.like(like),
                BizProductionMaterialUsage.catalog_no.like(like),
                BizProductionMaterialUsage.receiver.like(like),
            )
        )
    total = q0.order_by(None).with_entities(func.count(BizProductionMaterialUsage.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 200)
    rows = (
        q0.order_by(BizProductionMaterialUsage.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    bmap = _batch_map(db, [x.batch_id for x in rows if x.batch_id])
    name_map = _user_nick_name_map(db, [x.create_by for x in rows if x.create_by])
    receivers = []
    for r in rows:
        rcv, _ = _resolve_receiver_purpose(r)
        if rcv:
            receivers.append(rcv)
    receiver_map = _receiver_display_map(db, receivers)
    consumed_map = _step_consumed_qty_map(db, tenant_no, [x.id for x in rows])
    out: List[Dict[str, Any]] = []
    for r in rows:
        batch = bmap.get(r.batch_id or "")
        rcv, purpose = _resolve_receiver_purpose(r)
        rcv_name = receiver_map.get(rcv, rcv) if rcv else ""
        qty_abs = abs(float(r.change_qty or 0))
        unit = _usage_unit_label(r, batch)
        use_time = r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else ""
        task_consumed = consumed_map.get(r.id, 0)
        out.append(
            {
                "id": r.id,
                "materialId": r.material_id,
                "materialType": r.material_type or "",
                "materialCode": r.material_code or "",
                "materialName": r.material_name or "",
                "specModel": (batch.spec_model if batch else "") or "",
                "batchId": r.batch_id or "",
                "stockInNo": r.stock_in_no or "",
                "stockInDate": _stock_in_date_for_row(r, batch),
                "expireDate": _fmt_date_val(batch.expire_date if batch else None),
                "catalogNo": r.catalog_no or "",
                "batchNo": r.batch_no or "",
                "manufacturer": r.manufacturer or "",
                "storageLocation": (r.storage_location or "").strip() or (batch.storage_location if batch else "") or "",
                "unitPrice": _unit_price_for_row(r, batch),
                "unit": unit,
                "requisitionQty": qty_abs,
                "taskConsumedQty": task_consumed,
                "receiver": rcv,
                "receiverName": rcv_name,
                "purpose": purpose,
                "useTime": use_time,
                "requisitionStatus": (r.requisition_status or "open").strip() or "open",
                "createByName": name_map.get(r.create_by or "", r.create_by or ""),
            }
        )
    return out, int(total)


def mark_requisition_fully_used(
    db: Session,
    usage_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    user_info=None,
) -> None:
    row = (
        db.query(BizProductionMaterialUsage)
        .filter(
            BizProductionMaterialUsage.id == usage_id,
            BizProductionMaterialUsage.tenant_no == tenant_no,
            BizProductionMaterialUsage.del_flag == 0,
            BizProductionMaterialUsage.change_type == "use",
            BizProductionMaterialUsage.source_type == "manual_out",
        )
        .first()
    )
    if not row:
        raise ValueError("领用记录不存在或不可标记")
    if user_info:
        from app.core.data_scope import apply_data_scope

        scoped = apply_data_scope(
            db.query(BizProductionMaterialUsage).filter(BizProductionMaterialUsage.id == usage_id),
            BizProductionMaterialUsage,
            user_info,
            db,
            tenant_no,
        )
        if not scoped.first():
            raise PermissionError("无权操作该领用记录")
    if (row.requisition_status or "open") == "closed":
        return
    now = datetime.now()
    row.requisition_status = "closed"
    row.update_by = user_id
    row.update_time = now
    db.add(row)
    db.commit()


def _prep_consumption_to_dict(
    r: BizProductionMaterialUsage,
    *,
    prep_no: str = "",
    configured_material_name: str = "",
    create_by_name: str = "",
) -> Dict[str, Any]:
    """配制消耗明细行转字典。"""
    return {
        "id": r.id,
        "materialId": r.material_id,
        "materialType": r.material_type or "",
        "materialCode": r.material_code or "",
        "materialName": r.material_name or "",
        "batchId": r.batch_id or "",
        "stockInNo": r.stock_in_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,
        "changeQty": float(r.change_qty) if r.change_qty is not None else 0,
        "usageUnit": r.usage_unit or "",
        "prepId": r.prep_id or "",
        "prepNo": prep_no,
        "configuredMaterialName": configured_material_name,
        "sourceType": r.source_type or "",
        "remark": r.remark or "",
        "createBy": r.create_by or "",
        "createByName": create_by_name or r.create_by or "",
        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
    }


def list_prep_consumptions(
    db: Session,
    tenant_no: str,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    material_type: Optional[str] = None,
    start_time: Optional[str] = None,
    end_time: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """查询配制消耗明细（sub_consume）。"""
    q0 = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type == "prep_consume",
    )

    if material_type:
        mt = material_type.strip().lower()
        if mt in ("reagent", "consumable"):
            q0 = q0.filter(BizProductionMaterialUsage.material_type == mt)

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

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

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

    # 收集 prep_id 并查询配制单信息
    prep_ids = list({(r.prep_id or "").strip() for r in rows if (r.prep_id or "").strip()})
    prep_map: Dict[str, Dict[str, Any]] = {}
    if prep_ids:
        preps = (
            db.query(BizProductionMaterialPrep)
            .filter(
                BizProductionMaterialPrep.id.in_(prep_ids),
                BizProductionMaterialPrep.tenant_no == tenant_no,
                BizProductionMaterialPrep.del_flag == 0,
            )
            .all()
        )
        prep_map = {p.id: {"prepNo": p.prep_no, "configuredMaterialName": p.configured_material_name or ""} for p in preps}

    name_map = _user_nick_name_map(db, [r.create_by for r in rows if r.create_by])

    items = [
        _prep_consumption_to_dict(
            r,
            prep_no=prep_map.get((r.prep_id or "").strip(), {}).get("prepNo", ""),
            configured_material_name=prep_map.get((r.prep_id or "").strip(), {}).get("configuredMaterialName", ""),
            create_by_name=name_map.get(r.create_by or "", ""),
        )
        for r in rows
    ]
    return items, int(total)


def list_my_prep_consumptions(
    db: Session,
    tenant_no: str,
    *,
    user_id: str,
    nick_name: str = "",
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    material_type: Optional[str] = None,
    start_time: Optional[str] = None,
    end_time: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    """当前登录人领用批次的配制消耗明细（按领用人，非配制操作人）。"""
    uid = (user_id or "").strip()
    nn = (nick_name or "").strip()
    parent_conds = [BizProductionMaterialUsage.source_type == "manual_out"]
    receiver_match: List[Any] = []
    if nn:
        receiver_match.append(BizProductionMaterialUsage.receiver == nn)
        receiver_match.append(BizProductionMaterialUsage.remark.like(f"%领用人：{nn}%"))
    if uid:
        receiver_match.append(BizProductionMaterialUsage.receiver == uid)
    if not receiver_match:
        return [], 0
    parent_conds.append(or_(*receiver_match))

    parent_id_q = (
        db.query(BizProductionMaterialUsage.id)
        .filter(
            BizProductionMaterialUsage.tenant_no == tenant_no,
            BizProductionMaterialUsage.del_flag == 0,
            and_(*parent_conds),
        )
    )

    q0 = db.query(BizProductionMaterialUsage).filter(
        BizProductionMaterialUsage.tenant_no == tenant_no,
        BizProductionMaterialUsage.del_flag == 0,
        BizProductionMaterialUsage.change_type == "use",
        BizProductionMaterialUsage.source_type == "prep_consume",
        BizProductionMaterialUsage.parent_usage_id.in_(parent_id_q),
    )

    if material_type:
        mt = material_type.strip().lower()
        if mt in ("reagent", "consumable"):
            q0 = q0.filter(BizProductionMaterialUsage.material_type == mt)

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

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

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

    prep_ids = list({(r.prep_id or "").strip() for r in rows if (r.prep_id or "").strip()})
    prep_map: Dict[str, Dict[str, Any]] = {}
    if prep_ids:
        preps = (
            db.query(BizProductionMaterialPrep)
            .filter(
                BizProductionMaterialPrep.id.in_(prep_ids),
                BizProductionMaterialPrep.tenant_no == tenant_no,
                BizProductionMaterialPrep.del_flag == 0,
            )
            .all()
        )
        prep_map = {p.id: {"prepNo": p.prep_no, "configuredMaterialName": p.configured_material_name or ""} for p in preps}

    items = [
        _prep_consumption_to_dict(
            r,
            prep_no=prep_map.get((r.prep_id or "").strip(), {}).get("prepNo", ""),
            configured_material_name=prep_map.get((r.prep_id or "").strip(), {}).get("configuredMaterialName", ""),
        )
        for r in rows
    ]
    return items, int(total)