From 461c5bb508c13547b8e4e9335fb3233a6090c2d8 Mon Sep 17 00:00:00 2001 From: bwstudio <7469504@qq.com> Date: Wed, 17 Sep 2025 21:38:45 +0800 Subject: [PATCH] 20250917 --- app.py | 805 ++++++++++++++++++++++ car_info.db | Bin 0 -> 36864 bytes run.ps1 | 15 + start_server.py | 14 + templates/login.html | 81 +++ templates/query.html | 1001 ++++++++++++++++++++++++++++ templates/register.html | 106 +++ templates/users_management.html | 1105 +++++++++++++++++++++++++++++++ update_database.py | 72 ++ verify_database.py | 41 ++ 10 files changed, 3240 insertions(+) create mode 100644 app.py create mode 100644 car_info.db create mode 100644 run.ps1 create mode 100644 start_server.py create mode 100644 templates/login.html create mode 100644 templates/query.html create mode 100644 templates/register.html create mode 100644 templates/users_management.html create mode 100644 update_database.py create mode 100644 verify_database.py diff --git a/app.py b/app.py new file mode 100644 index 0000000..cec506a --- /dev/null +++ b/app.py @@ -0,0 +1,805 @@ +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/') +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) \ No newline at end of file diff --git a/car_info.db b/car_info.db new file mode 100644 index 0000000000000000000000000000000000000000..88856efc6daf022860e5b671ad29ef2ff170030a GIT binary patch literal 36864 zcmeI*S!~;68~|{;>5=q^5my)zs;fP8RcY|&+hv_7!|N(aH%f{&ULvpYwm^z$1U#u^l&Sk&vb#^miNA*Z(_y z-)Fb69ruaHr?M53Jy<%Xuoz55ch2bJDVo9&9#ml*>np zrA&CUV12PNpPnmb)+H0H-?e8(pPZO>O0vz%bBLYMcj+_w^iKT=)?QdsggGfm5r?2d@=^S%Lt(dTJt$rnJ!fG2h5VQP1CBw#ey3bEGwDi zYOYd-2B<==UgLXh8EoffoShG6>q0Ny+wS(g7ag-Xqm(WlEkKh5Qyn3#VEMnMYN_TX z{m-wqzfDlDY$gq**|2Z6cJCn6%p2L- zzvJ2Hx@8mEpfG2yn{5xtu3NdyF8}=yIkfly-IYir=$FTwG@U)h8MTGt8+($!i{Fw1 z!e;36n>FS???2^=QK`Dk!#<+#8lRm?v4h53C40mi^m~Wfa6@gKiLt~+hYtrTO`IPQ z_;nJsf2H)ni)OA-@Qnispa2S>01BW03ZMWApa2S>01BYMe_lYOqg3L){;i2EMkb#v zI3vOft6!a8edE>Io2QAOa-1{bCR5j~Pu)`p-B2%@`ICZg98drSPyhu`00mG01yBG5 zPyhu`00sVE0?|k!=A0q0um86*KTymc%qsH>OyGb5D1ZVefC4Ch0w{n2D1ZVefC4D+ zUlh2#Ju=u9`ZN#yM?4Z4po33mbo8}F24cZyhN3jQ|93tJV88$Gh>uas$IMjxxA?m- zi319t01BW03ZMWApa2S>01BW03WNprbwsKDtz-)!^^5Q6k|^+m2$Fr5xib3Nv~bJSbOEm=MNjjT(%lY`oag00wi5qT5^-#7x4D@$h~<<-|etY5qYg?_Tl zO?iaWzy9pu28UgH!5INUP-weG^>e3cXU^oZ*wC9eRZsMa``E^v)^Wx z$jb5;;|&b&W`06K>Dp^+-+j2UbPO;Ghss8wS)+FHT}T3oT0eX4`sr^VZTtTJ4(1BQ z{La*vD{u$E)o>Hz915TS3ZMWApa2S>01BW03ZMWApujB^=!^8z{&|-Ud(J;s(r(ZB zr#B+bmfk6ew%$m8%s-J}U;n?m`xA=!k$H`oU^?SJ#gD~z$76lh`i}MO?u+(b?LFT6 zXm3x?&poGl_V(P-{VQyY0}7x33ZMWApa2S>01Dhp0m(W7&$be_Y>D(9DEVMs@idPZ5+qh+q++W)TmW~>uo|m6&r*rWwo=* z@USc|M^0?y$jknp*F@u!L{^i6lGMZpt#TuHXzkmNYRAtv1^dP2<=_HgBbW$cQWmyb zrAGAd%JTBc@(Dj$9T7M`YyaOzF8JXsnX77&C}hlvS$0dRB5QuQ2H6v7yH)V&-YORq-hb!f zWllukSG4U`e#3B1_DkOyF3U-x@?%!ct$g$bl@H|M`m;=+u8QiH9_5gT^fFA&4eSq~PAjB_0OXCwx zgzgLBhwX5?^;+`}_-mKWO{gj-bCQ?63iTJ%9o8YY_|ew-hdK{Dg^uTW1#T7adOimY zpz)*DL3>}Nwnm4;`*@Hy+)6-rO%r8-^MZL|?|aa7%|_u{bLYH+1TU{$_*_>ciQ`p2 z-p*VhJFJXb`q*$w<(oQB@CZG>^@$1z2+(?S8z<0IgWyI3uLQ*If0{UI9k_Azg98px zR9TZWL4-Z8VaH2=&k3q!*ui$&h3uM&4u^~hQ4>_ncjqM`sR{1$|LYV}znKScOdADI z00mG01yBG5Pyhu`00mG01yJDT3*19@!2bpa|6>3}6p5tb^Z#$YKIVV|D1ZVefC4Ch U0w{n2D1ZVefC9Hjz<>V#--S6Ng#Z8m literal 0 HcmV?d00001 diff --git a/run.ps1 b/run.ps1 new file mode 100644 index 0000000..f5d7870 --- /dev/null +++ b/run.ps1 @@ -0,0 +1,15 @@ +# 设置执行策略以允许运行脚本（仅在首次运行时需要取消注释） +# Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope Process + +# 激活虚拟环境 +& '.\venv\Scripts\Activate.ps1' + +# 设置Flask应用程序环境变量 +$env:FLASK_APP = "app.py" +$env:FLASK_ENV = "development" + +# 启动Flask服务器，允许外部访问（使用python -m flask确保命令可用） +python -m flask run --host=0.0.0.0 + +# 保持窗口打开 +Read-Host "Press Enter to exit" \ No newline at end of file diff --git a/start_server.py b/start_server.py new file mode 100644 index 0000000..f72275e --- /dev/null +++ b/start_server.py @@ -0,0 +1,14 @@ +import os +import subprocess + +# 设置Flask应用程序环境变量 +os.environ['FLASK_APP'] = 'app.py' +os.environ['FLASK_ENV'] = 'development' + +print("Starting Flask server...") +print("Server will be available at http://localhost:5000/") +print("To allow external access, it's listening on all interfaces (0.0.0.0)") +print("Press Ctrl+C to stop the server") + +# 启动Flask服务器，允许外部访问 +subprocess.run(['python', '-m', 'flask', 'run', '--host=0.0.0.0']) \ No newline at end of file diff --git a/templates/login.html b/templates/login.html new file mode 100644 index 0000000..a3c8694 --- /dev/null +++ b/templates/login.html @@ -0,0 +1,81 @@ + + + + + + 车主信息查询系统 - 登录 + + + + + + + + + + + + +
+
+

车主信息查询系统

+
+
+
+ + +
+
+ + +
+ {% if error %} +
+ {{ error }} +
+ {% endif %} + {% if success %} +
+ {{ success }} +
+ {% endif %} + +
+ 注册账号 +
+
+
+ + \ No newline at end of file diff --git a/templates/query.html b/templates/query.html new file mode 100644 index 0000000..405b229 --- /dev/null +++ b/templates/query.html @@ -0,0 +1,1001 @@ + + + + + + 车主信息查询系统 - 查询 + + + + + + + + + + + + + +
+
+
+ +

车主信息查询系统

+
+
+ +
+
+ 欢迎 {{ session.department }} {{ session.name }} + +
+ + +
+
+
+
+ + +
+ + {% if error %} +
+ + {{ error }} +
+ {% endif %} + {% if success %} +
+ + {{ success }} +
+ {% endif %} + + + {% if recent_queries %} +
+
+

+ 最近查询记录 +

+ +
+ +
+ + + + + + + + + + + + + + {% for query in recent_queries %} + + + + + + + + + + {% endfor %} + +
序号查询日期车牌号手机号查询结果登录部门登录IP
{{ loop.index }}{{ query.query_date }}{{ query.plate_number }}{{ query.phone }} + + {{ query.results_count }} 条记录 + + {{ query.department }}{{ query.query_ip }}
+
+
+ {% endif %} +
+ + + + + + + + +
+
+

© 2025 车主信息查询系统 - 内部使用

+

免责声明：本系统数据仅供内部人员工作参考，请勿外传或用于其他用途。

+
+
+ + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/templates/register.html b/templates/register.html new file mode 100644 index 0000000..99f220e --- /dev/null +++ b/templates/register.html @@ -0,0 +1,106 @@ + + + + + + 车主信息查询系统 - 注册 + + + + + + + + + + + + +
+
+

车主信息查询系统

+

用户注册

+
+
+
+
+ + +
+
+ + +
+
+
+
+ + +
+
+ + +
+
+
+ + +
+
+ + +
+ {% if error %} +
+ {{ error }} +
+ {% endif %} + {% if success %} +
+ {{ success }} +
+ {% endif %} + +
+ 已有账号？返回登录 +
+
+
+ + \ No newline at end of file diff --git a/templates/users_management.html b/templates/users_management.html new file mode 100644 index 0000000..3db1a80 --- /dev/null +++ b/templates/users_management.html @@ -0,0 +1,1105 @@ + + + + + + 车主信息查询系统 - 用户管理 + + + + + + + + + + + + + +
+
+
+ +

车主信息查询系统

+
+
+ +
+
+ 欢迎 {{ session.department }} {{ session.name }} + +
+ + +
+
+
+
+ + +
+ +
+

用户管理

+

管理系统用户、审核登录权限、修改部门信息和初始化密码

+
+ + +
+
+
+
+ + +
+
+
+ + + +
+
+
+ + + + + +
+
+ + + + + + + + + + + + + +
用户名姓名部门状态操作
+
+ +
+ +
+
+
+ + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/update_database.py b/update_database.py new file mode 100644 index 0000000..5a55f2c --- /dev/null +++ b/update_database.py @@ -0,0 +1,72 @@ +import sqlite3 + +# 连接到数据库并确保users表结构正确 +def update_users_table(): + try: + # 连接数据库 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 检查users表的当前结构 + cursor.execute("PRAGMA table_info(users)") + columns = [column[1] for column in cursor.fetchall()] + print(f"当前users表的字段: {columns}") + + # 使用不同的方法来添加字段，避免NOT NULL约束可能导致的问题 + if 'name' not in columns: + # 先添加可为空的字段 + cursor.execute("ALTER TABLE users ADD COLUMN name TEXT") + # 然后设置默认值并更新现有数据 + cursor.execute("UPDATE users SET name = '未知用户' WHERE name IS NULL") + print("已添加name字段并设置默认值") + + if 'phone' not in columns: + # 先添加可为空的字段 + cursor.execute("ALTER TABLE users ADD COLUMN phone TEXT") + # 然后设置默认值并更新现有数据 + cursor.execute("UPDATE users SET phone = '00000000000' WHERE phone IS NULL") + print("已添加phone字段并设置默认值") + + if 'department' not in columns: + # 先添加可为空的字段 + cursor.execute("ALTER TABLE users ADD COLUMN department TEXT") + # 然后设置默认值并更新现有数据 + cursor.execute("UPDATE users SET department = '未知部门' WHERE department IS NULL") + print("已添加department字段并设置默认值") + + # 无条件更新admin用户的信息，确保所有字段都有值 + cursor.execute( + "UPDATE users SET name = ?, phone = ?, department = ? WHERE username = ?", + ('管理员', '13800138000', 'IT部', 'admin') + ) + print("已更新admin用户信息") + + # 提交更改 + conn.commit() + + # 再次验证表结构 + cursor.execute("PRAGMA table_info(users)") + updated_columns = cursor.fetchall() + print("\n更新后的users表字段:") + for col in updated_columns: + print(f"字段名: {col[1]}, 类型: {col[2]}") + + # 检查admin用户的信息是否正确更新 + cursor.execute("SELECT id, username, name, phone, department FROM users WHERE username = ?", ('admin',)) + admin_user = cursor.fetchone() + if admin_user: + print("\n更新后的admin用户信息:") + print(f"ID: {admin_user[0]}, 用户名: {admin_user[1]}, 姓名: {admin_user[2]}, 电话: {admin_user[3]}, 部门: {admin_user[4]}") + + # 关闭连接 + conn.close() + print("\n数据库更新完成") + + except Exception as e: + print(f"更新数据库时出错: {str(e)}") + # 确保在异常情况下也关闭连接 + if 'conn' in locals(): + conn.close() + +if __name__ == "__main__": + update_users_table() \ No newline at end of file diff --git a/verify_database.py b/verify_database.py new file mode 100644 index 0000000..ee8d24d --- /dev/null +++ b/verify_database.py @@ -0,0 +1,41 @@ +import sqlite3 + +# 连接到数据库并验证users表结构 +def verify_users_table(): + try: + # 连接数据库 + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() + + # 检查users表的当前结构 + print("验证users表结构:") + cursor.execute("PRAGMA table_info(users)") + columns = cursor.fetchall() + for column in columns: + print(f"字段名: {column[1]}, 类型: {column[2]}, 是否可空: {column[3]}") + + # 检查admin用户的信息 + print("\n验证admin用户信息:") + cursor.execute("SELECT id, username, name, phone, department FROM users WHERE username = ?", ('admin',)) + admin_user = cursor.fetchone() + if admin_user: + print(f"ID: {admin_user[0]}") + print(f"用户名: {admin_user[1]}") + print(f"姓名: {admin_user[2]}") + print(f"电话: {admin_user[3]}") + print(f"部门: {admin_user[4]}") + else: + print("未找到admin用户") + + # 关闭连接 + conn.close() + print("\n数据库验证完成") + + except Exception as e: + print(f"验证数据库时出错: {str(e)}") + # 确保在异常情况下也关闭连接 + if 'conn' in locals(): + conn.close() + +if __name__ == "__main__": + verify_users_table() \ No newline at end of file