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' ) ''') # 检查是否存在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("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 ) 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') 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) ) 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'])) 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('/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)) 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)}'}) # 创建参数配置表（如果不存在） def create_parameter_config_table(): conn = sqlite3.connect('car_info.db') cursor = conn.cursor() # 创建参数配置表 cursor.execute(''' CREATE TABLE IF NOT EXISTS parameter_config ( id INTEGER PRIMARY KEY AUTOINCREMENT, param1 TEXT, param2 TEXT, param3 TEXT, param4 TEXT, param5 TEXT, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ''') # 检查是否存在参数配置记录，如果不存在则插入第一条记录 cursor.execute("SELECT COUNT(*) FROM parameter_config") count = cursor.fetchone()[0] if count == 0: cursor.execute( "INSERT INTO parameter_config (param1, param2, param3, param4, param5, update_time) VALUES (?, ?, ?, ?, ?, ?)", ('默认参数1', '默认参数2', '默认参数3', '默认参数4', '默认参数5', datetime.now().strftime('%Y-%m-%d %H:%M:%S')) ) conn.commit() conn.close() # 获取参数配置路由 @bp.route('/get_parameter_config', methods=['GET']) def get_parameter_config(): if 'username' not in session: return jsonify({'success': False, 'error': '未登录'}), 401 try: conn = sqlite3.connect('car_info.db') cursor = conn.cursor() # 查询参数配置 cursor.execute("SELECT param1, param2, param3, param4, param5, update_time FROM parameter_config ORDER BY update_time DESC LIMIT 1") config = cursor.fetchone() conn.close() if not config: return jsonify({'success': False, 'error': '未找到参数配置'}) # 构造返回数据 parameters = { 'param1': config[0], 'param2': config[1], 'param3': config[2], 'param4': config[3], 'param5': config[4], 'update_time': config[5] } return jsonify({'success': True, 'parameters': parameters}) except Exception as e: return jsonify({'success': False, 'error': f'获取参数配置失败：{str(e)}'}) # 保存参数配置路由 @bp.route('/save_parameter_config', methods=['POST']) def save_parameter_config(): if 'username' not in session: return jsonify({'success': False, 'error': '未登录'}), 401 # 只有管理员可以修改参数配置 if session['username'] != 'admin': return jsonify({'success': False, 'error': '无权限修改参数配置'}), 403 # 获取请求数据 data = request.get_json() if not data: return jsonify({'success': False, 'error': '请求数据格式错误'}), 400 param1 = data.get('param1', '').strip() param2 = data.get('param2', '').strip() param3 = data.get('param3', '').strip() param4 = data.get('param4', '').strip() param5 = data.get('param5', '').strip() # 验证参数 if not param1: return jsonify({'success': False, 'error': '参数1不能为空'}) try: conn = sqlite3.connect('car_info.db') cursor = conn.cursor() # 更新参数配置 cursor.execute( "UPDATE parameter_config SET param1 = ?, param2 = ?, param3 = ?, param4 = ?, param5 = ?, update_time = ? WHERE id = 1", (param1, param2, param3, param4, param5, datetime.now().strftime('%Y-%m-%d %H:%M:%S')) ) # 如果没有更新任何记录，插入新记录 if cursor.rowcount == 0: cursor.execute( "INSERT INTO parameter_config (param1, param2, param3, param4, param5, update_time) VALUES (?, ?, ?, ?, ?, ?)", (param1, param2, param3, param4, param5, datetime.now().strftime('%Y-%m-%d %H:%M:%S')) ) conn.commit() conn.close() return jsonify({'success': True, 'message': '参数配置保存成功'}) except Exception as e: return jsonify({'success': False, 'error': f'保存参数配置失败：{str(e)}'}) # 参数配置页面路由 @bp.route('/parameter_config_page', methods=['GET', 'POST']) def parameter_config_page(): if 'username' not in session: return redirect(url_for('user.login')) # 只有管理员可以访问参数配置页面 if session['username'] != 'admin': return redirect(url_for('user.user_profile_page', error='无权限访问参数配置页面')) # 渲染参数配置页面 return render_template('parameter_config.html', current_year=datetime.now().year) # 修改密码页面路由 @bp.route('/change_password_page', methods=['GET', 'POST']) def change_password_page(): if 'username' not in session: return redirect(url_for('user.login')) if request.method == 'POST': 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 render_template('change_password.html', error='请填写所有必填字段') # 验证新密码和确认密码是否一致 if new_password != confirm_password: return render_template('change_password.html', 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 render_template('change_password.html', 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 render_template('change_password.html', error='旧密码不正确') # 更新密码 cursor.execute("UPDATE users SET password = ? WHERE username = ?", (new_password, session['username'])) conn.commit() conn.close() # 密码修改成功，先执行退出功能，再跳转到登录页面 session.clear() return redirect(url_for('user.login', success='修改密码成功，请重新登录')) # GET请求，渲染修改密码页面 return render_template('change_password.html', current_year=datetime.now().year) # 个人信息页面路由 @bp.route('/user_profile_page', methods=['GET', 'POST']) def user_profile_page(): if 'username' not in session: return redirect(url_for('user.login')) # 连接数据库，获取用户信息 conn = sqlite3.connect('car_info.db') cursor = conn.cursor() cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],)) user_tuple = cursor.fetchone() conn.close() if not user_tuple: return redirect(url_for('user.login', error='用户不存在')) # 将元组转换为字典，以便在模板中使用属性访问 user = { 'username': user_tuple[0], 'name': user_tuple[1], 'phone': user_tuple[2], 'department': user_tuple[3] } if request.method == 'POST': 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 render_template('user_profile.html', user=user, error='请填写所有必填字段', current_year=datetime.now().year) # 连接数据库，更新用户信息 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() # 更新会话中的部门信息 session['department'] = department session['name'] = name conn.close() # 重新获取用户信息 conn = sqlite3.connect('car_info.db') cursor = conn.cursor() cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],)) updated_user_tuple = cursor.fetchone() conn.close() # 将元组转换为字典 updated_user = { 'username': updated_user_tuple[0], 'name': updated_user_tuple[1], 'phone': updated_user_tuple[2], 'department': updated_user_tuple[3] } return render_template('user_profile.html', user=updated_user, success='个人信息更新成功', current_year=datetime.now().year) except Exception as e: conn.close() return render_template('user_profile.html', user=user, error=f'更新失败，请重试。错误：{str(e)}', current_year=datetime.now().year) # GET请求，渲染个人信息页面 return render_template('user_profile.html', user=user, current_year=datetime.now().year)