#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 将 compressed_group_time_issues.json 数据导入到 MySQL 数据库 """ import json import mysql.connector from mysql.connector import Error # MySQL 连接配置 DB_CONFIG = { 'host': '127.0.0.1', 'port': 3306, 'user': 'root', 'password': 'root123456', 'database': 'testdb', 'charset': 'utf8mb4' } # JSON 文件路径 JSON_FILE_PATH = 'server/jsons/compressed_group_time_issues.json' # 建表 SQL CREATE_TABLE_SQL = """ CREATE TABLE IF NOT EXISTS compressed_group_time_issues ( id INT AUTO_INCREMENT PRIMARY KEY COMMENT '主键', row_index INT COMMENT '行索引', -- Group1 字段 group1_latest_extracted_mtime DOUBLE COMMENT 'group1 最新提取文件的 mtime', group1_title VARCHAR(1000) COMMENT 'group1 标题', group1_url VARCHAR(1000) COMMENT 'group1 URL', group1_time VARCHAR(100) COMMENT 'group1 时间', group1_file_name VARCHAR(1000) COMMENT 'group1 第一个文件名', group1_file_mtime DOUBLE COMMENT 'group1 第一个文件 mtime', group1_file_datetime VARCHAR(50) COMMENT 'group1 第一个文件 datetime', -- Group2 字段 group2_latest_extracted_mtime DOUBLE COMMENT 'group2 最新提取文件的 mtime', group2_title VARCHAR(1000) COMMENT 'group2 标题', group2_url VARCHAR(1000) COMMENT 'group2 URL', group2_time VARCHAR(100) COMMENT 'group2 时间', group2_file_name VARCHAR(1000) COMMENT 'group2 第一个文件名', group2_file_mtime DOUBLE COMMENT 'group2 第一个文件 mtime', group2_file_datetime VARCHAR(50) COMMENT 'group2 第一个文件 datetime', -- 来源信息 source_folder VARCHAR(500) COMMENT '来源文件夹', word_doc VARCHAR(500) COMMENT '关联的 word 文档', -- 索引 INDEX idx_row_index (row_index), INDEX idx_source_folder (source_folder) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='压缩组时间问题数据表'; """ def create_connection(): """创建数据库连接""" try: connection = mysql.connector.connect(**DB_CONFIG) if connection.is_connected(): print(f"成功连接到 MySQL 数据库") print(f"数据库版本：{connection.get_server_info()}") return connection except Error as e: print(f"连接 MySQL 失败：{e}") return None def create_table(connection): """创建表""" try: cursor = connection.cursor() cursor.execute(CREATE_TABLE_SQL) connection.commit() print("表 'compressed_group_time_issues' 创建成功!") cursor.close() except Error as e: print(f"创建表失败：{e}") def load_json_data(file_path): """加载 JSON 数据""" try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) print(f"成功加载 JSON 文件，共 {len(data)} 条记录") return data except FileNotFoundError: print(f"文件不存在：{file_path}") return None except json.JSONDecodeError as e: print(f"JSON 解析错误：{e}") return None def flatten_record(record): """ 扁平化记录 """ # 获取 group1 的第一个文件 group1_files = record.get('group1', {}).get('files', []) group1_first_file = group1_files[0] if group1_files else {} # 获取 group2 的第一个文件 group2_files = record.get('group2', {}).get('files', []) group2_first_file = group2_files[0] if group2_files else {} return { 'row_index': record.get('row_index'), # Group1 字段 'group1_latest_extracted_mtime': record.get('group1_latest_extracted_mtime'), 'group1_title': record.get('group1', {}).get('title'), 'group1_url': record.get('group1', {}).get('url'), 'group1_time': record.get('group1', {}).get('time'), 'group1_file_name': group1_first_file.get('name'), 'group1_file_mtime': group1_first_file.get('mtime'), 'group1_file_datetime': group1_first_file.get('datetime'), # Group2 字段 'group2_latest_extracted_mtime': record.get('group2_latest_extracted_mtime'), 'group2_title': record.get('group2', {}).get('title'), 'group2_url': record.get('group2', {}).get('url'), 'group2_time': record.get('group2', {}).get('time'), 'group2_file_name': group2_first_file.get('name'), 'group2_file_mtime': group2_first_file.get('mtime'), 'group2_file_datetime': group2_first_file.get('datetime'), # 来源信息 'source_folder': record.get('source_folder'), 'word_doc': record.get('word_doc') } def insert_data(connection, data): """插入数据""" insert_sql = """ INSERT INTO compressed_group_time_issues ( row_index, group1_latest_extracted_mtime, group1_title, group1_url, group1_time, group1_file_name, group1_file_mtime, group1_file_datetime, group2_latest_extracted_mtime, group2_title, group2_url, group2_time, group2_file_name, group2_file_mtime, group2_file_datetime, source_folder, word_doc ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) """ try: cursor = connection.cursor() inserted_count = 0 for record in data: flattened = flatten_record(record) values = ( flattened['row_index'], flattened['group1_latest_extracted_mtime'], flattened['group1_title'], flattened['group1_url'], flattened['group1_time'], flattened['group1_file_name'], flattened['group1_file_mtime'], flattened['group1_file_datetime'], flattened['group2_latest_extracted_mtime'], flattened['group2_title'], flattened['group2_url'], flattened['group2_time'], flattened['group2_file_name'], flattened['group2_file_mtime'], flattened['group2_file_datetime'], flattened['source_folder'], flattened['word_doc'] ) cursor.execute(insert_sql, values) inserted_count += 1 if inserted_count % 50 == 0: print(f"已插入 {inserted_count} 条记录...") connection.commit() print(f"成功插入 {inserted_count} 条记录!") cursor.close() return inserted_count except Error as e: print(f"插入数据失败：{e}") connection.rollback() return 0 def verify_data(connection): """验证导入的数据""" try: cursor = connection.cursor() # 查询总记录数 cursor.execute("SELECT COUNT(*) FROM compressed_group_time_issues") count = cursor.fetchone()[0] print(f"\n验证：表中总记录数 = {count}") # 查询前 2 条记录作为示例 cursor.execute(""" SELECT row_index, group1_title, group2_title, source_folder FROM compressed_group_time_issues LIMIT 2 """) rows = cursor.fetchall() print("\n示例数据:") for row in rows: print(f" row_index={row[0]}") print(f" group1_title={row[1][:50]}...") print(f" group2_title={row[2][:50]}...") print(f" source_folder={row[3]}") print("-" * 50) cursor.close() except Error as e: print(f"验证数据失败：{e}") def main(): """主函数""" print("=" * 60) print("开始导入 compressed_group_time_issues.json 到 MySQL") print("=" * 60) # 1. 创建数据库连接 connection = create_connection() if not connection: return # 2. 创建表 create_table(connection) # 3. 加载 JSON 数据 data = load_json_data(JSON_FILE_PATH) if not data: connection.close() return # 4. 插入数据 insert_data(connection, data) # 5. 验证数据 verify_data(connection) # 关闭连接 connection.close() print("\n数据库连接已关闭") print("=" * 60) print("导入完成!") print("=" * 60) if __name__ == '__main__': main()