业务模块数据权限运行时过滤 — 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: 将 apply_data_scope 从「系统管理 + 门户工作流 + 请假 Demo」扩展到 CRM、PL、BX、DB 及系统管理辅助接口，使角色 data_scope 在业务列表/详情/统计查询中一致生效。

Architecture: 在 app/core/data_scope.py 增强通用能力与 CRM 负责人字段适配；新增 assert_row_in_data_scope 供详情/CUD 校验；各模块 CRUD/Service 列表函数统一增加 user_info 参数并在查询末尾调用过滤；API 层传入当前用户上下文。PL/DB 通过 DbUserContext.to_user_info() 对齐门户 JWT 字段。公开 token 接口（/public/pl/*、/public/db/*）不在范围内。

Tech Stack: FastAPI · SQLAlchemy · Pydantic v2 · pytest · 现有 sys_role.data_scope / sys_role_data_scope_detail

依据: OpenSpec permission-conventions · 现状扫描（2026-06-28）

现状摘要

状态	模块	端点约数
✅ 列表已接入	系统管理 9 类主列表、门户工作流 5 类列表、请假列表	~20
⚠️ 部分接入	系统管理辅助下拉/详情、工作流写操作/详情	~15
❌ 未接入	CRM、PL、BX、DB、系统监控	~385

过滤依据字段（默认）: create_dept_id（部门范围）+ create_by（仅本人）

CRM 额外字段: 客户/线索/商机/合同/报价等含 owner_user_id，「仅本人/自定义用户」应同时匹配 owner_user_id。

文件结构（新增/核心修改）

文件	职责
`backend/app/core/data_scope.py`	扩展 `apply_data_scope`、新增 `assert_row_in_data_scope`、`build_user_info_from_context`
`backend/app/core/db_auth.py`	`DbUserContext.to_user_info()` 供 PL/DB 复用
`backend/tests/test_data_scope.py`	数据权限单元测试（新建）
`backend/app/crud/crm/*.py`	CRM 各域 list/query 接入
`backend/app/api/crm*.py`	传入 `user_info`；详情/CUD 校验
`backend/app/services/pl_.py` / `backend/app/api/pl/.py`	PL 列表/详情接入
`backend/app/services/bx/.py` / `backend/app/api/bx/.py`	BX 列表/详情接入
`backend/app/api/db/*.py`	DB 列表/详情接入
`backend/app/api/admin.py`	补全辅助接口与详情校验
`backend/app/crud/monitor/oper_log.py`	操作日志列表过滤（可选按 `oper_name`）

设计决策（实施前确认）

D1 — CRM「仅本人」语义

规则: data_scope=4 时，CRM 主实体（客户/线索/商机/合同/报价）过滤条件为
(owner_user_id = 当前用户) OR (create_by = 当前用户)
子表（跟进记录、产品行等）仍按 create_by。
实现: apply_data_scope(query, model, user_info, db, tenant_no, owner_field="owner_user_id") 可选参数。

D2 — 公海/池规则/产品主数据

实体	策略
公海客户 `list_pool_customers`	不按 owner 过滤；全员可见公海（仅租户隔离）
公海规则 `pool_rule`	管理配置，`create_by` 过滤
产品/产品分类	主数据，建议 `data_scope=1` 角色可见全部；列表仍走 `apply_data_scope`
统计接口	与对应 list 使用相同 base_query 再聚合

D3 — BX「我的/待办/全部」

接口类型	策略
`/reimburses/my`、`/advances/my`	在「我的」语义上 AND `apply_data_scope`（取交集，防止越权）
`/pending-approval`	审批人维度为主，不叠加 data_scope（审批可见性独立）
`/reimburses` 管理列表	必须 `apply_data_scope`

D4 — PL/DB 用户上下文

扩展 DbUserContext.to_user_info():

def to_user_info(self) -> dict:
    return {
        "user_id": self.id,
        "role": self.role,
        "tenant_no": self.tenant_no,
        "dept_id": self.dept_id,
        "region_id": self.dept_id,
    }

API 层：user_info = user.to_user_info() 传入 service。

D5 — 详情/CUD 校验

列表过滤不足以防 ID 枚举。新增:

def assert_row_in_data_scope(db, row, model, user_info, tenant_no, owner_field=None) -> None:
    """不在范围内则 raise HTTPException(403)"""

所有 get_by_id / update / delete 在业务校验后调用。

Phase 0 — 基础设施（必须先完成）

Task 0.1: 扩展 data_scope 核心

Files:

Modify: backend/app/core/data_scope.py
Create: backend/tests/test_data_scope.py
[ ] Step 1: 编写失败测试 — tests/test_data_scope.py

import pytest
from unittest.mock import MagicMock, patch
from app.core.data_scope import apply_data_scope, assert_row_in_data_scope


class _FakeModel:
    create_dept_id = "dept-a"
    create_by = "user-1"
    owner_user_id = "user-2"


def test_scope4_filters_by_create_by_when_no_owner_field():
    query = MagicMock()
    user_info = {"user_id": "user-1", "role": "admin", "dept_id": "dept-a"}
    with patch("app.core.data_scope._resolve_user_role_scopes", return_value=[{"data_scope": "4", "details": []}]):
        apply_data_scope(query, _FakeModel, user_info, MagicMock(), "shanxia2.0")
    assert query.filter.called


def test_super_admin_bypass():
    query = MagicMock()
    result = apply_data_scope(query, _FakeModel, {"role": "super_admin"}, MagicMock(), "shanxia2.0")
    assert result is query
    query.filter.assert_not_called()

Step 2: 运行测试确认 FAIL

cd backend && source venv/bin/activate && pytest tests/test_data_scope.py -v

Step 3: 实现扩展

在 apply_data_scope 签名增加 owner_field: str | None = None；scope=4 且 model 有 owner_field 时:

if scope == "4":
    user_ids.add(user_id)
    if owner_field and hasattr(model, owner_field):
        conditions.append(getattr(model, owner_field).in_(user_ids))
    elif has_user:
        conditions.append(model.create_by.in_(user_ids))

新增 assert_row_in_data_scope：复用 scope 解析，对单行做 membership 检查。

[ ] Step 4: 测试 PASS
[ ] Step 5: Commit — feat: extend data_scope for owner field and row assert

Task 0.2: DbUserContext 对齐

Files:

Modify: backend/app/core/db_auth.py
[ ] Step 1: 新增 to_user_info()（见 D4），保留 to_tenant_user_dict() 兼容旧调用。
[ ] Step 2: Commit — feat: DbUserContext.to_user_info for data scope

Task 0.3: CRUD 接入模板（文档化约定）

约定（后续所有 Task 遵循）:

列表函数增加 user_info=None 参数（已有则跳过）。
租户过滤 → 业务条件 → if user_info: q = apply_data_scope(q, Model, user_info, db, tenant_no[, owner_field=...]) → count/paginate。
API 路由将 current_user / DbUserContext.to_user_info() 传入 CRUD。
详情/更新/删除：取 row 后 assert_row_in_data_scope(...)。

Phase 1 — 系统管理补全（⚠️ 部分改造）

预估: 0.5 人日

Task 1.1: 辅助接口

Files:

Modify: backend/app/api/admin.py
Modify: backend/app/crud/user.py（如需）

接口	改造
`GET /users/options`	查询后 `apply_data_scope(SysUser, ...)`
`GET /roles/options`	已有 tenant；可选 `apply_data_scope(SysRole, ...)`
`GET /dict-data/options`	`list_dict_data_options` 增加 `user_info`
`GET /dict-data/all`	改为走 `list_dict_data` + user_info

逐项改造并手动验证下拉数据量随角色变化。

Task 1.2: 详情接口校验

Files:

Modify: backend/app/api/admin.py
[ ] GET /users/{id}: 取 user 后 assert_row_in_data_scope(db, row, SysUser, user, tenant_no)，403 替代仅 tenant 检查。
[ ] Commit — feat: data scope on admin auxiliary and user detail

Phase 2 — CRM（最高优先级）

预估: 5–8 人日 · ~228 端点 · 按域分批 PR

Task 2.1: 客户域（试点）

Files:

Modify: backend/app/crud/crm/customer.py
Modify: backend/app/crud/crm/customer_pool.py
Modify: backend/app/crud/crm/customer_follow.py
Modify: backend/app/crud/crm/pool_rule.py
Modify: backend/app/api/crm.py（customer / pool / follow / pool/rule 相关路由）
[ ] list_customers: 增加 user_info，owner_field="owner_user_id"（排除 list_pool_customers 不过 data_scope）。
[ ] get_customer / update / delete: assert_row_in_data_scope。
[ ] API customer/list、customer/detail 等传入 _user 的 user_info。
[ ] 保留 customer/mine 等接口原有 owner 过滤，与 data_scope AND。
[ ] 手工用例：角色 data_scope=4 仅能看到本人负责客户；=2 能看到本部门。

Task 2.2: 线索域

Files:

Modify: backend/app/crud/crm/clue.py, clue_follow.py, clue_statistics.py
Modify: backend/app/api/crm.py（clue 段）
[ ] 同 Task 2.1 模式，owner_field="owner_user_id"。
[ ] 统计 clue_statistics/*: base query 先 apply_data_scope 再聚合。

Task 2.3: 商机域

Files:

Modify: backend/app/crud/crm/opportunity.py, opportunity_follow.py, opportunity_product.py
Modify: backend/app/api/crm.py（opportunity 段）

Task 2.4: 联系人 / 供应商 / 开票

Files:

Modify: backend/app/crud/crm/contact.py, supplier_info.py, invoice_info.py
Modify: backend/app/api/crm.py
[ ] 联系人按 customer_id 关联时：先校验客户 in scope，或 contact 自身 create_by。

Task 2.5: 产品域

Files:

Modify: backend/app/crud/crm/product.py
Modify: backend/app/api/crm.py（product / product-categories）
[ ] 仅 create_by / create_dept_id，无 owner_field。

Task 2.6: 报价域

Files:

Modify: backend/app/crud/crm/quote.py, quote_product.py, quote_approval.py, quote_share_log.py
Modify: backend/app/api/crm_quote_routes.py
[ ] owner_field="owner_user_id"（报价负责人）。

Task 2.7: 合同域

Files:

Modify: backend/app/crud/crm/contract.py, contract_product.py, contract_approval.py
Modify: backend/app/api/crm_contract_routes.py, crm_contract_sub_routes.py
[ ] 合同主表 owner_field="owner_user_id"；变更/模板/审批子模块按 create_by。
[ ] Phase 2 验收: CRM 各 list 在 Postman 用不同 data_scope 角色对比 total；detail 越权 ID 返回 403。

Phase 3 — 运输计划 PL

预估: 3–4 人日 · ~60 端点（不含 public）

Task 3.1: 基础数据

Files:

Modify: backend/app/api/pl/base_data.py（carrier / vehicle / personnel / driver 列表查询段）
Modify: backend/app/services/pl_personnel_service.py（如有独立 query）
[ ] 各 GET 列表：user_info = user.to_user_info() + apply_data_scope。
[ ] 详情/改/删：assert_row_in_data_scope。

Task 3.2: 计划 / 运输 / 运费 / 盘存

Files:

Modify: backend/app/api/pl/plan.py
Modify: backend/app/api/pl/transport.py
Modify: backend/app/api/pl/freight.py
Modify: backend/app/services/pl_freight_settlement_service.py
Modify: backend/app/services/pl_plan_service.py
Modify: backend/app/api/pl/inventory.py
[ ] 优先改造：plan 列表、freight 结算单列表、transport 列表。
[ ] 审批/待办类接口：按产品确认是否叠加 scope（默认：列表加，待办不加）。
[ ] 排除: public_site.py, public_transport.py（无 JWT）。

Phase 4 — 报销 BX

预估: 2–3 人日 · ~64 端点

Task 4.1: 费用项 + 流程配置

Files:

Modify: backend/app/services/bx/expense_item.py, bx/workflow.py
Modify: backend/app/api/bx/expense_item.py, bx/workflow.py

Task 4.2: 备用金 + 报销 + 支付

Files:

Modify: backend/app/services/bx/advance.py, bx/reimburse.py, bx/payment.py
Modify: backend/app/api/bx/advance.py, bx/reimburse.py, bx/payment.py
[ ] 管理列表 /advances、/reimburses（非 my）：apply_data_scope。
[ ] /my 系列：(create_by = me) AND apply_data_scope。
[ ] /pending-approval：保持审批人逻辑，不叠加 scope（见 D3）。

Task 4.3: 附件

Files:

Modify: backend/app/services/bx/attachment.py, backend/app/api/bx/attachment.py
[ ] 按关联业务单校验 scope（下载/删除时 assert 主单 in scope）。

Phase 5 — 地磅 DB

预估: 2 人日 · ~28 端点（不含 public）

Task 5.1: 出入库 / 计划 / 过磅 / 看板

Files:

Modify: backend/app/api/db/inventory.py
Modify: backend/app/api/db/plans.py
Modify: backend/app/api/db/weighing.py
Modify: backend/app/api/db/dashboard.py
[ ] 统一 user.to_user_info() 传入查询。
[ ] Dashboard 统计基于已过滤子查询。
[ ] 排除: public_inventory.py。

Phase 6 — 系统监控 + 门户工作流补全

预估: 1 人日

Task 6.1: 操作日志

Files:

Modify: backend/app/crud/monitor/oper_log.py
Modify: backend/app/api/monitor.py
[ ] 若日志表无 create_dept_id：按 oper_user_id 映射为等效 create_by 过滤，或文档声明「监控仅 super_admin」。

Task 6.2: 工作流详情/写操作

Files:

Modify: backend/app/api/workflow.py
Modify: backend/app/crud/workflow.py
[ ] GET /instances/{id}、POST /tasks/{id}/approve 前校验实例/任务关联业务或 create_by in scope。

Phase 7 — 测试与回归

Task 7.1: 自动化测试

Files:

Create/Expand: backend/tests/test_data_scope.py
[ ] 覆盖：scope 1/2/3/4/5、多角色并集、super_admin 放行、CRM owner_field、空 scope 返回空集。

cd backend && pytest tests/test_data_scope.py -v

Task 7.2: 手工回归矩阵

角色 data_scope	模块	预期
1 全部	CRM 客户 list	租户内全量
4 仅本人	CRM 客户 list	仅 owner/create_by=本人
2 本部门	PL 计划 list	仅本部门 create_dept_id
5 自定义部门	系统用户 list	自定义部门用户
super_admin	任意	不过滤

Task 7.3: 字段一致性（可选）

cd backend && python scripts/check_field_consistency.py

排期建议

阶段	内容	人日	依赖
Phase 0	基础设施 + 测试	1	—
Phase 1	系统管理补全	0.5	Phase 0
Phase 2	CRM 全域	5–8	Phase 0
Phase 3	PL	3–4	Phase 0
Phase 4	BX	2–3	Phase 0
Phase 5	DB	2	Phase 0
Phase 6	监控 + 工作流补全	1	Phase 0
Phase 7	测试回归	1	All
合计		15–20

推荐 PR 切分（便于评审）:

feat/data-scope-core — Phase 0 + 1
feat/data-scope-crm-customer-clue — Phase 2.1–2.2
feat/data-scope-crm-opp-quote-contract — Phase 2.3–2.7
feat/data-scope-pl — Phase 3
feat/data-scope-bx — Phase 4
feat/data-scope-db-monitor-wf — Phase 5–6

风险与缓解

风险	缓解
CRM `/mine` 与 data_scope 双重过滤导致结果为空	集成测试 + 产品确认：mine = owner 条件 AND scope
历史数据 `create_dept_id` 为空	迁移脚本补全；空 dept 记录仅 create_by 命中
PL/DB service 层多处裸 query	优先抽 `build_scoped_query(model, user_info)` 减少遗漏
性能（scope=3 递归部门）	已有 `_get_child_dept_ids`；大数据量时可缓存部门树
公海/审批/待办语义冲突	按 D2/D3 写死规则并在 API docstring 注明

不在本次范围

小程序端菜单/数据权限（独立鉴权体系）
公开 token 接口（/public/pl/*、/public/db/*）
前端改造（后端过滤即可；前端无需改 unless 403 提示优化）
新角色类型或新 data_scope 枚举值

完成定义（Definition of Done）

backend/app/core/data_scope.py 含 owner_field + assert_row_in_data_scope
CRM / PL / BX / DB 所有 列表/统计 查询均调用 apply_data_scope（公开接口除外）
各模块 详情/更新/删除 均调用 assert_row_in_data_scope
系统管理辅助接口补全
pytest tests/test_data_scope.py 通过
手工回归矩阵至少覆盖 CRM + PL 各 1 条
./scripts/update-graph.sh backend 已执行

2026-06-28-data-scope-business-modules.md 16 KB Histórico Raw

业务模块数据权限运行时过滤 — Implementation Plan

现状摘要

文件结构（新增/核心修改）

设计决策（实施前确认）

D1 — CRM「仅本人」语义

D2 — 公海/池规则/产品主数据

D3 — BX「我的/待办/全部」

D4 — PL/DB 用户上下文

D5 — 详情/CUD 校验

Phase 0 — 基础设施（必须先完成）

Task 0.1: 扩展 data_scope 核心

Task 0.2: DbUserContext 对齐

Task 0.3: CRUD 接入模板（文档化约定）

Phase 1 — 系统管理补全（⚠️ 部分改造）

Task 1.1: 辅助接口

Task 1.2: 详情接口校验

Phase 2 — CRM（最高优先级）

Task 2.1: 客户域（试点）

Task 2.2: 线索域

Task 2.3: 商机域

Task 2.4: 联系人 / 供应商 / 开票

Task 2.5: 产品域

Task 2.6: 报价域

Task 2.7: 合同域

Phase 3 — 运输计划 PL

Task 3.1: 基础数据

Task 3.2: 计划 / 运输 / 运费 / 盘存

Phase 4 — 报销 BX

Task 4.1: 费用项 + 流程配置

Task 4.2: 备用金 + 报销 + 支付

Task 4.3: 附件

Phase 5 — 地磅 DB

Task 5.1: 出入库 / 计划 / 过磅 / 看板

Phase 6 — 系统监控 + 门户工作流补全

Task 6.1: 操作日志

Task 6.2: 工作流详情/写操作

Phase 7 — 测试与回归

Task 7.1: 自动化测试

Task 7.2: 手工回归矩阵

Task 7.3: 字段一致性（可选）

排期建议

风险与缓解

不在本次范围

完成定义（Definition of Done）

2026-06-28-data-scope-business-modules.md 16 KB

Histórico Raw