汤学会 e90d296047 chore: customize branding for Baihua

2026-06-14 21:05:49 +08:00

66 KiB

Raw Blame History

Special Warehouse Adjustment Implementation Plan

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 为百华仓库 6 大库的入库与出库操作区分别增加“特殊入库”“特殊出库”，允许在强制填写说明后，精确到库存明细行进行受控库存调整，并把说明写入所有变更记录。

Architecture: 新增独立的“特殊调整”后端服务与审计主/明细表，库存变动仍复用现有 wh_stock_lot、wh_stock_balance、wh_inventory_txn 三层账本。前端在现有 InventoryLedgerView.vue 的仓库出入库操作矩阵中增加特殊操作入口，打开专用抽屉完成说明填写、明细选择、调整前后预览和提交。

Tech Stack: FastAPI, SQLAlchemy ORM, MySQL 8, SQLite unit tests, Vue 3, Vite, Node.js static UI scripts.

Scope And Invariants

本计划只实现“特殊入库/特殊出库”，不改变正常采购、销售、生产、盘库链路。

必须保持这些业务不变量：

原材料库：只按重量调整，精确到原材料库存批次号。
半成品库：按该半成品库存既有口径调整，按件库存只能调数量，按重库存只能调重量。
成品库：主维度按数量调整，重量作为辅助字段随同记录。
辅料库：只按数量调整。
废料库：只按重量调整，并保留来源原材料库存批次追溯展示。
退货库：按退货品库存明细调整，保留发货来源、来源原材料库存批次追溯。
特殊出库不能超过所选明细当前可用量。
盘库锁库期间禁止特殊入库和特殊出库。
操作说明必填，且必须写入特殊调整主单、明细、库存流水 remark。
特殊调整不联动采购订单、销售订单、生产工单状态，避免污染正常业务状态。
用户界面展示必须使用中文业务名，不显示英文状态或英文标志。

File Structure

Create:

backend/sql/add_special_warehouse_adjustment_patch.sql
- MySQL 生产/测试库补丁，新增特殊调整主表与明细表。
backend/app/services/special_inventory_adjustment.py
- 特殊调整核心服务，负责校验、明细变更、汇总库存更新、流水生成、审计记录。
backend/tests/test_special_warehouse_adjustment.py
- 后端 SQLite 单元测试，覆盖强制说明、原材料重量调整、辅料数量调整、成品数量调整、锁库拦截。
frontend/src/utils/specialAdjustmentRules.js
- 前端特殊调整口径与展示规则，避免把口径判断全部堆进 Vue 单文件。
frontend/scripts/test-special-adjustment-ui.mjs
- 前端静态脚本测试，检查 6 大库都出现特殊入库/特殊出库入口，检查口径规则。

Modify:

backend/app/models/operations.py
- 新增 SpecialWarehouseAdjustment、SpecialWarehouseAdjustmentLine ORM 模型。
backend/app/schemas/operations.py
- 新增特殊调整请求/响应 schema。
backend/app/api/routes/inventory.py
- 新增 /inventory/special-adjustments 接口，接入服务。
backend/app/services/operations.py
- 扩展仓库流水方向与类型中文展示能力，确保特殊调整可在流水中查到。
frontend/src/views/InventoryLedgerView.vue
- 增加 6 大库操作按钮、专用抽屉、提交逻辑、调整前后预览。

Task 1: Backend Failing Tests For Special Adjustment

Files:

Create: backend/tests/test_special_warehouse_adjustment.py
Future dependency: backend/app/services/special_inventory_adjustment.py
Future dependency: backend/app/schemas/operations.py
Step 1: Write the failing backend test file

Create backend/tests/test_special_warehouse_adjustment.py with this content:

from __future__ import annotations

import unittest
from datetime import UTC, datetime
from decimal import Decimal

from sqlalchemy import BigInteger, create_engine, select
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import Session, sessionmaker


@compiles(BigInteger, "sqlite")
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
    _ = type_, compiler, kw
    return "INTEGER"


import app.models.master_data  # noqa: E402,F401
import app.models.operations  # noqa: E402,F401
from app.models.base import Base  # noqa: E402
from app.models.master_data import Item, StockBalance, Warehouse  # noqa: E402
from app.models.operations import InventoryTxn, StockLot, WarehouseLocation  # noqa: E402
from app.schemas.operations import SpecialWarehouseAdjustmentCreate, SpecialWarehouseAdjustmentLineCreate  # noqa: E402
from app.services.special_inventory_adjustment import create_special_inventory_adjustment  # noqa: E402


class SpecialWarehouseAdjustmentTest(unittest.TestCase):
    def setUp(self) -> None:
        engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
        Base.metadata.create_all(engine)
        self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
        self.db: Session = self.SessionLocal()
        self._seed()

    def tearDown(self) -> None:
        self.db.close()

    def _seed(self) -> None:
        now = datetime.now(UTC)
        self.raw = Warehouse(
            id=1,
            warehouse_code="WH-RAW",
            warehouse_name="原材料库",
            warehouse_type="RAW",
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.aux = Warehouse(
            id=2,
            warehouse_code="WH-AUX",
            warehouse_name="辅料库",
            warehouse_type="AUX",
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.finished = Warehouse(
            id=3,
            warehouse_code="WH-FG",
            warehouse_name="成品库",
            warehouse_type="FINISHED",
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.raw_loc = WarehouseLocation(
            id=1,
            warehouse_id=1,
            location_code="RAW-A",
            location_name="原材料默认库位",
            is_default=1,
            is_locked=0,
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.aux_loc = WarehouseLocation(
            id=2,
            warehouse_id=2,
            location_code="AUX-A",
            location_name="辅料默认库位",
            is_default=1,
            is_locked=0,
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.finished_loc = WarehouseLocation(
            id=3,
            warehouse_id=3,
            location_code="FG-A",
            location_name="成品默认库位",
            is_default=1,
            is_locked=0,
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.raw_item = Item(
            id=1,
            item_code="原材料00001",
            item_name="冷轧钢板",
            item_type="RAW_MATERIAL",
            unit_weight_kg=Decimal("1"),
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.aux_item = Item(
            id=2,
            item_code="原材料00002",
            item_name="劳保手套",
            item_type="RAW_MATERIAL",
            unit_weight_kg=Decimal("0"),
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.finished_item = Item(
            id=3,
            item_code="产品需规00001",
            item_name="钢制碗",
            item_type="FINISHED_GOOD",
            unit_weight_kg=Decimal("0.6"),
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.raw_lot = StockLot(
            id=1,
            lot_no="JH_冷轧钢板00001_0001",
            lot_role="OPENING_STOCK",
            item_id=1,
            warehouse_id=1,
            location_id=1,
            source_doc_type="OPENING_STOCK",
            source_doc_id=1,
            inbound_qty=Decimal("0"),
            inbound_weight_kg=Decimal("500"),
            remaining_qty=Decimal("0"),
            remaining_weight_kg=Decimal("300"),
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=Decimal("4.5"),
            quality_status="PASS",
            status="AVAILABLE",
            created_at=now,
            updated_at=now,
        )
        self.aux_lot = StockLot(
            id=2,
            lot_no="AUX-20260606-001",
            lot_role="OPENING_STOCK",
            item_id=2,
            warehouse_id=2,
            location_id=2,
            source_doc_type="OPENING_STOCK",
            source_doc_id=2,
            inbound_qty=Decimal("120"),
            inbound_weight_kg=Decimal("0"),
            remaining_qty=Decimal("80"),
            remaining_weight_kg=Decimal("0"),
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=Decimal("2"),
            quality_status="PASS",
            status="AVAILABLE",
            created_at=now,
            updated_at=now,
        )
        self.finished_lot = StockLot(
            id=3,
            lot_no="FG-20260606-001",
            lot_role="INBOUND_FINISHED",
            item_id=3,
            warehouse_id=3,
            location_id=3,
            source_doc_type="FG_RECEIPT",
            source_doc_id=3,
            inbound_qty=Decimal("100"),
            inbound_weight_kg=Decimal("60"),
            remaining_qty=Decimal("70"),
            remaining_weight_kg=Decimal("42"),
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=Decimal("2.7"),
            quality_status="PASS",
            status="AVAILABLE",
            created_at=now,
            updated_at=now,
        )
        self.raw_balance = StockBalance(
            item_id=1,
            warehouse_id=1,
            location_id=1,
            qty_on_hand=Decimal("0"),
            weight_on_hand_kg=Decimal("300"),
            qty_available=Decimal("0"),
            weight_available_kg=Decimal("300"),
            qty_allocated=Decimal("0"),
            weight_allocated_kg=Decimal("0"),
            avg_unit_cost=Decimal("4.5"),
            created_at=now,
            updated_at=now,
        )
        self.aux_balance = StockBalance(
            item_id=2,
            warehouse_id=2,
            location_id=2,
            qty_on_hand=Decimal("80"),
            weight_on_hand_kg=Decimal("0"),
            qty_available=Decimal("80"),
            weight_available_kg=Decimal("0"),
            qty_allocated=Decimal("0"),
            weight_allocated_kg=Decimal("0"),
            avg_unit_cost=Decimal("2"),
            created_at=now,
            updated_at=now,
        )
        self.finished_balance = StockBalance(
            item_id=3,
            warehouse_id=3,
            location_id=3,
            qty_on_hand=Decimal("70"),
            weight_on_hand_kg=Decimal("42"),
            qty_available=Decimal("70"),
            weight_available_kg=Decimal("42"),
            qty_allocated=Decimal("0"),
            weight_allocated_kg=Decimal("0"),
            avg_unit_cost=Decimal("2.7"),
            created_at=now,
            updated_at=now,
        )
        self.db.add_all([
            self.raw,
            self.aux,
            self.finished,
            self.raw_loc,
            self.aux_loc,
            self.finished_loc,
            self.raw_item,
            self.aux_item,
            self.finished_item,
            self.raw_lot,
            self.aux_lot,
            self.finished_lot,
            self.raw_balance,
            self.aux_balance,
            self.finished_balance,
        ])
        self.db.commit()

    def test_special_outbound_requires_reason_before_stock_change(self) -> None:
        payload = SpecialWarehouseAdjustmentCreate(
            warehouse_id=self.raw.id,
            direction="OUT",
            reason="",
            lines=[
                SpecialWarehouseAdjustmentLineCreate(
                    item_id=self.raw_item.id,
                    lot_id=self.raw_lot.id,
                    adjust_weight_kg=Decimal("10"),
                )
            ],
        )

        with self.assertRaisesRegex(ValueError, "特殊出库必须填写说明"):
            create_special_inventory_adjustment(self.db, payload, user_id=9)

        unchanged_lot = self.db.get(StockLot, self.raw_lot.id)
        self.assertEqual(Decimal(str(unchanged_lot.remaining_weight_kg)), Decimal("300.000000"))

    def test_raw_special_outbound_updates_lot_balance_and_inventory_txn(self) -> None:
        payload = SpecialWarehouseAdjustmentCreate(
            warehouse_id=self.raw.id,
            direction="OUT",
            reason="现场发现边角料已线下报废，补做特殊出库",
            lines=[
                SpecialWarehouseAdjustmentLineCreate(
                    item_id=self.raw_item.id,
                    lot_id=self.raw_lot.id,
                    adjust_weight_kg=Decimal("25.5"),
                )
            ],
        )

        result = create_special_inventory_adjustment(self.db, payload, user_id=9)

        adjusted_lot = self.db.get(StockLot, self.raw_lot.id)
        balance = self.db.scalar(select(StockBalance).where(StockBalance.item_id == self.raw_item.id))
        txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "SPECIAL_OUT"))

        self.assertEqual(result.direction, "OUT")
        self.assertEqual(Decimal(str(adjusted_lot.remaining_weight_kg)), Decimal("274.500000"))
        self.assertEqual(Decimal(str(balance.weight_on_hand_kg)), Decimal("274.500000"))
        self.assertEqual(Decimal(str(txn.weight_change_kg)), Decimal("-25.500000"))
        self.assertIn("特殊出库说明：现场发现边角料已线下报废，补做特殊出库", txn.remark)
        self.assertIn("调整前重量 300", txn.remark)
        self.assertIn("调整后重量 274.5", txn.remark)

    def test_special_outbound_cannot_exceed_available_lot_quantity_or_weight(self) -> None:
        payload = SpecialWarehouseAdjustmentCreate(
            warehouse_id=self.aux.id,
            direction="OUT",
            reason="盘点前发现辅料账实差异，补做出库",
            lines=[
                SpecialWarehouseAdjustmentLineCreate(
                    item_id=self.aux_item.id,
                    lot_id=self.aux_lot.id,
                    adjust_qty=Decimal("81"),
                )
            ],
        )

        with self.assertRaisesRegex(ValueError, "特殊出库数量不能超过当前可用数量"):
            create_special_inventory_adjustment(self.db, payload, user_id=9)

    def test_aux_special_inbound_uses_quantity_only(self) -> None:
        payload = SpecialWarehouseAdjustmentCreate(
            warehouse_id=self.aux.id,
            direction="IN",
            reason="供应商赠送辅料漏入账，补做特殊入库",
            lines=[
                SpecialWarehouseAdjustmentLineCreate(
                    item_id=self.aux_item.id,
                    lot_id=self.aux_lot.id,
                    adjust_qty=Decimal("12"),
                    adjust_weight_kg=Decimal("99"),
                )
            ],
        )

        create_special_inventory_adjustment(self.db, payload, user_id=9)

        adjusted_lot = self.db.get(StockLot, self.aux_lot.id)
        balance = self.db.scalar(select(StockBalance).where(StockBalance.item_id == self.aux_item.id))
        txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "SPECIAL_IN"))

        self.assertEqual(Decimal(str(adjusted_lot.remaining_qty)), Decimal("92.000000"))
        self.assertEqual(Decimal(str(adjusted_lot.remaining_weight_kg)), Decimal("0.000000"))
        self.assertEqual(Decimal(str(balance.qty_on_hand)), Decimal("92.000000"))
        self.assertEqual(Decimal(str(txn.qty_change)), Decimal("12.000000"))
        self.assertEqual(Decimal(str(txn.weight_change_kg)), Decimal("0.000000"))

    def test_finished_special_outbound_uses_qty_and_keeps_weight_auxiliary(self) -> None:
        payload = SpecialWarehouseAdjustmentCreate(
            warehouse_id=self.finished.id,
            direction="OUT",
            reason="客户样品领用，补做成品特殊出库",
            lines=[
                SpecialWarehouseAdjustmentLineCreate(
                    item_id=self.finished_item.id,
                    lot_id=self.finished_lot.id,
                    adjust_qty=Decimal("5"),
                    adjust_weight_kg=Decimal("3"),
                )
            ],
        )

        create_special_inventory_adjustment(self.db, payload, user_id=9)

        adjusted_lot = self.db.get(StockLot, self.finished_lot.id)
        balance = self.db.scalar(select(StockBalance).where(StockBalance.item_id == self.finished_item.id))

        self.assertEqual(Decimal(str(adjusted_lot.remaining_qty)), Decimal("65.000000"))
        self.assertEqual(Decimal(str(adjusted_lot.remaining_weight_kg)), Decimal("39.000000"))
        self.assertEqual(Decimal(str(balance.qty_on_hand)), Decimal("65.000000"))
        self.assertEqual(Decimal(str(balance.weight_on_hand_kg)), Decimal("39.000000"))

Step 2: Run the failing backend tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_special_warehouse_adjustment.py -v

Expected: FAIL with ModuleNotFoundError: No module named 'app.services.special_inventory_adjustment' or import errors for SpecialWarehouseAdjustmentCreate.

Step 3: Commit the failing test

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/tests/test_special_warehouse_adjustment.py
git commit -m "test: cover special warehouse adjustments"

Task 2: Database And ORM Audit Tables

Files:

Create: backend/sql/add_special_warehouse_adjustment_patch.sql
Modify: backend/app/models/operations.py
Step 1: Add MySQL migration patch

Create backend/sql/add_special_warehouse_adjustment_patch.sql:

-- 特殊出入库调整主表和明细表
-- 说明：
-- 1. 特殊调整不替代库存流水，所有库存变动仍写入 wh_inventory_txn。
-- 2. 该表用于把一次特殊操作中涉及的多行明细聚合成一张审计单。

CREATE TABLE IF NOT EXISTS wh_special_adjustment (
  id BIGINT NOT NULL AUTO_INCREMENT,
  adjustment_no VARCHAR(50) NOT NULL,
  warehouse_id BIGINT NOT NULL,
  warehouse_type VARCHAR(32) NOT NULL,
  direction VARCHAR(16) NOT NULL,
  reason VARCHAR(1000) NOT NULL,
  status VARCHAR(32) NOT NULL DEFAULT 'CONFIRMED',
  operator_user_id BIGINT NULL,
  adjusted_at DATETIME NOT NULL,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (id),
  UNIQUE KEY uk_wh_special_adjustment_no (adjustment_no),
  KEY idx_wh_special_adjustment_warehouse (warehouse_id, adjusted_at),
  KEY idx_wh_special_adjustment_direction (direction, adjusted_at),
  CONSTRAINT fk_wh_special_adjustment_warehouse
    FOREIGN KEY (warehouse_id) REFERENCES wh_warehouse(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='仓库特殊出入库调整主表';

CREATE TABLE IF NOT EXISTS wh_special_adjustment_line (
  id BIGINT NOT NULL AUTO_INCREMENT,
  adjustment_id BIGINT NOT NULL,
  line_no INT NOT NULL,
  item_id BIGINT NOT NULL,
  warehouse_id BIGINT NOT NULL,
  location_id BIGINT NULL,
  lot_id BIGINT NOT NULL,
  inventory_txn_id BIGINT NULL,
  before_qty DECIMAL(18,6) NOT NULL DEFAULT 0,
  change_qty DECIMAL(18,6) NOT NULL DEFAULT 0,
  after_qty DECIMAL(18,6) NOT NULL DEFAULT 0,
  before_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  change_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  after_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  unit_cost DECIMAL(18,4) NOT NULL DEFAULT 0,
  line_reason VARCHAR(1000) NULL,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (id),
  KEY idx_wh_special_adjustment_line_adjustment (adjustment_id, line_no),
  KEY idx_wh_special_adjustment_line_lot (lot_id),
  KEY idx_wh_special_adjustment_line_txn (inventory_txn_id),
  CONSTRAINT fk_wh_special_adjustment_line_adjustment
    FOREIGN KEY (adjustment_id) REFERENCES wh_special_adjustment(id),
  CONSTRAINT fk_wh_special_adjustment_line_item
    FOREIGN KEY (item_id) REFERENCES md_item(id),
  CONSTRAINT fk_wh_special_adjustment_line_warehouse
    FOREIGN KEY (warehouse_id) REFERENCES wh_warehouse(id),
  CONSTRAINT fk_wh_special_adjustment_line_lot
    FOREIGN KEY (lot_id) REFERENCES wh_stock_lot(id),
  CONSTRAINT fk_wh_special_adjustment_line_txn
    FOREIGN KEY (inventory_txn_id) REFERENCES wh_inventory_txn(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='仓库特殊出入库调整明细表';

Step 2: Add ORM models

Modify backend/app/models/operations.py after InventoryTxn:

class SpecialWarehouseAdjustment(Base):
    __tablename__ = "wh_special_adjustment"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    adjustment_no: Mapped[str] = mapped_column(String(50))
    warehouse_id: Mapped[int] = mapped_column(ForeignKey("wh_warehouse.id"))
    warehouse_type: Mapped[str] = mapped_column(String(32))
    direction: Mapped[str] = mapped_column(String(16))
    reason: Mapped[str] = mapped_column(String(1000))
    status: Mapped[str] = mapped_column(String(32), server_default=text("'CONFIRMED'"))
    operator_user_id: Mapped[int | None] = mapped_column(ForeignKey("sys_user.id"), nullable=True)
    adjusted_at: Mapped[object] = mapped_column(DateTime)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class SpecialWarehouseAdjustmentLine(Base):
    __tablename__ = "wh_special_adjustment_line"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    adjustment_id: Mapped[int] = mapped_column(ForeignKey("wh_special_adjustment.id"))
    line_no: Mapped[int] = mapped_column(Integer)
    item_id: Mapped[int] = mapped_column(ForeignKey("md_item.id"))
    warehouse_id: Mapped[int] = mapped_column(ForeignKey("wh_warehouse.id"))
    location_id: Mapped[int | None] = mapped_column(ForeignKey("wh_location.id"), nullable=True)
    lot_id: Mapped[int] = mapped_column(ForeignKey("wh_stock_lot.id"))
    inventory_txn_id: Mapped[int | None] = mapped_column(ForeignKey("wh_inventory_txn.id"), nullable=True)
    before_qty: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    change_qty: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    after_qty: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    before_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    change_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    after_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    unit_cost: Mapped[float] = mapped_column(DECIMAL(18, 4), server_default=text("0"))
    line_reason: Mapped[str | None] = mapped_column(String(1000), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)

Step 3: Run model import check

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python - <<'PY'
import app.models.operations as operations
print(operations.SpecialWarehouseAdjustment.__tablename__)
print(operations.SpecialWarehouseAdjustmentLine.__tablename__)
PY

Expected:

wh_special_adjustment
wh_special_adjustment_line

Step 4: Re-run backend tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_special_warehouse_adjustment.py -v

Expected: still FAIL because schemas and service are not implemented yet.

Step 5: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/sql/add_special_warehouse_adjustment_patch.sql backend/app/models/operations.py
git commit -m "feat: add special adjustment audit tables"

Task 3: Backend Schemas And Core Service

Files:

Modify: backend/app/schemas/operations.py
Create: backend/app/services/special_inventory_adjustment.py
Step 1: Add request and response schemas

Modify backend/app/schemas/operations.py near the warehouse operation schemas:

class SpecialWarehouseAdjustmentLineCreate(BaseModel):
    item_id: int
    lot_id: int | None = None
    location_id: int | None = None
    adjust_qty: float | None = 0
    adjust_weight_kg: float | None = 0
    unit_cost: float | None = None
    line_reason: str | None = None


class SpecialWarehouseAdjustmentCreate(BaseModel):
    warehouse_id: int
    direction: str
    reason: str
    lines: list[SpecialWarehouseAdjustmentLineCreate] = Field(min_length=1)


class SpecialWarehouseAdjustmentLineRead(BaseModel):
    line_id: int
    line_no: int
    item_id: int
    lot_id: int
    inventory_txn_id: int | None = None
    before_qty: float
    change_qty: float
    after_qty: float
    before_weight_kg: float
    change_weight_kg: float
    after_weight_kg: float
    unit_cost: float
    line_reason: str | None = None


class SpecialWarehouseAdjustmentRead(BaseModel):
    adjustment_id: int
    adjustment_no: str
    warehouse_id: int
    warehouse_type: str
    direction: str
    reason: str
    status: str
    adjusted_at: datetime
    lines: list[SpecialWarehouseAdjustmentLineRead] = Field(default_factory=list)

Step 2: Implement service constants and helpers

Create backend/app/services/special_inventory_adjustment.py with these imports and helpers:

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime
from decimal import Decimal

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.master_data import Item, StockBalance, Warehouse
from app.models.operations import (
    InventoryTxn,
    SpecialWarehouseAdjustment,
    SpecialWarehouseAdjustmentLine,
    StockLot,
)
from app.schemas.operations import (
    SpecialWarehouseAdjustmentCreate,
    SpecialWarehouseAdjustmentLineRead,
    SpecialWarehouseAdjustmentRead,
)
from app.services.operations import create_inventory_txn, get_location_or_default, next_doc_no, to_decimal, upsert_stock_balance
from app.services.stocktake import ensure_warehouses_unlocked

SPECIAL_SOURCE_DOC_TYPE = "SPECIAL_ADJUSTMENT"
SPECIAL_IN_TXN_TYPE = "SPECIAL_IN"
SPECIAL_OUT_TXN_TYPE = "SPECIAL_OUT"
WEIGHT_ONLY_WAREHOUSE_TYPES = {"RAW", "SCRAP"}
QTY_ONLY_WAREHOUSE_TYPES = {"AUX"}
QTY_PRIMARY_WAREHOUSE_TYPES = {"FINISHED", "RETURN"}


@dataclass(frozen=True)
class AdjustmentMeasure:
    qty: Decimal
    weight: Decimal
    amount_basis: str


def _clean_required_reason(direction: str, reason: str | None) -> str:
    label = "特殊入库" if direction == "IN" else "特殊出库"
    cleaned = str(reason or "").strip()
    if not cleaned:
        raise ValueError(f"{label}必须填写说明")
    if len(cleaned) < 4:
        raise ValueError(f"{label}说明不能少于4个字")
    return cleaned


def _direction_label(direction: str) -> str:
    return "特殊入库" if direction == "IN" else "特殊出库"


def _normalize_direction(value: str | None) -> str:
    direction = str(value or "").upper()
    if direction not in {"IN", "OUT"}:
        raise ValueError("特殊调整方向只能是特殊入库或特殊出库")
    return direction


def _measure_for_warehouse(warehouse_type: str, raw_qty: Decimal, raw_weight: Decimal) -> AdjustmentMeasure:
    if warehouse_type in WEIGHT_ONLY_WAREHOUSE_TYPES:
        if raw_weight <= 0:
            raise ValueError("该仓库特殊调整必须填写重量")
        return AdjustmentMeasure(qty=Decimal("0"), weight=raw_weight, amount_basis="WEIGHT")
    if warehouse_type in QTY_ONLY_WAREHOUSE_TYPES:
        if raw_qty <= 0:
            raise ValueError("该仓库特殊调整必须填写数量")
        return AdjustmentMeasure(qty=raw_qty, weight=Decimal("0"), amount_basis="QTY")
    if warehouse_type in QTY_PRIMARY_WAREHOUSE_TYPES:
        if raw_qty <= 0:
            raise ValueError("该仓库特殊调整必须填写数量")
        return AdjustmentMeasure(qty=raw_qty, weight=max(raw_weight, Decimal("0")), amount_basis="QTY")
    if raw_qty > 0 and raw_weight > 0:
        raise ValueError("半成品特殊调整只能按件或按重选择一种口径")
    if raw_qty > 0:
        return AdjustmentMeasure(qty=raw_qty, weight=Decimal("0"), amount_basis="QTY")
    if raw_weight > 0:
        return AdjustmentMeasure(qty=Decimal("0"), weight=raw_weight, amount_basis="WEIGHT")
    raise ValueError("半成品特殊调整必须填写数量或重量")


def _format_decimal(value: Decimal) -> str:
    text = f"{value.normalize():f}"
    return text.rstrip("0").rstrip(".") if "." in text else text

Step 3: Implement new lot creation for special inbound

Append this helper to backend/app/services/special_inventory_adjustment.py:

def _build_special_lot_no(db: Session, item: Item, warehouse: Warehouse, now: datetime) -> str:
    warehouse_type = str(warehouse.warehouse_type or "").upper()
    if warehouse_type == "RAW":
        from app.services.operations import build_inventory_lot_no

        return build_inventory_lot_no(db, item.id)
    prefix = {
        "SEMI": "半成品特殊",
        "FINISHED": "成品特殊",
        "AUX": "辅料特殊",
        "SCRAP": "废料特殊",
        "RETURN": "退货特殊",
    }.get(warehouse_type, "库存特殊")
    base = f"{prefix}{now.strftime('%Y%m%d')}"
    existing_count = db.scalar(
        select(SpecialWarehouseAdjustmentLine.id)
        .join(SpecialWarehouseAdjustment, SpecialWarehouseAdjustment.id == SpecialWarehouseAdjustmentLine.adjustment_id)
        .where(SpecialWarehouseAdjustment.warehouse_id == warehouse.id)
        .order_by(SpecialWarehouseAdjustmentLine.id.desc())
        .limit(1)
    )
    next_seq = int(existing_count or 0) + 1
    candidate = f"{base}-{next_seq:04d}"
    while db.scalar(select(StockLot.id).where(StockLot.lot_no == candidate)):
        next_seq += 1
        candidate = f"{base}-{next_seq:04d}"
    return candidate


def _create_special_inbound_lot(
    db: Session,
    *,
    item: Item,
    warehouse: Warehouse,
    location_id: int | None,
    measure: AdjustmentMeasure,
    unit_cost: Decimal,
    adjustment_id: int,
    line_id: int,
    reason: str,
    now: datetime,
) -> StockLot:
    location = get_location_or_default(db, warehouse.id, location_id)
    lot = StockLot(
        lot_no=_build_special_lot_no(db, item, warehouse, now),
        lot_role="SPECIAL_ADJUSTMENT",
        item_id=item.id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        source_doc_type=SPECIAL_SOURCE_DOC_TYPE,
        source_doc_id=adjustment_id,
        source_line_id=line_id,
        source_material_summary=f"特殊入库说明：{reason}",
        inbound_qty=measure.qty,
        inbound_weight_kg=measure.weight,
        remaining_qty=measure.qty,
        remaining_weight_kg=measure.weight,
        locked_qty=Decimal("0"),
        locked_weight_kg=Decimal("0"),
        unit_cost=unit_cost,
        production_date=now.date(),
        quality_status="PASS",
        status="AVAILABLE",
        remark=f"特殊入库说明：{reason}"[:255],
        created_at=now,
        updated_at=now,
    )
    db.add(lot)
    db.flush()
    return lot

Step 4: Implement main service

Append this function to backend/app/services/special_inventory_adjustment.py:

def create_special_inventory_adjustment(
    db: Session,
    payload: SpecialWarehouseAdjustmentCreate,
    *,
    user_id: int | None,
) -> SpecialWarehouseAdjustmentRead:
    direction = _normalize_direction(payload.direction)
    reason = _clean_required_reason(direction, payload.reason)
    warehouse = db.get(Warehouse, payload.warehouse_id)
    if not warehouse:
        raise ValueError("仓库不存在")
    ensure_warehouses_unlocked(db, [warehouse.id], _direction_label(direction))
    warehouse_type = str(warehouse.warehouse_type or "").upper()
    now = datetime.now()
    adjustment = SpecialWarehouseAdjustment(
        adjustment_no=next_doc_no(db, SpecialWarehouseAdjustment, "adjustment_no", f"特殊调整{now.strftime('%Y%m%d')}"),
        warehouse_id=warehouse.id,
        warehouse_type=warehouse_type,
        direction=direction,
        reason=reason,
        status="CONFIRMED",
        operator_user_id=user_id,
        adjusted_at=now,
        created_at=now,
        updated_at=now,
    )
    db.add(adjustment)
    db.flush()

    read_lines: list[SpecialWarehouseAdjustmentLineRead] = []
    for index, payload_line in enumerate(payload.lines, start=1):
        item = db.get(Item, payload_line.item_id)
        if not item:
            raise ValueError(f"第 {index} 行物料不存在")

        raw_qty = to_decimal(payload_line.adjust_qty)
        raw_weight = to_decimal(payload_line.adjust_weight_kg)
        if raw_qty < 0 or raw_weight < 0:
            raise ValueError("特殊调整数量和重量不能小于0")
        measure = _measure_for_warehouse(warehouse_type, raw_qty, raw_weight)

        lot = db.get(StockLot, payload_line.lot_id) if payload_line.lot_id else None
        if direction == "OUT" and lot is None:
            raise ValueError("特殊出库必须选择库存明细")
        if lot is not None:
            if lot.item_id != item.id or lot.warehouse_id != warehouse.id:
                raise ValueError("库存明细与所选物料或仓库不一致")
            if payload_line.location_id and lot.location_id != payload_line.location_id:
                raise ValueError("库存明细不属于所选库位")

        if direction == "IN" and lot is None:
            unit_cost = to_decimal(payload_line.unit_cost, "0.0001")
            placeholder_line = SpecialWarehouseAdjustmentLine(
                adjustment_id=adjustment.id,
                line_no=index,
                item_id=item.id,
                warehouse_id=warehouse.id,
                location_id=payload_line.location_id,
                lot_id=0,
                before_qty=Decimal("0"),
                change_qty=measure.qty,
                after_qty=measure.qty,
                before_weight_kg=Decimal("0"),
                change_weight_kg=measure.weight,
                after_weight_kg=measure.weight,
                unit_cost=unit_cost,
                line_reason=payload_line.line_reason,
                created_at=now,
                updated_at=now,
            )
            db.add(placeholder_line)
            db.flush()
            lot = _create_special_inbound_lot(
                db,
                item=item,
                warehouse=warehouse,
                location_id=payload_line.location_id,
                measure=measure,
                unit_cost=unit_cost,
                adjustment_id=adjustment.id,
                line_id=placeholder_line.id,
                reason=reason,
                now=now,
            )
            placeholder_line.lot_id = lot.id
            db.add(placeholder_line)
            line = placeholder_line
            before_qty = Decimal("0")
            before_weight = Decimal("0")
        else:
            assert lot is not None
            before_qty = to_decimal(lot.remaining_qty)
            before_weight = to_decimal(lot.remaining_weight_kg)
            unit_cost = to_decimal(payload_line.unit_cost, "0.0001") if payload_line.unit_cost is not None else to_decimal(lot.unit_cost, "0.0001")
            if direction == "OUT":
                available_qty = before_qty - to_decimal(lot.locked_qty)
                available_weight = before_weight - to_decimal(lot.locked_weight_kg)
                if measure.qty > 0 and measure.qty > available_qty:
                    raise ValueError("特殊出库数量不能超过当前可用数量")
                if measure.weight > 0 and measure.weight > available_weight:
                    raise ValueError("特殊出库重量不能超过当前可用重量")
                change_qty = -measure.qty
                change_weight = -measure.weight
            else:
                change_qty = measure.qty
                change_weight = measure.weight
            line = SpecialWarehouseAdjustmentLine(
                adjustment_id=adjustment.id,
                line_no=index,
                item_id=item.id,
                warehouse_id=warehouse.id,
                location_id=lot.location_id,
                lot_id=lot.id,
                before_qty=before_qty,
                change_qty=change_qty,
                after_qty=before_qty + change_qty,
                before_weight_kg=before_weight,
                change_weight_kg=change_weight,
                after_weight_kg=before_weight + change_weight,
                unit_cost=unit_cost,
                line_reason=payload_line.line_reason,
                created_at=now,
                updated_at=now,
            )
            db.add(line)
            db.flush()
            lot.remaining_qty = line.after_qty
            lot.remaining_weight_kg = line.after_weight_kg
            lot.status = "AVAILABLE" if line.after_qty > 0 or line.after_weight_kg > 0 else "DEPLETED"
            lot.updated_at = now
            db.add(lot)

        line_reason = str(payload_line.line_reason or "").strip()
        direction_text = _direction_label(direction)
        remark_parts = [
            f"{direction_text}说明：{reason}",
            f"调整前数量 {_format_decimal(to_decimal(line.before_qty))}",
            f"调整数量 {_format_decimal(to_decimal(line.change_qty))}",
            f"调整后数量 {_format_decimal(to_decimal(line.after_qty))}",
            f"调整前重量 {_format_decimal(to_decimal(line.before_weight_kg))}",
            f"调整重量 {_format_decimal(to_decimal(line.change_weight_kg))}",
            f"调整后重量 {_format_decimal(to_decimal(line.after_weight_kg))}",
        ]
        if line_reason:
            remark_parts.append(f"明细说明：{line_reason}")
        txn = create_inventory_txn(
            db,
            txn_type=SPECIAL_IN_TXN_TYPE if direction == "IN" else SPECIAL_OUT_TXN_TYPE,
            item_id=item.id,
            warehouse_id=warehouse.id,
            location_id=lot.location_id,
            lot_id=lot.id,
            qty_change=to_decimal(line.change_qty),
            weight_change=to_decimal(line.change_weight_kg),
            unit_cost=unit_cost,
            source_doc_type=SPECIAL_SOURCE_DOC_TYPE,
            source_doc_id=adjustment.id,
            source_line_id=line.id,
            biz_time=now,
            operator_user_id=user_id,
            remark="；".join(remark_parts)[:255],
            amount_basis=measure.amount_basis,
        )
        line.inventory_txn_id = txn.id
        db.add(line)
        upsert_stock_balance(
            db,
            item_id=item.id,
            warehouse_id=warehouse.id,
            location_id=lot.location_id,
            qty_delta=to_decimal(line.change_qty),
            weight_delta=to_decimal(line.change_weight_kg),
            available_qty_delta=to_decimal(line.change_qty),
            available_weight_delta=to_decimal(line.change_weight_kg),
            unit_cost=unit_cost,
        )
        read_lines.append(
            SpecialWarehouseAdjustmentLineRead(
                line_id=line.id,
                line_no=line.line_no,
                item_id=line.item_id,
                lot_id=line.lot_id,
                inventory_txn_id=txn.id,
                before_qty=float(line.before_qty),
                change_qty=float(line.change_qty),
                after_qty=float(line.after_qty),
                before_weight_kg=float(line.before_weight_kg),
                change_weight_kg=float(line.change_weight_kg),
                after_weight_kg=float(line.after_weight_kg),
                unit_cost=float(line.unit_cost),
                line_reason=line.line_reason,
            )
        )

    db.commit()
    return SpecialWarehouseAdjustmentRead(
        adjustment_id=adjustment.id,
        adjustment_no=adjustment.adjustment_no,
        warehouse_id=adjustment.warehouse_id,
        warehouse_type=adjustment.warehouse_type,
        direction=adjustment.direction,
        reason=adjustment.reason,
        status=adjustment.status,
        adjusted_at=adjustment.adjusted_at,
        lines=read_lines,
    )

Step 5: Run the backend tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_special_warehouse_adjustment.py -v

Expected: PASS for the new test file. If SQLite rejects lot_id=0 because of a foreign key in the placeholder line, adjust the implementation by creating the special inbound StockLot first with source_line_id=0, then create SpecialWarehouseAdjustmentLine, then update StockLot.source_line_id to the line id before commit.

Step 6: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/schemas/operations.py backend/app/services/special_inventory_adjustment.py backend/tests/test_special_warehouse_adjustment.py
git commit -m "feat: implement special warehouse adjustment service"

Task 4: API Route And Ledger Visibility

Files:

Modify: backend/app/api/routes/inventory.py
Modify: backend/app/services/operations.py
Modify: backend/tests/test_special_warehouse_adjustment.py
Step 1: Extend tests with API-style route behavior

Append to SpecialWarehouseAdjustmentTest:

    def test_special_adjustment_ledger_direction_and_chinese_remark(self) -> None:
        from app.services.operations import get_inventory_txn_ledger_query

        payload = SpecialWarehouseAdjustmentCreate(
            warehouse_id=self.raw.id,
            direction="OUT",
            reason="仓管复核后确认线下已处理，补做特殊出库",
            lines=[
                SpecialWarehouseAdjustmentLineCreate(
                    item_id=self.raw_item.id,
                    lot_id=self.raw_lot.id,
                    adjust_weight_kg=Decimal("6"),
                )
            ],
        )
        create_special_inventory_adjustment(self.db, payload, user_id=9)

        row = self.db.execute(
            get_inventory_txn_ledger_query(warehouse_type="RAW", txn_type="SPECIAL_OUT")
        ).mappings().first()

        self.assertIsNotNone(row)
        self.assertEqual(row["direction"], "OUT")
        self.assertEqual(row["txn_type"], "SPECIAL_OUT")
        self.assertIn("特殊出库说明：仓管复核后确认线下已处理，补做特殊出库", row["remark"])

Step 2: Run route/ledger test before modifying ledger labels

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_special_warehouse_adjustment.py::SpecialWarehouseAdjustmentTest::test_special_adjustment_ledger_direction_and_chinese_remark -v

Expected: PASS or FAIL only if SPECIAL_OUT is not included in ledger filtering/summary logic. If it already passes, still complete Step 3 to add explicit label coverage.

Step 3: Add route imports and endpoint

Modify imports in backend/app/api/routes/inventory.py:

from app.schemas.operations import (
    # existing imports...
    SpecialWarehouseAdjustmentCreate,
    SpecialWarehouseAdjustmentRead,
)
from app.services.special_inventory_adjustment import create_special_inventory_adjustment

Add this endpoint near the generic inbound/outbound endpoints:

@router.post("/special-adjustments", response_model=SpecialWarehouseAdjustmentRead)
def create_inventory_special_adjustment(
    payload: SpecialWarehouseAdjustmentCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> SpecialWarehouseAdjustmentRead:
    try:
        return create_special_inventory_adjustment(db, payload, user_id=context.user.id)
    except ValueError as exc:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(exc)) from exc

Step 4: Add Chinese business labels for special types

Modify _INVENTORY_BIZ_TYPE_LABELS in backend/app/api/routes/inventory.py:

    "SPECIAL_IN": "特殊入库",
    "SPECIAL_OUT": "特殊出库",

Modify any front/back ledger display helper that maps raw txn_type to labels if one exists. If no helper exists, keep txn_type as internal code but make sure remark contains the Chinese label because current warehouse ledger UI searches and displays remarks.

Step 5: Run focused backend tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_special_warehouse_adjustment.py tests/test_warehouse_transaction_ledger.py -v

Expected: PASS.

Step 6: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/api/routes/inventory.py backend/app/services/operations.py backend/tests/test_special_warehouse_adjustment.py
git commit -m "feat: expose special warehouse adjustment api"

Task 5: Frontend Rules And Static Tests

Files:

Create: frontend/src/utils/specialAdjustmentRules.js
Create: frontend/scripts/test-special-adjustment-ui.mjs
Step 1: Add frontend measure rules

Create frontend/src/utils/specialAdjustmentRules.js:

export function normalizeWarehouseType(value) {
  return String(value || "").toUpperCase();
}

export function specialAdjustmentDirectionLabel(direction) {
  return String(direction || "").toUpperCase() === "OUT" ? "特殊出库" : "特殊入库";
}

export function specialAdjustmentMeasureMode(warehouseType, row = {}) {
  const normalizedType = normalizeWarehouseType(warehouseType || row.warehouse_type);
  if (["RAW", "SCRAP"].includes(normalizedType)) {
    return { mode: "WEIGHT", primaryLabel: "调整重量(kg)", showQty: false, showWeight: true };
  }
  if (normalizedType === "AUX") {
    return { mode: "QTY", primaryLabel: "调整数量", showQty: true, showWeight: false };
  }
  if (["FINISHED", "RETURN"].includes(normalizedType)) {
    return { mode: "QTY_WITH_WEIGHT", primaryLabel: "调整数量", showQty: true, showWeight: true };
  }
  const hasQty = Number(row?.remaining_qty || row?.qty_available || 0) > 0;
  const hasWeight = Number(row?.remaining_weight_kg || row?.weight_available_kg || 0) > 0;
  if (hasQty && !hasWeight) {
    return { mode: "QTY", primaryLabel: "调整数量", showQty: true, showWeight: false };
  }
  if (hasWeight && !hasQty) {
    return { mode: "WEIGHT", primaryLabel: "调整重量(kg)", showQty: false, showWeight: true };
  }
  return { mode: "SEMI_SELECT", primaryLabel: "调整口径", showQty: true, showWeight: true };
}

export function specialAdjustmentOperationKeys() {
  return {
    raw: ["raw-special-in", "raw-special-out"],
    semi: ["semi-special-in", "semi-special-out"],
    finished: ["finished-special-in", "finished-special-out"],
    auxiliary: ["aux-special-in", "aux-special-out"],
    scrap: ["scrap-special-in", "scrap-special-out"],
    return: ["return-special-in", "return-special-out"]
  };
}

Step 2: Add static UI test script

Create frontend/scripts/test-special-adjustment-ui.mjs:

import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import { specialAdjustmentMeasureMode, specialAdjustmentOperationKeys } from "../src/utils/specialAdjustmentRules.js";

const viewSource = readFileSync(new URL("../src/views/InventoryLedgerView.vue", import.meta.url), "utf8");
const operationKeys = specialAdjustmentOperationKeys();

for (const [warehouseTab, keys] of Object.entries(operationKeys)) {
  for (const key of keys) {
    assert.ok(viewSource.includes(`makeOperation("${key}"`), `${warehouseTab} 缺少 ${key}`);
  }
}

assert.ok(viewSource.includes("特殊入库"), "仓库页面必须展示特殊入库");
assert.ok(viewSource.includes("特殊出库"), "仓库页面必须展示特殊出库");
assert.ok(viewSource.includes("/inventory/special-adjustments"), "仓库页面必须调用特殊调整接口");

assert.deepEqual(specialAdjustmentMeasureMode("RAW"), {
  mode: "WEIGHT",
  primaryLabel: "调整重量(kg)",
  showQty: false,
  showWeight: true
});

assert.deepEqual(specialAdjustmentMeasureMode("AUX"), {
  mode: "QTY",
  primaryLabel: "调整数量",
  showQty: true,
  showWeight: false
});

assert.equal(specialAdjustmentMeasureMode("FINISHED").mode, "QTY_WITH_WEIGHT");
assert.equal(specialAdjustmentMeasureMode("SCRAP").mode, "WEIGHT");
assert.equal(specialAdjustmentMeasureMode("SEMI", { remaining_qty: 10, remaining_weight_kg: 0 }).mode, "QTY");
assert.equal(specialAdjustmentMeasureMode("SEMI", { remaining_qty: 0, remaining_weight_kg: 10 }).mode, "WEIGHT");

Step 3: Run the failing frontend test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-special-adjustment-ui.mjs

Expected: FAIL because InventoryLedgerView.vue does not yet include special adjustment operation keys or API call.

Step 4: Commit failing frontend tests

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add frontend/src/utils/specialAdjustmentRules.js frontend/scripts/test-special-adjustment-ui.mjs
git commit -m "test: cover special adjustment warehouse ui"

Task 6: Frontend Operation Buttons And Drawer

Files:

Modify: frontend/src/views/InventoryLedgerView.vue
Step 1: Import frontend rules

In frontend/src/views/InventoryLedgerView.vue, add:

import { specialAdjustmentMeasureMode, specialAdjustmentDirectionLabel } from "../utils/specialAdjustmentRules.js";

Step 2: Add special operation buttons to all six warehouses

Modify operationMatrix so each inbound list ends with 特殊入库 and each outbound list ends with 特殊出库:

makeOperation("raw-special-in", "in", "SPECIAL_IN", "特殊入库", "强制填写说明后，直接调整原材料库库存明细。", "specialAdjustment")
makeOperation("raw-special-out", "out", "SPECIAL_OUT", "特殊出库", "强制填写说明后，直接扣减原材料库库存明细。", "specialAdjustment")
makeOperation("semi-special-in", "in", "SPECIAL_IN", "特殊入库", "强制填写说明后，直接调整半成品库库存明细。", "specialAdjustment")
makeOperation("semi-special-out", "out", "SPECIAL_OUT", "特殊出库", "强制填写说明后，直接扣减半成品库库存明细。", "specialAdjustment")
makeOperation("finished-special-in", "in", "SPECIAL_IN", "特殊入库", "强制填写说明后，直接调整成品库库存明细。", "specialAdjustment")
makeOperation("finished-special-out", "out", "SPECIAL_OUT", "特殊出库", "强制填写说明后，直接扣减成品库库存明细。", "specialAdjustment")
makeOperation("aux-special-in", "in", "SPECIAL_IN", "特殊入库", "强制填写说明后，直接调整辅料库库存明细。", "specialAdjustment")
makeOperation("aux-special-out", "out", "SPECIAL_OUT", "特殊出库", "强制填写说明后，直接扣减辅料库库存明细。", "specialAdjustment")
makeOperation("scrap-special-in", "in", "SPECIAL_IN", "特殊入库", "强制填写说明后，直接调整废料库库存明细。", "specialAdjustment")
makeOperation("scrap-special-out", "out", "SPECIAL_OUT", "特殊出库", "强制填写说明后，直接扣减废料库库存明细。", "specialAdjustment")
makeOperation("return-special-in", "in", "SPECIAL_IN", "特殊入库", "强制填写说明后，直接调整退货库库存明细。", "specialAdjustment")
makeOperation("return-special-out", "out", "SPECIAL_OUT", "特殊出库", "强制填写说明后，直接扣减退货库库存明细。", "specialAdjustment")

Use the same visual style as existing operation cards. Do not introduce a separate visual language.

Step 3: Add drawer state

Add near other drawer state:

const specialAdjustmentDrawerOpen = ref(false);
const specialAdjustmentSubmitting = ref(false);
const specialAdjustmentForm = reactive({
  direction: "IN",
  warehouse_id: "",
  reason: "",
  lines: []
});

Add reset helpers:

function resetSpecialAdjustmentForm() {
  specialAdjustmentForm.direction = activeOperation.value?.direction === "out" ? "OUT" : "IN";
  specialAdjustmentForm.warehouse_id = activeWarehouses.value[0]?.id || "";
  specialAdjustmentForm.reason = "";
  specialAdjustmentForm.lines = [];
}

function closeSpecialAdjustmentDrawer() {
  specialAdjustmentDrawerOpen.value = false;
}

Step 4: Route special operations to the drawer

Modify openOperationDrawer(operation) before the generic drawer fallback:

  if (operation.drawerType === "specialAdjustment") {
    resetSpecialAdjustmentForm();
    specialAdjustmentDrawerOpen.value = true;
    return;
  }

Step 5: Add computed lot options and preview helpers

Add computed helpers:

const specialAdjustmentDirectionText = computed(() => specialAdjustmentDirectionLabel(specialAdjustmentForm.direction));

const specialAdjustmentLotOptions = computed(() => {
  const warehouseId = Number(specialAdjustmentForm.warehouse_id || 0);
  return stockLots.value
    .filter((lot) => Number(lot.warehouse_id || 0) === warehouseId)
    .sort((left, right) => {
      const leftAvailable = Number(left.remaining_qty || 0) > 0 || Number(left.remaining_weight_kg || 0) > 0 ? 0 : 1;
      const rightAvailable = Number(right.remaining_qty || 0) > 0 || Number(right.remaining_weight_kg || 0) > 0 ? 0 : 1;
      return leftAvailable - rightAvailable || String(left.lot_no || "").localeCompare(String(right.lot_no || ""), "zh-CN", { numeric: true });
    });
});

function buildSpecialAdjustmentEmptyLine() {
  return {
    lot_id: "",
    item_id: "",
    item_name: "",
    lot_no: "",
    adjust_qty: "",
    adjust_weight_kg: "",
    unit_cost: "",
    line_reason: ""
  };
}

function addSpecialAdjustmentLine() {
  if (!specialAdjustmentForm.reason.trim()) {
    errorMessage.value = `${specialAdjustmentDirectionText.value}必须先填写说明`;
    return;
  }
  specialAdjustmentForm.lines.push(buildSpecialAdjustmentEmptyLine());
}

function removeSpecialAdjustmentLine(index) {
  specialAdjustmentForm.lines.splice(index, 1);
}

function applySpecialAdjustmentLot(line) {
  const lot = specialAdjustmentLotOptions.value.find((item) => Number(item.lot_id) === Number(line.lot_id));
  if (!lot) {
    return;
  }
  line.item_id = lot.item_id;
  line.item_name = lot.item_name;
  line.lot_no = lot.lot_no;
  line.unit_cost = lot.unit_cost || 0;
  const measure = specialAdjustmentMeasureMode(activeWarehouseType.value, lot);
  line.measure_mode = measure.mode;
}

function specialAdjustmentLinePreview(line) {
  const lot = specialAdjustmentLotOptions.value.find((item) => Number(item.lot_id) === Number(line.lot_id));
  if (!lot) {
    return "请选择库存明细";
  }
  const directionMultiplier = specialAdjustmentForm.direction === "OUT" ? -1 : 1;
  const qtyChange = Number(line.adjust_qty || 0) * directionMultiplier;
  const weightChange = Number(line.adjust_weight_kg || 0) * directionMultiplier;
  const afterQty = Number(lot.remaining_qty || 0) + qtyChange;
  const afterWeight = Number(lot.remaining_weight_kg || 0) + weightChange;
  return `调整前：${formatQty(lot.remaining_qty)} / ${formatWeight(lot.remaining_weight_kg)}；调整后：${formatQty(afterQty)} / ${formatWeight(afterWeight)}`;
}

Step 6: Add drawer template

Add a SideDrawer after the generic drawer or near other operation drawers:

<SideDrawer
  :open="specialAdjustmentDrawerOpen"
  :eyebrow="`${activeInventoryLabel} · ${specialAdjustmentDirectionText}`"
  :title="specialAdjustmentDirectionText"
  tag="特殊调整"
  @close="closeSpecialAdjustmentDrawer"
>
  <form class="stack-form" @submit.prevent="submitSpecialAdjustment">
    <div class="warning-card">
      <strong>特殊调整会直接改变库存明细</strong>
      <span>请只在补录、纠错、线下已处理等特殊场景使用。保存后系统会写入库存流水和特殊调整记录。</span>
    </div>

    <label class="field">
      <span>调整说明</span>
      <textarea
        v-model.trim="specialAdjustmentForm.reason"
        required
        rows="3"
        placeholder="必须填写，例如：现场复核发现该批次已线下处理，补做特殊出库"
      />
    </label>

    <label class="field">
      <span>仓库</span>
      <select v-model.number="specialAdjustmentForm.warehouse_id" required>
        <option disabled value="">请选择仓库</option>
        <option v-for="warehouse in activeWarehouses" :key="warehouse.id" :value="warehouse.id">
          {{ warehouse.warehouse_name }}
        </option>
      </select>
    </label>

    <div class="drawer-table-actions">
      <button class="ghost-button" type="button" @click="addSpecialAdjustmentLine">添加调整明细</button>
    </div>

    <div class="drawer-table-wrap">
      <table class="drawer-data-table">
        <thead>
          <tr>
            <th>库存明细</th>
            <th>调整数量</th>
            <th>调整重量(kg)</th>
            <th>明细说明</th>
            <th>调整预览</th>
            <th>操作</th>
          </tr>
        </thead>
        <tbody>
          <tr v-for="(line, index) in specialAdjustmentForm.lines" :key="index">
            <td>
              <select v-model.number="line.lot_id" required @change="applySpecialAdjustmentLot(line)">
                <option disabled value="">请选择库存明细</option>
                <option v-for="lot in specialAdjustmentLotOptions" :key="lot.lot_id" :value="lot.lot_id">
                  {{ lot.item_name }} · {{ lot.lot_no }} · 数量 {{ formatQty(lot.remaining_qty) }} · 重量 {{ formatWeight(lot.remaining_weight_kg) }}
                </option>
              </select>
            </td>
            <td>
              <input v-model.number="line.adjust_qty" type="number" min="0" step="0.000001" />
            </td>
            <td>
              <input v-model.number="line.adjust_weight_kg" type="number" min="0" step="0.001" />
            </td>
            <td>
              <input v-model.trim="line.line_reason" type="text" placeholder="可选，补充单行说明" />
            </td>
            <td>{{ specialAdjustmentLinePreview(line) }}</td>
            <td>
              <button class="link-button danger" type="button" @click="removeSpecialAdjustmentLine(index)">删除</button>
            </td>
          </tr>
          <tr v-if="!specialAdjustmentForm.lines.length">
            <td colspan="6">请先填写调整说明，再添加调整明细。</td>
          </tr>
        </tbody>
      </table>
    </div>

    <div class="drawer-actions">
      <button class="ghost-button" type="button" @click="closeSpecialAdjustmentDrawer">取消</button>
      <button class="primary-button" type="submit" :disabled="specialAdjustmentSubmitting">
        {{ specialAdjustmentSubmitting ? "保存中..." : "确认保存" }}
      </button>
    </div>
  </form>
</SideDrawer>

Step 7: Add submit validation and API call

Add:

function validateSpecialAdjustmentBeforeSubmit() {
  if (!specialAdjustmentForm.reason.trim()) {
    return `${specialAdjustmentDirectionText.value}必须填写说明`;
  }
  if (!specialAdjustmentForm.lines.length) {
    return `${specialAdjustmentDirectionText.value}必须至少添加一条明细`;
  }
  for (const [index, line] of specialAdjustmentForm.lines.entries()) {
    if (!Number(line.lot_id || 0)) {
      return `第${index + 1}行必须选择库存明细`;
    }
    const lot = specialAdjustmentLotOptions.value.find((item) => Number(item.lot_id) === Number(line.lot_id));
    if (!lot) {
      return `第${index + 1}行库存明细不存在`;
    }
    const measure = specialAdjustmentMeasureMode(activeWarehouseType.value, lot);
    const qty = Number(line.adjust_qty || 0);
    const weight = Number(line.adjust_weight_kg || 0);
    if (measure.mode === "WEIGHT" && weight <= 0) {
      return `第${index + 1}行必须填写调整重量`;
    }
    if (["QTY", "QTY_WITH_WEIGHT"].includes(measure.mode) && qty <= 0) {
      return `第${index + 1}行必须填写调整数量`;
    }
    if (specialAdjustmentForm.direction === "OUT") {
      if (qty > Number(lot.remaining_qty || 0)) {
        return `第${index + 1}行调整数量不能超过当前可用数量`;
      }
      if (weight > Number(lot.remaining_weight_kg || 0)) {
        return `第${index + 1}行调整重量不能超过当前可用重量`;
      }
    }
  }
  return "";
}

async function submitSpecialAdjustment() {
  resetFeedback();
  const validationError = validateSpecialAdjustmentBeforeSubmit();
  if (validationError) {
    errorMessage.value = validationError;
    return;
  }
  const confirmed = window.confirm(`确认执行${specialAdjustmentDirectionText.value}？保存后会直接改变库存，并写入库存流水。`);
  if (!confirmed) {
    return;
  }
  specialAdjustmentSubmitting.value = true;
  try {
    const payload = {
      warehouse_id: Number(specialAdjustmentForm.warehouse_id),
      direction: specialAdjustmentForm.direction,
      reason: specialAdjustmentForm.reason.trim(),
      lines: specialAdjustmentForm.lines.map((line) => ({
        item_id: Number(line.item_id),
        lot_id: Number(line.lot_id),
        adjust_qty: Number(line.adjust_qty || 0),
        adjust_weight_kg: Number(line.adjust_weight_kg || 0),
        unit_cost: Number(line.unit_cost || 0),
        line_reason: line.line_reason || ""
      }))
    };
    const result = await postResource("/inventory/special-adjustments", payload);
    feedbackMessage.value = `${specialAdjustmentDirectionText.value} ${result.adjustment_no} 已保存`;
    closeSpecialAdjustmentDrawer();
    await loadAll();
  } catch (error) {
    errorMessage.value = error.message || `${specialAdjustmentDirectionText.value}保存失败`;
  } finally {
    specialAdjustmentSubmitting.value = false;
  }
}

Step 8: Run frontend static test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-special-adjustment-ui.mjs

Expected: PASS.

Step 9: Run frontend build

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build

Expected: PASS and Vite build completes.

Step 10: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add frontend/src/views/InventoryLedgerView.vue frontend/src/utils/specialAdjustmentRules.js frontend/scripts/test-special-adjustment-ui.mjs
git commit -m "feat: add special adjustment warehouse UI"

Task 7: Integration Verification And Manual Smoke Checks

Files:

No new files expected.
Verify changed files from prior tasks.
Step 1: Run focused backend suite

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest \
  tests/test_special_warehouse_adjustment.py \
  tests/test_warehouse_transaction_ledger.py \
  tests/test_stocktake_locking.py \
  tests/test_raw_material_weight_only_flow.py \
  tests/test_auxiliary_quantity_flow.py \
  tests/test_semi_finished_operation_inventory.py \
  tests/test_scrap_warehouse_flow.py \
  tests/test_return_warehouse_flow.py \
  -v

Expected: PASS.

Step 2: Run focused frontend scripts

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-special-adjustment-ui.mjs
node scripts/test-inventory-ledger-options.mjs
node scripts/test-warehouse-drawer-table-overflow.mjs
npm run build

Expected: all scripts pass and build completes.

Step 3: Apply SQL patch to test database

Run this only after confirming the target database is the intended test database:

mysql \
  -h <测试库IP> \
  -P <测试库端口> \
  -u <用户名> \
  -p \
  --default-character-set=utf8mb4 \
  jiaheng_erp < /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend/sql/add_special_warehouse_adjustment_patch.sql

Expected: no SQL error. Verify:

SHOW TABLES LIKE 'wh_special_adjustment%';

Expected:

wh_special_adjustment
wh_special_adjustment_line

Step 4: Browser smoke check

Start backend and frontend in the project’s normal local way, then open:

http://127.0.0.1:5173/

Manual checks:

百华仓库 6 大库的入库操作区都有“特殊入库”。
百华仓库 6 大库的出库操作区都有“特殊出库”。
未填写说明时，不能添加/提交特殊调整明细。
原材料库特殊出库只能填写重量，提交后对应库存批次重量减少。
辅料库特殊入库只按数量增加，不写入重量。
成品库特殊出库按数量减少，重量作为辅助字段同步减少。
保存后打开当前库“流水”图标，能看到 特殊入库 或 特殊出库 对应流水，备注里包含完整说明。
盘库锁库状态下，特殊入库/特殊出库接口返回禁止操作提示。
Step 5: Final commit if smoke checks required small fixes

If manual smoke checks required fixes:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend frontend
git commit -m "fix: polish special warehouse adjustment flow"

If no fixes were needed, do not create an empty commit.

Self-Review

Spec coverage:

6 大库入库/出库均添加特殊操作：Task 6 Step 2 and Task 5 static test.
可直接操作仓库数据并精确到每行明细：Task 3 service updates StockLot and StockBalance per selected lot_id.
操作前必须写说明：Task 1 mandatory reason test, Task 3 _clean_required_reason, Task 6 front-end validation.
说明体现在各变更记录中：Task 3 writes reason into SpecialWarehouseAdjustment.reason, SpecialWarehouseAdjustmentLine.line_reason, and InventoryTxn.remark.
现有 6 大库不同口径：Task 3 _measure_for_warehouse, Task 5 specialAdjustmentMeasureMode.
不能只有图标没有业务逻辑：Task 3 backend service, Task 4 API, Task 6 frontend submit, Task 7 smoke checks.

Placeholder scan:

No unresolved placeholder markers or deferred implementation phrases.
The only conditional note is an execution-time SQLite foreign key fallback in Task 3 Step 5; it gives the exact alternate implementation path.

Type consistency:

SpecialWarehouseAdjustmentCreate, SpecialWarehouseAdjustmentLineCreate, and SpecialWarehouseAdjustmentRead are defined before service use.
Service function is consistently named create_special_inventory_adjustment.
Frontend API endpoint is consistently /inventory/special-adjustments.
Internal transaction types are consistently SPECIAL_IN and SPECIAL_OUT; user-facing labels remain Chinese.

66 KiB Raw Blame History Unescape Escape

Special Warehouse Adjustment Implementation Plan

Scope And Invariants

File Structure

Task 1: Backend Failing Tests For Special Adjustment

Task 2: Database And ORM Audit Tables

Task 3: Backend Schemas And Core Service

Task 4: API Route And Ledger Visibility

Task 5: Frontend Rules And Static Tests

Task 6: Frontend Operation Buttons And Drawer

Task 7: Integration Verification And Manual Smoke Checks

Self-Review

66 KiB

Raw Blame History