2dbeec7d6d
new file: app/blueprints/__pycache__/__init__.cpython-311.pyc new file: app/blueprints/__pycache__/__init__.cpython-39.pyc new file: app/blueprints/__pycache__/admin.cpython-311.pyc new file: app/blueprints/__pycache__/admin.cpython-39.pyc new file: app/blueprints/__pycache__/auth.cpython-311.pyc new file: app/blueprints/__pycache__/auth.cpython-39.pyc new file: app/blueprints/__pycache__/brand.cpython-311.pyc new file: app/blueprints/__pycache__/brand.cpython-39.pyc new file: app/blueprints/__pycache__/image.cpython-311.pyc new file: app/blueprints/__pycache__/image.cpython-39.pyc new file: app/blueprints/__pycache__/main.cpython-311.pyc new file: app/blueprints/__pycache__/main.cpython-39.pyc new file: app/blueprints/admin.py new file: app/blueprints/auth.py new file: app/blueprints/brand.py new file: app/blueprints/image.py new file: app/blueprints/main.py new file: app/brand_spider/__pycache__/main.cpython-39.pyc new file: app/brand_spider/__pycache__/web_dec.cpython-39.pyc new file: app/brand_spider/main.py new file: app/brand_spider/web_dec.py new file: app/static/bg.jpg new file: "app/static/\345\223\201\347\211\214\346\226\207\346\241\243\346\240\274\345\274\217_\346\250\241\346\235\277.xlsx" new file: "app/static/\346\250\241\346\235\2772-\344\273\245\346\226\207\344\273\266\345\244\271\346\226\271\345\274\217\344\270\212\344\274\240.zip" new file: app/tool/.device_id new file: app/tool/__pycache__/devices.cpython-311.pyc new file: app/tool/__pycache__/devices.cpython-39.pyc new file: app/tool/devices.py
108 lines
4.6 KiB
Python
108 lines
4.6 KiB
Python
"""
|
|
认证蓝图：登录、登出、登录状态校验
|
|
"""
|
|
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'))
|