Files
QueryCarInfo2/user.py
T
2025-09-23 20:15:03 +08:00

645 lines
23 KiB
Python

from datetime import datetime
from flask import Blueprint, render_template, request, jsonify, redirect, url_for, session
# 导入数据库工具
from db_utils import get_db_connection, close_db_connection
# 创建用户相关的蓝图
bp = Blueprint('user', __name__)
# 初始化数据库(用户相关部分)
def init_user_db():
conn = get_db_connection()
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
)
try:
conn.commit()
except Exception as e:
print(f"初始化用户数据库时出错: {e}")
finally:
close_db_connection(conn)
# 登录路由
@bp.route('/login', methods=['GET', 'POST'])
def login():
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
# 连接数据库
conn = get_db_connection()
cursor = conn.cursor()
# 验证用户名和密码
cursor.execute("SELECT * FROM users WHERE username = ? AND password = ?", (username, password))
user = cursor.fetchone()
if user:
# 检查用户状态
if user[6] != 'active':
close_db_connection(conn)
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')
close_db_connection(conn)
# 登录成功,跳转到查询页面
return redirect(url_for('query'))
else:
close_db_connection(conn)
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 = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE username = ?", (username,))
if cursor.fetchone():
close_db_connection(conn)
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()
close_db_connection(conn)
# 注册成功,跳转到登录页面
return redirect(url_for('user.login', success='注册成功,请登录'))
except Exception as e:
close_db_connection(conn)
return render_template('register.html', error=f'注册失败:{str(e)}')
return render_template('register.html')
# 获取个人信息路由
@bp.route('/get_user_profile', methods=['GET'])
def get_user_profile():
if 'username' not in session:
return jsonify({'success': False, 'error': '未登录'}), 401
# 连接数据库,获取用户信息
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],))
user = cursor.fetchone()
close_db_connection(conn)
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})
# 获取所有用户列表的API
@bp.route('/api/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
# 获取URL参数中的页码、每页数量、排序字段和排序方式
page = request.args.get('page', 1, type=int)
page_size = request.args.get('pageSize', 10, type=int)
sort_field = request.args.get('sortField', 'id')
sort_order = request.args.get('sortOrder', 'asc')
department = request.args.get('department', '')
search = request.args.get('search', '')
# 验证排序字段是否有效
valid_sort_fields = ['id', 'username', 'name', 'phone', 'department', 'status']
if sort_field not in valid_sort_fields:
sort_field = 'id'
# 验证排序方式是否有效
if sort_order not in ['asc', 'desc']:
sort_order = 'asc'
# 计算偏移量
offset = (page - 1) * page_size
try:
# 连接数据库
conn = get_db_connection()
cursor = conn.cursor()
# 构建查询条件
conditions = []
params = []
if department:
conditions.append("department = ?")
params.append(department)
if search:
conditions.append("(username LIKE ? OR name LIKE ? OR phone LIKE ?)")
params.extend([f'%{search}%', f'%{search}%', f'%{search}%'])
# 构建SQL查询语句
where_clause = " AND ".join(conditions) if conditions else ""
if where_clause:
where_clause = "WHERE " + where_clause
# 查询用户列表
cursor.execute(
f"SELECT id, username, name, phone, department, status FROM users {where_clause} ORDER BY {sort_field} {sort_order} LIMIT ? OFFSET ?",
params + [page_size, offset]
)
users = cursor.fetchall()
# 查询总记录数
cursor.execute(f"SELECT COUNT(*) FROM users {where_clause}", params)
total = cursor.fetchone()[0]
# 关闭数据库连接
close_db_connection(conn)
# 格式化用户数据
formatted_users = []
for user in users:
formatted_users.append({
'id': user[0],
'username': user[1],
'name': user[2],
'phone': user[3],
'department': user[4],
'status': user[5]
})
# 返回分页结果
return jsonify({
'success': True,
'data': formatted_users,
'total': total,
'page': page,
'pageSize': page_size
})
except Exception as e:
# 确保在异常情况下也关闭连接
close_db_connection(conn)
return jsonify({'success': False, 'error': str(e)}), 500
# 更新用户状态路由
@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': '缺少必要参数'})
# 不能禁用admin用户
if user_id == 1:
return jsonify({'success': False, 'error': '不能修改管理员用户的状态'})
# 更新用户状态
try:
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("UPDATE users SET status = ? WHERE id = ?", (status, user_id))
conn.commit()
close_db_connection(conn)
return jsonify({'success': True, 'message': '用户状态更新成功'})
except Exception as e:
close_db_connection(conn)
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 = get_db_connection()
cursor = conn.cursor()
cursor.execute("UPDATE users SET department = ? WHERE id = ?", (department, user_id))
conn.commit()
close_db_connection(conn)
return jsonify({'success': True, 'message': '部门信息更新成功'})
except Exception as e:
close_db_connection(conn)
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 = get_db_connection()
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:
close_db_connection(conn)
return jsonify({'success': False, 'error': '用户信息不匹配'})
# 更新users表中的密码
cursor.execute("UPDATE users SET password = ? WHERE id = ?", (default_password, user_id))
conn.commit()
close_db_connection(conn)
return jsonify({'success': True, 'message': '密码已初始化为:' + default_password})
except Exception as e:
close_db_connection(conn)
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 = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT department FROM users")
departments = [dept[0] for dept in cursor.fetchall()]
close_db_connection(conn)
return render_template('users_management.html', departments=departments)
# 创建参数配置表(如果不存在)
def create_parameter_config_table():
conn = get_db_connection()
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'))
)
try:
conn.commit()
except Exception as e:
print(f"插入参数配置数据时出错: {e}")
close_db_connection(conn)
# 获取参数配置路由
@bp.route('/get_parameter_config', methods=['GET'])
def get_parameter_config():
if 'username' not in session:
return jsonify({'success': False, 'error': '未登录'}), 401
# 确保参数配置表存在
create_parameter_config_table()
try:
conn = get_db_connection()
cursor = conn.cursor()
# 获取最新的参数配置
cursor.execute("SELECT param1, param2, param3, param4, param5 FROM parameter_config ORDER BY update_time DESC LIMIT 1")
config = cursor.fetchone()
close_db_connection(conn)
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': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
}
return jsonify({
'success': True,
'parameters': parameters
})
except Exception as e:
close_db_connection(conn)
return jsonify({'success': False, 'error': str(e)})
# 保存参数配置路由
@bp.route('/save_parameter_config', methods=['POST'])
def save_parameter_config():
if 'username' not in session:
return jsonify({'success': False, 'error': '未登录'}), 401
# 确保参数配置表存在
create_parameter_config_table()
data = request.get_json()
param1 = data.get('param1')
param2 = data.get('param2')
param3 = data.get('param3')
param5 = data.get('param5')
# 获取当前param4的值(保持不变)
current_param4 = None
try:
conn = get_db_connection()
cursor = conn.cursor()
# 获取最新的参数配置中的param4值
cursor.execute("SELECT param4 FROM parameter_config ORDER BY update_time DESC LIMIT 1")
latest_config = cursor.fetchone()
if latest_config:
current_param4 = latest_config[0]
# 插入新的参数配置记录(不更新旧记录,而是插入新记录)
# 注意:param4使用当前数据库中已有的值,确保其保持不变
cursor.execute(
"INSERT INTO parameter_config (param1, param2, param3, param4, param5, update_time) VALUES (?, ?, ?, ?, ?, ?)",
(param1, param2, param3, current_param4, param5, datetime.now().strftime('%Y-%m-%d %H:%M:%S'))
)
conn.commit()
close_db_connection(conn)
return jsonify({'success': True, 'message': '参数配置保存成功'})
except Exception as e:
if 'conn' in locals() and conn:
close_db_connection(conn)
return jsonify({'success': False, 'error': str(e)})
# 参数配置页面路由
@bp.route('/parameter_config_page')
def parameter_config_page():
if 'username' not in session:
return redirect(url_for('user.login'))
return render_template('parameter_config.html')
# 修改密码页面路由
@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 = get_db_connection()
cursor = conn.cursor()
# 验证旧密码是否正确
cursor.execute("SELECT password FROM users WHERE username = ?", (session['username'],))
result = cursor.fetchone()
if not result or result[0] != old_password:
close_db_connection(conn)
return render_template('change_password.html', error='旧密码不正确')
# 更新密码
try:
cursor.execute(
"UPDATE users SET password = ? WHERE username = ?",
(new_password, session['username'])
)
conn.commit()
close_db_connection(conn)
# 密码修改成功,跳转到登录页面
return redirect(url_for('user.login', success='密码修改成功,请重新登录'))
except Exception as e:
close_db_connection(conn)
return render_template('change_password.html', error=f'密码修改失败:{str(e)}')
return render_template('change_password.html')
# 个人信息页面路由
@bp.route('/user_profile_page', methods=['GET', 'POST'])
def user_profile_page():
if 'username' not in session:
return redirect(url_for('user.login'))
# 连接数据库,获取用户信息
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],))
user_tuple = cursor.fetchone()
close_db_connection(conn)
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 = get_db_connection()
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
close_db_connection(conn)
# 重新获取用户信息
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute("SELECT username, name, phone, department FROM users WHERE username = ?", (session['username'],))
updated_user_tuple = cursor.fetchone()
close_db_connection(conn)
# 将元组转换为字典
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:
close_db_connection(conn)
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)