"""Shared stock-operation documents and atomic posting for receipts and vendor returns.""" from dataclasses import dataclass from decimal import Decimal from types import SimpleNamespace import json from datetime import date from fastapi import HTTPException from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.plugins.inventory.services import assert_period_open def _serial_numbers(value) -> list[str]: return list(json.loads(value) if isinstance(value, str) else value or []) def stock_qty_from_purchase_qty(qty: Decimal, uom_factor: Decimal) -> Decimal: return Decimal(str(qty)) * Decimal(str(uom_factor)) def purchase_qty_from_stock_qty(qty: Decimal, uom_factor: Decimal) -> Decimal: return Decimal(str(qty)) / Decimal(str(uom_factor)) def stock_unit_cost(unit_cost: Decimal, uom_factor: Decimal, fx_rate: Decimal) -> Decimal: return Decimal(str(unit_cost)) * Decimal(str(fx_rate)) / Decimal(str(uom_factor)) @dataclass(frozen=True) class ReceiptProgress: ordered: Decimal received: Decimal @property def remaining(self) -> Decimal: return max(Decimal("0"), Decimal(str(self.ordered)) - Decimal(str(self.received))) def apply(self, qty: Decimal) -> "ReceiptProgress": value = Decimal(str(qty)) if value <= 0: raise HTTPException(status_code=400, detail="实际数量必须大于 0") if value > self.remaining: raise HTTPException(status_code=400, detail=f"实际数量 {value} 超过未完成数量 {self.remaining}") return ReceiptProgress(Decimal(str(self.ordered)), Decimal(str(self.received)) + value) async def _existing_operation(db: AsyncSession, tenant_id: int, idempotency_key: str | None) -> int | None: if not idempotency_key: return None return (await db.execute( text("SELECT id FROM inventory_operations WHERE tenant_id=:tid AND idempotency_key=:key LIMIT 1"), {"tid": tenant_id, "key": idempotency_key}, )).scalar() async def _validate_receipt_draft_lines(db, tenant_id: int, purchase_order_id: int, lines: list, operation_id: int = 0) -> None: for line in lines: row = (await db.execute( text("SELECT qty,received_qty FROM inventory_purchase_order_lines " "WHERE id=:line AND tenant_id=:tid AND po_id=:po FOR UPDATE"), {"line": line.purchase_order_line_id, "tid": tenant_id, "po": purchase_order_id}, )).fetchone() if row is None: raise HTTPException(status_code=400, detail="收货单包含不属于该采购单的行") reserved = (await db.execute( text("SELECT COALESCE(SUM(m.done_qty / pol.uom_factor),0) FROM inventory_moves m " "JOIN inventory_operations op ON op.id=m.operation_id AND op.tenant_id=m.tenant_id " "JOIN inventory_purchase_order_lines pol ON pol.id=m.purchase_order_line_id AND pol.tenant_id=m.tenant_id " "WHERE m.tenant_id=:tid AND m.purchase_order_line_id=:line AND op.operation_type='receipt' " "AND op.state='draft' AND op.id<>:operation_id"), {"tid": tenant_id, "line": line.purchase_order_line_id, "operation_id": operation_id}, )).scalar() available = Decimal(str(row[0])) - Decimal(str(row[1] or 0)) - Decimal(str(reserved or 0)) if Decimal(str(line.done_qty)) <= 0 or Decimal(str(line.done_qty)) > available: raise HTTPException(status_code=409, detail=f"采购行 {line.purchase_order_line_id} 可收数量不足") async def _insert_operation(db: AsyncSession, tenant_id: int, *, operation_type: str, source_doc_type: str | None, source_doc_id: int | None, warehouse_id: int, planned_at: str | None = None, reference: str | None = None, reason: str | None = None, idempotency_key: str | None = None) -> int: existing = await _existing_operation(db, tenant_id, idempotency_key) if existing: return int(existing) await db.execute( text("INSERT INTO inventory_operations " "(tenant_id,operation_type,state,source_doc_type,source_doc_id,warehouse_id,planned_at," "idempotency_key,reference,reason,created_at,updated_at) " "VALUES (:tid,:kind,'draft',:source_type,:source_id,:wh,:planned,:idem,:reference,:reason,NOW(3),NOW(3))"), {"tid": tenant_id, "kind": operation_type, "source_type": source_doc_type, "source_id": source_doc_id, "wh": warehouse_id, "planned": planned_at, "idem": idempotency_key, "reference": reference, "reason": reason}, ) return int((await db.execute(text("SELECT LAST_INSERT_ID()"))).scalar()) async def create_purchase_receipt(db: AsyncSession, tenant_id: int, purchase_order_id: int, lines: list, *, planned_at: str | None = None, reference: str | None = None, idempotency_key: str | None = None, operation_id: int = 0) -> int: existing = await _existing_operation(db, tenant_id, idempotency_key) if not operation_id else None if existing: return int(existing) line_ids = [int(line.purchase_order_line_id) for line in lines] if len(line_ids) != len(set(line_ids)): raise HTTPException(status_code=400, detail="一次收货单中同一采购行只能出现一次") po = (await db.execute( text("SELECT status, warehouse_id, fx_rate FROM inventory_purchase_orders " "WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": purchase_order_id, "tid": tenant_id}, )).fetchone() if po is None: raise HTTPException(status_code=404, detail="采购单不存在") if po[0] not in ("confirmed", "partially_received"): raise HTTPException(status_code=409, detail=f"采购单状态为 {po[0]}，不可创建收货单") if not lines: raise HTTPException(status_code=400, detail="收货单至少一行") await _validate_receipt_draft_lines(db, tenant_id, purchase_order_id, lines) placeholders = ",".join(str(line_id) for line_id in sorted(line_ids)) rows = (await db.execute( text("SELECT id,product_id,variant_id,qty,received_qty,unit_cost,batch_no,expires_on,uom_factor " f"FROM inventory_purchase_order_lines WHERE tenant_id=:tid AND po_id=:po AND id IN ({placeholders}) " "FOR UPDATE"), {"tid": tenant_id, "po": purchase_order_id}, )).fetchall() by_id = {int(row[0]): row for row in rows} if set(by_id) != set(line_ids): raise HTTPException(status_code=400, detail="收货单包含不属于该采购单的行") if not operation_id: operation_id = await _insert_operation( db, tenant_id, operation_type="receipt", source_doc_type="purchase_order", source_doc_id=purchase_order_id, warehouse_id=int(po[1]), planned_at=planned_at, reference=reference, idempotency_key=idempotency_key, ) else: await db.execute(text("UPDATE inventory_operations SET planned_at=:planned,reference=:reference,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"planned": planned_at, "reference": reference, "id": operation_id, "tid": tenant_id}) fx_rate = Decimal(str(po[2])) for line in lines: row = by_id[int(line.purchase_order_line_id)] ReceiptProgress(Decimal(str(row[3])), Decimal(str(row[4] or 0))).apply(line.done_qty) uom_factor = Decimal(str(row[8])) stock_qty = stock_qty_from_purchase_qty(line.done_qty, uom_factor) batch_no = line.batch_no or row[6] from app.plugins.inventory.traceability import calculate_expiry, load_product_profile, validate_move_tracking profile = await load_product_profile(db, tenant_id, int(row[1]), int(row[2] or 0)) if profile["tracking_type"] == "lot" and not batch_no: batch_no = f"AUTO-{date.today():%Y%m%d}-{operation_id}" serial_numbers = list(line.serial_numbers) validate_move_tracking(profile["tracking_type"], stock_qty, serial_numbers=serial_numbers, batch_no=batch_no) expires_on = line.expires_on or row[7] if not expires_on and line.produced_on and profile["shelf_life_days"]: from datetime import date expires_on = calculate_expiry(date.fromisoformat(line.produced_on), profile["shelf_life_days"]) destination_state = "qc" if profile["qc_required"] else "sellable" await db.execute( text("INSERT INTO inventory_moves " "(tenant_id,operation_id,purchase_order_line_id,product_id,variant_id,source_warehouse_id," "destination_warehouse_id,source_location_id,destination_location_id,planned_qty,done_qty," "batch_id,batch_no,expires_on,serial_no,serial_numbers,produced_on,source_state,destination_state," "unit_cost,created_at,updated_at) " "VALUES (:tid,:op,:line,:pid,:vid,0,:wh,0,:location,:qty,:qty,0,:batch,:expires,NULL,:serials," ":produced,'vendor',:destination_state,:cost,NOW(3),NOW(3))"), {"tid": tenant_id, "op": operation_id, "line": row[0], "pid": row[1], "vid": row[2] or 0, "wh": po[1], "location": line.destination_location_id or 0, "qty": stock_qty, "batch": batch_no, "expires": expires_on, "serials": json.dumps(serial_numbers), "produced": line.produced_on, "destination_state": destination_state, "cost": stock_unit_cost(Decimal(str(row[5])), uom_factor, fx_rate)}, ) return operation_id async def update_purchase_receipt(db, tenant_id: int, operation_id: int, body) -> None: operation = (await db.execute( text("SELECT source_doc_id,state FROM inventory_operations WHERE id=:id AND tenant_id=:tid " "AND operation_type='receipt' FOR UPDATE"), {"id": operation_id, "tid": tenant_id}, )).fetchone() if operation is None: raise HTTPException(status_code=404, detail="收货单不存在") if operation[1] != "draft": raise HTTPException(status_code=409, detail="仅草稿收货单可编辑") if int(operation[0]) != int(body.purchase_order_id): raise HTTPException(status_code=400, detail="收货单不能更换来源采购单") await db.execute(text("DELETE FROM inventory_moves WHERE tenant_id=:tid AND operation_id=:id"), {"tid": tenant_id, "id": operation_id}) await create_purchase_receipt(db, tenant_id, body.purchase_order_id, body.lines, planned_at=body.planned_at, reference=body.reference, operation_id=operation_id) async def cancel_purchase_receipt(db, tenant_id: int, operation_id: int) -> None: result = await db.execute( text("UPDATE inventory_operations SET state='cancelled',updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid AND operation_type='receipt' AND state='draft'"), {"id": operation_id, "tid": tenant_id}, ) if not result.rowcount: raise HTTPException(status_code=409, detail="仅草稿收货单可作废") async def create_full_purchase_receipt(db: AsyncSession, tenant_id: int, purchase_order_id: int) -> int: rows = (await db.execute( text("SELECT id,qty,received_qty,batch_no,expires_on FROM inventory_purchase_order_lines " "WHERE tenant_id=:tid AND po_id=:po ORDER BY id"), {"tid": tenant_id, "po": purchase_order_id}, )).fetchall() lines = [ SimpleNamespace(purchase_order_line_id=row[0], done_qty=Decimal(str(row[1])) - Decimal(str(row[2] or 0)), destination_location_id=None, batch_no=row[3], produced_on=None, expires_on=row[4], serial_numbers=[]) for row in rows if Decimal(str(row[1])) > Decimal(str(row[2] or 0)) ] if not lines: raise HTTPException(status_code=409, detail="采购单没有待收货数量") return await create_purchase_receipt(db, tenant_id, purchase_order_id, lines) async def create_vendor_return(db: AsyncSession, tenant_id: int, *, warehouse_id: int, purchase_order_id: int | None, lines: list, reason: str, reference: str | None = None, idempotency_key: str | None = None) -> int: from app.plugins.inventory.purchasing import _validate_line_products existing = await _existing_operation(db, tenant_id, idempotency_key) if existing: return int(existing) warehouse = (await db.execute( text("SELECT 1 FROM inventory_warehouses WHERE id=:id AND tenant_id=:tid AND is_active=1"), {"id": warehouse_id, "tid": tenant_id}, )).scalar() if not warehouse: raise HTTPException(status_code=400, detail="退货仓库不存在或不属于当前租户") await _validate_line_products(db, tenant_id, lines) operation_id = await _insert_operation( db, tenant_id, operation_type="vendor_return", source_doc_type="purchase_order" if purchase_order_id else None, source_doc_id=purchase_order_id, warehouse_id=warehouse_id, reference=reference, reason=reason, idempotency_key=idempotency_key, ) for line in lines: from app.plugins.inventory.traceability import load_product_profile, validate_move_tracking profile = await load_product_profile(db, tenant_id, int(line.product_id), int(line.variant_id or 0)) validate_move_tracking(profile["tracking_type"], line.done_qty, serial_numbers=line.serial_numbers, batch_no=line.batch_no) await db.execute( text("INSERT INTO inventory_moves " "(tenant_id,operation_id,purchase_order_line_id,product_id,variant_id,source_warehouse_id," "destination_warehouse_id,source_location_id,destination_location_id,planned_qty,done_qty," "batch_id,batch_no,expires_on,serial_no,serial_numbers,produced_on,source_state,destination_state," "unit_cost,created_at,updated_at) " "VALUES (:tid,:op,:po_line,:pid,:vid,:wh,0,:location,0,:qty,:qty,:batch_id,:batch_no," ":expires,NULL,:serials,:produced,'sellable','vendor',:cost,NOW(3),NOW(3))"), {"tid": tenant_id, "op": operation_id, "po_line": line.purchase_order_line_id, "pid": line.product_id, "vid": line.variant_id or 0, "wh": warehouse_id, "location": line.source_location_id or 0, "qty": line.done_qty, "batch_id": line.batch_id or 0, "batch_no": line.batch_no, "expires": line.expires_on, "serials": json.dumps(line.serial_numbers), "produced": line.produced_on, "cost": line.unit_cost}, ) return operation_id async def _resolve_batch(db: AsyncSession, tenant_id: int, move, batch_enabled: bool, *, qc_required: bool = False, operation_id: int | None = None) -> int: if move[12]: valid = (await db.execute( text("SELECT 1 FROM inventory_batches WHERE id=:id AND tenant_id=:tid " "AND product_id=:pid AND (variant_id IS NULL OR variant_id=:vid)"), {"id": move[12], "tid": tenant_id, "pid": move[3], "vid": move[4] or None}, )).scalar() if not valid: raise HTTPException(status_code=400, detail="批次不存在或不属于当前商品/租户") return int(move[12]) if not batch_enabled: return 0 batch_no = move[13] or (f"QC-{operation_id}-{move[0]}" if qc_required and operation_id else None) if not batch_no: raise HTTPException(status_code=400, detail=f"商品 {move[3]} 启用批次管理，必须填写批次号") await db.execute( text("INSERT INTO inventory_batches " "(tenant_id,product_id,variant_id,batch_no,produced_on,expires_on,qc_state,source_operation_id," "created_at,updated_at) " "VALUES (:tid,:pid,:vid,:batch,:produced,:expires,:qc,:op,NOW(3),NOW(3)) " "ON DUPLICATE KEY UPDATE produced_on=COALESCE(VALUES(produced_on),produced_on)," "expires_on=COALESCE(VALUES(expires_on),expires_on),updated_at=NOW(3)"), {"tid": tenant_id, "pid": move[3], "vid": move[4] or None, "batch": batch_no, "produced": move[19], "expires": move[14], "qc": "pending" if qc_required else "passed", "op": operation_id}, ) return int((await db.execute( text("SELECT id FROM inventory_batches WHERE tenant_id=:tid AND product_id=:pid AND batch_no=:batch"), {"tid": tenant_id, "pid": move[3], "batch": batch_no}, )).scalar()) async def _resolve_location(db: AsyncSession, tenant_id: int, warehouse_id: int, requested_location_id: int | None) -> int: if requested_location_id: valid = (await db.execute( text("SELECT 1 FROM inventory_locations WHERE id=:id AND tenant_id=:tid AND warehouse_id=:wh " "AND is_active=1"), {"id": requested_location_id, "tid": tenant_id, "wh": warehouse_id}, )).scalar() if not valid: raise HTTPException(status_code=400, detail="库位不存在或不属于当前仓库/租户") return int(requested_location_id) default_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 default_id: raise HTTPException(status_code=400, detail="仓库没有可用的默认库位") return int(default_id) async def _write_operation_leg(db: AsyncSession, tenant_id: int, *, operation_id: int, move_id: int, product_id: int, variant_id: int, warehouse_id: int, location_id: int, batch_id: int, qty_delta: Decimal, unit_cost: Decimal | None, operator_id: int, reason: str, stock_state: str = "sellable", leg_suffix: str = "main") -> None: from app.plugins.inventory.ledger import TxnRequest, build_txn before = (await db.execute( text("SELECT qty FROM inventory_balances WHERE tenant_id=:tid AND product_id=:pid " "AND variant_id=:vid AND warehouse_id=:wh AND location_id=:location " "AND batch_id=:batch AND stock_state=:state FOR UPDATE"), {"tid": tenant_id, "pid": product_id, "vid": variant_id, "wh": warehouse_id, "location": location_id, "batch": batch_id, "state": stock_state}, )).scalar() before = Decimal(str(before or 0)) if qty_delta < 0 and before + qty_delta < 0: raise HTTPException(status_code=400, detail=f"商品 {product_id} 在指定仓库/库位/批次库存不足") transaction = build_txn( TxnRequest(tenant_id=tenant_id, product_id=product_id, variant_id=variant_id or None, warehouse_id=warehouse_id, location_id=location_id or None, batch_id=batch_id or None, qty_delta=qty_delta, doc_type="inventory_operation", doc_id=operation_id, unit_cost=unit_cost, operator_id=operator_id, reason=reason, stock_state=stock_state), qty_before=before, ) transaction.idempotency_key = f"inventory_operation:{operation_id}:{move_id}:{leg_suffix}"[:120] db.add(transaction) 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,:batch,:state,:qty,NOW(3),NOW(3)) " "ON DUPLICATE KEY UPDATE qty=qty+:qty,updated_at=NOW(3)"), {"tid": tenant_id, "pid": product_id, "vid": variant_id, "wh": warehouse_id, "location": location_id, "batch": batch_id, "state": stock_state, "qty": qty_delta}, ) await db.execute(text("UPDATE products SET stock_qty=stock_qty+:qty WHERE id=:id AND tenant_id=:tid"), {"qty": qty_delta, "id": product_id, "tid": tenant_id}) if variant_id: await db.execute(text("UPDATE product_variants SET stock_qty=stock_qty+:qty " "WHERE id=:id AND product_id=:pid AND tenant_id=:tid"), {"qty": qty_delta, "id": variant_id, "pid": product_id, "tid": tenant_id}) async def post_operation(db: AsyncSession, tenant_id: int, operation_id: int, operator_id: int) -> str: await assert_period_open(db, tenant_id) operation = (await db.execute( text("SELECT operation_type,state,source_doc_type,source_doc_id,warehouse_id,reason " "FROM inventory_operations WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": operation_id, "tid": tenant_id}, )).fetchone() if operation is None: raise HTTPException(status_code=404, detail="库存操作单不存在") if operation[1] != "draft": raise HTTPException(status_code=409, detail=f"库存操作单状态为 {operation[1]}，仅草稿可确认") moves = (await db.execute( text("SELECT id,operation_id,purchase_order_line_id,product_id,variant_id,source_warehouse_id," "destination_warehouse_id,source_location_id,destination_location_id,planned_qty,done_qty," "unit_cost,batch_id,batch_no,expires_on,serial_no,source_state,destination_state " ",serial_numbers,produced_on " "FROM inventory_moves WHERE tenant_id=:tid AND operation_id=:op ORDER BY id FOR UPDATE"), {"tid": tenant_id, "op": operation_id}, )).fetchall() if not moves: raise HTTPException(status_code=400, detail="库存操作单没有明细") if operation[0] == "receipt": po_state = (await db.execute( text("SELECT status FROM inventory_purchase_orders WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": operation[3], "tid": tenant_id}, )).scalar() if po_state not in ("confirmed", "partially_received"): raise HTTPException(status_code=409, detail=f"采购单状态为 {po_state}，不可确认收货") batch_enabled = bool((await db.execute( text("SELECT batch_enabled FROM inventory_settings WHERE tenant_id=:tid"), {"tid": tenant_id} )).scalar() or 0) for move in moves: po_line = (await db.execute( text("SELECT qty,received_qty,product_id,variant_id,uom_factor FROM inventory_purchase_order_lines " "WHERE id=:id AND tenant_id=:tid AND po_id=:po FOR UPDATE"), {"id": move[2], "tid": tenant_id, "po": operation[3]}, )).fetchone() if po_line is None or int(po_line[2]) != int(move[3]) or int(po_line[3] or 0) != int(move[4] or 0): raise HTTPException(status_code=400, detail="收货行与采购行不匹配") purchase_qty = purchase_qty_from_stock_qty(Decimal(str(move[10])), Decimal(str(po_line[4]))) ReceiptProgress(Decimal(str(po_line[0])), Decimal(str(po_line[1] or 0))).apply(purchase_qty) from app.plugins.inventory.traceability import ( load_product_profile, register_received_serials, validate_move_tracking, ) profile = await load_product_profile(db, tenant_id, int(move[3]), int(move[4] or 0)) serial_numbers = _serial_numbers(move[18]) validate_move_tracking(profile["tracking_type"], move[10], serial_numbers=serial_numbers, batch_no=move[13]) batch_id = await _resolve_batch( db, tenant_id, move, batch_enabled or profile["tracking_type"] == "lot" or profile["qc_required"], qc_required=profile["qc_required"], operation_id=operation_id, ) location_id = await _resolve_location(db, tenant_id, int(move[6]), int(move[8] or 0)) qty = Decimal(str(move[10])) cost = Decimal(str(move[11] or 0)) destination_state = "qc" if profile["qc_required"] else "sellable" await _write_operation_leg( db, tenant_id, operation_id=operation_id, move_id=int(move[0]), product_id=int(move[3]), variant_id=int(move[4] or 0), warehouse_id=int(move[6]), location_id=location_id, batch_id=batch_id, qty_delta=qty, unit_cost=cost, operator_id=operator_id, reason="采购收货", stock_state=destination_state, ) if serial_numbers: await register_received_serials( db, tenant_id, serial_numbers=serial_numbers, product_id=int(move[3]), variant_id=int(move[4] or 0), warehouse_id=int(move[6]), location_id=location_id, batch_id=batch_id, stock_state=destination_state, operation_id=operation_id, ) await db.execute( text("UPDATE inventory_moves SET destination_location_id=:location,batch_id=:batch,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"location": location_id, "batch": batch_id, "id": move[0], "tid": tenant_id}, ) await db.execute( text("INSERT INTO inventory_cost_layers " "(tenant_id,product_id,variant_id,batch_id,source_operation_id,source_move_id,qty_in,qty_consumed," "unit_cost,created_at,updated_at) " "VALUES (:tid,:pid,:vid,:batch,:op,:move,:qty,0,:cost,NOW(3),NOW(3))"), {"tid": tenant_id, "pid": move[3], "vid": move[4] or 0, "batch": batch_id, "op": operation_id, "move": move[0], "qty": qty, "cost": cost}, ) await db.execute( text("UPDATE inventory_purchase_order_lines SET received_qty=received_qty+:qty,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"qty": purchase_qty, "id": move[2], "tid": tenant_id}, ) outstanding = (await db.execute( text("SELECT COUNT(*) FROM inventory_purchase_order_lines WHERE tenant_id=:tid AND po_id=:po " "AND received_qty < qty"), {"tid": tenant_id, "po": operation[3]}, )).scalar() po_status = "partially_received" if int(outstanding or 0) else "received" await db.execute(text("UPDATE inventory_purchase_orders SET status=:status,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"status": po_status, "id": operation[3], "tid": tenant_id}) elif operation[0] == "vendor_return": from app.plugins.inventory.services import _consume_batch_layer, _consume_cost_pool, _cost_method batch_enabled = bool((await db.execute( text("SELECT batch_enabled FROM inventory_settings WHERE tenant_id=:tid"), {"tid": tenant_id} )).scalar() or 0) cost_method = await _cost_method(db, tenant_id) for move in moves: qty = Decimal(str(move[10])) from app.plugins.inventory.traceability import load_product_profile, move_serials, validate_move_tracking profile = await load_product_profile(db, tenant_id, int(move[3]), int(move[4] or 0)) serial_numbers = _serial_numbers(move[18]) validate_move_tracking(profile["tracking_type"], qty, serial_numbers=serial_numbers, batch_no=move[13]) if move[2]: po_line = (await db.execute( text("SELECT received_qty,returned_qty,product_id,variant_id FROM inventory_purchase_order_lines " "WHERE id=:id AND tenant_id=:tid AND (:po IS NULL OR po_id=:po) FOR UPDATE"), {"id": move[2], "tid": tenant_id, "po": operation[3]}, )).fetchone() if po_line is None or int(po_line[2]) != int(move[3]) or int(po_line[3] or 0) != int(move[4] or 0): raise HTTPException(status_code=400, detail="供应商退货行与采购行不匹配") ReceiptProgress(Decimal(str(po_line[0])), Decimal(str(po_line[1] or 0))).apply(qty) batch_id = int(move[12] or 0) if batch_enabled and not batch_id: raise HTTPException(status_code=400, detail=f"商品 {move[3]} 启用批次管理，供应商退货必须指定批次") if batch_id: await _resolve_batch(db, tenant_id, move, batch_enabled) location_id = await _resolve_location(db, tenant_id, int(move[5]), int(move[7] or 0)) if cost_method == "batch_actual" and batch_id: unit_cost = await _consume_batch_layer( db, tenant_id, int(move[3]), int(move[4] or 0), batch_id, qty ) else: total_cost = await _consume_cost_pool( db, tenant_id, int(move[3]), int(move[4] or 0), qty, cost_method ) unit_cost = (total_cost / qty).quantize(Decimal("0.0001")) if qty else Decimal("0") await _write_operation_leg( db, tenant_id, operation_id=operation_id, move_id=int(move[0]), product_id=int(move[3]), variant_id=int(move[4] or 0), warehouse_id=int(move[5]), location_id=location_id, batch_id=batch_id, qty_delta=-qty, unit_cost=unit_cost, operator_id=operator_id, reason=operation[5] or "供应商退货", ) if serial_numbers: await move_serials( db, tenant_id, serial_numbers=serial_numbers, product_id=int(move[3]), variant_id=int(move[4] or 0), operation_id=operation_id, event_type="vendor_return", from_warehouse_id=int(move[5]), to_warehouse_id=0, from_location_id=location_id, to_location_id=0, from_state="sellable", to_state="returned", batch_id=batch_id, ) await db.execute( text("UPDATE inventory_moves SET source_location_id=:location,unit_cost=:cost,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"location": location_id, "cost": unit_cost, "id": move[0], "tid": tenant_id}, ) if move[2]: await db.execute(text("UPDATE inventory_purchase_order_lines SET returned_qty=returned_qty+:qty," "updated_at=NOW(3) WHERE id=:id AND tenant_id=:tid"), {"qty": qty, "id": move[2], "tid": tenant_id}) po_status = "done" else: raise HTTPException(status_code=400, detail=f"暂不支持确认操作类型 {operation[0]}") await db.execute( text("UPDATE inventory_operations SET state='done',actual_at=NOW(3),error_message=NULL,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"id": operation_id, "tid": tenant_id}, ) return po_status async def create_transfer_operation(db: AsyncSession, tenant_id: int, body) -> int: from app.plugins.inventory.purchasing import _validate_line_products existing = await _existing_operation(db, tenant_id, body.idempotency_key) if existing: return int(existing) if int(body.source_warehouse_id) == int(body.destination_warehouse_id) and not ( body.source_location_id and body.destination_location_id and int(body.source_location_id) != int(body.destination_location_id) ): raise HTTPException(status_code=400, detail="调拨的来源和目标不能相同") warehouses = (await db.execute( text("SELECT id FROM inventory_warehouses WHERE tenant_id=:tid AND is_active=1 " "AND id IN (:source,:destination)"), {"tid": tenant_id, "source": body.source_warehouse_id, "destination": body.destination_warehouse_id}, )).fetchall() if {int(row[0]) for row in warehouses} != {int(body.source_warehouse_id), int(body.destination_warehouse_id)}: raise HTTPException(status_code=400, detail="调拨仓库不存在或不属于当前租户") await _validate_line_products(db, tenant_id, [body]) from app.plugins.inventory.traceability import load_product_profile, validate_move_tracking profile = await load_product_profile(db, tenant_id, int(body.product_id), int(body.variant_id or 0)) validate_move_tracking(profile["tracking_type"], body.qty, serial_numbers=body.serial_numbers, batch_no="existing" if body.batch_id else None) source_location = await _resolve_location( db, tenant_id, int(body.source_warehouse_id), body.source_location_id ) destination_location = await _resolve_location( db, tenant_id, int(body.destination_warehouse_id), body.destination_location_id ) operation_id = await _insert_operation( db, tenant_id, operation_type="transfer", source_doc_type=None, source_doc_id=None, warehouse_id=int(body.source_warehouse_id), reference=body.reference, idempotency_key=body.idempotency_key, ) await db.execute( text("UPDATE inventory_operations SET transfer_steps=:steps WHERE id=:id AND tenant_id=:tid"), {"steps": body.steps, "id": operation_id, "tid": tenant_id}, ) await db.execute( text("INSERT INTO inventory_moves " "(tenant_id,operation_id,product_id,variant_id,source_warehouse_id,destination_warehouse_id," "source_location_id,destination_location_id,planned_qty,done_qty,batch_id,serial_numbers,source_state," "destination_state,created_at,updated_at) " "VALUES (:tid,:op,:pid,:vid,:source_wh,:destination_wh,:source_location,:destination_location," ":qty,:qty,:batch,:serials,'sellable','sellable',NOW(3),NOW(3))"), {"tid": tenant_id, "op": operation_id, "pid": body.product_id, "vid": body.variant_id or 0, "source_wh": body.source_warehouse_id, "destination_wh": body.destination_warehouse_id, "source_location": source_location, "destination_location": destination_location, "qty": body.qty, "batch": body.batch_id or 0, "serials": json.dumps(body.serial_numbers)}, ) return operation_id async def ship_transfer(db: AsyncSession, tenant_id: int, operation_id: int, operator_id: int) -> str: await assert_period_open(db, tenant_id) operation = (await db.execute( text("SELECT operation_type,state,transfer_steps FROM inventory_operations " "WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": operation_id, "tid": tenant_id}, )).fetchone() if operation is None: raise HTTPException(status_code=404, detail="调拨单不存在") if operation[0] not in ("transfer", "putaway") or operation[1] != "draft": raise HTTPException(status_code=409, detail="仅草稿调拨/上架单可发出") moves = (await db.execute( text("SELECT id,product_id,variant_id,source_warehouse_id,destination_warehouse_id," "source_location_id,destination_location_id,done_qty,batch_id,serial_numbers FROM inventory_moves " "WHERE tenant_id=:tid AND operation_id=:op ORDER BY id FOR UPDATE"), {"tid": tenant_id, "op": operation_id}, )).fetchall() if not moves: raise HTTPException(status_code=400, detail="调拨单没有明细") three_step = int(operation[2] or 1) == 3 for move in moves: qty = Decimal(str(move[7])) common = {"db": db, "tenant_id": tenant_id, "operation_id": operation_id, "move_id": int(move[0]), "product_id": int(move[1]), "variant_id": int(move[2] or 0), "batch_id": int(move[8] or 0), "unit_cost": None, "operator_id": operator_id, "reason": "库存调拨"} await _write_operation_leg( **common, warehouse_id=int(move[3]), location_id=int(move[5]), qty_delta=-qty, stock_state="sellable", leg_suffix="ship_out", ) await _write_operation_leg( **common, warehouse_id=int(move[4]), location_id=int(move[6]), qty_delta=qty, stock_state="in_transit" if three_step else "sellable", leg_suffix="ship_in_transit" if three_step else "ship_in", ) serial_numbers = list(move[9] or []) if serial_numbers: from app.plugins.inventory.traceability import move_serials await move_serials( db, tenant_id, serial_numbers=serial_numbers, product_id=int(move[1]), variant_id=int(move[2] or 0), operation_id=operation_id, event_type="transfer_ship", from_warehouse_id=int(move[3]), to_warehouse_id=int(move[4]), from_location_id=int(move[5]), to_location_id=int(move[6]), from_state="sellable", to_state="in_transit" if three_step else "sellable", batch_id=int(move[8] or 0), ) target = "in_transit" if three_step else "done" await db.execute( text("UPDATE inventory_operations SET state=:state,actual_at=CASE WHEN :state='done' THEN NOW(3) ELSE actual_at END," "updated_at=NOW(3) WHERE id=:id AND tenant_id=:tid"), {"state": target, "id": operation_id, "tid": tenant_id}, ) return target async def receive_transfer(db: AsyncSession, tenant_id: int, operation_id: int, operator_id: int) -> None: await assert_period_open(db, tenant_id) state = (await db.execute( text("SELECT state FROM inventory_operations WHERE id=:id AND tenant_id=:tid " "AND operation_type='transfer' FOR UPDATE"), {"id": operation_id, "tid": tenant_id}, )).scalar() if state is None: raise HTTPException(status_code=404, detail="调拨单不存在") if state != "in_transit": raise HTTPException(status_code=409, detail="仅在途调拨单可收货") moves = (await db.execute( text("SELECT id,product_id,variant_id,destination_warehouse_id,destination_location_id,done_qty,batch_id " ",serial_numbers " "FROM inventory_moves WHERE tenant_id=:tid AND operation_id=:op ORDER BY id FOR UPDATE"), {"tid": tenant_id, "op": operation_id}, )).fetchall() for move in moves: qty = Decimal(str(move[5])) common = {"db": db, "tenant_id": tenant_id, "operation_id": operation_id, "move_id": int(move[0]), "product_id": int(move[1]), "variant_id": int(move[2] or 0), "warehouse_id": int(move[3]), "location_id": int(move[4]), "batch_id": int(move[6] or 0), "unit_cost": None, "operator_id": operator_id, "reason": "调拨收货"} await _write_operation_leg(**common, qty_delta=-qty, stock_state="in_transit", leg_suffix="receive_out") await _write_operation_leg(**common, qty_delta=qty, stock_state="sellable", leg_suffix="receive_in") serial_numbers = list(move[7] or []) if serial_numbers: from app.plugins.inventory.traceability import move_serials await move_serials( db, tenant_id, serial_numbers=serial_numbers, product_id=int(move[1]), variant_id=int(move[2] or 0), operation_id=operation_id, event_type="transfer_receive", from_warehouse_id=int(move[3]), to_warehouse_id=int(move[3]), from_location_id=int(move[4]), to_location_id=int(move[4]), from_state="in_transit", to_state="sellable", batch_id=int(move[6] or 0), ) await db.execute(text("UPDATE inventory_operations SET state='done',actual_at=NOW(3),updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"id": operation_id, "tid": tenant_id}) async def create_and_post_putaway(db: AsyncSession, tenant_id: int, receipt_operation_id: int, operator_id: int) -> int | None: from app.plugins.inventory.locations import resolve_putaway_location receipt = (await db.execute( text("SELECT state,operation_type FROM inventory_operations WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": receipt_operation_id, "tid": tenant_id}, )).fetchone() if receipt is None or receipt[0] != "done" or receipt[1] != "receipt": raise HTTPException(status_code=409, detail="仅已确认的收货单可执行上架") existing = (await db.execute( text("SELECT id FROM inventory_operations WHERE tenant_id=:tid AND operation_type='putaway' " "AND source_doc_type='inventory_operation' AND source_doc_id=:source LIMIT 1"), {"tid": tenant_id, "source": receipt_operation_id}, )).scalar() if existing: return int(existing) rows = (await db.execute( text("SELECT id,product_id,variant_id,destination_warehouse_id,destination_location_id,done_qty,batch_id " ",serial_numbers " "FROM inventory_moves WHERE tenant_id=:tid AND operation_id=:op ORDER BY id"), {"tid": tenant_id, "op": receipt_operation_id}, )).fetchall() planned = [] for row in rows: destination = await resolve_putaway_location(db, tenant_id, int(row[3]), int(row[1])) if destination != int(row[4]): planned.append((row, destination)) if not planned: return None operation_id = await _insert_operation( db, tenant_id, operation_type="putaway", source_doc_type="inventory_operation", source_doc_id=receipt_operation_id, warehouse_id=int(planned[0][0][3]), ) for row, destination in planned: await db.execute( text("INSERT INTO inventory_moves " "(tenant_id,operation_id,product_id,variant_id,source_warehouse_id,destination_warehouse_id," "source_location_id,destination_location_id,planned_qty,done_qty,batch_id,source_state," "destination_state,serial_numbers,created_at,updated_at) VALUES " "(:tid,:op,:pid,:vid,:wh,:wh,:source,:destination,:qty,:qty,:batch,'sellable','sellable'," ":serials,NOW(3),NOW(3))"), {"tid": tenant_id, "op": operation_id, "pid": row[1], "vid": row[2] or 0, "wh": row[3], "source": row[4], "destination": destination, "qty": row[5], "batch": row[6] or 0, "serials": json.dumps(list(row[7] or []))}, ) await ship_transfer(db, tenant_id, operation_id, operator_id) return operation_id