# Production Inbound Archive Entrypoints 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:** Expose preview/download/regenerate entrypoints for `生产入库结算单` in the places users naturally look after production ledger inbound creates multiple warehouse transactions. **Architecture:** Keep `生产台账入库` as one business settlement action and keep all generated warehouse transactions linked to the same `生产入库结算单` archive. The backend warehouse transaction ledger will return archive metadata for production settlement transactions; the frontend will reuse `DocumentArchiveActions` in the warehouse ledger and show a post-save shortcut when `生产台账入库` succeeds. **Tech Stack:** FastAPI, SQLAlchemy, Pydantic, existing `document_archives` service/routes, Vue 3, existing `DocumentArchiveActions`, existing `openResource`/`downloadResource`/`postResource`. --- ## File Structure ### Backend - Modify: `backend/app/schemas/operations.py` - Add archive metadata fields to `InventoryTxnRead`, inherited by `InventoryTxnLedgerRowRead`. - Modify: `backend/app/services/operations.py` - Add `DocumentArchive` and production archive constants imports if missing. - Add helper functions for production settlement archive grouping on warehouse inventory transactions. - Extend `get_inventory_txn_ledger_query` to select archive metadata for production settlement rows. - Test: `backend/tests/test_inventory_transaction_ledger_archive_fields.py` - Add a focused test that proves finished, surplus, and scrap rows from the same production settlement batch all expose the same archive business ID and status. ### Frontend - Modify: `frontend/src/views/InventoryLedgerView.vue` - Import `DocumentArchiveActions`. - Import `downloadResource` and `openResource` if not already imported. - Add a `单据留档` column to the warehouse流水 drawer. - Render `DocumentArchiveActions` for rows with `archive_business_id`. - Add archive preview/download/regenerate handlers. - Add a post-save shortcut panel after `生产台账入库` succeeds. - Modify: `frontend/src/views/InventoryLedgerView.test.js` - Add static assertions for the new warehouse ledger archive column, document type key, and post-save shortcut state. - Modify: `frontend/src/styles/main.css` - Add compact archive action styling for table cells and feedback shortcut panels. --- ## Current Reality To Preserve - `生产台账入库` already calls `POST /production/production-ledger-inbounds`. - That endpoint already returns: - `archive_status` - `archive_business_id` - `archive_document_type` - `archive_error_message` - `ProductionLedgerView.vue` already uses `DocumentArchiveActions` for `production-material-out` and `production-inbound-settlement`. - `InventoryLedgerView.vue` warehouse ledger currently has no archive column. - `GET /inventory/transaction-ledger` currently returns `InventoryTxnLedgerRowRead`, which does not yet include archive metadata. --- ### Task 1: Backend Test For Warehouse Ledger Archive Metadata **Files:** - Create: `backend/tests/test_inventory_transaction_ledger_archive_fields.py` - [ ] **Step 1: Write the failing test** Create `backend/tests/test_inventory_transaction_ledger_archive_fields.py`: ```python from __future__ import annotations import unittest from datetime import datetime 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_big_integer_for_sqlite(type_, compiler, **kw) -> str: _ = type_, compiler, kw return "INTEGER" import app.models.document_archive # noqa: E402,F401 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.document_archive import DocumentArchive # noqa: E402 from app.models.master_data import Item, Warehouse # noqa: E402 from app.models.operations import InventoryTxn, StockLot # noqa: E402 from app.services.document_archives import DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT # noqa: E402 from app.services.operations import get_inventory_txn_ledger_query # noqa: E402 class InventoryTransactionLedgerArchiveFieldsTest(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.now = datetime(2026, 6, 11, 14, 0, 0) self._seed_data() def tearDown(self) -> None: self.db.close() def _seed_data(self) -> None: finished_wh = Warehouse( id=1, warehouse_code="WH-FIN", warehouse_name="成品库", warehouse_type="FINISHED", status="ACTIVE", created_at=self.now, updated_at=self.now, ) raw_wh = Warehouse( id=2, warehouse_code="WH-RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE", created_at=self.now, updated_at=self.now, ) scrap_wh = Warehouse( id=3, warehouse_code="WH-SCRAP", warehouse_name="废料库", warehouse_type="SCRAP", status="ACTIVE", created_at=self.now, updated_at=self.now, ) material = Item( id=10, item_code="RM-010", item_name="测试原材料", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("0"), status="ACTIVE", created_at=self.now, updated_at=self.now, ) product = Item( id=20, item_code="FG-020", item_name="测试成品", item_type="FINISHED", unit_weight_kg=Decimal("0.5"), status="ACTIVE", created_at=self.now, updated_at=self.now, ) source_lot = StockLot( id=100, lot_no="YL0001", item_id=material.id, warehouse_id=raw_wh.id, location_id=None, inbound_qty=0, inbound_weight_kg=500, remaining_qty=0, remaining_weight_kg=200, locked_qty=0, locked_weight_kg=0, unit_cost=Decimal("6.5"), quality_status="PASS", status="AVAILABLE", created_at=self.now, updated_at=self.now, ) finished_lot = StockLot( id=101, lot_no="FGL0001", parent_lot_id=source_lot.id, item_id=product.id, warehouse_id=finished_wh.id, location_id=None, source_material_lot_id=source_lot.id, source_material_sub_batch_no="YL0001", inbound_qty=100, inbound_weight_kg=50, remaining_qty=100, remaining_weight_kg=50, locked_qty=0, locked_weight_kg=0, unit_cost=Decimal("3.25"), quality_status="PASS", status="AVAILABLE", created_at=self.now, updated_at=self.now, ) scrap_lot = StockLot( id=102, lot_no="SCRAPI0001", parent_lot_id=source_lot.id, item_id=material.id, warehouse_id=scrap_wh.id, location_id=None, source_material_lot_id=source_lot.id, source_material_sub_batch_no="YL0001", inbound_qty=0, inbound_weight_kg=5, remaining_qty=0, remaining_weight_kg=5, locked_qty=0, locked_weight_kg=0, unit_cost=Decimal("6.5"), quality_status="SCRAP", status="AVAILABLE", created_at=self.now, updated_at=self.now, ) batch_no = "生产结算20260611140000000001" self.finished_txn = InventoryTxn( id=1000, txn_no="TXN-FIN-001", txn_type="FG_IN", item_id=product.id, warehouse_id=finished_wh.id, location_id=None, lot_id=finished_lot.id, qty_change=Decimal("100"), weight_change_kg=Decimal("50"), unit_cost=Decimal("3.25"), amount=Decimal("325"), source_doc_type="生产台账", source_doc_id=2000, source_line_id=finished_lot.id, biz_time=self.now, remark=f"生产台账入库;成品入库;单据批次号:{batch_no}", created_at=self.now, updated_at=self.now, ) self.surplus_txn = InventoryTxn( id=1001, txn_no="TXN-SURPLUS-001", txn_type="PRODUCTION_SURPLUS_IN", item_id=material.id, warehouse_id=raw_wh.id, location_id=None, lot_id=source_lot.id, qty_change=Decimal("0"), weight_change_kg=Decimal("20"), unit_cost=Decimal("6.5"), amount=Decimal("130"), source_doc_type="生产台账", source_doc_id=2000, source_line_id=source_lot.id, biz_time=self.now, remark=f"生产台账入库;生产余料入库;单据批次号:{batch_no}", created_at=self.now, updated_at=self.now, ) self.scrap_txn = InventoryTxn( id=1002, txn_no="TXN-SCRAP-001", txn_type="PRODUCTION_SCRAP_IN", item_id=material.id, warehouse_id=scrap_wh.id, location_id=None, lot_id=scrap_lot.id, qty_change=Decimal("0"), weight_change_kg=Decimal("5"), unit_cost=Decimal("6.5"), amount=Decimal("32.5"), source_doc_type="生产台账", source_doc_id=2000, source_line_id=scrap_lot.id, biz_time=self.now, remark=f"生产台账入库;生产废料入库;单据批次号:{batch_no}", created_at=self.now, updated_at=self.now, ) archive = DocumentArchive( document_type=DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT, business_id=self.finished_txn.id, document_no="生产入库结算-TXN-FIN-001", archive_version=1, file_format="PDF", file_name="生产入库结算-TXN-FIN-001_V1.pdf", file_path="/tmp/生产入库结算-TXN-FIN-001_V1.pdf", status="已归档", error_message=None, created_by=None, created_at=self.now, ) self.db.add_all([ finished_wh, raw_wh, scrap_wh, material, product, source_lot, finished_lot, scrap_lot, self.finished_txn, self.surplus_txn, self.scrap_txn, archive, ]) self.db.commit() def test_finished_warehouse_row_exposes_own_settlement_archive(self) -> None: rows = self.db.execute(get_inventory_txn_ledger_query(warehouse_type="FINISHED")).mappings().all() self.assertEqual(len(rows), 1) row = dict(rows[0]) self.assertEqual(row["inventory_txn_id"], self.finished_txn.id) self.assertEqual(row["archive_business_id"], self.finished_txn.id) self.assertEqual(row["archive_document_type"], DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT) self.assertEqual(row["archive_status"], "已归档") def test_surplus_warehouse_row_exposes_same_settlement_archive(self) -> None: rows = self.db.execute(get_inventory_txn_ledger_query(warehouse_type="RAW")).mappings().all() self.assertEqual(len(rows), 1) row = dict(rows[0]) self.assertEqual(row["inventory_txn_id"], self.surplus_txn.id) self.assertEqual(row["archive_business_id"], self.finished_txn.id) self.assertEqual(row["archive_document_type"], DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT) self.assertEqual(row["archive_status"], "已归档") def test_scrap_warehouse_row_exposes_same_settlement_archive(self) -> None: rows = self.db.execute(get_inventory_txn_ledger_query(warehouse_type="SCRAP")).mappings().all() self.assertEqual(len(rows), 1) row = dict(rows[0]) self.assertEqual(row["inventory_txn_id"], self.scrap_txn.id) self.assertEqual(row["archive_business_id"], self.finished_txn.id) self.assertEqual(row["archive_document_type"], DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT) self.assertEqual(row["archive_status"], "已归档") if __name__ == "__main__": unittest.main() ``` - [ ] **Step 2: Run the test and verify it fails** Run: ```bash cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend python -m unittest backend.tests.test_inventory_transaction_ledger_archive_fields -v ``` Expected: fail with missing `archive_business_id`/`archive_document_type`/`archive_status` columns on the warehouse transaction ledger query. --- ### Task 2: Add Archive Metadata To Warehouse Transaction Ledger **Files:** - Modify: `backend/app/schemas/operations.py` - Modify: `backend/app/services/operations.py` - [ ] **Step 1: Extend the response schema** In `backend/app/schemas/operations.py`, add these fields to `InventoryTxnRead` after `remark`: ```python archive_status: str | None = None archive_business_id: int | None = None archive_document_type: str | None = None archive_error_message: str | None = None ``` - [ ] **Step 2: Import production archive constants** In `backend/app/services/operations.py`, add this import near the other service imports: ```python from app.services.document_archives import DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT, production_document_batch_no_from_text ``` - [ ] **Step 3: Add production settlement helper constants** In `backend/app/services/operations.py`, near `WAREHOUSE_LEDGER_SORT_COLUMNS`, add: ```python PRODUCTION_SETTLEMENT_INVENTORY_TXN_TYPES = { "FG_IN", "PRODUCTION_SURPLUS_IN", "PRODUCTION_SCRAP_IN", } PRODUCTION_SETTLEMENT_MAIN_TXN_PRIORITY = { "FG_IN": 1, "PRODUCTION_SURPLUS_IN": 2, "PRODUCTION_SCRAP_IN": 3, } ``` - [ ] **Step 4: Add a settlement main transaction subquery helper** In `backend/app/services/operations.py`, add this helper before `get_inventory_txn_ledger_query`: ```python def _production_settlement_archive_subquery(): settlement_txn = aliased(InventoryTxn) settlement_batch_no = func.coalesce( production_document_batch_no_from_text(settlement_txn.remark), func.date_format(settlement_txn.biz_time, "%Y-%m-%d %H:%i:%s.%f"), ).label("settlement_group_key") priority_expr = case( *[ (settlement_txn.txn_type == txn_type, priority) for txn_type, priority in PRODUCTION_SETTLEMENT_MAIN_TXN_PRIORITY.items() ], else_=99, ) grouped = ( select( settlement_txn.source_doc_id.label("production_ledger_id"), settlement_batch_no, func.min( case( *[ (settlement_txn.txn_type == txn_type, priority) for txn_type, priority in PRODUCTION_SETTLEMENT_MAIN_TXN_PRIORITY.items() ], else_=99, ) * 1_000_000_000 + settlement_txn.id ).label("main_rank"), ) .where( settlement_txn.source_doc_type == "生产台账", settlement_txn.source_doc_id.is_not(None), settlement_txn.txn_type.in_(PRODUCTION_SETTLEMENT_INVENTORY_TXN_TYPES), ) .group_by(settlement_txn.source_doc_id, settlement_batch_no) .subquery() ) return grouped ``` If MySQL-specific `date_format` breaks SQLite tests, replace the fallback group key with `func.coalesce(production_document_batch_no_from_text(settlement_txn.remark), settlement_txn.biz_time)`. The seed data uses `单据批次号` in all remarks, so the SQLite tests still exercise the intended path. - [ ] **Step 5: Join archive metadata in `get_inventory_txn_ledger_query`** In `get_inventory_txn_ledger_query`, add aliases before building `stmt`: ```python settlement_grouped = _production_settlement_archive_subquery() settlement_main_txn = aliased(InventoryTxn) settlement_archive = aliased(DocumentArchive) current_batch_no = func.coalesce( production_document_batch_no_from_text(InventoryTxn.remark), func.date_format(InventoryTxn.biz_time, "%Y-%m-%d %H:%i:%s.%f"), ) ``` Add these selected columns after `InventoryTxn.remark.label("remark")`: ```python settlement_archive.status.label("archive_status"), settlement_main_txn.id.label("archive_business_id"), literal(DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT).label("archive_document_type"), settlement_archive.error_message.label("archive_error_message"), ``` Add these joins after the operator join: ```python .outerjoin( settlement_grouped, and_( InventoryTxn.source_doc_type == "生产台账", InventoryTxn.source_doc_id == settlement_grouped.c.production_ledger_id, current_batch_no == settlement_grouped.c.settlement_group_key, InventoryTxn.txn_type.in_(PRODUCTION_SETTLEMENT_INVENTORY_TXN_TYPES), ), ) .outerjoin( settlement_main_txn, settlement_main_txn.id == settlement_grouped.c.main_rank % 1_000_000_000, ) .outerjoin( settlement_archive, and_( settlement_archive.document_type == DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT, settlement_archive.business_id == settlement_main_txn.id, settlement_archive.file_format == "PDF", settlement_archive.archive_version == select(func.max(DocumentArchive.archive_version)) .where( DocumentArchive.document_type == DOCUMENT_TYPE_PRODUCTION_INBOUND_SETTLEMENT, DocumentArchive.business_id == settlement_main_txn.id, DocumentArchive.file_format == "PDF", ) .scalar_subquery(), ), ) ``` - [ ] **Step 6: Run the backend test** Run: ```bash cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend python -m unittest backend.tests.test_inventory_transaction_ledger_archive_fields -v ``` Expected: all 3 tests pass. --- ### Task 3: Frontend Static Tests For Warehouse Ledger Archive Entrypoints **Files:** - Modify: `frontend/src/views/InventoryLedgerView.test.js` - [ ] **Step 1: Add static assertions** Append this `describe` block to `frontend/src/views/InventoryLedgerView.test.js`: ```javascript describe("InventoryLedgerView document archive entrypoints", () => { it("renders a warehouse ledger archive column using production settlement archive actions", () => { assert.match(source, /