打包项目

2026-03-20 18:13:47 +08:00
parent 6791869e0c
commit 7efb86ed3a
59 changed files with 1872 additions and 4187 deletions
--- a/source_code/blueprints/auth.py
+++ b/source_code/blueprints/auth.py
@@ -0,0 +1,107 @@
+"""
+认证蓝图：登录、登出、登录状态校验
+"""
+from flask import Blueprint, request, redirect, url_for, session, jsonify
+from werkzeug.security import check_password_hash
+
+from app_common import (
+    get_db,
+    _render_html,
+    _is_session_user_valid,
+    login_required,
+    BASE_DIR,
+)
+from tool.devices import DeviceIDGenerator
+
+auth_bp = Blueprint('auth', __name__)
+
+
+@auth_bp.route('/login', methods=['GET', 'POST'])
+def login():
+    if session.get('user_id') and _is_session_user_valid():
+        return redirect(url_for('main.home'))
+    if request.method == 'POST':
+        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 request.is_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()
+            if row and check_password_hash(row['password_hash'], password):
+                current_machine = DeviceIDGenerator().get_device_id()
+                stored_machine = (row.get('machine') or '').strip()
+                if not stored_machine:
+                    with conn.cursor() as cur:
+                        cur.execute("UPDATE users SET machine = %s WHERE id = %s", (current_machine, row['id']))
+                    conn.commit()
+                    conn.close()
+                    session.permanent = True
+                    session['user_id'] = row['id']
+                    session['username'] = username
+                    if request.is_json:
+                        return jsonify({'success': True, 'redirect': url_for('main.home')})
+                    return redirect(url_for('main.home'))
+                print("验证设备",stored_machine)
+                print("当前设备",current_machine)
+                if stored_machine != current_machine and row.get("is_admin") != 1:
+                    conn.close()
+                    err_msg = '当前设备与首次登录设备不一致，请在原设备上登录'
+                    if request.is_json:
+                        return jsonify({'success': False, 'error': err_msg})
+                    return _render_html('login.html', error=err_msg)
+                conn.close()
+                session.permanent = True
+                session['user_id'] = row['id']
+                session['username'] = username
+                if request.is_json:
+                    return jsonify({'success': True, 'redirect': url_for('main.home')})
+                return redirect(url_for('main.home'))
+            conn.close()
+        except Exception as e:
+            if request.is_json:
+                return jsonify({'success': False, 'error': str(e)})
+            return _render_html('login.html', error='登录失败，请稍后重试')
+        if request.is_json:
+            return jsonify({'success': False, 'error': '用户名或密码错误'})
+        return _render_html('login.html', error='用户名或密码错误')
+    return _render_html('login.html')
+
+
+@auth_bp.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})
+        stored_machine = (row.get('machine') or '').strip()
+        if stored_machine:
+            current_machine = DeviceIDGenerator().get_device_id()
+            if stored_machine != current_machine and row.get("is_admin") != 1:
+                session.clear()
+                return jsonify({'logged_in': False, 'error': '当前设备与首次登录设备不一致'})
+    except Exception:
+        return jsonify({'logged_in': False})
+    return jsonify({'logged_in': True, 'redirect': url_for('main.home')})
+
+
+@auth_bp.route('/logout')
+def logout():
+    session.clear()
+    return redirect(url_for('auth.login'))