From 54943470af65f1d444feedcb1591777530a6fbdc Mon Sep 17 00:00:00 2001 From: Shuming Liu Date: Fri, 3 Apr 2026 09:17:09 +0800 Subject: [PATCH] save a version --- server/file_association.py | 175 +++++++++++++++++++++++++++++-------- 1 file changed, 140 insertions(+), 35 deletions(-) diff --git a/server/file_association.py b/server/file_association.py index 0acca4c..89371b8 100644 --- a/server/file_association.py +++ b/server/file_association.py @@ -18,47 +18,60 @@ import shutil import shutil +def clear_directory(dir_path: Path) -> None: + """清空目录中的所有文件和子目录""" + if dir_path.exists(): + for item in dir_path.iterdir(): + if item.is_file(): + item.unlink() + elif item.is_dir(): + shutil.rmtree(item) + + def extract_all_zips( - source_dir: Union[str, Path] = "data", - output_dir: Optional[Union[str, Path]] = None, + source_dir: Union[str, Path] = "cc", + output_dir: Union[str, Path] = "data", delete_original: bool = True ) -> List[Path]: """ 解压指定目录中的所有 zip 文件 - + Args: source_dir: 包含压缩文件的目录 output_dir: 输出目录 delete_original: 解压后删除原文件 - + Returns: 所有解压后的目录路径列表 """ source_dir = Path(source_dir) - output_dir = Path(output_dir) if output_dir else source_dir - + output_dir = Path(output_dir) + + # 清空 output_dir + clear_directory(output_dir) + zip_files = list(source_dir.glob("*.zip")) - + if not zip_files: print(f"在 {source_dir} 中未发现 zip 文件") return [] - - print(f"找到 {len(zip_files)} 个压缩文件，开始解压...") + + print(f"找到 {len(zip_files)} 个压缩文件，开始解压到 {output_dir}...") results = [] for zip_file in zip_files: # 解压到指定目录 extract_dir = output_dir / zip_file.stem extract_dir.mkdir(parents=True, exist_ok=True) - + with zipfile.ZipFile(zip_file, 'r') as zip_ref: # 解压所有文件并保留原始时间戳 for member in zip_ref.infolist(): # 解压文件 zip_ref.extract(member, extract_dir) - + # 获取解压后的文件路径 member_path = extract_dir / member.filename - + # 如果文件存在，设置时间戳 if member_path.exists(): # 将 ZipInfo 的日期时间转换为时间戳 @@ -66,7 +79,7 @@ def extract_all_zips( timestamp = date_time.timestamp() # 设置访问时间和修改时间 os.utime(member_path, (timestamp, timestamp)) - + # 检查是否多了一层目录（zip 内部只有一个同名文件夹） inner_dir = extract_dir / zip_file.stem if inner_dir.is_dir() and len(list(extract_dir.iterdir())) == 1: @@ -78,13 +91,13 @@ def extract_all_zips( print(f"✓ 已解压：{zip_file.name} (移除了多余目录层)") else: print(f"✓ 已解压：{zip_file.name}") - + results.append(extract_dir) - + # 删除原压缩包 if delete_original: zip_file.unlink() - + return results @@ -240,14 +253,15 @@ def get_files_with_times(folder: Path) -> List[Tuple[Path, datetime]]: return files -def extract_zip_recursive(zip_path: Path, extract_dir: Path) -> List[Path]: +def extract_zip_recursive(zip_path: Path, extract_dir: Path, clear_dir: bool = True) -> List[Path]: """递归解压 zip 文件，如果解压后的文件还有 zip，继续解压""" # 先清空 extract_dir 中的内容 - for item in extract_dir.iterdir(): - if item.is_file(): - item.unlink() - elif item.is_dir(): - shutil.rmtree(item) + if clear_dir: + for item in extract_dir.iterdir(): + if item.is_file(): + item.unlink() + elif item.is_dir(): + shutil.rmtree(item) # 解压当前层 with zipfile.ZipFile(zip_path, 'r') as zip_ref: @@ -306,14 +320,8 @@ def build_file_associations( tmp_dir.mkdir(exist_ok=True) extract_target = tmp_dir / file_path.stem # 清空 extract_target 目录 - if extract_target.exists(): - for item in extract_target.iterdir(): - if item.is_file(): - item.unlink() - elif item.is_dir(): - shutil.rmtree(item) - else: - extract_target.mkdir(parents=True, exist_ok=True) + clear_directory(extract_target) + extract_target.mkdir(parents=True, exist_ok=True) # 根据压缩格式选择解压方式 if file_path.suffix.lower() == '.zip': @@ -373,17 +381,114 @@ def main(): parser = argparse.ArgumentParser(description='数据文件关联处理程序') parser.add_argument('--extract', action='store_true', help='解压压缩文件') parser.add_argument('--associate', action='store_true', help='建立文件关联') + parser.add_argument('--batch', action='store_true', help='批量处理所有 zip 文件') parser.add_argument('--base-dir', type=str, help='基础目录') parser.add_argument('--delete-original', action='store_true', help='解压后删除原文件') - + args = parser.parse_args() - + # 解压压缩文件 if args.extract: - source_dir = Path(args.base_dir) if args.base_dir else Path("data") - extract_all_zips(source_dir,"tmp", delete_original=args.delete_original) + source_dir = Path(args.base_dir) if args.base_dir else Path("cc") + extract_all_zips(source_dir, "data", delete_original=args.delete_original) return - + + # 批量处理所有 zip 文件 + if True or args.batch: + source_dir = Path(args.base_dir) if args.base_dir else Path("cc") + data_dir = Path("data") + server_jsons_dir = Path("server") / "jsons" + server_jsons_dir.mkdir(parents=True, exist_ok=True) + + zip_files = list(source_dir.glob("*.zip")) + + if not zip_files: + print(f"在 {source_dir} 中未发现 zip 文件") + return + + # 清空 data 目录 + clear_directory(data_dir) + + print(f"找到 {len(zip_files)} 个压缩文件，开始批量处理...") + success_count = 0 + fail_count = 0 + + for zip_file in zip_files: + print(f"\n【开始处理】{zip_file.name}") + + # 清空当前 zip 文件的提取目录 + extract_dir = data_dir / zip_file.stem + if extract_dir.exists(): + clear_directory(extract_dir) + + try: + # 解压 + with zipfile.ZipFile(zip_file, 'r') as zip_ref: + for member in zip_ref.infolist(): + zip_ref.extract(member, data_dir) + member_path = data_dir / zip_file.stem / member.filename + if member_path.exists(): + date_time = datetime(*member.date_time[:5]) + timestamp = date_time.timestamp() + os.utime(member_path, (timestamp, timestamp)) + + # 检查是否多了一层目录 + inner_dir = extract_dir / zip_file.stem + if inner_dir.is_dir() and len(list(extract_dir.iterdir())) == 1: + for item in inner_dir.iterdir(): + item.rename(extract_dir / item.name) + inner_dir.rmdir() + + # 查找 Word 文档 + word_doc = find_word_doc_recursive(extract_dir) + if not word_doc: + print(f"⚠️ 在 {zip_file.name} 中未找到 Word 文档") + fail_count += 1 + continue + + # 读取 Word 内容 + word_content = read_word_content(word_doc) + if not word_content: + print(f"⚠️ 无法读取 Word 文档内容") + fail_count += 1 + continue + + base_dir = word_doc.parent + print(f"✓ 找到 Word 文档：{word_doc}") + print(f" 关联目录：{base_dir}") + + # 建立文件关联 + twole_folder = base_dir / "21世纪教育网" + zxxk_folder = base_dir / "学科网" + + if not twole_folder.exists() or not zxxk_folder.exists(): + print("❌ 文件夹不存在，请先解压文件") + fail_count += 1 + continue + + associations = build_file_associations(word_content, twole_folder, zxxk_folder) + + # 保存 JSON 文件 + output_file = server_jsons_dir / f"{word_doc.stem}.json" + with open(output_file, 'w', encoding='utf-8') as f: + json.dump(associations, f, ensure_ascii=False, indent=2) + + print(f"✓ JSON 文件已保存到：{output_file}") + + # 删除原压缩包 + if args.delete_original: + zip_file.unlink() + + success_count += 1 + + except Exception as e: + print(f"❌ 处理失败：{e}") + fail_count += 1 + + print(f"\n===== 批量处理完成 =====") + print(f"成功：{success_count}, 失败：{fail_count}") + return + # 建立文件关联 if args.associate: base_dir = Path(args.base_dir) if args.base_dir else Path("data")