293 lines
9.8 KiB
Python
293 lines
9.8 KiB
Python
from flask import Flask, render_template, request, redirect, url_for, session, jsonify
|
|
from datetime import datetime, timedelta
|
|
import os
|
|
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:
|
|
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:
|
|
# 存储错误信息到session
|
|
session['error_message'] = '请至少输入车牌号或手机号'
|
|
return redirect(url_for('query'))
|
|
|
|
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
|
|
|
|
# 从用户信息表获取用户ID
|
|
user_id = None
|
|
try:
|
|
# 可以复用同一个连接，不需要创建新连接
|
|
cursor.execute("SELECT id FROM users WHERE username = ?", (session['username'],))
|
|
user_info = cursor.fetchone()
|
|
if user_info:
|
|
user_id = user_info[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 (query_date, plate_number, phone, results_count, query_ip, user_id) VALUES (?, ?, ?, ?, ?, ?)",
|
|
(query_date, plate_number if plate_number else None, phone if phone else None, results_count, query_ip, user_id)
|
|
)
|
|
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)
|
|
}
|
|
|
|
# 将查询结果和历史信息存储到session中
|
|
session['query_results'] = {
|
|
'car_owners': car_owners,
|
|
'query_history': query_history
|
|
}
|
|
|
|
# 重定向到GET请求，避免刷新页面时的重复提交提示
|
|
return redirect(url_for('query'))
|
|
|
|
# GET请求时，从session中获取查询结果（如果有）
|
|
query_results = session.pop('query_results', None)
|
|
error_message = session.pop('error_message', None)
|
|
|
|
# 重新获取最近查询历史，确保数据是最新的
|
|
formatted_recent_queries = get_recent_queries()
|
|
|
|
if query_results:
|
|
return render_template('query.html',
|
|
query_history=query_results['query_history'],
|
|
car_owners=query_results['car_owners'],
|
|
recent_queries=formatted_recent_queries,
|
|
error=error_message)
|
|
|
|
return render_template('query.html', recent_queries=formatted_recent_queries, error=error_message)
|
|
|
|
# 获取查询历史详情的路由
|
|
@app.route('/query_history_detail/<int:history_id>')
|
|
def query_history_detail(history_id):
|
|
"""
|
|
根据查询历史ID获取对应的车辆所有者信息
|
|
|
|
Args:
|
|
history_id: 查询历史记录的ID
|
|
|
|
Returns:
|
|
JSON: 包含车辆所有者信息的JSON响应
|
|
"""
|
|
# 检查用户是否已登录
|
|
if 'username' not in session:
|
|
return jsonify({'error': '未登录'}), 401
|
|
|
|
# 验证history_id是否为正整数
|
|
if not isinstance(history_id, int) or history_id <= 0:
|
|
return jsonify({'error': '无效的查询历史ID'}), 400
|
|
|
|
conn = get_db_connection()
|
|
cursor = conn.cursor()
|
|
|
|
try:
|
|
# 直接根据query_history_id从car_owners表查询对应记录
|
|
cursor.execute("SELECT * FROM car_owners WHERE query_history_id = ?", (history_id,))
|
|
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] if result[5] else '',
|
|
'address': result[6] if result[6] else ''
|
|
})
|
|
|
|
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:
|
|
# 使用JOIN操作获取用户姓名和部门信息
|
|
cursor.execute("""
|
|
SELECT qh.id, qh.query_date, qh.plate_number, qh.phone, qh.results_count, qh.query_ip, qh.user_id,
|
|
u.name AS user_name, u.department AS user_department
|
|
FROM query_history qh
|
|
LEFT JOIN users u ON qh.user_id = u.id
|
|
ORDER BY qh.query_date DESC LIMIT 20
|
|
""")
|
|
recent_queries = cursor.fetchall()
|
|
|
|
# 格式化查询历史数据
|
|
formatted_recent_queries = []
|
|
for q in recent_queries:
|
|
formatted_recent_queries.append({
|
|
'id': q[0],
|
|
'query_date': q[1],
|
|
'plate_number': q[2] if q[2] else '未提供',
|
|
'phone': q[3] if q[3] else '未提供',
|
|
'results_count': q[4],
|
|
'query_ip': q[5],
|
|
'user_id': q[6] if len(q) > 6 else None,
|
|
'user_name': q[7] if len(q) > 7 else '未知',
|
|
'user_department': q[8] if len(q) > 8 else '未知'
|
|
})
|
|
|
|
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,host="0.0.0.0",port=99)
|
|
finally:
|
|
# 确保在程序退出时关闭数据库连接
|
|
close_global_connection() |