From 04f77853baa5d8e1a6912395e742d61b2290ea1b Mon Sep 17 00:00:00 2001 From: bwstudio <7469504@qq.com> Date: Wed, 17 Sep 2025 22:33:49 +0800 Subject: [PATCH] 2025017a --- app.py | 537 ++--------------------------------------- car_info.db | Bin 36864 -> 36864 bytes templates/query.html | 9 +- user.py | 552 +++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 573 insertions(+), 525 deletions(-) create mode 100644 user.py diff --git a/app.py b/app.py index cec506a..98d25ba 100644 --- a/app.py +++ b/app.py @@ -1,13 +1,18 @@ -from flask import Flask, render_template, request, redirect, url_for, session +from flask import Flask, render_template, request, redirect, url_for, session, jsonify from datetime import datetime, timedelta import sqlite3 import os import random -from flask import Flask, render_template, request, redirect, url_for, session, jsonify + +# 导入用户模块 +from user import bp as user_bp, init_user_db app = Flask(__name__) app.secret_key = 'your_secret_key' # 生产环境中应该使用更安全的密钥 +# 注册用户蓝图 +app.register_blueprint(user_bp) + # 确保模板文件夹存在 templates_dir = os.path.join(os.path.dirname(__file__), 'templates') if not os.path.exists(templates_dir): @@ -15,20 +20,12 @@ if not os.path.exists(templates_dir): # 初始化数据库 def init_db(): + # 初始化用户相关的数据库 + init_user_db() + conn = sqlite3.connect('car_info.db') cursor = conn.cursor() - # 创建用户表 - cursor.execute(''' - CREATE TABLE IF NOT EXISTS users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT NOT NULL UNIQUE, - password TEXT NOT NULL, - name TEXT NOT NULL, - phone TEXT NOT NULL, - department TEXT NOT NULL, - status TEXT DEFAULT 'active' NOT NULL - ) - ''') + # 创建车辆信息表 cursor.execute(''' CREATE TABLE IF NOT EXISTS car_owners ( @@ -41,6 +38,7 @@ def init_db(): address TEXT ) ''') + # 创建查询历史表 cursor.execute(''' CREATE TABLE IF NOT EXISTS query_history ( @@ -54,30 +52,8 @@ def init_db(): ) ''') - # 创建登录信息表 - cursor.execute(''' - CREATE TABLE IF NOT EXISTS login_info ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT NOT NULL UNIQUE, - password TEXT NOT NULL, - otp_code TEXT NOT NULL, - FOREIGN KEY (username) REFERENCES users (username) - ) - ''') - # 插入测试数据 - # 插入用户和车辆数据 + # 插入一些测试车辆数据 try: - # 先检查用户表结构是否有新字段，如果没有则跳过插入，避免错误 - cursor.execute("PRAGMA table_info(users)") - columns = [column[1] for column in cursor.fetchall()] - if 'name' in columns and 'phone' in columns and 'department' in columns: - # 检查admin用户是否已存在 - cursor.execute("SELECT id FROM users WHERE username = ?", ('admin',)) - if not cursor.fetchone(): - cursor.execute("INSERT INTO users (username, password, name, phone, department) VALUES (?, ?, ?, ?, ?)", ('admin', 'admin123', '管理员', '13800138000', 'IT部')) - # 同时插入到登录信息表 - cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", ('admin', 'admin123', str(random.randint(100000, 999999)))) - # 插入一些测试车辆数据 test_cars = [ ('京A12345', '13800138001', '110101199001011234', '张三', 'zhangsan@example.com', '北京市朝阳区'), ('沪B54321', '13900139001', '310101199001012345', '李四', 'lisi@example.com', '上海市静安区'), @@ -160,63 +136,16 @@ def init_db(): conn.commit() conn.close() -# 登录路由 -@app.route('/', methods=['GET', 'POST']) -def login(): - # 检查用户是否已登录，如果已登录则自动跳转到查询页面 - if 'username' in session: - return redirect(url_for('query')) - - if request.method == 'POST': - username = request.form['username'] - password = request.form['password'] - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - # 先检查用户是否存在于users表中 - cursor.execute("SELECT * FROM users WHERE username = ?", (username,)) - user = cursor.fetchone() - - if user: - # 从login_info表中验证密码 - cursor.execute("SELECT password, otp_code FROM login_info WHERE username = ?", (username,)) - login_info = cursor.fetchone() - - # 如果login_info表中没有该用户的记录，创建一个 - if not login_info: - # 不自动生成OTP码，设置为空字符串 - otp_code = "" - # 插入到login_info表 - cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", - (username, password, otp_code)) - conn.commit() - is_valid = True - else: - # 验证密码 - is_valid = (login_info[0] == password) - otp_code = login_info[1] - - if is_valid: - # 登录成功，设置会话信息 - session['username'] = username - session['department'] = user[5] if len(user) > 5 else '未知部门' - session['name'] = user[3] if len(user) > 3 else username # 存储用户姓名，默认使用用户名 - # 存储登录时间 - session['login_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') - # 存储OTP码 - session['otp_code'] = otp_code - conn.close() - return redirect(url_for('query')) - - conn.close() - return render_template('login.html', error='用户名或密码错误') - return render_template('login.html') +# 首页重定向到登录页 +@app.route('/') +def index(): + return redirect(url_for('user.login')) # 查询路由 @app.route('/query', methods=['GET', 'POST']) def query(): if 'username' not in session: - return redirect(url_for('login')) + return redirect(url_for('user.login')) # 获取最近查询历史 formatted_recent_queries = get_recent_queries() @@ -370,436 +299,6 @@ def get_recent_queries(): }) return formatted_recent_queries -# 注册路由 -@app.route('/register', methods=['GET', 'POST']) -def register(): - if request.method == 'POST': - username = request.form.get('username', '').strip() - name = request.form.get('name', '').strip() - phone = request.form.get('phone', '').strip() - department = request.form.get('department', '').strip() - password = request.form.get('password', '') - confirm_password = request.form.get('confirm_password', '') - - # 表单验证 - if not username or not name or not phone or not department or not password: - return render_template('register.html', error='请填写所有必填字段') - - # 密码匹配检查 - if password != confirm_password: - return render_template('register.html', error='两次输入的密码不一致') - - # 检查用户名是否已存在 - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - cursor.execute("SELECT id FROM users WHERE username = ?", (username,)) - if cursor.fetchone(): - conn.close() - return render_template('register.html', error='该用户名已被注册') - - # 插入新用户 - try: - cursor.execute( - "INSERT INTO users (username, password, name, phone, department) VALUES (?, ?, ?, ?, ?)", - (username, password, name, phone, department) - ) - # 同时插入到登录信息表，不自动生成OTP码 - cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", - (username, password, "")) - conn.commit() - conn.close() - # 注册成功，跳转到登录页并显示成功消息 - return render_template('login.html', success='注册成功，请登录') - except Exception as e: - conn.close() - return render_template('register.html', error=f'注册失败，请重试。错误：{str(e)}') - - return render_template('register.html') - -# 修改密码路由 -@app.route('/change_password', methods=['POST']) -def change_password(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - old_password = request.form.get('old_password') - new_password = request.form.get('new_password') - confirm_password = request.form.get('confirm_password') - - # 验证参数 - if not old_password or not new_password or not confirm_password: - return jsonify({'success': False, 'error': '请填写所有必填字段'}) - - # 验证新密码和确认密码是否一致 - if new_password != confirm_password: - return jsonify({'success': False, 'error': '两次输入的新密码不一致'}) - - # 验证新密码复杂度 - if len(new_password) < 8 or not any(c.isalpha() for c in new_password) or not any(c.isdigit() for c in new_password): - return jsonify({'success': False, 'error': '新密码长度至少8位，且必须包含字母和数字'}) - - # 连接数据库，验证旧密码并更新新密码 - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - # 验证旧密码是否正确 - cursor.execute("SELECT password FROM users WHERE username = ?", (session['username'],)) - result = cursor.fetchone() - - if not result or result[0] != old_password: - conn.close() - return jsonify({'success': False, 'error': '旧密码不正确'}) - - # 更新密码 - cursor.execute("UPDATE users SET password = ? WHERE username = ?", (new_password, session['username'])) - # 同时更新login_info表中的密码 - cursor.execute("UPDATE login_info SET password = ? WHERE username = ?", (new_password, session['username'])) - conn.commit() - conn.close() - - return jsonify({'success': True, 'message': '修改密码成功，请重新登录'}) - -# 获取个人信息路由 -@app.route('/get_user_profile', methods=['GET']) -def get_user_profile(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - # 连接数据库，获取用户信息 - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],)) - user = cursor.fetchone() - conn.close() - - if not user: - return jsonify({'success': False, 'error': '用户不存在'}) - - # 构造返回数据 - user_data = { - 'username': user[0], - 'name': user[1], - 'phone': user[2], - 'department': user[3] - } - - return jsonify({'success': True, 'user': user_data}) - -# 更新个人信息路由 -@app.route('/update_user_profile', methods=['POST']) -def update_user_profile(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - name = request.form.get('name', '').strip() - phone = request.form.get('phone', '').strip() - department = request.form.get('department', '').strip() - - # 验证参数 - if not name or not phone or not department: - return jsonify({'success': False, 'error': '请填写所有必填字段'}) - - # 连接数据库，更新用户信息 - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - try: - cursor.execute( - "UPDATE users SET name = ?, phone = ?, department = ? WHERE username = ?", - (name, phone, department, session['username']) - ) - conn.commit() - - # 获取更新后的用户信息 - cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],)) - updated_user = cursor.fetchone() - conn.close() - - # 更新会话中的部门信息 - session['department'] = department - - # 构造返回数据 - user_data = { - 'username': updated_user[0], - 'name': updated_user[1], - 'phone': updated_user[2], - 'department': updated_user[3] - } - - return jsonify({'success': True, 'message': '个人信息更新成功', 'user': user_data}) - except Exception as e: - conn.close() - return jsonify({'success': False, 'error': f'更新失败，请重试。错误：{str(e)}'}) - -# 获取登录信息路由 -@app.route('/get_login_info', methods=['GET']) -def get_login_info(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - # 获取登录时间 - 如果会话中没有存储登录时间，可以使用当前时间 - login_time = session.get('login_time', datetime.now().strftime('%Y-%m-%d %H:%M:%S')) - - # 获取登录IP地址 - login_ip = request.remote_addr - - # 构造返回数据 - return jsonify({ - 'success': True, - 'login_time': login_time, - 'login_ip': login_ip - }) - -# 登录信息更新路由 -@app.route('/update_login_info', methods=['POST']) -def update_login_info(): - import re - if 'username' not in session: - return jsonify({'success': False, 'error': '请先登录'}), 401 - - data = request.get_json() - username = data.get('username') - password = data.get('password') - otp_code = data.get('otp_code') - - # 验证输入 - if not username or not otp_code: - return jsonify({'success': False, 'error': '用户名和OTP码不能为空'}) - - # 验证OTP码长度必须是6位 - if len(otp_code) != 6: - return jsonify({'success': False, 'error': 'OTP码长度必须是6位'}) - - # 确保用户只能更新自己的信息 - if username != session['username'] and session['username'] != 'admin': - return jsonify({'success': False, 'error': '无权更新其他用户的信息'}) - - # 连接数据库 - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - try: - # 构建更新语句 - if session['username'] == 'admin': - # admin用户可以更新所有字段 - if password: - # 只验证密码不为空，不限制格式 - cursor.execute("UPDATE login_info SET password = ?, otp_code = ? WHERE username = ?", - (password, otp_code, username)) - else: - cursor.execute("UPDATE login_info SET otp_code = ? WHERE username = ?", (otp_code, username)) - else: - # 普通用户只能更新自己的OTP码 - cursor.execute("UPDATE login_info SET otp_code = ? WHERE username = ?", (otp_code, session['username'])) - - # 检查是否有行被更新 - if cursor.rowcount == 0: - # 如果没有找到记录，则插入新记录 - if session['username'] == 'admin': - # admin用户可以为任何用户创建登录信息 - cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", - (username, password or 'default_password', otp_code)) - else: - # 普通用户只能创建自己的登录信息 - cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", - (session['username'], 'default_password', otp_code)) - - # 提交更改 - conn.commit() - - # 如果更新的是当前登录用户的信息，更新session中的OTP码 - if username == session['username']: - session['otp_code'] = otp_code - - conn.close() - return jsonify({'success': True, 'message': '登录信息更新成功'}) - except Exception as e: - conn.close() - return jsonify({'success': False, 'error': f'更新失败：{str(e)}'}) - -# 登出路由 -@app.route('/logout') -def logout(): - session.clear() - return redirect(url_for('login')) - -# 用户管理页面路由 -@app.route('/users_management') -def users_management(): - if 'username' not in session: - return redirect(url_for('login')) - - # 只有admin用户才能访问用户管理页面 - if session['username'] != 'admin': - return redirect(url_for('query')) - - # 获取所有部门列表（用于筛选和修改） - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - cursor.execute("SELECT DISTINCT department FROM users") - departments = [dept[0] for dept in cursor.fetchall()] - conn.close() - - return render_template('users_management.html', departments=departments) - -# 获取用户列表路由 -@app.route('/get_users', methods=['GET']) -def get_users(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - # 只有admin用户才能获取用户列表 - if session['username'] != 'admin': - return jsonify({'success': False, 'error': '无权限访问'}), 403 - - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - # 获取所有用户信息 - cursor.execute("SELECT id, username, name, department, status FROM users") - users = cursor.fetchall() - conn.close() - - # 格式化用户数据 - formatted_users = [] - for user in users: - formatted_users.append({ - 'id': user[0], - 'username': user[1], - 'name': user[2], - 'department': user[3], - 'status': user[4] - }) - - return jsonify({'success': True, 'users': formatted_users}) - -# 更新用户状态路由 -@app.route('/update_user_status', methods=['POST']) -def update_user_status(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - # 只有admin用户才能更新用户状态 - if session['username'] != 'admin': - return jsonify({'success': False, 'error': '无权限访问'}), 403 - - data = request.get_json() - user_id = data.get('user_id') - status = data.get('status') - - # 验证输入 - if not user_id or not status: - return jsonify({'success': False, 'error': '缺少必要参数'}) - - # 验证状态值 - if status not in ['active', 'inactive']: - return jsonify({'success': False, 'error': '状态值无效'}) - - # 不能禁用admin用户 - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - cursor.execute("SELECT username FROM users WHERE id = ?", (user_id,)) - user = cursor.fetchone() - - if user and user[0] == 'admin' and status == 'inactive': - conn.close() - return jsonify({'success': False, 'error': '不能禁用管理员用户'}) - - # 更新用户状态 - try: - cursor.execute("UPDATE users SET status = ? WHERE id = ?", (status, user_id)) - conn.commit() - conn.close() - - return jsonify({'success': True, 'message': '用户状态更新成功'}) - except Exception as e: - conn.close() - return jsonify({'success': False, 'error': f'更新失败：{str(e)}'}) - -# 更新用户部门路由 -@app.route('/update_user_department', methods=['POST']) -def update_user_department(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - # 只有admin用户才能更新用户部门 - if session['username'] != 'admin': - return jsonify({'success': False, 'error': '无权限访问'}), 403 - - data = request.get_json() - user_id = data.get('user_id') - department = data.get('department') - - # 验证输入 - if not user_id or not department: - return jsonify({'success': False, 'error': '缺少必要参数'}) - - # 更新用户部门 - try: - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - cursor.execute("UPDATE users SET department = ? WHERE id = ?", (department, user_id)) - conn.commit() - conn.close() - - return jsonify({'success': True, 'message': '部门信息更新成功'}) - except Exception as e: - conn.close() - return jsonify({'success': False, 'error': f'更新失败：{str(e)}'}) - -# 重置用户密码路由 -@app.route('/reset_user_password', methods=['POST']) -def reset_user_password(): - if 'username' not in session: - return jsonify({'success': False, 'error': '未登录'}), 401 - - # 只有admin用户才能重置用户密码 - if session['username'] != 'admin': - return jsonify({'success': False, 'error': '无权限访问'}), 403 - - data = request.get_json() - user_id = data.get('user_id') - username = data.get('username') - - # 验证输入 - if not user_id or not username: - return jsonify({'success': False, 'error': '缺少必要参数'}) - - # 重置密码为默认值 - default_password = 'default123' - - try: - conn = sqlite3.connect('car_info.db') - cursor = conn.cursor() - - # 检查用户ID和用户名是否匹配 - cursor.execute("SELECT id FROM users WHERE username = ?", (username,)) - db_user_id = cursor.fetchone() - - if not db_user_id or db_user_id[0] != user_id: - conn.close() - return jsonify({'success': False, 'error': '用户信息不匹配'}) - - # 更新users表中的密码 - cursor.execute("UPDATE users SET password = ? WHERE id = ?", (default_password, user_id)) - - # 更新login_info表中的密码 - cursor.execute("UPDATE login_info SET password = ? WHERE username = ?", (default_password, username)) - - # 如果login_info表中没有该用户的记录，则插入 - if cursor.rowcount == 0: - cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", - (username, default_password, "")) - - conn.commit() - conn.close() - - return jsonify({'success': True, 'message': '密码已初始化为：' + default_password}) - except Exception as e: - conn.close() - return jsonify({'success': False, 'error': f'密码重置失败：{str(e)}'}) - if __name__ == '__main__': init_db() app.run(debug=True) \ No newline at end of file diff --git a/car_info.db b/car_info.db index 88856efc6daf022860e5b671ad29ef2ff170030a..eeb643cc403708bcadebca1c5148088973516bf8 100644 GIT binary patch delta 335 zcmZozz|^pSX@WGP%0wAwRuu-lWX6pt3-~#h_?I&9pX6V`A;$MpV};_aDZRjgpq?mTvlFLTrx2w zH!}}R7#bN*ekiXDRQH{M|2t6KJ^qOS{E{5Z;+$Z4UIqpR7XHHw{NMPW^Izvb3{

sECxXHo-y!01FF8muOh+73bYs1 UQgag%LsJVbpy{YKbHZ%~09(I0Hvj+t diff --git a/templates/query.html b/templates/query.html index 405b229..ae9968f 100644 --- a/templates/query.html +++ b/templates/query.html @@ -695,13 +695,10 @@ .then(response => response.json()) .then(data => { if (data.success) { - // 密码修改成功 + // 密码修改成功，先执行退出功能，再跳转到登录页面 alert(data.message); - closeChangePasswordModal(); - // 跳转到登录页面 - setTimeout(() => { - window.location.href = '{{ url_for('login') }}'; - }, 1000); + // 重定向到退出路由，系统会自动完成退出并跳转到登录页面 + window.location.href = '{{ url_for('logout') }}'; } else if (data.error) { // 显示服务器返回的错误信息 showPasswordError(data.error); diff --git a/user.py b/user.py new file mode 100644 index 0000000..f78dc78 --- /dev/null +++ b/user.py @@ -0,0 +1,552 @@ +import sqlite3 +from datetime import datetime +from flask import Blueprint, render_template, request, jsonify, redirect, url_for, session + +# 创建用户相关的蓝图 +bp = Blueprint('user', __name__) + +# 初始化数据库（用户相关部分） +def init_user_db(): + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 创建用户表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT NOT NULL UNIQUE, + password TEXT NOT NULL, + name TEXT NOT NULL, + phone TEXT NOT NULL, + department TEXT NOT NULL, + status TEXT DEFAULT 'active' + ) + ''') + + # 创建登录信息表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS login_info ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT NOT NULL UNIQUE, + password TEXT NOT NULL, + otp_code TEXT + ) + ''') + + # 检查是否存在admin用户，如果不存在则创建 + cursor.execute("SELECT * FROM users WHERE username = 'admin'") + if not cursor.fetchone(): + # 创建admin用户，默认密码为'admin123' + cursor.execute( + "INSERT INTO users (username, password, name, phone, department) VALUES (?, ?, ?, ?, ?)", + ('admin', 'admin123', '管理员', '13800138000', '管理部') + ) + # 同时插入到登录信息表 + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + ('admin', 'admin123', '')) + + # 插入一些测试数据 + cursor.execute("SELECT COUNT(*) FROM users") + if cursor.fetchone()[0] <= 1: # 只有admin用户时 + # 插入几个测试用户 + test_users = [ + ('user1', 'password1', '张三', '13800138001', '技术部'), + ('user2', 'password2', '李四', '13800138002', '销售部'), + ('user3', 'password3', '王五', '13800138003', '财务部') + ] + + for user in test_users: + cursor.execute( + "INSERT INTO users (username, password, name, phone, department) VALUES (?, ?, ?, ?, ?)", + user + ) + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + (user[0], user[1], '')) + + conn.commit() + conn.close() + +# 登录路由 +@bp.route('/login', methods=['GET', 'POST']) +def login(): + if request.method == 'POST': + username = request.form.get('username') + password = request.form.get('password') + + # 连接数据库 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 验证用户名和密码 + cursor.execute("SELECT * FROM users WHERE username = ? AND password = ?", (username, password)) + user = cursor.fetchone() + + if user: + # 检查用户状态 + if user[6] != 'active': + conn.close() + return render_template('login.html', error='用户已被禁用') + + # 设置会话信息 + session['username'] = user[1] # 用户名 + session['name'] = user[3] # 姓名 + session['department'] = user[5] # 部门 + session['login_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') + + # 检查是否有login_info记录，如果没有则创建 + cursor.execute("SELECT * FROM login_info WHERE username = ?", (username,)) + login_info = cursor.fetchone() + + if not login_info: + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + (username, password, "")) + conn.commit() + # 设置session中的OTP码为空 + session['otp_code'] = "" + else: + # 设置session中的OTP码 + session['otp_code'] = login_info[3] if login_info[3] else "" + + conn.close() + + # 登录成功，跳转到查询页面 + return redirect(url_for('query')) + else: + conn.close() + return render_template('login.html', error='用户名或密码错误') + + return render_template('login.html') + +# 注册路由 +@bp.route('/register', methods=['GET', 'POST']) +def register(): + if request.method == 'POST': + username = request.form.get('username') + password = request.form.get('password') + confirm_password = request.form.get('confirm_password') + name = request.form.get('name') + phone = request.form.get('phone') + department = request.form.get('department') + + # 验证参数 + if not username or not password or not confirm_password or not name or not phone or not department: + return render_template('register.html', error='请填写所有必填字段') + + # 验证密码是否一致 + if password != confirm_password: + return render_template('register.html', error='两次输入的密码不一致') + + # 验证密码复杂度 + if len(password) < 8 or not any(c.isalpha() for c in password) or not any(c.isdigit() for c in password): + return render_template('register.html', error='密码长度至少8位，且必须包含字母和数字') + + # 连接数据库，检查用户名是否已存在 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + cursor.execute("SELECT * FROM users WHERE username = ?", (username,)) + if cursor.fetchone(): + conn.close() + return render_template('register.html', error='用户名已存在') + + try: + cursor.execute( + "INSERT INTO users (username, password, name, phone, department) VALUES (?, ?, ?, ?, ?)", + (username, password, name, phone, department) + ) + # 同时插入到登录信息表，不自动生成OTP码 + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + (username, password, "")) + conn.commit() + conn.close() + # 注册成功，跳转到登录页并显示成功消息 + return render_template('login.html', success='注册成功，请登录') + except Exception as e: + conn.close() + return render_template('register.html', error=f'注册失败，请重试。错误：{str(e)}') + + return render_template('register.html') + +# 修改密码路由 +@bp.route('/change_password', methods=['POST']) +def change_password(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + old_password = request.form.get('old_password') + new_password = request.form.get('new_password') + confirm_password = request.form.get('confirm_password') + + # 验证参数 + if not old_password or not new_password or not confirm_password: + return jsonify({'success': False, 'error': '请填写所有必填字段'}) + + # 验证新密码和确认密码是否一致 + if new_password != confirm_password: + return jsonify({'success': False, 'error': '两次输入的新密码不一致'}) + + # 验证新密码复杂度 + if len(new_password) < 8 or not any(c.isalpha() for c in new_password) or not any(c.isdigit() for c in new_password): + return jsonify({'success': False, 'error': '新密码长度至少8位，且必须包含字母和数字'}) + + # 连接数据库，验证旧密码并更新新密码 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 验证旧密码是否正确 + cursor.execute("SELECT password FROM users WHERE username = ?", (session['username'],)) + result = cursor.fetchone() + + if not result or result[0] != old_password: + conn.close() + return jsonify({'success': False, 'error': '旧密码不正确'}) + + # 更新密码 + cursor.execute("UPDATE users SET password = ? WHERE username = ?", (new_password, session['username'])) + # 同时更新login_info表中的密码 + cursor.execute("UPDATE login_info SET password = ? WHERE username = ?", (new_password, session['username'])) + conn.commit() + conn.close() + + return jsonify({'success': True, 'message': '修改密码成功，请重新登录'}) + +# 获取个人信息路由 +@bp.route('/get_user_profile', methods=['GET']) +def get_user_profile(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + # 连接数据库，获取用户信息 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],)) + user = cursor.fetchone() + conn.close() + + if not user: + return jsonify({'success': False, 'error': '用户不存在'}) + + # 构造返回数据 + user_data = { + 'username': user[0], + 'name': user[1], + 'phone': user[2], + 'department': user[3] + } + + return jsonify({'success': True, 'user': user_data}) + +# 更新个人信息路由 +@bp.route('/update_user_profile', methods=['POST']) +def update_user_profile(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + name = request.form.get('name', '').strip() + phone = request.form.get('phone', '').strip() + department = request.form.get('department', '').strip() + + # 验证参数 + if not name or not phone or not department: + return jsonify({'success': False, 'error': '请填写所有必填字段'}) + + # 连接数据库，更新用户信息 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + try: + cursor.execute( + "UPDATE users SET name = ?, phone = ?, department = ? WHERE username = ?", + (name, phone, department, session['username']) + ) + conn.commit() + + # 获取更新后的用户信息 + cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],)) + updated_user = cursor.fetchone() + conn.close() + + # 更新会话中的部门信息 + session['department'] = department + + # 构造返回数据 + user_data = { + 'username': updated_user[0], + 'name': updated_user[1], + 'phone': updated_user[2], + 'department': updated_user[3] + } + + return jsonify({'success': True, 'message': '个人信息更新成功', 'user': user_data}) + except Exception as e: + conn.close() + return jsonify({'success': False, 'error': f'更新失败，请重试。错误：{str(e)}'}) + +# 获取登录信息路由 +@bp.route('/get_login_info', methods=['GET']) +def get_login_info(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + # 获取登录时间 - 如果会话中没有存储登录时间，可以使用当前时间 + login_time = session.get('login_time', datetime.now().strftime('%Y-%m-%d %H:%M:%S')) + + # 获取登录IP地址 + login_ip = request.remote_addr + + # 构造返回数据 + return jsonify({ + 'success': True, + 'login_time': login_time, + 'login_ip': login_ip + }) + +# 登录信息更新路由 +@bp.route('/update_login_info', methods=['POST']) +def update_login_info(): + import re + if 'username' not in session: + return jsonify({'success': False, 'error': '请先登录'}), 401 + + data = request.get_json() + username = data.get('username') + password = data.get('password') + otp_code = data.get('otp_code') + + # 验证输入 + if not username or not otp_code: + return jsonify({'success': False, 'error': '用户名和OTP码不能为空'}) + + # 验证OTP码长度必须是6位 + if len(otp_code) != 6: + return jsonify({'success': False, 'error': 'OTP码长度必须是6位'}) + + # 确保用户只能更新自己的信息 + if username != session['username'] and session['username'] != 'admin': + return jsonify({'success': False, 'error': '无权更新其他用户的信息'}) + + # 连接数据库 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + try: + # 构建更新语句 + if session['username'] == 'admin': + # admin用户可以更新所有字段 + if password: + # 只验证密码不为空，不限制格式 + cursor.execute("UPDATE login_info SET password = ?, otp_code = ? WHERE username = ?", + (password, otp_code, username)) + else: + cursor.execute("UPDATE login_info SET otp_code = ? WHERE username = ?", (otp_code, username)) + else: + # 普通用户只能更新自己的OTP码 + cursor.execute("UPDATE login_info SET otp_code = ? WHERE username = ?", (otp_code, session['username'])) + + # 检查是否有行被更新 + if cursor.rowcount == 0: + # 如果没有找到记录，则插入新记录 + if session['username'] == 'admin': + # admin用户可以为任何用户创建登录信息 + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + (username, password or 'default_password', otp_code)) + else: + # 普通用户只能创建自己的登录信息 + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + (session['username'], 'default_password', otp_code)) + + # 提交更改 + conn.commit() + + # 如果更新的是当前登录用户的信息，更新session中的OTP码 + if username == session['username']: + session['otp_code'] = otp_code + + conn.close() + return jsonify({'success': True, 'message': '登录信息更新成功'}) + except Exception as e: + conn.close() + return jsonify({'success': False, 'error': f'更新失败：{str(e)}'}) + +# 登出路由 +@bp.route('/logout') +def logout(): + session.clear() + return redirect(url_for('user.login')) + +# 用户管理页面路由 +@bp.route('/users_management') +def users_management(): + if 'username' not in session: + return redirect(url_for('user.login')) + + # 只有admin用户才能访问用户管理页面 + if session['username'] != 'admin': + return redirect(url_for('query')) + + # 获取所有部门列表（用于筛选和修改） + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + cursor.execute("SELECT DISTINCT department FROM users") + departments = [dept[0] for dept in cursor.fetchall()] + conn.close() + + return render_template('users_management.html', departments=departments) + +# 获取用户列表路由 +@bp.route('/get_users', methods=['GET']) +def get_users(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + # 只有admin用户才能获取用户列表 + if session['username'] != 'admin': + return jsonify({'success': False, 'error': '无权限访问'}), 403 + + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 获取所有用户信息 + cursor.execute("SELECT id, username, name, department, status FROM users") + users = cursor.fetchall() + conn.close() + + # 格式化用户数据 + formatted_users = [] + for user in users: + formatted_users.append({ + 'id': user[0], + 'username': user[1], + 'name': user[2], + 'department': user[3], + 'status': user[4] + }) + + return jsonify({'success': True, 'users': formatted_users}) + +# 更新用户状态路由 +@bp.route('/update_user_status', methods=['POST']) +def update_user_status(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + # 只有admin用户才能更新用户状态 + if session['username'] != 'admin': + return jsonify({'success': False, 'error': '无权限访问'}), 403 + + data = request.get_json() + user_id = data.get('user_id') + status = data.get('status') + + # 验证输入 + if not user_id or not status: + return jsonify({'success': False, 'error': '缺少必要参数'}) + + # 验证状态值 + if status not in ['active', 'inactive']: + return jsonify({'success': False, 'error': '状态值无效'}) + + # 不能禁用admin用户 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + cursor.execute("SELECT username FROM users WHERE id = ?", (user_id,)) + user = cursor.fetchone() + + if user and user[0] == 'admin' and status == 'inactive': + conn.close() + return jsonify({'success': False, 'error': '不能禁用管理员用户'}) + + # 更新用户状态 + try: + cursor.execute("UPDATE users SET status = ? WHERE id = ?", (status, user_id)) + conn.commit() + conn.close() + + return jsonify({'success': True, 'message': '用户状态更新成功'}) + except Exception as e: + conn.close() + return jsonify({'success': False, 'error': f'更新失败：{str(e)}'}) + +# 更新用户部门路由 +@bp.route('/update_user_department', methods=['POST']) +def update_user_department(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + # 只有admin用户才能更新用户部门 + if session['username'] != 'admin': + return jsonify({'success': False, 'error': '无权限访问'}), 403 + + data = request.get_json() + user_id = data.get('user_id') + department = data.get('department') + + # 验证输入 + if not user_id or not department: + return jsonify({'success': False, 'error': '缺少必要参数'}) + + # 更新用户部门 + try: + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + cursor.execute("UPDATE users SET department = ? WHERE id = ?", (department, user_id)) + conn.commit() + conn.close() + + return jsonify({'success': True, 'message': '部门信息更新成功'}) + except Exception as e: + conn.close() + return jsonify({'success': False, 'error': f'更新失败：{str(e)}'}) + +# 重置用户密码路由 +@bp.route('/reset_user_password', methods=['POST']) +def reset_user_password(): + if 'username' not in session: + return jsonify({'success': False, 'error': '未登录'}), 401 + + # 只有admin用户才能重置用户密码 + if session['username'] != 'admin': + return jsonify({'success': False, 'error': '无权限访问'}), 403 + + data = request.get_json() + user_id = data.get('user_id') + username = data.get('username') + + # 验证输入 + if not user_id or not username: + return jsonify({'success': False, 'error': '缺少必要参数'}) + + # 重置密码为默认值 + default_password = 'default123' + + try: + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 检查用户ID和用户名是否匹配 + cursor.execute("SELECT id FROM users WHERE username = ?", (username,)) + db_user_id = cursor.fetchone() + + if not db_user_id or db_user_id[0] != user_id: + conn.close() + return jsonify({'success': False, 'error': '用户信息不匹配'}) + + # 更新users表中的密码 + cursor.execute("UPDATE users SET password = ? WHERE id = ?", (default_password, user_id)) + + # 更新login_info表中的密码 + cursor.execute("UPDATE login_info SET password = ? WHERE username = ?", (default_password, username)) + + # 如果login_info表中没有该用户的记录，则插入 + if cursor.rowcount == 0: + cursor.execute("INSERT INTO login_info (username, password, otp_code) VALUES (?, ?, ?)", + (username, default_password, "")) + + conn.commit() + conn.close() + + return jsonify({'success': True, 'message': '密码已初始化为：' + default_password}) + except Exception as e: + conn.close() + return jsonify({'success': False, 'error': f'密码重置失败：{str(e)}'}) \ No newline at end of file