bwadmin/QueryCarInfo2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
461c5bb508c13547b8e4e9335fb3233a6090c2d8
QueryCarInfo2/app.py
T
bwadmin 461c5bb508 20250917
2025-09-17 21:38:45 +08:00

805 lines
31 KiB
Python
Raw Blame History

from flask import Flask, render_template, request, redirect, url_for, session
from datetime import datetime, timedelta
import sqlite3
import os
import random
from flask import Flask, render_template, request, redirect, url_for, session, jsonify
app = Flask(__name__)
app.secret_key = 'your_secret_key' # 生产环境中应该使用更安全的密钥
# 确保模板文件夹存在
templates_dir = os.path.join(os.path.dirname(__file__), 'templates')
if not os.path.exists(templates_dir):
os.makedirs(templates_dir)
# 初始化数据库
def init_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 (
id INTEGER PRIMARY KEY AUTOINCREMENT,
plate_number TEXT NOT NULL UNIQUE,
phone TEXT NOT NULL,
id_card TEXT NOT NULL,
name TEXT NOT NULL,
email TEXT,
address TEXT
)
''')
# 创建查询历史表
cursor.execute('''
CREATE TABLE IF NOT EXISTS query_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
department TEXT NOT NULL,
query_date TEXT NOT NULL,
plate_number TEXT,
phone TEXT,
results_count INTEGER NOT NULL,
query_ip TEXT NOT NULL
)
''')
# 创建登录信息表
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', '上海市静安区'),
('粤C67890', '13700137001', '440401199001013456', '王五', 'wangwu@example.com', '广东省珠海市'),
('苏D12345', '13600136001', '320401199001014567', '赵六', 'zhaoliu@example.com', '江苏省常州市'),
('浙E54321', '13500135001', '330501199001015678', '钱七', 'qianqi@example.com', '浙江省湖州市')
]
cursor.executemany("INSERT INTO car_owners (plate_number, phone, id_card, name, email, address) VALUES (?, ?, ?, ?, ?, ?)", test_cars)
except sqlite3.IntegrityError:
# 数据已存在，忽略
pass
# 单独处理查询历史演示数据，确保每次都能插入20条记录
try:
# 先删除现有的演示数据
cursor.execute("DELETE FROM query_history WHERE department IN ('IT部', '财务部', '人事部', '市场部', '行政部')")
# 生成最近20天的随机查询历史
provinces = ['京', '沪', '粤', '苏', '浙', '鲁', '豫', '川', '湘', '鄂']
cities = ['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', 'J', 'K']
departments = ['IT部', '财务部', '人事部', '市场部', '行政部']
# 获取现有的车辆信息，用于生成能匹配的查询历史
cursor.execute("SELECT plate_number, phone FROM car_owners")
existing_cars = cursor.fetchall()
for i in range(20):
# 生成随机日期（最近20天内）
days_ago = random.randint(0, 19)
date = (datetime.now() - timedelta(days=days_ago)).strftime('%Y-%m-%d %H:%M:%S')
# 70%的概率使用已存在的车辆信息，30%的概率生成随机信息
use_existing = random.random() < 0.7
if use_existing and existing_cars:
# 使用已存在的车辆信息
car_info = random.choice(existing_cars)
if random.choice([True, False]):
plate_number = car_info[0]
phone = None
else:
plate_number = None
phone = car_info[1]
# 确保结果数量正确
if plate_number:
cursor.execute("SELECT COUNT(*) FROM car_owners WHERE plate_number = ?", (plate_number,))
results_count = cursor.fetchone()[0]
else:
cursor.execute("SELECT COUNT(*) FROM car_owners WHERE phone = ?", (phone,))
results_count = cursor.fetchone()[0]
else:
# 随机生成车牌号或手机号
use_plate = random.choice([True, False])
if use_plate:
plate_number = f"{random.choice(provinces)}{random.choice(cities)}{random.randint(100000, 999999)}"
phone = None
else:
plate_number = None
phone = f"1{random.randint(3000000000, 9999999999)}"
# 检查是否存在匹配结果
if plate_number:
cursor.execute("SELECT COUNT(*) FROM car_owners WHERE plate_number = ?", (plate_number,))
results_count = cursor.fetchone()[0]
else:
cursor.execute("SELECT COUNT(*) FROM car_owners WHERE phone = ?", (phone,))
results_count = cursor.fetchone()[0]
# 随机部门
department = random.choice(departments)
# 生成随机IP地址
query_ip = f"192.168.{random.randint(0, 255)}.{random.randint(1, 254)}"
cursor.execute(
"INSERT INTO query_history (department, query_date, plate_number, phone, results_count, query_ip) VALUES (?, ?, ?, ?, ?, ?)",
(department, date, plate_number, phone, results_count, query_ip)
)
except Exception as e:
print(f"插入查询历史数据时出错: {e}")
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('/query', methods=['GET', 'POST'])
def query():
if 'username' not in session:
return redirect(url_for('login'))
# 获取最近查询历史
formatted_recent_queries = get_recent_queries()
if request.method == 'POST':
plate_number = request.form.get('plate_number', '').strip()
phone = request.form.get('phone', '').strip()
if not plate_number and not phone:
return render_template('query.html', error='请至少输入车牌号或手机号', recent_queries=formatted_recent_queries)
conn = sqlite3.connect('car_info.db')
cursor = conn.cursor()
query_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
if plate_number and phone:
cursor.execute("SELECT * FROM car_owners WHERE plate_number = ? AND phone = ?", (plate_number, phone))
elif plate_number:
cursor.execute("SELECT * FROM car_owners WHERE plate_number = ?", (plate_number,))
else:
cursor.execute("SELECT * FROM car_owners WHERE phone = ?", (phone,))
results = cursor.fetchall()
# 获取客户端IP地址
query_ip = request.remote_addr
# 从用户信息表获取部门信息
try:
conn_dept = sqlite3.connect('car_info.db')
cursor_dept = conn_dept.cursor()
cursor_dept.execute("SELECT department FROM users WHERE username = ?", (session['username'],))
user_dept = cursor_dept.fetchone()
department = user_dept[0] if user_dept else '未知部门'
except Exception as e:
print(f"获取部门信息时出错: {e}")
department = '未知部门'
finally:
if 'conn_dept' in locals():
conn_dept.close()
# 记录查询历史到数据库
try:
conn_history = sqlite3.connect('car_info.db')
cursor_history = conn_history.cursor()
cursor_history.execute(
"INSERT INTO query_history (department, query_date, plate_number, phone, results_count, query_ip) VALUES (?, ?, ?, ?, ?, ?)",
(department, query_date, plate_number if plate_number else None, phone if phone else None, len(results), query_ip)
)
conn_history.commit()
except Exception as e:
print(f"记录查询历史时出错: {e}")
finally:
if 'conn_history' in locals():
conn_history.close()
# 处理查询结果
car_owners = []
for result in results:
car_owners.append({
'id': result[0],
'plate_number': result[1],
'phone': result[2],
'id_card': result[3],
'name': result[4],
'email': result[5],
'address': result[6]
})
# 构建查询历史信息
query_history = {
'date': query_date,
'plate_number': plate_number,
'phone': phone,
'results_count': len(car_owners)
}
# 使用辅助函数获取更新后的最近查询历史
formatted_recent_queries = get_recent_queries()
return render_template('query.html', query_history=query_history, car_owners=car_owners, recent_queries=formatted_recent_queries)
return render_template('query.html', recent_queries=formatted_recent_queries)
# 获取查询历史详情的路由
@app.route('/query_history_detail/<int:history_id>')
def query_history_detail(history_id):
if 'username' not in session:
return jsonify({'error': '未登录'}), 401
conn = sqlite3.connect('car_info.db')
cursor = conn.cursor()
# 获取查询历史记录
cursor.execute("SELECT plate_number, phone FROM query_history WHERE id = ?", (history_id,))
history = cursor.fetchone()
if not history:
conn.close()
return jsonify({'error': '查询历史不存在'}), 404
plate_number, phone = history
# 根据车牌号和手机号查询车辆信息
if plate_number and phone:
cursor.execute("SELECT * FROM car_owners WHERE plate_number = ? AND phone = ?", (plate_number, phone))
elif plate_number:
cursor.execute("SELECT * FROM car_owners WHERE plate_number = ?", (plate_number,))
else:
cursor.execute("SELECT * FROM car_owners WHERE phone = ?", (phone,))
results = cursor.fetchall()
conn.close()
# 格式化结果
car_owners = []
for result in results:
car_owners.append({
'id': result[0],
'plate_number': result[1],
'phone': result[2],
'id_card': result[3],
'name': result[4],
'email': result[5],
'address': result[6]
})
return jsonify({'car_owners': car_owners})
# 获取最近查询历史的辅助函数
def get_recent_queries():
# 获取最近20条查询历史记录，按查询日期降序排列
conn = sqlite3.connect('car_info.db')
cursor = conn.cursor()
cursor.execute("SELECT id, department, query_date, plate_number, phone, results_count, query_ip FROM query_history ORDER BY query_date DESC LIMIT 20")
recent_queries = cursor.fetchall()
conn.close()
# 格式化查询历史数据
formatted_recent_queries = []
for q in recent_queries:
formatted_recent_queries.append({
'id': q[0],
'department': q[1],
'query_date': q[2],
'plate_number': q[3] if q[3] else '未提供',
'phone': q[4] if q[4] else '未提供',
'results_count': q[5],
'query_ip': q[6]
})
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)
Reference in New Issue View Git Blame Copy Permalink