20250819corp

2025-09-19 16:47:15 +08:00
parent fc5fb355b4
commit b0e8a68204
12 changed files with 1148 additions and 809 deletions
@@ -1,13 +1,15 @@
 from flask import Flask, render_template, request, redirect, url_for, session, jsonify
 from datetime import datetime, timedelta
-import sqlite3
 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'  # 生产环境中应该使用更安全的密钥
@@ -27,7 +29,7 @@ def init_db():
    # 初始化用户相关的数据库
    init_user_db()
    
-    conn = sqlite3.connect('car_info.db')
+    conn = get_db_connection()
    cursor = conn.cursor()
    
    # 创建车辆信息表
@@ -66,9 +68,9 @@ def init_db():
            ('浙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
+    except Exception as e:
+        # 数据已存在或其他错误，忽略
+        print(f"插入测试数据时出错: {e}")
    
    # 单独处理查询历史演示数据，确保每次都能插入20条记录
    try:
@@ -137,8 +139,12 @@ def init_db():
            )
    except Exception as e:
        print(f"插入查询历史数据时出错: {e}")
-    conn.commit()
-    conn.close()
+    try:
+        conn.commit()
+    except Exception as e:
+        print(f"提交数据库更改时出错: {e}")
+    finally:
+        close_db_connection(conn)

 # 首页重定向到登录页
@app.route('/')
@@ -161,7 +167,7 @@ def query():
        if not plate_number and not phone:
            return render_template('query.html', error='请至少输入车牌号或手机号', recent_queries=formatted_recent_queries)
        
-        conn = sqlite3.connect('car_info.db')
+        conn = get_db_connection()
        cursor = conn.cursor()
        
        query_date = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
@@ -180,32 +186,35 @@ def query():
        
        # 从用户信息表获取部门信息
        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 '未知部门'
+            # 可以复用同一个连接，不需要创建新连接
+            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}")
-            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(
+            # 继续使用同一个连接
+            # 检查是否是从"增加"按钮提交的请求
+            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, len(results), query_ip)
+                (department, query_date, plate_number if plate_number else None, phone if phone else None, results_count, query_ip)
            )
-            conn_history.commit()
+            conn.commit()
        except Exception as e:
            print(f"记录查询历史时出错: {e}")
        finally:
-            if 'conn_history' in locals():
-                conn_history.close()
+            close_db_connection(conn)
        
        # 处理查询结果
        car_owners = []
@@ -241,68 +250,97 @@ def query_history_detail(history_id):
    if 'username' not in session:
        return jsonify({'error': '未登录'}), 401
    
-    conn = sqlite3.connect('car_info.db')
+    conn = get_db_connection()
    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})
+    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 = sqlite3.connect('car_info.db')
+    conn = get_db_connection()
    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
+    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__':
-    init_db()
-    app.run(debug=True)
+    # 注册信号处理程序
+    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()