tangxh/ForgeFlow-ERP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
c424633030
ForgeFlow-ERP/docs/superpowers/plans/2026-05-31-inventory-lot-unified-trace.md
汤学会 c927886888 Initial commit
2026-06-12 16:00:56 +08:00

2304 lines
75 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

# Inventory Lot Unified Trace 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:** Replace the separate “材料分批次号” concept with a single canonical “库存批次号” trace identity across raw material purchase receipt, production issue, miniapp report sync, finished goods receipt, delivery trace, DAG, and purchase-order-to-sales-order linkage.
**Architecture:** `wh_stock_lot.lot_no` becomes the canonical raw material trace key. Existing downstream records keep foreign-key links such as `source_lot_id` and store display snapshots only for compatibility; new UI and APIs expose “库存批次号” instead of “材料分批次号”. Purchase orders support multiple linked sales orders via a normalized join table, while purchase items still carry demand-level links for cost and MRP detail.
**Tech Stack:** FastAPI, SQLAlchemy ORM, Pydantic schemas, MySQL/PyMySQL migrations, SQLite-backed unit tests, Vue 3, Vite.
---
## Scope And Sequencing
This is a cross-module business refactor. Execute in this order:
1. Add compatibility schema and tests.
2. Introduce a single inventory-lot-number service.
3. Change purchase receipt lot generation and remove forced same-PO-line merge.
4. Change production issue to select stock lots instead of generating material sub-batches.
5. Rename and remap miniapp report sync from “原材料分批次号” to “库存批次号”.
6. Update finished goods receipt, delivery trace, sales order trace, and DAG labels/data.
7. Add multi-sales-order purchase order links.
8. Update frontend forms and labels.
9. Run full verification.
Do not remove legacy database columns in the first pass. Keep them synchronized for compatibility:
- `pp_work_order_material_issue.material_sub_batch_no`: write the selected `StockLot.lot_no` into this field during the transition.
- `wh_stock_lot.material_sub_batch_no`: keep nullable and unused for new raw material lots.
- `wh_stock_lot.source_material_sub_batch_no`: keep nullable, but new finished/semi-finished lots should store source lot summaries and source lot ids instead of relying on this field.
The user confirmed that the miniapp field previously named “原材料分批次号” should be treated as “库存批次号” going forward.
---
## File Structure
### Backend Files
- Modify: `backend/app/models/operations.py`
- Add purchase-order-to-sales-order association model.
- Keep legacy material-sub-batch fields for compatibility.
- Modify: `backend/app/schemas/operations.py`
- Add `sales_order_ids` to purchase order create/update payloads.
- Add selected stock lot issue payload types.
- Add lot availability fields for frontend multi-select.
- Rename user-facing schema aliases/comments where safe.
- Modify: `backend/app/services/operations.py`
- Replace `build_material_sub_batch_no` usage with `build_inventory_lot_no`.
- Add raw stock lot availability query sorted by usable-first order.
- Change material issue service to consume selected lots.
- Update purchase-order query source summaries for multiple sales orders.
- Modify: `backend/app/api/routes/purchase.py`
- Use `JH_{material_code}_{0001}` lot number generation for raw purchase receipts.
- Stop forcing subsequent arrivals on the same purchase line to merge.
- Preserve explicit user-selected merge behavior.
- Support multiple linked sales orders on create/update/read.
- Modify: `backend/app/api/routes/production.py`
- Change work-order creation to accept selected stock lot issues.
- Stop assigning work order number from generated material sub-batch number.
- Resolve miniapp report sync by库存批次号.
- Update completion receipt source summaries.
- Modify: `backend/app/api/routes/inventory.py`
- Update raw warehouse production-out endpoint wording and payload mapping.
- Ensure stock lot list exposes purchase order, receipt, unit cost, available/depleted state.
- Modify: `backend/app/api/routes/dashboard.py`
- Rename DAG nodes and details from “材料分批次号” to “库存批次号”.
- Link production/report/finished/delivery nodes through source stock lot ids.
- Modify: `backend/app/api/routes/sales.py`
- Ensure delivery trace reports source inventory lot numbers and no longer depends on material-sub-batch labels.
- Create: `backend/sql/unify_inventory_lot_trace_patch.sql`
- Add normalized purchase-order sales-order link table.
- Add indexes for lot-number sequence and selected lot lookup.
- Backfill compatibility values safely.
### Backend Tests
- Create: `backend/tests/test_inventory_lot_numbering.py`
- Create: `backend/tests/test_selected_stock_lot_production_issue.py`
- Create: `backend/tests/test_purchase_order_multi_sales_links.py`
- Modify: `backend/tests/test_raw_material_weight_only_flow.py`
- Modify: `backend/tests/test_sales_order_delivery_trace.py`
- Modify: `backend/tests/test_sales_order_purchase_lock_flow.py`
### Frontend Files
- Modify: `frontend/src/components/GlobalSearchableSelect.vue`
- Confirm it supports searchable multi-select tag mode, or add this mode without changing existing single-select behavior.
- Create: `frontend/src/components/StockLotTagSelect.vue`
- Focused searchable multi-select for inventory lots.
- Modify: `frontend/src/views/PurchaseOrderView.vue`
- Replace single sales order select with multi-select tags for sales-order purchase.
- Display multiple linked sales orders in source cards.
- Modify: `frontend/src/views/PurchaseReceiptView.vue`
- Use new default lot number from backend response or frontend helper.
- Remove forced same-purchase-line merge messaging.
- Modify: `frontend/src/views/InventoryLedgerView.vue`
- Rename production out operation from “生成材料分批次号” to selecting/using库存批次号.
- Show raw stock lot detail as `原材料物料 -> 库存批次号`.
- Modify: `frontend/src/views/WorkOrderManagementView.vue`
- Rename list/form labels away from材料分批次号.
- Add stock lot tag selector and per-lot issue weight input.
- Modify: `frontend/src/views/OperationReportView.vue`
- Rename field display to库存批次号 while accepting miniapp’s existing field data.
- Modify: `frontend/src/views/CompletionReceiptView.vue`
- Replace source material sub-batch labels with source inventory lot labels.
- Modify: `frontend/src/views/DeliveryManagementView.vue`
- Modify: `frontend/src/views/SalesPlanningView.vue`
- Modify: `frontend/src/views/DashboardView.vue`
- Replace material-sub-batch labels and search fields with source inventory lot wording.
- Modify: `frontend/src/utils/dictionaries.js`
- Modify: `frontend/src/utils/statusBadges.js`
- Add any missing display label mappings caused by the rename.
---
### Task 1: Add Regression Tests For The New Canonical Batch Model
**Files:**
- Create: `backend/tests/test_inventory_lot_numbering.py`
- Create: `backend/tests/test_selected_stock_lot_production_issue.py`
- Create: `backend/tests/test_purchase_order_multi_sales_links.py`
- Modify: `backend/tests/test_raw_material_weight_only_flow.py`
- [ ] **Step 1: Write failing lot-numbering tests**
Create `backend/tests/test_inventory_lot_numbering.py` with this content:
```python
from __future__ import annotations
import unittest
from decimal import Decimal
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.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.master_data import Item, Warehouse # noqa: E402
from app.models.operations import StockLot # noqa: E402
from app.services.operations import build_inventory_lot_no # noqa: E402
class InventoryLotNumberingTest(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.material = Item(
id=1,
item_code="110501000185",
item_name="左右后纵梁后段毛坯",
item_type="RAW_MATERIAL",
unit_weight_kg=Decimal("1"),
status="ACTIVE",
)
self.warehouse = Warehouse(
id=1,
warehouse_code="WH-RAW",
warehouse_name="原材料库",
warehouse_type="RAW",
status="ACTIVE",
)
self.db.add_all([self.material, self.warehouse])
self.db.commit()
def tearDown(self) -> None:
self.db.close()
def test_inventory_lot_no_starts_from_one_per_material(self) -> None:
self.assertEqual(build_inventory_lot_no(self.db, self.material.id), "JH_110501000185_0001")
def test_inventory_lot_no_increments_from_existing_stock_lots(self) -> None:
self.db.add(
StockLot(
lot_no="JH_110501000185_0001",
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.add(
StockLot(
lot_no="JH_110501000185_0003",
lot_role="INBOUND_RAW",
item_id=self.material.id,
warehouse_id=self.warehouse.id,
source_doc_type="TEST",
source_doc_id=2,
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), "JH_110501000185_0004")
```
- [ ] **Step 2: Write failing production selected-lot tests**
Create `backend/tests/test_selected_stock_lot_production_issue.py` with this content:
```python
from __future__ import annotations
import unittest
from decimal import Decimal
from fastapi import HTTPException
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.master_data import Bom, BomItem, Item, StockBalance, Warehouse # noqa: E402
from app.models.operations import ProcessRoute, ProcessRouteOperation, StockLot, WorkCenter, WorkOrderMaterialIssue # noqa: E402
from app.services.operations import issue_selected_stock_lots # noqa: E402
class SelectedStockLotProductionIssueTest(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.material = Item(id=1, item_code="RM-SEL-001", item_name="选择批次原料", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE")
self.product = Item(id=2, item_code="FG-SEL-001", item_name="选择批次成品", item_type="FINISHED_GOOD", unit_weight_kg=Decimal("2"), status="ACTIVE")
self.raw_wh = Warehouse(id=1, warehouse_code="WH-RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE")
self.db.add_all([self.material, self.product, self.raw_wh])
self.db.flush()
self.bom = Bom(id=1, bom_code="BOM-SEL-001", product_item_id=self.product.id, version_no="V1.0", is_default=1, status="ACTIVE")
self.bom_item = BomItem(id=1, bom_id=1, seq_no=1, material_item_id=self.material.id, usage_type="WEIGHT", gross_weight_per_piece_kg=Decimal("1.5"), loss_rate=Decimal("0"))
self.work_center = WorkCenter(id=1, center_code="WC-001", center_name="冲压", dept_id=1, center_type="INTERNAL", capacity_hours_per_day=Decimal("8"), status="ACTIVE")
self.route = ProcessRoute(id=1, route_code="RT-SEL-001", product_item_id=self.product.id, version_no="V1.0", is_default=1, status="ACTIVE")
self.route_op = ProcessRouteOperation(id=1, route_id=1, seq_no=1, operation_name="1序", work_center_id=1, standard_seconds_per_piece=Decimal("10"), status="ACTIVE")
self.db.add_all([self.bom, self.bom_item, self.work_center, self.route, self.route_op])
self.lot1 = StockLot(
id=1,
lot_no="JH_RM-SEL-001_0001",
lot_role="INBOUND_RAW",
item_id=self.material.id,
warehouse_id=self.raw_wh.id,
source_doc_type="PURCHASE_RECEIPT",
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("4"),
quality_status="PASS",
status="AVAILABLE",
)
self.lot2 = StockLot(
id=2,
lot_no="JH_RM-SEL-001_0002",
lot_role="INBOUND_RAW",
item_id=self.material.id,
warehouse_id=self.raw_wh.id,
source_doc_type="PURCHASE_RECEIPT",
source_doc_id=2,
inbound_qty=Decimal("0"),
inbound_weight_kg=Decimal("50"),
remaining_qty=Decimal("0"),
remaining_weight_kg=Decimal("50"),
locked_qty=Decimal("0"),
locked_weight_kg=Decimal("0"),
unit_cost=Decimal("5"),
quality_status="PASS",
status="AVAILABLE",
)
self.balance = StockBalance(
id=1,
item_id=self.material.id,
warehouse_id=self.raw_wh.id,
location_id=None,
qty_on_hand=Decimal("0"),
weight_on_hand_kg=Decimal("150"),
qty_available=Decimal("0"),
weight_available_kg=Decimal("150"),
qty_allocated=Decimal("0"),
weight_allocated_kg=Decimal("0"),
avg_unit_cost=Decimal("4.3333"),
)
self.db.add_all([self.lot1, self.lot2, self.balance])
self.db.commit()
def tearDown(self) -> None:
self.db.close()
def test_issue_selected_stock_lots_uses_lot_no_as_trace_key(self) -> None:
issues = issue_selected_stock_lots(
self.db,
work_order_id=1,
work_order_material_id=1,
material_item_id=self.material.id,
selected_lots=[
{"source_lot_id": self.lot1.id, "issued_weight_kg": Decimal("80")},
{"source_lot_id": self.lot2.id, "issued_weight_kg": Decimal("30")},
],
operator_user_id=1,
)
self.assertEqual([issue.material_sub_batch_no for issue in issues], ["JH_RM-SEL-001_0001", "JH_RM-SEL-001_0002"])
self.assertEqual(float(self.db.get(StockLot, self.lot1.id).remaining_weight_kg), 20)
self.assertEqual(float(self.db.get(StockLot, self.lot2.id).remaining_weight_kg), 20)
persisted = self.db.scalars(select(WorkOrderMaterialIssue).order_by(WorkOrderMaterialIssue.id)).all()
self.assertEqual(len(persisted), 2)
self.assertEqual(float(persisted[0].amount), 320)
self.assertEqual(float(persisted[1].amount), 150)
def test_issue_selected_stock_lots_rejects_depleted_or_overdrawn_lots(self) -> None:
self.lot2.remaining_weight_kg = Decimal("0")
self.lot2.status = "DEPLETED"
self.db.commit()
with self.assertRaises(HTTPException) as exc:
issue_selected_stock_lots(
self.db,
work_order_id=1,
work_order_material_id=1,
material_item_id=self.material.id,
selected_lots=[{"source_lot_id": self.lot2.id, "issued_weight_kg": Decimal("1")}],
operator_user_id=1,
)
self.assertEqual(exc.exception.status_code, 400)
self.assertIn("不可用", exc.exception.detail)
```
- [ ] **Step 3: Write failing multi-sales-order purchase tests**
Create `backend/tests/test_purchase_order_multi_sales_links.py` with this content:
```python
from __future__ import annotations
import unittest
from datetime import date
from decimal import Decimal
from sqlalchemy import BigInteger, create_engine, select
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import Session, sessionmaker
@compiles(BigInteger, "sqlite")
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
_ = type_, compiler, kw
return "INTEGER"
import app.models.master_data # noqa: E402,F401
import app.models.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.api.routes.purchase import create_purchase_order, update_purchase_order # noqa: E402
from app.models.base import Base # noqa: E402
from app.models.master_data import Item, Material # noqa: E402
from app.models.operations import PurchaseOrderSalesOrderLink, Supplier # noqa: E402
from app.models.sales import Customer, SalesOrder, SalesOrderItem # noqa: E402
from app.schemas.operations import PurchaseOrderCreate, PurchaseOrderItemCreate, PurchaseOrderUpdate # noqa: E402
class PurchaseOrderMultiSalesLinksTest(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.supplier = Supplier(id=1, supplier_code="SUP-MULTI-001", supplier_name="多订单供应商", short_name="多订单供应商", lead_time_days=0, default_tax_rate=Decimal("0"), status="ACTIVE")
self.customer = Customer(id=1, customer_code="CUS-MULTI-001", customer_name="多订单客户", short_name="客户", status="ACTIVE")
self.product = Item(id=1, item_code="FG-MULTI-001", item_name="多订单成品", item_type="FINISHED_GOOD", unit_weight_kg=Decimal("1"), status="ACTIVE")
self.material = Item(id=2, item_code="RM-MULTI-001", item_name="多订单原料", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE")
self.db.add_all([self.supplier, self.customer, self.product, self.material])
self.db.flush()
self.db.add(Material(item_id=self.material.id, default_supplier_id=self.supplier.id, purchase_calc_mode="BY_WEIGHT", min_purchase_qty=Decimal("0"), purchase_multiple_qty=Decimal("0"), purchase_multiple_weight_kg=Decimal("0"), lead_time_days=0))
self.order1 = SalesOrder(id=1, order_no="销售2026-00001", customer_id=self.customer.id, order_date=date.today(), status="OPEN")
self.order2 = SalesOrder(id=2, order_no="销售2026-00002", customer_id=self.customer.id, order_date=date.today(), status="OPEN")
self.item1 = SalesOrderItem(id=1, sales_order_id=1, line_no=1, product_item_id=self.product.id, order_qty=Decimal("100"), delivered_qty=Decimal("0"), unit_price=Decimal("10"), line_amount=Decimal("1000"), status="OPEN")
self.item2 = SalesOrderItem(id=2, sales_order_id=2, line_no=1, product_item_id=self.product.id, order_qty=Decimal("200"), delivered_qty=Decimal("0"), unit_price=Decimal("10"), line_amount=Decimal("2000"), status="OPEN")
self.db.add_all([self.order1, self.order2, self.item1, self.item2])
self.db.commit()
def tearDown(self) -> None:
self.db.close()
def test_purchase_order_can_link_multiple_sales_orders(self) -> None:
result = create_purchase_order(
PurchaseOrderCreate(
supplier_id=self.supplier.id,
expected_date=date.today(),
target_warehouse_type="RAW",
sales_order_ids=[self.order1.id, self.order2.id],
items=[PurchaseOrderItemCreate(material_item_id=self.material.id, order_weight_kg=Decimal("300"), unit_price=Decimal("6"))],
),
db=self.db,
)
links = self.db.scalars(select(PurchaseOrderSalesOrderLink).order_by(PurchaseOrderSalesOrderLink.sales_order_id)).all()
self.assertEqual([link.sales_order_id for link in links], [self.order1.id, self.order2.id])
self.assertEqual(result.source_type, "SALES_ORDER")
self.assertEqual(result.source_order_nos, "销售2026-00001、销售2026-00002")
def test_purchase_order_update_replaces_sales_order_links(self) -> None:
result = create_purchase_order(
PurchaseOrderCreate(
supplier_id=self.supplier.id,
expected_date=date.today(),
target_warehouse_type="RAW",
sales_order_ids=[self.order1.id, self.order2.id],
items=[PurchaseOrderItemCreate(material_item_id=self.material.id, order_weight_kg=Decimal("300"), unit_price=Decimal("6"))],
),
db=self.db,
)
updated = update_purchase_order(
result.purchase_order_id,
PurchaseOrderUpdate(
supplier_id=self.supplier.id,
expected_date=date.today(),
target_warehouse_type="RAW",
sales_order_ids=[self.order2.id],
items=[PurchaseOrderItemCreate(material_item_id=self.material.id, order_weight_kg=Decimal("250"), unit_price=Decimal("6"))],
),
db=self.db,
)
links = self.db.scalars(select(PurchaseOrderSalesOrderLink).order_by(PurchaseOrderSalesOrderLink.sales_order_id)).all()
self.assertEqual([link.sales_order_id for link in links], [self.order2.id])
self.assertEqual(updated.source_order_nos, "销售2026-00002")
```
- [ ] **Step 4: Run new tests and verify they fail for missing implementation**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_inventory_lot_numbering.py tests/test_selected_stock_lot_production_issue.py tests/test_purchase_order_multi_sales_links.py -q
```
Expected: FAIL because `build_inventory_lot_no`, `issue_selected_stock_lots`, `PurchaseOrderSalesOrderLink`, and `sales_order_ids` do not exist yet.
- [ ] **Step 5: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
Expected: The current workspace may not be a git repository. If it is a git repository, commit the failing tests with:
```bash
git add backend/tests/test_inventory_lot_numbering.py backend/tests/test_selected_stock_lot_production_issue.py backend/tests/test_purchase_order_multi_sales_links.py
git commit -m "test: cover unified inventory lot trace model"
```
---
### Task 2: Add Database Compatibility Schema
**Files:**
- Modify: `backend/app/models/operations.py`
- Create: `backend/sql/unify_inventory_lot_trace_patch.sql`
- [ ] **Step 1: Add purchase-order sales-order link ORM model**
In `backend/app/models/operations.py`, add this model near `PurchaseOrder` and `PurchaseOrderItem`:
```python
class PurchaseOrderSalesOrderLink(Base):
__tablename__ = "po_purchase_order_sales_order"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
purchase_order_id: Mapped[int] = mapped_column(ForeignKey("po_purchase_order.id"))
sales_order_id: Mapped[int] = mapped_column(ForeignKey("so_sales_order.id"))
created_at: Mapped[object] = mapped_column(DateTime)
updated_at: Mapped[object] = mapped_column(DateTime)
```
- [ ] **Step 2: Add SQL migration**
Create `backend/sql/unify_inventory_lot_trace_patch.sql`:
```sql
CREATE TABLE IF NOT EXISTS po_purchase_order_sales_order (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
purchase_order_id BIGINT NOT NULL,
sales_order_id BIGINT NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uq_po_so_link (purchase_order_id, sales_order_id),
KEY idx_po_so_link_po (purchase_order_id),
KEY idx_po_so_link_so (sales_order_id),
CONSTRAINT fk_po_so_link_po FOREIGN KEY (purchase_order_id) REFERENCES po_purchase_order(id) ON DELETE CASCADE,
CONSTRAINT fk_po_so_link_so FOREIGN KEY (sales_order_id) REFERENCES so_sales_order(id)
);
CREATE INDEX idx_wh_stock_lot_item_lot_no ON wh_stock_lot (item_id, lot_no);
CREATE INDEX idx_pp_issue_source_lot ON pp_work_order_material_issue (source_lot_id);
INSERT INTO po_purchase_order_sales_order (purchase_order_id, sales_order_id, created_at, updated_at)
SELECT DISTINCT po.id, so.id, NOW(), NOW()
FROM po_purchase_order po
JOIN po_purchase_order_item poi ON poi.purchase_order_id = po.id
JOIN pl_material_demand md ON md.id = poi.source_demand_id
JOIN so_sales_order_item soi ON soi.id = md.sales_order_item_id
JOIN so_sales_order so ON so.id = soi.sales_order_id
WHERE po.source_type = 'SALES_ORDER'
AND NOT EXISTS (
SELECT 1
FROM po_purchase_order_sales_order existing
WHERE existing.purchase_order_id = po.id
AND existing.sales_order_id = so.id
);
UPDATE pp_work_order_material_issue issue
JOIN wh_stock_lot lot ON lot.id = issue.source_lot_id
SET issue.material_sub_batch_no = lot.lot_no
WHERE issue.material_sub_batch_no IS NULL
OR issue.material_sub_batch_no = ''
OR issue.material_sub_batch_no NOT LIKE 'JH\\_%';
```
- [ ] **Step 3: Run model/test import smoke check**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_purchase_order_multi_sales_links.py::PurchaseOrderMultiSalesLinksTest::test_purchase_order_can_link_multiple_sales_orders -q
```
Expected: The test should still fail on missing schema fields/service logic, but no longer fail with `ImportError` for `PurchaseOrderSalesOrderLink`.
- [ ] **Step 4: Apply migration to local database when implementing against MySQL**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
mysql --default-character-set=utf8mb4 -u root -p jiaheng_erp < backend/sql/unify_inventory_lot_trace_patch.sql
```
Expected: MySQL prints no error. If the local project uses a different DB user, use the same command style already used for prior project SQL patches.
- [ ] **Step 5: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/models/operations.py backend/sql/unify_inventory_lot_trace_patch.sql
git commit -m "feat: add unified inventory lot trace schema"
```
---
### Task 3: Implement Inventory Lot Number Service And Availability Query
**Files:**
- Modify: `backend/app/services/operations.py`
- Modify: `backend/app/schemas/operations.py`
- Modify: `backend/app/api/routes/inventory.py`
- [ ] **Step 1: Add lot availability schema**
In `backend/app/schemas/operations.py`, add:
```python
class StockLotOptionRead(BaseModel):
lot_id: int
lot_no: str
item_id: int
item_code: str
item_name: str
warehouse_id: int
warehouse_name: str
remaining_weight_kg: float
unit_cost: float
quality_status: str
status: str
is_available_for_issue: bool
disabled_reason: str | None = None
```
- [ ] **Step 2: Replace generator service**
In `backend/app/services/operations.py`, keep `build_material_sub_batch_no` temporarily but add the new canonical generator above it:
```python
def build_inventory_lot_no(db: Session, material_item_id: int) -> str:
material = db.get(Item, material_item_id)
material_code = material.item_code if material else f"MAT{material_item_id}"
prefix = f"JH_{material_code}_"
existing_lot_nos = db.scalars(
select(StockLot.lot_no).where(
StockLot.item_id == material_item_id,
StockLot.lot_no.like(f"{prefix}%"),
)
).all()
max_sequence = 0
for lot_no in existing_lot_nos:
tail = str(lot_no).replace(prefix, "", 1)
if len(tail) == 4 and tail.isdigit():
max_sequence = max(max_sequence, int(tail))
sequence = max_sequence + 1
candidate = f"{prefix}{sequence:04d}"
while db.scalar(select(func.count(StockLot.id)).where(StockLot.lot_no == candidate)):
sequence += 1
candidate = f"{prefix}{sequence:04d}"
return candidate
```
- [ ] **Step 3: Add usable-first stock lot query**
In `backend/app/services/operations.py`, add:
```python
def get_stock_lot_options_query(material_item_id: int | None = None) -> Select:
is_available_expr = case(
(
(StockLot.quality_status == "PASS")
& (StockLot.status == "AVAILABLE")
& (StockLot.remaining_weight_kg > 0),
1,
),
else_=0,
)
stmt = (
select(
StockLot.id.label("lot_id"),
StockLot.lot_no.label("lot_no"),
StockLot.item_id.label("item_id"),
Item.item_code.label("item_code"),
Item.item_name.label("item_name"),
StockLot.warehouse_id.label("warehouse_id"),
Warehouse.warehouse_name.label("warehouse_name"),
StockLot.remaining_weight_kg.label("remaining_weight_kg"),
StockLot.unit_cost.label("unit_cost"),
StockLot.quality_status.label("quality_status"),
StockLot.status.label("status"),
is_available_expr.label("is_available_for_issue"),
case(
(StockLot.quality_status != "PASS", "未质检放行"),
(StockLot.remaining_weight_kg <= 0, "已用完"),
(StockLot.status != "AVAILABLE", "当前状态不可用"),
else_=None,
).label("disabled_reason"),
)
.join(Item, Item.id == StockLot.item_id)
.join(Warehouse, Warehouse.id == StockLot.warehouse_id)
.where(Warehouse.warehouse_type == "RAW")
.order_by(is_available_expr.desc(), StockLot.lot_no.asc(), StockLot.id.asc())
)
if material_item_id:
stmt = stmt.where(StockLot.item_id == material_item_id)
return stmt
```
Ensure `case` is imported from SQLAlchemy in this file:
```python
from sqlalchemy import Select, and_, case, false, func, or_, select
```
- [ ] **Step 4: Add stock lot options endpoint**
In `backend/app/api/routes/inventory.py`, import `StockLotOptionRead` and `get_stock_lot_options_query`, then add:
```python
@router.get("/stock-lot-options", response_model=list[StockLotOptionRead])
def list_stock_lot_options(
material_item_id: int | None = Query(default=None),
limit: int = Query(default=500, ge=1, le=1000),
db: Session = Depends(get_db),
) -> list[StockLotOptionRead]:
rows = db.execute(get_stock_lot_options_query(material_item_id=material_item_id).limit(limit)).mappings().all()
return [StockLotOptionRead.model_validate(dict(row)) for row in rows]
```
- [ ] **Step 5: Run lot-numbering test**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_inventory_lot_numbering.py -q
```
Expected: PASS.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/services/operations.py backend/app/schemas/operations.py backend/app/api/routes/inventory.py backend/tests/test_inventory_lot_numbering.py
git commit -m "feat: add canonical inventory lot numbering"
```
---
### Task 4: Change Purchase Receipt To Generate Canonical Inventory Lot Numbers
**Files:**
- Modify: `backend/app/api/routes/purchase.py`
- Modify: `backend/tests/test_raw_material_weight_only_flow.py`
- [ ] **Step 1: Update receipt tests for new lot numbers and no forced merge**
In `backend/tests/test_raw_material_weight_only_flow.py`, add this test to `RawMaterialWeightOnlyFlowTest`:
```python
def test_split_arrivals_create_sequential_inventory_lots_for_same_purchase_line(self) -> None:
order = create_purchase_order(
PurchaseOrderCreate(
supplier_id=self.supplier.id,
expected_date=date.today(),
items=[
PurchaseOrderItemCreate(
material_item_id=self.material_item.id,
order_weight_kg=Decimal("500"),
unit_price=Decimal("4"),
)
],
),
db=self.db,
)
po_item = self.db.scalar(select(PurchaseOrderItem).where(PurchaseOrderItem.purchase_order_id == order.purchase_order_id))
lock_purchase_order(order.purchase_order_id, db=self.db)
create_purchase_receipt(
PurchaseReceiptCreate(
purchase_order_id=order.purchase_order_id,
warehouse_id=self.warehouse.id,
receiver_employee_id=self.employee.id,
items=[
PurchaseReceiptItemCreate(
purchase_order_item_id=po_item.id,
material_item_id=self.material_item.id,
received_weight_kg=Decimal("300"),
unit_cost=Decimal("4"),
)
],
),
context=self._auth_context(),
db=self.db,
)
create_purchase_receipt(
PurchaseReceiptCreate(
purchase_order_id=order.purchase_order_id,
warehouse_id=self.warehouse.id,
receiver_employee_id=self.employee.id,
items=[
PurchaseReceiptItemCreate(
purchase_order_item_id=po_item.id,
material_item_id=self.material_item.id,
received_weight_kg=Decimal("250"),
unit_cost=Decimal("4.5"),
)
],
),
context=self._auth_context(),
db=self.db,
)
lots = self.db.scalars(select(StockLot).order_by(StockLot.id)).all()
receipt_items = self.db.scalars(select(PurchaseReceiptItem).order_by(PurchaseReceiptItem.id)).all()
self.assertEqual([lot.lot_no for lot in lots], ["JH_RM-WEIGHT-001_0001", "JH_RM-WEIGHT-001_0002"])
self.assertEqual(float(receipt_items[0].unit_cost), 4)
self.assertEqual(float(receipt_items[1].unit_cost), 4.5)
self.assertEqual(float(lots[0].unit_cost), 4)
self.assertEqual(float(lots[1].unit_cost), 4.5)
```
- [ ] **Step 2: Run the updated raw-material test and verify failure**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_raw_material_weight_only_flow.py::RawMaterialWeightOnlyFlowTest::test_split_arrivals_create_sequential_inventory_lots_for_same_purchase_line -q
```
Expected: FAIL because existing receipt logic forces merge and uses the old lot number format.
- [ ] **Step 3: Replace purchase receipt lot number creation**
In `backend/app/api/routes/purchase.py`, import:
```python
from app.services.operations import build_inventory_lot_no
```
In `create_purchase_receipt`, remove this block:
```python
existing_po_item_lot = _find_existing_receipt_lot_for_po_item(db, item_payload.purchase_order_item_id)
if existing_po_item_lot:
if item_payload.merge_to_lot_id and item_payload.merge_to_lot_id != existing_po_item_lot.id:
raise HTTPException(
status_code=400,
detail=f"采购行 {po_item.line_no} 已有入库批次 {existing_po_item_lot.lot_no}，本次到货必须合并到该批次",
)
merge_target_lot = existing_po_item_lot
elif item_payload.merge_to_lot_id:
merge_target_lot = db.get(StockLot, item_payload.merge_to_lot_id)
if not merge_target_lot:
raise HTTPException(status_code=404, detail=f"合并目标批次不存在: {item_payload.merge_to_lot_id}")
```
Replace it with:
```python
if item_payload.merge_to_lot_id:
merge_target_lot = db.get(StockLot, item_payload.merge_to_lot_id)
if not merge_target_lot:
raise HTTPException(status_code=404, detail=f"合并目标批次不存在: {item_payload.merge_to_lot_id}")
```
Then replace old default lot generation:
```python
lot_no = item_payload.lot_no or _build_purchase_lot_no(
db,
material_code=material_item.item_code,
receipt_dt=receipt.receipt_date,
po_no=purchase_order.po_no,
line_no=po_item.line_no,
)
```
with:
```python
lot_no = item_payload.lot_no or build_inventory_lot_no(db, item_payload.material_item_id)
```
Keep explicit merge behavior unchanged:
```python
if merge_target_lot:
final_unit_cost = to_decimal(merge_target_lot.unit_cost, "0.0001")
else:
final_unit_cost = entered_unit_cost
```
- [ ] **Step 4: Run purchase receipt tests**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_raw_material_weight_only_flow.py -q
```
Expected: PASS, including the existing explicit merge test where merge uses the first lot unit cost.
- [ ] **Step 5: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/api/routes/purchase.py backend/tests/test_raw_material_weight_only_flow.py
git commit -m "feat: create sequential inventory lots on receipt"
```
---
### Task 5: Implement Selected Stock Lot Production Issue
**Files:**
- Modify: `backend/app/schemas/operations.py`
- Modify: `backend/app/services/operations.py`
- Modify: `backend/app/api/routes/production.py`
- [ ] **Step 1: Add selected lot payload schemas**
In `backend/app/schemas/operations.py`, add:
```python
class SelectedStockLotIssueCreate(BaseModel):
source_lot_id: int
issued_weight_kg: float
class WorkOrderCreate(BaseModel):
product_item_id: int
source_sales_order_item_id: int | None = None
planned_qty: float = 0
issue_weight_kg: float = 0
auto_issue_materials: bool = True
selected_stock_lots: list[SelectedStockLotIssueCreate] = Field(default_factory=list)
planned_start_time: datetime | None = None
planned_end_time: datetime | None = None
priority_level: int | None = None
remark: str | None = None
```
`WorkOrderCreate` already exists in `backend/app/schemas/operations.py`; modify that existing class and do not add a duplicate class.
- [ ] **Step 2: Add selected-lot issue service**
In `backend/app/services/operations.py`, add:
```python
def issue_selected_stock_lots(
db: Session,
*,
work_order_id: int,
work_order_material_id: int,
material_item_id: int,
selected_lots: list[dict[str, Decimal]],
operator_user_id: int | None = None,
) -> list[WorkOrderMaterialIssue]:
if not selected_lots:
raise HTTPException(status_code=400, detail="请选择至少一个库存批次")
issues: list[WorkOrderMaterialIssue] = []
for selected in selected_lots:
source_lot_id = int(selected["source_lot_id"])
issued_weight = to_decimal(selected["issued_weight_kg"])
if issued_weight <= 0:
raise HTTPException(status_code=400, detail="领用重量必须大于0")
lot = db.get(StockLot, source_lot_id)
if not lot or lot.item_id != material_item_id:
raise HTTPException(status_code=404, detail="库存批次不存在或物料不匹配")
if lot.quality_status != "PASS" or lot.status != "AVAILABLE" or to_decimal(lot.remaining_weight_kg) <= 0:
raise HTTPException(status_code=400, detail=f"库存批次 {lot.lot_no} 不可用")
if to_decimal(lot.remaining_weight_kg) < issued_weight:
raise HTTPException(status_code=400, detail=f"库存批次 {lot.lot_no} 剩余重量不足")
unit_cost = to_decimal(lot.unit_cost, "0.0001")
amount = (issued_weight * unit_cost).quantize(Decimal("0.01"))
lot.remaining_qty = to_decimal(0)
lot.remaining_weight_kg = to_decimal(lot.remaining_weight_kg) - issued_weight
if to_decimal(lot.remaining_weight_kg) <= 0:
lot.remaining_weight_kg = to_decimal(0)
lot.status = "DEPLETED"
db.add(lot)
issue = WorkOrderMaterialIssue(
work_order_material_id=work_order_material_id,
work_order_id=work_order_id,
material_item_id=material_item_id,
source_lot_id=source_lot_id,
material_sub_batch_no=lot.lot_no,
issued_qty=to_decimal(0),
issued_weight_kg=issued_weight,
unit_cost=unit_cost,
amount=amount,
issue_time=datetime.now(),
operator_user_id=operator_user_id,
remark="按库存批次生产出库",
)
db.add(issue)
issues.append(issue)
txn = create_inventory_txn(
db,
txn_type="ISSUE_TO_PRODUCTION",
item_id=material_item_id,
warehouse_id=lot.warehouse_id,
location_id=lot.location_id,
lot_id=lot.id,
qty_change=to_decimal(0),
weight_change=-issued_weight,
unit_cost=unit_cost,
source_doc_type="WORK_ORDER",
source_doc_id=work_order_id,
source_line_id=work_order_material_id,
biz_time=datetime.now(),
operator_user_id=operator_user_id,
remark=f"生产出库：{lot.lot_no}",
amount_basis="WEIGHT",
)
txn.amount = -abs(to_decimal(txn.amount, "0.01"))
db.add(txn)
upsert_stock_balance(
db,
item_id=material_item_id,
warehouse_id=lot.warehouse_id,
location_id=lot.location_id,
qty_delta=to_decimal(0),
weight_delta=-issued_weight,
available_qty_delta=to_decimal(0),
available_weight_delta=-issued_weight,
unit_cost=unit_cost,
)
db.flush()
return issues
```
Call `create_inventory_txn` and `upsert_stock_balance` from `backend/app/services/operations.py`; do not create a second transaction-number helper.
- [ ] **Step 3: Wire selected lots into work order creation**
In `backend/app/api/routes/production.py`, change `create_work_order` so that:
```python
selected_lots = [
{"source_lot_id": int(row.source_lot_id), "issued_weight_kg": to_decimal(row.issued_weight_kg)}
for row in payload.selected_stock_lots
]
```
When `selected_lots` is non-empty:
```python
payload.auto_issue_materials = False
issue_weight = sum((to_decimal(row["issued_weight_kg"]) for row in selected_lots), Decimal("0"))
```
After `create_work_order_materials_and_operations(...)`, find the first material row and call:
```python
if selected_lots:
first_material = db.scalar(
select(WorkOrderMaterial)
.where(WorkOrderMaterial.work_order_id == work_order.id)
.order_by(WorkOrderMaterial.id)
.limit(1)
)
if not first_material:
raise HTTPException(status_code=400, detail="产品需规缺少默认用料，无法按库存批次出库")
issue_selected_stock_lots(
db,
work_order_id=work_order.id,
work_order_material_id=first_material.id,
material_item_id=first_material.material_item_id,
selected_lots=selected_lots,
operator_user_id=context.user.id,
)
```
Remove this legacy assignment:
```python
if first_issue:
work_order.work_order_no = first_issue.material_sub_batch_no
```
Keep `_build_work_order_no(db, product)` as the production order identifier. The UI should no longer call it “材料分批次号”.
- [ ] **Step 4: Run selected-lot tests**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_selected_stock_lot_production_issue.py -q
```
Expected: PASS.
- [ ] **Step 5: Run production-adjacent tests**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_raw_material_weight_only_flow.py tests/test_scrap_warehouse_flow.py -q
```
Expected: PASS.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/schemas/operations.py backend/app/services/operations.py backend/app/api/routes/production.py backend/tests/test_selected_stock_lot_production_issue.py
git commit -m "feat: issue production materials by inventory lot"
```
---
### Task 6: Update Miniapp Report Sync To Treat The Miniapp Field As 库存批次号
**Files:**
- Modify: `backend/app/api/routes/production.py`
- Modify: `frontend/src/views/OperationReportView.vue`
- [ ] **Step 1: Add a backend resolver for miniapp lot number**
In `backend/app/api/routes/production.py`, replace resolver logic that queries:
```python
WorkOrderMaterialIssue.material_sub_batch_no == batch_no
```
with a resolver that first uses stock lot number:
```python
def _resolve_work_order_by_inventory_lot_no(db: Session, lot_no: str) -> WorkOrderMaterialIssue | None:
return db.scalar(
select(WorkOrderMaterialIssue)
.join(StockLot, StockLot.id == WorkOrderMaterialIssue.source_lot_id)
.where(StockLot.lot_no == lot_no)
.order_by(WorkOrderMaterialIssue.id.desc())
.limit(1)
)
```
For compatibility, if this returns `None`, fall back to:
```python
db.scalar(
select(WorkOrderMaterialIssue)
.where(WorkOrderMaterialIssue.material_sub_batch_no == lot_no)
.order_by(WorkOrderMaterialIssue.id.desc())
.limit(1)
)
```
- [ ] **Step 2: Change miniapp error text**
Replace:
```python
"小程序报工缺少有效原材料分批次号，不能同步为 ERP 报工"
```
with:
```python
"小程序报工缺少有效库存批次号，不能同步为 ERP 报工"
```
Replace:
```python
"该材料分批次缺少可匹配的小程序报工工序"
```
with:
```python
"该库存批次缺少可匹配的小程序报工工序"
```
- [ ] **Step 3: Rename frontend operation report labels**
In `frontend/src/views/OperationReportView.vue`, replace user-facing strings:
```text
材料分批次号
原材料分批次号
```
with:
```text
库存批次号
```
Keep the raw API field name if the miniapp still sends `raw_material_batch_no`.
- [ ] **Step 4: Run production route tests**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_selected_stock_lot_production_issue.py tests/test_product_spec_miniapp_material_sync.py -q
```
Expected: PASS.
- [ ] **Step 5: Run frontend build smoke**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build
```
Expected: PASS. Existing Vite chunk-size warnings are acceptable if no new error is introduced.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/api/routes/production.py frontend/src/views/OperationReportView.vue
git commit -m "feat: sync miniapp reports by inventory lot"
```
---
### Task 7: Update Finished Goods Receipt And Delivery Trace
**Files:**
- Modify: `backend/app/api/routes/production.py`
- Modify: `backend/app/api/routes/sales.py`
- Modify: `backend/app/services/operations.py`
- Modify: `backend/tests/test_sales_order_delivery_trace.py`
- [ ] **Step 1: Update sales delivery trace test expectation**
In `backend/tests/test_sales_order_delivery_trace.py`, change the seeded finished lot:
```python
source_material_sub_batch_no="JH_MAT001_0001",
source_material_summary="JH_MAT001_0001 / 冷轧钢板 / 40kg / ¥200",
```
to:
```python
source_material_sub_batch_no="RML-001",
source_material_summary="RML-001 / 冷轧钢板 / 40kg / ¥200",
```
Then update assertions so trace rows expect source inventory lot number:
```python
self.assertEqual(trace_row["source_material_lot_no"], "RML-001")
self.assertEqual(trace_row["source_material_sub_batch_no"], "RML-001")
```
- [ ] **Step 2: Change completion source summary builder**
In `backend/app/api/routes/production.py`, where source material rows are summarized, replace summary pieces using:
```python
row["material_sub_batch_no"]
```
with:
```python
row["source_lot_no"]
```
The summary line should be:
```python
f"{row['source_lot_no']} / {row['material_name']} / {weight}kg / ¥{amount}"
```
Set completion source fields as:
```python
"source_material_lot_id": first["source_lot_id"],
"source_material_sub_batch_no": first["source_lot_no"],
```
- [ ] **Step 3: Update delivery item query labels**
In `backend/app/services/operations.py`, update `get_delivery_items_query` so:
```python
source_material_lot_no
```
comes from `source_lot.lot_no` where available, and `source_material_sub_batch_no` is a compatibility alias equal to that same inventory lot number for new rows.
Use this expression:
```python
func.coalesce(source_material_lot.lot_no, StockLot.source_material_sub_batch_no).label("source_material_lot_no")
```
And:
```python
func.coalesce(source_material_lot.lot_no, StockLot.source_material_sub_batch_no).label("source_material_sub_batch_no")
```
- [ ] **Step 4: Run delivery trace test**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_sales_order_delivery_trace.py -q
```
Expected: PASS.
- [ ] **Step 5: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/api/routes/production.py backend/app/services/operations.py backend/tests/test_sales_order_delivery_trace.py
git commit -m "feat: trace finished goods to inventory lots"
```
---
### Task 8: Implement Multiple Sales Orders Per Purchase Order
**Files:**
- Modify: `backend/app/schemas/operations.py`
- Modify: `backend/app/api/routes/purchase.py`
- Modify: `backend/app/services/operations.py`
- Modify: `backend/tests/test_purchase_order_multi_sales_links.py`
- Modify: `backend/tests/test_sales_order_purchase_lock_flow.py`
- [ ] **Step 1: Add schema fields**
In `backend/app/schemas/operations.py`, add to `PurchaseOrderCreate`:
```python
sales_order_ids: list[int] = Field(default_factory=list)
```
Add to `PurchaseOrderRead`:
```python
sales_order_ids: list[int] = Field(default_factory=list)
```
- [ ] **Step 2: Add helper functions in purchase route**
In `backend/app/api/routes/purchase.py`, add:
```python
def _replace_purchase_order_sales_links(db: Session, purchase_order_id: int, sales_order_ids: list[int]) -> None:
normalized_ids = sorted({int(order_id) for order_id in sales_order_ids if int(order_id) > 0})
db.query(PurchaseOrderSalesOrderLink).filter(
PurchaseOrderSalesOrderLink.purchase_order_id == purchase_order_id
).delete(synchronize_session=False)
for sales_order_id in normalized_ids:
sales_order = db.get(SalesOrder, sales_order_id)
if not sales_order:
raise HTTPException(status_code=404, detail=f"销售订单不存在: {sales_order_id}")
db.add(
PurchaseOrderSalesOrderLink(
purchase_order_id=purchase_order_id,
sales_order_id=sales_order_id,
created_at=datetime.now(),
updated_at=datetime.now(),
)
)
def _sales_order_ids_from_items(db: Session, items: list[PurchaseOrderItemCreate]) -> list[int]:
ids: set[int] = set()
for item in items:
if not item.source_demand_id:
continue
row = db.execute(
select(SalesOrder.id)
.join(SalesOrderItem, SalesOrderItem.sales_order_id == SalesOrder.id)
.join(MaterialDemand, MaterialDemand.sales_order_item_id == SalesOrderItem.id)
.where(MaterialDemand.id == item.source_demand_id)
).scalar_one_or_none()
if row:
ids.add(int(row))
return sorted(ids)
```
Import `PurchaseOrderSalesOrderLink`, `SalesOrder`, `SalesOrderItem`, `MaterialDemand`, and `datetime` if they are not already imported in this file.
- [ ] **Step 3: Call helper from create/update**
After `purchase_order` is flushed in `create_purchase_order`, call:
```python
explicit_sales_order_ids = payload.sales_order_ids or []
inferred_sales_order_ids = _sales_order_ids_from_items(db, payload.items)
linked_sales_order_ids = sorted({*explicit_sales_order_ids, *inferred_sales_order_ids})
_replace_purchase_order_sales_links(db, purchase_order.id, linked_sales_order_ids)
if linked_sales_order_ids:
purchase_order.source_type = "SALES_ORDER"
```
In `update_purchase_order`, after replacing items, call the same logic.
- [ ] **Step 4: Update purchase order query source summary**
In `backend/app/api/routes/purchase.py`, add a post-query enrichment helper. This avoids dialect-specific `GROUP_CONCAT` syntax and works in both MySQL and SQLite tests:
```python
def _enrich_purchase_order_sources(db: Session, rows: list[dict]) -> list[dict]:
purchase_order_ids = [int(row["purchase_order_id"]) for row in rows]
if not purchase_order_ids:
return rows
link_rows = db.execute(
select(
PurchaseOrderSalesOrderLink.purchase_order_id.label("purchase_order_id"),
SalesOrder.id.label("sales_order_id"),
SalesOrder.order_no.label("order_no"),
)
.join(SalesOrder, SalesOrder.id == PurchaseOrderSalesOrderLink.sales_order_id)
.where(PurchaseOrderSalesOrderLink.purchase_order_id.in_(purchase_order_ids))
.order_by(PurchaseOrderSalesOrderLink.purchase_order_id, SalesOrder.order_no)
).mappings().all()
links_by_po: dict[int, list[dict]] = {}
for link in link_rows:
links_by_po.setdefault(int(link["purchase_order_id"]), []).append(dict(link))
for row in rows:
links = links_by_po.get(int(row["purchase_order_id"]), [])
if links:
row["sales_order_ids"] = [int(link["sales_order_id"]) for link in links]
row["source_order_nos"] = "、".join(str(link["order_no"]) for link in links)
row["source_type"] = "SALES_ORDER"
else:
row["sales_order_ids"] = []
return rows
```
Call `_enrich_purchase_order_sources` in list/detail read paths before converting rows to `PurchaseOrderRead`.
- [ ] **Step 5: Run multi-sales-order tests**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_purchase_order_multi_sales_links.py tests/test_sales_order_purchase_lock_flow.py -q
```
Expected: PASS.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/schemas/operations.py backend/app/api/routes/purchase.py backend/app/services/operations.py backend/tests/test_purchase_order_multi_sales_links.py backend/tests/test_sales_order_purchase_lock_flow.py
git commit -m "feat: link purchase orders to multiple sales orders"
```
---
### Task 9: Update Frontend Stock Lot Multi-Select And Production Flow
**Files:**
- Create: `frontend/src/components/StockLotTagSelect.vue`
- Modify: `frontend/src/views/WorkOrderManagementView.vue`
- Modify: `frontend/src/views/InventoryLedgerView.vue`
- Modify: `frontend/src/services/api.js`
- [ ] **Step 1: Create stock lot tag select component**
Create `frontend/src/components/StockLotTagSelect.vue`:
```vue
<template>
<div class="stock-lot-tag-select" :class="{ open: dropdownOpen }">
<div class="tag-input-shell" @click="focusInput">
<span v-for="lot in selectedLots" :key="lot.lot_id" class="selected-lot-tag">
{{ lot.lot_no }}
<button type="button" class="tag-remove" :title="`移除 ${lot.lot_no}`" @click.stop="removeLot(lot.lot_id)">×</button>
</span>
<input
ref="inputRef"
v-model.trim="keyword"
type="text"
:placeholder="selectedLots.length ? '' : placeholder"
@focus="dropdownOpen = true"
/>
</div>
<div v-if="dropdownOpen" class="stock-lot-dropdown">
<button
v-for="lot in filteredLots"
:key="lot.lot_id"
type="button"
class="stock-lot-option"
:class="{ disabled: !lot.is_available_for_issue, selected: selectedIds.includes(lot.lot_id) }"
:disabled="!lot.is_available_for_issue"
@click="toggleLot(lot)"
>
<span class="lot-main">{{ lot.lot_no }} · {{ lot.item_name }}</span>
<span class="lot-meta">剩余 {{ formatWeight(lot.remaining_weight_kg) }} · ¥{{ formatAmount(lot.unit_cost) }}/kg</span>
<span v-if="!lot.is_available_for_issue" class="lot-disabled">{{ lot.disabled_reason || '不可用' }}</span>
</button>
<p v-if="!filteredLots.length" class="lot-empty">没有匹配的库存批次</p>
</div>
</div>
</template>
<script setup>
import { computed, ref } from "vue";
import { formatAmount, formatWeight } from "../utils/formatters";
const props = defineProps({
modelValue: { type: Array, default: () => [] },
options: { type: Array, default: () => [] },
placeholder: { type: String, default: "搜索库存批次号" }
});
const emit = defineEmits(["update:modelValue"]);
const keyword = ref("");
const dropdownOpen = ref(false);
const inputRef = ref(null);
const selectedIds = computed(() => props.modelValue.map((item) => Number(item.lot_id)));
const selectedLots = computed(() => props.modelValue);
const filteredLots = computed(() => {
const text = keyword.value.toLowerCase();
return props.options
.filter((lot) => {
if (!text) return true;
return [lot.lot_no, lot.item_code, lot.item_name, lot.disabled_reason].some((value) =>
String(value || "").toLowerCase().includes(text)
);
})
.sort((left, right) => {
if (Number(left.is_available_for_issue) !== Number(right.is_available_for_issue)) {
return Number(right.is_available_for_issue) - Number(left.is_available_for_issue);
}
return String(left.lot_no || "").localeCompare(String(right.lot_no || ""));
});
});
function focusInput() {
dropdownOpen.value = true;
inputRef.value?.focus();
}
function toggleLot(lot) {
if (!lot.is_available_for_issue) return;
if (selectedIds.value.includes(Number(lot.lot_id))) {
removeLot(lot.lot_id);
return;
}
emit("update:modelValue", [...props.modelValue, lot]);
keyword.value = "";
}
function removeLot(lotId) {
emit("update:modelValue", props.modelValue.filter((lot) => Number(lot.lot_id) !== Number(lotId)));
}
</script>
```
Add styles in the same file matching existing Ruoyi/card UI style:
```vue
<style scoped>
.stock-lot-tag-select {
position: relative;
}
.tag-input-shell {
min-height: 40px;
display: flex;
flex-wrap: wrap;
gap: 6px;
align-items: center;
padding: 6px 10px;
border: 1px solid var(--border-color, #dcdfe6);
border-radius: 8px;
background: rgba(255, 255, 255, 0.92);
}
.tag-input-shell input {
flex: 1;
min-width: 140px;
border: 0;
outline: 0;
background: transparent;
}
.selected-lot-tag {
display: inline-flex;
align-items: center;
gap: 4px;
padding: 3px 8px;
border-radius: 999px;
color: #155bd4;
background: #e8f1ff;
border: 1px solid #b8d5ff;
font-size: 12px;
}
.tag-remove {
border: 0;
background: transparent;
color: #155bd4;
cursor: pointer;
}
.stock-lot-dropdown {
position: absolute;
z-index: 30;
top: calc(100% + 4px);
left: 0;
right: 0;
max-height: 280px;
overflow: auto;
border: 1px solid #dcdfe6;
border-radius: 10px;
background: #fff;
box-shadow: 0 14px 30px rgba(15, 23, 42, 0.16);
}
.stock-lot-option {
width: 100%;
display: grid;
gap: 2px;
padding: 9px 12px;
border: 0;
border-bottom: 1px solid #f1f5f9;
background: #fff;
text-align: left;
cursor: pointer;
}
.stock-lot-option:hover:not(.disabled),
.stock-lot-option.selected {
background: #eef6ff;
}
.stock-lot-option.disabled {
color: #9ca3af;
background: #f8fafc;
cursor: not-allowed;
}
.lot-main {
font-weight: 700;
}
.lot-meta,
.lot-disabled,
.lot-empty {
font-size: 12px;
color: #64748b;
}
.lot-empty {
padding: 12px;
}
</style>
```
- [ ] **Step 2: Wire component into work order creation**
In `frontend/src/views/WorkOrderManagementView.vue`, import:
```js
import StockLotTagSelect from "../components/StockLotTagSelect.vue";
```
Add state:
```js
const stockLotOptions = ref([]);
const selectedStockLots = ref([]);
```
Load options:
```js
async function loadStockLotOptions(materialItemId) {
if (!materialItemId) {
stockLotOptions.value = [];
return;
}
stockLotOptions.value = await fetchResource(`/inventory/stock-lot-options?material_item_id=${Number(materialItemId)}&limit=1000`);
}
```
In the create drawer, replace material-sub-batch generation wording with:
```vue
<StockLotTagSelect
v-model="selectedStockLots"
:options="stockLotOptions"
placeholder="搜索并选择库存批次号，可多选"
/>
```
For each selected lot, render an issue-weight input:
```vue
<div v-for="lot in selectedStockLots" :key="lot.lot_id" class="selected-lot-weight-row">
<span>{{ lot.lot_no }}</span>
<input v-model.number="lot.issued_weight_kg" type="number" min="0" step="0.001" :max="lot.remaining_weight_kg" />
<span>可用 {{ formatWeight(lot.remaining_weight_kg) }}</span>
</div>
```
Build payload:
```js
selected_stock_lots: selectedStockLots.value.map((lot) => ({
source_lot_id: Number(lot.lot_id),
issued_weight_kg: Number(lot.issued_weight_kg || lot.remaining_weight_kg || 0)
}))
```
- [ ] **Step 3: Rename production wording in inventory ledger**
In `frontend/src/views/InventoryLedgerView.vue`, replace:
```text
生成材料分批次号
材料分批次号
领用原材料生成材料分批次号，后续小程序报工围绕该分批次号推进。
```
with:
```text
生产出库
库存批次号
选择原材料库存批次出库，后续小程序报工按库存批次号推进。
```
- [ ] **Step 4: Run frontend build**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build
```
Expected: PASS.
- [ ] **Step 5: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add frontend/src/components/StockLotTagSelect.vue frontend/src/views/WorkOrderManagementView.vue frontend/src/views/InventoryLedgerView.vue frontend/src/services/api.js
git commit -m "feat: select production inventory lots in frontend"
```
---
### Task 10: Update Purchase Order Frontend For Multiple Sales Orders
**Files:**
- Modify: `frontend/src/views/PurchaseOrderView.vue`
- Reuse: `frontend/src/components/GlobalSearchableSelect.vue`
- [ ] **Step 1: Replace single sales order field with multi-tag state**
In `frontend/src/views/PurchaseOrderView.vue`, replace:
```js
sales_order_id: "",
```
with:
```js
sales_order_ids: [],
```
Add computed selected orders:
```js
const selectedSalesOrders = computed(() =>
salesOrders.value.filter((order) => form.sales_order_ids.map(Number).includes(Number(order.order_id)))
);
```
- [ ] **Step 2: Add multi-select UI**
Replace the single `<select v-model.number="form.sales_order_id">` with a local searchable multi-select tag field:
```vue
<div v-if="form.purchase_type === 'SALES_ORDER'" class="order-tag-picker">
<label class="form-field">
<span>关联销售订单</span>
<input v-model.trim="salesOrderKeyword" type="text" placeholder="搜索销售订单号、客户、产品" @focus="salesOrderPickerOpen = true" />
</label>
<div class="selected-order-tags">
<span v-for="order in selectedSalesOrders" :key="order.order_id" class="selected-lot-tag">
{{ order.order_no }}
<button type="button" class="tag-remove" @click="removeSalesOrder(order.order_id)">×</button>
</span>
</div>
<div v-if="salesOrderPickerOpen" class="stock-lot-dropdown">
<button
v-for="order in filteredSalesOrders"
:key="order.order_id"
type="button"
class="stock-lot-option"
@click="toggleSalesOrder(order.order_id)"
>
<span class="lot-main">{{ order.order_no }} · {{ order.customer_name }}</span>
<span class="lot-meta">{{ order.product_names || '-' }} · {{ order.status }}</span>
</button>
</div>
</div>
```
Add functions:
```js
function toggleSalesOrder(orderId) {
const id = Number(orderId);
if (form.sales_order_ids.map(Number).includes(id)) {
form.sales_order_ids = form.sales_order_ids.filter((value) => Number(value) !== id);
return;
}
form.sales_order_ids = [...form.sales_order_ids, id];
}
function removeSalesOrder(orderId) {
form.sales_order_ids = form.sales_order_ids.filter((value) => Number(value) !== Number(orderId));
}
```
- [ ] **Step 3: Update payload**
When submitting purchase orders, send:
```js
sales_order_ids: form.purchase_type === "SALES_ORDER" ? form.sales_order_ids.map(Number) : [],
```
Remove all remaining payload uses of:
```js
sales_order_id: form.sales_order_id
```
- [ ] **Step 4: Update demand calculation flow**
Change `applySalesOrderPurchase` to loop selected order ids:
```js
const allDemandRows = [];
for (const orderId of form.sales_order_ids) {
const result = await postResource(`/planning/generate-mrp/${Number(orderId)}`, {});
allDemandRows.push(...(result.demands || []));
}
applyDemandRowsToPurchaseItems(allDemandRows);
```
Add `applyDemandRowsToPurchaseItems(rows)` that groups by `material_item_id` and sums `suggested_purchase_weight_kg`, preserving one source demand id per material row for detail:
```js
function applyDemandRowsToPurchaseItems(rows) {
const grouped = new Map();
rows.forEach((row) => {
const key = Number(row.material_item_id);
if (!grouped.has(key)) {
grouped.set(key, { ...row, suggested_purchase_weight_kg: 0 });
}
grouped.get(key).suggested_purchase_weight_kg += Number(row.suggested_purchase_weight_kg || 0);
});
form.items = Array.from(grouped.values()).map((row, index) => ({
line_no: index + 1,
material_item_id: Number(row.material_item_id),
source_demand_id: row.demand_id || row.source_demand_id || null,
order_weight_kg: Number(row.suggested_purchase_weight_kg || 0),
unit_price: Number(row.reference_unit_price || 0),
remark: `销售订单采购：${form.sales_order_ids.length}张订单合并需求`
}));
}
```
- [ ] **Step 5: Run frontend build**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build
```
Expected: PASS.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add frontend/src/views/PurchaseOrderView.vue
git commit -m "feat: select multiple sales orders for purchase"
```
---
### Task 11: Rename Downstream UI Labels And Search Fields
**Files:**
- Modify: `frontend/src/views/CompletionReceiptView.vue`
- Modify: `frontend/src/views/DeliveryManagementView.vue`
- Modify: `frontend/src/views/SalesPlanningView.vue`
- Modify: `frontend/src/views/DashboardView.vue`
- Modify: `frontend/src/components/StocktakeDialog.vue`
- [ ] **Step 1: Replace user-facing material-sub-batch wording**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
rg -n "材料分批次|原材料分批次|来源分批次|材料分批号|来源分批号" frontend/src
```
Expected: The command prints remaining frontend occurrences.
For each occurrence that is user-facing, replace with:
```text
库存批次号
来源库存批次号
```
Keep JavaScript object keys unchanged only when they match backend compatibility fields.
- [ ] **Step 2: Update key table columns**
In `frontend/src/views/CompletionReceiptView.vue`, change table headers:
```text
材料分批次号 -> 来源库存批次号
```
In `frontend/src/views/DeliveryManagementView.vue`, change table headers:
```text
来源材料分批次号 -> 来源库存批次号
```
In `frontend/src/views/SalesPlanningView.vue`, change contract and trace labels:
```text
材料分批次号 -> 来源库存批次号
```
In `frontend/src/views/DashboardView.vue`, change:
```text
原料批次号 -> 库存批次号
分批领料 -> 生产出库
```
- [ ] **Step 3: Update search placeholder text**
Replace placeholders such as:
```text
搜索报废单号、材料分批次号、工序、产品、状态
```
with:
```text
搜索报废单号、库存批次号、工序、产品、状态
```
- [ ] **Step 4: Run frontend text scan**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
rg -n "材料分批次|原材料分批次|来源分批次|材料分批号|来源分批号" frontend/src
```
Expected: No user-facing labels remain. Compatibility field names such as `source_material_sub_batch_no` may remain in script code.
- [ ] **Step 5: Run frontend build**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build
```
Expected: PASS.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add frontend/src/views/CompletionReceiptView.vue frontend/src/views/DeliveryManagementView.vue frontend/src/views/SalesPlanningView.vue frontend/src/views/DashboardView.vue frontend/src/components/StocktakeDialog.vue
git commit -m "refactor: rename material sub-batch UI to inventory lot"
```
---
### Task 12: Update Dashboard DAG Backend To Use Inventory Lot Nodes
**Files:**
- Modify: `backend/app/api/routes/dashboard.py`
- Modify: `backend/app/schemas/dashboard.py`
- [ ] **Step 1: Rename DAG node detail labels**
In `backend/app/api/routes/dashboard.py`, replace DAG detail labels:
```python
("材料分批号", row["material_sub_batch_no"])
("来源分批号", row["source_material_sub_batch_no"])
```
with:
```python
("库存批次号", row["source_lot_no"])
("来源库存批次号", row["source_material_lot_no"])
```
Add `source_lot_no` to the production issue query:
```python
StockLot.lot_no.label("source_lot_no")
```
on the join to `WorkOrderMaterialIssue.source_lot_id`.
- [ ] **Step 2: Rename production issue nodes**
Change issue node titles from:
```python
title=row["material_sub_batch_no"]
summary="原材料分批领料..."
```
to:
```python
title=row["source_lot_no"]
summary="原材料按库存批次生产出库，记录领用重量、单价和成本。"
```
- [ ] **Step 3: Keep compatibility fallback**
For older rows without `source_lot_no`, use:
```python
source_lot_no = row["source_lot_no"] or row["material_sub_batch_no"] or "未记录库存批次号"
```
Use `source_lot_no` in node titles and details.
- [ ] **Step 4: Run backend dashboard import smoke**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python - <<'PY'
import app.api.routes.dashboard as dashboard
print("dashboard import ok", dashboard.router.prefix if hasattr(dashboard.router, "prefix") else "router")
PY
```
Expected: Prints `dashboard import ok`.
- [ ] **Step 5: Run backend tests**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests/test_sales_order_delivery_trace.py tests/test_selected_stock_lot_production_issue.py -q
```
Expected: PASS.
- [ ] **Step 6: Checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend/app/api/routes/dashboard.py backend/app/schemas/dashboard.py
git commit -m "refactor: show inventory lot trace in dashboard dag"
```
---
### Task 13: Final Full Verification
**Files:**
- No new files.
- Validate all modified backend and frontend files.
- [ ] **Step 1: Run full backend test suite**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
../backend/.venv/bin/python -m pytest tests -q
```
Expected: All tests PASS.
- [ ] **Step 2: Run frontend build**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build
```
Expected: Build PASS. Existing Vite chunk warnings are acceptable; syntax/type errors are not.
- [ ] **Step 3: Run terminology scan**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
rg -n "材料分批次|原材料分批次|材料分批号|来源分批号" frontend/src backend/app
```
Expected: Only backend compatibility field names, migration comments, and intentional fallback logic remain. No user-facing Chinese labels should remain.
- [ ] **Step 4: Run raw material quantity regression scan**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
rg -n "原材料.*数量|收货数量|领用数量|remaining_qty|received_qty|issued_qty" frontend/src/views backend/app/api backend/app/services
```
Expected: Raw material UI does not show quantity concepts. Backend compatibility decimal fields may remain but should be written as zero for raw material flows.
- [ ] **Step 5: Manual smoke scenario**
Use the local ERP UI and run this business scenario:
```text
1. 新建或选择一个原材料。
2. 新建采购订单，选择原材料库、备料采购，采购重量 500kg，锁单。
3. 创建第 1 张到货入库单，入库 200kg，单价默认采购单价，可手动修改。
4. 创建第 2 张到货入库单，入库 300kg。
5. 确认原材料库详情显示 JH_原材料编码_0001 和 JH_原材料编码_0002。
6. 生产出库时搜索该原材料，选择两个库存批次，分别填写领用重量。
7. 确认工序报工和成品入库能按库存批次号追溯。
8. 创建销售订单采购，关联两个销售订单，确认采购来源小卡片展示两个销售订单。
9. 打开 DAG，确认链路为库存批次号 -> 生产出库/报工 -> 成品批次 -> 发货。
```
Expected: No “材料分批次号”用户概念出现，所有追溯都围绕库存批次号。
- [ ] **Step 6: Final checkpoint**
Run:
```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short || true
```
If git is available:
```bash
git add backend frontend docs/superpowers/plans/2026-05-31-inventory-lot-unified-trace.md
git commit -m "feat: unify raw material trace by inventory lot"
```
---
## Self-Review
**Spec coverage:** Covered removal of “材料分批次号” as a user concept, new `原材料物料 -> 库存批次号` hierarchy, `JH_材料号_0001` numbering, split purchase arrivals, final receipt unit cost, selected multi-lot production issue, disabled depleted lots, miniapp field remapping, sales order multi-select for purchase orders, and downstream DAG/trace updates.
**Placeholder scan:** This plan contains no `TBD`, no unresolved `TODO`, and no intentionally vague “handle errors” step. Compatibility cases are spelled out.
**Type consistency:** Backend payload names use `sales_order_ids`, `selected_stock_lots`, `source_lot_id`, and `issued_weight_kg`. Frontend payload names match those schemas. Legacy `source_material_sub_batch_no` remains only as compatibility alias.
Reference in New Issue View Git Blame Copy Permalink