zkxw/backend/app/services/web_auth_roles.py

"""Web 管理端登录：基于 V0.3 sys_user / sys_user_role / sys_role 的角色与菜单权限解析

有效角色 = 用户直接角色 ∪ 经启用职位继承的角色（sys_post_role）。
"""
from __future__ import annotations

from typing import List, Optional, Set, Collection

from sqlalchemy import and_
from sqlalchemy.orm import Session

from app.core.config import settings
from app.models.models import (
    SysMenu,
    SysPost,
    SysPostRole,
    SysRole,
    SysRoleMenu,
    SysUserPost,
    SysUserRole,
)

# 无有效角色绑定时的 Web 端角色（不可为 super_admin，避免越权）
NO_ROLE_WEB_ROLE = "guest"

# 一级菜单权限码（用于顶部菜单权限控制）
TOP_MENU_PERMS = frozenset({"menu:market", "menu:production", "menu:quality", "menu:operations", "menu:system-manage"})


def _map_role_name_to_web(role_name: str) -> str:
    """将 sys_role.role_name 映射为 web-admin 使用的 role 字符串。"""
    n = (role_name or "").strip()
    if not n:
        return "admin"
    if "超级" in n or n in ("系统管理员", "超级管理员"):
        return "super_admin"
    if "管理" in n or "运营" in n:
        return "admin"
    return "admin"


def _dedupe_role_id_rows(rows) -> List[str]:
    out: List[str] = []
    seen: Set[str] = set()
    for row in rows:
        rid = str(row[0]).strip() if row and row[0] else ""
        if rid and rid not in seen:
            seen.add(rid)
            out.append(rid)
    return out


def get_direct_enabled_role_ids(db: Session, user_id: str) -> List[str]:
    """用户直接绑定且启用的角色 ID（sys_user_role → sys_role）。"""
    if not user_id:
        return []
    tenant = settings.DEFAULT_TENANT_NO
    rows = (
        db.query(SysRole.id)
        .join(SysUserRole, SysUserRole.role_id == SysRole.id)
        .filter(
            SysUserRole.user_id == user_id,
            SysUserRole.del_flag == 0,
            SysUserRole.tenant_no == tenant,
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant,
        )
        .order_by(SysRole.create_time.desc())
        .all()
    )
    return _dedupe_role_id_rows(rows)


def get_post_inherited_role_ids(db: Session, user_id: str) -> List[str]:
    """经启用职位继承的启用角色 ID（sys_user_post → sys_post → sys_post_role → sys_role）。"""
    if not user_id:
        return []
    tenant = settings.DEFAULT_TENANT_NO
    rows = (
        db.query(SysRole.id)
        .join(SysPostRole, SysPostRole.role_id == SysRole.id)
        .join(
            SysPost,
            and_(
                SysPost.id == SysPostRole.post_id,
                SysPost.del_flag == 0,
                SysPost.status == "1",
                SysPost.tenant_no == tenant,
            ),
        )
        .join(
            SysUserPost,
            and_(
                SysUserPost.post_id == SysPost.id,
                SysUserPost.user_id == user_id,
                SysUserPost.del_flag == 0,
                SysUserPost.tenant_no == tenant,
            ),
        )
        .filter(
            SysPostRole.del_flag == 0,
            SysPostRole.tenant_no == tenant,
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant,
        )
        .order_by(SysRole.create_time.desc())
        .all()
    )
    return _dedupe_role_id_rows(rows)


def get_effective_role_ids(db: Session, user_id: str) -> List[str]:
    """用户有效角色 ID：直接绑定 ∪ 启用职位关联的启用角色（去重保序）。"""
    if not user_id:
        return []
    out: List[str] = []
    seen: Set[str] = set()
    for rid in get_direct_enabled_role_ids(db, user_id) + get_post_inherited_role_ids(
        db, user_id
    ):
        if rid not in seen:
            seen.add(rid)
            out.append(rid)
    return out


def _normalize_candidate_user_ids(
    candidate_user_ids: Optional[Collection[str]],
) -> Optional[List[str]]:
    if candidate_user_ids is None:
        return None
    return list(dict.fromkeys([str(u).strip() for u in candidate_user_ids if u and str(u).strip()]))


def list_user_ids_with_effective_role_name(
    db: Session,
    *,
    tenant_no: str,
    role_name: str,
    candidate_user_ids: Optional[Collection[str]] = None,
) -> Set[str]:
    """拥有指定角色名的用户 ID（直接绑定 ∪ 职位继承）。"""
    tenant = (tenant_no or "").strip()
    rn = (role_name or "").strip()
    if not tenant or not rn:
        return set()
    candidates = _normalize_candidate_user_ids(candidate_user_ids)
    if candidates is not None and not candidates:
        return set()

    direct_q = (
        db.query(SysUserRole.user_id)
        .join(SysRole, SysRole.id == SysUserRole.role_id)
        .filter(
            SysUserRole.del_flag == 0,
            SysUserRole.tenant_no == tenant,
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant,
            SysRole.role_name == rn,
        )
    )
    post_q = (
        db.query(SysUserPost.user_id)
        .join(
            SysPost,
            and_(
                SysPost.id == SysUserPost.post_id,
                SysPost.del_flag == 0,
                SysPost.status == "1",
                SysPost.tenant_no == tenant,
            ),
        )
        .join(
            SysPostRole,
            and_(
                SysPostRole.post_id == SysPost.id,
                SysPostRole.del_flag == 0,
                SysPostRole.tenant_no == tenant,
            ),
        )
        .join(
            SysRole,
            and_(
                SysRole.id == SysPostRole.role_id,
                SysRole.del_flag == 0,
                SysRole.status == "1",
                SysRole.tenant_no == tenant,
                SysRole.role_name == rn,
            ),
        )
        .filter(SysUserPost.del_flag == 0, SysUserPost.tenant_no == tenant)
    )
    if candidates is not None:
        direct_q = direct_q.filter(SysUserRole.user_id.in_(candidates))
        post_q = post_q.filter(SysUserPost.user_id.in_(candidates))

    out: Set[str] = set()
    for row in direct_q.all():
        uid = str(row[0]).strip() if row and row[0] else ""
        if uid:
            out.add(uid)
    for row in post_q.all():
        uid = str(row[0]).strip() if row and row[0] else ""
        if uid:
            out.add(uid)
    return out


def list_user_ids_with_any_effective_role_ids(
    db: Session,
    *,
    tenant_no: str,
    role_ids: List[str],
    candidate_user_ids: Optional[Collection[str]] = None,
) -> List[str]:
    """拥有任一指定角色 ID 的用户 ID 列表（直接绑定 ∪ 职位继承，去重）。"""
    tenant = (tenant_no or "").strip()
    rids = list(dict.fromkeys([str(r).strip() for r in (role_ids or []) if r and str(r).strip()]))
    if not tenant or not rids:
        return []
    candidates = _normalize_candidate_user_ids(candidate_user_ids)
    if candidates is not None and not candidates:
        return []

    direct_q = (
        db.query(SysUserRole.user_id)
        .join(SysRole, SysRole.id == SysUserRole.role_id)
        .filter(
            SysUserRole.del_flag == 0,
            SysUserRole.tenant_no == tenant,
            SysUserRole.role_id.in_(rids),
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant,
        )
    )
    post_q = (
        db.query(SysUserPost.user_id)
        .join(
            SysPost,
            and_(
                SysPost.id == SysUserPost.post_id,
                SysPost.del_flag == 0,
                SysPost.status == "1",
                SysPost.tenant_no == tenant,
            ),
        )
        .join(
            SysPostRole,
            and_(
                SysPostRole.post_id == SysPost.id,
                SysPostRole.del_flag == 0,
                SysPostRole.tenant_no == tenant,
                SysPostRole.role_id.in_(rids),
            ),
        )
        .join(
            SysRole,
            and_(
                SysRole.id == SysPostRole.role_id,
                SysRole.del_flag == 0,
                SysRole.status == "1",
                SysRole.tenant_no == tenant,
            ),
        )
        .filter(SysUserPost.del_flag == 0, SysUserPost.tenant_no == tenant)
    )
    if candidates is not None:
        direct_q = direct_q.filter(SysUserRole.user_id.in_(candidates))
        post_q = post_q.filter(SysUserPost.user_id.in_(candidates))

    out: List[str] = []
    seen: Set[str] = set()
    for row in list(direct_q.all()) + list(post_q.all()):
        uid = str(row[0]).strip() if row and row[0] else ""
        if uid and uid not in seen:
            seen.add(uid)
            out.append(uid)
    return out


def get_role_names_by_ids(db: Session, role_ids: List[str]) -> List[str]:
    """按给定角色 ID 返回启用角色名称（保序、去重）。"""
    ordered = list(dict.fromkeys([str(r).strip() for r in (role_ids or []) if r]))
    if not ordered:
        return []
    tenant = settings.DEFAULT_TENANT_NO
    rows = (
        db.query(SysRole.id, SysRole.role_name)
        .filter(
            SysRole.id.in_(ordered),
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant,
        )
        .all()
    )
    name_by_id = {
        str(rid).strip(): str(name).strip()
        for rid, name in rows
        if rid and name
    }
    out: List[str] = []
    seen: Set[str] = set()
    for rid in ordered:
        name = name_by_id.get(rid)
        if name and name not in seen:
            seen.add(name)
            out.append(name)
    return out


def _active_role_name_rows(db: Session, user_id: str) -> list:
    """查询用户当前有效的 sys_role.role_name 列表（含职位继承）。"""
    return get_role_names_by_ids(db, get_effective_role_ids(db, user_id))


def user_has_active_roles(db: Session, user_id: str) -> bool:
    """用户是否至少拥有一个有效角色（直接或经职位继承）。"""
    return bool(get_effective_role_ids(db, user_id))


def get_user_role_names(db: Session, user_id: str) -> list:
    """用户当前有效的 sys_role.role_name 列表（去重、保序）。"""
    names = _active_role_name_rows(db, user_id)
    return list(dict.fromkeys(names))


def resolve_web_role_for_user(db: Session, user_id: str) -> str:
    """根据用户有效角色解析 Web 端 role（admin / super_admin / guest）。"""
    names = _active_role_name_rows(db, user_id)
    if not names:
        return NO_ROLE_WEB_ROLE
    mapped = [_map_role_name_to_web(n) for n in names]
    if "super_admin" in mapped:
        return "super_admin"
    return "admin"


def iter_user_menu_perm_codes(db: Session, user_id: str) -> Set[str]:
    """用户通过有效角色关联得到的菜单权限标识（sys_menu.perms，空则退回菜单 id）。

    注：不过滤 visible——F 类型按钮等 visible=0 的菜单仍可通过角色授权生效；
    visible 仅控制侧栏/菜单树展示，不影响权限码解析。
    一级菜单（menu:market 等）同样经本函数返回，用于前端顶部菜单权限控制。
    """
    if not user_id:
        return set()
    role_ids = get_effective_role_ids(db, user_id)
    if not role_ids:
        return set()
    tenant = settings.DEFAULT_TENANT_NO
    try:
        q = (
            db.query(SysMenu.perms, SysMenu.id)
            .join(
                SysRoleMenu,
                and_(
                    SysRoleMenu.menu_id == SysMenu.id,
                    SysRoleMenu.del_flag == 0,
                    SysRoleMenu.tenant_no == tenant,
                    SysRoleMenu.role_id.in_(role_ids),
                ),
            )
            .filter(
                SysMenu.del_flag == 0,
                SysMenu.tenant_no == tenant,
                SysMenu.status == "1",
            )
            .distinct()
        )
        out: Set[str] = set()
        for perms, mid in q.all():
            p = (perms or "").strip()
            if p:
                out.add(p)
            elif mid:
                out.add(str(mid).strip())
        return out
    except Exception:
        return set()