track-LICA/backend/app/services/data_scope_service.py

"""业务分组数据范围判定 —— 列表 / 扫码 / 统计三类接口共用的**唯一**判定来源

设计红线（照抄部门隔离那套的精神，别在这里走样）：

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)]