# OA 审批任务中心检索区 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:** 为待办/已办/我发起的/抄送我的四中心增加服务端筛选(流程名称、申请人、申请时间、流程状态)与即时检索 UI。 **Architecture:** 扩展 `list_tasks` 在分页前按 Query 过滤(join `OaInstance`/`OaApp`/`SysUser`);API 透传参数;四页 Vue 使用 Fresh Mint `filter-block` + `watchDebounced(300)`,保留搜索/重置。 **Tech Stack:** FastAPI · SQLAlchemy · unittest;Vue 3 · Element Plus · `@vueuse/core` `watchDebounced`(auto-import) **Spec:** `docs/superpowers/specs/2026-08-11-oa-approval-task-filter-design.md` **UI Demo:** `docs/demos/oa-approval-task-filter-demo.html` --- ## File map | Path | Responsibility | |------|----------------| | Create: `backend/tests/test_oa_tasks_filters.py` | 筛选单测(TDD) | | Modify: `backend/app/services/oa_engine.py` | `list_tasks` 过滤参数与 join | | Modify: `backend/app/api/oa_approval.py` | `list_oa_tasks` Query 透传 | | Modify: `web-admin/src/api/oaApproval.ts` | `fetchOaTasks` params | | Modify: `web-admin/src/views/oa-approval/pending/index.vue` | 筛选区 + 即时检索 | | Modify: `web-admin/src/views/oa-approval/done/index.vue` | 同上 | | Modify: `web-admin/src/views/oa-approval/mine/index.vue` | 同上(无申请人) | | Modify: `web-admin/src/views/oa-approval/cc/index.vue` | 扩展筛选 + 即时检索 | | Run: `./scripts/update-graph.sh backend` | 图谱(签名变更) | --- ### Task 1: 后端筛选失败测试(TDD) **Files:** - Create: `backend/tests/test_oa_tasks_filters.py` - [ ] **Step 1: 写入测试文件** ```python """OA list_tasks 检索过滤。""" from __future__ import annotations import unittest from datetime import datetime from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from app.models.base import Base from app.models.models import OaApp, OaAppVersion, OaInstance, OaTask, SysUser class TestOaTasksFilters(unittest.TestCase): def setUp(self): self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ OaApp.__table__, OaAppVersion.__table__, SysUser.__table__, OaInstance.__table__, OaTask.__table__, ], ) Session = sessionmaker(bind=self.engine, autoflush=False, autocommit=False) self.db = Session() t1 = datetime(2026, 7, 10, 9, 0, 0) t2 = datetime(2026, 7, 20, 15, 30, 0) self.db.add_all( [ OaApp( id="app-leave", app_name="请假审批", initiator_scope="all", admin_scope="all_oa_admin", status="1", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaApp( id="app-ot", app_name="加班申请", initiator_scope="all", admin_scope="all_oa_admin", status="1", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaAppVersion( id="v-leave", app_id="app-leave", version_no=1, status="published", is_latest_published="1", form_schema_json="{}", flow_tree_json="{}", advanced_settings_json="{}", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaAppVersion( id="v-ot", app_id="app-ot", version_no=1, status="published", is_latest_published="1", form_schema_json="{}", flow_tree_json="{}", advanced_settings_json="{}", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), SysUser( id="u-a", user_no="a", nick_name="张三", password="x", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), SysUser( id="u-b", user_no="b", nick_name="李四", password="x", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), SysUser( id="u-approver", user_no="ap", nick_name="审批人", password="x", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaInstance( id="i-leave", app_id="app-leave", app_version_id="v-leave", title="请假-忽略标题", initiator_user_id="u-a", form_data_json="{}", status="running", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaInstance( id="i-ot", app_id="app-ot", app_version_id="v-ot", title="加班标题含报销", initiator_user_id="u-b", form_data_json="{}", status="completed", tenant_no="zkxw", create_dept_id="", create_time=t2, update_time=t2, del_flag=0, ), OaTask( id="t-leave-pending", instance_id="i-leave", node_key="n1", node_name="审批", task_type="approval", assignee_user_id="u-approver", status="pending", tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaTask( id="t-ot-done", instance_id="i-ot", node_key="n1", node_name="审批", task_type="approval", assignee_user_id="u-approver", status="approved", tenant_no="zkxw", create_dept_id="", create_time=t2, update_time=t2, del_flag=0, ), OaTask( id="t-leave-cc-unread", instance_id="i-leave", node_key="cc1", node_name="抄送", task_type="cc", assignee_user_id="u-approver", status="done", is_read=0, tenant_no="zkxw", create_dept_id="", create_time=t1, update_time=t1, del_flag=0, ), OaTask( id="t-ot-cc-read", instance_id="i-ot", node_key="cc1", node_name="抄送", task_type="cc", assignee_user_id="u-approver", status="done", is_read=1, tenant_no="zkxw", create_dept_id="", create_time=t2, update_time=t2, del_flag=0, ), ] ) self.db.commit() def tearDown(self): self.db.close() self.engine.dispose() def test_mine_filter_by_app_name(self): from app.services.oa_engine import list_tasks res = list_tasks( self.db, tenant_no="zkxw", user_id="u-a", kind="mine", app_name="请假", ) self.assertEqual(res["total"], 1) self.assertEqual(res["list"][0]["id"], "i-leave") def test_mine_filter_by_app_name_matches_title(self): from app.services.oa_engine import list_tasks # u-b 发起的加班实例 title 含「报销」 res = list_tasks( self.db, tenant_no="zkxw", user_id="u-b", kind="mine", app_name="报销", ) self.assertEqual(res["total"], 1) self.assertEqual(res["list"][0]["id"], "i-ot") def test_pending_filter_by_initiator_name(self): from app.services.oa_engine import list_tasks res = list_tasks( self.db, tenant_no="zkxw", user_id="u-approver", kind="pending", initiator_name="张", ) self.assertEqual(res["total"], 1) self.assertEqual(res["list"][0]["instanceId"], "i-leave") def test_done_filter_by_apply_time_range(self): from app.services.oa_engine import list_tasks res = list_tasks( self.db, tenant_no="zkxw", user_id="u-approver", kind="done", apply_time_from="2026-07-15", apply_time_to="2026-07-25", ) self.assertEqual(res["total"], 1) self.assertEqual(res["list"][0]["instanceId"], "i-ot") def test_mine_filter_by_instance_status(self): from app.services.oa_engine import list_tasks res = list_tasks( self.db, tenant_no="zkxw", user_id="u-b", kind="mine", instance_status="completed", ) self.assertEqual(res["total"], 1) self.assertEqual(res["list"][0]["instanceStatus"], "completed") def test_invalid_instance_status_ignored(self): from app.services.oa_engine import list_tasks res = list_tasks( self.db, tenant_no="zkxw", user_id="u-a", kind="mine", instance_status="not-a-real-status", ) self.assertEqual(res["total"], 1) def test_cc_combine_read_status_and_app_name(self): from app.services.oa_engine import list_tasks res = list_tasks( self.db, tenant_no="zkxw", user_id="u-approver", kind="cc", read_status="unread", app_name="请假", ) self.assertEqual(res["total"], 1) self.assertEqual(res["list"][0]["id"], "t-leave-cc-unread") if __name__ == "__main__": unittest.main() ``` - [ ] **Step 2: 跑测确认失败** ```bash cd backend && source venv/bin/activate && pytest tests/test_oa_tasks_filters.py -v ``` Expected: FAIL(`list_tasks()` 不接受 `app_name` 等关键字参数,或过滤未生效) - [ ] **Step 3: Commit 测试** ```bash git add backend/tests/test_oa_tasks_filters.py git commit -m "test: add failing OA task list filter coverage" ``` --- ### Task 2: 实现 `list_tasks` 过滤 **Files:** - Modify: `backend/app/services/oa_engine.py`(`list_tasks` 及其上方辅助函数) - [ ] **Step 1: 在 `list_tasks` 上方增加常量与辅助函数** 在 `list_tasks` 定义之前加入: ```python _INSTANCE_STATUS_FILTER_VALUES = frozenset( {"running", "completed", "rejected", "returned", "cancelled"} ) def _parse_day_start(s: Optional[str]) -> Optional[datetime]: raw = (s or "").strip() if not raw: return None try: d = datetime.strptime(raw[:10], "%Y-%m-%d") except ValueError: return None return d.replace(hour=0, minute=0, second=0, microsecond=0) def _parse_day_end(s: Optional[str]) -> Optional[datetime]: raw = (s or "").strip() if not raw: return None try: d = datetime.strptime(raw[:10], "%Y-%m-%d") except ValueError: return None return d.replace(hour=23, minute=59, second=59, microsecond=999999) def _apply_instance_list_filters( q, *, app_name: Optional[str] = None, initiator_name: Optional[str] = None, apply_time_from: Optional[str] = None, apply_time_to: Optional[str] = None, instance_status: Optional[str] = None, joined_app: bool = False, joined_initiator: bool = False, ): """在已包含 OaInstance 的 query 上叠加列表筛选。返回 (query, joined_app, joined_initiator)。""" from sqlalchemy import or_ kw = (app_name or "").strip() if kw: if not joined_app: q = q.outerjoin( OaApp, (OaApp.id == OaInstance.app_id) & (OaApp.del_flag == 0), ) joined_app = True like = f"%{kw}%" q = q.filter(or_(OaApp.app_name.like(like), OaInstance.title.like(like))) iname = (initiator_name or "").strip() if iname: if not joined_initiator: q = q.outerjoin( SysUser, (SysUser.id == OaInstance.initiator_user_id) & (SysUser.del_flag == 0), ) joined_initiator = True q = q.filter(SysUser.nick_name.like(f"%{iname}%")) t0 = _parse_day_start(apply_time_from) if t0 is not None: q = q.filter(OaInstance.create_time >= t0) t1 = _parse_day_end(apply_time_to) if t1 is not None: q = q.filter(OaInstance.create_time <= t1) st = (instance_status or "").strip() if st in _INSTANCE_STATUS_FILTER_VALUES: q = q.filter(OaInstance.status == st) return q, joined_app, joined_initiator ``` 确认文件顶部已 import:`OaApp`, `OaInstance`, `SysUser`, `Optional`, `datetime`(已有则不重复)。 - [ ] **Step 2: 扩展 `list_tasks` 签名并在分页前调用过滤** 将签名改为: ```python def list_tasks( db: Session, *, tenant_no: str, user_id: str, kind: str, read_status: Optional[str] = None, app_name: Optional[str] = None, initiator_name: Optional[str] = None, apply_time_from: Optional[str] = None, apply_time_to: Optional[str] = None, instance_status: Optional[str] = None, page: int = 1, page_size: int = 20, ) -> Dict[str, Any]: ``` 在 `kind == "mine"` 分支中,基础 `q` 建好后、`count` 之前: ```python q, _, _ = _apply_instance_list_filters( q, app_name=app_name, initiator_name=initiator_name, apply_time_from=apply_time_from, apply_time_to=apply_time_to, instance_status=instance_status, ) ``` 在 `pending` / `done` / `cc` 分支中,在各自 `q` 与 `read_status` 条件之后、`count` 之前: ```python need_inst_join = any( [ (app_name or "").strip(), (initiator_name or "").strip(), (apply_time_from or "").strip(), (apply_time_to or "").strip(), (instance_status or "").strip() in _INSTANCE_STATUS_FILTER_VALUES, ] ) if need_inst_join: q = q.join( OaInstance, (OaInstance.id == OaTask.instance_id) & (OaInstance.tenant_no == tenant_no) & (OaInstance.del_flag == 0), ) q, _, _ = _apply_instance_list_filters( q, app_name=app_name, initiator_name=initiator_name, apply_time_from=apply_time_from, apply_time_to=apply_time_to, instance_status=instance_status, ) ``` 注意:若 `instance_status` 非法,`need_inst_join` 中不要因非法 status 单独 join;上面写法已用 `in _INSTANCE_STATUS_FILTER_VALUES`。 - [ ] **Step 3: 跑测通过** ```bash cd backend && source venv/bin/activate && pytest tests/test_oa_tasks_filters.py tests/test_oa_engine.py -q ``` Expected: PASS - [ ] **Step 4: Commit** ```bash git add backend/app/services/oa_engine.py backend/tests/test_oa_tasks_filters.py git commit -m "feat: filter OA task lists by app name, initiator, time, status" ``` --- ### Task 3: API 透传 Query **Files:** - Modify: `backend/app/api/oa_approval.py`(`list_oa_tasks`,约 1040 行) - [ ] **Step 1: 扩展端点参数并传入 `list_tasks`** ```python @router.get("/tasks/{kind}") async def list_oa_tasks( kind: str, readStatus: Optional[str] = Query(None, alias="readStatus"), appName: Optional[str] = Query(None, alias="appName"), initiatorName: Optional[str] = Query(None, alias="initiatorName"), applyTimeFrom: Optional[str] = Query(None, alias="applyTimeFrom"), applyTimeTo: Optional[str] = Query(None, alias="applyTimeTo"), instanceStatus: Optional[str] = Query(None, alias="instanceStatus"), page: int = Query(1, ge=1), pageSize: int = Query(20, ge=1, le=100), user: dict = Depends(require_admin_or_super_admin), db: Session = Depends(get_db), ): from app.services.oa_engine import OaEngineError, list_tasks try: tenant_no = _get_tenant_no(db, user) data = list_tasks( db, tenant_no=tenant_no, user_id=user.get("user_id") or "", kind=kind, read_status=readStatus, app_name=appName, initiator_name=initiatorName, apply_time_from=applyTimeFrom, apply_time_to=applyTimeTo, instance_status=instanceStatus, page=page, page_size=pageSize, ) return success_response( {"list": data.get("list") or [], "total": int(data.get("total") or 0)} ) except OaEngineError as e: return error_response(400, str(e)) ``` - [ ] **Step 2: Commit** ```bash git add backend/app/api/oa_approval.py git commit -m "feat: expose OA task list filter query params on API" ``` --- ### Task 4: 前端 `fetchOaTasks` 参数 **Files:** - Modify: `web-admin/src/api/oaApproval.ts` - [ ] **Step 1: 扩展 params 并透传** 将 `fetchOaTasks` 替换为: ```typescript export function fetchOaTasks( kind: 'pending' | 'done' | 'mine' | 'cc', params?: { readStatus?: 'unread' | 'read' | 'all' appName?: string initiatorName?: string applyTimeFrom?: string applyTimeTo?: string instanceStatus?: string page?: number pageSize?: number } ) { const appName = (params?.appName || '').trim() const initiatorName = (params?.initiatorName || '').trim() const applyTimeFrom = (params?.applyTimeFrom || '').trim() const applyTimeTo = (params?.applyTimeTo || '').trim() const instanceStatus = (params?.instanceStatus || '').trim() return request.get<{ list: any[]; total: number }>({ url: `${BASE}/tasks/${kind}`, params: { page: params?.page ?? 1, pageSize: params?.pageSize ?? 20, ...(kind === 'cc' ? { readStatus: params?.readStatus ?? 'unread' } : {}), ...(appName ? { appName } : {}), ...(initiatorName ? { initiatorName } : {}), ...(applyTimeFrom ? { applyTimeFrom } : {}), ...(applyTimeTo ? { applyTimeTo } : {}), ...(instanceStatus ? { instanceStatus } : {}) } }) } ``` - [ ] **Step 2: Commit** ```bash git add web-admin/src/api/oaApproval.ts git commit -m "feat: pass OA task list filter params from web-admin API" ``` --- ### Task 5: 待办中心筛选 UI + 即时检索 **Files:** - Modify: `web-admin/src/views/oa-approval/pending/index.vue` - [ ] **Step 1: 在 `panel-top` 与 `list-body` 之间插入筛选区** ```vue