2026-06-12 16:00:56 +08:00

74 KiB

Raw Permalink Blame History

Production Batch Ledger And Smart Reporting Toggle 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 production execution core from “生产工单” to “生产台账”, where each row is uniquely identified by 材料库存批次号 + 产品, while keeping the intelligent operation-reporting miniapp as an optional integration controlled by a system switch.

Architecture: Add a new production ledger aggregate table keyed by material_lot_id + product_item_id, with an internal numeric ID for stable references and a ledger transaction table for every production issue, finished inbound, surplus inbound, scrap inbound, material lock, reopen, and write-off event. New production flows write to this ledger instead of creating new pp_work_order rows; old work-order tables remain for historical compatibility and rework flows until a separate cleanup project. The system switch 对接智能报工小程序 controls whether miniapp operation reports drive progress calculations and UI visibility; when disabled, progress is inferred from ERP inbound data and BOM gross/net weights.

Tech Stack: FastAPI, SQLAlchemy, Pydantic, MySQL patch SQL, Vue 3, Vite, Node static UI checks, Python unittest/pytest, miniapp backend FastAPI service under /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint.

Business Rules Locked In

Production ledger business key:

材料库存批次号 + 产品

Physical database primary key:

production_ledger_id

Unique database constraint:

material_lot_id + product_item_id

Ledger statuses are Chinese user/business values:

在生产
锁单

“该批材料结单” means this material lot is no longer outside the warehouse for this product. Any remaining outside weight is confirmed by a write-off ledger transaction and the row status becomes 锁单.
If the same 材料库存批次号 + 产品 is issued again after locking, the same ledger row reopens to 在生产, the new issue weight is accumulated, and a 重新开工 transaction is recorded.
Miniapp dropdown source is active production ledger rows, not pp_work_order. The miniapp shows material stock lot numbers for rows where 状态 = 在生产 and product matches the scanned product/process context.
When the smart reporting switch is enabled, operation reports are linked to production ledger rows and drive operation-detail progress.
When the smart reporting switch is disabled, 工序报工 UI is hidden and production calculations use ERP data:

理论已用料 = 已入库成品数量 × BOM毛重
理论废料 = 已入库成品数量 × (BOM毛重 - BOM净重)
理论余料 = 累计领料重量 - 理论已用料

Scrap rate is not separately modeled in disabled mode; it is handled through the 15% deviation explanation mechanism.

File Structure

Create: backend/sql/add_production_batch_ledger_patch.sql
- Create pp_production_batch_ledger and pp_production_batch_ledger_txn.
- Add optional production_ledger_id columns to existing ERP operation report tables for compatibility.
- Seed SMART_OPERATION_REPORT_ENABLED config as enabled.
Modify: backend/app/models/operations.py
- Add ProductionBatchLedger and ProductionBatchLedgerTxn.
- Add nullable production_ledger_id to OperationReport.
Modify: backend/app/schemas/operations.py
- Add read/create schemas for production ledger, ledger transactions, ledger inbound preview, ledger inbound create, and material lock request.
Modify: backend/app/services/system_config.py
- Add SMART_OPERATION_REPORT_ENABLED helpers.
Modify: backend/app/schemas/database.py
- Add SmartOperationReportConfigRead and SmartOperationReportConfigUpdate.
Modify: backend/app/api/routes/system_extension.py
- Add switch GET/PUT endpoints.
Create: backend/app/services/production_batch_ledger.py
- Own all new production ledger aggregation, reopen, lock, progress, and inbound calculations.
Modify: backend/app/api/routes/production.py
- Add new production ledger endpoints.
- Change production issue endpoint to upsert ledger rows instead of creating new pp_work_order rows.
- Keep legacy work-order endpoints as read-compatible routes while frontend migrates.
- Bind miniapp report sync to production ledger IDs.
Modify: backend/app/api/routes/inventory.py
- Change production surplus and scrap branch inbound to select and update production ledger.
- Keep rework work-order logic untouched.
Modify: backend/app/services/production_work_order_ledger.py
- Keep only as legacy compatibility, delegating normal production ledger reads to the new service if needed.
Modify: backend/app/services/production_work_order_inbound.py
- Keep legacy compatibility, but normal unified production inbound should call the new production ledger service.
Create: backend/tests/test_production_batch_ledger.py
- Covers issue accumulation, reopen, lock, write-off, enabled/disabled calculations.
Create: backend/tests/test_smart_operation_report_config.py
- Covers system switch defaults and updates.
Modify: backend/tests/test_selected_stock_lot_production_issue.py
- Update expectations from work-order creation to production-ledger upsert.
Modify: backend/tests/test_production_work_order_inbound.py
- Split legacy tests and add new ledger inbound tests, or migrate assertions to production ledger APIs.
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/app/routers/reports.py
- Change material batch options SQL from pp_work_order to pp_production_batch_ledger.
Modify: /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_work_order_batch_options.py
- Rename test intent to production ledger material batch options.
Modify: frontend/src/router/index.js
- Add production-ledger route and redirect old work-order-ledger to it.
- Hide direct operation-reports when smart reporting switch is disabled.
Modify: frontend/src/App.vue
- Rename menu 工单台账 to 生产台账.
- Hide 工序报工 when disabled.
Create: frontend/src/views/ProductionLedgerView.vue
- Replace WorkOrderManagementView.vue as the main production ledger page.
Modify: frontend/src/views/ProductionWorkspaceView.vue
- Rename and explain the new production ledger concept.
Modify: frontend/src/views/InventoryLedgerView.vue
- Rename 生产工单入库 to 生产台账入库.
- Branch selectors use production ledger rows.
- Main close action becomes 该批材料结单.
Modify: frontend/src/views/SystemExtensionView.vue
- Add 对接智能报工小程序 switch.
Create: frontend/src/services/systemFeatures.js
- Cache feature switch config.
Create: frontend/scripts/test-production-batch-ledger-ui.mjs
- Static checks for renamed UI and route migration.
Create: frontend/scripts/test-smart-operation-report-toggle.mjs
- Static checks for switch and hidden miniapp UI.

Task 1: Add Database Patch And ORM Models

Files:

Create: backend/sql/add_production_batch_ledger_patch.sql
Modify: backend/app/models/operations.py
Modify: backend/app/schemas/operations.py
Test: backend/tests/test_production_batch_ledger.py
Step 1: Write failing model test

Create backend/tests/test_production_batch_ledger.py:

from __future__ import annotations

import unittest
from datetime import datetime
from decimal import Decimal

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


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


import app.models.master_data  # noqa: E402,F401
import app.models.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 ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot  # noqa: E402


class ProductionBatchLedgerModelTest(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_models_persist_chinese_status_and_business_key(self) -> None:
        raw_item = Item(id=1, item_code="RM-001", item_name="冷轧钢板", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE")
        product = Item(id=2, item_code="FG-001", item_name="钢制碗", item_type="FINISHED_GOOD", unit_weight_kg=Decimal("0.5"), status="ACTIVE")
        warehouse = Warehouse(id=1, warehouse_code="WH-RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE")
        lot = StockLot(
            id=1,
            lot_no="YL0001",
            lot_role="RAW_MATERIAL",
            item_id=1,
            warehouse_id=1,
            inbound_qty=Decimal("0"),
            inbound_weight_kg=Decimal("500"),
            remaining_qty=Decimal("0"),
            remaining_weight_kg=Decimal("300"),
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=Decimal("4"),
            quality_status="PASS",
            status="AVAILABLE",
        )
        self.db.add_all([raw_item, product, warehouse, lot])
        self.db.flush()

        ledger = ProductionBatchLedger(
            material_lot_id=1,
            material_lot_no="YL0001",
            material_item_id=1,
            product_item_id=2,
            status="在生产",
            miniapp_selectable=1,
            total_issued_weight_kg=Decimal("200"),
            outside_weight_kg=Decimal("200"),
            finished_inbound_qty=Decimal("0"),
            surplus_inbound_weight_kg=Decimal("0"),
            scrap_inbound_weight_kg=Decimal("0"),
            writeoff_weight_kg=Decimal("0"),
            first_issue_time=datetime(2026, 6, 8, 8, 30),
            last_issue_time=datetime(2026, 6, 8, 8, 30),
        )
        self.db.add(ledger)
        self.db.flush()
        self.db.add(
            ProductionBatchLedgerTxn(
                production_ledger_id=ledger.id,
                txn_type="生产出库",
                qty_delta=Decimal("0"),
                weight_delta_kg=Decimal("200"),
                outside_weight_after_kg=Decimal("200"),
                source_doc_type="库存流水",
                source_doc_id=10,
                biz_time=datetime(2026, 6, 8, 8, 30),
                operator_user_id=1,
                remark="生产出库累计到生产台账",
            )
        )
        self.db.commit()

        saved = self.db.scalar(select(ProductionBatchLedger).where(ProductionBatchLedger.material_lot_no == "YL0001"))
        self.assertEqual(saved.status, "在生产")
        self.assertEqual(saved.miniapp_selectable, 1)
        self.assertEqual(float(saved.outside_weight_kg), 200)
        txns = self.db.scalars(select(ProductionBatchLedgerTxn)).all()
        self.assertEqual(txns[0].txn_type, "生产出库")

Step 2: Run model test and verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py -q

Expected: FAIL because ProductionBatchLedger and ProductionBatchLedgerTxn do not exist.

Step 3: Add SQL patch

Create backend/sql/add_production_batch_ledger_patch.sql:

CREATE TABLE IF NOT EXISTS pp_production_batch_ledger (
  id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '生产台账ID',
  material_lot_id BIGINT NOT NULL COMMENT '材料库存批次ID',
  material_lot_no VARCHAR(128) NOT NULL COMMENT '材料库存批次号',
  material_item_id BIGINT NOT NULL COMMENT '原材料物料ID',
  product_item_id BIGINT NOT NULL COMMENT '产品物料ID',
  status VARCHAR(32) NOT NULL DEFAULT '在生产' COMMENT '状态：在生产/锁单',
  miniapp_selectable TINYINT(1) NOT NULL DEFAULT 1 COMMENT '小程序是否可选',
  total_issued_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '累计领料重量',
  outside_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '当前库外未闭环重量',
  finished_inbound_qty DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '已入库成品数量',
  surplus_inbound_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '已归还余料重量',
  scrap_inbound_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '已入库废料重量',
  writeoff_weight_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '结单核销重量',
  first_issue_time DATETIME NULL COMMENT '首次生产出库时间',
  last_issue_time DATETIME NULL COMMENT '最近生产出库时间',
  locked_at DATETIME NULL COMMENT '最近锁单时间',
  reopened_at DATETIME NULL COMMENT '最近重新开工时间',
  lock_count INT NOT NULL DEFAULT 0 COMMENT '锁单次数',
  reopen_count INT NOT NULL DEFAULT 0 COMMENT '重新开工次数',
  remark VARCHAR(500) NULL COMMENT '备注',
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  UNIQUE KEY uk_production_batch_ledger_lot_product (material_lot_id, product_item_id),
  KEY idx_production_batch_ledger_status (status, miniapp_selectable),
  KEY idx_production_batch_ledger_product (product_item_id),
  CONSTRAINT fk_production_batch_ledger_lot FOREIGN KEY (material_lot_id) REFERENCES wh_stock_lot(id),
  CONSTRAINT fk_production_batch_ledger_material FOREIGN KEY (material_item_id) REFERENCES md_item(id),
  CONSTRAINT fk_production_batch_ledger_product FOREIGN KEY (product_item_id) REFERENCES md_item(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='生产台账：材料库存批次号+产品';

CREATE TABLE IF NOT EXISTS pp_production_batch_ledger_txn (
  id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '生产台账流水ID',
  production_ledger_id BIGINT NOT NULL COMMENT '生产台账ID',
  txn_type VARCHAR(32) NOT NULL COMMENT '流水类型：生产出库/成品入库/生产余料入库/生产废料入库/该批材料结单/重新开工/结单核销',
  qty_delta DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '数量变化',
  weight_delta_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '重量变化',
  outside_weight_after_kg DECIMAL(18,6) NOT NULL DEFAULT 0 COMMENT '流水后库外未闭环重量',
  source_doc_type VARCHAR(64) NULL COMMENT '来源单据类型',
  source_doc_id BIGINT NULL COMMENT '来源单据ID',
  source_line_id BIGINT NULL COMMENT '来源单据明细ID',
  biz_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '业务时间',
  operator_user_id BIGINT NULL COMMENT '操作人ID',
  remark VARCHAR(500) NULL COMMENT '备注',
  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  KEY idx_production_batch_ledger_txn_ledger (production_ledger_id, biz_time),
  KEY idx_production_batch_ledger_txn_type (txn_type),
  CONSTRAINT fk_production_batch_ledger_txn_ledger FOREIGN KEY (production_ledger_id) REFERENCES pp_production_batch_ledger(id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='生产台账流水';

ALTER TABLE pp_operation_report
  ADD COLUMN production_ledger_id BIGINT NULL COMMENT '生产台账ID' AFTER work_order_operation_id;

INSERT INTO sys_system_config (config_code, config_name, config_value, remark, status)
SELECT
  'SMART_OPERATION_REPORT_ENABLED',
  '对接智能报工小程序',
  '开启',
  '开启后显示工序报工并按小程序报工数据计算生产进度；关闭后按ERP入库数据反推生产进度',
  'ACTIVE'
FROM DUAL
WHERE NOT EXISTS (
  SELECT 1 FROM sys_system_config WHERE config_code = 'SMART_OPERATION_REPORT_ENABLED'
);

Use this duplicate-column-safe form on MySQL if the simple ALTER TABLE fails because the column already exists:

SET @column_exists := (
  SELECT COUNT(*)
  FROM information_schema.COLUMNS
  WHERE TABLE_SCHEMA = DATABASE()
    AND TABLE_NAME = 'pp_operation_report'
    AND COLUMN_NAME = 'production_ledger_id'
);
SET @ddl := IF(
  @column_exists = 0,
  'ALTER TABLE pp_operation_report ADD COLUMN production_ledger_id BIGINT NULL COMMENT ''生产台账ID'' AFTER work_order_operation_id',
  'SELECT ''pp_operation_report.production_ledger_id already exists'''
);
PREPARE stmt FROM @ddl;
EXECUTE stmt;
DEALLOCATE PREPARE stmt;

Step 4: Add ORM models

Modify backend/app/models/operations.py:

class ProductionBatchLedger(Base):
    __tablename__ = "pp_production_batch_ledger"
    __table_args__ = (UniqueConstraint("material_lot_id", "product_item_id", name="uk_production_batch_ledger_lot_product"),)

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    material_lot_id: Mapped[int] = mapped_column(ForeignKey("wh_stock_lot.id"))
    material_lot_no: Mapped[str] = mapped_column(String(128))
    material_item_id: Mapped[int] = mapped_column(ForeignKey("md_item.id"))
    product_item_id: Mapped[int] = mapped_column(ForeignKey("md_item.id"))
    status: Mapped[str] = mapped_column(String(32), server_default=text("'在生产'"))
    miniapp_selectable: Mapped[int] = mapped_column(Integer, server_default=text("1"))
    total_issued_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    outside_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    finished_inbound_qty: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    surplus_inbound_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    scrap_inbound_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    writeoff_weight_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    first_issue_time: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    last_issue_time: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    locked_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    reopened_at: Mapped[object | None] = mapped_column(DateTime, nullable=True)
    lock_count: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    reopen_count: Mapped[int] = mapped_column(Integer, server_default=text("0"))
    remark: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)
    updated_at: Mapped[object] = mapped_column(DateTime)


class ProductionBatchLedgerTxn(Base):
    __tablename__ = "pp_production_batch_ledger_txn"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    production_ledger_id: Mapped[int] = mapped_column(ForeignKey("pp_production_batch_ledger.id"))
    txn_type: Mapped[str] = mapped_column(String(32))
    qty_delta: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    weight_delta_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    outside_weight_after_kg: Mapped[float] = mapped_column(DECIMAL(18, 6), server_default=text("0"))
    source_doc_type: Mapped[str | None] = mapped_column(String(64), nullable=True)
    source_doc_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    source_line_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    biz_time: Mapped[object] = mapped_column(DateTime)
    operator_user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
    remark: Mapped[str | None] = mapped_column(String(500), nullable=True)
    created_at: Mapped[object] = mapped_column(DateTime)

Add to OperationReport:

    production_ledger_id: Mapped[int | None] = mapped_column(ForeignKey("pp_production_batch_ledger.id"), nullable=True)

Step 5: Add schemas

Modify backend/app/schemas/operations.py:

class ProductionBatchLedgerTxnRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    production_ledger_id: int
    txn_type: str
    qty_delta: float = 0
    weight_delta_kg: float = 0
    outside_weight_after_kg: float = 0
    source_doc_type: str | None = None
    source_doc_id: int | None = None
    source_line_id: int | None = None
    biz_time: datetime
    operator_user_id: int | None = None
    remark: str | None = None


class ProductionBatchLedgerRead(BaseModel):
    production_ledger_id: int
    material_lot_id: int
    material_lot_no: str
    material_item_id: int
    material_code: str | None = None
    material_name: str | None = None
    product_item_id: int
    product_code: str | None = None
    product_name: str | None = None
    status: str
    miniapp_selectable: bool = True
    total_issued_weight_kg: float = 0
    outside_weight_kg: float = 0
    finished_inbound_qty: float = 0
    surplus_inbound_weight_kg: float = 0
    scrap_inbound_weight_kg: float = 0
    writeoff_weight_kg: float = 0
    first_issue_time: datetime | None = None
    last_issue_time: datetime | None = None
    locked_at: datetime | None = None
    reopened_at: datetime | None = None
    lock_count: int = 0
    reopen_count: int = 0
    remark: str | None = None
    txns: list[ProductionBatchLedgerTxnRead] = []
    operations: list[WorkOrderLedgerOperationRow] = []

Use existing operation-row schema names if they differ; keep response fields readable for frontend.

Step 6: Run model test

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py -q
python3 -m py_compile backend/app/models/operations.py backend/app/schemas/operations.py

Expected: PASS.

Step 7: Commit

git add backend/sql/add_production_batch_ledger_patch.sql backend/app/models/operations.py backend/app/schemas/operations.py backend/tests/test_production_batch_ledger.py
git commit -m "feat: add production batch ledger schema"

Task 2: Build Production Ledger Service

Files:

Create: backend/app/services/production_batch_ledger.py
Modify: backend/tests/test_production_batch_ledger.py
Step 1: Add failing service tests

Append to backend/tests/test_production_batch_ledger.py:

    def test_record_issue_accumulates_same_lot_and_product(self) -> None:
        from app.services.production_batch_ledger import record_production_issue_to_ledger

        self._seed_basic_items_and_lot()
        first = record_production_issue_to_ledger(
            self.db,
            material_lot_id=1,
            product_item_id=2,
            issued_weight_kg=Decimal("120"),
            operator_user_id=1,
            source_doc_type="库存流水",
            source_doc_id=100,
            source_line_id=1001,
            biz_time=datetime(2026, 6, 8, 9, 0),
            remark="第一次出库",
        )
        second = record_production_issue_to_ledger(
            self.db,
            material_lot_id=1,
            product_item_id=2,
            issued_weight_kg=Decimal("80"),
            operator_user_id=1,
            source_doc_type="库存流水",
            source_doc_id=101,
            source_line_id=1002,
            biz_time=datetime(2026, 6, 8, 14, 0),
            remark="第二次出库",
        )
        self.db.commit()

        self.assertEqual(first.id, second.id)
        self.assertEqual(second.status, "在生产")
        self.assertEqual(float(second.total_issued_weight_kg), 200)
        self.assertEqual(float(second.outside_weight_kg), 200)
        txns = self.db.scalars(select(ProductionBatchLedgerTxn).order_by(ProductionBatchLedgerTxn.id)).all()
        self.assertEqual([row.txn_type for row in txns], ["生产出库", "生产出库"])

    def test_lock_and_reopen_same_lot_product(self) -> None:
        from app.services.production_batch_ledger import lock_production_batch_ledger, record_production_issue_to_ledger

        self._seed_basic_items_and_lot()
        ledger = record_production_issue_to_ledger(
            self.db,
            material_lot_id=1,
            product_item_id=2,
            issued_weight_kg=Decimal("200"),
            operator_user_id=1,
            source_doc_type="库存流水",
            source_doc_id=100,
            source_line_id=1001,
            biz_time=datetime(2026, 6, 8, 9, 0),
            remark="生产出库",
        )
        locked = lock_production_batch_ledger(
            self.db,
            production_ledger_id=ledger.id,
            operator_user_id=1,
            biz_time=datetime(2026, 6, 8, 18, 0),
            remark="现场确认该批材料已无库外余量",
        )
        self.assertEqual(locked.status, "锁单")
        self.assertEqual(locked.miniapp_selectable, 0)
        self.assertEqual(float(locked.outside_weight_kg), 0)
        self.assertEqual(float(locked.writeoff_weight_kg), 200)

        reopened = record_production_issue_to_ledger(
            self.db,
            material_lot_id=1,
            product_item_id=2,
            issued_weight_kg=Decimal("50"),
            operator_user_id=1,
            source_doc_type="库存流水",
            source_doc_id=102,
            source_line_id=1003,
            biz_time=datetime(2026, 6, 9, 9, 0),
            remark="继续出库生产",
        )

        self.assertEqual(reopened.id, locked.id)
        self.assertEqual(reopened.status, "在生产")
        self.assertEqual(reopened.miniapp_selectable, 1)
        self.assertEqual(float(reopened.total_issued_weight_kg), 250)
        self.assertEqual(float(reopened.outside_weight_kg), 50)
        txns = self.db.scalars(select(ProductionBatchLedgerTxn).order_by(ProductionBatchLedgerTxn.id)).all()
        self.assertEqual([row.txn_type for row in txns], ["生产出库", "结单核销", "该批材料结单", "重新开工", "生产出库"])

Add helper inside the test class:

    def _seed_basic_items_and_lot(self) -> None:
        raw_item = Item(id=1, item_code="RM-001", item_name="冷轧钢板", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE")
        product = Item(id=2, item_code="FG-001", item_name="钢制碗", item_type="FINISHED_GOOD", unit_weight_kg=Decimal("0.5"), status="ACTIVE")
        warehouse = Warehouse(id=1, warehouse_code="WH-RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE")
        lot = StockLot(
            id=1,
            lot_no="YL0001",
            lot_role="RAW_MATERIAL",
            item_id=1,
            warehouse_id=1,
            inbound_qty=Decimal("0"),
            inbound_weight_kg=Decimal("500"),
            remaining_qty=Decimal("0"),
            remaining_weight_kg=Decimal("300"),
            locked_qty=Decimal("0"),
            locked_weight_kg=Decimal("0"),
            unit_cost=Decimal("4"),
            quality_status="PASS",
            status="AVAILABLE",
        )
        self.db.add_all([raw_item, product, warehouse, lot])
        self.db.flush()

Step 2: Run tests and verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py -q

Expected: FAIL because production_batch_ledger.py service does not exist.

Step 3: Implement service

Create backend/app/services/production_batch_ledger.py:

from __future__ import annotations

from datetime import datetime
from decimal import Decimal, ROUND_FLOOR

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

from app.models.master_data import BomItem, Item
from app.models.operations import ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot
from app.services.operations import to_decimal


INBOUND_TOLERANCE_MULTIPLIER = Decimal("1.15")


def _round_weight(value: Decimal) -> Decimal:
    return to_decimal(value, "0.001")


def _non_negative(value: Decimal) -> Decimal:
    return value if value > 0 else Decimal("0")


def _append_txn(
    db: Session,
    *,
    ledger: ProductionBatchLedger,
    txn_type: str,
    qty_delta: Decimal = Decimal("0"),
    weight_delta_kg: Decimal = Decimal("0"),
    source_doc_type: str | None = None,
    source_doc_id: int | None = None,
    source_line_id: int | None = None,
    biz_time: datetime | None = None,
    operator_user_id: int | None = None,
    remark: str | None = None,
) -> ProductionBatchLedgerTxn:
    txn = ProductionBatchLedgerTxn(
        production_ledger_id=ledger.id,
        txn_type=txn_type,
        qty_delta=qty_delta,
        weight_delta_kg=weight_delta_kg,
        outside_weight_after_kg=to_decimal(ledger.outside_weight_kg),
        source_doc_type=source_doc_type,
        source_doc_id=source_doc_id,
        source_line_id=source_line_id,
        biz_time=biz_time or datetime.now(),
        operator_user_id=operator_user_id,
        remark=remark,
    )
    db.add(txn)
    return txn


def _get_locked_ledger(db: Session, material_lot_id: int, product_item_id: int) -> ProductionBatchLedger | None:
    return db.scalar(
        select(ProductionBatchLedger)
        .where(
            ProductionBatchLedger.material_lot_id == material_lot_id,
            ProductionBatchLedger.product_item_id == product_item_id,
        )
        .with_for_update()
    )


def record_production_issue_to_ledger(
    db: Session,
    *,
    material_lot_id: int,
    product_item_id: int,
    issued_weight_kg: Decimal,
    operator_user_id: int | None,
    source_doc_type: str | None,
    source_doc_id: int | None,
    source_line_id: int | None,
    biz_time: datetime | None = None,
    remark: str | None = None,
) -> ProductionBatchLedger:
    if issued_weight_kg <= 0:
        raise HTTPException(status_code=400, detail="生产出库重量必须大于0")
    lot = db.get(StockLot, material_lot_id)
    if not lot:
        raise HTTPException(status_code=404, detail="材料库存批次号不存在")
    product = db.get(Item, product_item_id)
    if not product:
        raise HTTPException(status_code=404, detail="产品不存在")
    now = biz_time or datetime.now()
    ledger = _get_locked_ledger(db, material_lot_id, product_item_id)
    if ledger and str(ledger.status or "") == "锁单":
        ledger.status = "在生产"
        ledger.miniapp_selectable = 1
        ledger.reopened_at = now
        ledger.reopen_count = int(ledger.reopen_count or 0) + 1
        _append_txn(
            db,
            ledger=ledger,
            txn_type="重新开工",
            source_doc_type=source_doc_type,
            source_doc_id=source_doc_id,
            source_line_id=source_line_id,
            biz_time=now,
            operator_user_id=operator_user_id,
            remark="锁单后再次生产出库，生产台账重新进入在生产状态",
        )
    if not ledger:
        ledger = ProductionBatchLedger(
            material_lot_id=material_lot_id,
            material_lot_no=lot.lot_no,
            material_item_id=lot.item_id,
            product_item_id=product_item_id,
            status="在生产",
            miniapp_selectable=1,
            total_issued_weight_kg=Decimal("0"),
            outside_weight_kg=Decimal("0"),
            finished_inbound_qty=Decimal("0"),
            surplus_inbound_weight_kg=Decimal("0"),
            scrap_inbound_weight_kg=Decimal("0"),
            writeoff_weight_kg=Decimal("0"),
            first_issue_time=now,
            last_issue_time=now,
            remark=remark,
        )
        db.add(ledger)
        db.flush()
    ledger.status = "在生产"
    ledger.miniapp_selectable = 1
    ledger.total_issued_weight_kg = _round_weight(to_decimal(ledger.total_issued_weight_kg) + issued_weight_kg)
    ledger.outside_weight_kg = _round_weight(to_decimal(ledger.outside_weight_kg) + issued_weight_kg)
    ledger.last_issue_time = now
    ledger.updated_at = now
    db.add(ledger)
    db.flush()
    _append_txn(
        db,
        ledger=ledger,
        txn_type="生产出库",
        weight_delta_kg=issued_weight_kg,
        source_doc_type=source_doc_type,
        source_doc_id=source_doc_id,
        source_line_id=source_line_id,
        biz_time=now,
        operator_user_id=operator_user_id,
        remark=remark or "生产出库累计到生产台账",
    )
    return ledger


def lock_production_batch_ledger(
    db: Session,
    *,
    production_ledger_id: int,
    operator_user_id: int | None,
    biz_time: datetime | None = None,
    remark: str | None = None,
) -> ProductionBatchLedger:
    ledger = db.get(ProductionBatchLedger, production_ledger_id)
    if not ledger:
        raise HTTPException(status_code=404, detail="生产台账不存在")
    now = biz_time or datetime.now()
    remaining = _round_weight(to_decimal(ledger.outside_weight_kg))
    if remaining > 0:
        ledger.writeoff_weight_kg = _round_weight(to_decimal(ledger.writeoff_weight_kg) + remaining)
        ledger.outside_weight_kg = Decimal("0")
        db.add(ledger)
        db.flush()
        _append_txn(
            db,
            ledger=ledger,
            txn_type="结单核销",
            weight_delta_kg=-remaining,
            biz_time=now,
            operator_user_id=operator_user_id,
            remark=remark or "该批材料结单时确认库外剩余重量已无现场存量",
        )
    ledger.status = "锁单"
    ledger.miniapp_selectable = 0
    ledger.locked_at = now
    ledger.lock_count = int(ledger.lock_count or 0) + 1
    ledger.updated_at = now
    db.add(ledger)
    db.flush()
    _append_txn(
        db,
        ledger=ledger,
        txn_type="该批材料结单",
        biz_time=now,
        operator_user_id=operator_user_id,
        remark=remark or "该批材料已结单，小程序不再显示该材料库存批次号",
    )
    return ledger

Step 4: Run service tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py -q
python3 -m py_compile backend/app/services/production_batch_ledger.py

Expected: PASS.

Step 5: Commit

git add backend/app/services/production_batch_ledger.py backend/tests/test_production_batch_ledger.py
git commit -m "feat: manage production ledger issue lock and reopen"

Task 3: Refactor Production Issue To Upsert Production Ledger

Files:

Modify: backend/app/api/routes/production.py
Modify: backend/tests/test_selected_stock_lot_production_issue.py
Modify: backend/tests/test_production_batch_ledger.py
Step 1: Write failing API/service behavior test

Update backend/tests/test_selected_stock_lot_production_issue.py so production issue asserts:

def test_production_issue_creates_or_updates_production_batch_ledger(...):
    response = client.post("/api/production/work-orders", json=payload)
    assert response.status_code == 200
    data = response.json()
    assert data["production_ledger_id"] > 0
    assert data["material_lot_no"] == "YL0001"
    assert data["product_name"] == "钢制碗"
    assert data["status"] == "在生产"
    assert data["total_issued_weight_kg"] == 200

If this test file is not client-based, add a direct route/service test using its existing setup style. Keep endpoint path /production/work-orders for one release as compatibility, but make the returned object a production ledger read model.

Step 2: Run test and verify failure

Run:

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

Expected: FAIL because route still creates pp_work_order.

Step 3: Change production issue route

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

from app.models.operations import ProductionBatchLedger
from app.schemas.operations import ProductionBatchLedgerRead
from app.services.production_batch_ledger import record_production_issue_to_ledger, production_batch_ledger_to_read

Update route decorator:

@router.post("/work-orders", response_model=ProductionBatchLedgerRead)

Inside create_work_order, after stock lot validation and after issue_selected_stock_lots or equivalent inventory deduction, call:

ledger = record_production_issue_to_ledger(
    db,
    material_lot_id=int(selected_lot.id),
    product_item_id=int(payload.product_item_id),
    issued_weight_kg=issue_weight,
    operator_user_id=context.user.id,
    source_doc_type="生产出库",
    source_doc_id=None,
    source_line_id=None,
    biz_time=biz_time,
    remark=payload.remark or "原材料库生产出库形成生产台账",
)
db.commit()
return production_batch_ledger_to_read(db, ledger.id)

Do not create a new WorkOrder for normal production issue. Rework flows still use WorkOrder.

Step 4: Preserve legacy endpoint names with new semantics

Add new endpoint alias:

@router.get("/production-ledger", response_model=list[ProductionBatchLedgerRead])
def list_production_batch_ledger(...):
    ...

Keep old endpoint:

@router.get("/work-order-ledger", response_model=list[ProductionBatchLedgerRead])
def list_work_order_ledger(...):
    return list_production_batch_ledger(...)

This lets frontend migrate route names without breaking old bookmarks immediately.

Step 5: Run focused tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_selected_stock_lot_production_issue.py backend/tests/test_production_batch_ledger.py -q

Expected: PASS.

Step 6: Commit

git add backend/app/api/routes/production.py backend/tests/test_selected_stock_lot_production_issue.py backend/tests/test_production_batch_ledger.py
git commit -m "feat: make production issue update batch ledger"

Task 4: Implement Ledger Progress Calculations And Inbound APIs

Files:

Modify: backend/app/services/production_batch_ledger.py
Modify: backend/app/api/routes/production.py
Modify: backend/app/api/routes/inventory.py
Modify: backend/app/schemas/operations.py
Modify: backend/tests/test_production_batch_ledger.py
Step 1: Add calculation tests

Append tests:

    def test_disabled_mode_calculates_from_finished_inbound_and_bom(self) -> None:
        from app.services.production_batch_ledger import calculate_production_batch_ledger_snapshot

        self._seed_basic_items_bom_and_lot()
        ledger = self._seed_ledger(total_issued_weight_kg=Decimal("200"))
        ledger.finished_inbound_qty = Decimal("100")
        snapshot = calculate_production_batch_ledger_snapshot(self.db, ledger.id, smart_reporting_enabled=False)

        self.assertEqual(float(snapshot.reference_finished_qty), 333)
        self.assertEqual(float(snapshot.used_material_weight_kg), 60)
        self.assertEqual(float(snapshot.theoretical_scrap_weight_kg), 10)
        self.assertEqual(float(snapshot.theoretical_surplus_weight_kg), 140)

    def test_lock_writes_off_remaining_outside_weight_after_inbounds(self) -> None:
        from app.services.production_batch_ledger import lock_production_batch_ledger, post_finished_inbound_to_ledger

        self._seed_basic_items_bom_and_lot()
        ledger = self._seed_ledger(total_issued_weight_kg=Decimal("200"), outside_weight_kg=Decimal("200"))
        post_finished_inbound_to_ledger(
            self.db,
            production_ledger_id=ledger.id,
            finished_qty=Decimal("100"),
            operator_user_id=1,
            source_doc_type="成品入库",
            source_doc_id=1,
            source_line_id=1,
            biz_time=datetime(2026, 6, 8, 12, 0),
            remark="成品入库",
        )
        locked = lock_production_batch_ledger(
            self.db,
            production_ledger_id=ledger.id,
            operator_user_id=1,
            biz_time=datetime(2026, 6, 8, 18, 0),
            remark="该批材料结单",
        )
        self.assertEqual(locked.status, "锁单")
        self.assertEqual(float(locked.outside_weight_kg), 0)

Step 2: Run tests and verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py -q

Expected: FAIL until calculation and inbound helpers exist.

Step 3: Add snapshot and inbound helpers

In backend/app/services/production_batch_ledger.py, add:

@dataclass
class ProductionBatchLedgerSnapshot:
    production_ledger_id: int
    material_lot_no: str
    product_item_id: int
    gross_weight_per_piece_kg: Decimal
    net_weight_per_piece_kg: Decimal
    total_issued_weight_kg: Decimal
    finished_inbound_qty: Decimal
    surplus_inbound_weight_kg: Decimal
    scrap_inbound_weight_kg: Decimal
    writeoff_weight_kg: Decimal
    outside_weight_kg: Decimal
    reference_finished_qty: Decimal
    used_material_weight_kg: Decimal
    theoretical_scrap_weight_kg: Decimal
    theoretical_surplus_weight_kg: Decimal

Implement:

def calculate_production_batch_ledger_snapshot(
    db: Session,
    production_ledger_id: int,
    *,
    smart_reporting_enabled: bool,
    selected_finished_qty: Decimal | None = None,
) -> ProductionBatchLedgerSnapshot:
    ledger = db.get(ProductionBatchLedger, production_ledger_id)
    if not ledger:
        raise HTTPException(status_code=404, detail="生产台账不存在")
    gross, net = _get_bom_weight_profile(db, ledger.product_item_id, ledger.material_item_id)
    total_issued = to_decimal(ledger.total_issued_weight_kg)
    finished_qty = to_decimal(ledger.finished_inbound_qty)
    basis_qty = finished_qty + (selected_finished_qty or Decimal("0"))
    reference_finished_qty = Decimal("0")
    if gross > 0:
        reference_finished_qty = (total_issued / gross).to_integral_value(rounding=ROUND_FLOOR)
    used_weight = _round_weight(basis_qty * gross)
    theoretical_scrap = _round_weight(_non_negative(basis_qty * _non_negative(gross - net) - to_decimal(ledger.scrap_inbound_weight_kg)))
    theoretical_surplus = _round_weight(_non_negative(total_issued - used_weight - to_decimal(ledger.surplus_inbound_weight_kg)))
    return ProductionBatchLedgerSnapshot(
        production_ledger_id=ledger.id,
        material_lot_no=ledger.material_lot_no,
        product_item_id=ledger.product_item_id,
        gross_weight_per_piece_kg=gross,
        net_weight_per_piece_kg=net,
        total_issued_weight_kg=total_issued,
        finished_inbound_qty=finished_qty,
        surplus_inbound_weight_kg=to_decimal(ledger.surplus_inbound_weight_kg),
        scrap_inbound_weight_kg=to_decimal(ledger.scrap_inbound_weight_kg),
        writeoff_weight_kg=to_decimal(ledger.writeoff_weight_kg),
        outside_weight_kg=to_decimal(ledger.outside_weight_kg),
        reference_finished_qty=reference_finished_qty,
        used_material_weight_kg=used_weight,
        theoretical_scrap_weight_kg=theoretical_scrap,
        theoretical_surplus_weight_kg=theoretical_surplus,
    )

Add helpers:

def post_finished_inbound_to_ledger(...): ...
def post_surplus_inbound_to_ledger(...): ...
def post_scrap_inbound_to_ledger(...): ...

Each helper updates the matching aggregate field, subtracts the correct weight from outside_weight_kg, writes a Chinese transaction type, and returns the refreshed ledger.

Step 4: Update APIs

Add in backend/app/api/routes/production.py:

@router.get("/production-ledger-inbounds/preview", response_model=ProductionBatchLedgerInboundPreviewRead)
def preview_production_batch_ledger_inbound(...): ...

@router.post("/production-ledger-inbounds", response_model=ProductionBatchLedgerInboundRead)
def create_production_batch_ledger_inbound(...): ...

Payload fields:

production_ledger_id: int
finished_qty: float = 0
surplus_weight_kg: float = 0
scrap_weight_kg: float = 0
lock_material_batch: bool = False
lock_confirmed: bool = False
deviation_remark: str | None = None
remark: str | None = None

If lock_material_batch is true, call lock_production_batch_ledger after posting inbound rows.

Step 5: Update branch entries

In backend/app/api/routes/inventory.py, for normal production surplus/scrap only:

Read payload.source_doc_type == "PRODUCTION_LEDGER".
Read payload.source_doc_id as production_ledger_id.
Call post_surplus_inbound_to_ledger or post_scrap_inbound_to_ledger.
If payload requests lock, call lock_production_batch_ledger.
Keep rework paths using WORK_ORDER.
Step 6: Run tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py backend/tests/test_production_work_order_inbound.py backend/tests/test_scrap_warehouse_flow.py -q

Expected: PASS after updating legacy tests to the new API where normal production is involved.

Step 7: Commit

git add backend/app/services/production_batch_ledger.py backend/app/api/routes/production.py backend/app/api/routes/inventory.py backend/app/schemas/operations.py backend/tests/test_production_batch_ledger.py backend/tests/test_production_work_order_inbound.py backend/tests/test_scrap_warehouse_flow.py
git commit -m "feat: calculate and post production ledger inbounds"

Task 5: Add Smart Operation Reporting Switch

Files:

Modify: backend/app/services/system_config.py
Modify: backend/app/schemas/database.py
Modify: backend/app/api/routes/system_extension.py
Create: backend/tests/test_smart_operation_report_config.py
Step 1: Write config tests

Create backend/tests/test_smart_operation_report_config.py:

from __future__ import annotations

import unittest

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.services.system_config import (  # noqa: E402
    get_smart_operation_report_enabled,
    normalize_enabled_config_value,
    upsert_smart_operation_report_config,
)


class SmartOperationReportConfigTest(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_defaults_enabled(self) -> None:
        self.assertTrue(get_smart_operation_report_enabled(self.db))

    def test_normalizes_chinese_values(self) -> None:
        self.assertEqual(normalize_enabled_config_value("开启"), "开启")
        self.assertEqual(normalize_enabled_config_value("关闭"), "关闭")
        self.assertEqual(normalize_enabled_config_value(True), "开启")
        self.assertEqual(normalize_enabled_config_value(False), "关闭")

    def test_upsert_changes_enabled_state(self) -> None:
        upsert_smart_operation_report_config(self.db, enabled=False, updated_by=1, remark="未购买小程序")
        self.db.commit()
        self.assertFalse(get_smart_operation_report_enabled(self.db))

Step 2: Run tests and verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_smart_operation_report_config.py -q

Expected: FAIL until helpers exist.

Step 3: Implement config helpers

Use Chinese persisted values:

SMART_OPERATION_REPORT_CONFIG_CODE = "SMART_OPERATION_REPORT_ENABLED"
SMART_OPERATION_REPORT_DEFAULT = "开启"


def normalize_enabled_config_value(value: object) -> str:
    if isinstance(value, bool):
        return "开启" if value else "关闭"
    normalized = str(value or "").strip().upper()
    if normalized in {"1", "TRUE", "YES", "ON", "ENABLED", "开启", "打开", "启用", "是"}:
        return "开启"
    if normalized in {"0", "FALSE", "NO", "OFF", "DISABLED", "关闭", "停用", "否"}:
        return "关闭"
    return SMART_OPERATION_REPORT_DEFAULT


def get_smart_operation_report_enabled(db: Session) -> bool:
    value = get_system_config_value(db, SMART_OPERATION_REPORT_CONFIG_CODE, SMART_OPERATION_REPORT_DEFAULT)
    return normalize_enabled_config_value(value) == "开启"


def upsert_smart_operation_report_config(...):
    return upsert_system_config(
        db,
        config_code=SMART_OPERATION_REPORT_CONFIG_CODE,
        config_name="对接智能报工小程序",
        config_value=normalize_enabled_config_value(enabled),
        remark=remark or "开启后显示工序报工并按小程序报工数据计算生产进度；关闭后按ERP入库数据反推生产进度",
        updated_by=updated_by,
        now=now,
    )

Step 4: Add API schemas and endpoints

Add to backend/app/schemas/database.py:

class SmartOperationReportConfigUpdate(BaseModel):
    enabled: bool = True
    remark: str | None = Field(default=None, max_length=500)


class SmartOperationReportConfigRead(BaseModel):
    config_code: str
    config_name: str
    enabled: bool
    config_value: str
    remark: str | None = None
    updated_at: datetime | None = None

Add endpoints in backend/app/api/routes/system_extension.py:

@router.get("/smart-operation-report-config", response_model=SmartOperationReportConfigRead)
def get_smart_operation_report_config(...): ...

@router.put("/smart-operation-report-config", response_model=SmartOperationReportConfigRead)
def save_smart_operation_report_config(...): ...

Step 5: Run tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_smart_operation_report_config.py -q
python3 -m py_compile backend/app/services/system_config.py backend/app/schemas/database.py backend/app/api/routes/system_extension.py

Expected: PASS.

Step 6: Commit

git add backend/app/services/system_config.py backend/app/schemas/database.py backend/app/api/routes/system_extension.py backend/tests/test_smart_operation_report_config.py
git commit -m "feat: add smart reporting integration switch"

Task 6: Refactor Miniapp Batch Options And ERP Report Sync

Files:

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
Modify: backend/app/api/routes/production.py
Modify: backend/tests/test_miniapp_operation_report_date_filter.py
Modify: backend/tests/test_production_batch_ledger.py
Step 1: Update miniapp backend tests

In /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint/tests/test_work_order_batch_options.py, replace setup tables from pp_work_order to pp_production_batch_ledger:

create table pp_production_batch_ledger (
  id integer primary key,
  material_lot_id integer,
  material_lot_no text,
  material_item_id integer,
  product_item_id integer,
  status text,
  miniapp_selectable integer
);

Expected assertions:

def test_raw_material_batch_options_returns_active_production_ledger_lot_numbers_for_product_process(monkeypatch):
    ...
    assert options == ["YL0001"]


def test_raw_material_batch_options_excludes_locked_ledgers(monkeypatch):
    ...
    assert options == ["YL0002"]

Step 2: Run miniapp backend test and verify failure

Run:

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

Expected: FAIL because _raw_material_batch_options still queries pp_work_order.

Step 3: Change miniapp dropdown SQL

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

select distinct ledger.material_lot_no
from pp_production_batch_ledger ledger
join md_item product on product.id = ledger.product_item_id
join md_process_route route on route.product_item_id = product.id and route.status = 'ACTIVE'
join md_process_route_operation op on op.route_id = route.id
where ledger.material_lot_no is not null
  and trim(ledger.material_lot_no) <> ''
  and ledger.status = '在生产'
  and coalesce(ledger.miniapp_selectable, 1) = 1
  and (:process_name = '' or op.operation_name = :process_name)
  and (
    (:product_name <> '' and :project_no <> '' and product.item_name = :product_name and product.item_code = :project_no)
    or (:product_name <> '' and :project_no = '' and product.item_name = :product_name)
    or (:product_name = '' and :project_no <> '' and product.item_code = :project_no)
  )
order by ledger.material_lot_no asc

Step 4: Bind ERP operation sync to production ledger

In backend/app/api/routes/production.py, replace _find_work_order_for_miniapp_item for normal production with _find_production_ledger_for_miniapp_item.

Rules:

raw_material_batch_no = miniapp submitted material stock lot number
product = matched ERP product from miniapp product name/project no
production ledger = pp_production_batch_ledger where material_lot_no = raw_material_batch_no and product_item_id = product.id and status = 在生产

Operation reports should write:

report.production_ledger_id = ledger.id
report.work_order_id = None
report.work_order_operation_id = None

Keep work_order_id for legacy/rework only.

Step 5: Add ERP sync test

Add to backend/tests/test_production_batch_ledger.py:

    def test_miniapp_report_resolution_uses_production_ledger_lot_and_product(self) -> None:
        from app.api.routes.production import _find_production_ledger_for_miniapp_item

        self._seed_basic_items_bom_and_lot()
        ledger = self._seed_ledger(total_issued_weight_kg=Decimal("200"), status="在生产")
        row = {
            "raw_material_batch_no": "YL0001",
            "product_name": "钢制碗",
            "project_no": "FG-001",
        }
        matched = _find_production_ledger_for_miniapp_item(self.db, row, cache=None)
        self.assertEqual(matched.id, ledger.id)

Step 6: Run miniapp and ERP tests

Run:

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

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest backend/tests/test_production_batch_ledger.py backend/tests/test_miniapp_operation_report_date_filter.py -q

Expected: PASS.

Step 7: Commit

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git add backend/app/api/routes/production.py backend/tests/test_production_batch_ledger.py backend/tests/test_miniapp_operation_report_date_filter.py
git commit -m "feat: bind miniapp reports to production ledger"

cd /Users/souplearn/Gitlab/app/JhHardwareWRS_BackPoint
git add app/routers/reports.py tests/test_work_order_batch_options.py
git commit -m "feat: source material batch options from ERP production ledger"

Task 7: Frontend Route And Page Migration

Files:

Create: frontend/src/services/systemFeatures.js
Create: frontend/src/views/ProductionLedgerView.vue
Modify: frontend/src/router/index.js
Modify: frontend/src/App.vue
Modify: frontend/src/views/ProductionWorkspaceView.vue
Modify: frontend/src/views/SystemExtensionView.vue
Create: frontend/scripts/test-production-batch-ledger-ui.mjs
Create: frontend/scripts/test-smart-operation-report-toggle.mjs
Step 1: Write static UI tests

Create frontend/scripts/test-production-batch-ledger-ui.mjs:

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

const root = process.cwd();
const read = (relativePath) => readFileSync(path.join(root, relativePath), "utf8");

const router = read("src/router/index.js");
const app = read("src/App.vue");
const productionLedger = read("src/views/ProductionLedgerView.vue");
const inventory = read("src/views/InventoryLedgerView.vue");

assert.match(router, /ProductionLedgerView/, "router should import ProductionLedgerView");
assert.match(router, /name:\s*"production-ledger"/, "router should expose production-ledger route");
assert.match(router, /path:\s*"\/work-order-ledger"[\s\S]*redirect:\s*\{\s*name:\s*"production-ledger"\s*\}/, "old work-order-ledger route should redirect to production-ledger");
assert.match(app, /生产台账/, "sidebar should show 生产台账");
assert.doesNotMatch(app, /工单台账/, "sidebar should no longer show 工单台账");
assert.match(productionLedger, /材料库存批次号/, "production ledger page should use material stock lot number");
assert.match(productionLedger, /产品/, "production ledger page should show product");
assert.match(productionLedger, /库外未闭环重量/, "production ledger page should show outside weight");
assert.match(productionLedger, /该批材料结单/, "production ledger page should expose material batch lock action");
assert.match(inventory, /生产台账入库/, "inventory unified inbound should be renamed to production ledger inbound");
assert.match(inventory, /该批材料结单/, "inventory close action should use material batch close wording");

console.log("production batch ledger UI static checks passed");

Create frontend/scripts/test-smart-operation-report-toggle.mjs:

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

const root = process.cwd();
const read = (relativePath) => readFileSync(path.join(root, relativePath), "utf8");

const featureService = read("src/services/systemFeatures.js");
const router = read("src/router/index.js");
const app = read("src/App.vue");
const systemExtension = read("src/views/SystemExtensionView.vue");
const productionLedger = read("src/views/ProductionLedgerView.vue");

assert.match(featureService, /smart-operation-report-config/, "feature service should load switch config");
assert.match(router, /operation-reports/, "router should still define operation report route");
assert.match(router, /loadSmartOperationReportConfig/, "router should guard operation-reports by switch");
assert.match(app, /smartOperationReportEnabled/, "App should track switch state");
assert.match(app, /child\.key\s*===\s*"operation-reports"/, "App should hide operation reports child when disabled");
assert.match(systemExtension, /对接智能报工小程序/, "system extension should show switch");
assert.match(productionLedger, /工序明细/, "production ledger page should show operation detail only when enabled");
assert.match(productionLedger, /当前未对接智能报工小程序/, "production ledger page should explain disabled mode");

console.log("smart operation report toggle static checks passed");

Step 2: Run static tests and verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-production-batch-ledger-ui.mjs
node scripts/test-smart-operation-report-toggle.mjs

Expected: FAIL until frontend files are migrated.

Step 3: Add feature service

Create frontend/src/services/systemFeatures.js:

import { fetchResource } from "./api";

let smartOperationReportConfigCache = null;
let smartOperationReportConfigPromise = null;

export async function loadSmartOperationReportConfig(force = false) {
  if (!force && smartOperationReportConfigCache) {
    return smartOperationReportConfigCache;
  }
  if (!force && smartOperationReportConfigPromise) {
    return smartOperationReportConfigPromise;
  }
  smartOperationReportConfigPromise = fetchResource("/system-extension/smart-operation-report-config", {
    enabled: true,
    config_value: "开启",
    config_name: "对接智能报工小程序",
    remark: ""
  }).then((config) => {
    smartOperationReportConfigCache = {
      ...config,
      enabled: config?.enabled !== false
    };
    smartOperationReportConfigPromise = null;
    return smartOperationReportConfigCache;
  });
  return smartOperationReportConfigPromise;
}

export function resetSmartOperationReportConfigCache(config = null) {
  smartOperationReportConfigCache = config;
  smartOperationReportConfigPromise = null;
}

Step 4: Create ProductionLedgerView

Create frontend/src/views/ProductionLedgerView.vue using the structure of WorkOrderManagementView.vue, but rename concepts:

<template>
  <section class="page-grid">
    <section class="panel">
      <div class="panel-header">
        <div>
          <p class="eyebrow">生产台账</p>
          <h3>按材料库存批次号和产品追踪仓库外生产流转</h3>
        </div>
        <span class="panel-tag">材料批次 + 产品</span>
      </div>
      <div v-if="!smartOperationReportEnabled" class="workflow-tip">
        当前未对接智能报工小程序，生产进度按成品入库数量和BOM毛重/净重反推。
      </div>
      <TableControls ... placeholder="搜索材料库存批次号、产品、状态" />
      <table class="data-table compact-table">
        <thead>
          <tr>
            <th>材料库存批次号</th>
            <th>产品</th>
            <th>累计领料重量</th>
            <th>库外未闭环重量</th>
            <th>已入库成品</th>
            <th>已归还余料</th>
            <th>已入库废料</th>
            <th>状态</th>
          </tr>
        </thead>
      </table>
    </section>
  </section>
</template>

Load rows from:

fetchResource("/production/production-ledger?limit=500", [])

Step 5: Update router and sidebar

Modify frontend/src/router/index.js:

import ProductionLedgerView from "../views/ProductionLedgerView.vue";
import { loadSmartOperationReportConfig } from "../services/systemFeatures";

Routes:

{
  path: "/production-ledger",
  name: "production-ledger",
  component: ProductionLedgerView,
  meta: { title: "生产台账", requiresAuth: true, permission: "MENU_WORK_ORDER" }
},
{
  path: "/work-order-ledger",
  name: "work-order-ledger",
  redirect: { name: "production-ledger" },
  meta: { title: "生产台账", requiresAuth: true, permission: "MENU_WORK_ORDER" }
}

Route guard:

if (to.name === "operation-reports") {
  const config = await loadSmartOperationReportConfig();
  if (config.enabled === false) {
    return { name: "production-ledger" };
  }
}

Modify frontend/src/App.vue child menu:

{
  key: "production-ledger",
  label: "生产台账",
  to: { name: "production-ledger" },
  routeNames: ["production-ledger", "work-order-ledger"],
  permission: "MENU_WORK_ORDER",
  ...
}

Step 6: Add SystemExtension switch UI

Add a panel in frontend/src/views/SystemExtensionView.vue:

<section class="panel">
  <div class="panel-header">
    <div>
      <p class="eyebrow">生产模式配置</p>
      <h3>对接智能报工小程序</h3>
    </div>
    <span class="panel-tag">{{ smartOperationReportForm.enabled ? "已对接" : "未对接" }}</span>
  </div>
  <form class="stack-form compact-extension-form" @submit.prevent="saveSmartOperationReportConfig">
    <label class="checkbox-field extension-switch-field">
      <input v-model="smartOperationReportForm.enabled" type="checkbox" />
      <span>对接智能报工小程序</span>
    </label>
    <div class="workflow-tip">
      {{ smartOperationReportForm.enabled ? "开启后显示工序报工，并按小程序报工数据计算生产进度。" : "关闭后隐藏工序报工，生产进度按ERP入库数据和BOM反推。" }}
    </div>
    <button class="primary-button" type="submit">保存生产模式</button>
  </form>
</section>

Step 7: Run frontend checks

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-production-batch-ledger-ui.mjs
node scripts/test-smart-operation-report-toggle.mjs
npm run build

Expected: PASS.

Step 8: Commit

git add frontend/src/services/systemFeatures.js frontend/src/views/ProductionLedgerView.vue frontend/src/router/index.js frontend/src/App.vue frontend/src/views/ProductionWorkspaceView.vue frontend/src/views/SystemExtensionView.vue frontend/scripts/test-production-batch-ledger-ui.mjs frontend/scripts/test-smart-operation-report-toggle.mjs
git commit -m "feat: migrate frontend to production batch ledger"

Task 8: Update Inventory Inbound UI To Production Ledger

Files:

Modify: frontend/src/views/InventoryLedgerView.vue
Modify: frontend/scripts/test-production-batch-ledger-ui.mjs
Modify: frontend/scripts/test-production-work-order-inbound-ui.mjs
Step 1: Extend static checks

Add to frontend/scripts/test-production-batch-ledger-ui.mjs:

assert.doesNotMatch(inventory, /生产工单入库/, "inventory should not show old production work-order inbound wording");
assert.doesNotMatch(inventory, /工单号/, "production ledger inbound summary should not use 工单号");
assert.match(inventory, /生产台账入库/, "inventory should show production ledger inbound");
assert.match(inventory, /生产台账/, "branch selectors should use production ledger wording");
assert.match(inventory, /该批材料结单/, "close action should lock material batch production ledger");
assert.match(inventory, /库外未闭环重量/, "summary should show outside unclosed weight");

Step 2: Run static test and verify failure

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-production-batch-ledger-ui.mjs

Expected: FAIL until inventory UI wording and payloads are migrated.

Step 3: Rename unified inbound drawer

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

生产工单入库 -> 生产台账入库
工单号 -> 生产台账
本次入库后结单 -> 本次入库后该批材料结单
保存生产工单入库 -> 保存生产台账入库

Preview endpoint:

fetchResource(`/production/production-ledger-inbounds/preview?production_ledger_id=${id}&finished_qty=${finishedQty}`, null)

Submit endpoint:

postResource("/production/production-ledger-inbounds", {
  production_ledger_id: Number(selectedProductionLedgerId.value),
  finished_qty: finishedQty,
  surplus_weight_kg: surplusWeight,
  scrap_weight_kg: scrapWeight,
  lock_material_batch: Boolean(productionLedgerInboundForm.lock_material_batch),
  lock_confirmed: Boolean(productionLedgerInboundForm.lock_material_batch),
  deviation_remark: productionLedgerInboundForm.deviation_remark || null,
  remark: productionLedgerInboundForm.remark || null
})

Step 4: Update branch payloads

For 原材料库 · 生产余料入库:

source_doc_type: "PRODUCTION_LEDGER",
source_doc_id: Number(selectedProductionLedger.value.production_ledger_id)

For 废料库 · 生产废料入库:

source_doc_type: "PRODUCTION_LEDGER",
source_doc_id: Number(selectedProductionLedger.value.production_ledger_id)

Close checkbox label:

该批材料结单

Confirmation text:

确认后该材料库存批次号将从小程序待选列表中剔除；系统会把当前库外未闭环重量作为结单核销记录。

Step 5: Run frontend checks

Run:

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

Expected: PASS after updating old test names or replacing the old script with production ledger checks.

Step 6: Commit

git add frontend/src/views/InventoryLedgerView.vue frontend/scripts/test-production-batch-ledger-ui.mjs frontend/scripts/test-production-work-order-inbound-ui.mjs
git commit -m "feat: use production ledger for warehouse production inbound"

Task 9: Compatibility And Data Migration

Files:

Create: backend/sql/migrate_work_orders_to_production_batch_ledger.sql
Modify: backend/app/api/routes/production.py
Test: manual SQL verification
Step 1: Create migration SQL

Create backend/sql/migrate_work_orders_to_production_batch_ledger.sql:

INSERT INTO pp_production_batch_ledger (
  material_lot_id,
  material_lot_no,
  material_item_id,
  product_item_id,
  status,
  miniapp_selectable,
  total_issued_weight_kg,
  outside_weight_kg,
  finished_inbound_qty,
  surplus_inbound_weight_kg,
  scrap_inbound_weight_kg,
  writeoff_weight_kg,
  first_issue_time,
  last_issue_time,
  remark
)
SELECT
  issue.source_lot_id,
  lot.lot_no,
  issue.material_item_id,
  wo.product_item_id,
  CASE WHEN UPPER(COALESCE(wo.status, '')) = 'SETTLED' THEN '锁单' ELSE '在生产' END,
  CASE WHEN UPPER(COALESCE(wo.status, '')) = 'SETTLED' THEN 0 ELSE 1 END,
  SUM(COALESCE(issue.issued_weight_kg, 0)),
  GREATEST(SUM(COALESCE(issue.issued_weight_kg, 0)) - COALESCE(MAX(wom.returned_weight_kg), 0), 0),
  COALESCE(MAX(wo.finished_qty), 0),
  COALESCE(MAX(wom.returned_weight_kg), 0),
  0,
  0,
  MIN(issue.issue_time),
  MAX(issue.issue_time),
  '由历史生产工单迁移生成'
FROM pp_work_order wo
JOIN pp_work_order_material_issue issue ON issue.work_order_id = wo.id
JOIN wh_stock_lot lot ON lot.id = issue.source_lot_id
LEFT JOIN pp_work_order_material wom ON wom.work_order_id = wo.id AND wom.material_item_id = issue.material_item_id
WHERE wo.work_order_type = 'NORMAL'
GROUP BY issue.source_lot_id, lot.lot_no, issue.material_item_id, wo.product_item_id, wo.status
ON DUPLICATE KEY UPDATE
  total_issued_weight_kg = VALUES(total_issued_weight_kg),
  outside_weight_kg = VALUES(outside_weight_kg),
  finished_inbound_qty = VALUES(finished_inbound_qty),
  surplus_inbound_weight_kg = VALUES(surplus_inbound_weight_kg),
  status = VALUES(status),
  miniapp_selectable = VALUES(miniapp_selectable),
  remark = CONCAT(COALESCE(pp_production_batch_ledger.remark, ''), '；历史工单迁移已更新');

Follow with transaction inserts:

INSERT INTO pp_production_batch_ledger_txn (
  production_ledger_id,
  txn_type,
  qty_delta,
  weight_delta_kg,
  outside_weight_after_kg,
  source_doc_type,
  source_doc_id,
  source_line_id,
  biz_time,
  remark
)
SELECT
  ledger.id,
  '生产出库',
  0,
  issue.issued_weight_kg,
  ledger.outside_weight_kg,
  '历史生产工单',
  issue.work_order_id,
  issue.id,
  issue.issue_time,
  '由历史生产工单领料记录迁移'
FROM pp_work_order_material_issue issue
JOIN pp_work_order wo ON wo.id = issue.work_order_id
JOIN pp_production_batch_ledger ledger
  ON ledger.material_lot_id = issue.source_lot_id
 AND ledger.product_item_id = wo.product_item_id
WHERE wo.work_order_type = 'NORMAL'
  AND NOT EXISTS (
    SELECT 1
    FROM pp_production_batch_ledger_txn existing
    WHERE existing.production_ledger_id = ledger.id
      AND existing.source_doc_type = '历史生产工单'
      AND existing.source_doc_id = issue.work_order_id
      AND existing.source_line_id = issue.id
      AND existing.txn_type = '生产出库'
  );

Step 2: Run migration in test DB

Run:

mysql -h "$DB_HOST" -P "${DB_PORT:-3306}" -u "$DB_USER" -p"$DB_PASSWORD" --default-character-set=utf8mb4 "$DB_NAME" < /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend/sql/add_production_batch_ledger_patch.sql
mysql -h "$DB_HOST" -P "${DB_PORT:-3306}" -u "$DB_USER" -p"$DB_PASSWORD" --default-character-set=utf8mb4 "$DB_NAME" < /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend/sql/migrate_work_orders_to_production_batch_ledger.sql

Expected: both commands exit 0.

Step 3: Verify migration

Run:

SELECT material_lot_no, product_item_id, status, total_issued_weight_kg, outside_weight_kg
FROM pp_production_batch_ledger
ORDER BY id DESC
LIMIT 20;

Expected: rows exist for historical normal work-order issue data.

Step 4: Commit

git add backend/sql/migrate_work_orders_to_production_batch_ledger.sql
git commit -m "chore: add legacy work order to production ledger migration"

Task 10: Full Verification

Files:

Verify all touched backend, frontend, and miniapp backend files.
Step 1: Backend focused tests

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
python3 -m pytest \
  backend/tests/test_production_batch_ledger.py \
  backend/tests/test_smart_operation_report_config.py \
  backend/tests/test_selected_stock_lot_production_issue.py \
  backend/tests/test_production_work_order_inbound.py \
  backend/tests/test_scrap_warehouse_flow.py \
  backend/tests/test_miniapp_operation_report_date_filter.py \
  -q

Expected: PASS.

Step 2: Frontend static checks and build

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-production-batch-ledger-ui.mjs
node scripts/test-smart-operation-report-toggle.mjs
node scripts/test-operation-report-date-filter.mjs
npm run build

Expected: PASS. Vite chunk-size warnings are acceptable.

Step 3: Miniapp backend checks

Run:

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

Expected: PASS.

Step 4: Manual smoke checks

Start ERP frontend/backend locally and verify:

生产执行 menu shows 生产台账, not 工单台账.
生产台账 table rows are keyed by 材料库存批次号 + 产品. -原材料库生产出库 twice for the same material lot and product updates one ledger row.
该批材料结单 changes status to 锁单 and removes the material lot from miniapp dropdown.
Issuing the same material lot/product after locking reopens the same ledger row to 在生产.
When 对接智能报工小程序 is enabled, 工序报工 is visible and ledger detail shows 工序明细.
When the switch is disabled, 工序报工 is hidden and ledger detail hides 工序明细.
生产台账入库 calculates theoretical surplus/scrap immediately after finished quantity input.
Branch entries for production surplus and production scrap update the same production ledger row.
Step 5: Final status check

Run:

cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP
git status --short

Expected: only intentional files are modified.

Self-Review

The plan no longer treats pp_work_order as the normal production core.
The new core is pp_production_batch_ledger, uniquely representing 材料库存批次号 + 产品.
The old 工单台账 wording is replaced by 生产台账.
The old “工单结单” concept is replaced by 该批材料结单.
The miniapp dropdown source is changed from work orders to active production ledger rows.
The smart reporting switch remains part of the plan and controls UI/calculation mode.
Rework work orders are intentionally left on legacy pp_work_order because they are a separate return/rework lifecycle.

74 KiB Raw Permalink Blame History Unescape Escape

Production Batch Ledger And Smart Reporting Toggle Implementation Plan

Business Rules Locked In

File Structure

Task 1: Add Database Patch And ORM Models

Task 2: Build Production Ledger Service

Task 3: Refactor Production Issue To Upsert Production Ledger

Task 4: Implement Ledger Progress Calculations And Inbound APIs

Task 5: Add Smart Operation Reporting Switch

Task 6: Refactor Miniapp Batch Options And ERP Report Sync

Task 7: Frontend Route And Page Migration

Task 8: Update Inventory Inbound UI To Production Ledger

Task 9: Compatibility And Data Migration

Task 10: Full Verification

Self-Review

74 KiB

Raw Permalink Blame History