ljwx-health/ljwx-bigscreen/app/services/admin_helper.py

"""
Admin Helper Module
提供管理员用户判断和查询功能

基于 sys_user 表的字段：
- admin_level: 管理员级别（0=普通员工, 1=部门/租户管理员, 3=超级管理员）
- customer_id: 租户ID
- org_id: 所属组织/部门ID
- level: 通过 sys_org_units 表关联获取组织层级
"""

import logging
from typing import List, Optional, Dict, Any

logger = logging.getLogger(__name__)


class AdminHelper:
    """管理员辅助工具类"""

    # 管理员级别常量
    SUPER_ADMIN_LEVEL = 3  # 超级管理员
    TENANT_ADMIN_LEVEL = 1  # 租户/部门管理员
    NORMAL_USER_LEVEL = 0  # 普通用户

    def __init__(self):
        """初始化"""
        pass

    def is_admin_user(self, user_id: int) -> bool:
        """
        判断用户是否为管理员（任何级别）

        Args:
            user_id: 用户ID

        Returns:
            bool: True表示是管理员，False表示不是
        """
        try:
            from app.models import db, UserInfo

            user = db.session.query(UserInfo).filter(
                UserInfo.id == user_id,
                UserInfo.is_deleted == False
            ).first()

            if user and hasattr(user, 'admin_level'):
                is_admin = user.admin_level > self.NORMAL_USER_LEVEL
                logger.debug(f"User {user_id} admin_level={user.admin_level}, is_admin={is_admin}")
                return is_admin

            logger.debug(f"User {user_id} not found or has no admin_level field")
            return False

        except Exception as e:
            logger.error(f"Failed to check if user {user_id} is admin: {e}")
            return False

    def is_super_admin(self, user_id: int) -> bool:
        """
        判断用户是否为超级管理员

        Args:
            user_id: 用户ID

        Returns:
            bool: True表示是超级管理员
        """
        try:
            from app.models import db, UserInfo

            user = db.session.query(UserInfo).filter(
                UserInfo.id == user_id,
                UserInfo.is_deleted == False
            ).first()

            if user and hasattr(user, 'admin_level'):
                return user.admin_level == self.SUPER_ADMIN_LEVEL

            return False

        except Exception as e:
            logger.error(f"Failed to check if user {user_id} is super admin: {e}")
            return False

    def is_tenant_admin(self, user_id: int, customer_id: int) -> bool:
        """
        判断用户是否为指定租户的管理员

        Args:
            user_id: 用户ID
            customer_id: 租户ID

        Returns:
            bool: True表示是该租户的管理员
        """
        try:
            from app.models import db, UserInfo

            user = db.session.query(UserInfo).filter(
                UserInfo.id == user_id,
                UserInfo.is_deleted == False
            ).first()

            if not user or not hasattr(user, 'admin_level'):
                return False

            # 超级管理员有所有租户的管理权限
            if user.admin_level == self.SUPER_ADMIN_LEVEL:
                return True

            # 租户管理员需要匹配 customer_id
            if user.admin_level >= self.TENANT_ADMIN_LEVEL:
                return user.customer_id == customer_id

            return False

        except Exception as e:
            logger.error(f"Failed to check if user {user_id} is admin of tenant {customer_id}: {e}")
            return False

    def is_org_admin(self, user_id: int, org_id: int, include_parent: bool = True) -> bool:
        """
        判断用户是否为指定组织/部门的管理员

        Args:
            user_id: 用户ID
            org_id: 组织/部门ID
            include_parent: 是否包含上级组织的管理员（默认True）

        Returns:
            bool: True表示是该组织的管理员
        """
        try:
            from app.models import db, UserInfo, OrgInfo

            user = db.session.query(UserInfo).filter(
                UserInfo.id == user_id,
                UserInfo.is_deleted == False
            ).first()

            if not user or not hasattr(user, 'admin_level'):
                return False

            # 超级管理员有所有组织的管理权限
            if user.admin_level == self.SUPER_ADMIN_LEVEL:
                return True

            # 非管理员直接返回False
            if user.admin_level < self.TENANT_ADMIN_LEVEL:
                return False

            # 管理员用户，检查组织权限
            user_org_id = user.org_id

            # 直接匹配
            if user_org_id == org_id:
                return True

            # 如果需要检查上级组织权限
            if include_parent:
                # 获取目标组织信息
                target_org = db.session.query(OrgInfo).filter(
                    OrgInfo.id == org_id,
                    OrgInfo.is_deleted == False
                ).first()

                # 获取用户组织信息
                user_org = db.session.query(OrgInfo).filter(
                    OrgInfo.id == user_org_id,
                    OrgInfo.is_deleted == False
                ).first()

                if target_org and user_org:
                    # 如果用户组织层级更高（数字更小），且是目标组织的祖先
                    # 通过检查 target_org 的 ancestors 字段或递归查询 parent_id
                    if user_org.level < target_org.level:
                        # 检查用户组织是否是目标组织的祖先
                        current_org = target_org
                        while current_org.parent_id and current_org.parent_id != 0:
                            if current_org.parent_id == user_org_id:
                                return True
                            current_org = db.session.query(OrgInfo).filter(
                                OrgInfo.id == current_org.parent_id,
                                OrgInfo.is_deleted == False
                            ).first()
                            if not current_org:
                                break

            return False

        except Exception as e:
            logger.error(f"Failed to check if user {user_id} is admin of org {org_id}: {e}")
            return False

    def get_admin_user_ids(
        self,
        customer_id: Optional[int] = None,
        org_id: Optional[int] = None,
        min_admin_level: int = TENANT_ADMIN_LEVEL,
        include_super_admin: bool = True
    ) -> List[str]:
        """
        获取管理员用户的ID列表

        Args:
            customer_id: 可选的租户ID，用于过滤特定租户的管理员
            org_id: 可选的组织ID，用于过滤特定组织的管理员
            min_admin_level: 最低管理员级别（默认1）
            include_super_admin: 是否包含超级管理员（默认True）

        Returns:
            List[str]: 管理员用户ID列表（字符串格式，用于SQL拼接）
        """
        try:
            from app.models import db, UserInfo

            # 构建基础查询
            query = db.session.query(UserInfo.id).filter(
                UserInfo.is_deleted == False,
                UserInfo.status == '1'  # 只查询启用的用户
            )

            # 管理员级别过滤
            if include_super_admin:
                # 包含超级管理员和指定级别以上的管理员
                query = query.filter(UserInfo.admin_level >= min_admin_level)
            else:
                # 只包含指定级别的管理员（排除超级管理员）
                query = query.filter(
                    UserInfo.admin_level >= min_admin_level,
                    UserInfo.admin_level < self.SUPER_ADMIN_LEVEL
                )

            # 租户过滤
            if customer_id is not None and not include_super_admin:
                query = query.filter(UserInfo.customer_id == customer_id)

            # 组织过滤
            if org_id is not None:
                query = query.filter(UserInfo.org_id == org_id)

            admin_users = query.distinct().all()
            admin_ids = [str(user.id) for user in admin_users]

            logger.debug(
                f"Found {len(admin_ids)} admin users "
                f"(customerId={customer_id}, orgId={org_id}, minLevel={min_admin_level})"
            )
            return admin_ids

        except Exception as e:
            logger.error(f"Failed to get admin user IDs: {e}")
            return []

    def get_user_admin_info(self, user_id: int) -> Dict[str, Any]:
        """
        获取用户的完整管理员信息

        Args:
            user_id: 用户ID

        Returns:
            Dict: 包含用户管理员相关信息的字典
        """
        try:
            from app.models import db, UserInfo, OrgInfo

            user = db.session.query(UserInfo).filter(
                UserInfo.id == user_id,
                UserInfo.is_deleted == False
            ).first()

            if not user:
                return {
                    'user_id': user_id,
                    'is_admin': False,
                    'admin_level': 0,
                    'customer_id': None,
                    'org_id': None,
                    'org_name': None,
                    'org_level': None
                }

            # 获取组织信息
            org_info = None
            org_level = None
            if user.org_id:
                org_info = db.session.query(OrgInfo).filter(
                    OrgInfo.id == user.org_id,
                    OrgInfo.is_deleted == False
                ).first()
                if org_info:
                    org_level = org_info.level

            return {
                'user_id': user_id,
                'is_admin': user.admin_level > self.NORMAL_USER_LEVEL,
                'is_super_admin': user.admin_level == self.SUPER_ADMIN_LEVEL,
                'admin_level': user.admin_level,
                'customer_id': user.customer_id,
                'org_id': user.org_id,
                'org_name': user.org_name or (org_info.name if org_info else None),
                'org_level': org_level
            }

        except Exception as e:
            logger.error(f"Failed to get admin info for user {user_id}: {e}")
            return {
                'user_id': user_id,
                'is_admin': False,
                'admin_level': 0,
                'error': str(e)
            }


# 创建全局单例实例
admin_helper = AdminHelper()


# 便捷函数
def is_admin_user(user_id: int) -> bool:
    """判断用户是否为管理员（便捷函数）"""
    return admin_helper.is_admin_user(user_id)


def is_super_admin(user_id: int) -> bool:
    """判断用户是否为超级管理员（便捷函数）"""
    return admin_helper.is_super_admin(user_id)


def is_tenant_admin(user_id: int, customer_id: int) -> bool:
    """判断用户是否为指定租户的管理员（便捷函数）"""
    return admin_helper.is_tenant_admin(user_id, customer_id)


def is_org_admin(user_id: int, org_id: int, include_parent: bool = True) -> bool:
    """判断用户是否为指定组织的管理员（便捷函数）"""
    return admin_helper.is_org_admin(user_id, org_id, include_parent)


def get_admin_user_ids(
    customer_id: Optional[int] = None,
    org_id: Optional[int] = None,
    min_admin_level: int = 1,
    include_super_admin: bool = True
) -> List[str]:
    """获取管理员用户ID列表（便捷函数）"""
    return admin_helper.get_admin_user_ids(customer_id, org_id, min_admin_level, include_super_admin)


def get_user_admin_info(user_id: int) -> Dict[str, Any]:
    """获取用户的完整管理员信息（便捷函数）"""
    return admin_helper.get_user_admin_info(user_id)