zkxw/backend/app/crud/qc_material_stock_batch.py

"""质检物料入库批次 CRUD"""
from __future__ import annotations

from datetime import date, 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.crud.qc_material import _dec, _fmt_dt, _is_enabled_status, _norm_status
from app.crud.qc_material_batch_stock_status import normalize_batch_stock_status
from app.models.qc_material import BizQcMaterial, BizQcMaterialStockBatch, BizQcMaterialUsage


def _fmt_date(d: Optional[date]) -> str:
    return d.strftime("%Y-%m-%d") if d else ""


def _parse_date(v: Any, *, label: str = "入库日期") -> date:
    if isinstance(v, date) and not isinstance(v, datetime):
        return v
    s = str(v or "").strip()
    if not s:
        raise ValueError(f"{label}不能为空")
    if len(s) >= 10:
        s = s[:10]
    try:
        return datetime.strptime(s, "%Y-%m-%d").date()
    except ValueError as e:
        raise ValueError("入库日期格式无效，应为 YYYY-MM-DD") from e


def _parse_expire_date(v: Any) -> date:
    return _parse_date(v, label="过期日期")


def _require_str(v: Any, label: str) -> str:
    s = str(v or "").strip()
    if not s:
        raise ValueError(f"{label}不能为空")
    return s


def _require_decimal(v: Any, label: str) -> Decimal:
    if v is None or str(v).strip() == "":
        raise ValueError(f"{label}不能为空")
    try:
        d = Decimal(str(v))
    except Exception as e:
        raise ValueError(f"{label}格式无效") from e
    if d < 0:
        raise ValueError(f"{label}不能为负数")
    return d


def stock_in_no_exists(
    db: Session, stock_in_no: str, tenant_no: str, exclude_id: Optional[str] = None
) -> bool:
    s = (stock_in_no or "").strip()
    if not s:
        return False
    q = db.query(BizQcMaterialStockBatch).filter(
        BizQcMaterialStockBatch.stock_in_no == s,
        BizQcMaterialStockBatch.tenant_no == tenant_no,
        BizQcMaterialStockBatch.del_flag == 0,
    )
    if exclude_id:
        q = q.filter(BizQcMaterialStockBatch.id != exclude_id)
    return q.first() is not None


def _batch_row_to_dict(b: BizQcMaterialStockBatch, material: Optional[BizQcMaterial] = None) -> Dict[str, Any]:
    mat = material
    warning_qty = _dec(mat.warning_qty) if mat else 0
    stock_qty = _dec(mat.stock_qty) if mat else _dec(b.remaining_qty)
    status = _norm_status(mat.status) if mat else "1"
    return {
        "id": b.id,
        "batchId": b.id,
        "materialId": b.material_id,
        "materialCode": b.material_code or "",
        "materialName": b.material_name or "",
        "specModel": b.spec_model or "",
        "unit": b.unit or "",
        "stockInNo": b.stock_in_no or "",
        "stockInDate": _fmt_date(b.stock_in_date),
        "expireDate": _fmt_date(b.expire_date),
        "catalogNo": b.catalog_no or "",
        "batchNo": b.batch_no or "",
        "manufacturer": b.manufacturer or "",
        "unitPrice": float(b.unit_price) if b.unit_price is not None else 0,
        "amount": float(b.amount) if b.amount is not None else 0,
        "inQty": _dec(b.in_qty),
        "remainingQty": _dec(b.remaining_qty),
        "qcAttachmentUrls": b.qc_attachment_urls or "",
        "stockQty": stock_qty,
        "warningQty": warning_qty,
        "status": status,
        "remark": b.remark or "",
        "createTime": _fmt_dt(b.create_time),
        "updateTime": _fmt_dt(b.update_time),
    }


def _usage_batch_snapshots(
    batch: BizQcMaterialStockBatch,
    *,
    out_qty: Optional[Decimal] = None,
) -> Dict[str, Any]:
    unit_price = batch.unit_price
    if out_qty is not None:
        amount = (Decimal(str(unit_price or 0)) * abs(out_qty)).quantize(Decimal("0.01"))
    else:
        amount = batch.amount
    return {
        "batch_id": batch.id,
        "stock_in_no": batch.stock_in_no,
        "catalog_no": batch.catalog_no,
        "batch_no": batch.batch_no,
        "manufacturer": batch.manufacturer,
        "unit_price": unit_price,
        "amount": amount,
    }


def _get_material_for_update(
    db: Session, material_id: str, tenant_no: str
) -> BizQcMaterial:
    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("物料不存在")
    return obj


def list_batches(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    material_id: Optional[str] = None,
    status: Optional[str] = None,
    stock_status: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = (
            db.query(BizQcMaterialStockBatch, BizQcMaterial)
            .join(
                BizQcMaterial,
                BizQcMaterial.id == BizQcMaterialStockBatch.material_id,
            )
            .filter(
                BizQcMaterialStockBatch.del_flag == 0,
                BizQcMaterial.del_flag == 0,
            )
        )
        if tenant_no:
            q0 = q0.filter(BizQcMaterialStockBatch.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizQcMaterialStockBatch.material_code.like(like),
                    BizQcMaterialStockBatch.material_name.like(like),
                    BizQcMaterialStockBatch.stock_in_no.like(like),
                    BizQcMaterialStockBatch.batch_no.like(like),
                    BizQcMaterialStockBatch.catalog_no.like(like),
                )
            )
        if material_id:
            q0 = q0.filter(BizQcMaterialStockBatch.material_id == material_id)
        st = (status or "").strip()
        if st in ("0", "1"):
            q0 = q0.filter(BizQcMaterial.status == st)
        mode = normalize_batch_stock_status(stock_status)
        if mode == "available":
            q0 = q0.filter(BizQcMaterialStockBatch.remaining_qty > 0)
        elif mode == "exhausted":
            q0 = q0.filter(BizQcMaterialStockBatch.remaining_qty == 0)
        if user_info:
            from app.core.data_scope import apply_data_scope

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

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


def batch_options(
    db: Session,
    tenant_no: str,
    *,
    keyword: Optional[str] = None,
    material_id: Optional[str] = None,
    only_available: bool = True,
) -> List[Dict[str, Any]]:
    q = (
        db.query(BizQcMaterialStockBatch, BizQcMaterial)
        .join(BizQcMaterial, BizQcMaterial.id == BizQcMaterialStockBatch.material_id)
        .filter(
            BizQcMaterialStockBatch.tenant_no == tenant_no,
            BizQcMaterialStockBatch.del_flag == 0,
            BizQcMaterial.del_flag == 0,
            BizQcMaterial.status == "1",
        )
    )
    if only_available:
        q = q.filter(BizQcMaterialStockBatch.remaining_qty > 0)
    if material_id:
        q = q.filter(BizQcMaterialStockBatch.material_id == material_id)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                BizQcMaterialStockBatch.stock_in_no.like(like),
                BizQcMaterialStockBatch.material_name.like(like),
                BizQcMaterialStockBatch.material_code.like(like),
                BizQcMaterialStockBatch.batch_no.like(like),
                BizQcMaterialStockBatch.catalog_no.like(like),
            )
        )
    rows = q.order_by(BizQcMaterialStockBatch.stock_in_date.asc()).all()
    out: List[Dict[str, Any]] = []
    for b, m in rows:
        unit = b.unit or ""
        rem = _dec(b.remaining_qty)
        mfr = (b.manufacturer or "").strip()
        cat = (b.catalog_no or "").strip()
        label = (
            f"{b.stock_in_no} / {b.material_name}"
            f" / 厂家:{mfr or '—'}"
            f" / 批号:{b.batch_no or '—'}"
            f" / 货号:{cat or '—'}"
            f" / 剩:{rem}{unit}"
        )
        unit_price = float(b.unit_price) if b.unit_price is not None else 0
        out.append(
            {
                "id": b.id,
                "batchId": b.id,
                "stockInNo": b.stock_in_no or "",
                "stockInDate": _fmt_date(b.stock_in_date),
                "expireDate": _fmt_date(b.expire_date),
                "materialId": b.material_id,
                "materialCode": b.material_code or "",
                "materialName": b.material_name or "",
                "specModel": b.spec_model or "",
                "unit": unit,
                "batchNo": b.batch_no or "",
                "catalogNo": b.catalog_no or "",
                "manufacturer": b.manufacturer or "",
                "unitPrice": unit_price,
                "amount": float(b.amount) if b.amount is not None else 0,
                "remainingQty": rem,
                "label": label,
            }
        )
    return out


def _create_batch_and_usage(
    db: Session,
    *,
    material: BizQcMaterial,
    stock_in_no: str,
    stock_in_date: date,
    expire_date: date,
    catalog_no: str,
    batch_no: str,
    manufacturer: str,
    unit_price: Decimal,
    amount: Decimal,
    qty: Decimal,
    remark: Optional[str],
    tenant_no: str,
    user_id: Optional[str],
    operate_time: datetime,
    commit: bool = False,
) -> BizQcMaterialStockBatch:
    if qty <= 0:
        raise ValueError("入库数量必须大于0")
    sno = _require_str(stock_in_no, "入库编号")
    if stock_in_no_exists(db, sno, tenant_no):
        raise ValueError(f"入库编号「{sno}」已存在")

    if not _is_enabled_status(material.status):
        raise ValueError(f"物料「{material.material_name}」已停用，无法入库")

    old_stock = material.stock_qty or Decimal("0")
    new_stock = old_stock + qty
    material.stock_qty = new_stock
    material.update_by = user_id
    material.update_time = datetime.now()
    db.add(material)

    batch = BizQcMaterialStockBatch(
        id=generate_uuid(),
        material_id=material.id,
        material_code=material.material_code,
        material_name=material.material_name,
        spec_model=material.spec_model,
        unit=material.unit,
        stock_in_no=sno,
        stock_in_date=stock_in_date,
        expire_date=expire_date,
        catalog_no=catalog_no,
        batch_no=batch_no,
        manufacturer=manufacturer,
        unit_price=unit_price,
        amount=amount,
        in_qty=qty,
        remaining_qty=qty,
        tenant_no=tenant_no,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=material.create_dept_id or "",
        create_time=operate_time,
        update_by=user_id,
        update_time=operate_time,
        del_flag=0,
    )
    db.add(batch)

    snaps = _usage_batch_snapshots(batch)
    usage = BizQcMaterialUsage(
        id=generate_uuid(),
        material_id=material.id,
        material_code=material.material_code,
        material_name=material.material_name,
        change_type="in",
        change_qty=qty,
        stock_before=old_stock,
        stock_after=new_stock,
        source_type="manual_in",
        tenant_no=tenant_no,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=material.create_dept_id or "",
        create_time=operate_time,
        update_by=user_id,
        update_time=operate_time,
        del_flag=0,
        **snaps,
    )
    db.add(usage)
    if commit:
        db.commit()
        db.refresh(batch)
    return batch


def stock_in_one(
    db: Session,
    material_id: str,
    *,
    stock_in_no: str,
    stock_in_date: Any,
    expire_date: Any,
    catalog_no: str,
    batch_no: str,
    manufacturer: str,
    unit_price: Any,
    amount: Any,
    qty: Decimal,
    remark: Optional[str] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
    operate_time: Optional[datetime] = None,
) -> BizQcMaterialStockBatch:
    op_time = operate_time or datetime.now()
    material = _get_material_for_update(db, material_id, tenant_no)
    batch = _create_batch_and_usage(
        db,
        material=material,
        stock_in_no=stock_in_no,
        stock_in_date=_parse_date(stock_in_date),
        expire_date=_parse_expire_date(expire_date),
        catalog_no=_require_str(catalog_no, "货号"),
        batch_no=_require_str(batch_no, "批号"),
        manufacturer=_require_str(manufacturer, "厂家"),
        unit_price=_require_decimal(unit_price, "单价"),
        amount=_require_decimal(amount, "金额"),
        qty=qty,
        remark=remark,
        tenant_no=tenant_no,
        user_id=user_id,
        operate_time=op_time,
        commit=True,
    )
    return batch


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
    seen_nos: set = set()

    try:
        for it in items:
            material_id = (it.get("material_id") or it.get("materialId") or "").strip()
            if not material_id:
                raise ValueError("请选择物资")
            sno = _require_str(it.get("stock_in_no") or it.get("stockInNo"), "入库编号")
            if sno in seen_nos:
                raise ValueError(f"入库编号「{sno}」在本单中重复")
            seen_nos.add(sno)
            qty = it.get("qty")
            if qty is None or Decimal(str(qty)) <= 0:
                raise ValueError("入库数量必须大于0")
            qty_dec = Decimal(str(qty))
            line_remark = (it.get("remark") or "").strip() or shared_remark

            material = _get_material_for_update(db, material_id, tenant_no)
            _create_batch_and_usage(
                db,
                material=material,
                stock_in_no=sno,
                stock_in_date=_parse_date(it.get("stock_in_date") or it.get("stockInDate")),
                expire_date=_parse_expire_date(it.get("expire_date") or it.get("expireDate")),
                catalog_no=_require_str(it.get("catalog_no") or it.get("catalogNo"), "货号"),
                batch_no=_require_str(it.get("batch_no") or it.get("batchNo"), "批号"),
                manufacturer=_require_str(it.get("manufacturer"), "厂家"),
                unit_price=_require_decimal(it.get("unit_price") or it.get("unitPrice"), "单价"),
                amount=_require_decimal(it.get("amount"), "金额"),
                qty=qty_dec,
                remark=line_remark,
                tenant_no=tenant_no,
                user_id=user_id,
                operate_time=op_time,
                commit=False,
            )
        db.commit()
        return len(items)
    except Exception:
        db.rollback()
        raise


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

    batch = (
        db.query(BizQcMaterialStockBatch)
        .filter(
            BizQcMaterialStockBatch.id == batch_id,
            BizQcMaterialStockBatch.tenant_no == tenant_no,
            BizQcMaterialStockBatch.del_flag == 0,
        )
        .with_for_update()
        .first()
    )
    if not batch:
        raise ValueError("入库批次不存在")

    material = _get_material_for_update(db, batch.material_id, tenant_no)
    if not _is_enabled_status(material.status):
        raise ValueError(f"物料「{material.material_name}」已停用，无法出库")

    rem = batch.remaining_qty or Decimal("0")
    if rem < qty:
        raise ValueError(
            f"批次「{batch.stock_in_no}」库存不足，当前剩余 {rem}，需出库 {qty}"
        )

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

    stock_before = material.stock_qty or Decimal("0")
    new_stock = stock_before - qty
    material.stock_qty = new_stock
    material.update_by = user_id
    material.update_time = datetime.now()
    db.add(material)

    snaps = _usage_batch_snapshots(batch, out_qty=qty)
    unit_label = (usage_unit or "").strip() or (batch.unit or "").strip() or None
    usage = BizQcMaterialUsage(
        id=generate_uuid(),
        material_id=material.id,
        material_code=material.material_code,
        material_name=material.material_name,
        change_type="use",
        change_qty=-qty,
        usage_unit=unit_label,
        stock_before=stock_before,
        stock_after=new_stock,
        source_type="manual_out",
        tenant_no=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=material.create_dept_id or "",
        create_time=operate_time,
        update_by=user_id,
        update_time=operate_time,
        del_flag=0,
        **snaps,
    )
    db.add(usage)


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("请至少添加一条出库明细")
    op_time = operate_time or datetime.now()
    shared_remark = (remark or "").strip() or None
    seen_batches: set = set()

    try:
        for it in items:
            batch_id = (it.get("batch_id") or it.get("batchId") or "").strip()
            if not batch_id:
                raise ValueError("请选择入库批次")
            if batch_id in seen_batches:
                raise ValueError("同一入库批次不可重复添加")
            seen_batches.add(batch_id)
            qty = it.get("qty")
            if qty is None or Decimal(str(qty)) <= 0:
                raise ValueError("出库数量必须大于0")
            qty_dec = Decimal(str(qty))
            line_remark = (it.get("remark") or "").strip() or shared_remark
            usage_unit = (it.get("usageUnit") or it.get("usage_unit") or "").strip() or None
            _stock_out_batch(
                db,
                batch_id,
                qty=qty_dec,
                receiver=receiver,
                purpose=purpose,
                remark=line_remark,
                usage_unit=usage_unit,
                tenant_no=tenant_no,
                user_id=user_id,
                operate_time=op_time,
            )
        db.commit()
        return len(items)
    except Exception:
        db.rollback()
        raise


DISCARD_REASONS = frozenset({"过期", "质量问题", "污染/破损", "其他"})


def _validate_discard_reason(discard_reason: Optional[str], remark: Optional[str]) -> str:
    reason = (discard_reason or "").strip()
    if reason not in DISCARD_REASONS:
        raise ValueError("请选择有效的废弃原因")
    if reason == "其他" and not (remark or "").strip():
        raise ValueError("选择「其他」时请填写备注说明")
    return reason


def _discard_batch(
    db: Session,
    batch_id: str,
    *,
    qty: Decimal,
    discard_reason: str,
    remark: Optional[str],
    usage_unit: Optional[str] = None,
    tenant_no: str,
    user_id: Optional[str],
    operate_time: datetime,
) -> None:
    if qty <= 0:
        raise ValueError("废弃数量必须大于0")

    batch = (
        db.query(BizQcMaterialStockBatch)
        .filter(
            BizQcMaterialStockBatch.id == batch_id,
            BizQcMaterialStockBatch.tenant_no == tenant_no,
            BizQcMaterialStockBatch.del_flag == 0,
        )
        .with_for_update()
        .first()
    )
    if not batch:
        raise ValueError("入库批次不存在")

    material = _get_material_for_update(db, batch.material_id, tenant_no)
    if not _is_enabled_status(material.status):
        raise ValueError(f"物料「{material.material_name}」已停用，无法废弃")

    rem = batch.remaining_qty or Decimal("0")
    if rem < qty:
        raise ValueError(
            f"批次「{batch.stock_in_no}」库存不足，当前剩余 {rem}，需废弃 {qty}"
        )

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

    stock_before = material.stock_qty or Decimal("0")
    new_stock = stock_before - qty
    material.stock_qty = new_stock
    material.update_by = user_id
    material.update_time = datetime.now()
    db.add(material)

    snaps = _usage_batch_snapshots(batch, out_qty=qty)
    unit_label = (usage_unit or "").strip() or (batch.unit or "").strip() or None
    usage = BizQcMaterialUsage(
        id=generate_uuid(),
        material_id=material.id,
        material_code=material.material_code,
        material_name=material.material_name,
        change_type="discard",
        change_qty=-qty,
        usage_unit=unit_label,
        stock_before=stock_before,
        stock_after=new_stock,
        source_type="manual_discard",
        tenant_no=tenant_no,
        purpose=discard_reason,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=material.create_dept_id or "",
        create_time=operate_time,
        update_by=user_id,
        update_time=operate_time,
        del_flag=0,
        **snaps,
    )
    db.add(usage)


def batch_discard(
    db: Session,
    *,
    items: List[Dict[str, Any]],
    discard_reason: 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("请至少添加一条废弃明细")
    op_time = operate_time or datetime.now()
    shared_remark = (remark or "").strip() or None
    reason = _validate_discard_reason(discard_reason, shared_remark)
    seen_batches: set = set()

    try:
        for it in items:
            batch_id = (it.get("batch_id") or it.get("batchId") or "").strip()
            if not batch_id:
                raise ValueError("请选择入库批次")
            if batch_id in seen_batches:
                raise ValueError("同一入库批次不可重复添加")
            seen_batches.add(batch_id)
            qty = it.get("qty")
            if qty is None or Decimal(str(qty)) <= 0:
                raise ValueError("废弃数量必须大于0")
            qty_dec = Decimal(str(qty))
            line_remark = (it.get("remark") or "").strip() or shared_remark
            if reason == "其他" and not line_remark:
                raise ValueError("选择「其他」时请填写备注说明")
            usage_unit = (it.get("usageUnit") or it.get("usage_unit") or "").strip() or None
            _discard_batch(
                db,
                batch_id,
                qty=qty_dec,
                discard_reason=reason,
                remark=line_remark,
                usage_unit=usage_unit,
                tenant_no=tenant_no,
                user_id=user_id,
                operate_time=op_time,
            )
        db.commit()
        return len(items)
    except Exception:
        db.rollback()
        raise


def get_batch(db: Session, batch_id: str, tenant_no: str) -> Optional[BizQcMaterialStockBatch]:
    return (
        db.query(BizQcMaterialStockBatch)
        .filter(
            BizQcMaterialStockBatch.id == batch_id,
            BizQcMaterialStockBatch.tenant_no == tenant_no,
            BizQcMaterialStockBatch.del_flag == 0,
        )
        .first()
    )


def _sum_batch_out_total(
    db: Session, batch_id: str, tenant_no: str
) -> Tuple[Decimal, int]:
    row = (
        db.query(
            func.coalesce(func.sum(func.abs(BizQcMaterialUsage.change_qty)), 0),
            func.count(BizQcMaterialUsage.id),
        )
        .filter(
            BizQcMaterialUsage.batch_id == batch_id,
            BizQcMaterialUsage.tenant_no == tenant_no,
            BizQcMaterialUsage.del_flag == 0,
            BizQcMaterialUsage.change_type.in_(["use", "discard"]),
        )
        .one()
    )
    total = Decimal(str(row[0] or 0))
    count = int(row[1] or 0)
    return total, count


def get_batch_detail(
    db: Session, batch_id: str, tenant_no: str
) -> Dict[str, Any]:
    batch = get_batch(db, batch_id, tenant_no)
    if not batch:
        raise ValueError("入库批次不存在")
    material = (
        db.query(BizQcMaterial)
        .filter(
            BizQcMaterial.id == batch.material_id,
            BizQcMaterial.tenant_no == tenant_no,
            BizQcMaterial.del_flag == 0,
        )
        .first()
    )
    out_total, out_count = _sum_batch_out_total(db, batch_id, tenant_no)
    payload = _batch_row_to_dict(batch, material)
    payload["outTotal"] = _dec(out_total)
    payload["outCount"] = out_count
    return payload


def _append_batch_qty_adjust(
    db: Session,
    *,
    batch: BizQcMaterialStockBatch,
    material: BizQcMaterial,
    change_qty: Decimal,
    stock_before: Decimal,
    stock_after: Decimal,
    remark: str,
    tenant_no: str,
    user_id: Optional[str],
) -> None:
    if change_qty == 0:
        return
    now = datetime.now()
    snaps = _usage_batch_snapshots(batch)
    db.add(
        BizQcMaterialUsage(
            id=generate_uuid(),
            material_id=material.id,
            material_code=material.material_code,
            material_name=material.material_name,
            change_type="adjust",
            change_qty=change_qty,
            stock_before=stock_before,
            stock_after=stock_after,
            source_type="batch_qty_edit",
            tenant_no=tenant_no,
            remark=remark,
            create_by=user_id,
            create_dept_id=material.create_dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
            **snaps,
        )
    )


def update_batch(
    db: Session,
    batch_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    stock_in_no: Optional[str] = None,
    stock_in_date: Optional[Any] = None,
    expire_date: Optional[Any] = None,
    catalog_no: Optional[str] = None,
    batch_no: Optional[str] = None,
    manufacturer: Optional[str] = None,
    unit_price: Optional[Any] = None,
    amount: Optional[Any] = None,
    in_qty: Optional[Any] = None,
    remaining_qty: Optional[Any] = None,
    remark: Optional[str] = None,
    qc_attachment_urls: Optional[str] = None,
    material_code: Optional[str] = None,
    material_name: Optional[str] = None,
    spec_model: Optional[str] = None,
    unit: Optional[str] = None,
    warning_qty: Optional[Any] = None,
    status: Optional[str] = None,
) -> BizQcMaterialStockBatch:
    from app.crud import qc_material as mat_svc

    batch = (
        db.query(BizQcMaterialStockBatch)
        .filter(
            BizQcMaterialStockBatch.id == batch_id,
            BizQcMaterialStockBatch.tenant_no == tenant_no,
            BizQcMaterialStockBatch.del_flag == 0,
        )
        .with_for_update()
        .first()
    )
    if not batch:
        raise ValueError("入库批次不存在")

    material = _get_material_for_update(db, batch.material_id, tenant_no)
    mat_data: Dict[str, Any] = {}
    if material_code is not None:
        mat_data["material_code"] = material_code
    if material_name is not None:
        mat_data["material_name"] = material_name
    if spec_model is not None:
        mat_data["spec_model"] = spec_model
    if unit is not None:
        mat_data["unit"] = unit
    if warning_qty is not None:
        mat_data["warning_qty"] = Decimal(str(warning_qty))
    if status is not None:
        mat_data["status"] = _norm_status(status)
    if mat_data:
        material = mat_svc.update_material(
            db,
            batch.material_id,
            tenant_no=tenant_no,
            user_id=user_id,
            **mat_data,
        )

    if stock_in_no is not None:
        sno = _require_str(stock_in_no, "入库编号")
        if stock_in_no_exists(db, sno, tenant_no, exclude_id=batch_id):
            raise ValueError(f"入库编号「{sno}」已存在")
        batch.stock_in_no = sno
    if stock_in_date is not None:
        batch.stock_in_date = _parse_date(stock_in_date)
    if expire_date is not None:
        batch.expire_date = _parse_expire_date(expire_date)
    if catalog_no is not None:
        batch.catalog_no = _require_str(catalog_no, "货号")
    if batch_no is not None:
        batch.batch_no = _require_str(batch_no, "批号")
    if manufacturer is not None:
        batch.manufacturer = _require_str(manufacturer, "厂家")
    if unit_price is not None:
        batch.unit_price = _require_decimal(unit_price, "单价")
    if amount is not None:
        batch.amount = _require_decimal(amount, "金额")
    if remark is not None:
        batch.remark = (remark or "").strip() or None
    if qc_attachment_urls is not None:
        batch.qc_attachment_urls = (qc_attachment_urls or "").strip() or None

    old_in_qty = batch.in_qty or Decimal("0")
    old_remaining = batch.remaining_qty or Decimal("0")
    qty_updated = in_qty is not None or remaining_qty is not None
    qty_values_changed = False

    if qty_updated:
        new_in_qty = _require_decimal(in_qty, "入库数量") if in_qty is not None else old_in_qty
        new_remaining = (
            _require_decimal(remaining_qty, "剩余数量")
            if remaining_qty is not None
            else old_remaining
        )
        qty_values_changed = new_in_qty != old_in_qty or new_remaining != old_remaining

        if qty_values_changed:
            if new_in_qty <= 0:
                raise ValueError("入库数量必须大于0")
            if new_remaining < 0:
                raise ValueError("剩余数量不能为负数")
            if new_remaining > new_in_qty:
                raise ValueError("剩余数量不能大于入库数量")

            out_total, _ = _sum_batch_out_total(db, batch_id, tenant_no)
            if new_in_qty < out_total:
                raise ValueError(f"入库数量不能小于已出库量 {_dec(out_total)}")

            delta_remaining = new_remaining - old_remaining

            if in_qty is not None and amount is None:
                unit_p = batch.unit_price or Decimal("0")
                batch.amount = (unit_p * new_in_qty).quantize(Decimal("0.01"))

            stock_before = material.stock_qty or Decimal("0")
            material.stock_qty = stock_before + delta_remaining
            material.update_by = user_id
            material.update_time = datetime.now()
            db.add(material)

            batch.in_qty = new_in_qty
            batch.remaining_qty = new_remaining

            in_qty_delta = new_in_qty - old_in_qty
            remaining_extra = new_remaining - (old_remaining + in_qty_delta)
            running_stock = stock_before
            user_remark = (remark or "").strip() if remark is not None else ""

            if in_qty_delta != 0:
                next_stock = running_stock + in_qty_delta
                adjust_remark = (
                    f"编辑修正入库数量（{_dec(old_in_qty)} → {_dec(new_in_qty)}）"
                )
                if user_remark:
                    adjust_remark = f"{adjust_remark}；{user_remark}"
                _append_batch_qty_adjust(
                    db,
                    batch=batch,
                    material=material,
                    change_qty=in_qty_delta,
                    stock_before=running_stock,
                    stock_after=next_stock,
                    remark=adjust_remark,
                    tenant_no=tenant_no,
                    user_id=user_id,
                )
                running_stock = next_stock

            if remaining_extra != 0:
                expected_remaining = old_remaining + in_qty_delta
                next_stock = running_stock + remaining_extra
                adjust_remark = (
                    f"编辑修正剩余数量（{_dec(expected_remaining)} → {_dec(new_remaining)}）"
                )
                if user_remark and in_qty_delta == 0:
                    adjust_remark = f"{adjust_remark}；{user_remark}"
                _append_batch_qty_adjust(
                    db,
                    batch=batch,
                    material=material,
                    change_qty=remaining_extra,
                    stock_before=running_stock,
                    stock_after=next_stock,
                    remark=adjust_remark,
                    tenant_no=tenant_no,
                    user_id=user_id,
                )

    batch.material_code = material.material_code
    batch.material_name = material.material_name
    batch.spec_model = material.spec_model
    batch.unit = material.unit
    batch.update_by = user_id
    batch.update_time = datetime.now()
    db.add(batch)

    in_usage = (
        db.query(BizQcMaterialUsage)
        .filter(
            BizQcMaterialUsage.batch_id == batch_id,
            BizQcMaterialUsage.tenant_no == tenant_no,
            BizQcMaterialUsage.del_flag == 0,
            BizQcMaterialUsage.change_type == "in",
            BizQcMaterialUsage.source_type == "manual_in",
        )
        .first()
    )
    if in_usage:
        in_usage.material_code = material.material_code
        in_usage.material_name = material.material_name
        in_usage.stock_in_no = batch.stock_in_no
        in_usage.catalog_no = batch.catalog_no
        in_usage.batch_no = batch.batch_no
        in_usage.manufacturer = batch.manufacturer
        in_usage.unit_price = batch.unit_price
        if not qty_values_changed:
            in_usage.amount = batch.amount
        in_usage.remark = batch.remark
        in_usage.update_by = user_id
        in_usage.update_time = datetime.now()
        db.add(in_usage)

    db.commit()
    db.refresh(batch)
    return batch


def delete_batch(
    db: Session,
    batch_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    batch = get_batch(db, batch_id, tenant_no)
    if not batch:
        raise ValueError("入库批次不存在")

    used = (
        db.query(BizQcMaterialUsage)
        .filter(
            BizQcMaterialUsage.batch_id == batch_id,
            BizQcMaterialUsage.tenant_no == tenant_no,
            BizQcMaterialUsage.del_flag == 0,
            BizQcMaterialUsage.change_type == "use",
        )
        .count()
    )
    if used > 0:
        raise ValueError("该批次已有领用或出库记录，无法删除")

    material = _get_material_for_update(db, batch.material_id, tenant_no)
    rem = batch.remaining_qty or Decimal("0")
    material.stock_qty = (material.stock_qty or Decimal("0")) - rem
    material.update_by = user_id
    material.update_time = datetime.now()
    db.add(material)

    now = datetime.now()
    batch.del_flag = 1
    batch.update_by = user_id
    batch.update_time = now
    db.add(batch)

    usages = (
        db.query(BizQcMaterialUsage)
        .filter(
            BizQcMaterialUsage.batch_id == batch_id,
            BizQcMaterialUsage.tenant_no == tenant_no,
            BizQcMaterialUsage.del_flag == 0,
        )
        .all()
    )
    for u in usages:
        u.del_flag = 1
        u.update_by = user_id
        u.update_time = now
        db.add(u)

    db.commit()