346 lines
13 KiB
Python
346 lines
13 KiB
Python
from flask import Flask, render_template, request, redirect, url_for, session, jsonify
|
|
from datetime import datetime, timedelta
|
|
import os
|
|
import random
|
|
from flask import Flask, render_template, request, redirect, url_for, session
|
|
|
|
# 导入用户模块
|
|
from user import bp as user_bp, init_user_db
|
|
# 导入API模块
|
|
from api import api_bp
|
|
# 导入数据库工具
|
|
from db_utils import get_db_connection, close_db_connection, close_global_connection
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = 'your_secret_key' # 生产环境中应该使用更安全的密钥
|
|
|
|
# 注册用户蓝图
|
|
app.register_blueprint(user_bp)
|
|
# 注册API蓝图
|
|
app.register_blueprint(api_bp)
|
|
|
|
# 确保模板文件夹存在
|
|
templates_dir = os.path.join(os.path.dirname(__file__), 'templates')
|
|
if not os.path.exists(templates_dir):
|
|
os.makedirs(templates_dir)
|
|
|
|
# 初始化数据库
|
|
def init_db():
|
|
# 初始化用户相关的数据库
|
|
init_user_db()
|
|
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
# 创建车辆信息表
|
|
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
|
|
)
|
|
''')
|
|
|
|
# 插入一些测试车辆数据
|
|
try:
|
|
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 Exception as e:
|
|
# 数据已存在或其他错误，忽略
|
|
print(f"插入测试数据时出错: {e}")
|
|
|
|
# 单独处理查询历史演示数据，确保每次都能插入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}")
|
|
try:
|
|
conn.commit()
|
|
except Exception as e:
|
|
print(f"提交数据库更改时出错: {e}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
# 首页重定向到登录页
|
|
@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('user.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 = get_db_connection()
|
|
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:
|
|
# 可以复用同一个连接，不需要创建新连接
|
|
cursor.execute("SELECT department FROM users WHERE username = ?", (session['username'],))
|
|
user_dept = cursor.fetchone()
|
|
if user_dept:
|
|
department = user_dept[0]
|
|
except Exception as e:
|
|
print(f"获取部门信息时出错: {e}")
|
|
|
|
# 记录查询历史到数据库
|
|
try:
|
|
# 继续使用同一个连接
|
|
# 检查是否是从"增加"按钮提交的请求
|
|
is_add_query = request.form.get('is_add_query', 'false').lower() == 'true'
|
|
|
|
# 如果是增加按钮提交的请求，将results_count设置为-1
|
|
if is_add_query:
|
|
results_count = -1
|
|
else:
|
|
results_count = len(results)
|
|
|
|
cursor.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, results_count, query_ip)
|
|
)
|
|
conn.commit()
|
|
except Exception as e:
|
|
print(f"记录查询历史时出错: {e}")
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
# 处理查询结果
|
|
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 = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
# 获取查询历史记录
|
|
cursor.execute("SELECT plate_number, phone FROM query_history WHERE id = ?", (history_id,))
|
|
history = cursor.fetchone()
|
|
|
|
if not history:
|
|
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()
|
|
|
|
# 格式化结果
|
|
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})
|
|
except Exception as e:
|
|
print(f"查询历史详情时出错: {e}")
|
|
return jsonify({'error': '查询失败'}), 500
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
# 获取最近查询历史的辅助函数
|
|
def get_recent_queries():
|
|
# 获取最近20条查询历史记录，按查询日期降序排列
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
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()
|
|
|
|
# 格式化查询历史数据
|
|
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
|
|
except Exception as e:
|
|
print(f"获取最近查询历史时出错: {e}")
|
|
return []
|
|
finally:
|
|
close_db_connection(conn)
|
|
|
|
import signal
|
|
import sys
|
|
|
|
def signal_handler(sig, frame):
|
|
"""处理终止信号，确保在程序退出时关闭数据库连接"""
|
|
print('\n接收到终止信号，正在关闭数据库连接...')
|
|
close_global_connection()
|
|
print('数据库连接已关闭')
|
|
sys.exit(0)
|
|
|
|
if __name__ == '__main__':
|
|
# 注册信号处理程序
|
|
signal.signal(signal.SIGINT, signal_handler) # 处理Ctrl+C
|
|
signal.signal(signal.SIGTERM, signal_handler) # 处理终止信号
|
|
|
|
try:
|
|
init_db()
|
|
app.run(debug=True)
|
|
finally:
|
|
# 确保在程序退出时关闭数据库连接
|
|
close_global_connection() |