diff --git a/add_query_history_id_to_car_owners.py b/add_query_history_id_to_car_owners.py deleted file mode 100644 index 01e1866..0000000 --- a/add_query_history_id_to_car_owners.py +++ /dev/null @@ -1,118 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -为car_owners表添加query_history_id字段，并设置为query_history表id字段的外键 -""" - -import sqlite3 -import logging - -# 配置日志 -handlers = [logging.FileHandler('db_update.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# 数据库文件路径 -db_path = 'car_info.db' - -def main(): - """ - 主函数：连接数据库并为car_owners表添加query_history_id字段 - """ - logger.info("开始为car_owners表添加query_history_id字段") - - try: - # 连接数据库 - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 检查car_owners表是否已经存在query_history_id字段 - cursor.execute("PRAGMA table_info(car_owners)") - columns = cursor.fetchall() - - has_query_history_id = any(column[1] == 'query_history_id' for column in columns) - - if has_query_history_id: - logger.info("car_owners表已存在query_history_id字段，无需添加") - else: - # 添加query_history_id字段（先不添加外键约束，确保字段可以添加成功） - logger.info("开始添加query_history_id字段") - cursor.execute("ALTER TABLE car_owners ADD COLUMN query_history_id INTEGER") - conn.commit() - logger.info("query_history_id字段添加成功") - - # 尝试添加外键约束 - try: - logger.info("开始添加外键约束") - cursor.execute(""" - ALTER TABLE car_owners - ADD CONSTRAINT fk_car_owners_query_history_id - FOREIGN KEY (query_history_id) REFERENCES query_history(id) - """) - conn.commit() - logger.info("外键约束添加成功") - except sqlite3.Error as e: - # 在SQLite中，某些版本不支持通过ALTER TABLE直接添加外键约束 - # 如果添加外键约束失败，记录警告信息并继续 - logger.warning(f"添加外键约束失败: {str(e)}") - logger.warning("将创建临时表并重新创建car_owners表来添加外键约束") - - # 创建临时表 - cursor.execute(""" - CREATE TABLE car_owners_temp ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - plate_number TEXT NOT NULL, - phone TEXT NOT NULL, - id_card TEXT, - name TEXT, - email TEXT, - address TEXT, - query_history_id INTEGER, - FOREIGN KEY (query_history_id) REFERENCES query_history(id) - ) - """) - - # 复制数据到临时表 - cursor.execute(""" - INSERT INTO car_owners_temp ( - id, plate_number, phone, id_card, - name, email, address - ) - SELECT - id, plate_number, phone, id_card, - name, email, address - FROM car_owners - """) - - # 删除原表 - cursor.execute("DROP TABLE car_owners") - - # 重命名临时表 - cursor.execute("ALTER TABLE car_owners_temp RENAME TO car_owners") - - # 提交更改 - conn.commit() - logger.info("通过重建表的方式成功添加外键约束") - - # 关闭数据库连接 - conn.close() - logger.info("数据库连接已关闭，操作完成") - - except sqlite3.Error as e: - logger.error(f"数据库操作失败: {str(e)}") - # 如果发生异常，尝试回滚 - if 'conn' in locals() and conn: - try: - conn.rollback() - logger.info("事务已回滚") - except sqlite3.Error as rollback_error: - logger.error(f"回滚失败: {str(rollback_error)}") - except Exception as e: - logger.error(f"程序执行出错: {str(e)}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/add_sqlite3_import.py b/add_sqlite3_import.py deleted file mode 100644 index 55ca2a6..0000000 --- a/add_sqlite3_import.py +++ /dev/null @@ -1,53 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -为api.py添加sqlite3导入 -""" - -import logging - -# 配置日志 -handlers = [logging.FileHandler('add_import_log.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -def main(): - """ - 主函数：为api.py添加sqlite3导入 - """ - logger.info("开始为api.py添加sqlite3导入") - - try: - # 读取当前api.py文件内容 - with open('api.py', 'r', encoding='utf-8') as f: - lines = f.readlines() - - # 检查是否已经导入了sqlite3 - has_sqlite3_import = False - for line in lines: - if line.strip() == 'import sqlite3': - has_sqlite3_import = True - break - - # 如果没有导入，添加导入语句 - if not has_sqlite3_import: - # 添加到文件开头 - lines.insert(0, 'import sqlite3\n') - - # 写入更新后的文件 - with open('api.py', 'w', encoding='utf-8') as f: - f.writelines(lines) - - logger.info("成功为api.py添加sqlite3导入") - else: - logger.info("api.py已经导入了sqlite3") - - except Exception as e: - logger.error(f"添加导入时发生错误: {str(e)}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/add_user_id_to_query_history.py b/add_user_id_to_query_history.py deleted file mode 100644 index b722209..0000000 --- a/add_user_id_to_query_history.py +++ /dev/null @@ -1,115 +0,0 @@ -#!/usr/bin/env python3 -# -*- coding: utf-8 -*- - -""" -为query_history表添加user_id字段，并设置为users表id字段的外键 -""" - -import sqlite3 -import logging - -# 配置日志 -handlers = [logging.FileHandler('db_update.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# 数据库文件路径 -db_path = 'car_info.db' - - -def main(): - """ - 主函数：连接数据库并为query_history表添加user_id字段 - """ - logger.info("开始为query_history表添加user_id字段") - - try: - # 连接数据库 - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 检查query_history表是否已经存在user_id字段 - cursor.execute("PRAGMA table_info(query_history)") - columns = cursor.fetchall() - - has_user_id = any(column[1] == 'user_id' for column in columns) - - if has_user_id: - logger.info("query_history表已存在user_id字段，无需添加") - else: - # 添加user_id字段（先不添加外键约束，确保字段可以添加成功） - logger.info("开始添加user_id字段") - cursor.execute("ALTER TABLE query_history ADD COLUMN user_id INTEGER") - conn.commit() - logger.info("user_id字段添加成功") - - # 尝试添加外键约束 - try: - logger.info("开始添加外键约束") - cursor.execute(""" - ALTER TABLE query_history - ADD CONSTRAINT fk_query_history_user_id - FOREIGN KEY (user_id) REFERENCES users(id) - """) - conn.commit() - logger.info("外键约束添加成功") - except sqlite3.Error as e: - # 在SQLite中，某些版本不支持通过ALTER TABLE直接添加外键约束 - # 如果添加外键约束失败，记录警告信息并继续 - logger.warning(f"添加外键约束失败: {str(e)}") - logger.warning("将创建临时表并重新创建query_history表来添加外键约束") - - # 创建临时表 - cursor.execute(""" - CREATE TABLE query_history_temp ( - 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, - user_id INTEGER, - FOREIGN KEY (user_id) REFERENCES users(id) - ) - """) - - # 复制数据到临时表 - cursor.execute(""" - INSERT INTO query_history_temp ( - id, department, query_date, plate_number, - phone, results_count, query_ip - ) - SELECT - id, department, query_date, plate_number, - phone, results_count, query_ip - FROM query_history - """) - - # 删除原表 - cursor.execute("DROP TABLE query_history") - - # 重命名临时表 - cursor.execute("ALTER TABLE query_history_temp RENAME TO query_history") - - # 提交更改 - conn.commit() - logger.info("通过重新创建表的方式成功添加外键约束") - - # 关闭连接 - conn.close() - logger.info("操作完成") - - except Exception as e: - logger.error(f"操作过程中发生错误: {str(e)}") - if 'conn' in locals() and conn: - conn.rollback() - conn.close() - - -if __name__ == '__main__': - main() \ No newline at end of file diff --git a/app.py b/app.py index ebd60ed..a51aab6 100644 --- a/app.py +++ b/app.py @@ -83,7 +83,9 @@ def query(): phone = request.form.get('phone', '').strip() if not plate_number and not phone: - return render_template('query.html', error='请至少输入车牌号或手机号', recent_queries=formatted_recent_queries) + # 存储错误信息到session + session['error_message'] = '请至少输入车牌号或手机号' + return redirect(url_for('query')) conn = get_db_connection() cursor = conn.cursor() @@ -156,40 +158,57 @@ def query(): 'results_count': len(car_owners) } - # 使用辅助函数获取更新后的最近查询历史 - formatted_recent_queries = get_recent_queries() + # 将查询结果和历史信息存储到session中 + session['query_results'] = { + 'car_owners': car_owners, + 'query_history': query_history + } - return render_template('query.html', query_history=query_history, car_owners=car_owners, recent_queries=formatted_recent_queries) + # 重定向到GET请求，避免刷新页面时的重复提交提示 + return redirect(url_for('query')) - return render_template('query.html', recent_queries=formatted_recent_queries) + # 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/') 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: - # 获取查询历史记录 - 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,)) - + # 直接根据query_history_id从car_owners表查询对应记录 + cursor.execute("SELECT * FROM car_owners WHERE query_history_id = ?", (history_id,)) results = cursor.fetchall() # 格式化结果 @@ -201,8 +220,8 @@ def query_history_detail(history_id): 'phone': result[2], 'id_card': result[3], 'name': result[4], - 'email': result[5], - 'address': result[6] + 'email': result[5] if result[5] else '', + 'address': result[6] if result[6] else '' }) return jsonify({'car_owners': car_owners}) diff --git a/car_info.db b/car_info.db index 1dc54b9..c39b527 100644 Binary files a/car_info.db and b/car_info.db differ diff --git a/check_car_owners_constraints.py b/check_car_owners_constraints.py deleted file mode 100644 index e5f0a76..0000000 --- a/check_car_owners_constraints.py +++ /dev/null @@ -1,79 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -检查car_owners表的约束条件 -""" - -import sqlite3 -import logging - -# 配置日志 -handlers = [logging.FileHandler('constraints_check.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# 数据库文件路径 -db_path = 'car_info.db' - -def check_car_owners_constraints(): - """ - 检查car_owners表的约束条件 - """ - logger.info("开始检查car_owners表的约束条件") - - try: - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 检查表的唯一约束 - cursor.execute("PRAGMA index_list(car_owners)") - indexes = cursor.fetchall() - - logger.info("car_owners表的索引列表:") - for index in indexes: - index_id, name, unique, origin, partial = index - logger.info(f"索引ID: {index_id}, 名称: {name}, 唯一: {unique}, 来源: {origin}, 部分索引: {partial}") - - # 获取索引的列信息 - if name: - cursor.execute(f"PRAGMA index_info({name})") - index_columns = cursor.fetchall() - logger.info(f"索引 {name} 的列信息:") - for col in index_columns: - col_id, seq_no, col_name = col - logger.info(f"列ID: {col_id}, 序列号: {seq_no}, 列名: {col_name}") - - # 获取表的创建语句（如果可能） - cursor.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name='car_owners'") - create_sql = cursor.fetchone() - - if create_sql: - logger.info(f"car_owners表的创建语句: {create_sql[0]}") - else: - logger.warning("无法获取car_owners表的创建语句") - - # 关闭数据库连接 - conn.close() - - except sqlite3.Error as e: - logger.error(f"数据库操作失败: {str(e)}") - except Exception as e: - logger.error(f"程序执行出错: {str(e)}") - -def main(): - """ - 主函数 - """ - logger.info("开始检查数据库约束") - - # 检查car_owners表的约束条件 - check_car_owners_constraints() - - logger.info("数据库约束检查完成") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/check_car_owners_table.py b/check_car_owners_table.py deleted file mode 100644 index 3350a9d..0000000 --- a/check_car_owners_table.py +++ /dev/null @@ -1,76 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -检查car_owners表的结构，确认query_history_id字段是否正确添加 -""" - -import sqlite3 -import logging - -# 配置日志 -handlers = [logging.FileHandler('db_check.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# 数据库文件路径 -db_path = 'car_info.db' - -def main(): - """ - 主函数：连接数据库并检查car_owners表的结构 - """ - logger.info("开始检查car_owners表的结构") - - try: - # 连接数据库 - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 检查car_owners表的结构 - cursor.execute("PRAGMA table_info(car_owners)") - columns = cursor.fetchall() - - logger.info("car_owners表的字段信息:") - for column in columns: - logger.info(f"字段ID: {column[0]}, 字段名: {column[1]}, 数据类型: {column[2]}, 非空约束: {column[3]}, 默认值: {column[4]}, 主键: {column[5]}") - - # 检查外键约束 - cursor.execute("PRAGMA foreign_key_list(car_owners)") - foreign_keys = cursor.fetchall() - - logger.info("car_owners表的外键约束:") - for fk in foreign_keys: - logger.info(f"ID: {fk[0]}, 列名: {fk[3]}, 引用表: {fk[2]}, 引用列: {fk[4]}") - - # 检查query_history_id字段是否存在 - has_query_history_id = any(column[1] == 'query_history_id' for column in columns) - - if has_query_history_id: - logger.info("car_owners表已成功添加query_history_id字段") - else: - logger.warning("car_owners表中不存在query_history_id字段") - - # 检查query_history表是否存在 - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='query_history'") - query_history_exists = cursor.fetchone() is not None - - if query_history_exists: - logger.info("query_history表存在") - else: - logger.error("query_history表不存在，外键约束无法建立") - - # 关闭数据库连接 - conn.close() - logger.info("数据库连接已关闭，检查完成") - - except sqlite3.Error as e: - logger.error(f"数据库操作失败: {str(e)}") - except Exception as e: - logger.error(f"程序执行出错: {str(e)}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/check_db.py b/check_db.py deleted file mode 100644 index 1dadf0d..0000000 --- a/check_db.py +++ /dev/null @@ -1,55 +0,0 @@ -# 导入数据库工具 -from db_utils import get_db_connection, close_db_connection - -# 检查数据库连接是否正常 -def check_database_connection(): - try: - # 连接数据库 - conn = get_db_connection() - cursor = conn.cursor() - - # 执行简单的SQL查询来测试连接 - cursor.execute("SELECT sqlite_version()") - version = cursor.fetchone()[0] - - # 检查是否存在必要的表 - necessary_tables = ['users', 'query_records'] - tables_info = [] - - for table in necessary_tables: - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)) - exists = cursor.fetchone() is not None - tables_info.append({"name": table, "exists": exists}) - - # 关闭数据库连接 - close_db_connection(conn) - - # 构造返回结果 - result = { - "success": True, - "version": version, - "tables": tables_info - } - - return result - except Exception as e: - # 确保在异常情况下也关闭连接 - if 'conn' in locals() and conn: - close_db_connection(conn) - - return { - "success": False, - "error": str(e) - } - -# 如果直接运行此脚本，则执行检查操作 -if __name__ == "__main__": - result = check_database_connection() - print(f"数据库连接状态: {'成功' if result['success'] else '失败'}") - if result['success']: - print(f"SQLite版本: {result['version']}") - print("表状态:") - for table in result['tables']: - print(f" - {table['name']}: {'存在' if table['exists'] else '不存在'}") - else: - print(f"错误信息: {result['error']}") \ No newline at end of file diff --git a/check_table_structure.py b/check_table_structure.py index 95b95b9..c5977aa 100644 --- a/check_table_structure.py +++ b/check_table_structure.py @@ -1,85 +1,27 @@ -# -*- coding: utf-8 -*- - -""" -检查数据库表的实际结构 -""" - import sqlite3 -import logging -# 配置日志 -handlers = [logging.FileHandler('table_structure.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# 数据库文件路径 -db_path = 'car_info.db' - -def check_table_structure(table_name): - """ - 检查指定表的结构 - """ - logger.info(f"开始检查表 {table_name} 的结构") +# 连接到数据库 +try: + conn = sqlite3.connect('car_info.db') + cursor = conn.cursor() - try: - conn = sqlite3.connect(db_path) - cursor = conn.cursor() - - # 检查表是否存在 - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,)) - table_exists = cursor.fetchone() is not None - - if not table_exists: - logger.error(f"表 {table_name} 不存在") - return - - # 获取表的字段信息 - cursor.execute(f"PRAGMA table_info({table_name})") - columns = cursor.fetchall() - - logger.info(f"表 {table_name} 的字段信息:") - for column in columns: - logger.info(f"字段ID: {column[0]}, 字段名: {column[1]}, 数据类型: {column[2]}, 非空约束: {column[3]}, 默认值: {column[4]}, 主键: {column[5]}") - - # 获取表中的记录数量 - cursor.execute(f"SELECT COUNT(*) FROM {table_name}") - count = cursor.fetchone()[0] - logger.info(f"表 {table_name} 中的记录数量: {count}") - - # 如果有记录，显示前5条记录的内容 - if count > 0: - cursor.execute(f"SELECT * FROM {table_name} LIMIT 5") - rows = cursor.fetchall() - - logger.info(f"表 {table_name} 的前5条记录:") - for row in rows: - logger.info(f"记录: {row}") - - # 关闭数据库连接 - conn.close() - - except sqlite3.Error as e: - logger.error(f"数据库操作失败: {str(e)}") - except Exception as e: - logger.error(f"程序执行出错: {str(e)}") - -def main(): - """ - 主函数 - """ - logger.info("开始检查数据库表结构") + # 获取car_owners表的结构 + cursor.execute('PRAGMA table_info(car_owners)') + rows = cursor.fetchall() - # 检查query_history表的结构 - check_table_structure('query_history') + print('car_owners表结构:') + for row in rows: + print(row) - # 检查car_owners表的结构 - check_table_structure('car_owners') + # 也检查query_history表的结构，了解两个表的关系 + print('\nquery_history表结构:') + cursor.execute('PRAGMA table_info(query_history)') + rows = cursor.fetchall() + for row in rows: + print(row) - logger.info("数据库表结构检查完成") - -if __name__ == "__main__": - main() \ No newline at end of file +except Exception as e: + print(f'查询表结构时出错: {e}') +finally: + if conn: + conn.close() \ No newline at end of file diff --git a/final_verification.py b/final_verification.py deleted file mode 100644 index 58ddc59..0000000 --- a/final_verification.py +++ /dev/null @@ -1,193 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -最终验证脚本：确保所有问题都已修复 -1. 创建测试数据 -2. 测试update_query_result接口 -3. 验证数据是否正确保存 -""" - -import sqlite3 -import requests -import json -import logging - -# 配置日志 -handlers = [logging.FileHandler('final_verification.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# API地址 -BASE_URL = 'http://localhost:99' -API_ENDPOINT = f'{BASE_URL}/api/update-query-result' -DB_PATH = 'car_info.db' - -def create_test_data(): - """ - 在数据库中创建测试数据 - """ - logger.info("开始创建测试数据") - - try: - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() - - # 创建一个测试查询历史记录 - cursor.execute(""" - INSERT INTO query_history (department, query_date, plate_number, phone, results_count, query_ip) - VALUES (?, datetime('now'), ?, ?, ?, ?) - """, ('测试部门', '湘AB1234', '13800138000', 0, '127.0.0.1')) - - # 获取新插入的记录ID - query_id = cursor.lastrowid - - conn.commit() - conn.close() - - logger.info(f"成功创建测试数据，生成的查询ID: {query_id}") - return query_id - - except sqlite3.Error as e: - logger.error(f"创建测试数据时发生数据库错误: {str(e)}") - return None - except Exception as e: - logger.error(f"创建测试数据时发生错误: {str(e)}") - return None - -def test_update_query_result_with_real_data(query_id): - """ - 使用真实的测试数据测试update_query_result接口 - """ - logger.info(f"使用查询ID {query_id} 测试update_query_result接口") - - # 测试数据 - test_data = { - 'id': query_id, - 'results_count': 2, - 'cars_info': [ - { - 'plate_number': '湘AB1234', - 'phone': '13800138000', - 'name': '张三', - 'email': 'zhangsan@example.com', - 'address': '北京市海淀区' - }, - { - 'plate_number': '湘AB5678', - 'phone': '13800138001', - 'name': '李四', - 'email': 'lisi@example.com', - 'address': '上海市浦东新区' - } - ] - } - - try: - logger.info(f"发送测试请求，数据: {json.dumps(test_data, ensure_ascii=False)}") - - # 发送POST请求 - response = requests.post( - API_ENDPOINT, - json=test_data, - headers={'Content-Type': 'application/json'}, - timeout=30 - ) - - logger.info(f"请求完成，状态码: {response.status_code}") - - # 输出响应内容 - if response.status_code == 200: - try: - result = response.json() - logger.info(f"响应数据: {json.dumps(result, ensure_ascii=False)}") - return True, result - except json.JSONDecodeError: - logger.warning(f"响应不是有效的JSON格式: {response.text}") - return False, None - else: - logger.error(f"请求失败，状态码: {response.status_code}，响应内容: {response.text}") - return False, None - - except requests.exceptions.RequestException as e: - logger.error(f"请求异常: {str(e)}") - return False, None - -def verify_data_in_db(query_id, car_owner_ids): - """ - 验证数据是否正确保存到数据库 - """ - logger.info(f"验证查询ID {query_id} 的数据是否正确保存到数据库") - - try: - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() - - # 检查query_history表是否更新成功 - cursor.execute("SELECT results_count FROM query_history WHERE id = ?", (query_id,)) - result = cursor.fetchone() - - if result and result[0] == 2: - logger.info("query_history表更新成功") - else: - logger.error(f"query_history表更新失败，当前值: {result[0] if result else '记录不存在'}") - return False - - # 检查car_owners表是否保存成功 - for car_id in car_owner_ids: - cursor.execute("SELECT plate_number, query_history_id FROM car_owners WHERE id = ?", (car_id,)) - result = cursor.fetchone() - - if result and result[1] == query_id: - logger.info(f"car_owners表记录 {car_id} (车牌号: {result[0]}) 保存成功，并正确关联了查询历史") - else: - logger.error(f"car_owners表记录 {car_id} 保存失败或关联错误") - return False - - conn.close() - return True - - except sqlite3.Error as e: - logger.error(f"验证数据时发生数据库错误: {str(e)}") - return False - except Exception as e: - logger.error(f"验证数据时发生错误: {str(e)}") - return False - -def main(): - """ - 主函数 - """ - logger.info("开始最终验证") - - # 创建测试数据 - query_id = create_test_data() - if not query_id: - logger.error("无法创建测试数据，验证失败") - return - - # 测试update_query_result接口 - success, result = test_update_query_result_with_real_data(query_id) - if not success: - logger.error("update_query_result接口测试失败") - return - - # 验证数据是否正确保存 - if result and 'data' in result and 'car_owner_ids' in result['data']: - car_owner_ids = result['data']['car_owner_ids'] - verify_success = verify_data_in_db(query_id, car_owner_ids) - - if verify_success: - logger.info("所有验证通过！系统功能正常工作") - else: - logger.error("数据验证失败") - else: - logger.error("响应数据中不包含car_owner_ids字段") - - logger.info("最终验证完成") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/final_verification_v2.py b/final_verification_v2.py deleted file mode 100644 index 810edfe..0000000 --- a/final_verification_v2.py +++ /dev/null @@ -1,179 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -最终验证脚本（版本2）：适应实际的表结构 -1. 创建测试数据 -2. 测试update_query_result接口 -3. 验证数据是否正确保存 -""" - -import sqlite3 -import requests -import json -import logging -import datetime - -# 配置日志 -handlers = [logging.FileHandler('final_verification.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# API地址 -BASE_URL = 'http://localhost:99' -API_ENDPOINT = f'{BASE_URL}/api/update-query-result' -DB_PATH = 'car_info.db' - -def create_test_data(): - """ - 在数据库中创建测试数据 - """ - logger.info("开始创建测试数据") - - try: - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() - - # 创建一个测试查询历史记录（根据实际表结构） - current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') - cursor.execute(""" - INSERT INTO query_history (query_date, plate_number, phone, results_count, query_ip) - VALUES (?, ?, ?, ?, ?) - """, (current_time, '测试车牌', '13800138000', 0, '127.0.0.1')) - - # 获取新插入的记录ID - query_id = cursor.lastrowid - - conn.commit() - conn.close() - - logger.info(f"成功创建测试数据，生成的查询ID: {query_id}") - return query_id - - except sqlite3.Error as e: - logger.error(f"创建测试数据时发生数据库错误: {str(e)}") - return None - except Exception as e: - logger.error(f"创建测试数据时发生错误: {str(e)}") - return None - -def get_existing_query_id(): - """ - 获取数据库中已存在的查询历史记录ID - """ - logger.info("尝试获取已存在的查询历史记录ID") - - try: - conn = sqlite3.connect(DB_PATH) - cursor = conn.cursor() - - # 获取最近的一条记录 - cursor.execute("SELECT id FROM query_history ORDER BY query_date DESC LIMIT 1") - result = cursor.fetchone() - - conn.close() - - if result: - query_id = result[0] - logger.info(f"成功获取已存在的查询历史记录ID: {query_id}") - return query_id - else: - logger.warning("数据库中没有查询历史记录") - return None - - except sqlite3.Error as e: - logger.error(f"获取已存在记录时发生数据库错误: {str(e)}") - return None - except Exception as e: - logger.error(f"获取已存在记录时发生错误: {str(e)}") - return None - -def test_update_query_result_with_real_data(query_id): - """ - 使用真实的测试数据测试update_query_result接口 - """ - logger.info(f"使用查询ID {query_id} 测试update_query_result接口") - - # 测试数据 - test_data = { - 'id': query_id, - 'results_count': 2, - 'cars_info': [ - { - 'plate_number': '测试车牌1', - 'phone': '13800138001', - 'name': '测试姓名1', - 'email': 'test1@example.com', - 'address': '测试地址1' - }, - { - 'plate_number': '测试车牌2', - 'phone': '13800138002', - 'name': '测试姓名2', - 'email': 'test2@example.com', - 'address': '测试地址2' - } - ] - } - - try: - logger.info(f"发送测试请求，数据: {json.dumps(test_data, ensure_ascii=False)}") - - # 发送POST请求 - response = requests.post( - API_ENDPOINT, - json=test_data, - headers={'Content-Type': 'application/json'}, - timeout=30 - ) - - logger.info(f"请求完成，状态码: {response.status_code}") - - # 输出响应内容 - if response.status_code == 200: - try: - result = response.json() - logger.info(f"响应数据: {json.dumps(result, ensure_ascii=False)}") - return True, result - except json.JSONDecodeError: - logger.warning(f"响应不是有效的JSON格式: {response.text}") - return False, None - else: - logger.error(f"请求失败，状态码: {response.status_code}，响应内容: {response.text}") - return False, None - - except requests.exceptions.RequestException as e: - logger.error(f"请求异常: {str(e)}") - return False, None - -def main(): - """ - 主函数 - """ - logger.info("开始最终验证") - - # 首先尝试获取已存在的查询ID - query_id = get_existing_query_id() - - # 如果没有已存在的查询ID，尝试创建新的 - if not query_id: - query_id = create_test_data() - - if not query_id: - logger.error("无法获取或创建测试数据，验证失败") - return - - # 测试update_query_result接口 - success, result = test_update_query_result_with_real_data(query_id) - if not success: - logger.error("update_query_result接口测试失败") - else: - logger.info("update_query_result接口测试成功！") - - logger.info("最终验证完成") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/fix_update_query_result.py b/fix_update_query_result.py deleted file mode 100644 index d8fb7fe..0000000 --- a/fix_update_query_result.py +++ /dev/null @@ -1,156 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -修复update_query_result函数中的问题 -主要解决以下问题： -1. 受影响行数的获取逻辑不准确 -2. 添加更详细的错误处理 -3. 确保数据库事务的正确性 -""" - -import logging - -# 配置日志 -handlers = [logging.FileHandler('fix_log.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -def main(): - """ - 主函数：执行修复操作 - """ - logger.info("开始修复update_query_result函数") - - try: - # 读取当前api.py文件内容 - with open('api.py', 'r', encoding='utf-8') as f: - content = f.read() - - # 定义需要替换的代码块和新代码块 - old_code = """ # 获取受影响的行数 - affected_rows = cursor.rowcount - - # 关闭数据库连接 - close_db_connection(conn) - - # 检查是否有记录被更新 - if affected_rows == 0: - return jsonify({ - 'success': False, - 'error': '未找到id为 {} 的记录'.format(query_id), - 'data': None - }), 404""" - - new_code = """ # 检查query_history表是否有记录被更新 - # 重新查询来确认更新是否成功 - cursor.execute("SELECT 1 FROM query_history WHERE id = ?", (query_id,)) - query_exists = cursor.fetchone() is not None - - # 关闭数据库连接 - close_db_connection(conn) - - # 检查是否存在该记录 - if not query_exists: - return jsonify({ - 'success': False, - 'error': '未找到id为 {} 的记录'.format(query_id), - 'data': None - }), 404""" - - # 执行替换 - if old_code in content: - new_content = content.replace(old_code, new_code) - - # 写入修复后的代码 - with open('api.py', 'w', encoding='utf-8') as f: - f.write(new_content) - - logger.info("成功修复update_query_result函数，解决了受影响行数的获取问题") - - # 添加更详细的异常处理 - add_detailed_exception_handling() - - else: - logger.warning("未找到需要修复的代码块，可能已经被修改过") - - except Exception as e: - logger.error(f"修复过程中发生错误: {str(e)}") - -def add_detailed_exception_handling(): - """ - 为update_query_result函数添加更详细的异常处理 - """ - try: - # 读取当前api.py文件内容 - with open('api.py', 'r', encoding='utf-8') as f: - content = f.read() - - # 定义需要替换的代码块和新代码块 - old_except_block = """ except Exception as e: - # 确保关闭数据库连接 - if 'conn' in locals(): - close_db_connection(conn) - return jsonify({ - 'success': False, - 'error': str(e), - 'data': None - }), 500""" - - new_except_block = """ except sqlite3.IntegrityError as e: - # 确保关闭数据库连接 - if 'conn' in locals(): - close_db_connection(conn) - logger.error(f"数据库完整性错误: {str(e)}") - return jsonify({ - 'success': False, - 'error': f'数据库完整性错误: {str(e)}', - 'data': None - }), 500 - except sqlite3.Error as e: - # 确保关闭数据库连接 - if 'conn' in locals(): - close_db_connection(conn) - logger.error(f"数据库错误: {str(e)}") - return jsonify({ - 'success': False, - 'error': f'数据库错误: {str(e)}', - 'data': None - }), 500 - except Exception as e: - # 确保关闭数据库连接 - if 'conn' in locals(): - close_db_connection(conn) - logger.error(f"服务器错误: {str(e)}") - return jsonify({ - 'success': False, - 'error': f'服务器错误: {str(e)}', - 'data': None - }), 500""" - - # 执行替换 - if old_except_block in content: - new_content = content.replace(old_except_block, new_except_block) - - # 同时添加sqlite3导入 - if 'import sqlite3' not in content: - import_line = "import sqlite3\n" - new_content = import_line + new_content - logger.info("成功添加sqlite3导入") - - # 写入修复后的代码 - with open('api.py', 'w', encoding='utf-8') as f: - f.write(new_content) - - logger.info("成功为update_query_result函数添加了详细的异常处理") - else: - logger.warning("未找到需要修复的异常处理代码块") - - except Exception as e: - logger.error(f"添加详细异常处理时发生错误: {str(e)}") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/query_processor.py b/query_processor.py index c14213b..6b7ef44 100644 --- a/query_processor.py +++ b/query_processor.py @@ -5,7 +5,7 @@ import requests from ewcc2 import Sinopec_ewcc # 配置日志 -handlers = [logging.FileHandler('query_processor.log'), logging.StreamHandler()] +handlers = [logging.FileHandler('log\query_processor.log'), logging.StreamHandler()] logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', diff --git a/remove_department_from_query_history.py b/remove_department_from_query_history.py deleted file mode 100644 index ba29b39..0000000 --- a/remove_department_from_query_history.py +++ /dev/null @@ -1,69 +0,0 @@ -import sqlite3 -from db_utils import get_db_connection, close_db_connection - - -def remove_department_from_query_history(): - """ - 从query_history表中删除department字段 - SQLite不支持直接删除列，采用以下步骤： - 1. 创建一个不包含department字段的新表 - 2. 将原始表中的数据（除department外）复制到新表 - 3. 删除原始表 - 4. 重命名新表为原始表名 - 5. 重新创建索引和外键约束 - """ - conn = None - try: - # 获取数据库连接 - conn = get_db_connection() - cursor = conn.cursor() - - print("开始从query_history表中删除department字段...") - - # 1. 创建临时表，不包含department字段 - print("1. 创建临时表query_history_new...") - cursor.execute(''' - CREATE TABLE IF NOT EXISTS query_history_new ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - query_date TEXT NOT NULL, - plate_number TEXT, - phone TEXT, - results_count INTEGER NOT NULL, - query_ip TEXT NOT NULL, - user_id INTEGER, - FOREIGN KEY (user_id) REFERENCES users (id) - ) - ''') - - # 2. 复制数据到新表，排除department字段 - print("2. 复制数据到新表，排除department字段...") - cursor.execute(''' - INSERT INTO query_history_new (id, query_date, plate_number, phone, results_count, query_ip, user_id) - SELECT id, query_date, plate_number, phone, results_count, query_ip, user_id - FROM query_history - ''') - - # 3. 删除原表 - print("3. 删除原始表query_history...") - cursor.execute('DROP TABLE IF EXISTS query_history') - - # 4. 重命名新表为原表名 - print("4. 重命名新表为query_history...") - cursor.execute('ALTER TABLE query_history_new RENAME TO query_history') - - # 5. 提交事务 - conn.commit() - print("操作完成：department字段已成功从query_history表中删除！") - - except sqlite3.Error as e: - print(f"数据库操作出错：{e}") - if conn: - conn.rollback() - raise - finally: - if conn: - close_db_connection(conn) - - -if __name__ == "__main__": - remove_department_from_query_history() \ No newline at end of file diff --git a/templates/parameter_config.html b/templates/parameter_config.html index d6b1737..8d66d12 100644 --- a/templates/parameter_config.html +++ b/templates/parameter_config.html @@ -79,7 +79,7 @@
- +
@@ -140,7 +140,7 @@ param1: document.getElementById('param1').value.trim(), param2: document.getElementById('param2').value.trim(), param3: document.getElementById('param3').value.trim(), - param4: document.getElementById('param4').value.trim(), + // 参数4为只读，不提交到服务器 param5: document.getElementById('param5').value.trim() }; diff --git a/test_update_query_result.py b/test_update_query_result.py deleted file mode 100644 index 3192afa..0000000 --- a/test_update_query_result.py +++ /dev/null @@ -1,97 +0,0 @@ -# -*- coding: utf-8 -*- - -""" -测试update_query_result接口的脚本 -用于模拟发送请求并获取详细的错误信息 -""" - -import requests -import json -import logging -import time - -# 配置日志 -handlers = [logging.FileHandler('test_api.log'), logging.StreamHandler()] -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=handlers -) -logger = logging.getLogger(__name__) - -# API地址 -BASE_URL = 'http://localhost:99' -API_ENDPOINT = f'{BASE_URL}/api/update-query-result' - -def test_update_query_result(): - """ - 测试update_query_result接口 - """ - logger.info(f"开始测试API接口: {API_ENDPOINT}") - - # 测试数据 - test_data = { - 'id': 1, # 假设query_history表中有id为1的记录 - 'results_count': 1, - 'cars_info': [ - { - 'plate_number': '测试车牌1', - 'phone': '13800138001', - 'name': '测试姓名1', - 'email': 'test1@example.com', - 'address': '测试地址1' - } - ] - } - - try: - logger.info(f"发送测试请求，数据: {json.dumps(test_data, ensure_ascii=False)}") - - # 发送POST请求 - start_time = time.time() - response = requests.post( - API_ENDPOINT, - json=test_data, - headers={'Content-Type': 'application/json'}, - timeout=30 - ) - end_time = time.time() - - logger.info(f"请求完成，耗时: {end_time - start_time:.2f}秒，状态码: {response.status_code}") - - # 输出响应内容 - if response.status_code == 200: - try: - result = response.json() - logger.info(f"响应数据: {json.dumps(result, ensure_ascii=False)}") - except json.JSONDecodeError: - logger.warning(f"响应不是有效的JSON格式: {response.text}") - else: - logger.error(f"请求失败，状态码: {response.status_code}，响应内容: {response.text}") - - return response.status_code - - except requests.exceptions.RequestException as e: - logger.error(f"请求异常: {str(e)}") - return None - -def main(): - """ - 主函数 - """ - logger.info("测试脚本启动") - - # 执行测试 - status_code = test_update_query_result() - - if status_code == 200: - logger.info("测试成功") - elif status_code is not None: - logger.error(f"测试失败，状态码: {status_code}") - else: - logger.error("测试失败，请求异常") - - logger.info("测试脚本结束") - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/update_database.py b/update_database.py deleted file mode 100644 index 227fb2b..0000000 --- a/update_database.py +++ /dev/null @@ -1,133 +0,0 @@ -import datetime - -# 导入数据库工具 -from db_utils import get_db_connection, close_db_connection - -# 连接到数据库并确保users表结构正确 -def update_users_table(): - try: - # 连接数据库 - conn = get_db_connection() - 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用户信息") - - # 提交更改 - try: - conn.commit() - except Exception as e: - print(f"提交更改时出错: {e}") - - # 再次验证表结构 - 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]}") - - # 关闭连接 - close_db_connection(conn) - print("\n数据库更新完成") - - except Exception as e: - print(f"更新数据库时出错: {str(e)}") - # 确保在异常情况下也关闭连接 - if 'conn' in locals() and conn: - close_db_connection(conn) - -# 更新数据库中的数据 -def update_database(): - try: - # 连接数据库 - conn = get_db_connection() - cursor = conn.cursor() - - # 检查是否存在测试数据表 - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='test_data'") - if not cursor.fetchone(): - # 如果不存在，创建测试数据表 - cursor.execute(''' - CREATE TABLE test_data ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - name TEXT NOT NULL, - value TEXT, - create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, - update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP - ) - ''') - conn.commit() - print("测试数据表创建成功") - - # 检查是否有测试数据，如果没有则插入 - cursor.execute("SELECT COUNT(*) FROM test_data") - count = cursor.fetchone()[0] - - if count == 0: - # 插入测试数据 - test_data = [ - ('测试数据1', 'value1', datetime.datetime.now()), - ('测试数据2', 'value2', datetime.datetime.now()), - ('测试数据3', 'value3', datetime.datetime.now()) - ] - - for data in test_data: - cursor.execute( - "INSERT INTO test_data (name, value, create_time, update_time) VALUES (?, ?, ?, ?)", - data - ) - conn.commit() - print("测试数据插入成功") - - # 关闭数据库连接 - close_db_connection(conn) - return True, "数据库更新成功" - - except Exception as e: - # 确保在异常情况下也关闭连接 - if 'conn' in locals() and conn: - close_db_connection(conn) - return False, f"数据库更新失败：{str(e)}" - -# 如果直接运行此脚本，则执行更新操作 -if __name__ == "__main__": - success, message = update_database() - print(message) \ No newline at end of file diff --git a/verify_database.py b/verify_database.py deleted file mode 100644 index 203a6e2..0000000 --- a/verify_database.py +++ /dev/null @@ -1,69 +0,0 @@ -# 导入数据库工具 -from db_utils import get_db_connection, close_db_connection - -# 验证数据库结构是否完整 -def verify_database_structure(): - try: - # 连接数据库 - conn = get_db_connection() - cursor = conn.cursor() - - # 需要检查的表 - tables_to_check = ['users', 'query_records'] - verification_results = [] - - for table in tables_to_check: - # 检查表是否存在 - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)) - table_exists = cursor.fetchone() is not None - - if table_exists: - # 检查表的列结构 - cursor.execute(f"PRAGMA table_info({table})") - columns = cursor.fetchall() - column_names = [col[1] for col in columns] - - verification_results.append({ - "table": table, - "exists": True, - "column_count": len(columns), - "columns": column_names - }) - else: - verification_results.append({ - "table": table, - "exists": False, - "column_count": 0, - "columns": [] - }) - - # 关闭数据库连接 - close_db_connection(conn) - - # 构造返回结果 - all_tables_exist = all(result['exists'] for result in verification_results) - - return { - "success": all_tables_exist, - "results": verification_results - } - except Exception as e: - # 确保在异常情况下也关闭连接 - if 'conn' in locals() and conn: - close_db_connection(conn) - - return { - "success": False, - "error": str(e) - } - -# 如果直接运行此脚本，则执行验证操作 -if __name__ == "__main__": - result = verify_database_structure() - print(f"数据库结构验证: {'通过' if result['success'] else '失败'}") - if result['success']: - print("验证详情:") - for table_result in result['results']: - print(f" - 表 {table_result['table']}: 存在，包含 {table_result['column_count']} 个字段") - else: - print(f"错误信息: {result['error']}") \ No newline at end of file