From f5c0ecc31f4d3d4a915bb5059384097950abefec Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Wed, 1 Jul 2026 15:52:29 +0800 Subject: [PATCH] modified pppp --- server/association_detail_smartold.py | 159 ++++++- server/import_json_to_mysql.py | 604 ++++++++++++++------------ 2 files changed, 464 insertions(+), 299 deletions(-) diff --git a/server/association_detail_smartold.py b/server/association_detail_smartold.py index 9ea57b1..ca43314 100644 --- a/server/association_detail_smartold.py +++ b/server/association_detail_smartold.py @@ -628,8 +628,27 @@ def has_audio_video_files(files_list: List[Dict]) -> bool: return any(file_entry.get('name', '').lower().endswith(ext) for file_entry in files_list for ext in audio_video_exts) +def create_empty_time_classification_report() -> Dict[str, List[Dict]]: + return { + "single_files": [], + "archives": [], + "same_child_archive_time": [], + "zxxk_child_earlier": [], + "twole_child_earlier": [], + } + + +def merge_time_classification_reports(target: Dict[str, List[Dict]], source: Dict[str, List[Dict]]) -> None: + for category in create_empty_time_classification_report(): + target.setdefault(category, []) + target[category].extend(source.get(category, [])) + + def build_file_associations(word_content: List[str], twole_folder: Path, zxxk_folder: Path) -> Dict: """构建文件关联关系""" + SAME_TIME_THRESHOLD = 300 + PAIR_TIME_DIFF_THRESHOLD = 30 + def process_file(file_path: Path) -> List[Dict]: files = [] mtime = file_path.stat().st_mtime @@ -657,35 +676,94 @@ def build_file_associations(word_content: List[str], twole_folder: Path, zxxk_fo files.append({"name": f.name, "mtime": mtime, "datetime": format_datetime(mtime), "path": str(f)}) return files - def get_latest_extracted_mtime(file_entries: List[Dict], archive_name: str) -> Optional[float]: + def get_archive_time_summary(file_entries: List[Dict], archive_name: str) -> Dict: extracted_times = [e["mtime"] for e in file_entries if e.get("name") != archive_name and "mtime" in e] - if not extracted_times: - return None archive_mtime = next((e["mtime"] for e in file_entries if e.get("name") == archive_name), None) + if not extracted_times: + return { + "archive_mtime": archive_mtime, + "archive_datetime": format_datetime(archive_mtime) if archive_mtime else None, + "latest_child_mtime": None, + "latest_child_datetime": None, + "child_archive_time_relation": "no_child_files", + "child_archive_time_diff_seconds": None, + } latest_extracted = max(extracted_times) - return None if archive_mtime and abs(latest_extracted - archive_mtime) < 300 else latest_extracted + time_diff = latest_extracted - archive_mtime if archive_mtime is not None else None + is_same_time = time_diff is not None and abs(time_diff) < SAME_TIME_THRESHOLD + return { + "archive_mtime": archive_mtime, + "archive_datetime": format_datetime(archive_mtime) if archive_mtime else None, + "latest_child_mtime": latest_extracted, + "latest_child_datetime": format_datetime(latest_extracted), + "child_archive_time_relation": "same" if is_same_time else "different", + "child_archive_time_diff_seconds": time_diff, + } - def match_group(title: str, url: str, time_val: str, group_name: str, twole_files: List, zxxk_files: List) -> Tuple[Dict, Optional[float]]: + def build_classification_record(row_index: int, group_name: str, entry: Dict, file_kind: str, archive_summary: Optional[Dict] = None) -> Dict: + site = get_url_site(entry.get("url", "")) + record = { + "row_index": row_index, + "group": group_name, + "site": site, + "site_name": "二一教育" if site == "twole" else "学科网" if site == "zxxk" else "", + "file_kind": file_kind, + "title": entry.get("title"), + "url": entry.get("url"), + "time": entry.get("time"), + "source_path": entry.get("source_path"), + "file": entry.get("files", [{}])[0] if entry.get("files") else {}, + } + if archive_summary: + record.update(archive_summary) + return record + + def build_pair_record(row_index: int, entry: Dict, twole_summary: Dict, zxxk_summary: Dict, earlier_site: str) -> Dict: + twole_entry = twole_summary["entry"] + zxxk_entry = zxxk_summary["entry"] + return { + "row_index": row_index, + "earlier_site": "二一教育" if earlier_site == "twole" else "学科网", + "earlier_site_key": earlier_site, + "twole_latest_child_mtime": twole_summary["archive"]["latest_child_mtime"], + "twole_latest_child_datetime": twole_summary["archive"]["latest_child_datetime"], + "zxxk_latest_child_mtime": zxxk_summary["archive"]["latest_child_mtime"], + "zxxk_latest_child_datetime": zxxk_summary["archive"]["latest_child_datetime"], + "child_time_diff_seconds": zxxk_summary["archive"]["latest_child_mtime"] - twole_summary["archive"]["latest_child_mtime"], + "twole": twole_entry, + "zxxk": zxxk_entry, + "group1": entry.get("group1"), + "group2": entry.get("group2"), + } + + def match_group(title: str, url: str, time_val: str, group_name: str, twole_files: List, zxxk_files: List) -> Tuple[Dict, Dict]: if 'www.21cnjy.com' in url and twole_files: candidates = twole_files elif 'www.zxxk.com' in url and zxxk_files: candidates = zxxk_files else: - return {}, None + return {}, {} best_match = find_best_match(title, candidates) if not best_match: - return {}, None + return {}, {} candidates.remove(best_match) file_path, _ = best_match files = process_file(file_path) entry = {"files": files, "title": title, "url": url, "time": time_val, "source_path": str(file_path)} - latest = get_latest_extracted_mtime(files, file_path.name) if file_path.suffix.lower() in ('.zip', '.rar', '.7z') else None - return entry, latest + is_archive = file_path.suffix.lower() in ARCHIVE_EXTENSIONS + summary = { + "entry": entry, + "group_name": group_name, + "site": get_url_site(url), + "is_archive": is_archive, + "archive": get_archive_time_summary(files, file_path.name) if is_archive else None, + } + return entry, summary result = [] - problematic_groups = [] + problematic_groups = create_empty_time_classification_report() twole_files = get_files_with_times(twole_folder) zxxk_files = get_files_with_times(zxxk_folder) @@ -698,8 +776,8 @@ def build_file_associations(word_content: List[str], twole_folder: Path, zxxk_fo title2, url2, time2 = parts[3].strip(), parts[4].strip(), parts[5].strip() entry = {} - group1_entry, group1_latest = match_group(title1, url1, time1, "group1", twole_files, zxxk_files) - group2_entry, group2_latest = match_group(title2, url2, time2, "group2", twole_files, zxxk_files) + group1_entry, group1_summary = match_group(title1, url1, time1, "group1", twole_files, zxxk_files) + group2_entry, group2_summary = match_group(title2, url2, time2, "group2", twole_files, zxxk_files) if group1_entry: entry["group1"] = group1_entry @@ -707,13 +785,44 @@ def build_file_associations(word_content: List[str], twole_folder: Path, zxxk_fo entry["group2"] = group2_entry if entry: - pair_sites = {get_url_site(entry.get("group1", {}).get("url", "")), get_url_site(entry.get("group2", {}).get("url", ""))} - if (pair_sites == {'twole', 'zxxk'} and group1_latest is not None and group2_latest is not None and - group1_latest < group2_latest and (group2_latest - group1_latest) > 30): - problematic_groups.append({ - "row_index": i, "group1_latest_extracted_mtime": group1_latest, "group2_latest_extracted_mtime": group2_latest, - "group1": entry.get("group1"), "group2": entry.get("group2") - }) + archive_summaries_by_site = {} + for summary in (group1_summary, group2_summary): + if not summary: + continue + record = build_classification_record( + i, + summary["group_name"], + summary["entry"], + "archive" if summary["is_archive"] else "single_file", + summary["archive"], + ) + if summary["is_archive"]: + problematic_groups["archives"].append(record) + archive_summaries_by_site[summary["site"]] = summary + if summary["archive"]["child_archive_time_relation"] == "same": + problematic_groups["same_child_archive_time"].append(record) + else: + problematic_groups["single_files"].append(record) + + twole_summary = archive_summaries_by_site.get("twole") + zxxk_summary = archive_summaries_by_site.get("zxxk") + if twole_summary and zxxk_summary: + twole_archive = twole_summary["archive"] + zxxk_archive = zxxk_summary["archive"] + if (twole_archive["child_archive_time_relation"] == "different" and + zxxk_archive["child_archive_time_relation"] == "different" and + twole_archive["latest_child_mtime"] is not None and + zxxk_archive["latest_child_mtime"] is not None): + child_time_diff = zxxk_archive["latest_child_mtime"] - twole_archive["latest_child_mtime"] + if abs(child_time_diff) > PAIR_TIME_DIFF_THRESHOLD: + if child_time_diff < 0: + problematic_groups["zxxk_child_earlier"].append( + build_pair_record(i, entry, twole_summary, zxxk_summary, "zxxk") + ) + else: + problematic_groups["twole_child_earlier"].append( + build_pair_record(i, entry, twole_summary, zxxk_summary, "twole") + ) result.append(entry) return {"items": result, "problematic_groups": problematic_groups} @@ -929,7 +1038,7 @@ def main(): total_failures = 0 total_conflicts = 0 total_duplicates = 0 - all_problem_groups = [] + all_problem_groups = create_empty_time_classification_report() all_write_reports = [] for source_item in source_items: @@ -964,10 +1073,12 @@ def main(): print(f" 关联目录：{base_dir}") associations = build_file_associations(word_content, twole_folder, zxxk_folder) - for group in associations.get("problematic_groups", []): - group["source_folder"] = source_item.name - group["word_doc"] = word_doc.name - all_problem_groups.extend(associations.get("problematic_groups", [])) + classification_report = associations.get("problematic_groups", create_empty_time_classification_report()) + for records in classification_report.values(): + for record in records: + record["source_folder"] = source_item.name + record["word_doc"] = word_doc.name + merge_time_classification_reports(all_problem_groups, classification_report) if not associations.get("items"): print("⚠️ 未构建出有效关联") diff --git a/server/import_json_to_mysql.py b/server/import_json_to_mysql.py index 0814086..df17319 100644 --- a/server/import_json_to_mysql.py +++ b/server/import_json_to_mysql.py @@ -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()