Neo-ZQYY/apps/backend/app/services/role.py

# -*- coding: utf-8 -*-
"""
角色权限服务 —— 查询用户在指定店铺下的角色和权限。

职责：
- get_user_permissions()：获取用户在指定 site_id 下的权限 code 列表
- get_user_sites()：获取用户关联的所有店铺及对应角色
- check_user_has_site_role()：检查用户在指定 site_id 下是否有任何角色绑定

所有数据库操作使用 psycopg2 原生 SQL，不引入 ORM。
"""

from __future__ import annotations

import logging

from app.database import get_connection
from app.trace.decorators import trace_service

logger = logging.getLogger(__name__)


@trace_service(description_zh="获取用户权限列表", description_en="Get user permissions")
async def get_user_permissions(user_id: int, site_id: int) -> list[str]:
    """
    获取用户在指定 site_id 下的权限 code 列表。

    通过 user_site_roles → role_permissions → permissions 三表联查，
    返回去重后的权限 code 列表。

    参数:
        user_id: 用户 ID
        site_id: 门店 ID

    返回:
        权限 code 字符串列表，如 ["view_tasks", "view_board"]
    """
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT DISTINCT p.code
                  FROM auth.user_site_roles usr
                  JOIN auth.role_permissions rp ON usr.role_id = rp.role_id
                  JOIN auth.permissions p ON rp.permission_id = p.id
                 WHERE usr.user_id = %s AND usr.site_id = %s
                   AND usr.is_removed = false
                """,
                (user_id, site_id),
            )
            rows = cur.fetchall()
    finally:
        conn.close()

    return [row[0] for row in rows]


@trace_service(description_zh="获取用户门店列表", description_en="Get user sites")
async def get_user_sites(user_id: int) -> list[dict]:
    """
    获取用户关联的所有店铺及对应角色。

    查询 user_site_roles JOIN roles，LEFT JOIN biz.sites 获取店铺名称，
    按 site_id 分组聚合角色列表。

    参数:
        user_id: 用户 ID

    返回:
        [{"site_id": int, "site_name": str, "roles": [{"code": str, "name": str}]}]
    """
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT usr.site_id,
                       COALESCE(scm.site_name, '') AS site_name,
                       r.code,
                       r.name
                  FROM auth.user_site_roles usr
                  JOIN auth.roles r ON usr.role_id = r.id
                  LEFT JOIN biz.sites scm ON scm.site_id = usr.site_id
                 WHERE usr.user_id = %s
                   AND usr.is_removed = false
                 ORDER BY usr.site_id, r.code
                """,
                (user_id,),
            )
            rows = cur.fetchall()
    finally:
        conn.close()

    # 按 site_id 分组聚合
    sites_map: dict[int, dict] = {}
    for site_id, site_name, role_code, role_name in rows:
        if site_id not in sites_map:
            sites_map[site_id] = {
                "site_id": site_id,
                "site_name": site_name,
                "roles": [],
            }
        sites_map[site_id]["roles"].append({"code": role_code, "name": role_name})

    return list(sites_map.values())


@trace_service(description_zh="检查用户门店角色", description_en="Check user site role")
async def check_user_has_site_role(user_id: int, site_id: int) -> bool:
    """
    检查用户在指定 site_id 下是否有任何角色绑定。

    参数:
        user_id: 用户 ID
        site_id: 门店 ID

    返回:
        True 表示有角色绑定，False 表示无
    """
    conn = get_connection()
    try:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT 1
                  FROM auth.user_site_roles
                 WHERE user_id = %s AND site_id = %s
                   AND is_removed = false
                 LIMIT 1
                """,
                (user_id, site_id),
            )
            return cur.fetchone() is not None
    finally:
        conn.close()