zkxw/backend/app/crud/material_stock_check.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.crud.material_stock_batch import _dec, _fmt_date, _fmt_dt, _parse_date, _require_str
from app.models.models import SysUser
from app.models.material import (
    BizMaterialStockBatch,
    BizMaterialStockCheck,
    BizMaterialStockCheckBatchLine,
    BizMaterialStockCheckUnusedLine,
)


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 _status_label(status: str) -> str:
    if status == "completed":
        return "已完成"
    if status == "cancelled":
        return "已作废"
    return "进行中"


def _batch_line_to_dict(row: BizMaterialStockCheckBatchLine) -> Dict[str, Any]:
    return {
        "id": row.id,
        "checkId": row.check_id,
        "batchId": row.batch_id,
        "materialId": row.material_id,
        "materialCode": row.material_code or "",
        "materialName": row.material_name or "",
        "specModel": row.spec_model or "",
        "unit": row.unit or "",
        "stockInNo": row.stock_in_no or "",
        "stockInDate": _fmt_date(row.stock_in_date),
        "expireDate": _fmt_date(row.expire_date),
        "catalogNo": row.catalog_no or "",
        "batchNo": row.batch_no or "",
        "manufacturer": row.manufacturer or "",
        "unitPrice": float(row.unit_price) if row.unit_price is not None else 0,
        "amount": float(row.amount) if row.amount is not None else 0,
        "remainingQty": _dec(row.remaining_qty),
        "remainingAmount": _unused_amount(row.unit_price, row.remaining_qty),
        "remark": row.remark or "",
    }


def _unused_amount(unit_price: Any, qty: Any) -> float:
    if unit_price is None:
        return 0
    return float((Decimal(str(unit_price)) * Decimal(str(qty))).quantize(Decimal("0.01")))


def _unused_line_to_dict(row: BizMaterialStockCheckUnusedLine) -> Dict[str, Any]:
    unit_price = float(row.unit_price) if row.unit_price is not None else None
    unused_qty = _dec(row.unused_qty)
    return {
        "id": row.id,
        "checkId": row.check_id,
        "usageId": row.usage_id or "",
        "batchId": row.batch_id or "",
        "materialCode": row.material_code or "",
        "materialName": row.material_name or "",
        "specModel": row.spec_model or "",
        "unit": row.unit or "",
        "stockInNo": row.stock_in_no or "",
        "stockInDate": _fmt_date(row.stock_in_date),
        "expireDate": _fmt_date(row.expire_date),
        "catalogNo": row.catalog_no or "",
        "batchNo": row.batch_no or "",
        "manufacturer": row.manufacturer or "",
        "unitPrice": unit_price,
        "remainingAmount": _unused_amount(row.unit_price, row.unused_qty) if unit_price is not None else None,
        "holder": row.holder or "",
        "unusedQty": unused_qty,
        "remark": row.remark or "",
    }


def _check_header_to_dict(
    row: BizMaterialStockCheck,
    *,
    create_by_name: str = "",
    batch_line_count: int = 0,
    unused_line_count: int = 0,
) -> Dict[str, Any]:
    return {
        "id": row.id,
        "title": row.title or "",
        "checkTime": _fmt_dt(row.check_time),
        "status": row.status or "draft",
        "statusLabel": _status_label(row.status or "draft"),
        "completeTime": _fmt_dt(row.complete_time) if row.complete_time else "",
        "remark": row.remark or "",
        "batchLineCount": batch_line_count,
        "unusedLineCount": unused_line_count,
        "createBy": row.create_by or "",
        "createByName": create_by_name or row.create_by or "",
        "createTime": _fmt_dt(row.create_time),
        "updateTime": _fmt_dt(row.update_time),
    }


def _get_check(db: Session, check_id: str, tenant_no: str) -> Optional[BizMaterialStockCheck]:
    return (
        db.query(BizMaterialStockCheck)
        .filter(
            BizMaterialStockCheck.id == check_id,
            BizMaterialStockCheck.tenant_no == tenant_no,
            BizMaterialStockCheck.del_flag == 0,
        )
        .first()
    )


def _line_counts(db: Session, check_ids: List[str]) -> Dict[str, Tuple[int, int]]:
    if not check_ids:
        return {}
    batch_rows = (
        db.query(
            BizMaterialStockCheckBatchLine.check_id,
            func.count(BizMaterialStockCheckBatchLine.id),
        )
        .filter(
            BizMaterialStockCheckBatchLine.check_id.in_(check_ids),
            BizMaterialStockCheckBatchLine.del_flag == 0,
        )
        .group_by(BizMaterialStockCheckBatchLine.check_id)
        .all()
    )
    unused_rows = (
        db.query(
            BizMaterialStockCheckUnusedLine.check_id,
            func.count(BizMaterialStockCheckUnusedLine.id),
        )
        .filter(
            BizMaterialStockCheckUnusedLine.check_id.in_(check_ids),
            BizMaterialStockCheckUnusedLine.del_flag == 0,
        )
        .group_by(BizMaterialStockCheckUnusedLine.check_id)
        .all()
    )
    batch_map = {cid: int(cnt) for cid, cnt in batch_rows}
    unused_map = {cid: int(cnt) for cid, cnt in unused_rows}
    out: Dict[str, Tuple[int, int]] = {}
    for cid in check_ids:
        out[cid] = (batch_map.get(cid, 0), unused_map.get(cid, 0))
    return out


def list_checks(
    db: Session,
    tenant_no: str,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    start_time: Optional[str] = None,
    end_time: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizMaterialStockCheck).filter(
            BizMaterialStockCheck.tenant_no == tenant_no,
            BizMaterialStockCheck.del_flag == 0,
        )
        kw = (keyword or "").strip()
        if kw:
            q0 = q0.filter(BizMaterialStockCheck.title.like(f"%{kw}%"))
        st = (status or "").strip()
        if st:
            q0 = q0.filter(BizMaterialStockCheck.status == st)
        if start_time:
            q0 = q0.filter(BizMaterialStockCheck.check_time >= start_time)
        if end_time:
            q0 = q0.filter(BizMaterialStockCheck.check_time <= end_time + " 23:59:59")
        if user_info:
            from app.core.data_scope import apply_data_scope

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

    total = _filtered().order_by(None).with_entities(func.count(BizMaterialStockCheck.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizMaterialStockCheck.check_time.desc(), BizMaterialStockCheck.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    ids = [r.id for r in rows]
    counts = _line_counts(db, ids)
    name_map = _user_nick_name_map(db, [r.create_by for r in rows if r.create_by])
    items = []
    for r in rows:
        bc, uc = counts.get(r.id, (0, 0))
        items.append(
            _check_header_to_dict(
                r,
                create_by_name=name_map.get(r.create_by or "", ""),
                batch_line_count=bc,
                unused_line_count=uc,
            )
        )
    return items, int(total)


def get_check_detail(db: Session, check_id: str, tenant_no: str) -> Dict[str, Any]:
    row = _get_check(db, check_id, tenant_no)
    if not row:
        raise ValueError("盘点单不存在")
    bc, uc = _line_counts(db, [check_id]).get(check_id, (0, 0))
    name_map = _user_nick_name_map(db, [row.create_by] if row.create_by else [])
    batch_lines = (
        db.query(BizMaterialStockCheckBatchLine)
        .filter(
            BizMaterialStockCheckBatchLine.check_id == check_id,
            BizMaterialStockCheckBatchLine.del_flag == 0,
        )
        .order_by(BizMaterialStockCheckBatchLine.stock_in_no.asc())
        .all()
    )
    unused_lines = (
        db.query(BizMaterialStockCheckUnusedLine)
        .filter(
            BizMaterialStockCheckUnusedLine.check_id == check_id,
            BizMaterialStockCheckUnusedLine.del_flag == 0,
        )
        .order_by(BizMaterialStockCheckUnusedLine.create_time.asc())
        .all()
    )
    data = _check_header_to_dict(
        row,
        create_by_name=name_map.get(row.create_by or "", ""),
        batch_line_count=bc,
        unused_line_count=uc,
    )
    data["batchLines"] = [_batch_line_to_dict(x) for x in batch_lines]
    data["unusedLines"] = [_unused_line_to_dict(x) for x in unused_lines]
    return data


def _parse_check_time(v: Any) -> datetime:
    if isinstance(v, datetime):
        return v
    s = str(v or "").strip()
    if not s:
        return datetime.now()
    for fmt, length in (("%Y-%m-%d %H:%M:%S", 19), ("%Y-%m-%d", 10)):
        try:
            return datetime.strptime(s[:length], fmt)
        except ValueError:
            continue
    raise ValueError("盘点时间格式无效")


def create_check(
    db: Session,
    *,
    title: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    dept_id: str = "",
    check_time: Optional[Any] = None,
    remark: Optional[str] = None,
) -> Dict[str, Any]:
    t = _require_str(title, "盘点标题")
    op_time = datetime.now()
    ck_time = _parse_check_time(check_time) if check_time else op_time

    check = BizMaterialStockCheck(
        id=generate_uuid(),
        title=t,
        check_time=ck_time,
        status="draft",
        tenant_no=tenant_no,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=dept_id or "",
        create_time=op_time,
        update_by=user_id,
        update_time=op_time,
        del_flag=0,
    )
    db.add(check)

    batches = (
        db.query(BizMaterialStockBatch)
        .filter(
            BizMaterialStockBatch.tenant_no == tenant_no,
            BizMaterialStockBatch.del_flag == 0,
            BizMaterialStockBatch.remaining_qty > 0,
        )
        .order_by(BizMaterialStockBatch.stock_in_date.desc(), BizMaterialStockBatch.create_time.desc())
        .all()
    )
    for b in batches:
        line = BizMaterialStockCheckBatchLine(
            id=generate_uuid(),
            check_id=check.id,
            batch_id=b.id,
            material_id=b.material_id,
            material_code=b.material_code,
            material_name=b.material_name,
            spec_model=b.spec_model,
            unit=b.unit,
            stock_in_no=b.stock_in_no,
            stock_in_date=b.stock_in_date,
            expire_date=b.expire_date,
            catalog_no=b.catalog_no,
            batch_no=b.batch_no,
            manufacturer=b.manufacturer,
            unit_price=b.unit_price,
            amount=b.amount,
            remaining_qty=b.remaining_qty,
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id=dept_id or "",
            create_time=op_time,
            update_by=user_id,
            update_time=op_time,
            del_flag=0,
        )
        db.add(line)

    db.commit()
    db.refresh(check)
    return {
        "id": check.id,
        "batchLineCount": len(batches),
    }


def _parse_optional_date(v: Any) -> Optional[Any]:
    if v is None or str(v).strip() == "":
        return None
    return _parse_date(v, label="日期")


def _resolve_unused_item_fields(
    db: Session,
    it: Dict[str, Any],
    tenant_no: str,
) -> Dict[str, Any]:
    """根据 usage_id / batch_id 补全登记行快照字段。"""
    usage_id = (it.get("usage_id") or it.get("usageId") or "").strip()
    batch_id = (it.get("batch_id") or it.get("batchId") or "").strip()
    out = dict(it)

    if usage_id:
        from app.models.material import BizMaterialUsage, BizMaterialStockBatch

        usage = (
            db.query(BizMaterialUsage)
            .filter(
                BizMaterialUsage.id == usage_id,
                BizMaterialUsage.tenant_no == tenant_no,
                BizMaterialUsage.del_flag == 0,
                BizMaterialUsage.source_type == "manual_out",
            )
            .first()
        )
        if not usage:
            raise ValueError("关联领用记录不存在")
        batch = None
        if usage.batch_id:
            batch = (
                db.query(BizMaterialStockBatch)
                .filter(
                    BizMaterialStockBatch.id == usage.batch_id,
                    BizMaterialStockBatch.tenant_no == tenant_no,
                    BizMaterialStockBatch.del_flag == 0,
                )
                .first()
            )
        out.setdefault("material_code", usage.material_code or "")
        out.setdefault("material_name", usage.material_name or "")
        out.setdefault("spec_model", (batch.spec_model if batch else "") or "")
        out.setdefault("unit", (batch.unit if batch else "") or usage.usage_unit or "")
        out.setdefault("stock_in_no", usage.stock_in_no or "")
        out.setdefault("stock_in_date", _fmt_date(batch.stock_in_date) if batch and batch.stock_in_date else "")
        out.setdefault("expire_date", _fmt_date(batch.expire_date) if batch and batch.expire_date else "")
        out.setdefault("catalog_no", usage.catalog_no or "")
        out.setdefault("batch_no", usage.batch_no or "")
        out.setdefault("manufacturer", usage.manufacturer or "")
        if out.get("unit_price") is None and out.get("unitPrice") is None:
            up = float(usage.unit_price) if usage.unit_price is not None else None
            if up is not None:
                out["unit_price"] = up
        rcv = (usage.receiver or "").strip()
        if rcv and not (out.get("holder") or "").strip():
            out["holder"] = rcv
        if batch and not batch_id:
            out["batch_id"] = batch.id
        return out

    if batch_id:
        from app.models.material import BizMaterialStockBatch

        batch = (
            db.query(BizMaterialStockBatch)
            .filter(
                BizMaterialStockBatch.id == batch_id,
                BizMaterialStockBatch.tenant_no == tenant_no,
                BizMaterialStockBatch.del_flag == 0,
            )
            .first()
        )
        if not batch:
            raise ValueError("关联入库批次不存在")
        out.setdefault("material_code", batch.material_code or "")
        out.setdefault("material_name", batch.material_name or "")
        out.setdefault("spec_model", batch.spec_model or "")
        out.setdefault("unit", batch.unit or "")
        out.setdefault("stock_in_no", batch.stock_in_no or "")
        out.setdefault("stock_in_date", _fmt_date(batch.stock_in_date) if batch.stock_in_date else "")
        out.setdefault("expire_date", _fmt_date(batch.expire_date) if batch.expire_date else "")
        out.setdefault("catalog_no", batch.catalog_no or "")
        out.setdefault("batch_no", batch.batch_no or "")
        out.setdefault("manufacturer", batch.manufacturer or "")
        if out.get("unit_price") is None and out.get("unitPrice") is None:
            out["unit_price"] = float(batch.unit_price) if batch.unit_price is not None else None
        return out

    return out


def save_unused_lines(
    db: Session,
    check_id: str,
    *,
    items: List[Dict[str, Any]],
    tenant_no: str,
    user_id: Optional[str] = None,
    dept_id: str = "",
) -> int:
    check = _get_check(db, check_id, tenant_no)
    if not check:
        raise ValueError("盘点单不存在")
    if (check.status or "") != "draft":
        raise ValueError("仅进行中的盘点单可登记未用完物料")

    now = datetime.now()
    old_rows = (
        db.query(BizMaterialStockCheckUnusedLine)
        .filter(
            BizMaterialStockCheckUnusedLine.check_id == check_id,
            BizMaterialStockCheckUnusedLine.del_flag == 0,
        )
        .all()
    )
    for r in old_rows:
        r.del_flag = 1
        r.update_by = user_id
        r.update_time = now
        db.add(r)

    count = 0
    for it in items or []:
        it = _resolve_unused_item_fields(db, it, tenant_no)
        name = _require_str(it.get("material_name") or it.get("materialName"), "物料名称")
        qty = it.get("unused_qty") if it.get("unused_qty") is not None else it.get("unusedQty")
        if qty is None or Decimal(str(qty)) <= 0:
            raise ValueError("未用完数量必须大于0")
        qty_dec = Decimal(str(qty))
        stock_in_date = _parse_optional_date(it.get("stock_in_date") or it.get("stockInDate"))
        expire_date = _parse_optional_date(it.get("expire_date") or it.get("expireDate"))
        usage_id = (it.get("usage_id") or it.get("usageId") or "").strip() or None
        batch_id = (it.get("batch_id") or it.get("batchId") or "").strip() or None

        row = BizMaterialStockCheckUnusedLine(
            id=generate_uuid(),
            check_id=check_id,
            usage_id=usage_id,
            batch_id=batch_id,
            material_code=(it.get("material_code") or it.get("materialCode") or "").strip() or None,
            material_name=name,
            spec_model=(it.get("spec_model") or it.get("specModel") or "").strip() or None,
            unit=(it.get("unit") or "").strip() or None,
            stock_in_no=(it.get("stock_in_no") or it.get("stockInNo") or "").strip() or None,
            stock_in_date=stock_in_date,
            expire_date=expire_date,
            catalog_no=(it.get("catalog_no") or it.get("catalogNo") or "").strip() or None,
            batch_no=(it.get("batch_no") or it.get("batchNo") or "").strip() or None,
            manufacturer=(it.get("manufacturer") or "").strip() or None,
            unit_price=(
                Decimal(str(it.get("unit_price") or it.get("unitPrice")))
                if (it.get("unit_price") is not None or it.get("unitPrice") is not None)
                else None
            ),
            holder=(it.get("holder") or "").strip() or None,
            unused_qty=qty_dec,
            tenant_no=tenant_no,
            remark=(it.get("remark") or "").strip() or None,
            create_by=user_id,
            create_dept_id=dept_id or "",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(row)
        count += 1

    check.update_by = user_id
    check.update_time = now
    db.add(check)
    db.commit()
    return count


def complete_check(
    db: Session,
    check_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    check = _get_check(db, check_id, tenant_no)
    if not check:
        raise ValueError("盘点单不存在")
    if (check.status or "") != "draft":
        raise ValueError("仅进行中的盘点单可完成")
    now = datetime.now()
    check.status = "completed"
    check.complete_time = now
    check.update_by = user_id
    check.update_time = now
    db.add(check)
    db.commit()


def cancel_check(
    db: Session,
    check_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> None:
    check = _get_check(db, check_id, tenant_no)
    if not check:
        raise ValueError("盘点单不存在")
    if (check.status or "") != "draft":
        raise ValueError("仅进行中的盘点单可作废")
    now = datetime.now()
    check.status = "cancelled"
    check.update_by = user_id
    check.update_time = now
    db.add(check)
    db.commit()