ForgeFlow-ERP/docs/superpowers/plans/2026-05-30-stocktake-locking.md

# Locked Warehouse Stocktake 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:** Build a "盘库" feature in 嘉恒仓库 that locks selected warehouses, exports stocktake sheets, imports counted results, shows a side-by-side diff, posts adjustments, records audit logs, and unlocks warehouses after confirmation.

**Architecture:** Backend owns stocktake locking, snapshots, Excel import/export, diff calculation, and inventory adjustment transactions. Frontend adds a stocktake entry above the warehouse bar, drives a wizard-style dialog, and renders a code-diff-like two-table comparison before confirmation. Inventory mutation endpoints must call a shared lock guard so locked warehouses cannot be changed outside stocktake.

**Tech Stack:** FastAPI, SQLAlchemy ORM, openpyxl, unittest, Vue 3 Composition API, existing `FormDrawer`, `TableControls`, `PaginationBar`, `downloadResource`, `uploadResource`, `postResource`, Vite.

---

## File Structure

- Create `backend/app/services/stocktake.py`
  - Builds stocktake numbers.
  - Checks active warehouse locks.
  - Creates stocktake snapshots.
  - Builds and parses Excel workbooks.
  - Calculates diff rows.
  - Applies confirmed adjustments.

- Modify `backend/app/models/operations.py`
  - Add `Stocktake`, `StocktakeWarehouse`, `StocktakeLine`, and `StocktakeAdjustment` SQLAlchemy models.

- Modify `backend/app/schemas/operations.py`
  - Add request/response schemas for stocktake start, import preview, diff rows, confirm, cancel, and history.

- Modify `backend/app/api/routes/inventory.py`
  - Add `/inventory/stocktakes` routes.
  - Add lock guard calls to `/inventory/inbound`, `/inventory/outbound`, and `/inventory/customer-supplied-inbound`.

- Modify `backend/app/api/routes/purchase.py`
  - Guard purchase receipt creation when target warehouse is locked.

- Modify `backend/app/api/routes/production.py`
  - Guard production work order auto issue and completion receipt posting when affected warehouse is locked.

- Modify `backend/app/api/routes/sales.py`
  - Guard delivery creation when finished warehouse is locked.

- Create `backend/sql/add_stocktake_tables.sql`
  - MySQL patch for production databases.

- Create `backend/tests/test_stocktake_locking.py`
  - Unit tests for lock guard and route-level blocking.

- Create `backend/tests/test_stocktake_excel_diff_confirm.py`
  - Unit tests for snapshot export, import diff, and confirmation adjustments.

- Modify `frontend/src/views/InventoryLedgerView.vue`
  - Add stocktake action above warehouse bar.
  - Add stocktake dialog state and API calls.
  - Include stocktake dialog component.

- Create `frontend/src/components/StocktakeDialog.vue`
  - Wizard UI: choose warehouses, export, import, diff preview, confirm, history.

- Modify `frontend/src/styles/main.css`
  - Add stocktake button, wizard, and side-by-side diff styles.

- Modify `frontend/src/utils/dictionaries.js`
  - Add stocktake status and diff type labels if central dictionary is already used for labels.

---

## Task 1: Backend Model and Lock Guard Tests

**Files:**
- Create: `backend/tests/test_stocktake_locking.py`
- Modify: `backend/app/models/operations.py`
- Create: `backend/app/services/stocktake.py`

- [ ] **Step 1: Write failing tests for active warehouse locks**

Create `backend/tests/test_stocktake_locking.py` with this content:

```python
from __future__ import annotations

import unittest
from datetime import UTC, datetime

from fastapi import HTTPException
from sqlalchemy import BigInteger, create_engine
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 Warehouse  # noqa: E402
from app.models.operations import Stocktake, StocktakeWarehouse  # noqa: E402
from app.services.stocktake import ensure_warehouses_unlocked, get_locked_warehouse_ids  # noqa: E402


class StocktakeLockingTest(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-01",
            warehouse_name="原材料库",
            warehouse_type="RAW",
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.finished = Warehouse(
            id=2,
            warehouse_code="WH-FG-01",
            warehouse_name="成品库",
            warehouse_type="FINISHED",
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.db.add_all([self.raw, self.finished])
        self.db.flush()
        stocktake = Stocktake(
            stocktake_no="PK-20260530-001",
            status="LOCKED",
            started_by_user_id=1,
            started_at=now,
            remark="测试锁库",
        )
        self.db.add(stocktake)
        self.db.flush()
        self.db.add(
            StocktakeWarehouse(
                stocktake_id=stocktake.id,
                warehouse_id=self.raw.id,
                warehouse_type="RAW",
                warehouse_name="原材料库",
                status="LOCKED",
            )
        )
        self.db.commit()

    def test_get_locked_warehouse_ids_returns_only_active_stocktakes(self) -> None:
        locked = get_locked_warehouse_ids(self.db, [self.raw.id, self.finished.id])

        self.assertEqual(locked, {self.raw.id})

    def test_ensure_warehouses_unlocked_raises_for_locked_warehouse(self) -> None:
        with self.assertRaises(HTTPException) as exc:
            ensure_warehouses_unlocked(self.db, [self.raw.id], "入库")

        self.assertEqual(exc.exception.status_code, 423)
        self.assertIn("原材料库正在盘库", exc.exception.detail)

    def test_ensure_warehouses_unlocked_allows_unlocked_warehouse(self) -> None:
        ensure_warehouses_unlocked(self.db, [self.finished.id], "销售出库")


if __name__ == "__main__":
    unittest.main()
```

- [ ] **Step 2: Run the lock tests and confirm they fail because models/service are missing**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_locking
```

Expected: FAIL with import errors for `Stocktake`, `StocktakeWarehouse`, or `app.services.stocktake`.

---

## Task 2: Add Stocktake Models, Schemas, SQL Patch, and Lock Service

**Files:**
- Modify: `backend/app/models/operations.py`
- Modify: `backend/app/schemas/operations.py`
- Create: `backend/app/services/stocktake.py`
- Create: `backend/sql/add_stocktake_tables.sql`
- Test: `backend/tests/test_stocktake_locking.py`

- [ ] **Step 1: Add stocktake ORM models**

Modify `backend/app/models/operations.py` by adding these classes after `InventoryTxn`:

```python
class Stocktake(Base):
    __tablename__ = "wh_stocktake"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    stocktake_no: Mapped[str] = mapped_column(String(50))
    status: Mapped[str] = mapped_column(String(32))
    started_by_user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    imported_by_user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    confirmed_by_user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    canceled_by_user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    started_at: Mapped[object] = mapped_column(DateTime)
    exported_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    imported_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    confirmed_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    canceled_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class StocktakeWarehouse(Base):
    __tablename__ = "wh_stocktake_warehouse"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    stocktake_id: Mapped[int] = mapped_column(ForeignKey("wh_stocktake.id"))
    warehouse_id: Mapped[int] = mapped_column(ForeignKey("wh_warehouse.id"))
    warehouse_type: Mapped[str] = mapped_column(String(32))
    warehouse_name: Mapped[str] = mapped_column(String(100))
    status: Mapped[str] = mapped_column(String(32))
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class StocktakeLine(Base):
    __tablename__ = "wh_stocktake_line"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    stocktake_id: Mapped[int] = mapped_column(ForeignKey("wh_stocktake.id"))
    stocktake_warehouse_id: Mapped[int | None] = mapped_column(ForeignKey("wh_stocktake_warehouse.id"), nullable=True)
    line_no: Mapped[int] = mapped_column(Integer)
    warehouse_id: Mapped[int] = mapped_column(ForeignKey("wh_warehouse.id"))
    warehouse_name: Mapped[str] = mapped_column(String(100))
    warehouse_type: Mapped[str] = mapped_column(String(32))
    location_id: Mapped[int | None] = mapped_column(ForeignKey("wh_location.id"), nullable=True)
    location_name: Mapped[str | None] = mapped_column(String(100), nullable=True)
    item_id: Mapped[int] = mapped_column(ForeignKey("md_item.id"))
    item_code: Mapped[str] = mapped_column(String(50))
    item_name: Mapped[str] = mapped_column(String(200))
    item_type: Mapped[str] = mapped_column(String(32))
    lot_id: Mapped[int | None] = mapped_column(ForeignKey("wh_stock_lot.id"), nullable=True)
    lot_no: Mapped[str | None] = mapped_column(String(80), nullable=True)
    source_material_sub_batch_no: Mapped[str | None] = mapped_column(String(80), nullable=True)
    snapshot_qty: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    snapshot_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    snapshot_unit_cost: Mapped[float] = mapped_column(DECIMAL(18, 4), server_default=text("0"))
    counted_qty: Mapped[float | None] = mapped_column(DECIMAL(18, 6), nullable=True)
    counted_weight_kg: Mapped[float | None] = mapped_column(DECIMAL(18, 6), nullable=True)
    diff_qty: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    diff_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    diff_amount: Mapped[float] = mapped_column(DECIMAL(18, 2), server_default=text("0"))
    diff_type: Mapped[str] = mapped_column(String(32))
    row_status: Mapped[str] = mapped_column(String(32))
    error_message: Mapped[str | None] = mapped_column(String(500), nullable=True)
    remark: Mapped[str | None] = mapped_column(String(255), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class StocktakeAdjustment(Base):
    __tablename__ = "wh_stocktake_adjustment"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    stocktake_id: Mapped[int] = mapped_column(ForeignKey("wh_stocktake.id"))
    stocktake_line_id: Mapped[int] = mapped_column(ForeignKey("wh_stocktake_line.id"))
    warehouse_id: Mapped[int] = mapped_column(ForeignKey("wh_warehouse.id"))
    item_id: Mapped[int] = mapped_column(ForeignKey("md_item.id"))
    lot_id: Mapped[int | None] = mapped_column(ForeignKey("wh_stock_lot.id"), nullable=True)
    adjustment_lot_id: Mapped[int | None] = mapped_column(ForeignKey("wh_stock_lot.id"), nullable=True)
    inventory_txn_id: Mapped[int | None] = mapped_column(ForeignKey("wh_inventory_txn.id"), nullable=True)
    adjustment_type: Mapped[str] = mapped_column(String(32))
    before_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"))
    after_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    qty_change: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    weight_change_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"))
    amount: Mapped[float] = mapped_column(DECIMAL(18, 2), server_default=text("0"))
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)
```

- [ ] **Step 2: Add stocktake schemas**

Modify `backend/app/schemas/operations.py` by adding:

```python
class StocktakeStartCreate(BaseModel):
    warehouse_ids: list[int] = Field(min_length=1)
    remark: str | None = None


class StocktakeWarehouseRead(BaseModel):
    id: int
    warehouse_id: int
    warehouse_name: str
    warehouse_type: str
    status: str


class StocktakeLineRead(BaseModel):
    id: int
    stocktake_id: int
    line_no: int
    warehouse_id: int
    warehouse_name: str
    warehouse_type: str
    location_id: int | None = None
    location_name: str | None = None
    item_id: int
    item_code: str
    item_name: str
    item_type: str
    lot_id: int | None = None
    lot_no: str | None = None
    source_material_sub_batch_no: str | None = None
    snapshot_qty: float
    snapshot_weight_kg: float
    snapshot_unit_cost: float
    counted_qty: float | None = None
    counted_weight_kg: float | None = None
    diff_qty: float
    diff_weight_kg: float
    diff_amount: float
    diff_type: str
    row_status: str
    error_message: str | None = None
    remark: str | None = None


class StocktakeRead(BaseModel):
    id: int
    stocktake_no: str
    status: str
    started_by_user_id: int | None = None
    imported_by_user_id: int | None = None
    confirmed_by_user_id: int | None = None
    canceled_by_user_id: int | None = None
    started_at: datetime
    exported_at: datetime | None = None
    imported_at: datetime | None = None
    confirmed_at: datetime | None = None
    canceled_at: datetime | None = None
    remark: str | None = None
    warehouses: list[StocktakeWarehouseRead] = Field(default_factory=list)
    summary: dict[str, float | int] = Field(default_factory=dict)


class StocktakeImportPreviewRead(BaseModel):
    stocktake: StocktakeRead
    lines: list[StocktakeLineRead]
    summary: dict[str, float | int]


class StocktakeConfirmCreate(BaseModel):
    confirm_text: str
    remark: str | None = None
```

If `Field` is not imported in `backend/app/schemas/operations.py`, add it to the existing Pydantic import.

- [ ] **Step 3: Create SQL patch for production MySQL**

Create `backend/sql/add_stocktake_tables.sql`:

```sql
CREATE TABLE IF NOT EXISTS wh_stocktake (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  stocktake_no VARCHAR(50) NOT NULL UNIQUE,
  status VARCHAR(32) NOT NULL,
  started_by_user_id BIGINT NULL,
  imported_by_user_id BIGINT NULL,
  confirmed_by_user_id BIGINT NULL,
  canceled_by_user_id BIGINT NULL,
  started_at DATETIME NOT NULL,
  exported_at DATETIME NULL,
  imported_at DATETIME NULL,
  confirmed_at DATETIME NULL,
  canceled_at DATETIME NULL,
  remark VARCHAR(255) NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  INDEX idx_wh_stocktake_status (status),
  INDEX idx_wh_stocktake_started_at (started_at)
);

CREATE TABLE IF NOT EXISTS wh_stocktake_warehouse (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  stocktake_id BIGINT NOT NULL,
  warehouse_id BIGINT NOT NULL,
  warehouse_type VARCHAR(32) NOT NULL,
  warehouse_name VARCHAR(100) NOT NULL,
  status VARCHAR(32) NOT NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  INDEX idx_wh_stocktake_warehouse_stocktake (stocktake_id),
  INDEX idx_wh_stocktake_warehouse_lock (warehouse_id, status),
  CONSTRAINT fk_stocktake_warehouse_stocktake FOREIGN KEY (stocktake_id) REFERENCES wh_stocktake(id),
  CONSTRAINT fk_stocktake_warehouse_warehouse FOREIGN KEY (warehouse_id) REFERENCES wh_warehouse(id)
);

CREATE TABLE IF NOT EXISTS wh_stocktake_line (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  stocktake_id BIGINT NOT NULL,
  stocktake_warehouse_id BIGINT NULL,
  line_no INT NOT NULL,
  warehouse_id BIGINT NOT NULL,
  warehouse_name VARCHAR(100) NOT NULL,
  warehouse_type VARCHAR(32) NOT NULL,
  location_id BIGINT NULL,
  location_name VARCHAR(100) NULL,
  item_id BIGINT NOT NULL,
  item_code VARCHAR(50) NOT NULL,
  item_name VARCHAR(200) NOT NULL,
  item_type VARCHAR(32) NOT NULL,
  lot_id BIGINT NULL,
  lot_no VARCHAR(80) NULL,
  source_material_sub_batch_no VARCHAR(80) NULL,
  snapshot_qty DECIMAL(18,6) NOT NULL DEFAULT 0,
  snapshot_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  snapshot_unit_cost DECIMAL(18,4) NOT NULL DEFAULT 0,
  counted_qty DECIMAL(18,6) NULL,
  counted_weight_kg DECIMAL(18,6) NULL,
  diff_qty DECIMAL(18,6) NOT NULL DEFAULT 0,
  diff_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  diff_amount DECIMAL(18,2) NOT NULL DEFAULT 0,
  diff_type VARCHAR(32) NOT NULL,
  row_status VARCHAR(32) NOT NULL,
  error_message VARCHAR(500) NULL,
  remark VARCHAR(255) NULL,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  INDEX idx_wh_stocktake_line_stocktake (stocktake_id),
  INDEX idx_wh_stocktake_line_lot (lot_id),
  INDEX idx_wh_stocktake_line_diff (diff_type, row_status),
  CONSTRAINT fk_stocktake_line_stocktake FOREIGN KEY (stocktake_id) REFERENCES wh_stocktake(id),
  CONSTRAINT fk_stocktake_line_warehouse FOREIGN KEY (warehouse_id) REFERENCES wh_warehouse(id),
  CONSTRAINT fk_stocktake_line_item FOREIGN KEY (item_id) REFERENCES md_item(id)
);

CREATE TABLE IF NOT EXISTS wh_stocktake_adjustment (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  stocktake_id BIGINT NOT NULL,
  stocktake_line_id BIGINT NOT NULL,
  warehouse_id BIGINT NOT NULL,
  item_id BIGINT NOT NULL,
  lot_id BIGINT NULL,
  adjustment_lot_id BIGINT NULL,
  inventory_txn_id BIGINT NULL,
  adjustment_type VARCHAR(32) NOT NULL,
  before_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,
  after_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  qty_change DECIMAL(18,6) NOT NULL DEFAULT 0,
  weight_change_kg DECIMAL(18,6) NOT NULL DEFAULT 0,
  unit_cost DECIMAL(18,4) NOT NULL DEFAULT 0,
  amount DECIMAL(18,2) NOT NULL DEFAULT 0,
  created_at DATETIME NOT NULL,
  updated_at DATETIME NOT NULL,
  INDEX idx_wh_stocktake_adjustment_stocktake (stocktake_id),
  CONSTRAINT fk_stocktake_adjustment_stocktake FOREIGN KEY (stocktake_id) REFERENCES wh_stocktake(id),
  CONSTRAINT fk_stocktake_adjustment_line FOREIGN KEY (stocktake_line_id) REFERENCES wh_stocktake_line(id)
);
```

- [ ] **Step 4: Create the stocktake lock service**

Create `backend/app/services/stocktake.py` with the lock constants and guard functions:

```python
from __future__ import annotations

from datetime import date, datetime
from decimal import Decimal

from fastapi import HTTPException
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.models.master_data import Warehouse
from app.models.operations import Stocktake, StocktakeWarehouse

ACTIVE_STOCKTAKE_STATUSES = {"LOCKED", "EXPORTED", "IMPORTED"}


def build_stocktake_no(db: Session) -> str:
    date_part = date.today().strftime("%Y%m%d")
    prefix = f"PK-{date_part}"
    count = db.scalar(select(func.count(Stocktake.id)).where(Stocktake.stocktake_no.like(f"{prefix}-%"))) or 0
    return f"{prefix}-{count + 1:03d}"


def stocktake_warehouse_type_label(value: str | None) -> str:
    return {
        "RAW": "原材料库",
        "SEMI": "半成品库",
        "FINISHED": "成品库",
        "AUX": "辅料库",
    }.get(str(value or "").upper(), str(value or "未知仓库"))


def get_locked_warehouse_ids(db: Session, warehouse_ids: list[int] | set[int]) -> set[int]:
    ids = {int(item) for item in warehouse_ids if item}
    if not ids:
        return set()
    rows = db.scalars(
        select(StocktakeWarehouse.warehouse_id)
        .join(Stocktake, Stocktake.id == StocktakeWarehouse.stocktake_id)
        .where(
            StocktakeWarehouse.warehouse_id.in_(ids),
            StocktakeWarehouse.status == "LOCKED",
            Stocktake.status.in_(ACTIVE_STOCKTAKE_STATUSES),
        )
    ).all()
    return {int(row) for row in rows}


def ensure_warehouses_unlocked(db: Session, warehouse_ids: list[int] | set[int], action_label: str) -> None:
    locked_ids = get_locked_warehouse_ids(db, warehouse_ids)
    if not locked_ids:
        return
    rows = db.scalars(select(Warehouse).where(Warehouse.id.in_(locked_ids)).order_by(Warehouse.id.asc())).all()
    labels = "、".join(row.warehouse_name for row in rows)
    raise HTTPException(status_code=423, detail=f"{labels}正在盘库，暂不能执行{action_label}")


def decimal_value(value: object, places: str = "0.000001") -> Decimal:
    return Decimal(str(value or 0)).quantize(Decimal(places))


def current_time() -> datetime:
    return datetime.now()
```

- [ ] **Step 5: Run lock tests and compile backend**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_locking
.venv/bin/python -m compileall app
```

Expected: lock tests PASS and compileall PASS.

---

## Task 3: Add Lock Guards to Inventory Mutation Routes

**Files:**
- Modify: `backend/app/api/routes/inventory.py`
- Test: `backend/tests/test_stocktake_locking.py`

- [ ] **Step 1: Add failing route-level test for generic inbound**

Append this test method to `StocktakeLockingTest` in `backend/tests/test_stocktake_locking.py`:

```python
    def test_inventory_inbound_is_blocked_when_warehouse_is_locked(self) -> None:
        from decimal import Decimal

        from app.api.routes.inventory import create_warehouse_inbound
        from app.models.master_data import Item
        from app.schemas.operations import WarehouseInboundCreate

        now = datetime.now(UTC)
        item = Item(
            id=1,
            item_code="原材料00001",
            item_name="冷轧钢板",
            item_type="RAW_MATERIAL",
            unit_weight_kg=Decimal("1"),
            safety_stock_weight_kg=Decimal("0"),
            scrap_sale_price=Decimal("0"),
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.db.add(item)
        self.db.commit()

        class User:
            id = 1

        class Context:
            user = User()

        with self.assertRaises(HTTPException) as exc:
            create_warehouse_inbound(
                WarehouseInboundCreate(
                    biz_type="OPENING",
                    item_id=item.id,
                    warehouse_id=self.raw.id,
                    inbound_weight_kg=10,
                    unit_cost=1,
                ),
                context=Context(),
                db=self.db,
            )

        self.assertEqual(exc.exception.status_code, 423)
        self.assertIn("正在盘库", exc.exception.detail)
```

- [ ] **Step 2: Run test and confirm failure**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_locking.StocktakeLockingTest.test_inventory_inbound_is_blocked_when_warehouse_is_locked
```

Expected: FAIL because `create_warehouse_inbound` does not yet check locks.

- [ ] **Step 3: Add lock checks to inventory routes**

In `backend/app/api/routes/inventory.py`, import the guard:

```python
from app.services.stocktake import ensure_warehouses_unlocked
```

Add these calls:

```python
# create_customer_supplied_inbound, after warehouse exists
ensure_warehouses_unlocked(db, [payload.warehouse_id], "客料入库")

# create_warehouse_inbound, after warehouse exists
ensure_warehouses_unlocked(db, [payload.warehouse_id], "入库")

# create_warehouse_outbound, after warehouse exists
ensure_warehouses_unlocked(db, [payload.warehouse_id], "出库")
```

- [ ] **Step 4: Run lock tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_locking
```

Expected: PASS.

---

## Task 4: Add Lock Guards to Purchase, Production, and Sales Posting Routes

**Files:**
- Modify: `backend/app/api/routes/purchase.py`
- Modify: `backend/app/api/routes/production.py`
- Modify: `backend/app/api/routes/sales.py`
- Test: `backend/tests/test_stocktake_locking.py`

- [ ] **Step 1: Add failing tests for posting-route lock helper calls**

Append a focused service-level test to `backend/tests/test_stocktake_locking.py` so the route imports are exercised without building every domain payload:

```python
    def test_lock_guard_message_names_action(self) -> None:
        with self.assertRaises(HTTPException) as exc:
            ensure_warehouses_unlocked(self.db, [self.raw.id], "到货入库")

        self.assertEqual(exc.exception.status_code, 423)
        self.assertIn("到货入库", exc.exception.detail)
```

Run it:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_locking.StocktakeLockingTest.test_lock_guard_message_names_action
```

Expected: PASS. This test locks the expected user-facing message before adding guard calls to larger routes.

- [ ] **Step 2: Guard purchase receipts**

In `backend/app/api/routes/purchase.py`, import:

```python
from app.services.stocktake import ensure_warehouses_unlocked
```

Inside `create_purchase_receipt`, after determining `warehouse_id` and before creating `PurchaseReceipt`, add:

```python
ensure_warehouses_unlocked(db, [warehouse_id], "到货入库")
```

Use the existing variable name in that function. If the function uses `receipt_warehouse_id`, call:

```python
ensure_warehouses_unlocked(db, [receipt_warehouse_id], "到货入库")
```

- [ ] **Step 3: Guard production work-order material issue**

In `backend/app/api/routes/production.py`, import:

```python
from app.services.stocktake import ensure_warehouses_unlocked
```

Inside `create_work_order`, before auto issue changes raw material lots, collect the warehouse ids of lots that would be consumed:

```python
locked_check_warehouse_ids = {lot.warehouse_id for lot in available_lots}
ensure_warehouses_unlocked(db, locked_check_warehouse_ids, "生产出库")
```

Place this after `available_lots` is loaded and before any `remaining_weight_kg` mutation.

- [ ] **Step 4: Guard completion receipts**

Inside `create_completion_receipt`, after validating the target completion warehouse and before creating `CompletionReceipt`, add:

```python
ensure_warehouses_unlocked(db, [payload.warehouse_id], "成品入库")
```

- [ ] **Step 5: Guard sales deliveries**

In `backend/app/api/routes/sales.py`, import:

```python
from app.services.stocktake import ensure_warehouses_unlocked
```

Inside `create_delivery`, after validating the warehouse and before creating `Delivery`, add:

```python
ensure_warehouses_unlocked(db, [payload.warehouse_id], "销售出库")
```

- [ ] **Step 6: Run backend compile and existing targeted tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m compileall app
.venv/bin/python -m unittest tests.test_stocktake_locking
.venv/bin/python -m unittest tests.test_sales_order_purchase_lock_flow
```

Expected: compileall PASS, lock tests PASS. Existing `test_sales_order_purchase_lock_flow` may expose unrelated status behavior already present in the repo; do not change purchase status semantics in this task.

---

## Task 5: Start Stocktake, Create Snapshot, and Export Workbook

**Files:**
- Modify: `backend/app/services/stocktake.py`
- Modify: `backend/app/api/routes/inventory.py`
- Test: `backend/tests/test_stocktake_excel_diff_confirm.py`

- [ ] **Step 1: Write failing snapshot/export tests**

Create `backend/tests/test_stocktake_excel_diff_confirm.py`:

```python
from __future__ import annotations

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

from openpyxl import load_workbook
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 StockLot, StocktakeLine  # noqa: E402
from app.services.stocktake import build_stocktake_workbook, start_stocktake  # noqa: E402


class StocktakeExcelDiffConfirmTest(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-01",
            warehouse_name="原材料库",
            warehouse_type="RAW",
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.item = Item(
            id=1,
            item_code="原材料00001",
            item_name="冷轧钢板",
            item_type="RAW_MATERIAL",
            unit_weight_kg=Decimal("1"),
            safety_stock_weight_kg=Decimal("0"),
            scrap_sale_price=Decimal("0"),
            status="ACTIVE",
            created_at=now,
            updated_at=now,
        )
        self.db.add_all([self.raw, self.item])
        self.db.flush()
        self.lot = StockLot(
            id=1,
            lot_no="LOT-RAW-001",
            lot_role="INVENTORY",
            item_id=self.item.id,
            warehouse_id=self.raw.id,
            source_doc_type="OPENING",
            source_doc_id=1,
            inbound_qty=Decimal("0"),
            inbound_weight_kg=Decimal("100"),
            remaining_qty=Decimal("0"),
            remaining_weight_kg=Decimal("100"),
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=Decimal("2.5"),
            quality_status="PASS",
            status="AVAILABLE",
            created_at=now,
            updated_at=now,
        )
        self.balance = StockBalance(
            item_id=self.item.id,
            warehouse_id=self.raw.id,
            qty_on_hand=Decimal("0"),
            weight_on_hand_kg=Decimal("100"),
            qty_available=Decimal("0"),
            weight_available_kg=Decimal("100"),
            qty_allocated=Decimal("0"),
            weight_allocated_kg=Decimal("0"),
            avg_unit_cost=Decimal("2.5"),
            created_at=now,
            updated_at=now,
        )
        self.db.add_all([self.lot, self.balance])
        self.db.commit()

    def test_start_stocktake_locks_warehouse_and_snapshots_lots(self) -> None:
        stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库")

        self.assertEqual(stocktake.status, "LOCKED")
        line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id))
        self.assertIsNotNone(line)
        self.assertEqual(line.lot_no, "LOT-RAW-001")
        self.assertEqual(float(line.snapshot_weight_kg), 100)
        self.assertEqual(line.diff_type, "MATCH")

    def test_export_workbook_has_one_sheet_per_selected_warehouse(self) -> None:
        stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库")

        workbook = build_stocktake_workbook(self.db, stocktake.id)
        output = BytesIO()
        workbook.save(output)
        loaded = load_workbook(BytesIO(output.getvalue()), read_only=True, data_only=True)

        self.assertEqual(loaded.sheetnames, ["原材料库"])
        rows = list(loaded["原材料库"].iter_rows(values_only=True))
        self.assertIn("快照行ID", rows[0])
        self.assertIn("盘点重量(kg)", rows[0])
        self.assertEqual(rows[1][rows[0].index("批次号")], "LOT-RAW-001")


if __name__ == "__main__":
    unittest.main()
```

- [ ] **Step 2: Run tests and confirm failure**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm
```

Expected: FAIL because `start_stocktake` and `build_stocktake_workbook` do not exist.

- [ ] **Step 3: Implement `start_stocktake` in `backend/app/services/stocktake.py`**

Add imports:

```python
from openpyxl import Workbook, load_workbook
from sqlalchemy import delete

from app.models.master_data import Item, StockBalance
from app.models.operations import InventoryTxn, StockLot, StocktakeAdjustment, StocktakeLine
from app.services.operations import create_inventory_txn, get_location_or_default, to_decimal, upsert_stock_balance
```

Add constants:

```python
STOCKTAKE_EXCEL_HEADERS = [
    "盘库单号",
    "快照行ID",
    "仓库",
    "库位",
    "物料编码",
    "物料名称",
    "批次号",
    "来源材料分批次号",
    "系统数量",
    "系统重量(kg)",
    "系统单价",
    "盘点数量",
    "盘点重量(kg)",
    "盘点备注",
]
```

Add `start_stocktake`:

```python
def start_stocktake(db: Session, *, warehouse_ids: list[int], user_id: int | None, remark: str | None = None) -> Stocktake:
    ids = [int(item) for item in warehouse_ids if item]
    if not ids:
        raise HTTPException(status_code=400, detail="请选择需要盘库的仓库")
    ensure_warehouses_unlocked(db, ids, "开始盘库")
    warehouses = db.scalars(select(Warehouse).where(Warehouse.id.in_(ids)).order_by(Warehouse.id.asc())).all()
    if len(warehouses) != len(set(ids)):
        raise HTTPException(status_code=404, detail="存在无效仓库，无法开始盘库")

    now = current_time()
    stocktake = Stocktake(
        stocktake_no=build_stocktake_no(db),
        status="LOCKED",
        started_by_user_id=user_id,
        started_at=now,
        remark=remark,
    )
    db.add(stocktake)
    db.flush()

    stocktake_warehouses_by_id: dict[int, StocktakeWarehouse] = {}
    for warehouse in warehouses:
        row = StocktakeWarehouse(
            stocktake_id=stocktake.id,
            warehouse_id=warehouse.id,
            warehouse_type=str(warehouse.warehouse_type or "").upper(),
            warehouse_name=warehouse.warehouse_name,
            status="LOCKED",
        )
        db.add(row)
        db.flush()
        stocktake_warehouses_by_id[warehouse.id] = row

    lot_rows = db.execute(
        select(
            StockLot,
            Item,
        )
        .join(Item, Item.id == StockLot.item_id)
        .where(
            StockLot.warehouse_id.in_(ids),
            StockLot.status == "AVAILABLE",
            StockLot.remaining_weight_kg > 0,
        )
        .order_by(StockLot.warehouse_id.asc(), Item.item_code.asc(), StockLot.lot_no.asc(), StockLot.id.asc())
    ).all()

    line_no = 1
    for lot, item in lot_rows:
        warehouse = next(row for row in warehouses if row.id == lot.warehouse_id)
        location_name = None
        if lot.location_id:
            from app.models.operations import WarehouseLocation

            location_name = db.scalar(select(WarehouseLocation.location_name).where(WarehouseLocation.id == lot.location_id))
        db.add(
            StocktakeLine(
                stocktake_id=stocktake.id,
                stocktake_warehouse_id=stocktake_warehouses_by_id[warehouse.id].id,
                line_no=line_no,
                warehouse_id=warehouse.id,
                warehouse_name=warehouse.warehouse_name,
                warehouse_type=str(warehouse.warehouse_type or "").upper(),
                location_id=lot.location_id,
                location_name=location_name,
                item_id=item.id,
                item_code=item.item_code,
                item_name=item.item_name,
                item_type=item.item_type,
                lot_id=lot.id,
                lot_no=lot.lot_no,
                source_material_sub_batch_no=lot.source_material_sub_batch_no or lot.material_sub_batch_no,
                snapshot_qty=Decimal("0") if item.item_type == "RAW_MATERIAL" else to_decimal(lot.remaining_qty),
                snapshot_weight_kg=to_decimal(lot.remaining_weight_kg),
                snapshot_unit_cost=to_decimal(lot.unit_cost, "0.0001"),
                counted_qty=None,
                counted_weight_kg=None,
                diff_qty=Decimal("0"),
                diff_weight_kg=Decimal("0"),
                diff_amount=Decimal("0"),
                diff_type="MATCH",
                row_status="SNAPSHOT",
                remark=None,
            )
        )
        line_no += 1

    db.commit()
    db.refresh(stocktake)
    return stocktake
```

- [ ] **Step 4: Implement workbook export**

Add:

```python
def _stocktake_lines(db: Session, stocktake_id: int) -> list[StocktakeLine]:
    return db.scalars(
        select(StocktakeLine)
        .where(StocktakeLine.stocktake_id == stocktake_id)
        .order_by(StocktakeLine.warehouse_id.asc(), StocktakeLine.line_no.asc())
    ).all()


def build_stocktake_workbook(db: Session, stocktake_id: int) -> Workbook:
    stocktake = db.get(Stocktake, stocktake_id)
    if not stocktake:
        raise HTTPException(status_code=404, detail="盘库单不存在")
    lines = _stocktake_lines(db, stocktake_id)
    workbook = Workbook()
    default_sheet = workbook.active
    workbook.remove(default_sheet)
    grouped: dict[str, list[StocktakeLine]] = {}
    for line in lines:
        grouped.setdefault(stocktake_warehouse_type_label(line.warehouse_type), []).append(line)
    if not grouped:
        for warehouse in db.scalars(select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake_id)).all():
            grouped.setdefault(stocktake_warehouse_type_label(warehouse.warehouse_type), [])
    for sheet_name, sheet_lines in grouped.items():
        worksheet = workbook.create_sheet(title=sheet_name[:31])
        worksheet.freeze_panes = "A2"
        worksheet.append(STOCKTAKE_EXCEL_HEADERS)
        for line in sheet_lines:
            worksheet.append(
                [
                    stocktake.stocktake_no,
                    line.id,
                    line.warehouse_name,
                    line.location_name or "",
                    line.item_code,
                    line.item_name,
                    line.lot_no or "",
                    line.source_material_sub_batch_no or "",
                    float(line.snapshot_qty or 0),
                    float(line.snapshot_weight_kg or 0),
                    float(line.snapshot_unit_cost or 0),
                    float(line.snapshot_qty or 0),
                    float(line.snapshot_weight_kg or 0),
                    "",
                ]
            )
        for index, width in enumerate([18, 12, 14, 14, 18, 28, 22, 24, 14, 16, 12, 14, 16, 24], start=1):
            worksheet.column_dimensions[worksheet.cell(row=1, column=index).column_letter].width = width
    stocktake.exported_at = current_time()
    stocktake.status = "EXPORTED"
    db.add(stocktake)
    db.commit()
    return workbook
```

- [ ] **Step 5: Add start/export API routes**

In `backend/app/api/routes/inventory.py`, import:

```python
from io import BytesIO
from urllib.parse import quote
from fastapi import File, UploadFile
from fastapi.responses import Response
from app.schemas.operations import StocktakeConfirmCreate, StocktakeImportPreviewRead, StocktakeRead, StocktakeStartCreate
from app.services.stocktake import build_stocktake_workbook, start_stocktake
```

Add response helper:

```python
def _excel_response(workbook, filename: str) -> Response:
    output = BytesIO()
    workbook.save(output)
    encoded_filename = quote(filename)
    return Response(
        content=output.getvalue(),
        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        headers={"Content-Disposition": f"attachment; filename=\"stocktake.xlsx\"; filename*=UTF-8''{encoded_filename}"},
    )
```

Add routes:

```python
@router.post("/stocktakes", response_model=StocktakeRead)
def start_inventory_stocktake(
    payload: StocktakeStartCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> StocktakeRead:
    stocktake = start_stocktake(db, warehouse_ids=payload.warehouse_ids, user_id=context.user.id, remark=payload.remark)
    return _stocktake_read(db, stocktake.id)


@router.get("/stocktakes/{stocktake_id}/export")
def export_inventory_stocktake(stocktake_id: int, db: Session = Depends(get_db)) -> Response:
    workbook = build_stocktake_workbook(db, stocktake_id)
    stocktake = db.get(Stocktake, stocktake_id)
    return _excel_response(workbook, f"{stocktake.stocktake_no}_盘库清单.xlsx")
```

Also add `_stocktake_read` in `inventory.py` or `stocktake.py`; keep it in `stocktake.py` if multiple routes need it.

- [ ] **Step 6: Run tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm
.venv/bin/python -m compileall app
```

Expected: tests PASS and compileall PASS.

---

## Task 6: Import Counted Workbook and Calculate Diff Preview

**Files:**
- Modify: `backend/app/services/stocktake.py`
- Modify: `backend/app/api/routes/inventory.py`
- Test: `backend/tests/test_stocktake_excel_diff_confirm.py`

- [ ] **Step 1: Add failing import/diff test**

Append this test to `StocktakeExcelDiffConfirmTest`:

```python
    def test_import_counted_workbook_marks_loss_gain_and_missing_rows(self) -> None:
        from openpyxl import Workbook
        from app.services.stocktake import import_stocktake_workbook

        stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库")
        line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id))
        workbook = Workbook()
        worksheet = workbook.active
        worksheet.title = "原材料库"
        worksheet.append([
            "盘库单号",
            "快照行ID",
            "仓库",
            "库位",
            "物料编码",
            "物料名称",
            "批次号",
            "来源材料分批次号",
            "系统数量",
            "系统重量(kg)",
            "系统单价",
            "盘点数量",
            "盘点重量(kg)",
            "盘点备注",
        ])
        worksheet.append([
            stocktake.stocktake_no,
            line.id,
            "原材料库",
            "",
            "原材料00001",
            "冷轧钢板",
            "LOT-RAW-001",
            "",
            0,
            100,
            2.5,
            0,
            90,
            "实盘少10kg",
        ])
        output = BytesIO()
        workbook.save(output)

        preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1)

        self.assertEqual(preview["summary"]["loss_count"], 1)
        loss_line = self.db.get(StocktakeLine, line.id)
        self.assertEqual(loss_line.diff_type, "LOSS")
        self.assertEqual(float(loss_line.diff_weight_kg), -10)
        self.assertEqual(loss_line.row_status, "READY")
```

- [ ] **Step 2: Run the test and confirm failure**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm.StocktakeExcelDiffConfirmTest.test_import_counted_workbook_marks_loss_gain_and_missing_rows
```

Expected: FAIL because `import_stocktake_workbook` is missing.

- [ ] **Step 3: Implement import parser and diff calculation**

In `backend/app/services/stocktake.py`, add:

```python
def _clean_text(value: object) -> str:
    return str(value or "").strip()


def _read_workbook_rows(content: bytes) -> list[dict[str, object]]:
    try:
        workbook = load_workbook(BytesIO(content), read_only=True, data_only=True)
    except Exception as exc:
        raise ValueError("Excel 文件读取失败，请确认上传的是盘库清单 xlsx 文件") from exc
    rows: list[dict[str, object]] = []
    for worksheet in workbook.worksheets:
        values = list(worksheet.iter_rows(values_only=True))
        if not values:
            continue
        headers = [_clean_text(item) for item in values[0]]
        missing = [header for header in STOCKTAKE_EXCEL_HEADERS if header not in headers]
        if missing:
            raise ValueError(f"{worksheet.title} 缺少必要列：{'、'.join(missing)}")
        for row in values[1:]:
            if not any(_clean_text(item) for item in row):
                continue
            payload = {headers[index]: value for index, value in enumerate(row) if index < len(headers)}
            payload["_sheet_name"] = worksheet.title
            rows.append(payload)
    return rows


def _line_diff_type(qty_diff: Decimal, weight_diff: Decimal, *, is_new: bool = False, is_missing: bool = False) -> str:
    if is_new:
        return "NEW_LOT"
    if is_missing:
        return "MISSING"
    basis = weight_diff if weight_diff != 0 else qty_diff
    if basis > 0:
        return "GAIN"
    if basis < 0:
        return "LOSS"
    return "MATCH"


def _stocktake_summary(lines: list[StocktakeLine]) -> dict[str, int | float]:
    summary = {
        "match_count": 0,
        "gain_count": 0,
        "loss_count": 0,
        "new_lot_count": 0,
        "missing_count": 0,
        "error_count": 0,
        "diff_amount": 0.0,
    }
    for line in lines:
        key = {
            "MATCH": "match_count",
            "GAIN": "gain_count",
            "LOSS": "loss_count",
            "NEW_LOT": "new_lot_count",
            "MISSING": "missing_count",
        }.get(line.diff_type)
        if key:
            summary[key] += 1
        if line.row_status == "ERROR":
            summary["error_count"] += 1
        summary["diff_amount"] += float(line.diff_amount or 0)
    return summary
```

Implement:

```python
def import_stocktake_workbook(db: Session, stocktake_id: int, content: bytes, user_id: int | None) -> dict[str, object]:
    stocktake = db.get(Stocktake, stocktake_id)
    if not stocktake:
        raise HTTPException(status_code=404, detail="盘库单不存在")
    if stocktake.status not in {"LOCKED", "EXPORTED", "IMPORTED"}:
        raise HTTPException(status_code=400, detail="当前盘库单状态不允许导入")

    rows = _read_workbook_rows(content)
    if not rows:
        raise ValueError("Excel 中没有可导入的盘点数据")

    existing_lines = {
        line.id: line
        for line in db.scalars(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake_id)).all()
    }
    seen_line_ids: set[int] = set()
    next_line_no = max((line.line_no for line in existing_lines.values()), default=0) + 1

    for row in rows:
        row_stocktake_no = _clean_text(row.get("盘库单号"))
        if row_stocktake_no != stocktake.stocktake_no:
            raise ValueError(f"存在不属于当前盘库单的行：{row_stocktake_no or '空盘库单号'}")
        line_id_text = _clean_text(row.get("快照行ID"))
        counted_qty = decimal_value(row.get("盘点数量"))
        counted_weight = decimal_value(row.get("盘点重量(kg)"))
        remark = _clean_text(row.get("盘点备注")) or None
        if line_id_text:
            line_id = int(float(line_id_text))
            line = existing_lines.get(line_id)
            if not line:
                raise ValueError(f"快照行ID {line_id} 不存在于当前盘库单")
            if line_id in seen_line_ids:
                line.row_status = "ERROR"
                line.error_message = "导入清单中该快照行重复"
            seen_line_ids.add(line_id)
            snapshot_qty = decimal_value(line.snapshot_qty)
            snapshot_weight = decimal_value(line.snapshot_weight_kg)
            qty_diff = Decimal("0") if line.item_type == "RAW_MATERIAL" else counted_qty - snapshot_qty
            weight_diff = counted_weight - snapshot_weight
            line.counted_qty = Decimal("0") if line.item_type == "RAW_MATERIAL" else counted_qty
            line.counted_weight_kg = counted_weight
            line.diff_qty = qty_diff
            line.diff_weight_kg = weight_diff
            line.diff_amount = (weight_diff * decimal_value(line.snapshot_unit_cost, "0.0001")).quantize(Decimal("0.01"))
            line.diff_type = _line_diff_type(qty_diff, weight_diff)
            line.row_status = "ERROR" if line.error_message else "READY"
            line.remark = remark
            db.add(line)
            continue

        warehouse_name = _clean_text(row.get("仓库"))
        item_code = _clean_text(row.get("物料编码"))
        item = db.scalar(select(Item).where(Item.item_code == item_code).limit(1))
        warehouse = db.scalar(select(Warehouse).where(Warehouse.warehouse_name == warehouse_name).limit(1))
        if not item or not warehouse:
            raise ValueError(f"新增批次行物料或仓库不存在：{warehouse_name} / {item_code}")
        unit_cost = decimal_value(row.get("系统单价"), "0.0001")
        db.add(
            StocktakeLine(
                stocktake_id=stocktake.id,
                stocktake_warehouse_id=None,
                line_no=next_line_no,
                warehouse_id=warehouse.id,
                warehouse_name=warehouse.warehouse_name,
                warehouse_type=str(warehouse.warehouse_type or "").upper(),
                location_id=None,
                location_name=_clean_text(row.get("库位")) or None,
                item_id=item.id,
                item_code=item.item_code,
                item_name=item.item_name,
                item_type=item.item_type,
                lot_id=None,
                lot_no=_clean_text(row.get("批次号")) or None,
                source_material_sub_batch_no=_clean_text(row.get("来源材料分批次号")) or None,
                snapshot_qty=Decimal("0"),
                snapshot_weight_kg=Decimal("0"),
                snapshot_unit_cost=unit_cost,
                counted_qty=Decimal("0") if item.item_type == "RAW_MATERIAL" else counted_qty,
                counted_weight_kg=counted_weight,
                diff_qty=Decimal("0") if item.item_type == "RAW_MATERIAL" else counted_qty,
                diff_weight_kg=counted_weight,
                diff_amount=(counted_weight * unit_cost).quantize(Decimal("0.01")),
                diff_type="NEW_LOT",
                row_status="READY",
                remark=remark,
            )
        )
        next_line_no += 1

    for line_id, line in existing_lines.items():
        if line_id in seen_line_ids:
            continue
        line.counted_qty = Decimal("0")
        line.counted_weight_kg = Decimal("0")
        line.diff_qty = -decimal_value(line.snapshot_qty)
        line.diff_weight_kg = -decimal_value(line.snapshot_weight_kg)
        line.diff_amount = (decimal_value(line.diff_weight_kg) * decimal_value(line.snapshot_unit_cost, "0.0001")).quantize(Decimal("0.01"))
        line.diff_type = "MISSING"
        line.row_status = "READY"
        line.remark = "导入清单缺失，按盘点为0处理"
        db.add(line)

    stocktake.status = "IMPORTED"
    stocktake.imported_by_user_id = user_id
    stocktake.imported_at = current_time()
    db.add(stocktake)
    db.commit()

    lines = db.scalars(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake_id).order_by(StocktakeLine.line_no.asc())).all()
    return {"stocktake": stocktake, "lines": lines, "summary": _stocktake_summary(lines)}
```

- [ ] **Step 4: Add import API route**

In `backend/app/api/routes/inventory.py`, import `import_stocktake_workbook` and add:

```python
@router.post("/stocktakes/{stocktake_id}/import", response_model=StocktakeImportPreviewRead)
async def import_inventory_stocktake(
    stocktake_id: int,
    file: UploadFile = File(...),
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> StocktakeImportPreviewRead:
    content = await file.read()
    try:
        result = import_stocktake_workbook(db, stocktake_id, content, user_id=context.user.id)
    except ValueError as exc:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    return _stocktake_preview_read(db, result)
```

- [ ] **Step 5: Run tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm
.venv/bin/python -m compileall app
```

Expected: tests PASS and compileall PASS.

---

## Task 7: Confirm Stocktake and Post Inventory Adjustments

**Files:**
- Modify: `backend/app/services/stocktake.py`
- Modify: `backend/app/api/routes/inventory.py`
- Test: `backend/tests/test_stocktake_excel_diff_confirm.py`

- [ ] **Step 1: Add failing confirmation test**

Append:

```python
    def test_confirm_stocktake_adjusts_lot_balance_and_writes_txn(self) -> None:
        from openpyxl import Workbook
        from app.models.operations import InventoryTxn, StocktakeAdjustment
        from app.services.stocktake import confirm_stocktake, import_stocktake_workbook

        stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库")
        line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id))
        workbook = Workbook()
        worksheet = workbook.active
        worksheet.title = "原材料库"
        worksheet.append([
            "盘库单号",
            "快照行ID",
            "仓库",
            "库位",
            "物料编码",
            "物料名称",
            "批次号",
            "来源材料分批次号",
            "系统数量",
            "系统重量(kg)",
            "系统单价",
            "盘点数量",
            "盘点重量(kg)",
            "盘点备注",
        ])
        worksheet.append([stocktake.stocktake_no, line.id, "原材料库", "", "原材料00001", "冷轧钢板", "LOT-RAW-001", "", 0, 100, 2.5, 0, 90, "实盘少10kg"])
        output = BytesIO()
        workbook.save(output)
        import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1)

        confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="确认盘库", remark="确认调整")

        self.assertEqual(confirmed.status, "CONFIRMED")
        self.db.refresh(self.lot)
        self.db.refresh(self.balance)
        self.assertEqual(float(self.lot.remaining_weight_kg), 90)
        self.assertEqual(float(self.balance.weight_on_hand_kg), 90)
        txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "STOCKTAKE_LOSS"))
        self.assertIsNotNone(txn)
        adjustment = self.db.scalar(select(StocktakeAdjustment).where(StocktakeAdjustment.stocktake_id == stocktake.id))
        self.assertIsNotNone(adjustment)
        self.assertEqual(float(adjustment.weight_change_kg), -10)
```

- [ ] **Step 2: Run test and confirm failure**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm.StocktakeExcelDiffConfirmTest.test_confirm_stocktake_adjusts_lot_balance_and_writes_txn
```

Expected: FAIL because `confirm_stocktake` is missing.

- [ ] **Step 3: Implement confirmation**

In `backend/app/services/stocktake.py`, add:

```python
def _stocktake_txn_type(diff_type: str) -> str:
    return {
        "GAIN": "STOCKTAKE_GAIN",
        "LOSS": "STOCKTAKE_LOSS",
        "MISSING": "STOCKTAKE_LOSS",
        "NEW_LOT": "STOCKTAKE_NEW_LOT",
    }.get(diff_type, "STOCKTAKE_GAIN")


def confirm_stocktake(db: Session, stocktake_id: int, *, user_id: int | None, confirm_text: str, remark: str | None = None) -> Stocktake:
    if confirm_text != "确认盘库":
        raise HTTPException(status_code=400, detail="请输入“确认盘库”后再确认")
    stocktake = db.get(Stocktake, stocktake_id)
    if not stocktake:
        raise HTTPException(status_code=404, detail="盘库单不存在")
    if stocktake.status != "IMPORTED":
        raise HTTPException(status_code=400, detail="请先导入盘点清单并确认差异后再过账")

    lines = db.scalars(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake_id).order_by(StocktakeLine.line_no.asc())).all()
    error_lines = [line for line in lines if line.row_status == "ERROR"]
    if error_lines:
        raise HTTPException(status_code=400, detail=f"存在 {len(error_lines)} 行异常，不能确认盘库")

    now = current_time()
    for line in lines:
        if line.diff_type == "MATCH":
            continue
        unit_cost = decimal_value(line.snapshot_unit_cost, "0.0001")
        qty_change = decimal_value(line.diff_qty)
        weight_change = decimal_value(line.diff_weight_kg)
        if line.diff_type == "NEW_LOT":
            lot = StockLot(
                lot_no=line.lot_no or f"{stocktake.stocktake_no}-NEW-{line.line_no:04d}",
                lot_role="STOCKTAKE_NEW",
                item_id=line.item_id,
                warehouse_id=line.warehouse_id,
                location_id=line.location_id,
                source_doc_type="STOCKTAKE",
                source_doc_id=stocktake.id,
                source_line_id=line.id,
                source_material_sub_batch_no=line.source_material_sub_batch_no,
                source_material_summary="盘库新增批次",
                inbound_qty=max(Decimal("0"), decimal_value(line.counted_qty)),
                inbound_weight_kg=max(Decimal("0"), decimal_value(line.counted_weight_kg)),
                remaining_qty=max(Decimal("0"), decimal_value(line.counted_qty)),
                remaining_weight_kg=max(Decimal("0"), decimal_value(line.counted_weight_kg)),
                locked_qty=Decimal("0"),
                locked_weight_kg=Decimal("0"),
                unit_cost=unit_cost,
                production_date=now.date(),
                quality_status="PASS",
                status="AVAILABLE",
                remark=line.remark or "盘库新增",
            )
            db.add(lot)
            db.flush()
            lot_id = lot.id
            before_qty = Decimal("0")
            before_weight = Decimal("0")
            after_qty = decimal_value(line.counted_qty)
            after_weight = decimal_value(line.counted_weight_kg)
        else:
            lot = db.get(StockLot, line.lot_id)
            if not lot:
                raise HTTPException(status_code=400, detail=f"盘库行 {line.line_no} 对应批次不存在")
            before_qty = Decimal("0") if line.item_type == "RAW_MATERIAL" else decimal_value(lot.remaining_qty)
            before_weight = decimal_value(lot.remaining_weight_kg)
            after_qty = Decimal("0") if line.item_type == "RAW_MATERIAL" else max(Decimal("0"), decimal_value(line.counted_qty))
            after_weight = max(Decimal("0"), decimal_value(line.counted_weight_kg))
            qty_change = after_qty - before_qty
            weight_change = after_weight - before_weight
            lot.remaining_qty = after_qty
            lot.remaining_weight_kg = after_weight
            lot.status = "AVAILABLE" if after_qty > 0 or after_weight > 0 else "DEPLETED"
            db.add(lot)
            lot_id = lot.id

        upsert_stock_balance(
            db,
            item_id=line.item_id,
            warehouse_id=line.warehouse_id,
            location_id=line.location_id,
            qty_delta=qty_change,
            weight_delta=weight_change,
            available_qty_delta=qty_change,
            available_weight_delta=weight_change,
            unit_cost=unit_cost,
        )
        txn = create_inventory_txn(
            db,
            txn_type=_stocktake_txn_type(line.diff_type),
            item_id=line.item_id,
            warehouse_id=line.warehouse_id,
            location_id=line.location_id,
            lot_id=lot_id,
            qty_change=qty_change,
            weight_change=weight_change,
            unit_cost=unit_cost,
            source_doc_type="STOCKTAKE",
            source_doc_id=stocktake.id,
            source_line_id=line.id,
            biz_time=now,
            operator_user_id=user_id,
            remark=line.remark or remark or "盘库调整",
            amount_basis="WEIGHT",
        )
        db.flush()
        db.add(
            StocktakeAdjustment(
                stocktake_id=stocktake.id,
                stocktake_line_id=line.id,
                warehouse_id=line.warehouse_id,
                item_id=line.item_id,
                lot_id=line.lot_id,
                adjustment_lot_id=lot_id,
                inventory_txn_id=txn.id,
                adjustment_type=line.diff_type,
                before_qty=before_qty,
                after_qty=after_qty,
                before_weight_kg=before_weight,
                after_weight_kg=after_weight,
                qty_change=qty_change,
                weight_change_kg=weight_change,
                unit_cost=unit_cost,
                amount=(abs(weight_change) * unit_cost).quantize(Decimal("0.01")),
            )
        )

    for warehouse in db.scalars(select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake_id)).all():
        warehouse.status = "UNLOCKED"
        db.add(warehouse)
    stocktake.status = "CONFIRMED"
    stocktake.confirmed_by_user_id = user_id
    stocktake.confirmed_at = now
    if remark:
        stocktake.remark = remark
    db.add(stocktake)
    db.commit()
    db.refresh(stocktake)
    return stocktake
```

- [ ] **Step 4: Add confirm and cancel API routes**

In `backend/app/api/routes/inventory.py`, import `confirm_stocktake` and add:

```python
@router.post("/stocktakes/{stocktake_id}/confirm", response_model=StocktakeRead)
def confirm_inventory_stocktake(
    stocktake_id: int,
    payload: StocktakeConfirmCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> StocktakeRead:
    stocktake = confirm_stocktake(
        db,
        stocktake_id,
        user_id=context.user.id,
        confirm_text=payload.confirm_text,
        remark=payload.remark,
    )
    return _stocktake_read(db, stocktake.id)
```

Add `cancel_stocktake` in service:

```python
def cancel_stocktake(db: Session, stocktake_id: int, *, user_id: int | None, remark: str | None = None) -> Stocktake:
    stocktake = db.get(Stocktake, stocktake_id)
    if not stocktake:
        raise HTTPException(status_code=404, detail="盘库单不存在")
    if stocktake.status not in ACTIVE_STOCKTAKE_STATUSES:
        raise HTTPException(status_code=400, detail="当前盘库单不能取消")
    now = current_time()
    for warehouse in db.scalars(select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake_id)).all():
        warehouse.status = "UNLOCKED"
        db.add(warehouse)
    stocktake.status = "CANCELED"
    stocktake.canceled_by_user_id = user_id
    stocktake.canceled_at = now
    stocktake.remark = remark or stocktake.remark
    db.add(stocktake)
    db.commit()
    db.refresh(stocktake)
    return stocktake
```

Add route:

```python
@router.post("/stocktakes/{stocktake_id}/cancel", response_model=StocktakeRead)
def cancel_inventory_stocktake(
    stocktake_id: int,
    payload: StocktakeConfirmCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> StocktakeRead:
    stocktake = cancel_stocktake(db, stocktake_id, user_id=context.user.id, remark=payload.remark)
    return _stocktake_read(db, stocktake.id)
```

- [ ] **Step 5: Run confirmation tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm
.venv/bin/python -m compileall app
```

Expected: tests PASS and compileall PASS.

---

## Task 8: Stocktake History and Read APIs

**Files:**
- Modify: `backend/app/services/stocktake.py`
- Modify: `backend/app/api/routes/inventory.py`
- Test: `backend/tests/test_stocktake_excel_diff_confirm.py`

- [ ] **Step 1: Add read helpers**

In `backend/app/services/stocktake.py`, add:

```python
def stocktake_lines_for_read(db: Session, stocktake_id: int) -> list[StocktakeLine]:
    return db.scalars(
        select(StocktakeLine)
        .where(StocktakeLine.stocktake_id == stocktake_id)
        .order_by(StocktakeLine.warehouse_id.asc(), StocktakeLine.line_no.asc())
    ).all()


def stocktake_warehouses_for_read(db: Session, stocktake_id: int) -> list[StocktakeWarehouse]:
    return db.scalars(
        select(StocktakeWarehouse)
        .where(StocktakeWarehouse.stocktake_id == stocktake_id)
        .order_by(StocktakeWarehouse.id.asc())
    ).all()
```

- [ ] **Step 2: Add route read serializers**

In `backend/app/api/routes/inventory.py`, add:

```python
def _stocktake_line_read(line: StocktakeLine) -> dict[str, object]:
    return {
        "id": line.id,
        "stocktake_id": line.stocktake_id,
        "line_no": line.line_no,
        "warehouse_id": line.warehouse_id,
        "warehouse_name": line.warehouse_name,
        "warehouse_type": line.warehouse_type,
        "location_id": line.location_id,
        "location_name": line.location_name,
        "item_id": line.item_id,
        "item_code": line.item_code,
        "item_name": line.item_name,
        "item_type": line.item_type,
        "lot_id": line.lot_id,
        "lot_no": line.lot_no,
        "source_material_sub_batch_no": line.source_material_sub_batch_no,
        "snapshot_qty": line.snapshot_qty,
        "snapshot_weight_kg": line.snapshot_weight_kg,
        "snapshot_unit_cost": line.snapshot_unit_cost,
        "counted_qty": line.counted_qty,
        "counted_weight_kg": line.counted_weight_kg,
        "diff_qty": line.diff_qty,
        "diff_weight_kg": line.diff_weight_kg,
        "diff_amount": line.diff_amount,
        "diff_type": line.diff_type,
        "row_status": line.row_status,
        "error_message": line.error_message,
        "remark": line.remark,
    }
```

Add `_stocktake_read`:

```python
def _stocktake_read(db: Session, stocktake_id: int) -> StocktakeRead:
    stocktake = db.get(Stocktake, stocktake_id)
    if not stocktake:
        raise HTTPException(status_code=404, detail="盘库单不存在")
    warehouses = stocktake_warehouses_for_read(db, stocktake_id)
    lines = stocktake_lines_for_read(db, stocktake_id)
    return StocktakeRead.model_validate(
        {
            "id": stocktake.id,
            "stocktake_no": stocktake.stocktake_no,
            "status": stocktake.status,
            "started_by_user_id": stocktake.started_by_user_id,
            "imported_by_user_id": stocktake.imported_by_user_id,
            "confirmed_by_user_id": stocktake.confirmed_by_user_id,
            "canceled_by_user_id": stocktake.canceled_by_user_id,
            "started_at": stocktake.started_at,
            "exported_at": stocktake.exported_at,
            "imported_at": stocktake.imported_at,
            "confirmed_at": stocktake.confirmed_at,
            "canceled_at": stocktake.canceled_at,
            "remark": stocktake.remark,
            "warehouses": [
                {
                    "id": row.id,
                    "warehouse_id": row.warehouse_id,
                    "warehouse_name": row.warehouse_name,
                    "warehouse_type": row.warehouse_type,
                    "status": row.status,
                }
                for row in warehouses
            ],
            "summary": _stocktake_summary(lines),
        }
    )
```

Add `_stocktake_preview_read`:

```python
def _stocktake_preview_read(db: Session, result: dict[str, object]) -> StocktakeImportPreviewRead:
    stocktake = result["stocktake"]
    lines = result["lines"]
    return StocktakeImportPreviewRead.model_validate(
        {
            "stocktake": _stocktake_read(db, stocktake.id),
            "lines": [_stocktake_line_read(line) for line in lines],
            "summary": result["summary"],
        }
    )
```

- [ ] **Step 3: Add list/detail/lines routes**

Add:

```python
@router.get("/stocktakes", response_model=list[StocktakeRead])
def list_inventory_stocktakes(
    limit: int = Query(default=100, ge=1, le=500),
    db: Session = Depends(get_db),
) -> list[StocktakeRead]:
    rows = db.scalars(select(Stocktake).order_by(Stocktake.started_at.desc(), Stocktake.id.desc()).limit(limit)).all()
    return [_stocktake_read(db, row.id) for row in rows]


@router.get("/stocktakes/{stocktake_id}", response_model=StocktakeRead)
def get_inventory_stocktake(stocktake_id: int, db: Session = Depends(get_db)) -> StocktakeRead:
    return _stocktake_read(db, stocktake_id)


@router.get("/stocktakes/{stocktake_id}/lines", response_model=list[StocktakeLineRead])
def list_inventory_stocktake_lines(stocktake_id: int, db: Session = Depends(get_db)) -> list[StocktakeLineRead]:
    return [StocktakeLineRead.model_validate(_stocktake_line_read(line)) for line in stocktake_lines_for_read(db, stocktake_id)]
```

- [ ] **Step 4: Run backend verification**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_excel_diff_confirm tests.test_stocktake_locking
.venv/bin/python -m compileall app
```

Expected: tests PASS and compileall PASS.

---

## Task 9: Frontend Stocktake Dialog API Wiring

**Files:**
- Modify: `frontend/src/views/InventoryLedgerView.vue`
- Create: `frontend/src/components/StocktakeDialog.vue`
- Modify: `frontend/src/utils/dictionaries.js`
- Modify: `frontend/src/styles/main.css`

- [ ] **Step 1: Add dictionary labels**

In `frontend/src/utils/dictionaries.js`, add exports:

```javascript
export function formatStocktakeStatusLabel(status) {
  return {
    LOCKED: "盘库中",
    EXPORTED: "已导出",
    IMPORTED: "已导入",
    CONFIRMED: "已确认",
    CANCELED: "已取消"
  }[String(status || "").toUpperCase()] || status || "-";
}

export function formatStocktakeDiffTypeLabel(type) {
  return {
    MATCH: "一致",
    GAIN: "盘盈",
    LOSS: "盘亏",
    NEW_LOT: "新增批次",
    MISSING: "缺失批次"
  }[String(type || "").toUpperCase()] || type || "-";
}
```

- [ ] **Step 2: Create dialog skeleton with props and events**

Create `frontend/src/components/StocktakeDialog.vue`:

```vue
<template>
  <Teleport to="body">
    <div v-if="open" class="workflow-drawer-mask stocktake-mask" @click.self="$emit('close')">
      <section class="stocktake-shell">
        <header class="stocktake-head">
          <div>
            <p class="eyebrow">嘉恒仓库 · 盘库</p>
            <h3>{{ activeStocktake ? activeStocktake.stocktake_no : "新建盘库" }}</h3>
          </div>
          <button class="ghost-button" type="button" @click="$emit('close')">关闭</button>
        </header>

        <nav class="stocktake-steps" aria-label="盘库步骤">
          <button v-for="step in steps" :key="step.key" type="button" class="stocktake-step" :class="{ active: activeStep === step.key }" @click="activeStep = step.key">
            <span>{{ step.no }}</span>
            <strong>{{ step.label }}</strong>
          </button>
        </nav>

        <section v-if="activeStep === 'select'" class="stocktake-panel">
          <div class="stocktake-warehouse-grid">
            <label v-for="warehouse in warehouses" :key="warehouse.id" class="stocktake-warehouse-card" :class="{ active: selectedWarehouseIds.includes(warehouse.id) }">
              <input v-model="selectedWarehouseIds" type="checkbox" :value="warehouse.id" />
              <strong>{{ warehouse.warehouse_name }}</strong>
              <span>{{ warehouseTypeText(warehouse.warehouse_type) }}</span>
            </label>
          </div>
          <div class="stocktake-footer">
            <button class="primary-button" type="button" :disabled="!selectedWarehouseIds.length || busy" @click="startStocktake">
              {{ busy ? "锁库中..." : "开始盘库并锁库" }}
            </button>
          </div>
        </section>

        <section v-if="activeStep === 'export'" class="stocktake-panel">
          <div class="stocktake-action-card">
            <strong>导出盘库清单</strong>
            <p>导出的 Excel 会按仓库分 sheet。锁库期间选中仓库不能执行出入库。</p>
            <button class="primary-button" type="button" :disabled="!activeStocktake || busy" @click="exportStocktake">导出盘库清单</button>
          </div>
        </section>

        <section v-if="activeStep === 'import'" class="stocktake-panel">
          <label class="import-dropzone">
            <input ref="fileInputRef" type="file" accept=".xlsx,.xlsm" @change="importStocktake" />
            <strong>{{ busy ? "导入中..." : "上传盘点后的 Excel" }}</strong>
            <span>系统会生成左右差异对比，不会直接改库存。</span>
          </label>
        </section>

        <section v-if="activeStep === 'diff'" class="stocktake-panel stocktake-diff-panel">
          <div class="stocktake-summary-grid">
            <article v-for="card in summaryCards" :key="card.key" class="stocktake-summary-card" :class="card.tone">
              <span>{{ card.label }}</span>
              <strong>{{ card.value }}</strong>
            </article>
          </div>
          <div class="stocktake-diff-toolbar">
            <button class="ghost-button" type="button" :class="{ active: diffFilter === 'ALL' }" @click="diffFilter = 'ALL'">全部</button>
            <button class="ghost-button" type="button" :class="{ active: diffFilter === 'DIFF' }" @click="diffFilter = 'DIFF'">只看差异</button>
            <button class="ghost-button" type="button" :class="{ active: diffFilter === 'ERROR' }" @click="diffFilter = 'ERROR'">只看异常</button>
          </div>
          <div class="stocktake-diff-table">
            <div class="stocktake-diff-column">
              <h4>系统库存</h4>
              <div v-for="line in filteredLines" :key="`sys-${line.id}`" class="stocktake-diff-row" :class="diffClass(line)">
                <strong>{{ line.item_code }} · {{ line.item_name }}</strong>
                <span>{{ line.lot_no || "-" }}</span>
                <span>数量 {{ formatQty(line.snapshot_qty) }} / 重量 {{ formatWeight(line.snapshot_weight_kg) }}</span>
              </div>
            </div>
            <div class="stocktake-diff-column">
              <h4>盘点结果</h4>
              <div v-for="line in filteredLines" :key="`count-${line.id}`" class="stocktake-diff-row" :class="diffClass(line)">
                <strong>{{ formatStocktakeDiffTypeLabel(line.diff_type) }}</strong>
                <span>{{ line.error_message || line.remark || "-" }}</span>
                <span>数量 {{ formatQty(line.counted_qty) }} / 重量 {{ formatWeight(line.counted_weight_kg) }}</span>
              </div>
            </div>
          </div>
          <div class="stocktake-footer">
            <input v-model.trim="confirmText" class="stocktake-confirm-input" placeholder="输入：确认盘库" />
            <button class="primary-button" type="button" :disabled="!canConfirm || busy" @click="confirmStocktake">
              {{ busy ? "确认中..." : "确认盘库并解锁" }}
            </button>
          </div>
        </section>
      </section>
    </div>
  </Teleport>
</template>
```

Add script:

```vue
<script setup>
import { computed, ref, watch } from "vue";

import { downloadResource, fetchResource, postResource, uploadResource } from "../services/api";
import { formatStocktakeDiffTypeLabel, formatStocktakeStatusLabel } from "../utils/dictionaries";
import { formatAmount, formatQty, formatWeight } from "../utils/formatters";

const props = defineProps({
  open: { type: Boolean, default: false },
  warehouses: { type: Array, default: () => [] },
  defaultWarehouseId: { type: Number, default: 0 }
});
const emit = defineEmits(["close", "changed", "message", "error"]);

const steps = [
  { key: "select", no: "01", label: "选择仓库" },
  { key: "export", no: "02", label: "导出清单" },
  { key: "import", no: "03", label: "导入盘点" },
  { key: "diff", no: "04", label: "差异确认" }
];

const activeStep = ref("select");
const selectedWarehouseIds = ref([]);
const activeStocktake = ref(null);
const previewLines = ref([]);
const previewSummary = ref({});
const busy = ref(false);
const fileInputRef = ref(null);
const diffFilter = ref("ALL");
const confirmText = ref("");

watch(
  () => props.open,
  (open) => {
    if (!open) return;
    activeStep.value = "select";
    selectedWarehouseIds.value = props.defaultWarehouseId ? [props.defaultWarehouseId] : [];
    activeStocktake.value = null;
    previewLines.value = [];
    previewSummary.value = {};
    diffFilter.value = "ALL";
    confirmText.value = "";
  }
);

const summaryCards = computed(() => [
  { key: "match", label: "一致", value: previewSummary.value.match_count || 0, tone: "neutral" },
  { key: "gain", label: "盘盈", value: previewSummary.value.gain_count || 0, tone: "gain" },
  { key: "loss", label: "盘亏", value: previewSummary.value.loss_count || 0, tone: "loss" },
  { key: "new", label: "新增批次", value: previewSummary.value.new_lot_count || 0, tone: "new" },
  { key: "missing", label: "缺失批次", value: previewSummary.value.missing_count || 0, tone: "missing" },
  { key: "error", label: "异常", value: previewSummary.value.error_count || 0, tone: "error" },
  { key: "amount", label: "影响金额", value: `¥${formatAmount(previewSummary.value.diff_amount || 0)}`, tone: "amount" }
]);

const filteredLines = computed(() => {
  if (diffFilter.value === "ERROR") {
    return previewLines.value.filter((line) => line.row_status === "ERROR");
  }
  if (diffFilter.value === "DIFF") {
    return previewLines.value.filter((line) => line.diff_type !== "MATCH");
  }
  return previewLines.value;
});

const canConfirm = computed(() => activeStocktake.value?.id && confirmText.value === "确认盘库" && Number(previewSummary.value.error_count || 0) === 0);

function warehouseTypeText(type) {
  return { RAW: "原材料库", SEMI: "半成品库", FINISHED: "成品库", AUX: "辅料库" }[String(type || "").toUpperCase()] || type || "-";
}

function diffClass(line) {
  return `diff-${String(line.diff_type || "MATCH").toLowerCase()} ${line.row_status === "ERROR" ? "diff-error" : ""}`;
}

async function startStocktake() {
  busy.value = true;
  try {
    activeStocktake.value = await postResource("/inventory/stocktakes", {
      warehouse_ids: selectedWarehouseIds.value,
      remark: "嘉恒仓库盘库"
    });
    activeStep.value = "export";
    emit("message", `盘库单 ${activeStocktake.value.stocktake_no} 已创建并锁库`);
    emit("changed");
  } catch (error) {
    emit("error", error.message || "开始盘库失败");
  } finally {
    busy.value = false;
  }
}

async function exportStocktake() {
  if (!activeStocktake.value?.id) return;
  busy.value = true;
  try {
    await downloadResource(`/inventory/stocktakes/${activeStocktake.value.id}/export`, `${activeStocktake.value.stocktake_no}_盘库清单.xlsx`);
    activeStep.value = "import";
    emit("message", "盘库清单已导出");
  } catch (error) {
    emit("error", error.message || "导出盘库清单失败");
  } finally {
    busy.value = false;
  }
}

async function importStocktake(event) {
  const file = event.target.files?.[0];
  if (!file || !activeStocktake.value?.id) return;
  busy.value = true;
  try {
    const formData = new FormData();
    formData.append("file", file);
    const result = await uploadResource(`/inventory/stocktakes/${activeStocktake.value.id}/import`, formData);
    activeStocktake.value = result.stocktake;
    previewLines.value = result.lines;
    previewSummary.value = result.summary;
    activeStep.value = "diff";
    emit("message", "盘点清单已导入，请核对差异");
  } catch (error) {
    emit("error", error.message || "导入盘点清单失败");
  } finally {
    busy.value = false;
    if (fileInputRef.value) fileInputRef.value.value = "";
  }
}

async function confirmStocktake() {
  if (!canConfirm.value) return;
  busy.value = true;
  try {
    activeStocktake.value = await postResource(`/inventory/stocktakes/${activeStocktake.value.id}/confirm`, {
      confirm_text: confirmText.value,
      remark: "盘库确认"
    });
    emit("message", `盘库单 ${activeStocktake.value.stocktake_no} 已确认并解锁`);
    emit("changed");
    emit("close");
  } catch (error) {
    emit("error", error.message || "确认盘库失败");
  } finally {
    busy.value = false;
  }
}
</script>
```

- [ ] **Step 3: Verify component compiles after wiring in next task**

Do not run build yet; `StocktakeDialog.vue` is not imported until Task 10.

---

## Task 10: Add Stocktake Entry Above Warehouse Bar

**Files:**
- Modify: `frontend/src/views/InventoryLedgerView.vue`
- Modify: `frontend/src/styles/main.css`

- [ ] **Step 1: Import dialog and API helpers**

Modify imports in `frontend/src/views/InventoryLedgerView.vue`:

```javascript
import StocktakeDialog from "../components/StocktakeDialog.vue";
```

Add state:

```javascript
const stocktakeDialogOpen = ref(false);
```

Update `useBodyScrollLock` condition:

```javascript
useBodyScrollLock(
  computed(
    () =>
      detailDrawerOpen.value ||
      genericDrawerOpen.value ||
      productionDrawerOpen.value ||
      completionDrawerOpen.value ||
      deliveryDrawerOpen.value ||
      stocktakeDialogOpen.value
  )
);
```

Add computed:

```javascript
const activeWarehouseId = computed(() => activeWarehouses.value[0]?.id || 0);
```

Add handlers:

```javascript
function openStocktakeDialog() {
  resetFeedback();
  stocktakeDialogOpen.value = true;
}

function closeStocktakeDialog() {
  stocktakeDialogOpen.value = false;
}

function handleStocktakeMessage(message) {
  feedbackMessage.value = message;
  errorMessage.value = "";
}

function handleStocktakeError(message) {
  errorMessage.value = message;
  feedbackMessage.value = "";
}
```

- [ ] **Step 2: Add stocktake button above warehouse strip**

In the template, insert between panel header and `<div class="inventory-warehouse-strip"...>`:

```vue
      <div class="inventory-stocktake-bar">
        <button class="stocktake-launch-button no-auto-icon" type="button" @click="openStocktakeDialog">
          <span class="stocktake-launch-icon" aria-hidden="true">
            <svg viewBox="0 0 24 24">
              <path d="M5 4h14v16H5z" />
              <path d="M8 8h8" />
              <path d="M8 12h8" />
              <path d="M8 16h5" />
            </svg>
          </span>
          <strong>盘库</strong>
        </button>
      </div>
```

Before closing `</section>` root, add:

```vue
    <StocktakeDialog
      :open="stocktakeDialogOpen"
      :warehouses="warehouses"
      :default-warehouse-id="activeWarehouseId"
      @close="closeStocktakeDialog"
      @changed="loadAll"
      @message="handleStocktakeMessage"
      @error="handleStocktakeError"
    />
```

- [ ] **Step 3: Add styles**

Append to `frontend/src/styles/main.css`:

```css
.inventory-stocktake-bar {
  display: flex;
  justify-content: flex-end;
  margin: 4px 0 12px;
}

.stocktake-launch-button {
  display: inline-flex;
  align-items: center;
  gap: 10px;
  border: 1px solid rgba(37, 99, 235, 0.38);
  border-radius: 16px;
  padding: 10px 18px;
  background: linear-gradient(135deg, rgba(255, 255, 255, 0.92), rgba(219, 234, 254, 0.82));
  color: #1d4ed8;
  box-shadow: 0 12px 28px rgba(37, 99, 235, 0.16);
  cursor: pointer;
}

.stocktake-launch-button:hover {
  transform: translateY(-2px);
  box-shadow: 0 18px 38px rgba(37, 99, 235, 0.22);
}

.stocktake-launch-icon {
  width: 26px;
  height: 26px;
  display: inline-grid;
  place-items: center;
}

.stocktake-launch-icon svg,
.stocktake-summary-card svg {
  width: 100%;
  height: 100%;
  fill: none;
  stroke: currentColor;
  stroke-width: 1.8;
  stroke-linecap: round;
  stroke-linejoin: round;
}

.stocktake-mask {
  align-items: center;
  justify-content: center;
}

.stocktake-shell {
  width: min(1500px, 92vw);
  height: min(860px, 90vh);
  display: grid;
  grid-template-rows: auto auto 1fr;
  border: 1px solid rgba(148, 163, 184, 0.55);
  border-radius: 26px;
  background: rgba(248, 250, 252, 0.94);
  box-shadow: 0 30px 90px rgba(15, 23, 42, 0.28);
  overflow: hidden;
}

.stocktake-head,
.stocktake-footer {
  display: flex;
  align-items: center;
  justify-content: space-between;
  gap: 16px;
  padding: 18px 22px;
  border-bottom: 1px solid rgba(148, 163, 184, 0.35);
}

.stocktake-steps {
  display: flex;
  gap: 10px;
  padding: 14px 20px;
  background: rgba(241, 245, 249, 0.78);
}

.stocktake-step,
.stocktake-warehouse-card,
.stocktake-summary-card,
.stocktake-action-card {
  border: 1px solid rgba(148, 163, 184, 0.42);
  border-radius: 18px;
  background: rgba(255, 255, 255, 0.86);
  box-shadow: 0 12px 30px rgba(15, 23, 42, 0.08);
}

.stocktake-step {
  display: inline-flex;
  align-items: center;
  gap: 10px;
  padding: 10px 14px;
  cursor: pointer;
}

.stocktake-step.active {
  border-color: rgba(37, 99, 235, 0.7);
  color: #1d4ed8;
  box-shadow: 0 16px 36px rgba(37, 99, 235, 0.16);
}

.stocktake-panel {
  overflow: auto;
  padding: 20px;
}

.stocktake-warehouse-grid {
  display: grid;
  grid-template-columns: repeat(auto-fit, minmax(220px, 1fr));
  gap: 14px;
}

.stocktake-warehouse-card {
  display: grid;
  gap: 8px;
  padding: 16px;
  cursor: pointer;
}

.stocktake-warehouse-card.active {
  border-color: rgba(22, 163, 74, 0.72);
  background: rgba(240, 253, 244, 0.9);
}

.stocktake-summary-grid {
  display: grid;
  grid-template-columns: repeat(7, minmax(120px, 1fr));
  gap: 12px;
  margin-bottom: 14px;
}

.stocktake-summary-card {
  padding: 12px 14px;
}

.stocktake-summary-card strong {
  display: block;
  margin-top: 6px;
  font-size: 22px;
}

.stocktake-summary-card.gain,
.stocktake-summary-card.new {
  background: rgba(240, 253, 244, 0.92);
  color: #166534;
}

.stocktake-summary-card.loss,
.stocktake-summary-card.missing {
  background: rgba(254, 242, 242, 0.92);
  color: #991b1b;
}

.stocktake-summary-card.error {
  background: rgba(127, 29, 29, 0.92);
  color: #fff;
}

.stocktake-diff-toolbar {
  display: flex;
  gap: 10px;
  margin-bottom: 12px;
}

.stocktake-diff-table {
  display: grid;
  grid-template-columns: 1fr 1fr;
  gap: 12px;
}

.stocktake-diff-column {
  border: 1px solid rgba(148, 163, 184, 0.42);
  border-radius: 18px;
  overflow: hidden;
  background: rgba(255, 255, 255, 0.86);
}

.stocktake-diff-column h4 {
  margin: 0;
  padding: 12px 14px;
  background: rgba(226, 232, 240, 0.8);
}

.stocktake-diff-row {
  display: grid;
  gap: 4px;
  min-height: 76px;
  padding: 12px 14px;
  border-top: 1px solid rgba(226, 232, 240, 0.9);
}

.stocktake-diff-row.diff-gain,
.stocktake-diff-row.diff-new_lot {
  background: rgba(220, 252, 231, 0.86);
}

.stocktake-diff-row.diff-loss,
.stocktake-diff-row.diff-missing {
  background: rgba(254, 226, 226, 0.9);
}

.stocktake-diff-row.diff-error {
  background: #7f1d1d;
  color: #fff;
}

.stocktake-confirm-input {
  min-width: 220px;
  border: 1px solid rgba(148, 163, 184, 0.7);
  border-radius: 12px;
  padding: 10px 12px;
}
```

- [ ] **Step 4: Run frontend build**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npx vite build --mode development
```

Expected: build PASS.

---

## Task 11: Add Stocktake History View Inside Dialog

**Files:**
- Modify: `frontend/src/components/StocktakeDialog.vue`

- [ ] **Step 1: Add history step**

Change `steps` in `StocktakeDialog.vue`:

```javascript
const steps = [
  { key: "select", no: "01", label: "选择仓库" },
  { key: "export", no: "02", label: "导出清单" },
  { key: "import", no: "03", label: "导入盘点" },
  { key: "diff", no: "04", label: "差异确认" },
  { key: "history", no: "05", label: "盘库记录" }
];
```

Add state:

```javascript
const historyRows = ref([]);
```

Add loader:

```javascript
async function loadHistory() {
  try {
    historyRows.value = await fetchResource("/inventory/stocktakes?limit=100", []);
  } catch (error) {
    emit("error", error.message || "读取盘库记录失败");
  }
}
```

Update watcher when opened:

```javascript
watch(
  () => props.open,
  async (open) => {
    if (!open) return;
    activeStep.value = "select";
    selectedWarehouseIds.value = props.defaultWarehouseId ? [props.defaultWarehouseId] : [];
    activeStocktake.value = null;
    previewLines.value = [];
    previewSummary.value = {};
    diffFilter.value = "ALL";
    confirmText.value = "";
    await loadHistory();
  }
);
```

- [ ] **Step 2: Add history template**

Add after diff section:

```vue
        <section v-if="activeStep === 'history'" class="stocktake-panel">
          <div class="table-wrap">
            <table class="data-table compact-table">
              <thead>
                <tr>
                  <th>盘库单号</th>
                  <th>仓库</th>
                  <th>状态</th>
                  <th>开始时间</th>
                  <th>确认时间</th>
                  <th>盘盈</th>
                  <th>盘亏</th>
                  <th>新增批次</th>
                  <th>缺失批次</th>
                  <th>影响金额</th>
                </tr>
              </thead>
              <tbody>
                <tr v-for="row in historyRows" :key="row.id">
                  <td>{{ row.stocktake_no }}</td>
                  <td>{{ row.warehouses.map((item) => item.warehouse_name).join("、") }}</td>
                  <td>{{ formatStocktakeStatusLabel(row.status) }}</td>
                  <td>{{ row.started_at }}</td>
                  <td>{{ row.confirmed_at || "-" }}</td>
                  <td>{{ row.summary.gain_count || 0 }}</td>
                  <td>{{ row.summary.loss_count || 0 }}</td>
                  <td>{{ row.summary.new_lot_count || 0 }}</td>
                  <td>{{ row.summary.missing_count || 0 }}</td>
                  <td>¥{{ formatAmount(row.summary.diff_amount || 0) }}</td>
                </tr>
                <tr v-if="!historyRows.length">
                  <td colspan="10" class="empty-row">暂无盘库记录</td>
                </tr>
              </tbody>
            </table>
          </div>
        </section>
```

- [ ] **Step 3: Run frontend build**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npx vite build --mode development
```

Expected: build PASS.

---

## Task 12: Final Verification and Regression Sweep

**Files:**
- Verify all touched backend and frontend files.

- [ ] **Step 1: Run stocktake unit tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest tests.test_stocktake_locking tests.test_stocktake_excel_diff_confirm
```

Expected: PASS.

- [ ] **Step 2: Run backend compile**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m compileall app
```

Expected: PASS.

- [ ] **Step 3: Run frontend build**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npx vite build --mode development
```

Expected: PASS.

- [ ] **Step 4: Manual browser workflow check**

Run the app using the project's normal backend/frontend commands, then verify:

```text
1. 打开 采购与库存 -> 嘉恒仓库。
2. 点击仓库 bar 上方“盘库”。
3. 默认勾选当前仓库。
4. 点击“开始盘库并锁库”。
5. 在锁库期间尝试对该仓库做入库或出库，页面应收到“正在盘库”错误提示。
6. 导出 Excel，确认 sheet 名按仓库分开。
7. 修改“盘点重量(kg)”或“盘点数量”后导入。
8. 差异页面左右两表对齐显示。
9. 盘盈、盘亏、新增批次、缺失批次颜色明显。
10. 输入“确认盘库”并确认。
11. 库存列表刷新后显示调整后的库存。
12. 再次进行入库或出库，锁已解除。
13. 打开盘库记录，能看到本次盘库和差异汇总。
```

- [ ] **Step 5: Run broader backend tests and document known unrelated failures**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m unittest discover -s tests
```

Expected: stocktake tests PASS. If existing purchase receipt or purchase lock tests fail with previously known failures, record exact test names in the final response and do not change unrelated purchase status behavior unless the user asks for that fix.

- [ ] **Step 6: Commit if this workspace is a git repository**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git rev-parse --is-inside-work-tree
```

If output is `true`, commit:

```bash
git add backend/app/models/operations.py backend/app/schemas/operations.py backend/app/services/stocktake.py backend/app/api/routes/inventory.py backend/app/api/routes/purchase.py backend/app/api/routes/production.py backend/app/api/routes/sales.py backend/sql/add_stocktake_tables.sql backend/tests/test_stocktake_locking.py backend/tests/test_stocktake_excel_diff_confirm.py frontend/src/views/InventoryLedgerView.vue frontend/src/components/StocktakeDialog.vue frontend/src/styles/main.css frontend/src/utils/dictionaries.js
git commit -m "feat: add locked warehouse stocktake"
```

If output says this is not a git repository, skip the commit and mention that the workspace has no `.git` directory.

---

## Self-Review

- Spec coverage: The plan implements lock-based stocktake, multi-warehouse Excel export with sheets, import, side-by-side diff, confirmation adjustment, audit records, history display, and lock guards on warehouse-changing flows.
- Placeholder scan: The plan contains concrete function names, file paths, validation behavior, commands, and expected outputs.
- Type consistency: `Stocktake`, `StocktakeWarehouse`, `StocktakeLine`, `StocktakeAdjustment`, `start_stocktake`, `build_stocktake_workbook`, `import_stocktake_workbook`, `confirm_stocktake`, and `ensure_warehouses_unlocked` are introduced before dependent tasks use them.