处理后台跳转错误

backend/utils/auth.py

@@ -2,13 +2,14 @@
 认证装饰器与 session 校验
 """
 from functools import wraps
+
 from flask import request, redirect, url_for, session, jsonify
 
 from utils.db import get_db
 
 
 def is_session_user_valid():
-    """校验 session 中的 user_id 是否在数据库中仍存在；不存在则清除 session 并返回 False"""
+    """校验 session 中的 user_id 是否仍存在；不存在则清空 session。"""
     uid = session.get('user_id')
     if not uid:
         return False
@@ -28,28 +29,41 @@ def is_session_user_valid():
 
 
 def get_current_admin_role():
-    """获取当前登录用户的管理角色：super_admin / admin / None（非管理员）"""
+    """返回当前登录用户的管理角色：super_admin / admin / None。"""
+    uid = session.get('user_id')
+    if not uid:
+        return None, None
     try:
         conn = get_db()
         with conn.cursor() as cur:
             cur.execute(
                 "SELECT id, username, is_admin, role, created_by_id FROM users WHERE id = %s",
-                (session['user_id'],)
+                (uid,)
             )
             row = cur.fetchone()
         conn.close()
         if not row or not row.get('is_admin'):
-            return None, None
-        return row.get('role') or ('super_admin' if row.get('created_by_id') is None else 'admin'), row
+            return None, row
+        role = row.get('role') or ('super_admin' if row.get('created_by_id') is None else 'admin')
+        return role, row
     except Exception:
         return None, None
 
 
+def is_current_user_admin():
+    role, _ = get_current_admin_role()
+    return role in ('super_admin', 'admin')
+
+
+def _is_ajax_request():
+    return request.headers.get('X-Requested-With') == 'XMLHttpRequest'
+
+
 def login_required(f):
     @wraps(f)
     def decorated(*args, **kwargs):
         if not session.get('user_id') or not is_session_user_valid():
-            if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+            if _is_ajax_request():
                 return jsonify({'success': False, 'error': '未登录'}), 401
             return redirect(url_for('auth.login'))
         return f(*args, **kwargs)
@@ -59,23 +73,20 @@ def login_required(f):
 def admin_required(f):
     @wraps(f)
     def decorated(*args, **kwargs):
-        if not session.get('user_id'):
-            if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+        if not session.get('user_id') or not is_session_user_valid():
+            if _is_ajax_request():
                 return jsonify({'success': False, 'error': '未登录'}), 401
             return redirect(url_for('auth.login'))
+
         try:
-            conn = get_db()
-            with conn.cursor() as cur:
-                cur.execute("SELECT is_admin, role FROM users WHERE id = %s", (session['user_id'],))
-                row = cur.fetchone()
-            conn.close()
-            if not row or not row.get('is_admin'):
-                if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
+            role, _ = get_current_admin_role()
+            if role not in ('super_admin', 'admin'):
+                if _is_ajax_request():
                     return jsonify({'success': False, 'error': '需要管理员权限'}), 403
-                return redirect(url_for('main.admin_page'))
-        except Exception as e:
-            if request.headers.get('X-Requested-With') == 'XMLHttpRequest':
-                return jsonify({'success': False, 'error': str(e)}), 500
-            return redirect(url_for('main.admin_page'))
+                return redirect(url_for('auth.login'))
+        except Exception as exc:
+            if _is_ajax_request():
+                return jsonify({'success': False, 'error': str(exc)}), 500
+            return redirect(url_for('auth.login'))
         return f(*args, **kwargs)
     return decorated