feat(分组权限): 业务分组模型 + DataScope 判定 + 列表接口接入
第一阶段：模型 → 判定 → /auth/me → 列表。统计接口与前端管理页随后。
1) 模型与迁移（head 从 k1l2m3n4o5p6 推进到 l1m2n3o4p5q6）
· business_groups 组定义，parent_id 表达「大组 > 小组」
· business_group_phases 可见范围，独立成表以支持多选 —— 需求要求
「范围可配置、不要写死」，单列存不下多个 phase
· business_group_members 成员，一人可属多组（这是「同时看生产+维修」的实现）
只建表、不写种子数据，所以可以先部署代码再建组。
2) DataScope 判定模块（app/services/data_scope_service.py）
全仓库唯一的权限谓词来源，业务代码里不准再出现 lifecycle_phase 过滤。
两条红线照抄部门隔离的教训：
· None(不限) 与 frozenset()(空) 语义相反，绝不共用一个哨兵值
· 空集合必须显式 false() —— SQLAlchemy 对 in_(()) 生 成 IN (NULL)，
一旦退化成不过滤就是全量泄漏
解析优先级：SUPER_ADMIN 硬放行（不可被分组覆盖）
> 显式分组（分组优先于角色）
> 未分组 SUPERVISOR 默认全厂
> 未分组普通用户
3) 过渡期开关 DATA_SCOPE_UNGROUPED（默认 ALL）
直接上严格模式会让所有未分组工人当场看不到自己的任务、现场停摆。
默认 ALL 先放行并打 WARNING 记录「谁还没分组」，配好组后再改 NONE。
4) /auth/me 返回 scope，phase 中文标签由服务端下发
—— 前端已有两份 phase 词表副本，不再加第三份。
5) 列表接口接入
· get_all_products：过滤加在 offset/limit 之前（其下有 6 段基于 product_ids
的批量预计算，过滤晚了等于算完再丢）
· get_all_tasks：谓词进【共享 filters】，保证 count 与 select 两条独立语句
同时生效，否则 total 与实际页不一致、移动端 hasMore 判断跟着错
· 扫码 get_product_by_serial 刻意不过滤，理由写死在 docstring 里
实测：空 scope 生成 false、受限 scope 生成 JOIN + IN 谓词；
/products 返回 2 条、/tasks 的 total 与 returned 一致。
This commit is contained in:
207
backend/app/services/data_scope_service.py
Normal file
207
backend/app/services/data_scope_service.py
Normal file
@ -0,0 +1,207 @@
|
||||
"""业务分组数据范围判定 —— 列表 / 扫码 / 统计三类接口共用的**唯一**判定来源
|
||||
|
||||
设计红线（照抄部门隔离那套的精神，别在这里走样）：
|
||||
|
||||
1. **宁可查不出，不可查过头。** 空范围必须落到 SQL 的 `false()`，绝不能因为
|
||||
写错而退化成「不过滤」—— 那就是全量泄漏。这是本模块存在的全部意义。
|
||||
|
||||
2. **`None`（不限）与 `frozenset()`（空）语义相反**，绝不能用同一个哨兵值表示。
|
||||
部门过滤当年踩过这个坑，这里显式区分。
|
||||
|
||||
3. **SQL 谓词只在本模块生成。** 业务代码里出现 `Product.lifecycle_phase.in_(...)`
|
||||
就是抄成了第二份口径 —— 评审时可用：
|
||||
grep -rn "lifecycle_phase.in_" backend/app/
|
||||
命中点应当**只有本文件一处**。
|
||||
|
||||
为什么放在 services 而不是 core：
|
||||
core/lifecycle.py 有明确约定「core 层不反向依赖 models」，而本模块需要
|
||||
AsyncSession 与三张 ORM 表。角色常量仍取自 app.core.roles、阶段常量取自
|
||||
app.core.lifecycle，单一事实来源不破。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from sqlalchemy import false, select, true
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.lifecycle import LIFECYCLE_PRODUCTION, PHASE_LABELS
|
||||
from app.core.roles import SUPER_ADMIN, SUPERVISOR
|
||||
from app.models.business_group import (
|
||||
BusinessGroup,
|
||||
BusinessGroupMember,
|
||||
BusinessGroupPhase,
|
||||
)
|
||||
from app.models.product import Product
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DataScope:
|
||||
"""当前登录用户能看到的数据范围。
|
||||
|
||||
phases:
|
||||
None → 不限（全厂）：SUPER_ADMIN，或未分组的 SUPERVISOR
|
||||
frozenset() → 空范围：未分组的普通用户。返回空列表，**不是报错**
|
||||
frozenset({...}) → 只含这些 lifecycle_phase
|
||||
"""
|
||||
phases: frozenset[str] | None = None
|
||||
group_ids: frozenset[int] = field(default_factory=frozenset)
|
||||
leader_of: frozenset[int] = field(default_factory=frozenset)
|
||||
reason: str = "" # 仅用于日志与 /auth/me 展示：super_admin / grouped / supervisor_default / ungrouped
|
||||
|
||||
# ---- 语义查询 ----
|
||||
|
||||
@property
|
||||
def is_unrestricted(self) -> bool:
|
||||
"""不限范围（全厂）。注意与 is_empty 语义相反，别混用。"""
|
||||
return self.phases is None
|
||||
|
||||
@property
|
||||
def is_empty(self) -> bool:
|
||||
"""有范围但范围为空 —— 什么都看不到。"""
|
||||
return self.phases is not None and not self.phases
|
||||
|
||||
def allows_phase(self, phase: str | None) -> bool:
|
||||
"""Python 侧判定（需要按单个产品/任务判断时用，与 SQL 侧同口径）"""
|
||||
if self.phases is None:
|
||||
return True
|
||||
return (phase or LIFECYCLE_PRODUCTION) in self.phases
|
||||
|
||||
# ---- SQL 谓词 ----
|
||||
|
||||
def product_where(self):
|
||||
"""Product 为主体（FROM products，或已 JOIN 到 products）的语句用。
|
||||
|
||||
⚠️ 空集合必须显式 false()。SQLAlchemy 对 `in_(())` 生成 `IN (NULL)`
|
||||
并抛 SAWarning，且历史版本行为有过变化 —— 一旦退化成不过滤就是
|
||||
全量泄漏，这是本方法存在的核心理由。
|
||||
"""
|
||||
if self.phases is None:
|
||||
return true()
|
||||
if not self.phases:
|
||||
return false()
|
||||
return Product.lifecycle_phase.in_(tuple(sorted(self.phases)))
|
||||
|
||||
def task_where(self):
|
||||
"""Task 为主体的语句用。
|
||||
|
||||
⚠️ 前置条件：调用方**必须已经** join 了 Product
|
||||
（`.join(Product, Task.product_id == Product.id)`）。
|
||||
tasks 表本身没有 lifecycle_phase，漏 join 会变成笛卡尔积或直接报错。
|
||||
|
||||
谓词内容与 product_where 相同，**分开命名只为把这个无法用类型系统
|
||||
表达的前置条件写在调用点** —— `grep "task_where"` 就能一次找出所有
|
||||
需要检查 join 的地方。
|
||||
"""
|
||||
return self.product_where()
|
||||
|
||||
|
||||
async def resolve_data_scope(db: AsyncSession, current_user: dict) -> DataScope:
|
||||
"""从 JWT payload 解析数据范围。
|
||||
|
||||
规则（顺序即优先级，改动前先跟业务确认）：
|
||||
|
||||
1. `SUPER_ADMIN` → 永远全厂。**硬编码，不可被分组覆盖** —— 防止管理员
|
||||
被误分进某个组后突然失去全部视野。
|
||||
2. 被显式分进组 → 该组生效。**分组优先于角色**：连 SUPERVISOR 被分进组
|
||||
也会受限（这是刻意的，见 groups.py 为何只有超管能改分组）。
|
||||
3. 未分组的 `SUPERVISOR`（部门主管）→ 默认全厂。
|
||||
4. 未分组的普通用户 → 空范围，或按 DATA_SCOPE_UNGROUPED 开关过渡放行。
|
||||
"""
|
||||
role = (current_user or {}).get("role") or ""
|
||||
username = (current_user or {}).get("username") or ""
|
||||
|
||||
# ---- 规则 1：超管硬放行 ----
|
||||
if role == SUPER_ADMIN:
|
||||
return DataScope(phases=None, reason="super_admin")
|
||||
|
||||
# ---- 规则 2：查我所属的活跃组（走 ix_business_group_members_user_id）----
|
||||
rows = []
|
||||
if username:
|
||||
result = await db.execute(
|
||||
select(
|
||||
BusinessGroup.id,
|
||||
BusinessGroup.parent_id,
|
||||
BusinessGroupMember.is_leader,
|
||||
)
|
||||
.join(BusinessGroup, BusinessGroup.id == BusinessGroupMember.group_id)
|
||||
.where(
|
||||
BusinessGroupMember.user_id == username,
|
||||
BusinessGroup.is_active.is_(True), # 停用组等同不存在
|
||||
)
|
||||
)
|
||||
rows = result.all()
|
||||
|
||||
if rows:
|
||||
# 范围继承：小组自己配了就用小组的，没配则向上取父组的。
|
||||
# 这样「生产大组配一次 PRODUCTION，下面的生产/测试小组都不用再配」。
|
||||
lookup_ids = {r[0] for r in rows} | {r[1] for r in rows if r[1]}
|
||||
phase_rows = await db.execute(
|
||||
select(BusinessGroupPhase.group_id, BusinessGroupPhase.phase)
|
||||
.where(BusinessGroupPhase.group_id.in_(lookup_ids))
|
||||
)
|
||||
by_group: dict[int, set[str]] = defaultdict(set)
|
||||
for gid, ph in phase_rows.all():
|
||||
if ph:
|
||||
by_group[gid].add(ph)
|
||||
|
||||
phases: set[str] = set()
|
||||
for gid, parent_id, _is_leader in rows:
|
||||
own = by_group.get(gid)
|
||||
if own:
|
||||
phases |= own
|
||||
elif parent_id:
|
||||
phases |= by_group.get(parent_id, set())
|
||||
|
||||
return DataScope(
|
||||
phases=frozenset(phases),
|
||||
group_ids=frozenset(r[0] for r in rows),
|
||||
leader_of=frozenset(r[0] for r in rows if r[2]),
|
||||
reason="grouped",
|
||||
)
|
||||
|
||||
# ---- 规则 3：未分组的主管 → 默认全厂 ----
|
||||
if role == SUPERVISOR:
|
||||
return DataScope(phases=None, reason="supervisor_default")
|
||||
|
||||
# ---- 规则 4：未分组的普通用户 ----
|
||||
if settings.DATA_SCOPE_UNGROUPED.upper() == "NONE":
|
||||
return DataScope(phases=frozenset(), reason="ungrouped")
|
||||
|
||||
# 过渡期：先放行，但把「谁还没分组」记下来 —— 这是把开关安全翻到 NONE 的前提
|
||||
logger.warning(
|
||||
"data_scope.ungrouped（过渡期默认放行，请尽快完成分组）",
|
||||
extra={"extra_fields": {"user": username, "role": role}},
|
||||
)
|
||||
return DataScope(phases=None, reason="ungrouped_fallback")
|
||||
|
||||
|
||||
async def scope_group_names(db: AsyncSession, scope: DataScope) -> list[str]:
|
||||
"""当前范围对应的组显示名 —— 供 /auth/me 让前端展示「我为什么只看到这些」"""
|
||||
if not scope.group_ids:
|
||||
return []
|
||||
result = await db.execute(
|
||||
select(BusinessGroup.name)
|
||||
.where(BusinessGroup.id.in_(tuple(scope.group_ids)))
|
||||
.order_by(BusinessGroup.sort_order, BusinessGroup.id)
|
||||
)
|
||||
return [r[0] for r in result.all()]
|
||||
|
||||
|
||||
def scope_phase_labels(scope: DataScope) -> list[str]:
|
||||
"""范围的中文标签。
|
||||
|
||||
刻意由服务端下发，避免前端再抄一份 phase 词表 —— 前端已有
|
||||
constants/task.ts 与 track-uniapp/utils/lifecycle.js 两份副本，
|
||||
不要再加第三份。
|
||||
"""
|
||||
if scope.phases is None:
|
||||
return ["全厂"]
|
||||
if not scope.phases:
|
||||
return ["未分组"]
|
||||
return [PHASE_LABELS.get(p, p) for p in sorted(scope.phases)]
|
||||
@ -19,6 +19,8 @@ from app.models.production_order import ProductionOrder
|
||||
from app.models.task import Task
|
||||
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse, ProductScanResponse
|
||||
from app.schemas.task import TaskSummaryResponse, TaskResponse, TaskRecordResponse
|
||||
# 业务分组数据范围 —— 只借用类型，谓词一律由 DataScope 生成，不在这里手写 IN 条件
|
||||
from app.services.data_scope_service import DataScope
|
||||
|
||||
|
||||
def _task_to_response(task: Task) -> TaskResponse:
|
||||
@ -62,7 +64,21 @@ async def _load_task_tree(db: AsyncSession, product_id: uuid.UUID) -> list[TaskR
|
||||
|
||||
|
||||
async def get_product_by_serial(db: AsyncSession, serial_number: str) -> ProductScanResponse:
|
||||
"""扫码查询：根据 16 位序列号查出产品 + 所属订单 + 完整任务树"""
|
||||
"""扫码查询：根据 16 位序列号查出产品 + 所属订单 + 完整任务树。
|
||||
|
||||
⚠️ 本函数**刻意不接受 DataScope、也不做业务分组过滤** —— 这是与
|
||||
「列表按组过滤」同等重要的设计决策，不是漏改：
|
||||
|
||||
· 维修组扫到一台生产中的设备必须能【看】。现场要判断的恰恰是
|
||||
「这台机器是不是走错了流程 / 该不该到我这儿」，看不见就无法判断。
|
||||
· 「不能操作」由操作类接口各自的属主校验（task_service._check_permission）
|
||||
保证，不在查询层做。
|
||||
· 在这里加范围过滤会让维修工扫自己的设备都可能 404 —— 功能性倒退。
|
||||
· 本函数另有 3 个内部调用点（product_finalize_service 两处、
|
||||
product_service 内部一处），加必传参数会波及它们。
|
||||
|
||||
如需调整这个决策，请先跟业务确认「跨组扫码要能看」这条是否仍然成立。
|
||||
"""
|
||||
result = await db.execute(
|
||||
select(Product)
|
||||
.options(
|
||||
@ -524,6 +540,7 @@ def _lookup_display_names(location_ids: list[str]) -> dict[str, str]:
|
||||
|
||||
async def get_all_products(
|
||||
db: AsyncSession,
|
||||
scope: DataScope,
|
||||
skip: int = 0,
|
||||
limit: int = 50,
|
||||
keyword: str | None = None,
|
||||
@ -534,6 +551,9 @@ async def get_all_products(
|
||||
|
||||
keyword: 同时模糊匹配 serial_number (产品身份证)、material_name/id (规格型号)、order_no (订单号)
|
||||
status_filter: 按产品状态过滤 (如 PENDING / WIP / COMPLETED / ARCHIVED)
|
||||
|
||||
⚠️ scope **不给默认值**：它是业务分组的权限边界，漏传必须当场 TypeError，
|
||||
而不是悄悄退化成「不过滤」——那等于全量泄漏。
|
||||
"""
|
||||
stmt = select(Product).options(selectinload(Product.order))
|
||||
|
||||
@ -620,6 +640,14 @@ async def get_all_products(
|
||||
# 兜底：其他状态码按"存在该状态任务"匹配（未完结）
|
||||
stmt = stmt.where(not_finished, _has_task_status(sf))
|
||||
|
||||
# 🚀 业务分组数据范围过滤。
|
||||
# 位置很关键：必须加在 offset/limit **之前**，因为下面有 6 段基于
|
||||
# product_ids 的批量预计算（macro_status / 最新任务 / 主干工序 / 滞留时长 /
|
||||
# 生产天数），过滤晚了等于把这些算完再丢掉，纯浪费。
|
||||
# 注：上面的 keyword 分支带了 .distinct()，在其后追加 .where() 语义正确
|
||||
# （distinct 是整体修饰，不是"立即去重"）。
|
||||
stmt = stmt.where(scope.product_where())
|
||||
|
||||
stmt = stmt.offset(skip).limit(limit).order_by(Product.created_at.desc())
|
||||
|
||||
result = await db.execute(stmt)
|
||||
|
||||
@ -23,6 +23,8 @@ from app.core.lifecycle import (
|
||||
from app.models.product import Product
|
||||
from app.models.task_log import TaskLog
|
||||
from app.core.roles import ADMIN_ROLES
|
||||
# 业务分组数据范围 —— 只借用类型，谓词一律由 DataScope 生成，不在这里手写 IN 条件
|
||||
from app.services.data_scope_service import DataScope
|
||||
from app.schemas.task import (
|
||||
TaskCreate,
|
||||
TaskUpdate,
|
||||
@ -455,22 +457,38 @@ async def update_task(db: AsyncSession, task_id: uuid.UUID, data: TaskUpdate) ->
|
||||
|
||||
|
||||
async def get_all_tasks(
|
||||
db: AsyncSession, product_id: uuid.UUID | None = None,
|
||||
db: AsyncSession, scope: DataScope,
|
||||
product_id: uuid.UUID | None = None,
|
||||
assignee_id: str | None = None, skip: int = 0, limit: int = 50
|
||||
) -> TaskListResponse:
|
||||
"""获取任务列表，可按产品/负责人筛选"""
|
||||
"""获取任务列表，可按产品/负责人筛选。
|
||||
|
||||
⚠️ tasks 表**没有** lifecycle_phase —— 业务分组的范围判定必须 join(Product)。
|
||||
所以仅在 scope 受限时补 join；不限范围（超管）时不加，避免给每次请求
|
||||
白搭一个 join。
|
||||
|
||||
⚠️ scope **不给默认值**：它是权限边界，漏传要当场 TypeError，不能静默全量。
|
||||
"""
|
||||
filters = []
|
||||
if product_id:
|
||||
filters.append(Task.product_id == product_id)
|
||||
if assignee_id:
|
||||
filters.append(Task.assignee_id == assignee_id)
|
||||
|
||||
# 🚀 数据范围谓词进【共享的 filters】—— 下面 count 与 select 是两条独立语句，
|
||||
# 谓词只写一次就天然同时生效。若只给其中一条加过滤，total 会与实际页内容
|
||||
# 不一致，移动端 hasMore（靠 tasks.length < total）判断跟着错。
|
||||
needs_join = scope.phases is not None
|
||||
if needs_join:
|
||||
filters.append(scope.task_where())
|
||||
|
||||
# 总数必须独立 COUNT：移动端「我的任务」用 total 判断 hasMore
|
||||
# （tasks.length < total），若 total 取当前页条数，首页满员时
|
||||
# hasMore 恒为 false，列表永远停在第一页。
|
||||
total = await db.scalar(
|
||||
select(func.count()).select_from(Task).where(*filters)
|
||||
) or 0
|
||||
total_stmt = select(func.count()).select_from(Task)
|
||||
if needs_join:
|
||||
total_stmt = total_stmt.join(Product, Task.product_id == Product.id)
|
||||
total = await db.scalar(total_stmt.where(*filters)) or 0
|
||||
|
||||
stmt = (
|
||||
select(Task)
|
||||
@ -483,6 +501,8 @@ async def get_all_tasks(
|
||||
.limit(limit)
|
||||
.order_by(Task.created_at.desc())
|
||||
)
|
||||
if needs_join:
|
||||
stmt = stmt.join(Product, Task.product_id == Product.id)
|
||||
|
||||
result = await db.execute(stmt)
|
||||
tasks = result.scalars().all()
|
||||
|
||||
Reference in New Issue
Block a user