zkxw/backend/app/crud/material_usage.py

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

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

from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.models.material import BizMaterialStockBatch, BizMaterialUsage
from app.models.models import SysUser


def _parse_receiver_purpose_from_remark(remark: Optional[str]) -> Tuple[str, str]:
    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: BizMaterialUsage) -> 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: BizMaterialUsage) -> 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]:
    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: BizMaterialUsage, batch: Optional[BizMaterialStockBatch]) -> 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: BizMaterialUsage, batch: Optional[BizMaterialStockBatch]) -> str:
    if batch and batch.stock_in_date:
        return batch.stock_in_date.strftime("%Y-%m-%d")
    return ""


def _unit_price_for_row(r: BizMaterialUsage, batch: Optional[BizMaterialStockBatch]) -> 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: BizMaterialUsage, batch: Optional[BizMaterialStockBatch]) -> 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: BizMaterialUsage,
    *,
    create_by_name: str = "",
    batch: Optional[BizMaterialStockBatch] = None,
    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)
    return {
        "id": r.id,
        "materialId": r.material_id,
        "materialCode": r.material_code or "",
        "materialName": r.material_name or "",
        "batchId": r.batch_id or "",
        "stockInNo": r.stock_in_no or "",
        "stockInDate": _stock_in_date_for_row(r, batch),
        "catalogNo": r.catalog_no or "",
        "batchNo": r.batch_no or "",
        "manufacturer": r.manufacturer 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,
        "receiver": receiver,
        "receiverName": receiver_name or receiver,
        "operatorName": (getattr(r, "operator_name", None) or "").strip(),
        "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, BizMaterialStockBatch]:
    ids = [x for x in batch_ids if x]
    if not ids:
        return {}
    rows = (
        db.query(BizMaterialStockBatch)
        .filter(BizMaterialStockBatch.id.in_(ids), BizMaterialStockBatch.del_flag == 0)
        .all()
    )
    return {r.id: r for r in rows}


def _list_usages_query(
    db: Session,
    tenant_no: str,
    *,
    material_id: 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,
):
    q0 = db.query(BizMaterialUsage).filter(
        BizMaterialUsage.tenant_no == tenant_no,
        BizMaterialUsage.del_flag == 0,
    )
    if material_id:
        q0 = q0.filter(BizMaterialUsage.material_id == material_id)
    ct = (change_type or "").strip()
    if ct == "out":
        q0 = q0.filter(
            BizMaterialUsage.change_type == "use",
            BizMaterialUsage.source_type == "manual_out",
        )
    elif ct == "in":
        q0 = q0.filter(BizMaterialUsage.change_type == "in")
    elif ct == "discard":
        q0 = q0.filter(BizMaterialUsage.change_type == "discard")
    sno = (stock_in_no or "").strip()
    if sno:
        q0 = q0.filter(BizMaterialUsage.stock_in_no.like(f"%{sno}%"))
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(
                BizMaterialUsage.material_code.like(like),
                BizMaterialUsage.material_name.like(like),
                BizMaterialUsage.stock_in_no.like(like),
                BizMaterialUsage.batch_no.like(like),
                BizMaterialUsage.catalog_no.like(like),
            )
        )
    if start_time:
        q0 = q0.filter(BizMaterialUsage.create_time >= start_time)
    if end_time:
        q0 = q0.filter(BizMaterialUsage.create_time <= end_time + " 23:59:59")
    if user_info:
        from app.core.data_scope import apply_data_scope

        q0 = apply_data_scope(q0, BizMaterialUsage, 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(BizMaterialUsage.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q0.order_by(BizMaterialUsage.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    bmap = _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,
    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,
        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 _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(BizMaterialUsage).filter(
        BizMaterialUsage.tenant_no == tenant_no,
        BizMaterialUsage.del_flag == 0,
        BizMaterialUsage.change_type == "use",
        BizMaterialUsage.source_type == "manual_out",
        BizMaterialUsage.requisition_status == "open",
    )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q0 = apply_data_scope(q0, BizMaterialUsage, user_info, db, tenant_no)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q0 = q0.filter(
            or_(
                BizMaterialUsage.material_code.like(like),
                BizMaterialUsage.material_name.like(like),
                BizMaterialUsage.stock_in_no.like(like),
                BizMaterialUsage.batch_no.like(like),
                BizMaterialUsage.catalog_no.like(like),
                BizMaterialUsage.receiver.like(like),
            )
        )
    total = q0.order_by(None).with_entities(func.count(BizMaterialUsage.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 200)
    rows = (
        q0.order_by(BizMaterialUsage.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)
    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 ""
        out.append(
            {
                "id": r.id,
                "materialId": r.material_id,
                "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 "",
                "unitPrice": _unit_price_for_row(r, batch),
                "unit": unit,
                "requisitionQty": qty_abs,
                "taskConsumedQty": 0,
                "receiver": rcv,
                "receiverName": rcv_name,
                "operatorName": (getattr(r, "operator_name", None) or "").strip(),
                "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(BizMaterialUsage)
        .filter(
            BizMaterialUsage.id == usage_id,
            BizMaterialUsage.tenant_no == tenant_no,
            BizMaterialUsage.del_flag == 0,
            BizMaterialUsage.change_type == "use",
            BizMaterialUsage.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(BizMaterialUsage).filter(BizMaterialUsage.id == usage_id),
            BizMaterialUsage,
            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()