KCGL/inventory-backend/app/services/auth_service.py

# app/services/auth_service.py
from app.models.system import SysUser, SysRolePermission # <== 引入 SysRolePermission
from app.extensions import db, redis_client, revoke_all_tokens_for_user
from sqlalchemy import func
from flask_jwt_extended import create_access_token, create_refresh_token, get_jwt_identity
from flask import current_app
from app.utils.constants import UserRole
from datetime import timedelta
import json


def _store_token_to_redis(user_id, token, token_type='access'):
    """
    将 Token 存储到 Redis，用于单设备登录互踢
    键名: user_token_{user_id}
    值: token
    过期时间: 与 JWT 过期时间一致
    """
    if redis_client is None:
        current_app.logger.warning("Redis not available, skipping token storage")
        return False

    try:
        # 获取 JWT 过期时间配置
        if token_type == 'access':
            expires_delta = current_app.config.get('JWT_ACCESS_TOKEN_EXPIRES', timedelta(hours=2))
        else:
            expires_delta = current_app.config.get('JWT_REFRESH_TOKEN_EXPIRES', timedelta(days=7))

        # 转换为秒数（兼容 timedelta 和 int 类型）
        if isinstance(expires_delta, timedelta):
            expires_seconds = int(expires_delta.total_seconds())
        else:
            expires_seconds = int(expires_delta)  # 已经是秒数（整数）

        # 存储到 Redis
        key = f"user_token_{user_id}"
        redis_client.setex(key, expires_seconds, token)
        return True
    except Exception as e:
        current_app.logger.error(f"Failed to store token to Redis: {e}")
        return False


def _get_token_from_redis(user_id):
    """从 Redis 获取当前有效的 Token"""
    if redis_client is None:
        return None
    try:
        key = f"user_token_{user_id}"
        return redis_client.get(key)
    except Exception as e:
        current_app.logger.error(f"Failed to get token from Redis: {e}")
        return None


class AuthService:
    # 硬编码的超级管理员凭证
    SUPER_ADMIN_USER = "IRIS"
    SUPER_ADMIN_PASS = "123321"

    @staticmethod
    def login(data):
        # 用户登录时输入的只是账号ID (例如: zhangsan)
        login_input = data.get('username', '').strip()
        password = data.get('password')

        user_role = None
        user_id = None
        user_info = {}

        # 1. 优先检查硬编码的超级管理员 (IRIS)
        if login_input == AuthService.SUPER_ADMIN_USER:
            if password == AuthService.SUPER_ADMIN_PASS:
                user_role = UserRole.SUPER_ADMIN
                user_id = 0
                user_info = {
                    'username': '超级管理员(IRIS)',
                    'account_id': 'IRIS',
                    'role': user_role,
                    'department': 'System',
                    'status': 'active'
                }
            else:
                raise ValueError("密码错误")

        # 2. 检查数据库用户
        # 数据库存的是 "张三/zhangsan"
        # 登录匹配逻辑: 查找以 "/login_input" 结尾的记录
        else:
            # 使用 like 进行后缀匹配: '%/zhangsan'
            user = SysUser.query.filter(SysUser.username.like(f"%/{login_input}")).first()

            if not user:
                raise ValueError("用户不存在")

            if not user.check_password(password):
                raise ValueError("密码错误")

            if user.status != 'active':
                raise ValueError("账号已被禁用，请联系管理员")

            user_role = user.role.upper() if user.role else None
            user_id = user.id
            user_info = user.to_dict()
            user_info['role'] = user_role
            # 获取用户所属公司（存于 department 字段）
            user_company = user.department or ''

        # 3. 生成 Token
        # Token 中 identity 存数据库ID，claims 存登录账号ID
        account_id = user_info.get('account_id', login_input)

        # Access Token: 2小时有效期
        access_token = create_access_token(
            identity=user_id,
            additional_claims={
                'role': user_role,
                'username': account_id,  # 存纯账号ID
                'display_name': user_info.get('username'),  # 存显示名
                'company_name': user_company  # 存所属公司
            }
        )

        # Refresh Token: 7天有效期
        refresh_token = create_refresh_token(
            identity=user_id,
            additional_claims={
                'role': user_role,
                'username': account_id,
                'display_name': user_info.get('display_name', account_id),
                'company_name': user_company
            }
        )

        # 4. 存储 Token 到 Redis (用于单设备登录互踢)
        # 使用 str(user_id) 作为 Redis key 的一部分，因为 user_id=0 是超级管理员
        _store_token_to_redis(str(user_id), access_token, 'access')
        _store_token_to_redis(str(user_id), refresh_token, 'refresh')

        return {
            'access_token': access_token,
            'refresh_token': refresh_token,
            'user': user_info
        }

    @staticmethod
    def refresh_access_token(refresh_token_str):
        """
        使用 refresh_token 换发新的 access_token
        """
        from flask_jwt_extended import decode_token
        from flask import current_app

        try:
            # 解码 refresh_token（不验证过期，仅获取 claims）
            decoded = decode_token(refresh_token_str)
            user_id = decoded.get('sub')
            role = decoded.get('role')
            username = decoded.get('username')

            if not user_id:
                raise ValueError("无效的 refresh_token")

            # 重新查询数据库获取用户的 display_name，避免刷新后丢失
            from app.models.system import SysUser
            user = SysUser.query.get(user_id)
            if user:
                user_info = user.to_dict()
                display_name = user_info.get('display_name', username)
            else:
                display_name = username

            # 生成新的 access_token
            new_access_token = create_access_token(
                identity=user_id,
                additional_claims={
                    'role': role,
                    'username': username,
                    'display_name': display_name
                }
            )

            return {
                'access_token': new_access_token
            }

        except Exception as e:
            raise ValueError(f"Token 刷新失败: {str(e)}")

    @staticmethod
    def create_user(data, operator_role):
        """
        创建新用户
        data 包含: cn_name(张三), username(zhangsan), ...
        """
        # 标准化操作者角色为全大写
        operator_role_upper = operator_role.upper() if operator_role else None
        if operator_role_upper not in [UserRole.SUPER_ADMIN, UserRole.SUPERVISOR]:
            raise Exception("权限不足：只有超级管理员或主管可以创建新用户")

        cn_name = data.get('cn_name')
        pinyin_base = data.get('username')  # 前端传来的基础拼音，如 zhangsan

        if not cn_name or not pinyin_base:
            raise Exception("姓名和账号不能为空")

        # 后端兜底正则校验：允许中英数，禁止纯数字，无特殊字符
        import re
        name_pattern = re.compile(r'^(?!\d+$)[a-zA-Z0-9\u4e00-\u9fa5]+$')

        if not name_pattern.match(cn_name):
            raise Exception("姓名格式错误：仅支持中英文和数字，不能为纯数字，且不支持特殊字符")

        if not name_pattern.match(pinyin_base):
            raise Exception("账号格式错误：仅支持中英文和数字，不能为纯数字，且不支持特殊字符")

        role_raw = data.get('role')
        role = role_raw.upper() if role_raw else None

        # 验证角色合法性
        valid_roles = [
            v for k, v in UserRole.__dict__.items()
            if not k.startswith('__') and isinstance(v, str)
        ]

        if role not in valid_roles:
            raise Exception(f"角色无效")

        if operator_role_upper == UserRole.SUPERVISOR and role == UserRole.SUPER_ADMIN:
            raise Exception("权限不足：主管无法创建超级管理员")

        email = data.get('email', '') or None  # 空字符串转 None，避免 unique 冲突
        if email and SysUser.query.filter_by(email=email).first():
            raise Exception("邮箱已被使用")

        # === 核心逻辑: 自动处理账号重复 (zhangsan -> zhangsan1 -> zhangsan2) ===
        final_account_id = pinyin_base
        counter = 1  # 如果重复，从1开始累加

        while True:
            # 检查数据库是否存在以 "/final_account_id" 结尾的记录
            existing = SysUser.query.filter(
                (SysUser.username.like(f"%/{final_account_id}")) |
                (SysUser.username == final_account_id)
            ).first()

            if not existing:
                break  # 找到了可用的ID，跳出循环

            # 如果存在，使用 base + counter
            final_account_id = f"{pinyin_base}{counter}"
            counter += 1

        # 拼接最终存储格式: 张三/zhangsan1
        full_username_storage = f"{cn_name}/{final_account_id}"

        new_user = SysUser(
            username=full_username_storage,  # 存组合串
            email=email,
            department=data.get('department', ''),
            role=role,
            status='active'
        )
        new_user.set_password(data.get('password'))

        db.session.add(new_user)
        db.session.commit()

        # 返回时，最好把生成的ID告诉前端
        return new_user.to_dict()

    @staticmethod
    def batch_create_users(data_list, operator_role):
        """
        批量创建新用户。复用 create_user 的核心防重逻辑。
        """
        results = []
        for data in data_list:
            try:
                # 复用单条创建逻辑，它自带张三/zhangsan1的防重机制
                new_user_dict = AuthService.create_user(data, operator_role)
                results.append({
                    "cn_name": data.get('cn_name'),
                    "account_id": new_user_dict.get('account_id'),
                    "status": "success"
                })
            except Exception as e:
                results.append({
                    "cn_name": data.get('cn_name'),
                    "error": str(e),
                    "status": "fail"
                })
        return results

    @staticmethod
    def update_user(user_id, data, operator_role):
        """
        更新用户信息
        注意: 这里暂时不允许修改用户名/账号，因为涉及 split 逻辑较复杂，且通常账号不开通后不改
        """
        # 标准化操作者角色为全大写
        operator_role_upper = operator_role.upper() if operator_role else None
        if operator_role_upper not in [UserRole.SUPER_ADMIN, UserRole.SUPERVISOR]:
            raise Exception("权限不足")

        user = SysUser.query.get(user_id)
        if not user:
            raise Exception("用户不存在")

        # 1. 更新基本信息
        if 'role' in data:
            valid_roles = [
                v for k, v in UserRole.__dict__.items()
                if not k.startswith('__') and isinstance(v, str)
            ]
            new_role_raw = data['role']
            new_role = new_role_raw.upper() if new_role_raw else None
            if new_role not in valid_roles:
                raise Exception(f"角色无效")
            if operator_role_upper == UserRole.SUPERVISOR and new_role == UserRole.SUPER_ADMIN:
                raise Exception("权限不足")
            user.role = new_role

        if 'department' in data:
            user.department = data['department']

        if 'email' in data:
            email = data['email']
            if email and email != user.email:
                existing = SysUser.query.filter_by(email=email).first()
                if existing:
                    raise Exception("该邮箱已被其他用户使用")
            user.email = email

        if 'status' in data:
            new_status = data['status']
            # ★ 幽灵令牌漏洞修复：用户被禁用时，立即吊销其所有 Token
            if new_status != 'active' and user.status == 'active':
                revoke_all_tokens_for_user(user_id)
            user.status = new_status

        new_password = data.get('password')
        if new_password and str(new_password).strip():
            if len(new_password) < 6:
                raise Exception("密码长度至少6位")
            user.set_password(new_password)

        db.session.commit()
        return user.to_dict()

    @staticmethod
    def get_all_users():
        """获取所有系统用户"""
        users = SysUser.query.order_by(SysUser.id.desc()).all()
        return [user.to_dict() for user in users]

    @staticmethod
    def delete_user(user_id, operator_role):
        """删除用户：删除前自动吊销该用户所有 JWT Token"""
        # 标准化操作者角色为全大写
        operator_role_upper = operator_role.upper() if operator_role else None
        if operator_role_upper != UserRole.SUPER_ADMIN:
            raise Exception("权限不足：只有超级管理员可以删除用户")

        user = SysUser.query.get(user_id)
        if not user:
            raise Exception("用户不存在")

        # 提前获取用户名用于审计日志
        username = user.username

        # ★ 幽灵令牌漏洞修复：删除用户前，先将 user_id 加入 JWT 黑名单
        # 效果：该用户持有的所有 Token 瞬间失效，无论是否已过期
        revoke_all_tokens_for_user(user_id)

        # 清除 Redis 中的单设备登录 Token（防止残留）
        if redis_client is not None:
            try:
                redis_client.delete(f"user_token_{user_id}")
            except Exception as e:
                current_app.logger.warning(f"Failed to delete user token from Redis: {e}")

        db.session.delete(user)
        db.session.commit()
        return username

    @staticmethod
    def get_user_permissions(role_code):
        """
        获取指定角色的所有权限代码列表
        返回格式: {
            'menus': ['inbound_buy', 'system_user'],
            'elements': ['inbound_buy:unit_price', ...]
        }
        """
        # 防御性编程：role_code 为空时直接返回空权限，避免后续 SQL 崩溃
        if not role_code:
            return {'menus': [], 'elements': []}

        # 超级管理员返回所有权限（通配符）
        from app.utils.constants import UserRole
        if role_code.upper() == UserRole.SUPER_ADMIN:
            # 返回通配符，表示拥有所有菜单和元素权限
            return {
                'menus': ['*'],
                'elements': ['*']
            }

        # 1. 查菜单权限
        # 使用 func.upper() 处理数据库字段的大小写
        menu_perms = SysRolePermission.query.filter(
            func.upper(SysRolePermission.role_code) == role_code.upper(),
            SysRolePermission.type == 'menu'
        ).all()
        menu_codes = [p.target_code for p in menu_perms]

        # 2. 查元素(列)权限
        # 注意：这里我们只返回用户拥有的。前端逻辑是："如果列配置了Key且用户没这个Key，则隐藏"
        element_perms = SysRolePermission.query.filter(
            func.upper(SysRolePermission.role_code) == role_code.upper(),
            SysRolePermission.type == 'element'
        ).all()
        element_codes = [p.target_code for p in element_perms]

        # 调试日志：输出查询结果便于排查字段权限问题
        from flask import current_app
        current_app.logger.info(
            f"[权限查询] role={role_code}, 查询到菜单权限={menu_codes}, 元素权限={element_codes}"
        )

        return {
            'menus': menu_codes,
            'elements': element_codes
        }