feat: generate permission sql for stocktake modules and implement single-device login restriction

inventory-backend/app/extensions.py

@@ -3,6 +3,7 @@ from flask_migrate import Migrate
 from flask_cors import CORS
 from flask_jwt_extended import JWTManager  # 确保引入了 JWTManager
 from datetime import datetime, timezone, timedelta
+import redis
 
 # 1. 创建扩展实例（此时未绑定具体的 App）
 db = SQLAlchemy()
@@ -10,6 +11,9 @@ migrate = Migrate()
 cors = CORS()
 jwt = JWTManager()  # 必须实例化
 
+# Redis 客户端 (单设备登录互踢用)
+redis_client = None
+
 
 def beijing_time():
     """获取北京时间 (UTC+8)"""
@@ -21,6 +25,8 @@ def init_extensions(app):
     """
     统一初始化所有 Flask 扩展
     """
+    global redis_client
+
     # 初始化数据库
     db.init_app(app)
 
@@ -31,4 +37,13 @@ def init_extensions(app):
     cors.init_app(app, resources={r"/api/*": {"origins": "*"}})
 
     # 初始化 JWT (这一步至关重要，缺少它会导致 500 错误)
-    jwt.init_app(app)
+    jwt.init_app(app)
+
+    # 初始化 Redis (单设备登录互踢)
+    redis_url = app.config.get('REDIS_URL', 'redis://localhost:6379/0')
+    try:
+        redis_client = redis.from_url(redis_url, decode_responses=True)
+        redis_client.ping()
+        app.logger.info("✅ Redis connected successfully")
+    except Exception as e:
+        app.logger.warning(f"⚠️ Redis connection failed: {e}, single-device login will be disabled")

inventory-backend/app/services/auth_service.py

@@ -1,10 +1,55 @@
 # app/services/auth_service.py
 from app.models.system import SysUser, SysRolePermission # <== 引入 SysRolePermission
-from app.extensions import db
+from app.extensions import db, redis_client
 from sqlalchemy import func
-from flask_jwt_extended import create_access_token, create_refresh_token
+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))
+
+        # 转换为秒数
+        expires_seconds = int(expires_delta.total_seconds())
+
+        # 存储到 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:
     # 硬编码的超级管理员凭证
@@ -80,6 +125,11 @@ class AuthService:
             }
         )
 
+        # 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,

inventory-backend/app/utils/decorators.py

@@ -6,6 +6,59 @@ import logging
 import json
 
 
+def _verify_token_in_redis():
+    """
+    验证当前 Token 是否与 Redis 中存储的 Token 一致（单设备登录互踢）
+    """
+    from app.extensions import redis_client
+    from flask import current_app
+
+    if redis_client is None:
+        # Redis 不可用，跳过验证
+        return True
+
+    try:
+        # 获取请求中的 Token
+        auth_header = request.headers.get('Authorization', '')
+        if not auth_header.startswith('Bearer '):
+            return True
+
+        request_token = auth_header[7:]  # 去掉 'Bearer ' 前缀
+
+        # 获取当前用户 ID
+        claims = get_jwt()
+        user_id = claims.get('sub')
+        if user_id is None:
+            return True
+
+        # 从 Redis 获取存储的 Token
+        stored_token = redis_client.get(f"user_token_{user_id}")
+
+        # 如果 Redis 中没有存储的 Token（可能是旧登录或 Redis 重启），允许通过
+        if stored_token is None:
+            return True
+
+        # 比较 Token 是否一致
+        if request_token != stored_token:
+            current_app.logger.warning(f"Token mismatch for user {user_id}: request token != stored token")
+            return False
+
+        return True
+    except Exception as e:
+        current_app.logger.error(f"Redis token verification error: {e}")
+        # 出错时默认放行，避免影响正常业务
+        return True
+
+
+def _raise_token_mismatch_error():
+    """抛出 Token 不一致的错误（用于单设备登录互踢）"""
+    return jsonify({
+        'msg': '您的账号已在其他设备登录，请重新登录',
+        'code': 401,
+        'reason': 'token_mismatch'
+    }), 401
+
+
 def role_required(*roles):
     """
     自定义装饰器：检查用户角色
@@ -44,6 +97,11 @@ def login_required(fn):
         except Exception as e:
             logging.warning(f"JWT verification failed: {e}")
             return jsonify(msg='登录已过期，请重新登录'), 401
+
+        # 单设备登录互踢检查
+        if not _verify_token_in_redis():
+            return _raise_token_mismatch_error()
+
         return fn(*args, **kwargs)
     return decorator
 
@@ -63,6 +121,10 @@ def permission_required(permission_code):
                 logging.warning(f"JWT verification failed: {e}")
                 return jsonify(msg='登录已过期，请重新登录'), 401
 
+            # 单设备登录互踢检查
+            if not _verify_token_in_redis():
+                return _raise_token_mismatch_error()
+
             claims = get_jwt()
             user_role = claims.get('role')
             # 超级管理员放行 (忽略大小写)

inventory-backend/config.py

@@ -43,4 +43,9 @@ class Config:
     # 上传文件存储路径
     UPLOAD_FOLDER = os.path.join(BASE_DIR, 'uploads')
     # 限制最大上传 16MB
-    MAX_CONTENT_LENGTH = 16 * 1024 * 1024
+    MAX_CONTENT_LENGTH = 16 * 1024 * 1024
+
+    # =========================================================
+    # 5. Redis 配置 (用于单设备登录互踢)
+    # =========================================================
+    REDIS_URL = os.getenv('REDIS_URL', 'redis://localhost:6379/0')

inventory-backend/requirements.txt

@@ -6,6 +6,8 @@ marshmallow-sqlalchemy==1.0.0
 psycopg2-binary==2.9.9
 python-dotenv==1.0.0
 flask-cors==4.0.0
+# Redis for token management (single device login)
+redis==5.0.1
 # 图片处理核心库
 Pillow>=10.0.0
 # [旧] 条形码生成库 (建议保留，防止旧代码报错)

inventory-web/src/utils/request.ts

@@ -162,6 +162,15 @@ service.interceptors.response.use(
         // 核心：401 错误处理 + 无感刷新
         // ============================================================
         if (status === 401) {
+            // 0. 检查是否是互踢情况（账号在其他设备登录）
+            if (data?.reason === 'token_mismatch') {
+                message = '您的账号已在其他设备登录，请重新登录'
+                ElMessage.warning(message)
+                localStorage.clear()
+                window.location.href = '/login'
+                return Promise.reject(error)
+            }
+
             // 1. 如果是登录接口的 401，不执行刷新
             if (isLoginEndpoint) {
                 message = data?.msg || '用户名或密码错误'