zkxw/backend/app/crud/material_stock_log.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 BizMaterialStockLog
from app.models.models import SysUser


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 _dec(v: Any) -> float:
    if v is None:
        return 0.0
    return float(v)


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _row_to_dict(r: BizMaterialStockLog, *, create_by_name: str = "") -> Dict[str, Any]:
    create_by = r.create_by or ""
    return {
        "id": r.id,
        "materialId": r.material_id,
        "materialCode": r.material_code or "",
        "materialName": r.material_name or "",
        "changeType": r.change_type,
        "changeQty": _dec(r.change_qty),
        "stockBefore": _dec(r.stock_before),
        "stockAfter": _dec(r.stock_after),
        "receiver": r.receiver or "",
        "purpose": r.purpose or "",
        "remark": r.remark or "",
        "createBy": create_by,
        "createByName": create_by_name or create_by,
        "createTime": _fmt_dt(r.create_time),
        "updateTime": _fmt_dt(r.update_time),
    }


def list_stock_logs(
    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,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizMaterialStockLog).filter(
            BizMaterialStockLog.tenant_no == tenant_no,
            BizMaterialStockLog.del_flag == 0,
        )
        if material_id:
            q0 = q0.filter(BizMaterialStockLog.material_id == material_id)
        if change_type:
            q0 = q0.filter(BizMaterialStockLog.change_type == change_type)
        if start_time:
            q0 = q0.filter(BizMaterialStockLog.create_time >= start_time)
        if end_time:
            q0 = q0.filter(BizMaterialStockLog.create_time <= end_time + " 23:59:59")
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizMaterialStockLog.material_code.ilike(like),
                    BizMaterialStockLog.material_name.ilike(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizMaterialStockLog, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizMaterialStockLog.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizMaterialStockLog.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    name_map = _user_nick_name_map(db, [x.create_by for x in rows if x.create_by])
    return [
        _row_to_dict(x, create_by_name=name_map.get(x.create_by or "", ""))
        for x in rows
    ], int(total)