tangxh/ForgeFlow-ERP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
main
ForgeFlow-ERP/docs/superpowers/plans/2026-06-07-raw-lot-work-order-inbound.md
汤学会 c927886888 Initial commit
2026-06-12 16:00:56 +08:00

79 KiB
Raw Permalink Blame History

Raw Lot Work Order Inbound 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: Implement原材料库存批次号全库递增、生产出库单批次合并工单、小程序按材料库存批次号报工、以及嘉恒仓库“生产工单入库”统一闭环成品/余料/废料。

Architecture: Add a small system-config service for the raw lot prefix, keep ERP production work orders as the internal container, and make raw stock lot IDs the external production-facing identifier. Production outbound becomes single-lot and merges into a deterministic daily work order; production work-order inbound becomes one transactional backend API that posts finished goods, returns surplus to the original raw lot, posts scrap, and optionally settles the work order.

Tech Stack: FastAPI, SQLAlchemy ORM, MySQL/SQLite-compatible tests, Vue 3 + Vite, WeChat miniapp frontend/backend, existing pytest and Node script tests.

File Structure

Backend ERP files:

  • Create backend/sql/add_raw_lot_work_order_inbound_patch.sql: create generic sys_system_config table and seed RAW_MATERIAL_LOT_PREFIX=YL.
  • Create backend/app/services/system_config.py: normalize config values and read/write sys_system_config.
  • Create backend/app/services/production_work_order_inbound.py: calculation and transactional posting for production work-order inbound.
  • Modify backend/app/models/org.py: add SystemConfig ORM model.
  • Modify backend/app/schemas/database.py: add config read/update schemas.
  • Modify backend/app/schemas/operations.py: add production work-order inbound preview/create/read schemas.
  • Modify backend/app/api/routes/system_extension.py: expose raw material lot prefix config endpoints.
  • Modify backend/app/services/operations.py: replace material-name lot numbering with raw-warehouse global prefix numbering and add deterministic work-order number helper.
  • Modify backend/app/api/routes/purchase.py: keep purchase receipt path on build_inventory_lot_no, add test coverage only unless code needs explicit import changes.
  • Modify backend/app/api/routes/inventory.py: use new raw lot numbering for raw opening/customer-supplied/special inbound paths and add read helpers used by frontend.
  • Modify backend/app/api/routes/production.py: enforce single stock lot production outbound, merge same-day same-lot same-product work orders, resolve miniapp reports by material stock lot no, and expose production work-order inbound endpoints.

ERP tests:

  • Create backend/tests/test_system_config_raw_lot_prefix.py.
  • Modify backend/tests/test_inventory_lot_numbering.py.
  • Modify backend/tests/test_customer_supplied_inbound_lot_no.py.
  • Modify backend/tests/test_opening_inventory_import.py.
  • Modify backend/tests/test_selected_stock_lot_production_issue.py.
  • Create backend/tests/test_production_work_order_inbound.py.

ERP frontend files:

  • Modify frontend/src/components/StockLotTagSelect.vue: add maxSelections support for single-select production outbound.
  • Modify frontend/src/views/SystemExtensionView.vue: add raw lot prefix config card.
  • Modify frontend/src/views/InventoryLedgerView.vue: production outbound single-select UI and new “生产工单入库” entry/drawer.
  • Create frontend/scripts/test-production-work-order-inbound-ui.mjs.
  • Modify frontend/scripts/test-special-adjustment-ui.mjs only if shared warehouse action ordering assertions need updating.

Miniapp files:

  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py: return material stock lot numbers instead of JH0001 work-order short codes.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_work_order_batch_options.py.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm/reportForm.wxml.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm/reportForm.js.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records/records.wxml.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxml.
  • Modify /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js.

Task 1: Add System Config Storage For Raw Lot Prefix

Files:

  • Create: backend/sql/add_raw_lot_work_order_inbound_patch.sql

  • Create: backend/app/services/system_config.py

  • Modify: backend/app/models/org.py

  • Modify: backend/app/schemas/database.py

  • Modify: backend/app/api/routes/system_extension.py

  • Test: backend/tests/test_system_config_raw_lot_prefix.py

  • Step 1: Write failing backend tests for config default, normalization, and update

Add backend/tests/test_system_config_raw_lot_prefix.py:

from __future__ import annotations

import unittest
from datetime import datetime

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.miniapp  # noqa: E402,F401
import app.models.operations  # noqa: E402,F401
import app.models.org  # noqa: E402,F401
import app.models.planning  # noqa: E402,F401
import app.models.sales  # noqa: E402,F401
from app.models.base import Base  # noqa: E402
from app.models.org import SystemConfig  # noqa: E402
from app.services.system_config import (  # noqa: E402
    RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE,
    get_raw_material_lot_prefix,
    normalize_raw_material_lot_prefix,
    upsert_system_config,
)


class RawLotPrefixSystemConfigTest(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()

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

    def test_default_prefix_is_yl_when_config_missing(self) -> None:
        self.assertEqual(get_raw_material_lot_prefix(self.db), "YL")

    def test_prefix_normalization_keeps_only_uppercase_letters_and_numbers(self) -> None:
        self.assertEqual(normalize_raw_material_lot_prefix(" yl-01 "), "YL01")
        self.assertEqual(normalize_raw_material_lot_prefix(""), "YL")
        self.assertEqual(normalize_raw_material_lot_prefix("材料"), "YL")

    def test_upsert_raw_lot_prefix_config(self) -> None:
        now = datetime.now()
        upsert_system_config(
            self.db,
            config_code=RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE,
            config_name="原材料库存批次号前缀",
            config_value="rm",
            remark="测试配置",
            updated_by=1,
            now=now,
        )
        self.db.commit()

        row = self.db.scalar(select(SystemConfig).where(SystemConfig.config_code == RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE))
        self.assertIsNotNone(row)
        self.assertEqual(row.config_value, "RM")
        self.assertEqual(get_raw_material_lot_prefix(self.db), "RM")
  • Step 2: Run failing config tests

Run:

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

Expected: FAIL because SystemConfig and app.services.system_config do not exist.

  • Step 3: Add SQL patch

Create backend/sql/add_raw_lot_work_order_inbound_patch.sql:

CREATE TABLE IF NOT EXISTS sys_system_config (
  id BIGINT PRIMARY KEY AUTO_INCREMENT,
  config_code VARCHAR(100) NOT NULL,
  config_name VARCHAR(100) NOT NULL,
  config_value VARCHAR(500) NOT NULL,
  remark VARCHAR(500) NULL,
  status VARCHAR(32) NOT NULL DEFAULT 'ACTIVE',
  created_by BIGINT NULL,
  updated_by BIGINT NULL,
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_sys_system_config_code (config_code),
  KEY idx_sys_system_config_status (status)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;

INSERT INTO sys_system_config (config_code, config_name, config_value, remark, status)
SELECT 'RAW_MATERIAL_LOT_PREFIX', '原材料库存批次号前缀', 'YL', '原材料库存批次号默认前缀，例如 YL0001', 'ACTIVE'
WHERE NOT EXISTS (
  SELECT 1 FROM sys_system_config WHERE config_code = 'RAW_MATERIAL_LOT_PREFIX'
);
  • Step 4: Add ORM model

Modify backend/app/models/org.py after AiAssistantConfig:

class SystemConfig(Base):
    __tablename__ = "sys_system_config"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    config_code: Mapped[str] = mapped_column(String(100))
    config_name: Mapped[str] = mapped_column(String(100))
    config_value: Mapped[str] = mapped_column(String(500))
    remark: Mapped[str | None] = mapped_column(String(500), nullable=True)
    status: Mapped[str] = mapped_column(String(32), server_default=text("'ACTIVE'"))
    created_by: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    updated_by: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)
  • Step 5: Add config service

Create backend/app/services/system_config.py:

from __future__ import annotations

import re
from datetime import datetime

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.org import SystemConfig

RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE = "RAW_MATERIAL_LOT_PREFIX"
RAW_MATERIAL_LOT_PREFIX_DEFAULT = "YL"


def normalize_raw_material_lot_prefix(value: object | None) -> str:
    normalized = "".join(re.findall(r"[A-Za-z0-9]", str(value or "").upper()))
    return normalized or RAW_MATERIAL_LOT_PREFIX_DEFAULT


def get_system_config_value(db: Session, config_code: str, default: str = "") -> str:
    row = db.scalar(
        select(SystemConfig)
        .where(SystemConfig.config_code == config_code, SystemConfig.status == "ACTIVE")
        .limit(1)
    )
    return str(row.config_value if row else default)


def get_raw_material_lot_prefix(db: Session) -> str:
    return normalize_raw_material_lot_prefix(
        get_system_config_value(db, RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE, RAW_MATERIAL_LOT_PREFIX_DEFAULT)
    )


def upsert_system_config(
    db: Session,
    *,
    config_code: str,
    config_name: str,
    config_value: object,
    remark: str | None = None,
    updated_by: int | None = None,
    now: datetime | None = None,
) -> SystemConfig:
    current_time = now or datetime.now()
    normalized_value = (
        normalize_raw_material_lot_prefix(config_value)
        if config_code == RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE
        else str(config_value or "").strip()
    )
    row = db.scalar(select(SystemConfig).where(SystemConfig.config_code == config_code).limit(1))
    if not row:
        row = SystemConfig(
            config_code=config_code,
            config_name=config_name,
            config_value=normalized_value,
            remark=remark,
            status="ACTIVE",
            created_by=updated_by,
            updated_by=updated_by,
            created_at=current_time,
            updated_at=current_time,
        )
    else:
        row.config_name = config_name
        row.config_value = normalized_value
        row.remark = remark
        row.status = "ACTIVE"
        row.updated_by = updated_by
        row.updated_at = current_time
    db.add(row)
    db.flush()
    return row
  • Step 6: Add schemas and API endpoints

Modify backend/app/schemas/database.py:

class SystemConfigUpdate(BaseModel):
    config_value: str = Field(min_length=1, max_length=500)
    remark: str | None = None


class SystemConfigRead(BaseModel):
    config_code: str
    config_name: str
    config_value: str
    remark: str | None = None
    updated_at: datetime | None = None

Modify backend/app/api/routes/system_extension.py imports:

from app.models.org import AiAssistantConfig, SystemConfig
from app.schemas.database import SystemConfigRead, SystemConfigUpdate
from app.services.system_config import RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE, get_raw_material_lot_prefix, upsert_system_config

Add endpoints near assistant config endpoints:

def _system_config_read(row: SystemConfig | None, *, config_code: str, config_name: str, default_value: str) -> SystemConfigRead:
    return SystemConfigRead(
        config_code=config_code,
        config_name=row.config_name if row else config_name,
        config_value=row.config_value if row else default_value,
        remark=row.remark if row else None,
        updated_at=row.updated_at if row else None,
    )


@router.get("/raw-material-lot-prefix", response_model=SystemConfigRead)
def get_raw_material_lot_prefix_config(
    context: AuthContext = Depends(require_system_admin),
    db: Session = Depends(get_db),
) -> SystemConfigRead:
    _ = context
    row = db.scalar(select(SystemConfig).where(SystemConfig.config_code == RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE).limit(1))
    return _system_config_read(
        row,
        config_code=RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE,
        config_name="原材料库存批次号前缀",
        default_value=get_raw_material_lot_prefix(db),
    )


@router.put("/raw-material-lot-prefix", response_model=SystemConfigRead)
def save_raw_material_lot_prefix_config(
    payload: SystemConfigUpdate,
    context: AuthContext = Depends(require_system_admin),
    db: Session = Depends(get_db),
) -> SystemConfigRead:
    row = upsert_system_config(
        db,
        config_code=RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE,
        config_name="原材料库存批次号前缀",
        config_value=payload.config_value,
        remark=payload.remark or "原材料库存批次号默认前缀，例如 YL0001",
        updated_by=context.user.id,
    )
    db.commit()
    db.refresh(row)
    return _system_config_read(
        row,
        config_code=RAW_MATERIAL_LOT_PREFIX_CONFIG_CODE,
        config_name="原材料库存批次号前缀",
        default_value=get_raw_material_lot_prefix(db),
    )
  • Step 7: Run config tests

Run:

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

Expected: PASS.

Task 2: Change Raw Material Lot Numbering To Global Prefix Sequence

Files:

  • Modify: backend/app/services/operations.py

  • Modify: backend/tests/test_inventory_lot_numbering.py

  • Step 1: Replace old numbering tests with new global sequence tests

Modify backend/tests/test_inventory_lot_numbering.py to assert:

def test_inventory_lot_no_starts_from_global_raw_sequence(self) -> None:
    self.assertEqual(build_inventory_lot_no(self.db, self.material.id), "YL0001")


def test_inventory_lot_no_increments_across_materials(self) -> None:
    other_material = Item(id=2, item_code="原材料00002", item_name="冷轧板", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE")
    self.db.add(other_material)
    self.db.add(
        StockLot(
            lot_no="YL0001",
            lot_role="INBOUND_RAW",
            item_id=other_material.id,
            warehouse_id=self.warehouse.id,
            source_doc_type="TEST",
            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("5"),
            quality_status="PASS",
            status="AVAILABLE",
        )
    )
    self.db.commit()

    self.assertEqual(build_inventory_lot_no(self.db, self.material.id), "YL0002")


def test_inventory_lot_no_expands_after_9999(self) -> None:
    self.db.add(
        StockLot(
            lot_no="YL9999",
            lot_role="INBOUND_RAW",
            item_id=self.material.id,
            warehouse_id=self.warehouse.id,
            source_doc_type="TEST",
            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("5"),
            quality_status="PASS",
            status="AVAILABLE",
        )
    )
    self.db.commit()

    self.assertEqual(build_inventory_lot_no(self.db, self.material.id), "YL10000")
  • Step 2: Run numbering tests to verify failure

Run:

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

Expected: FAIL because build_inventory_lot_no still returns JH_物料名称编码_0001.

  • Step 3: Implement global raw lot number helper

Modify backend/app/services/operations.py imports:

from app.services.system_config import get_raw_material_lot_prefix

Replace _build_inventory_lot_prefix and build_inventory_lot_no with:

def _raw_lot_sequence_width(next_sequence: int) -> int:
    return 4 if next_sequence <= 9999 else len(str(next_sequence))


def build_inventory_lot_no(db: Session, material_item_id: int) -> str:
    _ = material_item_id
    prefix = get_raw_material_lot_prefix(db)
    existing_lot_nos = db.scalars(select(StockLot.lot_no).where(StockLot.lot_no.like(f"{prefix}%"))).all()
    max_sequence = 0
    for lot_no in existing_lot_nos:
        match = re.fullmatch(rf"{re.escape(prefix)}(\d+)", str(lot_no or ""))
        if match:
            max_sequence = max(max_sequence, int(match.group(1)))
    sequence = max_sequence + 1
    candidate = f"{prefix}{sequence:0{_raw_lot_sequence_width(sequence)}d}"
    while db.scalar(select(func.count(StockLot.id)).where(StockLot.lot_no == candidate)):
        sequence += 1
        candidate = f"{prefix}{sequence:0{_raw_lot_sequence_width(sequence)}d}"
    return candidate
  • Step 4: Run numbering tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_inventory_lot_numbering.py tests/test_system_config_raw_lot_prefix.py -q

Expected: PASS.

Task 3: Apply New Lot Numbering To Raw Inbound Paths

Files:

  • Modify: backend/app/api/routes/inventory.py

  • Modify: backend/app/api/routes/purchase.py

  • Modify: backend/app/services/special_inventory_adjustment.py

  • Modify: backend/tests/test_customer_supplied_inbound_lot_no.py

  • Modify: backend/tests/test_opening_inventory_import.py

  • Modify: backend/tests/test_special_warehouse_adjustment.py

  • Step 1: Write or update tests for raw inbound paths

Update assertions so raw inbound paths generate YL0001, YL0002 and never hand-entered material-name batch numbers.

For backend/tests/test_customer_supplied_inbound_lot_no.py, assert:

self.assertEqual(created_lot.lot_no, "YL0001")
self.assertEqual(second_created_lot.lot_no, "YL0002")

For backend/tests/test_opening_inventory_import.py, assert imported raw opening lots use:

self.assertEqual([lot.lot_no for lot in raw_lots], ["YL0001", "YL0002"])

For backend/tests/test_special_warehouse_adjustment.py, assert special raw inbound uses:

self.assertRegex(raw_lot.lot_no, r"^YL\d{4,}$")
  • Step 2: Run inbound tests to verify failures

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest \
  tests/test_customer_supplied_inbound_lot_no.py \
  tests/test_opening_inventory_import.py \
  tests/test_special_warehouse_adjustment.py \
  -q

Expected: FAIL where old raw-specific lot builders still produce KLRK or material-name lot numbers.

  • Step 3: Route raw warehouse generic inbound through build_inventory_lot_no

Modify backend/app/api/routes/inventory.py in the # build lot_no block:

    if not lot_no:
        if warehouse_type == "RAW" and biz_type in {"OPENING", "CUSTOMER_SUPPLIED"}:
            lot_no = build_inventory_lot_no(db, item.id)
        elif biz_type in ("OPENING",):
            lot_no = _build_opening_lot_no(db, item, warehouse)
        elif biz_type in ("PRODUCTION_SURPLUS", "OUTSOURCING_SURPLUS"):
            tag = "SCSY" if biz_type == "PRODUCTION_SURPLUS" else "WWSY"
            lot_no = _build_surplus_lot_no(db, item, warehouse, tag)
        elif biz_type in ("PRODUCTION_SCRAP_IN", "OUTSOURCING_SCRAP_IN", "REWORK_SCRAP_IN"):
            tag = "SCFL" if biz_type == "PRODUCTION_SCRAP_IN" else "FWFL" if biz_type == "REWORK_SCRAP_IN" else "WWFL"
            lot_no = _build_surplus_lot_no(db, item, warehouse, tag)
        elif biz_type == "WIP_IN":
            lot_no = _build_wip_lot_no(db, item, warehouse)
        elif biz_type == "PRODUCTION_FINISHED":
            lot_no = _build_finished_lot_no(db, item, warehouse)
        elif biz_type == "OUTSOURCING_IN":
            lot_no = _build_surplus_lot_no(db, item, warehouse, "WWJK")
        elif biz_type == "CUSTOMER_SUPPLIED":
            lot_no = _build_customer_supplied_lot_no(db, item)
        else:
            lot_no = _build_opening_lot_no(db, item, warehouse)

Confirm backend/app/api/routes/purchase.py and backend/app/services/special_inventory_adjustment.py call build_inventory_lot_no(db, item_id) for raw material lots. If either file has a custom raw lot builder, replace that call with build_inventory_lot_no.

  • Step 4: Run inbound tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest \
  tests/test_customer_supplied_inbound_lot_no.py \
  tests/test_opening_inventory_import.py \
  tests/test_special_warehouse_adjustment.py \
  tests/test_inventory_lot_numbering.py \
  -q

Expected: PASS.

Task 4: Enforce Single-Lot Production Outbound And Merge Daily Work Orders

Files:

  • Modify: backend/app/services/operations.py

  • Modify: backend/app/api/routes/production.py

  • Modify: backend/tests/test_selected_stock_lot_production_issue.py

  • Step 1: Update production issue tests for new rules

In backend/tests/test_selected_stock_lot_production_issue.py, change lot numbers in seed data to YL0001 and YL0002.

Replace old yearly sequence tests with:

def test_create_work_order_no_uses_date_lot_and_product_name(self) -> None:
    class FrozenDateTime(datetime):
        @classmethod
        def now(cls, tz=None):  # noqa: ANN001
            _ = tz
            return cls(2026, 6, 7, 9, 15, 0)

    with patch.object(production_routes, "datetime", FrozenDateTime):
        result = create_work_order(
            WorkOrderCreate(
                product_item_id=self.product.id,
                selected_stock_lots=[
                    SelectedStockLotIssueCreate(
                        source_lot_id=self.lot1.id,
                        material_item_id=self.material.id,
                        issued_weight_kg=Decimal("10"),
                    )
                ],
            ),
            context=SimpleNamespace(user=SimpleNamespace(id=1)),
            db=self.db,
        )

    self.assertEqual(result.work_order_no, f"20260607-{self.lot1.lot_no}-{self.product.item_name}")

Add merge test:

def test_same_day_same_lot_same_product_merges_into_existing_work_order(self) -> None:
    class FrozenDateTime(datetime):
        @classmethod
        def now(cls, tz=None):  # noqa: ANN001
            _ = tz
            return cls(2026, 6, 7, 9, 15, 0)

    payload = WorkOrderCreate(
        product_item_id=self.product.id,
        selected_stock_lots=[
            SelectedStockLotIssueCreate(source_lot_id=self.lot1.id, material_item_id=self.material.id, issued_weight_kg=Decimal("10"))
        ],
    )
    with patch.object(production_routes, "datetime", FrozenDateTime):
        first = create_work_order(payload, context=SimpleNamespace(user=SimpleNamespace(id=1)), db=self.db)
        second = create_work_order(
            WorkOrderCreate(
                product_item_id=self.product.id,
                selected_stock_lots=[
                    SelectedStockLotIssueCreate(source_lot_id=self.lot1.id, material_item_id=self.material.id, issued_weight_kg=Decimal("8"))
                ],
            ),
            context=SimpleNamespace(user=SimpleNamespace(id=1)),
            db=self.db,
        )

    self.assertEqual(second.work_order_id, first.work_order_id)
    issues = self.db.scalars(select(WorkOrderMaterialIssue).where(WorkOrderMaterialIssue.work_order_id == first.work_order_id)).all()
    self.assertEqual(len(issues), 2)
    material_row = self.db.scalar(select(WorkOrderMaterial).where(WorkOrderMaterial.work_order_id == first.work_order_id))
    self.assertEqual(float(material_row.issued_weight_kg), 18)

Add rejection test:

def test_create_work_order_rejects_multiple_stock_lots(self) -> None:
    with self.assertRaises(HTTPException) as exc:
        create_work_order(
            WorkOrderCreate(
                product_item_id=self.product.id,
                selected_stock_lots=[
                    SelectedStockLotIssueCreate(source_lot_id=self.lot1.id, material_item_id=self.material.id, issued_weight_kg=Decimal("10")),
                    SelectedStockLotIssueCreate(source_lot_id=self.lot2.id, material_item_id=self.material.id, issued_weight_kg=Decimal("5")),
                ],
            ),
            context=SimpleNamespace(user=SimpleNamespace(id=1)),
            db=self.db,
        )

    self.assertEqual(exc.exception.status_code, 400)
    self.assertIn("一次生产出库只能选择一个材料库存批次号", exc.exception.detail)

Replace the old “requires lots for all BOM materials” expectation with a test that only the selected material is issued and other BOM rows remain unissued.

  • Step 2: Run focused production issue tests

Run:

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

Expected: FAIL on old multi-lot and old work-order number behavior.

  • Step 3: Add deterministic production work-order number helper

Modify backend/app/services/operations.py:

def build_raw_lot_product_work_order_no(*, biz_time: datetime, lot_no: str, product: Item) -> str:
    date_part = biz_time.strftime("%Y%m%d")
    product_name = " ".join(str(product.item_name or product.item_code or f"产品{product.id}").split()).strip("- ") or f"产品{product.id}"
    prefix = f"{date_part}-{lot_no}-"
    max_product_name_length = max(80 - len(prefix), 1)
    return f"{prefix}{product_name[:max_product_name_length]}"
  • Step 4: Refactor create_work_order to validate one selected lot

In backend/app/api/routes/production.py, after building selected_lots, add:

    if len(selected_lots) != 1:
        raise HTTPException(status_code=400, detail="一次生产出库只能选择一个材料库存批次号")

Remove the requirement that every BOM material must have a selected lot. Replace it with:

    selected_lot = selected_lot_by_id[int(selected_lots[0]["source_lot_id"])]
    selected_material_item_id = int(selected_lots[0]["material_item_id"] or selected_lot.item_id)
    selected_bom_item = next((item for item in bom_items if int(item.material_item_id) == selected_material_item_id), None)
    if not selected_bom_item:
        raise HTTPException(status_code=400, detail=f"库存批次 {selected_lot.lot_no} 不属于该产品需规默认用料")
  • Step 5: Implement same-day same-lot same-product merge

In backend/app/api/routes/production.py, before creating a new WorkOrder, compute:

    now = datetime.now()
    work_order_no = build_raw_lot_product_work_order_no(biz_time=now, lot_no=selected_lot.lot_no, product=product)
    existing_work_order = db.scalar(select(WorkOrder).where(WorkOrder.work_order_no == work_order_no).limit(1))
    if existing_work_order and str(existing_work_order.status or "").upper() == "SETTLED":
        raise HTTPException(status_code=400, detail="该库存批次今日对应产品的生产工单已结单，不能继续合并生产出库")

If existing_work_order is not settled, use it instead of adding a new one:

    if existing_work_order:
        work_order = existing_work_order
    else:
        work_order = WorkOrder(
            work_order_no=work_order_no,
            work_order_type="NORMAL",
            source_sales_order_item_id=payload.source_sales_order_item_id,
            product_item_id=payload.product_item_id,
            route_id=route.id,
            bom_id=bom.id,
            planned_qty=planned_qty,
            released_qty=planned_qty,
            finished_qty=to_decimal(0),
            scrap_qty=to_decimal(0),
            planned_start_time=payload.planned_start_time or now,
            planned_end_time=payload.planned_end_time,
            actual_start_time=None,
            actual_end_time=None,
            priority_level=payload.priority_level or 3,
            status="RELEASED",
            remark=payload.remark,
        )
        db.add(work_order)
        db.flush()
        create_work_order_materials_and_operations(
            db,
            work_order=work_order,
            bom=bom,
            route=route,
            auto_issue_materials=False,
            operator_user_id=context.user.id,
            issue_weight_override=None,
        )

When applying the selected lot issue, update the selected material row cumulatively:

    material_row.issued_qty = Decimal("0")
    material_row.issued_weight_kg = to_decimal(material_row.issued_weight_kg) + selected_lots[0]["issued_weight_kg"]
    material_row.required_weight_kg = material_row.issued_weight_kg
    material_row.status = "ISSUED"
    db.add(material_row)
    sync_work_order_reference_qty(db, work_order.id)

Ensure build_raw_lot_product_work_order_no is imported from app.services.operations.

  • Step 6: Run production issue tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_selected_stock_lot_production_issue.py tests/test_production_work_order_ledger.py -q

Expected: PASS.

Task 5: Match Miniapp Reports By Material Stock Lot Number

Files:

  • Modify: backend/app/api/routes/production.py

  • Modify: backend/tests/test_selected_stock_lot_production_issue.py

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_work_order_batch_options.py

  • Step 1: Update ERP miniapp resolver tests

Replace JH0001 tests with YL0001 tests:

def test_miniapp_material_lot_no_resolves_unsettled_work_order(self) -> None:
    work_order = self._add_issued_work_order(
        work_order_id=1,
        work_order_no=f"20260607-{self.lot1.lot_no}-{self.product.item_name}",
        product=self.product,
        issue_id=10,
        material_row_id=10,
    )
    self._add_work_order_operation(operation_id=101, work_order_id=work_order.id, operation_name="1序")
    self.db.commit()

    matched = _find_work_order_for_miniapp_item(
        self.db,
        {
            "raw_material_batch_no": self.lot1.lot_no,
            "product_name": self.product.item_name,
            "project_no": self.product.item_code,
        },
    )

    self.assertEqual(matched.id, work_order.id)

Add settled ignore test:

def test_miniapp_material_lot_no_ignores_settled_work_order(self) -> None:
    self._add_issued_work_order(
        work_order_id=1,
        work_order_no=f"20260607-{self.lot1.lot_no}-{self.product.item_name}",
        product=self.product,
        issue_id=10,
        material_row_id=10,
    ).status = "SETTLED"
    self.db.commit()

    matched = _find_work_order_for_miniapp_item(
        self.db,
        {
            "raw_material_batch_no": self.lot1.lot_no,
            "product_name": self.product.item_name,
            "project_no": self.product.item_code,
        },
    )

    self.assertIsNone(matched)

Keep one compatibility test for old JH0001 short code if existing historical data still needs syncing.

  • Step 2: Run ERP miniapp resolver tests

Run:

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

Expected: FAIL because resolver still prioritizes JH0001 short codes and does not filter settled work orders in lot-based candidate selection.

  • Step 3: Update ERP resolver

Modify _select_miniapp_issue_candidate in backend/app/api/routes/production.py to receive only unsettled work orders:

product_rows = db.execute(
    select(WorkOrder.id, Item.item_name, Item.item_code)
    .join(Item, Item.id == WorkOrder.product_item_id)
    .where(
        WorkOrder.id.in_(latest_by_work_order),
        func.upper(func.coalesce(WorkOrder.status, "")) != "SETTLED",
    )
).all()

Modify _resolve_work_order_issue_by_inventory_lot_no to join WorkOrder and ignore settled:

candidates = db.scalars(
    select(WorkOrderMaterialIssue)
    .join(StockLot, StockLot.id == WorkOrderMaterialIssue.source_lot_id)
    .join(WorkOrder, WorkOrder.id == WorkOrderMaterialIssue.work_order_id)
    .where(
        StockLot.lot_no == lot_no,
        func.upper(func.coalesce(WorkOrder.status, "")) != "SETTLED",
    )
    .order_by(WorkOrderMaterialIssue.id.desc())
).all()

Modify _find_work_order_for_miniapp_item to resolve stock lot numbers before JH0001 compatibility:

issue, has_issue_candidates = _resolve_work_order_issue_by_inventory_lot_no(db, batch_no, row)
work_order = db.get(WorkOrder, issue.work_order_id) if issue else None
if work_order:
    if cache is not None:
        cache["work_order"][cache_key] = work_order
    return work_order

if _is_miniapp_work_order_short_code(batch_no):
    work_order = _resolve_work_order_by_miniapp_short_code(db, batch_no, row)
    if cache is not None:
        cache["work_order"][cache_key] = work_order
    return work_order

Keep the direct work-order-number fallback only when has_issue_candidates is false.

  • Step 4: Update miniapp backend option query

Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py.

Replace _miniapp_work_order_short_code usage in _raw_material_batch_options with a query that returns source stock lot numbers:

rows = db.execute(
    text(
        """
        select distinct sl.lot_no
        from pp_work_order wo
        join pp_work_order_operation op on op.work_order_id = wo.id
        join md_item product on product.id = wo.product_item_id
        join pp_work_order_material_issue issue on issue.work_order_id = wo.id
        join wh_stock_lot sl on sl.id = issue.source_lot_id
        where sl.lot_no is not null
          and sl.lot_no <> ''
          and upper(coalesce(wo.status, '')) <> 'SETTLED'
          and (:product_name = '' or product.item_name = :product_name)
          and (:project_no = '' or product.item_code = :project_no)
          and (:process_name = '' or op.operation_name = :process_name)
        order by sl.lot_no asc
        """
    ),
    {
        "product_name": product_name,
        "project_no": project_no,
        "process_name": process_name,
    },
).all()
return [str(row[0]) for row in rows if row[0]]

If the miniapp backend connects to the ERP schema through a different SQLAlchemy engine name, use the existing _erp_engine() or session helper already used by the current _raw_material_batch_options function.

  • Step 5: Update miniapp backend tests

Modify /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_work_order_batch_options.py:

def test_raw_material_batch_options_returns_material_stock_lots_for_product_process(monkeypatch):
    # seed pp_work_order, pp_work_order_operation, md_item, pp_work_order_material_issue, wh_stock_lot
    options = reports._raw_material_batch_options(
        product_name="钢制碗",
        project_no="",
        process_name="1序",
    )
    assert options == ["YL0001", "YL0002"]

Also update the settled test to assert settled work orders do not return their lot number.

  • Step 6: Run ERP and miniapp backend tests

Run:

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

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
python -m pytest tests/test_work_order_batch_options.py -q

Expected: PASS.

Task 6: Add Production Work-Order Inbound Backend API

Files:

  • Create: backend/app/services/production_work_order_inbound.py

  • Modify: backend/app/schemas/operations.py

  • Modify: backend/app/api/routes/production.py

  • Test: backend/tests/test_production_work_order_inbound.py

  • Step 1: Write backend tests for preview, deviation, posting, and settle

Create backend/tests/test_production_work_order_inbound.py with the same SQLite setup pattern as test_production_work_order_ledger.py.

Core tests:

def test_preview_calculates_default_surplus_and_scrap_from_finished_qty(self) -> None:
    preview = preview_production_work_order_inbound(self.db, work_order_id=1, finished_qty=Decimal("198"))
    self.assertEqual(float(preview.theoretical_surplus_weight_kg), 81.2)
    self.assertEqual(float(preview.theoretical_scrap_weight_kg), 23.8)


def test_submit_requires_deviation_remark_when_surplus_diff_exceeds_15_percent(self) -> None:
    with self.assertRaises(HTTPException) as exc:
        create_production_work_order_inbound(
            self.db,
            payload=ProductionWorkOrderInboundCreate(
                work_order_id=1,
                finished_warehouse_id=2,
                scrap_warehouse_id=3,
                finished_qty=198,
                surplus_weight_kg=120,
                scrap_weight_kg=23.8,
                deviation_remark=None,
                settle_work_order=False,
            ),
            operator_user_id=1,
        )
    self.assertIn("偏差说明", exc.exception.detail)


def test_submit_posts_finished_surplus_and_scrap_in_one_transaction(self) -> None:
    result = create_production_work_order_inbound(
        self.db,
        payload=ProductionWorkOrderInboundCreate(
            work_order_id=1,
            finished_warehouse_id=2,
            scrap_warehouse_id=3,
            finished_qty=198,
            surplus_weight_kg=80,
            scrap_weight_kg=24,
            deviation_remark=None,
            settle_work_order=False,
        ),
        operator_user_id=1,
    )
    self.assertEqual(result.work_order_id, 1)
    self.assertEqual(float(self.db.get(WorkOrder, 1).finished_qty), 198)
    self.assertEqual(float(self.db.get(StockLot, 1).remaining_weight_kg), 380)
    txn_types = [row.txn_type for row in self.db.scalars(select(InventoryTxn).order_by(InventoryTxn.id)).all()]
    self.assertIn("FG_IN", txn_types)
    self.assertIn("PRODUCTION_SURPLUS_IN", txn_types)
    self.assertIn("PRODUCTION_SCRAP_IN", txn_types)


def test_submit_settles_work_order_only_when_requested(self) -> None:
    create_production_work_order_inbound(
        self.db,
        payload=ProductionWorkOrderInboundCreate(
            work_order_id=1,
            finished_warehouse_id=2,
            scrap_warehouse_id=3,
            finished_qty=198,
            surplus_weight_kg=80,
            scrap_weight_kg=24,
            settle_work_order=True,
            settle_confirmed=True,
            deviation_remark=None,
        ),
        operator_user_id=1,
    )
    self.assertEqual(self.db.get(WorkOrder, 1).status, "SETTLED")
  • Step 2: Run production inbound tests to verify failure

Run:

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

Expected: FAIL because schemas and service do not exist.

  • Step 3: Add schemas

Modify backend/app/schemas/operations.py:

class ProductionWorkOrderInboundPreviewRead(BaseModel):
    work_order_id: int
    work_order_no: str
    product_item_id: int
    product_name: str
    source_lot_id: int
    source_lot_no: str
    issued_weight_kg: float
    gross_weight_per_piece_kg: float
    net_weight_per_piece_kg: float
    reported_good_qty: float
    reported_scrap_qty: float
    finished_received_qty: float
    surplus_returned_weight_kg: float
    scrap_inbound_weight_kg: float
    theoretical_finished_qty: float
    theoretical_surplus_weight_kg: float
    theoretical_scrap_weight_kg: float


class ProductionWorkOrderInboundCreate(BaseModel):
    work_order_id: int = Field(gt=0)
    finished_warehouse_id: int = Field(gt=0)
    scrap_warehouse_id: int = Field(gt=0)
    finished_qty: float = 0
    surplus_weight_kg: float = 0
    scrap_weight_kg: float = 0
    deviation_remark: str | None = None
    remark: str | None = None
    settle_work_order: bool = False
    settle_confirmed: bool = False


class ProductionWorkOrderInboundRead(BaseModel):
    work_order_id: int
    work_order_no: str
    finished_qty: float
    surplus_weight_kg: float
    scrap_weight_kg: float
    deviation_remark: str | None = None
    settled: bool
    message: str
  • Step 4: Implement calculation helpers

Create backend/app/services/production_work_order_inbound.py:

from __future__ import annotations

from decimal import Decimal

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

from app.models.master_data import BomItem, Item, StockBalance, Warehouse
from app.models.operations import InventoryTxn, StockLot, WorkOrder, WorkOrderMaterial, WorkOrderMaterialIssue
from app.schemas.operations import ProductionWorkOrderInboundCreate, ProductionWorkOrderInboundPreviewRead, ProductionWorkOrderInboundRead
from app.services.operations import clamp_non_negative, create_inventory_txn, get_location_or_default, to_decimal, upsert_stock_balance
from app.services.production_work_order_ledger import get_work_order_limit_snapshot

DEVIATION_RATE = Decimal("0.15")


def _deviation_exceeds_limit(actual: Decimal, theoretical: Decimal) -> bool:
    if theoretical <= 0:
        return actual > 0
    return abs(actual - theoretical) / theoretical > DEVIATION_RATE


def _source_issue(db: Session, work_order_id: int) -> WorkOrderMaterialIssue:
    issue = db.scalar(
        select(WorkOrderMaterialIssue)
        .where(WorkOrderMaterialIssue.work_order_id == work_order_id)
        .order_by(WorkOrderMaterialIssue.id.asc())
        .limit(1)
    )
    if not issue:
        raise HTTPException(status_code=400, detail="该工单没有材料库存批次领用记录，不能进行生产工单入库")
    return issue


def preview_production_work_order_inbound(
    db: Session,
    *,
    work_order_id: int,
    finished_qty: Decimal | None = None,
) -> ProductionWorkOrderInboundPreviewRead:
    work_order = db.get(WorkOrder, work_order_id)
    if not work_order:
        raise HTTPException(status_code=404, detail="生产工单不存在")
    if str(work_order.status or "").upper() == "SETTLED":
        raise HTTPException(status_code=400, detail="该生产工单已结单，不能继续入库")
    product = db.get(Item, work_order.product_item_id)
    issue = _source_issue(db, work_order_id)
    source_lot = db.get(StockLot, issue.source_lot_id)
    snapshot = get_work_order_limit_snapshot(db, work_order_id)
    bom_item = db.scalar(
        select(BomItem)
        .join(WorkOrderMaterial, WorkOrderMaterial.bom_item_id == BomItem.id)
        .where(WorkOrderMaterial.work_order_id == work_order_id, WorkOrderMaterial.material_item_id == issue.material_item_id)
        .limit(1)
    )
    gross = to_decimal(bom_item.gross_weight_per_piece_kg if bom_item else 0)
    net = to_decimal(bom_item.net_weight_per_piece_kg if bom_item else 0)
    default_finished_qty = to_decimal(finished_qty if finished_qty is not None else snapshot.max_finished_inbound_qty)
    issued_weight = to_decimal(snapshot.issued_weight_kg)
    returned_weight = to_decimal(snapshot.returned_material_weight_kg)
    scrap_inbound_weight = to_decimal(snapshot.scrap_inbound_weight_kg)
    reported_scrap_qty = to_decimal(snapshot.total_operation_scrap_qty)
    theoretical_surplus = clamp_non_negative(issued_weight - returned_weight - default_finished_qty * gross)
    theoretical_scrap = clamp_non_negative(default_finished_qty * clamp_non_negative(gross - net) + reported_scrap_qty * gross - scrap_inbound_weight)
    return ProductionWorkOrderInboundPreviewRead(
        work_order_id=work_order.id,
        work_order_no=work_order.work_order_no,
        product_item_id=work_order.product_item_id,
        product_name=product.item_name if product else "",
        source_lot_id=source_lot.id if source_lot else issue.source_lot_id,
        source_lot_no=source_lot.lot_no if source_lot else issue.material_sub_batch_no,
        issued_weight_kg=float(issued_weight),
        gross_weight_per_piece_kg=float(gross),
        net_weight_per_piece_kg=float(net),
        reported_good_qty=float(snapshot.final_operation_good_qty),
        reported_scrap_qty=float(reported_scrap_qty),
        finished_received_qty=float(snapshot.finished_received_qty),
        surplus_returned_weight_kg=float(returned_weight),
        scrap_inbound_weight_kg=float(scrap_inbound_weight),
        theoretical_finished_qty=float(snapshot.max_finished_inbound_qty),
        theoretical_surplus_weight_kg=float(theoretical_surplus),
        theoretical_scrap_weight_kg=float(theoretical_scrap),
    )
  • Step 5: Implement transactional posting helpers

Continue in backend/app/services/production_work_order_inbound.py with helper functions that are called by the final submit function:

def _remark(base: str, payload: ProductionWorkOrderInboundCreate) -> str:
    parts = [base]
    if payload.deviation_remark:
        parts.append(f"入库偏差说明：{payload.deviation_remark}")
    if payload.remark:
        parts.append(payload.remark)
    return "；".join(part for part in parts if part)


def _source_lot(db: Session, preview: ProductionWorkOrderInboundPreviewRead) -> StockLot:
    source_lot = db.get(StockLot, preview.source_lot_id)
    if not source_lot:
        raise HTTPException(status_code=404, detail="来源材料库存批次号不存在")
    return source_lot


def _post_surplus_return(
    db: Session,
    *,
    work_order: WorkOrder,
    preview: ProductionWorkOrderInboundPreviewRead,
    payload: ProductionWorkOrderInboundCreate,
    surplus_weight: Decimal,
    operator_user_id: int | None,
) -> None:
    source_lot = _source_lot(db, preview)
    source_lot.remaining_weight_kg = to_decimal(source_lot.remaining_weight_kg) + surplus_weight
    if source_lot.status == "DEPLETED" and to_decimal(source_lot.remaining_weight_kg) > 0:
        source_lot.status = "AVAILABLE"
    db.add(source_lot)
    material_row = db.scalar(
        select(WorkOrderMaterial)
        .where(WorkOrderMaterial.work_order_id == work_order.id, WorkOrderMaterial.material_item_id == source_lot.item_id)
        .limit(1)
    )
    if material_row:
        material_row.returned_weight_kg = to_decimal(material_row.returned_weight_kg) + surplus_weight
        db.add(material_row)
    upsert_stock_balance(
        db,
        item_id=source_lot.item_id,
        warehouse_id=source_lot.warehouse_id,
        location_id=source_lot.location_id,
        qty_delta=Decimal("0"),
        weight_delta=surplus_weight,
        available_qty_delta=Decimal("0"),
        available_weight_delta=surplus_weight,
        unit_cost=to_decimal(source_lot.unit_cost, "0.0001"),
    )
    create_inventory_txn(
        db,
        txn_type="PRODUCTION_SURPLUS_IN",
        item_id=source_lot.item_id,
        warehouse_id=source_lot.warehouse_id,
        location_id=source_lot.location_id,
        lot_id=source_lot.id,
        qty_change=Decimal("0"),
        weight_change=surplus_weight,
        unit_cost=to_decimal(source_lot.unit_cost, "0.0001"),
        source_doc_type="WORK_ORDER",
        source_doc_id=work_order.id,
        source_line_id=source_lot.id,
        biz_time=datetime.now(),
        operator_user_id=operator_user_id,
        remark=_remark("生产工单入库：生产余料入库", payload),
        amount_basis="WEIGHT",
    )


def _post_finished_inbound(
    db: Session,
    *,
    work_order: WorkOrder,
    preview: ProductionWorkOrderInboundPreviewRead,
    payload: ProductionWorkOrderInboundCreate,
    finished_qty: Decimal,
    operator_user_id: int | None,
) -> None:
    product = db.get(Item, work_order.product_item_id)
    if not product:
        raise HTTPException(status_code=404, detail="工单产品不存在")
    warehouse = db.get(Warehouse, payload.finished_warehouse_id)
    if not warehouse or str(warehouse.warehouse_type or "").upper() != "FINISHED":
        raise HTTPException(status_code=400, detail="生产工单入库必须选择成品库")
    location = get_location_or_default(db, payload.finished_warehouse_id, None)
    source_lot = _source_lot(db, preview)
    gross_weight = to_decimal(preview.gross_weight_per_piece_kg)
    receipt_weight = finished_qty * gross_weight
    unit_cost = to_decimal(source_lot.unit_cost, "0.0001") * gross_weight
    lot_no = f"CP-{work_order.work_order_no}-{int(to_decimal(work_order.finished_qty) + finished_qty)}"
    lot = StockLot(
        lot_no=lot_no[:50],
        parent_lot_id=source_lot.id,
        lot_role="FINISHED_FROM_RAW_BATCH",
        material_sub_batch_no=None,
        item_id=product.id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        source_doc_type="WORK_ORDER_INBOUND",
        source_doc_id=work_order.id,
        source_line_id=None,
        source_material_lot_id=source_lot.id,
        source_material_sub_batch_no=source_lot.lot_no,
        source_material_summary=f"来源库存批次号：{source_lot.lot_no}",
        inbound_qty=finished_qty,
        inbound_weight_kg=receipt_weight,
        remaining_qty=finished_qty,
        remaining_weight_kg=receipt_weight,
        locked_qty=Decimal("0"),
        locked_weight_kg=Decimal("0"),
        unit_cost=unit_cost,
        quality_status="PASS",
        status="AVAILABLE",
        remark=_remark("生产工单入库：成品入库", payload),
    )
    db.add(lot)
    db.flush()
    upsert_stock_balance(
        db,
        item_id=product.id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        qty_delta=finished_qty,
        weight_delta=receipt_weight,
        available_qty_delta=finished_qty,
        available_weight_delta=receipt_weight,
        unit_cost=unit_cost,
    )
    create_inventory_txn(
        db,
        txn_type="FG_IN",
        item_id=product.id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        lot_id=lot.id,
        qty_change=finished_qty,
        weight_change=receipt_weight,
        unit_cost=unit_cost,
        source_doc_type="WORK_ORDER",
        source_doc_id=work_order.id,
        source_line_id=None,
        biz_time=datetime.now(),
        operator_user_id=operator_user_id,
        remark=_remark("生产工单入库：成品入库", payload),
        amount_basis="QTY",
    )
    work_order.finished_qty = to_decimal(work_order.finished_qty) + finished_qty
    db.add(work_order)


def _post_scrap_inbound(
    db: Session,
    *,
    work_order: WorkOrder,
    preview: ProductionWorkOrderInboundPreviewRead,
    payload: ProductionWorkOrderInboundCreate,
    scrap_weight: Decimal,
    operator_user_id: int | None,
) -> None:
    source_lot = _source_lot(db, preview)
    warehouse = db.get(Warehouse, payload.scrap_warehouse_id)
    if not warehouse or str(warehouse.warehouse_type or "").upper() != "SCRAP":
        raise HTTPException(status_code=400, detail="生产工单入库必须选择废料库")
    location = get_location_or_default(db, payload.scrap_warehouse_id, None)
    lot = StockLot(
        lot_no=f"FL-{work_order.work_order_no}"[:50],
        parent_lot_id=source_lot.id,
        lot_role="PRODUCTION_SCRAP",
        material_sub_batch_no=None,
        item_id=source_lot.item_id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        source_doc_type="WORK_ORDER",
        source_doc_id=work_order.id,
        source_line_id=None,
        source_material_lot_id=source_lot.id,
        source_material_sub_batch_no=source_lot.lot_no,
        source_material_summary=f"来源库存批次号：{source_lot.lot_no}",
        inbound_qty=Decimal("0"),
        inbound_weight_kg=scrap_weight,
        remaining_qty=Decimal("0"),
        remaining_weight_kg=scrap_weight,
        locked_qty=Decimal("0"),
        locked_weight_kg=Decimal("0"),
        unit_cost=to_decimal(source_lot.unit_cost, "0.0001"),
        quality_status="PASS",
        status="AVAILABLE",
        remark=_remark("生产工单入库：生产废料入库", payload),
    )
    db.add(lot)
    db.flush()
    upsert_stock_balance(
        db,
        item_id=source_lot.item_id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        qty_delta=Decimal("0"),
        weight_delta=scrap_weight,
        available_qty_delta=Decimal("0"),
        available_weight_delta=scrap_weight,
        unit_cost=to_decimal(source_lot.unit_cost, "0.0001"),
    )
    create_inventory_txn(
        db,
        txn_type="PRODUCTION_SCRAP_IN",
        item_id=source_lot.item_id,
        warehouse_id=warehouse.id,
        location_id=location.id if location else None,
        lot_id=lot.id,
        qty_change=Decimal("0"),
        weight_change=scrap_weight,
        unit_cost=to_decimal(source_lot.unit_cost, "0.0001"),
        source_doc_type="WORK_ORDER",
        source_doc_id=work_order.id,
        source_line_id=None,
        biz_time=datetime.now(),
        operator_user_id=operator_user_id,
        remark=_remark("生产工单入库：生产废料入库", payload),
        amount_basis="WEIGHT",
    )
  • Step 6: Implement submit function

Continue in backend/app/services/production_work_order_inbound.py:

def create_production_work_order_inbound(
    db: Session,
    *,
    payload: ProductionWorkOrderInboundCreate,
    operator_user_id: int | None,
) -> ProductionWorkOrderInboundRead:
    work_order = db.get(WorkOrder, payload.work_order_id)
    if not work_order:
        raise HTTPException(status_code=404, detail="生产工单不存在")
    if str(work_order.status or "").upper() == "SETTLED":
        raise HTTPException(status_code=400, detail="该生产工单已结单，不能继续入库")
    finished_qty = to_decimal(payload.finished_qty)
    surplus_weight = to_decimal(payload.surplus_weight_kg)
    scrap_weight = to_decimal(payload.scrap_weight_kg)
    if finished_qty <= 0 and surplus_weight <= 0 and scrap_weight <= 0:
        raise HTTPException(status_code=400, detail="本次至少填写一项入库数据")
    preview = preview_production_work_order_inbound(db, work_order_id=payload.work_order_id, finished_qty=finished_qty)
    if (
        _deviation_exceeds_limit(surplus_weight, to_decimal(preview.theoretical_surplus_weight_kg))
        or _deviation_exceeds_limit(scrap_weight, to_decimal(preview.theoretical_scrap_weight_kg))
    ) and not str(payload.deviation_remark or "").strip():
        raise HTTPException(status_code=400, detail="余料或废料入库重量与理论值偏差超过上下15%，请填写偏差说明")
    if payload.settle_work_order and not payload.settle_confirmed:
        raise HTTPException(status_code=400, detail="结单前必须确认本次入库明细及结单影响")

    if finished_qty > 0:
        _post_finished_inbound(db, work_order=work_order, preview=preview, payload=payload, finished_qty=finished_qty, operator_user_id=operator_user_id)
    if surplus_weight > 0:
        _post_surplus_return(db, work_order=work_order, preview=preview, payload=payload, surplus_weight=surplus_weight, operator_user_id=operator_user_id)
    if scrap_weight > 0:
        _post_scrap_inbound(db, work_order=work_order, preview=preview, payload=payload, scrap_weight=scrap_weight, operator_user_id=operator_user_id)

    if payload.settle_work_order:
        work_order.status = "SETTLED"
        work_order.actual_end_time = work_order.actual_end_time or datetime.now()
        work_order.updated_at = datetime.now()
        parts = [part.strip() for part in str(work_order.remark or "").replace(";", "；").split("；") if part.strip()]
        parts.append("生产工单入库已结单")
        if payload.deviation_remark:
            parts.append(f"入库偏差说明：{payload.deviation_remark}")
        work_order.remark = "；".join(dict.fromkeys(parts))[-255:]
        db.add(work_order)
    db.commit()
    return ProductionWorkOrderInboundRead(
        work_order_id=work_order.id,
        work_order_no=work_order.work_order_no,
        finished_qty=float(finished_qty),
        surplus_weight_kg=float(surplus_weight),
        scrap_weight_kg=float(scrap_weight),
        deviation_remark=payload.deviation_remark,
        settled=bool(payload.settle_work_order),
        message="生产工单入库已完成",
    )
  • Step 7: Add production routes

Modify backend/app/api/routes/production.py imports:

from app.schemas.operations import ProductionWorkOrderInboundCreate, ProductionWorkOrderInboundPreviewRead, ProductionWorkOrderInboundRead
from app.services.production_work_order_inbound import create_production_work_order_inbound, preview_production_work_order_inbound

Add routes:

@router.get("/work-order-inbounds/preview", response_model=ProductionWorkOrderInboundPreviewRead)
def get_work_order_inbound_preview(
    work_order_id: int = Query(gt=0),
    finished_qty: float | None = Query(default=None, ge=0),
    db: Session = Depends(get_db),
) -> ProductionWorkOrderInboundPreviewRead:
    return preview_production_work_order_inbound(
        db,
        work_order_id=work_order_id,
        finished_qty=to_decimal(finished_qty) if finished_qty is not None else None,
    )


@router.post("/work-order-inbounds", response_model=ProductionWorkOrderInboundRead)
def post_work_order_inbound(
    payload: ProductionWorkOrderInboundCreate,
    context: AuthContext = Depends(require_authenticated_user),
    db: Session = Depends(get_db),
) -> ProductionWorkOrderInboundRead:
    return create_production_work_order_inbound(db, payload=payload, operator_user_id=context.user.id)
  • Step 8: Run backend production inbound tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m pytest tests/test_production_work_order_inbound.py tests/test_production_work_order_ledger.py -q

Expected: PASS.

Task 7: Add ERP Frontend UI For Prefix, Single-Lot Production Out, And Work-Order Inbound

Files:

  • Modify: frontend/src/components/StockLotTagSelect.vue

  • Modify: frontend/src/views/SystemExtensionView.vue

  • Modify: frontend/src/views/InventoryLedgerView.vue

  • Create: frontend/scripts/test-production-work-order-inbound-ui.mjs

  • Step 1: Add frontend tests for single-select and production inbound text

Create frontend/scripts/test-production-work-order-inbound-ui.mjs:

import assert from "node:assert/strict";
import { readFileSync } from "node:fs";

const inventoryView = readFileSync(new URL("../src/views/InventoryLedgerView.vue", import.meta.url), "utf8");
const stockLotSelect = readFileSync(new URL("../src/components/StockLotTagSelect.vue", import.meta.url), "utf8");
const systemExtension = readFileSync(new URL("../src/views/SystemExtensionView.vue", import.meta.url), "utf8");

assert.match(stockLotSelect, /maxSelections/);
assert.match(inventoryView, /生产工单入库/);
assert.match(inventoryView, /一次生产出库只能选择一个材料库存批次号/);
assert.match(inventoryView, /work-order-inbounds\\/preview/);
assert.match(inventoryView, /work-order-inbounds/);
assert.match(inventoryView, /本次入库后结单/);
assert.match(systemExtension, /原材料库存批次号前缀/);
assert.match(systemExtension, /raw-material-lot-prefix/);

console.log("production work-order inbound UI checks passed");
  • Step 2: Run frontend UI test to verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-production-work-order-inbound-ui.mjs

Expected: FAIL before UI changes.

  • Step 3: Add maxSelections to StockLotTagSelect.vue

Modify props:

maxSelections: {
  type: Number,
  default: 0
}

Modify canSelect:

function canSelect(lot) {
  if (!lot || (!props.allowUnavailable && !lot.is_available_for_issue) || isSelected(lot)) {
    return false;
  }
  return !props.maxSelections || selectedLots.value.length < props.maxSelections;
}

Modify selectLot so single-select replaces existing selection:

const nextSelectedLots = props.maxSelections === 1 ? [] : selectedLots.value;
emit("update:modelValue", [
  ...nextSelectedLots,
  {
    ...lot,
    issued_weight_kg: normalizeSelectedIssuedWeight(lot)
  }
]);
  • Step 4: Add raw lot prefix config card to System Extension

Modify frontend/src/views/SystemExtensionView.vue:

Add reactive form:

const rawLotPrefixConfig = ref({});
const rawLotPrefixForm = reactive({ config_value: "YL", remark: "" });

Add loader:

async function loadRawLotPrefixConfig() {
  const result = await fetchResource("/system-extension/raw-material-lot-prefix", {});
  rawLotPrefixConfig.value = result;
  rawLotPrefixForm.config_value = result.config_value || "YL";
  rawLotPrefixForm.remark = result.remark || "";
}

Add saver:

async function saveRawLotPrefixConfig() {
  resetFeedback();
  try {
    const result = await putResource("/system-extension/raw-material-lot-prefix", rawLotPrefixForm);
    rawLotPrefixConfig.value = result;
    rawLotPrefixForm.config_value = result.config_value || "YL";
    rawLotPrefixForm.remark = result.remark || "";
    feedbackMessage.value = "原材料库存批次号前缀已保存";
  } catch (error) {
    errorMessage.value = error.message || "原材料库存批次号前缀保存失败";
  }
}

Call loadRawLotPrefixConfig() in loadAll().

Add panel:

<section class="panel">
  <div class="panel-header">
    <div>
      <p class="eyebrow">仓库编号配置</p>
      <h3>原材料库存批次号前缀</h3>
    </div>
    <span class="panel-tag">{{ rawLotPrefixConfig.config_value || "YL" }}</span>
  </div>
  <form class="stack-form" @submit.prevent="saveRawLotPrefixConfig">
    <label class="form-field">
      <span>前缀</span>
      <input v-model.trim="rawLotPrefixForm.config_value" type="text" maxlength="20" placeholder="默认 YL" />
    </label>
    <label class="form-field">
      <span>说明</span>
      <input v-model.trim="rawLotPrefixForm.remark" type="text" placeholder="例如：原材料库存批次号默认前缀" />
    </label>
    <button class="primary-button" type="submit">保存前缀</button>
  </form>
</section>
  • Step 5: Change production outbound UI to single-select

In InventoryLedgerView.vue, update production outbound StockLotTagSelect:

<StockLotTagSelect
  v-model="selectedProductionStockLots"
  :options="productionStockLotOptions"
  :max-selections="1"
  placeholder="搜索并选择一个材料库存批次号"
/>

Update validation:

function validateSelectedProductionStockLots(lots) {
  if (!lots.length) {
    return "请选择一个材料库存批次号";
  }
  if (lots.length > 1) {
    return "一次生产出库只能选择一个材料库存批次号";
  }
  const lot = lots[0];
  const issuedWeight = Number(lot.issued_weight_kg);
  const remainingWeight = Number(lot.remaining_weight_kg || 0);
  const lotNo = lot.lot_no || "未命名批次";
  if (!Number.isFinite(issuedWeight) || issuedWeight <= 0) {
    return `材料库存批次号 ${lotNo} 的本次出库重量必须大于0`;
  }
  if (issuedWeight > remainingWeight) {
    return `材料库存批次号 ${lotNo} 的本次出库重量不能超过剩余重量 ${formatWeight(remainingWeight)}`;
  }
  return "";
}
  • Step 6: Add “生产工单入库” button next to warehouse ledger button

Find the warehouse action area near the current “流水” button in InventoryLedgerView.vue and add:

<button class="ghost-button ledger-action-button" type="button" @click="openProductionWorkOrderInboundDrawer">
  生产工单入库
</button>

Add state:

const productionWorkOrderInboundDrawerOpen = ref(false);
const selectedInboundWorkOrderId = ref("");
const productionWorkOrderInboundPreview = ref(null);
const productionWorkOrderInboundForm = reactive({
  finished_qty: 0,
  surplus_weight_kg: 0,
  scrap_weight_kg: 0,
  deviation_remark: "",
  remark: "",
  settle_work_order: false
});

Add functions:

function openProductionWorkOrderInboundDrawer() {
  resetFeedback();
  selectedInboundWorkOrderId.value = "";
  productionWorkOrderInboundPreview.value = null;
  Object.assign(productionWorkOrderInboundForm, {
    finished_qty: 0,
    surplus_weight_kg: 0,
    scrap_weight_kg: 0,
    deviation_remark: "",
    remark: "",
    settle_work_order: false
  });
  productionWorkOrderInboundDrawerOpen.value = true;
}

async function loadProductionWorkOrderInboundPreview() {
  if (!selectedInboundWorkOrderId.value) {
    productionWorkOrderInboundPreview.value = null;
    return;
  }
  const query = new URLSearchParams({ work_order_id: String(selectedInboundWorkOrderId.value) });
  if (Number(productionWorkOrderInboundForm.finished_qty) > 0) {
    query.set("finished_qty", String(productionWorkOrderInboundForm.finished_qty));
  }
  const result = await fetchResource(`/production/work-order-inbounds/preview?${query.toString()}`, {});
  productionWorkOrderInboundPreview.value = result;
  if (!Number(productionWorkOrderInboundForm.finished_qty)) {
    productionWorkOrderInboundForm.finished_qty = Number(result.theoretical_finished_qty || 0);
  }
  productionWorkOrderInboundForm.surplus_weight_kg = Number(result.theoretical_surplus_weight_kg || 0);
  productionWorkOrderInboundForm.scrap_weight_kg = Number(result.theoretical_scrap_weight_kg || 0);
}

Add submit:

async function submitProductionWorkOrderInbound() {
  resetFeedback();
  if (!selectedInboundWorkOrderId.value) {
    errorMessage.value = "请选择生产工单";
    return;
  }
  if (productionWorkOrderInboundForm.settle_work_order) {
    const confirmed = window.confirm(
      `请确认本次生产工单入库\n\n工单：${productionWorkOrderInboundPreview.value?.work_order_no || ""}\n成品入库：${productionWorkOrderInboundForm.finished_qty || 0} 件\n余料入库：${productionWorkOrderInboundForm.surplus_weight_kg || 0} kg\n废料入库：${productionWorkOrderInboundForm.scrap_weight_kg || 0} kg\n\n勾选结单后，该工单将变为已结单，小程序不再显示该工单对应的材料库存批次号，后续不能再对该工单做生产工单入库。`
    );
    if (!confirmed) {
      return;
    }
  }
  const result = await postResource("/production/work-order-inbounds", {
    work_order_id: Number(selectedInboundWorkOrderId.value),
    finished_warehouse_id: finishedWarehouseId.value,
    scrap_warehouse_id: scrapWarehouseId.value,
    finished_qty: Number(productionWorkOrderInboundForm.finished_qty || 0),
    surplus_weight_kg: Number(productionWorkOrderInboundForm.surplus_weight_kg || 0),
    scrap_weight_kg: Number(productionWorkOrderInboundForm.scrap_weight_kg || 0),
    deviation_remark: productionWorkOrderInboundForm.deviation_remark || null,
    remark: productionWorkOrderInboundForm.remark || null,
    settle_work_order: Boolean(productionWorkOrderInboundForm.settle_work_order),
    settle_confirmed: Boolean(productionWorkOrderInboundForm.settle_work_order)
  });
  feedbackMessage.value = result.message || "生产工单入库已完成";
  productionWorkOrderInboundDrawerOpen.value = false;
  await loadAll();
}
  • Step 7: Add production work-order inbound drawer

Add a FormDrawer in InventoryLedgerView.vue:

<FormDrawer
  :open="productionWorkOrderInboundDrawerOpen"
  eyebrow="嘉恒仓库 · 生产工单入库"
  title="生产工单入库"
  tag="工单闭环"
  wide
  @close="productionWorkOrderInboundDrawerOpen = false"
>
  <form class="stack-form" @submit.prevent="submitProductionWorkOrderInbound">
    <label class="form-field">
      <span>生产工单</span>
      <select v-model.number="selectedInboundWorkOrderId" required @change="loadProductionWorkOrderInboundPreview">
        <option disabled value="">请选择未结单生产工单</option>
        <option v-for="workOrder in openProductionWorkOrders" :key="workOrder.work_order_id" :value="workOrder.work_order_id">
          {{ workOrder.work_order_no }} · {{ workOrder.product_name }} · 领料 {{ formatWeight(workOrder.issued_weight_kg) }}
        </option>
      </select>
    </label>
    <div v-if="productionWorkOrderInboundPreview" class="summary-grid compact-summary-grid">
      <article class="summary-card"><span>材料库存批次号</span><strong>{{ productionWorkOrderInboundPreview.source_lot_no }}</strong></article>
      <article class="summary-card"><span>领料重量</span><strong>{{ formatWeight(productionWorkOrderInboundPreview.issued_weight_kg) }}</strong></article>
      <article class="summary-card"><span>单件毛重</span><strong>{{ formatWeight(productionWorkOrderInboundPreview.gross_weight_per_piece_kg) }}</strong></article>
      <article class="summary-card"><span>单件净重</span><strong>{{ formatWeight(productionWorkOrderInboundPreview.net_weight_per_piece_kg) }}</strong></article>
    </div>
    <div class="triple-field">
      <label class="form-field">
        <span>成品入库数量</span>
        <input v-model.number="productionWorkOrderInboundForm.finished_qty" type="number" min="0" step="1" @change="loadProductionWorkOrderInboundPreview" />
      </label>
      <label class="form-field">
        <span>余料入库重量(kg)</span>
        <input v-model.number="productionWorkOrderInboundForm.surplus_weight_kg" type="number" min="0" step="0.001" />
      </label>
      <label class="form-field">
        <span>废料重量(kg)</span>
        <input v-model.number="productionWorkOrderInboundForm.scrap_weight_kg" type="number" min="0" step="0.001" />
      </label>
    </div>
    <label class="form-field">
      <span>偏差说明</span>
      <textarea v-model.trim="productionWorkOrderInboundForm.deviation_remark" rows="3" placeholder="余料或废料与理论值偏差超过上下15%时必须填写"></textarea>
    </label>
    <label class="form-field checkbox-field">
      <input v-model="productionWorkOrderInboundForm.settle_work_order" type="checkbox" />
      <span>本次入库后结单</span>
    </label>
    <label class="form-field">
      <span>备注</span>
      <textarea v-model.trim="productionWorkOrderInboundForm.remark" rows="3"></textarea>
    </label>
    <button class="primary-button" type="submit" :disabled="submittingOperation">保存生产工单入库</button>
  </form>
</FormDrawer>
  • Step 8: Run frontend tests and build

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-production-work-order-inbound-ui.mjs
node scripts/test-inventory-ledger-options.mjs
npm run build

Expected: scripts PASS and build PASS with only existing Vite chunk-size warning.

Task 8: Update Miniapp Frontend Wording To “材料库存批次号”

Files:

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm/reportForm.wxml

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm/reportForm.js

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records/records.wxml

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard/dashboard.wxml

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review/review.wxml

  • Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js

  • Step 1: Add a text scan test

Run before edits:

rg -n "工单号|原材料分批次号|请选择工单号|暂无可选工单号" \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js

Expected: output includes old wording.

  • Step 2: Replace visible wording

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm/reportForm.wxml:

<text class="label">材料库存批次号</text>
<view class="picker {{reportItem.rawMaterialBatchError ? 'field-error' : ''}}">{{reportItem.rawMaterialBatchNo || '请选择材料库存批次号'}}</view>
<view wx:else class="readonly">暂无可选材料库存批次号</view>

In /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm/reportForm.js, replace validation message:

title: `${blockDisplay(block)}请选择材料库存批次号`,

In records/dashboard/review WXML files, replace visible 原材料分批次号 with 材料库存批次号.

In /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js, keep payload field names unchanged but replace correction display label:

correctionPair(corrections, 'raw_material_batch_no', '材料库存批次号', row.raw_material_batch_no || '')
  • Step 3: Verify old wording is gone from user-facing miniapp files

Run:

rg -n "请选择工单号|暂无可选工单号|原材料分批次号" \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/reportForm \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/records \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/dashboard \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/pages/review \
  /Users/souplearn/Gitlab/app/JhHardwareWRS/utils/api.js

Expected: no output.

Task 9: Add SQL Execution And Deployment Checks

Files:

  • Use: backend/sql/add_raw_lot_work_order_inbound_patch.sql

  • No code files modified in this task.

  • Step 1: Execute SQL patch against the target test database

Run from ERP backend:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python - <<'PY'
from pathlib import Path
import re
import pymysql

env = {}
for line in Path(".env").read_text(encoding="utf-8").splitlines():
    line = line.strip()
    if not line or line.startswith("#") or "=" not in line:
        continue
    key, value = line.split("=", 1)
    env[key.strip()] = value.strip().strip('"').strip("'")

sql_path = Path("sql/add_raw_lot_work_order_inbound_patch.sql")
statements = [stmt.strip() for stmt in re.split(r";\s*(?:\n|$)", sql_path.read_text(encoding="utf-8")) if stmt.strip()]
conn = pymysql.connect(
    host=env["MYSQL_HOST"],
    port=int(env["MYSQL_PORT"]),
    user=env["MYSQL_USER"],
    password=env["MYSQL_PASSWORD"],
    database=env["MYSQL_DATABASE"],
    charset="utf8mb4",
)
try:
    with conn.cursor() as cur:
        for stmt in statements:
            cur.execute(stmt)
    conn.commit()
finally:
    conn.close()
print(f"executed {len(statements)} statements")
PY

Expected: executed 2 statements.

  • Step 2: Verify config table and seed row

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python - <<'PY'
from pathlib import Path
import pymysql

env = {}
for line in Path(".env").read_text(encoding="utf-8").splitlines():
    line = line.strip()
    if not line or line.startswith("#") or "=" not in line:
        continue
    key, value = line.split("=", 1)
    env[key.strip()] = value.strip().strip('"').strip("'")

conn = pymysql.connect(
    host=env["MYSQL_HOST"],
    port=int(env["MYSQL_PORT"]),
    user=env["MYSQL_USER"],
    password=env["MYSQL_PASSWORD"],
    database=env["MYSQL_DATABASE"],
    charset="utf8mb4",
)
try:
    with conn.cursor() as cur:
        cur.execute("select config_code, config_value from sys_system_config where config_code='RAW_MATERIAL_LOT_PREFIX'")
        print(cur.fetchall())
finally:
    conn.close()
PY

Expected: output contains RAW_MATERIAL_LOT_PREFIX and YL.

Task 10: Full Verification

Files:

  • No new files.

  • Step 1: Run backend full tests

Run:

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

Expected: all tests pass.

  • Step 2: Run frontend checks

Run:

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

Expected: script tests pass and build succeeds. Vite chunk-size warning is acceptable.

  • Step 3: Run miniapp backend tests

Run:

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
python -m pytest tests/test_work_order_batch_options.py tests/test_metrics.py tests/test_review_corrections.py -q

Expected: tests pass.

  • Step 4: Manual smoke test in browser

Start services:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
.venv/bin/python -m uvicorn app.main:app --host 0.0.0.0 --port 8000

In another terminal:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run dev -- --host 0.0.0.0 --port 5173

Smoke flow:

  1. Open http://127.0.0.1:5173.
  2. Go to 系统拓展 and verify 原材料库存批次号前缀 shows YL.
  3. Go to 嘉恒仓库原材料库 and create a raw inbound test lot; verify lot number is YL0001 or the next available YL sequence.
  4. Use 原材料库 · 生产出库, select one material stock lot and one product, submit.
  5. Verify 工单台账 has work order YYYYMMDD-YLxxxx-产品名称.
  6. Submit another production outbound for the same day, same lot, same product.
  7. Verify it merges into the same work order and adds another issue line.
  8. Open 嘉恒仓库 · 生产工单入库, select that work order, enter finished qty, verify surplus/scrap defaults are filled.
  9. Change surplus or scrap beyond 15%; verify deviation remark is required.
  10. Check “本次入库后结单”, submit, confirm second prompt, verify work order status becomes 已结单.

Expected: no red overlay, no uncaught console error, and warehouse流水 shows成品入库、生产余料入库、生产废料入库 records tied to the work order.

Self-Review

Spec coverage:

  • YL0001 prefix and YL10000 expansion: Task 1 and Task 2.
  • System拓展 prefix configuration: Task 1 and Task 7.
  • Single-lot production outbound: Task 4 and Task 7.
  • Same-day same-lot same-product merge: Task 4.
  • Miniapp wording and lot-number selection: Task 5 and Task 8.
  • Production work-order inbound entry and workflow: Task 6 and Task 7.
  • 15% deviation remark: Task 6 and Task 7.
  • No default settle and second confirmation: Task 6 and Task 7.
  • SQL execution and verification: Task 9.
  • Full test and smoke checks: Task 10.

Placeholder scan:

  • No unfinished placeholder markers remain in the plan.
  • Production work-order inbound now lists concrete helper functions and submit calls instead of leaving a comment-only implementation gap.

Type consistency:

  • Backend schemas use ProductionWorkOrderInboundPreviewRead, ProductionWorkOrderInboundCreate, and ProductionWorkOrderInboundRead.
  • Frontend routes use /production/work-order-inbounds/preview and /production/work-order-inbounds.
  • Miniapp keeps the persisted field raw_material_batch_no but changes visible wording to “材料库存批次号”.

Git note:

  • Current ERP directory has no .git folder. If implementation runs in this mounted directory, commit steps are replaced by checkpoint summaries after each task. If a git worktree is available at execution time, commit after each completed task with a focused message.