"""市场物料库存盘点 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.child_row_sync import sync_child_rows 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 _unused_line_key(usage_id: Optional[str], batch_id: Optional[str]) -> str: usage = (usage_id or "").strip() if usage: return f"usage:{usage}" batch = (batch_id or "").strip() if batch: return f"batch:{batch}" return "" def _prepare_unused_item(db: Session, it: Dict[str, Any], tenant_no: str) -> Dict[str, Any]: 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") unit_price = it.get("unit_price") if it.get("unit_price") is not None else it.get("unitPrice") return { "id": str(it.get("id") or "").strip(), "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, "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": _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")), "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(unit_price)) if unit_price is not None else None, "holder": (it.get("holder") or "").strip() or None, "unused_qty": Decimal(str(qty)), "remark": (it.get("remark") or "").strip() or None, } 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() prepared = [_prepare_unused_item(db, it, tenant_no) for it in (items or [])] def build_row(row_id: str): return BizMaterialStockCheckUnusedLine( id=row_id, check_id=check_id, material_name="", unused_qty=0, tenant_no=tenant_no, create_by=user_id, create_dept_id=dept_id or "", create_time=now, ) def apply_row(row, it, idx, _now): row.usage_id = it["usage_id"] row.batch_id = it["batch_id"] row.material_code = it["material_code"] row.material_name = it["material_name"] row.spec_model = it["spec_model"] row.unit = it["unit"] row.stock_in_no = it["stock_in_no"] row.stock_in_date = it["stock_in_date"] row.expire_date = it["expire_date"] row.catalog_no = it["catalog_no"] row.batch_no = it["batch_no"] row.manufacturer = it["manufacturer"] row.unit_price = it["unit_price"] row.holder = it["holder"] row.unused_qty = it["unused_qty"] row.remark = it["remark"] row.order_num = idx saved = sync_child_rows( db, BizMaterialStockCheckUnusedLine, parent_attr="check_id", parent_id=check_id, tenant_no=tenant_no, items=prepared, user_id=user_id, now=now, build_row=build_row, apply_fields=apply_row, natural_key=lambda it: _unused_line_key(it.get("usage_id"), it.get("batch_id")), row_natural_key=lambda row: _unused_line_key(row.usage_id, row.batch_id), duplicate_message="未用完明细重复", foreign_message="未用完明细不属于当前盘点单", ) check.update_by = user_id check.update_time = now db.add(check) db.commit() return len(saved) 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()