docs: add production waste-output list implementation plan.

Break the approved design into TDD-backed backend list helpers, menu SQL, and a P01 read-only web-admin page.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-08-18 10:48:56 +08:00
1 parent af337562ae
commit a089031994
1 file changed
+592
@@ -0,0 +1,592 @@
# 生产废弃物管理(只读列表)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**
```python
# 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:
```bash
cd backend && source venv/bin/activate && python -m pytest tests/test_production_waste_output.py -v
```
Expected: FAIL(模块不存在)
- [ ] **Step 3: Minimal implementation**
```python
# 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**
```bash
cd backend && source venv/bin/activate && python -m pytest tests/test_production_waste_output.py -v
```
Expected: PASS
- [ ] **Step 5: Commit**
```bash
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 区增加:
```python
from app.crud import production_waste_output as production_waste_output_svc
```
在合适位置(建议 `/production-tasks` 路由组附近)增加:
```python
@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: 冒烟导入**
```bash
cd backend && source venv/bin/activate && python -c "from app.api.production import list_production_waste_outputs; print('ok')"
```
Expected: `ok`
- [ ] **Step 3: Commit**
```bash
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**
脚本须包含:
1. 若不存在则 INSERT C / F(或按 id 更新 path/component/perms/order_num)。
2. 对「已拥有 PRM 下任一有效子菜单」的角色,幂等授予 C+F,并补祖先 `PRM000…`(避免 MySQL 1093:用临时表/派生表,勿同句边读边写 `sys_role_menu`)。
3. `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 示例值:
```sql
-- 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**
```bash
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 模块**
```typescript
// 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 项后面插入:
```typescript
{
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**
```bash
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,勿省略):
1. `defineOptions({ name: 'ProductionWasteOutput' })`
2. `panel-top`:`<h2 class="title-bar-heading">废弃物列表</h2>`(显式竖条类,无需改 App.vue)
3. `filter-block`:关键字(placeholder「产出名称/条码」)+ 登记时间 `el-date-picker` `type="daterange"` `value-format="YYYY-MM-DD"` → 请求 `createTimeStart` / `createTimeEnd`
4. **无** list-toolbar 新增/批量删除;可省略「已选 N 项」
5. `list-table` + `stripe`,列:产出名称、条码、工序/步骤、任务号、工单号、样本编码、备注、登记时间、操作(查看)
6. 空单元格显示 `—`;`el-empty` description=`暂无废弃物登记`
7. 分页:`list-body.list-wrap` 内 + `.pagination.list-pager-bar`
8. 操作列 `fixed="right"` `min-width` ≥ 120:`查看` 按钮带 `<el-icon><View /></el-icon>`;无 `taskId` 时 `disabled`
9. 页面底部挂载:
```vue
<TaskExecutionDialog
v-model:visible="executionDialogVisible"
:task-id="executionTaskId"
:task-name="executionTaskName"
:readonly="true"
append-to-body
/>
```
`handleView(row)`:若无 `taskId` → `ElMessage.warning('关联任务不可用')`;否则打开对话框。
10. `onMounted` / 搜索 / 重置 → `fetchProductionWasteOutputList`
- [ ] **Step 2: 本地打开页面冒烟**
- Hash:`/#/production-group/production/waste-output`
- 需先执行菜单 SQL;或临时用 frontend access 模式验证路由组件可加载
- [ ] **Step 3: Commit**
```bash
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: 更新知识图谱**
```bash
./scripts/update-graph.sh backend
./scripts/update-graph.sh frontend
```
- [ ] **Step 2: 手工验收(对照 spec §9)**
1. 任务执行登记废弃物(可空名称/空条码)→ 本页刷新可见;有效物不出现。
2. 关键字、登记时间、分页正确。
3. 「查看」打开只读任务执行;无 taskId 时禁用/提示。
4. 侧栏【生产管理】在洁净服后出现【废弃物管理】。
- [ ] **Step 3: Commit graphify(若有变更)**
```bash
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`)。