"""仓库操作 DB 装配：调拨、盘点、组合装组装/拆解。全部经账本流水。""" import hashlib import uuid from decimal import Decimal from fastapi import HTTPException from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.plugins.inventory.ledger import TxnRequest, build_txn from app.plugins.inventory.projection import project from app.plugins.inventory.services import ( _default_warehouse_id, _allow_negative, assert_period_open, _batch_enabled, _claim_operation, ) from app.plugins.inventory.warehouse_ops import transfer_legs, stocktake_diff, explode_kit, KitComponent async def _assert_warehouse(db, tenant_id: int, warehouse_id: int) -> None: r = await db.execute( text("SELECT id FROM inventory_warehouses WHERE id=:id AND tenant_id=:tid AND is_active=1"), {"id": warehouse_id, "tid": tenant_id}, ) if not r.fetchone(): raise HTTPException(status_code=400, detail="仓库不存在、已停用或不属于当前租户") async def _default_location_id(db, tenant_id: int, warehouse_id: int) -> int: location_id = (await db.execute( text("SELECT id FROM inventory_locations WHERE tenant_id=:tid AND warehouse_id=:wh " "AND is_default=1 AND is_active=1 ORDER BY id LIMIT 1"), {"tid": tenant_id, "wh": warehouse_id}, )).scalar() if not location_id: raise HTTPException(status_code=400, detail="仓库没有可用的默认库位") return int(location_id) async def _reject_if_batch(db, tenant_id: int, op: str) -> None: """批次租户禁用批次无关的通用操作，避免把批次库存写成 batch_id=0 丢失维度。""" if await _batch_enabled(db, tenant_id): raise HTTPException(status_code=400, detail=f"已启用批次管理，{op}需按批次操作；通用{op}接口对批次租户暂不可用") async def _balance_before(db, tenant_id, pid, vid, wh, batch_id: int = 0) -> Decimal: r = await db.execute( text("SELECT COALESCE(SUM(qty),0) FROM inventory_balances WHERE tenant_id=:tid " "AND product_id=:pid AND variant_id=:vid AND warehouse_id=:wh AND batch_id=:bk " "AND stock_state='sellable' FOR UPDATE"), {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "bk": batch_id}, ) return Decimal(str(r.scalar() or 0)) async def _leg(db, tenant_id, pid, vid, wh, delta, doc_type, doc_id, operator_id, reason, *, touch_product_projection: bool, allow_negative: bool = True, batch_id: int = 0, location_id: int = 0, idem: str | None = None) -> None: # 客户端给了幂等键则用「键+腿上下文」的确定键（超时重试去重，与在线一致）； # 未给则每腿唯一键（管理动作默认按逐次独立处理）。 if idem: from app.plugins.inventory.services import _txn_exists raw = f"{doc_type}:{idem}:{wh}:{batch_id}:{pid}:{vid}:{'+' if delta >= 0 else '-'}" key = hashlib.sha1(raw.encode()).hexdigest() if await _txn_exists(db, tenant_id, key): return # 已应用过的重复请求：跳过整条腿，优雅幂等 else: key = f"{doc_type}:{uuid.uuid4().hex}"[:120] before = await _balance_before(db, tenant_id, pid, vid, wh, batch_id) if delta < 0 and not allow_negative and before + delta < 0: raise HTTPException(status_code=400, detail=f"商品 {pid} 在仓库 {wh} 库存不足（剩余 {before}，需出 {-delta}）") _t = build_txn(TxnRequest(tenant_id=tenant_id, product_id=pid, variant_id=vid or None, warehouse_id=wh, location_id=location_id or None, qty_delta=delta, doc_type=doc_type, doc_id=doc_id, batch_id=batch_id or None, operator_id=operator_id, reason=reason), qty_before=before) _t.idempotency_key = key db.add(_t) await db.execute( text("INSERT INTO inventory_balances " "(tenant_id, product_id, variant_id, warehouse_id, location_id, batch_id, stock_state, qty, created_at, updated_at) " "VALUES (:tid,:pid,:vid,:wh,:location,:bk,'sellable',:q,NOW(3),NOW(3)) " "ON DUPLICATE KEY UPDATE qty = qty + :q, updated_at = NOW(3)"), {"tid": tenant_id, "pid": pid, "vid": vid, "wh": wh, "location": location_id, "bk": batch_id, "q": delta}, ) if touch_product_projection: await db.execute( text("UPDATE products SET stock_qty = stock_qty + :d WHERE id=:pid AND tenant_id=:tid"), {"d": delta, "pid": pid, "tid": tenant_id}, ) if vid: await db.execute( text("UPDATE product_variants SET stock_qty = stock_qty + :d WHERE id=:vid AND tenant_id=:tid"), {"d": delta, "vid": vid, "tid": tenant_id}, ) async def do_transfer(db: AsyncSession, tenant_id: int, *, product_id: int, variant_id: int | None, from_wh: int, to_wh: int, qty: Decimal, operator_id: int | None, idem_key: str | None = None) -> None: """仓间调拨：总在库不变，只在两仓间移动，不动商品级投影。 批次租户按 FEFO 逐批移动：调出的正是哪些批次，调入仓就落哪些批次，保住批次维度。 给了 idem_key 则超时重试去重（与在线一致）。 """ await assert_period_open(db, tenant_id) vid = variant_id or 0 q = Decimal(str(qty)) await _assert_warehouse(db, tenant_id, from_wh) await _assert_warehouse(db, tenant_id, to_wh) if from_wh == to_wh: raise HTTPException(status_code=400, detail="调出与调入仓库不能相同") # 先原子认领整次调拨，重放不再重新 FEFO 或基于已扣减余额报库存不足。 if idem_key and not await _claim_operation(db, tenant_id, "transfer", idem_key): return allow_neg = await _allow_negative(db, tenant_id) if await _batch_enabled(db, tenant_id): from app.plugins.inventory.allocation import allocate_fefo, BatchStock rows = (await db.execute( text("SELECT bal.batch_id, bal.qty, b.expires_on, b.id " "FROM inventory_balances bal JOIN inventory_batches b ON b.id=bal.batch_id " "WHERE bal.tenant_id=:tid AND bal.product_id=:pid AND bal.variant_id=:vid " "AND bal.warehouse_id=:wh AND bal.stock_state='sellable' AND bal.qty>0 AND bal.batch_id<>0 " "AND b.qc_state='passed' AND (b.expires_on IS NULL OR b.expires_on>=CURDATE()) " "FOR UPDATE"), {"tid": tenant_id, "pid": product_id, "vid": vid, "wh": from_wh}, )).fetchall() stocks = [BatchStock(batch_id=x[0], qty=Decimal(str(x[1])), expires_on=x[2], created_id=x[3]) for x in rows] allocs, gap = allocate_fefo(stocks, q) if q > 0 else ([], Decimal("0")) for a in allocs: await _leg(db, tenant_id, product_id, vid, from_wh, -a.qty, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=True, batch_id=a.batch_id, idem=idem_key) await _leg(db, tenant_id, product_id, vid, to_wh, a.qty, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=True, batch_id=a.batch_id, idem=idem_key) # 缺口用 batch0（启用批次前导入的期初库存）补：先看 from_wh 的 batch0 可调量 if gap > 0: b0 = await _balance_before(db, tenant_id, product_id, vid, from_wh, 0) move0 = gap if gap <= b0 else b0 if move0 <= 0 and not allow_neg: raise HTTPException(status_code=400, detail=f"商品 {product_id} 在仓 {from_wh} 可调库存不足") if move0 > 0: await _leg(db, tenant_id, product_id, vid, from_wh, -move0, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=True, idem=idem_key) await _leg(db, tenant_id, product_id, vid, to_wh, move0, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=True, idem=idem_key) gap -= move0 if gap > 0 and not allow_neg: raise HTTPException(status_code=400, detail=f"商品 {product_id} 在仓 {from_wh} 可调库存不足") if gap > 0: # allow_neg：仍缺就 batch0 走负 await _leg(db, tenant_id, product_id, vid, from_wh, -gap, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=True, idem=idem_key) await _leg(db, tenant_id, product_id, vid, to_wh, gap, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=True, idem=idem_key) return for leg in transfer_legs(from_wh, to_wh, q): # 调出腿受负库存限制（不能把某仓调成负）；调入腿恒为正不受限 await _leg(db, tenant_id, product_id, vid, leg.warehouse_id, leg.qty_delta, "transfer", None, operator_id, "调拨", touch_product_projection=False, allow_negative=allow_neg, idem=idem_key) async def do_stocktake(db: AsyncSession, tenant_id: int, *, product_id: int, variant_id: int | None, counted_qty: Decimal, operator_id: int | None, warehouse_id: int | None = None, batch_id: int | None = None, idem_key: str | None = None) -> Decimal: """盘点：按实盘与账面差异写调整流水。返回差异（正溢负损）。""" await assert_period_open(db, tenant_id) batch_on = await _batch_enabled(db, tenant_id) if batch_on and not batch_id: raise HTTPException(status_code=400, detail="已启用批次管理，盘点必须选择具体批次") if not batch_on and batch_id: raise HTTPException(status_code=400, detail="未启用批次管理，不能指定批次") if idem_key and not await _claim_operation(db, tenant_id, "stocktake", idem_key): return Decimal("0") vid = variant_id or 0 wh = warehouse_id or await _default_warehouse_id(db, tenant_id) await _assert_warehouse(db, tenant_id, wh) bkey = int(batch_id or 0) system = await _balance_before(db, tenant_id, product_id, vid, wh, bkey) diff = stocktake_diff(system, Decimal(str(counted_qty))) if diff != 0: # ponytail: warehouse-level count adjusts the default location; add a location-level count only if needed. await _leg(db, tenant_id, product_id, vid, wh, diff, "stocktake", None, operator_id, "盘点差异", touch_product_projection=True, batch_id=bkey, location_id=await _default_location_id(db, tenant_id, wh), idem=idem_key) return diff async def do_kit(db: AsyncSession, tenant_id: int, *, parent_product_id: int, kit_qty: Decimal, disassemble: bool, operator_id: int | None, idem_key: str | None = None) -> None: """组合装组装/拆解：按配方展开子件，父件与子件反向变动。""" await assert_period_open(db, tenant_id) await _reject_if_batch(db, tenant_id, "组合装组装/拆解") if idem_key and not await _claim_operation( db, tenant_id, "kit_disassemble" if disassemble else "kit_assemble", idem_key ): return comps = (await db.execute( text("SELECT child_product_id, qty FROM inventory_kit_recipes " "WHERE tenant_id=:tid AND parent_product_id=:pid"), {"tid": tenant_id, "pid": parent_product_id}, )).fetchall() if not comps: raise HTTPException(status_code=400, detail="该商品未配置组合装配方") components = [KitComponent(child_product_id=c[0], qty=Decimal(str(c[1]))) for c in comps] q = Decimal(str(kit_qty)) sign = -1 if disassemble else 1 # 组装 → 父件+、子件-；拆解相反 wh = await _default_warehouse_id(db, tenant_id) allow_neg = await _allow_negative(db, tenant_id) doc = "kit_disassemble" if disassemble else "kit_assemble" # 父件（拆解时父件出库，受负库存限制） await _leg(db, tenant_id, parent_product_id, 0, wh, sign * q, doc, None, operator_id, doc, touch_product_projection=True, allow_negative=allow_neg, idem=idem_key) # 子件反向（组装时子件出库，受负库存限制） for child_id, child_qty in explode_kit(components, q): await _leg(db, tenant_id, child_id, 0, wh, -sign * child_qty, doc, None, operator_id, doc, touch_product_projection=True, allow_negative=allow_neg, idem=idem_key)