smshine/hiddencode_project
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modified pppp

Browse Source
This commit is contained in:
Shuming Liu
2026-07-01 15:52:29 +08:00
parent 44a9c5a5d5
commit f5c0ecc31f
2 changed files with 464 additions and 299 deletions
Download Patch File Download Diff File Expand all files Collapse all files

604
server/import_json_to_mysql.py
View File

@@ -1,18 +1,26 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
将 compressed_group_time_issues.json 数据导入到 MySQL 数据库
将 jsons/compressed_group_time_issues.json 导入 MySQL。
只导入以下四类，并把同一 row 中二一教育和学科网的数据合并为同一条记录：
- single_files
- same_child_archive_time
- zxxk_child_earlier
- twole_child_earlier
"""
import argparse
import json
import mysql.connector
from mysql.connector import Error
import re
from datetime import datetime
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
import mysql.connector
from mysql.connector import Error
# MySQL 连接配置
DB_CONFIG = {
'host': '127.0.0.1',
'port': 3306,
@@ -22,316 +30,362 @@ DB_CONFIG = {
'charset': 'utf8mb4'
}
# JSON 文件路径
JSON_FILE_PATH = 'server/jsons/compressed_group_time_issues.json'
BASE_DIR = Path(__file__).resolve().parent
DEFAULT_JSON_FILE_PATH = BASE_DIR / "jsons" / "compressed_group_time_issues.json"
DEFAULT_TABLE_NAME = "compressed_group_time_records"
SELECTED_CATEGORIES = (
"single_files",
"same_child_archive_time",
"zxxk_child_earlier",
"twole_child_earlier",
)
# 建表 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',
CREATE_TABLE_TEMPLATE = """
CREATE TABLE IF NOT EXISTS `{table_name}` (
id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键',
category VARCHAR(64) NOT NULL COMMENT '分类',
source_folder VARCHAR(512) NOT NULL COMMENT '来源文件夹',
word_doc VARCHAR(512) NOT NULL COMMENT '关联 Word 文档',
row_index INT NOT NULL COMMENT 'Word 内容行索引',
-- 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',
earlier_site VARCHAR(64) COMMENT '更早的网站中文名',
earlier_site_key VARCHAR(32) COMMENT '更早的网站 key: twole/zxxk',
child_time_diff_seconds DOUBLE COMMENT '学科网子文件时间 - 二一教育子文件时间',
-- 来源信息
source_folder VARCHAR(500) COMMENT '来源文件夹',
word_doc VARCHAR(500) COMMENT '关联的 word 文档',
twole_title TEXT COMMENT '二一教育标题',
twole_url TEXT COMMENT '二一教育 URL',
twole_source_time VARCHAR(100) COMMENT 'Word 中二一教育时间',
twole_id VARCHAR(100) COMMENT '二一教育 URL ID',
twole_file_kind VARCHAR(32) COMMENT '二一教育文件类型',
twole_file_name VARCHAR(1024) COMMENT '二一教育文件名',
twole_file_mtime DOUBLE COMMENT '二一教育文件 mtime',
twole_file_datetime DATETIME COMMENT '二一教育文件时间',
twole_archive_mtime DOUBLE COMMENT '二一教育压缩包 mtime',
twole_archive_datetime DATETIME COMMENT '二一教育压缩包时间',
twole_latest_child_mtime DOUBLE COMMENT '二一教育压缩包内最新子文件 mtime',
twole_latest_child_datetime DATETIME COMMENT '二一教育压缩包内最新子文件时间',
twole_child_archive_time_relation VARCHAR(32) COMMENT '二一教育子文件与压缩包时间关系',
twole_child_archive_time_diff_seconds DOUBLE COMMENT '二一教育子文件与压缩包时间差',
-- 新增字段
21cn_son_file_datetime VARCHAR(50) COMMENT 'group1_latest_extracted_mtime 的日期时间格式',
xkw_son_file_datetime VARCHAR(50) COMMENT 'group2_latest_extracted_mtime 的日期时间格式',
21ID VARCHAR(100) COMMENT 'group1_url 中的数字部分',
xkwID VARCHAR(100) COMMENT 'group2_url 中的数字部分',
zxxk_title TEXT COMMENT '学科网标题',
zxxk_url TEXT COMMENT '学科网 URL',
zxxk_source_time VARCHAR(100) COMMENT 'Word 中学科网时间',
zxxk_id VARCHAR(100) COMMENT '学科网 URL ID',
zxxk_file_kind VARCHAR(32) COMMENT '学科网文件类型',
zxxk_file_name VARCHAR(1024) COMMENT '学科网文件名',
zxxk_file_mtime DOUBLE COMMENT '学科网文件 mtime',
zxxk_file_datetime DATETIME COMMENT '学科网文件时间',
zxxk_archive_mtime DOUBLE COMMENT '学科网压缩包 mtime',
zxxk_archive_datetime DATETIME COMMENT '学科网压缩包时间',
zxxk_latest_child_mtime DOUBLE COMMENT '学科网压缩包内最新子文件 mtime',
zxxk_latest_child_datetime DATETIME COMMENT '学科网压缩包内最新子文件时间',
zxxk_child_archive_time_relation VARCHAR(32) COMMENT '学科网子文件与压缩包时间关系',
zxxk_child_archive_time_diff_seconds DOUBLE COMMENT '学科网子文件与压缩包时间差',
-- 索引
INDEX idx_row_index (row_index),
INDEX idx_source_folder (source_folder)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='压缩组时间问题数据表';
twole_json JSON COMMENT '二一教育原始 JSON',
zxxk_json JSON COMMENT '学科网原始 JSON',
raw_json JSON NOT NULL COMMENT '合并前原始 JSON',
imported_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '导入时间',
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
UNIQUE KEY uniq_category_row (category, source_folder(191), word_doc(191), row_index),
KEY idx_category (category),
KEY idx_earlier_site_key (earlier_site_key),
KEY idx_source_row (source_folder(191), word_doc(191), row_index),
KEY idx_twole_id (twole_id),
KEY idx_zxxk_id (zxxk_id),
KEY idx_twole_latest_child_mtime (twole_latest_child_mtime),
KEY idx_zxxk_latest_child_mtime (zxxk_latest_child_mtime)
) 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}")
INSERT_COLUMNS = (
"category", "source_folder", "word_doc", "row_index",
"earlier_site", "earlier_site_key", "child_time_diff_seconds",
"twole_title", "twole_url", "twole_source_time", "twole_id", "twole_file_kind",
"twole_file_name", "twole_file_mtime", "twole_file_datetime",
"twole_archive_mtime", "twole_archive_datetime",
"twole_latest_child_mtime", "twole_latest_child_datetime",
"twole_child_archive_time_relation", "twole_child_archive_time_diff_seconds",
"zxxk_title", "zxxk_url", "zxxk_source_time", "zxxk_id", "zxxk_file_kind",
"zxxk_file_name", "zxxk_file_mtime", "zxxk_file_datetime",
"zxxk_archive_mtime", "zxxk_archive_datetime",
"zxxk_latest_child_mtime", "zxxk_latest_child_datetime",
"zxxk_child_archive_time_relation", "zxxk_child_archive_time_diff_seconds",
"twole_json", "zxxk_json", "raw_json",
)
def mysql_identifier(name: str) -> str:
if not re.match(r"^[A-Za-z0-9_]+$", name):
raise ValueError(f"非法表名：{name}")
return name
def epoch_to_datetime(value) -> Optional[str]:
if value in (None, ""):
return None
return datetime.fromtimestamp(float(value)).strftime("%Y-%m-%d %H:%M:%S")
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}")
def normalize_datetime(value) -> Optional[str]:
if not value:
return None
except json.JSONDecodeError as e:
print(f"JSON 解析错误：{e}")
if isinstance(value, str) and re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$", value):
return value
return None
def json_dumps(value) -> str:
return json.dumps(value or {}, ensure_ascii=False)
def extract_url_id(url: Optional[str], site: str) -> Optional[str]:
if not url:
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 {}
# 计算新字段
group1_mtime = record.get('group1_latest_extracted_mtime')
group1_datetime = datetime.fromtimestamp(group1_mtime).strftime('%Y-%m-%d %H:%M:%S') if group1_mtime else None
group2_mtime = record.get('group2_latest_extracted_mtime')
group2_datetime = datetime.fromtimestamp(group2_mtime).strftime('%Y-%m-%d %H:%M:%S') if group2_mtime else None
group1_url = record.get('group1', {}).get('url', '')
group1_id = None
if group1_url:
match = re.search(r'/(\d+)\.shtml$', group1_url)
if site == "twole":
match = re.search(r"/(\d+)\.shtml(?:[?#].*)?$", url)
if match:
group1_id = match.group(1)
group2_url = record.get('group2', {}).get('url', '')
group2_id = None
if group2_url:
match = re.search(r'/(\d+)\.html$', group2_url)
return match.group(1)
if site == "zxxk":
match = re.search(r"/(?:soft/)?(\d+)\.html(?:[?#].*)?$", url)
if match:
group2_id = match.group(1)
return match.group(1)
numbers = re.findall(r"\d+", url)
return numbers[-1] if numbers else None
def first_file_from_group(group: Dict) -> Dict:
files = group.get("files") or []
return files[0] if files else {}
def build_side_from_individual(record: Dict) -> Dict:
file_info = record.get("file") or {}
site = record.get("site") or ""
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'),
# 新增字段
'21cn_son_file_datetime': group1_datetime,
'xkw_son_file_datetime': group2_datetime,
'21ID': group1_id,
'xkwID': group2_id
"title": record.get("title"),
"url": record.get("url"),
"source_time": record.get("time"),
"id": extract_url_id(record.get("url"), site),
"file_kind": record.get("file_kind"),
"file_name": file_info.get("name"),
"file_mtime": file_info.get("mtime"),
"file_datetime": normalize_datetime(file_info.get("datetime")) or epoch_to_datetime(file_info.get("mtime")),
"archive_mtime": record.get("archive_mtime"),
"archive_datetime": normalize_datetime(record.get("archive_datetime")) or epoch_to_datetime(record.get("archive_mtime")),
"latest_child_mtime": record.get("latest_child_mtime"),
"latest_child_datetime": normalize_datetime(record.get("latest_child_datetime")) or epoch_to_datetime(record.get("latest_child_mtime")),
"child_archive_time_relation": record.get("child_archive_time_relation"),
"child_archive_time_diff_seconds": record.get("child_archive_time_diff_seconds"),
"json": record,
}
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,
21cn_son_file_datetime, xkw_son_file_datetime, 21ID, xkwID
) VALUES (
%s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s, %s, %s, %s, %s, %s,
%s, %s, %s, %s
def build_side_from_pair(group: Dict, site: str, pair_record: Dict) -> Dict:
file_info = first_file_from_group(group)
return {
"title": group.get("title"),
"url": group.get("url"),
"source_time": group.get("time"),
"id": extract_url_id(group.get("url"), site),
"file_kind": "archive",
"file_name": file_info.get("name"),
"file_mtime": file_info.get("mtime"),
"file_datetime": normalize_datetime(file_info.get("datetime")) or epoch_to_datetime(file_info.get("mtime")),
"archive_mtime": file_info.get("mtime"),
"archive_datetime": normalize_datetime(file_info.get("datetime")) or epoch_to_datetime(file_info.get("mtime")),
"latest_child_mtime": pair_record.get(f"{site}_latest_child_mtime"),
"latest_child_datetime": normalize_datetime(pair_record.get(f"{site}_latest_child_datetime")) or epoch_to_datetime(pair_record.get(f"{site}_latest_child_mtime")),
"child_archive_time_relation": "different",
"child_archive_time_diff_seconds": None,
"json": group,
}
def empty_import_record(category: str, source_folder: str, word_doc: str, row_index: int, raw_json) -> Dict:
record = {column: None for column in INSERT_COLUMNS}
record.update({
"category": category,
"source_folder": source_folder,
"word_doc": word_doc,
"row_index": row_index,
"raw_json": json_dumps(raw_json),
})
return record
def apply_side(record: Dict, site: str, side: Dict) -> None:
prefix = "twole" if site == "twole" else "zxxk"
for key in (
"title", "url", "source_time", "id", "file_kind", "file_name", "file_mtime", "file_datetime",
"archive_mtime", "archive_datetime", "latest_child_mtime", "latest_child_datetime",
"child_archive_time_relation", "child_archive_time_diff_seconds",
):
record[f"{prefix}_{key}"] = side.get(key)
record[f"{prefix}_json"] = json_dumps(side.get("json"))
def row_key(record: Dict) -> Tuple[str, str, int]:
return (record.get("source_folder") or "", record.get("word_doc") or "", int(record.get("row_index") or 0))
def normalize_individual_category(category: str, records: Iterable[Dict]) -> List[Dict]:
grouped: Dict[Tuple[str, str, int], Dict] = {}
raw_records: Dict[Tuple[str, str, int], List[Dict]] = {}
for source_record in records:
key = row_key(source_record)
raw_records.setdefault(key, []).append(source_record)
if key not in grouped:
grouped[key] = empty_import_record(category, key[0], key[1], key[2], raw_records[key])
grouped[key]["raw_json"] = json_dumps(raw_records[key])
site = source_record.get("site")
if site in ("twole", "zxxk"):
apply_side(grouped[key], site, build_side_from_individual(source_record))
return [grouped[key] for key in sorted(grouped)]
def normalize_pair_category(category: str, records: Iterable[Dict]) -> List[Dict]:
normalized = []
for source_record in records:
key = row_key(source_record)
record = empty_import_record(category, key[0], key[1], key[2], source_record)
record["earlier_site"] = source_record.get("earlier_site")
record["earlier_site_key"] = source_record.get("earlier_site_key")
record["child_time_diff_seconds"] = source_record.get("child_time_diff_seconds")
apply_side(record, "twole", build_side_from_pair(source_record.get("twole") or {}, "twole", source_record))
apply_side(record, "zxxk", build_side_from_pair(source_record.get("zxxk") or {}, "zxxk", source_record))
normalized.append(record)
return normalized
def normalize_records_for_import(data: Dict) -> List[Dict]:
normalized: List[Dict] = []
normalized.extend(normalize_individual_category("single_files", data.get("single_files", [])))
normalized.extend(normalize_individual_category("same_child_archive_time", data.get("same_child_archive_time", [])))
normalized.extend(normalize_pair_category("zxxk_child_earlier", data.get("zxxk_child_earlier", [])))
normalized.extend(normalize_pair_category("twole_child_earlier", data.get("twole_child_earlier", [])))
return normalized
def load_json_data(file_path: Path) -> Dict:
with file_path.open("r", encoding="utf-8") as f:
data = json.load(f)
if not isinstance(data, dict):
raise ValueError("当前导入脚本只支持新版分类字典格式 JSON")
return data
def create_connection():
return mysql.connector.connect(**DB_CONFIG)
def create_table(connection, table_name: str) -> None:
cursor = connection.cursor()
try:
cursor.execute(CREATE_TABLE_TEMPLATE.format(table_name=mysql_identifier(table_name)))
connection.commit()
finally:
cursor.close()
def truncate_table(connection, table_name: str) -> None:
cursor = connection.cursor()
try:
cursor.execute(f"TRUNCATE TABLE `{mysql_identifier(table_name)}`")
connection.commit()
finally:
cursor.close()
def build_insert_sql(table_name: str) -> str:
columns = ", ".join(f"`{column}`" for column in INSERT_COLUMNS)
placeholders = ", ".join(["%s"] * len(INSERT_COLUMNS))
updates = ", ".join(
f"`{column}` = VALUES(`{column}`)"
for column in INSERT_COLUMNS
if column not in {"category", "source_folder", "word_doc", "row_index"}
)
return f"""
INSERT INTO `{mysql_identifier(table_name)}` ({columns})
VALUES ({placeholders})
ON DUPLICATE KEY UPDATE {updates}
"""
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'],
flattened['21cn_son_file_datetime'],
flattened['xkw_son_file_datetime'],
flattened['21ID'],
flattened['xkwID']
)
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()
def insert_records(connection, table_name: str, records: List[Dict], batch_size: int = 500) -> int:
if not records:
return 0
def verify_data(connection):
"""验证导入的数据"""
insert_sql = build_insert_sql(table_name)
cursor = connection.cursor()
inserted = 0
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,
21cn_son_file_datetime, xkw_son_file_datetime, 21ID, xkwID
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(f" 21cn_son_file_datetime={row[4]}")
print(f" xkw_son_file_datetime={row[5]}")
print(f" 21ID={row[6]}")
print(f" xkwID={row[7]}")
print("-" * 50)
for start in range(0, len(records), batch_size):
batch = records[start:start + batch_size]
values = [tuple(record.get(column) for column in INSERT_COLUMNS) for record in batch]
cursor.executemany(insert_sql, values)
connection.commit()
inserted += len(batch)
print(f"已写入/更新 {inserted}/{len(records)} 条记录")
finally:
cursor.close()
except Error as e:
print(f"验证数据失败：{e}")
return inserted
def export_table_to_excel(connection, output_path: Path):
"""将数据库表导出为 Excel 文件"""
try:
import pandas as pd
except ImportError:
print("请先安装 pandas 和 openpyxl：pip install pandas openpyxl")
return False
def print_category_summary(records: List[Dict]) -> None:
counts = {category: 0 for category in SELECTED_CATEGORIES}
for record in records:
counts[record["category"]] = counts.get(record["category"], 0) + 1
print("待导入分类统计：")
for category in SELECTED_CATEGORIES:
print(f" {category}: {counts.get(category, 0)}")
print(f" total: {len(records)}")
try:
cursor = connection.cursor()
cursor.execute("SELECT * FROM compressed_group_time_issues")
rows = cursor.fetchall()
columns = cursor.column_names
cursor.close()
df = pd.DataFrame(rows, columns=columns)
output_path.parent.mkdir(parents=True, exist_ok=True)
df.to_excel(output_path, index=False)
print(f"成功导出 Excel 文件：{output_path}")
return True
except Exception as e:
print(f"导出 Excel 失败：{e}")
return False
def parse_args():
parser = argparse.ArgumentParser(description="导入 compressed_group_time_issues.json 到 MySQL")
parser.add_argument("--json-file", type=Path, default=DEFAULT_JSON_FILE_PATH, help="JSON 文件路径")
parser.add_argument("--table", default=DEFAULT_TABLE_NAME, help="MySQL 表名")
parser.add_argument("--truncate", action="store_true", help="导入前清空目标表")
parser.add_argument("--dry-run", action="store_true", help="只解析和统计，不连接 MySQL")
return parser.parse_args()
def main():
"""主函数"""
print("=" * 60)
print("开始导入 compressed_group_time_issues.json 到 MySQL")
print("=" * 60)
args = parse_args()
data = load_json_data(args.json_file)
records = normalize_records_for_import(data)
print_category_summary(records)
# 1. 创建数据库连接
connection = create_connection()
if not connection:
if args.dry_run:
print("dry-run 模式：未连接 MySQL，未写入数据")
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)
# 6. 导出 Excel
excel_path = Path('server/jsons/compressed_group_time_issues.xlsx')
export_table_to_excel(connection, excel_path)
# 关闭连接
connection.close()
print("\n数据库连接已关闭")
print("=" * 60)
print("导入完成!")
print("=" * 60)
connection = None
try:
connection = create_connection()
print("成功连接到 MySQL")
create_table(connection, args.table)
print(f"已确认目标表：{args.table}")
if args.truncate:
truncate_table(connection, args.table)
print(f"已清空目标表：{args.table}")
inserted = insert_records(connection, args.table, records)
print(f"导入完成：写入/更新 {inserted} 条记录")
except Error as exc:
print(f"MySQL 操作失败：{exc}")
raise
finally:
if connection and connection.is_connected():
connection.close()
print("MySQL 连接已关闭")
if __name__ == '__main__':
if __name__ == "__main__":
main()