""" 认证蓝图:登录、登出、登录状态校验 """ from flask import Blueprint, request, redirect, url_for, session, jsonify, make_response, current_app from werkzeug.security import check_password_hash from utils.db import get_db from utils.auth import login_required, is_session_user_valid from utils.render import render_html auth = Blueprint('auth', __name__, url_prefix='') @auth.route('/login', methods=['GET', 'POST']) def login(): force_relogin = request.args.get('logout') == '1' or request.args.get('switch') == '1' if request.method == 'GET' and force_relogin: session.clear() response = make_response(render_html('login.html')) response.delete_cookie(current_app.config.get('SESSION_COOKIE_NAME', 'session')) response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0' return response if request.method == 'GET' and session.get('user_id') and is_session_user_valid(): return redirect(url_for('main.admin_page')) if request.method == 'POST': session.clear() wants_json = request.is_json or request.headers.get('X-Requested-With') == 'XMLHttpRequest' data = request.get_json() if request.is_json else request.form username = (data.get('username') or '').strip() password = data.get('password') or '' if not username or not password: if wants_json: return jsonify({'success': False, 'error': '请输入用户名和密码'}) return render_html('login.html', error='请输入用户名和密码') try: conn = get_db() with conn.cursor() as cur: cur.execute( "SELECT id, password_hash, machine, is_admin FROM users WHERE username = %s", (username,) ) row = cur.fetchone() conn.close() if row and check_password_hash(row['password_hash'], password): session.permanent = True session['user_id'] = row['id'] session['username'] = username if wants_json: return jsonify({'success': True, 'redirect': url_for('main.admin_page')}) return redirect(url_for('main.admin_page')) except Exception as exc: if wants_json: return jsonify({'success': False, 'error': str(exc)}) return render_html('login.html', error='登录失败,请稍后重试') if wants_json: return jsonify({'success': False, 'error': '用户名或密码错误'}) return render_html('login.html', error='用户名或密码错误') return render_html('login.html') @auth.route('/api/auth/check') @login_required def api_auth_check(): """校验登录状态,用于页面加载时判断是否已登录。""" if not session.get('user_id'): return jsonify({'logged_in': False}) try: conn = get_db() with conn.cursor() as cur: cur.execute("SELECT machine, is_admin FROM users WHERE id = %s", (session['user_id'],)) row = cur.fetchone() conn.close() if not row: return jsonify({'logged_in': False}) except Exception: return jsonify({'logged_in': False}) return jsonify({'logged_in': True, 'redirect': url_for('main.admin_page')}) @auth.route('/logout') def logout(): session.clear() response = redirect(url_for('auth.login', logout='1')) response.delete_cookie(current_app.config.get('SESSION_COOKIE_NAME', 'session')) response.headers['Cache-Control'] = 'no-store, no-cache, must-revalidate, max-age=0' return response