生产废弃物管理（只读列表）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: 在【生产管理】下新增【废弃物管理】只读列表页，分页汇总所有生产任务工艺步骤中类型为废弃物（FQW）的步骤产出。

Architecture: 不新建业务表。后端按租户解析 biz_output_type（优先 type_code=FQW，兜底 type_name=废弃物），JOIN biz_production_task_step_output → step → task → work_order，应用与生产任务列表相同的数据权限后分页；前端 P01 只读列表 +「查看」打开只读 TaskExecutionDialog。

Tech Stack: FastAPI / SQLAlchemy / Vue 3 / Element Plus；设计依据 docs/superpowers/specs/2026-08-18-production-waste-output-list-design.md。

Spec: docs/superpowers/specs/2026-08-18-production-waste-output-list-design.md

File map

File	Responsibility
Create: `backend/app/crud/production_waste_output.py`	废弃物类型 id 解析 + 列表查询
Create: `backend/tests/test_production_waste_output.py`	类型解析纯函数 / 关键字匹配语义单测
Modify: `backend/app/api/production.py`	注册 `GET /production-waste-outputs`
Create: `docs/sql/alter_sys_menu_add_production_waste_output.sql`	幂等菜单 + 角色授权
Modify: `待同步到生产的sql脚本`	追加待上线 SQL 片段
Create: `web-admin/src/api/production-waste-output.ts`	列表 API
Create: `web-admin/src/views/production/waste-output/index.vue`	P01 只读列表页
Modify: `web-admin/src/router/modules/sqxp.ts`	路由（洁净服后）
Modify: `backend/app/api/menu.py`	静态菜单镜像（洁净服后）

Task 1: 后端 — 废弃物类型 id 解析（TDD）

Files:

Create: backend/app/crud/production_waste_output.py
Create: backend/tests/test_production_waste_output.py
[ ] Step 1: Write the failing test

# backend/tests/test_production_waste_output.py
"""生产废弃物列表：产出类型解析与关键字匹配语义。"""
import unittest

from app.crud.production_waste_output import (
    WASTE_OUTPUT_TYPE_CODE,
    WASTE_OUTPUT_TYPE_NAME,
    match_waste_output_keyword,
    pick_waste_output_type_ids,
)


class TestPickWasteOutputTypeIds(unittest.TestCase):
    def test_prefers_type_code_fqw(self):
        rows = [
            {"id": "a", "type_code": "YXW", "type_name": "有效物"},
            {"id": "b", "type_code": "FQW", "type_name": "废弃物"},
            {"id": "c", "type_code": "", "type_name": "废弃物"},
        ]
        self.assertEqual(pick_waste_output_type_ids(rows), {"b"})

    def test_fallback_type_name_when_no_fqw(self):
        rows = [
            {"id": "a", "type_code": "YXW", "type_name": "有效物"},
            {"id": "c", "type_code": "", "type_name": "废弃物"},
            {"id": "d", "type_code": None, "type_name": "废弃物"},
        ]
        self.assertEqual(pick_waste_output_type_ids(rows), {"c", "d"})

    def test_empty_when_no_match(self):
        self.assertEqual(pick_waste_output_type_ids([]), set())
        self.assertEqual(
            pick_waste_output_type_ids([{"id": "a", "type_code": "YXW", "type_name": "有效物"}]),
            set(),
        )


class TestMatchWasteOutputKeyword(unittest.TestCase):
    def test_matches_name_or_barcode(self):
        self.assertTrue(match_waste_output_keyword("废液", "废液A", "X1"))
        self.assertTrue(match_waste_output_keyword("X1", "废液", "BX1Y"))
        self.assertFalse(match_waste_output_keyword("zzz", "废液", "X1"))

    def test_blank_keyword_always_true(self):
        self.assertTrue(match_waste_output_keyword("", "废液", ""))
        self.assertTrue(match_waste_output_keyword(None, "", ""))


class TestConstants(unittest.TestCase):
    def test_constants(self):
        self.assertEqual(WASTE_OUTPUT_TYPE_CODE, "FQW")
        self.assertEqual(WASTE_OUTPUT_TYPE_NAME, "废弃物")


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

Step 2: Run test to verify it fails

Run:

cd backend && source venv/bin/activate && python -m pytest tests/test_production_waste_output.py -v

Expected: FAIL（模块不存在）

Step 3: Minimal implementation

# backend/app/crud/production_waste_output.py
"""生产任务步骤废弃物产出 — 只读汇总列表。"""
from __future__ import annotations

from typing import Any, Dict, Iterable, List, Optional, Set, Tuple

from sqlalchemy import and_, or_
from sqlalchemy.orm import Session

from app.models.models import (
    BizOutputType,
    BizProductionTask,
    BizProductionTaskStep,
    BizProductionTaskStepOutput,
    BizProductionWorkOrder,
)

WASTE_OUTPUT_TYPE_CODE = "FQW"
WASTE_OUTPUT_TYPE_NAME = "废弃物"


def pick_waste_output_type_ids(rows: Iterable[Dict[str, Any]]) -> Set[str]:
    """从产出类型行中选出废弃物类型 id：优先 type_code=FQW，否则兜底 type_name=废弃物。"""
    rows = list(rows or [])
    by_code = {
        str(r.get("id") or "").strip()
        for r in rows
        if str(r.get("type_code") or "").strip() == WASTE_OUTPUT_TYPE_CODE
        and str(r.get("id") or "").strip()
    }
    if by_code:
        return by_code
    return {
        str(r.get("id") or "").strip()
        for r in rows
        if str(r.get("type_name") or "").strip() == WASTE_OUTPUT_TYPE_NAME
        and str(r.get("id") or "").strip()
    }


def match_waste_output_keyword(
    keyword: Optional[str], output_name: Optional[str], output_barcode: Optional[str]
) -> bool:
    kw = (keyword or "").strip()
    if not kw:
        return True
    name = output_name or ""
    barcode = output_barcode or ""
    return kw in name or kw in barcode


def resolve_waste_output_type_ids(db: Session, tenant_no: str) -> Set[str]:
    rows = (
        db.query(BizOutputType.id, BizOutputType.type_code, BizOutputType.type_name)
        .filter(BizOutputType.tenant_no == tenant_no, BizOutputType.del_flag == 0)
        .all()
    )
    return pick_waste_output_type_ids(
        [{"id": r.id, "type_code": r.type_code, "type_name": r.type_name} for r in rows]
    )


def _fmt_dt(t) -> str:
    if t is None:
        return ""
    if hasattr(t, "strftime"):
        return t.strftime("%Y-%m-%d %H:%M:%S")
    return str(t)


def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
    v = (value or "").strip()
    if not v:
        return v
    if len(v) <= 10:
        return f"{v} 23:59:59" if end_of_day else f"{v} 00:00:00"
    return v


def list_production_waste_outputs(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    type_ids = resolve_waste_output_type_ids(db, tenant_no)
    if not type_ids:
        return [], 0

    q = (
        db.query(
            BizProductionTaskStepOutput,
            BizProductionTaskStep,
            BizProductionTask,
            BizProductionWorkOrder,
        )
        .join(
            BizProductionTaskStep,
            and_(
                BizProductionTaskStep.id == BizProductionTaskStepOutput.step_id,
                BizProductionTaskStep.del_flag == 0,
            ),
        )
        .join(
            BizProductionTask,
            and_(
                BizProductionTask.id == BizProductionTaskStep.task_id,
                BizProductionTask.del_flag == 0,
                BizProductionTask.tenant_no == tenant_no,
            ),
        )
        .outerjoin(
            BizProductionWorkOrder,
            and_(
                BizProductionWorkOrder.id == BizProductionTask.work_order_id,
                BizProductionWorkOrder.del_flag == 0,
            ),
        )
        .filter(
            BizProductionTaskStepOutput.tenant_no == tenant_no,
            BizProductionTaskStepOutput.del_flag == 0,
            BizProductionTaskStepOutput.output_type_id.in_(list(type_ids)),
        )
    )

    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                BizProductionTaskStepOutput.output_name.like(like),
                BizProductionTaskStepOutput.output_barcode.like(like),
            )
        )
    if create_time_start and create_time_start.strip():
        q = q.filter(
            BizProductionTaskStepOutput.create_time
            >= _normalize_date_bound(create_time_start.strip())
        )
    if create_time_end and create_time_end.strip():
        q = q.filter(
            BizProductionTaskStepOutput.create_time
            <= _normalize_date_bound(create_time_end.strip(), end_of_day=True)
        )

    if user_info is not None:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizProductionTask, user_info, db, tenant_no)

    q = q.order_by(BizProductionTaskStepOutput.create_time.desc())
    total = q.count()
    page = max(1, int(page or 1))
    page_size = max(1, min(int(page_size or 10), 100))
    rows = q.offset((page - 1) * page_size).limit(page_size).all()

    items: List[Dict[str, Any]] = []
    for out, step, task, wo in rows:
        items.append(
            {
                "id": out.id,
                "outputName": out.output_name or "",
                "outputBarcode": out.output_barcode or "",
                "remark": out.remark or "",
                "createTime": _fmt_dt(out.create_time),
                "stepId": out.step_id or "",
                "processName": (step.process_name if step else None) or "",
                "taskId": (task.id if task else None) or "",
                "taskNo": (task.task_no if task else None) or "",
                "workOrderId": (wo.id if wo else None) or (task.work_order_id if task else "") or "",
                "workOrderNo": (wo.work_order_no if wo else None) or "",
                "sampleCode": (wo.sample_code if wo else None) or "",
            }
        )
    return items, total


def total_pages(total: int, page_size: int) -> int:
    if page_size <= 0:
        return 0
    return (total + page_size - 1) // page_size

Step 4: Run tests to verify they pass

cd backend && source venv/bin/activate && python -m pytest tests/test_production_waste_output.py -v

Expected: PASS

Step 5: Commit

git add backend/app/crud/production_waste_output.py backend/tests/test_production_waste_output.py
git commit -m "feat(backend): add production waste-output list query helpers."

Task 2: 后端 — 注册列表 API

Files:

Modify: backend/app/api/production.py（在 list_production_tasks 附近新增路由）
[ ] Step 1: 增加 import 与路由

在 production.py 顶部 CRUD import 区增加：

from app.crud import production_waste_output as production_waste_output_svc

在合适位置（建议 /production-tasks 路由组附近）增加：

@router.get("/production-waste-outputs")
async def list_production_waste_outputs(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    createTimeStart: Optional[str] = Query(None, alias="createTimeStart"),
    createTimeEnd: Optional[str] = Query(None, alias="createTimeEnd"),
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    items, total = production_waste_output_svc.list_production_waste_outputs(
        db,
        tenant_no=tenant_no,
        page=page,
        page_size=pageSize,
        keyword=keyword,
        create_time_start=createTimeStart,
        create_time_end=createTimeEnd,
        user_info=user,
    )
    tp = production_waste_output_svc.total_pages(total, pageSize)
    return success_response(
        {"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp}
    )

Step 2: 冒烟导入

cd backend && source venv/bin/activate && python -c "from app.api.production import list_production_waste_outputs; print('ok')"

Expected: ok

Step 3: Commit

git add backend/app/api/production.py
git commit -m "feat(backend): expose GET /production-waste-outputs list API."

Task 3: 菜单 SQL + 待同步脚本

Files:

Create: docs/sql/alter_sys_menu_add_production_waste_output.sql
Modify: 待同步到生产的sql脚本（追加同一可执行片段）

约定：

项	值
C id	`95300000000000000000000000000001`
F id	`95300000000000000000000000000002`
parent	`PRM00000000000000000000000000001`
path	`waste-output`
component	`/production/waste-output/index`
perms	`production:waste-output:list`
order_num	`8`（洁净服当前为 `7`，紧挨其后）
F menu_name	废弃物查询

Step 1: 编写幂等 SQL

脚本须包含：

若不存在则 INSERT C / F（或按 id 更新 path/component/perms/order_num）。
对「已拥有 PRM 下任一有效子菜单」的角色，幂等授予 C+F，并补祖先 PRM000…（避免 MySQL 1093：用临时表/派生表，勿同句边读边写 sys_role_menu）。
sys_role_menu.id = UPPER(MD5(CONCAT(role_id, '|', menu_id)))；软删先恢复。

参考写法：docs/superpowers/specs/2026-08-17-production-pre-transit-sample-design.md §6.2（同 PRM 父菜单授权模式）。若仓库中仍有 alter_sys_menu_add_production_pre_transit_sample.sql 历史副本可对照；否则按该设计段落手写。

C 菜单 INSERT 示例值：

-- menu_name='废弃物管理', parent_id='PRM00000000000000000000000000001', menu_type='C',
-- path='waste-output', component='/production/waste-output/index',
-- perms='production:waste-output:list', order_num=8, tenant_no='zkxw', del_flag=0

F 按钮：menu_type='F', parent_id='95300000000000000000000000000001', menu_name='废弃物查询', 同 perms, order_num=1。

[ ] Step 2: 将可执行片段追加到 待同步到生产的sql脚本
[ ] Step 3: Commit

git add docs/sql/alter_sys_menu_add_production_waste_output.sql "待同步到生产的sql脚本"
git commit -m "docs(sql): add production waste-output menu migration."

Task 4: 前端 API + 路由 + 菜单镜像

Files:

Create: web-admin/src/api/production-waste-output.ts
Modify: web-admin/src/router/modules/sqxp.ts（在 ProductionCleanroomGarment 路由对象之后插入）
Modify: backend/app/api/menu.py（生产管理 children 中洁净服节点之后插入，字段与 sqxp 一致）
[ ] Step 1: API 模块

// web-admin/src/api/production-waste-output.ts
import request from '@/utils/http'

export interface ProductionWasteOutputItem {
  id: string
  outputName?: string
  outputBarcode?: string
  remark?: string
  createTime?: string
  stepId?: string
  processName?: string
  taskId?: string
  taskNo?: string
  workOrderId?: string
  workOrderNo?: string
  sampleCode?: string
}

export function fetchProductionWasteOutputList(params?: {
  page?: number
  pageSize?: number
  keyword?: string
  createTimeStart?: string
  createTimeEnd?: string
}) {
  return request.get<{
    list: ProductionWasteOutputItem[]
    total: number
    page: number
    pageSize: number
    totalPages: number
  }>({ url: '/web/admin/production-waste-outputs', params })
}

Step 2: sqxp.ts 路由

在 path: 'cleanroom-garment' 那个 children 项后面插入：

{
  id: '95300000000000000000000000000001',
  name: 'ProductionWasteOutput',
  path: 'waste-output',
  component: '/production/waste-output/index',
  meta: {
    title: '废弃物管理',
    keepAlive: true,
    menuCode: 'production:waste-output:list'
  }
}

Step 3: menu.py 镜像

同样在 ProductionCleanroomGarment 字典后插入等价 JSON（Python dict，True/False）。

Step 4: Commit

git add web-admin/src/api/production-waste-output.ts web-admin/src/router/modules/sqxp.ts backend/app/api/menu.py
git commit -m "feat: wire production waste-output API route and menu mirror."

Task 5: 前端只读列表页

Files:

Create: web-admin/src/views/production/waste-output/index.vue

参考： web-admin/src/views/system/unit/index.vue（P01 骨架）；「查看」打开任务执行对齐 web-admin/src/views/production/cell-product/SourceChainGraph.vue（TaskExecutionDialog + readonly）。

Step 1: 创建页面

要点（实现时写完整 Vue，勿省略）：

defineOptions({ name: 'ProductionWasteOutput' })
panel-top：<h2 class="title-bar-heading">废弃物列表</h2>（显式竖条类，无需改 App.vue）
filter-block：关键字（placeholder「产出名称/条码」）+ 登记时间 el-date-picker type="daterange" value-format="YYYY-MM-DD" → 请求 createTimeStart / createTimeEnd
无 list-toolbar 新增/批量删除；可省略「已选 N 项」
list-table + stripe，列：产出名称、条码、工序/步骤、任务号、工单号、样本编码、备注、登记时间、操作（查看）
空单元格显示 —；el-empty description=暂无废弃物登记
分页：list-body.list-wrap 内 + .pagination.list-pager-bar
操作列 fixed="right" min-width ≥ 120：查看 按钮带 <el-icon><View /></el-icon>；无 taskId 时 disabled
页面底部挂载：

<TaskExecutionDialog
  v-model:visible="executionDialogVisible"
  :task-id="executionTaskId"
  :task-name="executionTaskName"
  :readonly="true"
  append-to-body
/>

handleView(row)：若无 taskId → ElMessage.warning('关联任务不可用')；否则打开对话框。

onMounted / 搜索 / 重置 → fetchProductionWasteOutputList

[ ] Step 2: 本地打开页面冒烟
Hash：/#/production-group/production/waste-output
需先执行菜单 SQL；或临时用 frontend access 模式验证路由组件可加载
[ ] Step 3: Commit

git add web-admin/src/views/production/waste-output/index.vue
git commit -m "feat(web-admin): add production waste-output read-only list page."

Task 6: 图谱更新 + 验收

Step 1: 更新知识图谱

./scripts/update-graph.sh backend
./scripts/update-graph.sh frontend

Step 2: 手工验收（对照 spec §9）

任务执行登记废弃物（可空名称/空条码）→ 本页刷新可见；有效物不出现。
关键字、登记时间、分页正确。
「查看」打开只读任务执行；无 taskId 时禁用/提示。
侧栏【生产管理】在洁净服后出现【废弃物管理】。

Step 3: Commit graphify（若有变更）

git add backend/graphify-out web-admin/graphify-out
git commit -m "chore: update graphify after waste-output feature."

Spec coverage check

Spec 项	Task
只读列表 / 无写 API	Task 2、5
FQW 优先 + 名称兜底	Task 1
空名称/条码也展示	Task 1 查询无非空过滤
关键字 + 登记时间	Task 1、2、5
JOIN step/task/wo + 数据权限	Task 1
菜单同级紧挨洁净服 order 8	Task 3、4
查看 → 只读任务执行	Task 5
SQL 待同步	Task 3
graphify	Task 6

Placeholder scan

无 TBD；类型解析与列表函数签名前后一致（list_production_waste_outputs / fetchProductionWasteOutputList）。

2026-08-18-production-waste-output-list.md 19 KB Pysyvä linkki Historia Raaka

生产废弃物管理（只读列表）Implementation Plan

File map

Task 1: 后端 — 废弃物类型 id 解析（TDD）

Task 2: 后端 — 注册列表 API

Task 3: 菜单 SQL + 待同步脚本

Task 4: 前端 API + 路由 + 菜单镜像

Task 5: 前端只读列表页

Task 6: 图谱更新 + 验收

Spec coverage check

Placeholder scan

2026-08-18-production-waste-output-list.md 19 KB

Pysyvä linkki Historia Raaka