"""耗材变动记录 CRUD（biz_consumable_stock_log）""" from __future__ import annotations from datetime import datetime from decimal import Decimal from typing import Any, Dict, List, Optional, Tuple from fastapi import HTTPException from sqlalchemy import func, or_ from sqlalchemy.orm import Session from app.core.utils import generate_uuid from app.models.models import BizConsumable, BizConsumableStockLog, 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 _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: BizConsumableStockLog) -> 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: BizConsumableStockLog) -> 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 _row_to_dict(r: BizConsumableStockLog, *, create_by_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" return { "id": r.id, "materialId": r.material_id, "materialCode": r.material_code or "", "materialName": r.material_name or "", "changeType": r.change_type, "changeQty": float(r.change_qty) if r.change_qty is not None else 0, "stockBefore": float(r.stock_before) if r.stock_before is not None else 0, "stockAfter": float(r.stock_after) if r.stock_after is not None else 0, "stepId": r.step_id or "", "stepName": r.step_name or "", "taskId": r.task_id or "", "taskNo": r.task_no or "", "receiver": receiver, "purpose": purpose, "sourceType": r.source_type or "", "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 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(BizConsumableStockLog).filter( BizConsumableStockLog.tenant_no == tenant_no, BizConsumableStockLog.del_flag == 0, ) if material_id: q0 = q0.filter(BizConsumableStockLog.material_id == material_id) ct = (change_type or "").strip() if ct == "out": q0 = q0.filter( BizConsumableStockLog.change_type == "use", BizConsumableStockLog.source_type == "manual_out", ) elif ct == "use": q0 = q0.filter( BizConsumableStockLog.change_type == "use", BizConsumableStockLog.source_type == "step_save", ) elif ct == "in": q0 = q0.filter(BizConsumableStockLog.change_type == "in") if start_time: q0 = q0.filter(BizConsumableStockLog.create_time >= start_time) if end_time: q0 = q0.filter(BizConsumableStockLog.create_time <= end_time + " 23:59:59") kw = (keyword or "").strip() if kw: like = f"%{kw}%" q0 = q0.filter( or_( BizConsumableStockLog.material_code.like(like), BizConsumableStockLog.material_name.like(like), ) ) if user_info: from app.core.data_scope import apply_data_scope q0 = apply_data_scope(q0, BizConsumableStockLog, user_info, db, tenant_no) return q0 total = _filtered().order_by(None).with_entities(func.count(BizConsumableStockLog.id)).scalar() or 0 page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( _filtered() .order_by(BizConsumableStockLog.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) def revert_step_stock_changes( db: Session, step_id: str, tenant_no: str, user_id: Optional[str] = None, ) -> None: """撤销步骤此前 step_save 扣减，便于重新填报。""" rows = ( db.query(BizConsumableStockLog) .filter( BizConsumableStockLog.step_id == step_id, BizConsumableStockLog.tenant_no == tenant_no, BizConsumableStockLog.del_flag == 0, BizConsumableStockLog.source_type == "step_save", BizConsumableStockLog.change_type == "use", ) .all() ) if not rows: return now = datetime.now() for r in rows: restore_qty = abs(r.change_qty) if r.change_qty is not None else Decimal("0") if restore_qty <= 0: r.del_flag = 1 r.update_by = user_id r.update_time = now continue material = ( db.query(BizConsumable) .with_for_update() .filter( BizConsumable.id == r.material_id, BizConsumable.tenant_no == tenant_no, BizConsumable.del_flag == 0, ) .first() ) if material: material.stock_qty = (material.stock_qty or Decimal("0")) + restore_qty material.update_by = user_id material.update_time = now r.del_flag = 1 r.update_by = user_id r.update_time = now def deduct_for_step( db: Session, step_id: str, step_name: str, task_id: str, task_no: str, consumable_items: List[Dict[str, Any]], tenant_no: str, user_id: Optional[str] = None, dept_id: str = "", ) -> None: """扣减耗材库存并生成变动记录。调用方控制事务，本函数不 commit。""" for item in consumable_items: code = (item.get("consumable_code") or "").strip() actual_usage = item.get("actual_usage") if not code: continue # 跳过空/零/非数字的 actual_usage if actual_usage is None or str(actual_usage).strip() == "": continue try: usage_val = Decimal(str(actual_usage)) except Exception: continue if usage_val <= 0: continue # 按 consumable_code 查找耗材（跳过目录中不存在的自定义材料） material = ( db.query(BizConsumable) .with_for_update() .filter( BizConsumable.consumable_code == code, BizConsumable.tenant_no == tenant_no, BizConsumable.del_flag == 0, BizConsumable.status == "1", ) .first() ) if not material: continue stock_before = material.stock_qty if material.stock_qty is not None else Decimal("0") if stock_before < usage_val: raise HTTPException( status_code=400, detail=f"耗材「{material.consumable_name}」库存不足，当前库存 {stock_before}，需使用 {usage_val}", ) new_stock = stock_before - usage_val material.stock_qty = new_stock material.update_by = user_id material.update_time = datetime.now() log = BizConsumableStockLog( id=generate_uuid(), material_id=material.id, material_code=material.consumable_code, material_name=material.consumable_name, change_type="use", change_qty=-usage_val, stock_before=stock_before, stock_after=new_stock, source_type="step_save", step_id=step_id, step_name=(step_name or "").strip() or None, task_id=task_id, task_no=task_no, tenant_no=tenant_no, remark=str(item.get("remark") or "").strip()[:512] or None, create_by=user_id, create_dept_id=dept_id, create_time=datetime.now(), update_by=user_id, update_time=datetime.now(), del_flag=0, ) db.add(log)