#!/usr/bin/env python3 """ 悟空公文安装文件自动归类和重命名工具 (改进版) 将 original_file 文件夹中的文件根据平台和架构自动归类到对应文件夹 并重命名为包含版本号和缓存破坏机制的标准格式 同时更新 update.txt 中的版本号 改进内容： 1. 修复目录名称拼写错误 2. 使用移动而非复制操作 3. 改进版本号提取逻辑 4. 添加缓存破坏机制 5. 增强错误处理和日志记录 6. 支持更多文件格式和版本格式 """ import os import re import shutil import hashlib import json from pathlib import Path from datetime import datetime from typing import Tuple, Optional, Dict, List class WukongRenamerImproved: def __init__(self, debug: bool = False): # 设置基础路径 self.base_path = Path(__file__).parent self.original_folder = self.base_path / "original_file" # 修复拼写错误 self.update_file = self.base_path / "update.txt" self.log_file = self.base_path / "rename_log.json" self.debug = debug # 定义平台和架构的映射关系 self.platform_mapping = { 'linux-amd64.deb': ('linux', 'amd64'), 'mac-arm64.dmg': ('mac', 'arm64'), 'mac-x64.dmg': ('mac', 'x64'), 'mac-intel.dmg': ('mac', 'intel'), 'x64-setup.exe': ('win', 'x64'), 'ia32-setup.exe': ('win', 'x86'), 'win-x64.exe': ('win', 'x64'), 'win-x86.exe': ('win', 'x86'), 'win-x64.msi': ('win', 'x64'), 'win-x86.msi': ('win', 'x86'), 'linux-x64.tar.gz': ('linux', 'x64'), 'linux-arm64.tar.gz': ('linux', 'arm64'), } # 存储操作结果 self.processing_results = { 'timestamp': datetime.now().isoformat(), 'processed_files': [], 'skipped_files': [], 'errors': [], 'latest_version': None, 'success_count': 0, 'skip_count': 0 } def log_message(self, message: str, level: str = "info"): """记录日志消息""" timestamp = datetime.now().strftime("%H:%M:%S") log_entry = f"[{timestamp}] [{level.upper()}] {message}" print(log_entry) if self.debug: # 将日志写入文件 try: with open(self.base_path / "debug.log", "a", encoding="utf-8") as f: f.write(log_entry + "\n") except Exception: pass # 忽略日志写入错误 def extract_version(self, filename: str) -> Optional[str]: """从文件名中提取版本号，支持多种格式""" # 版本号匹配模式 patterns = [ r'(?:悟空公文|wukong)[-_]?v?(\d+\.\d+\.\d+)(?:[-_](?:linux|mac|win|amd64|x64|x86|arm64|setup))?[^-\d]*', # 标准三段版本号 r'(?:悟空公文|wukong)[-_]?v?(\d+\.\d+\.\d+[^-\d]*)', # 更宽松的三段版本号匹配 r'(?:悟空公文|wukong)[-_]?v?(\d+\.\d+\.\d+\.\d+)[^-\d]*', # 四段版本号 r'(?:悟空公文|wukong)[-_]?v?(\d+\.\d+)[^-\d]*', # 两段版本号 r'(\d+\.\d+\.\d+)(?:[-_](?:linux|mac|win|amd64|x64|x86|arm64|setup))?[^-\d]*', # 通用版本号匹配 ] for pattern in patterns: match = re.search(pattern, filename) if match: version = match.group(1) self.log_message(f"提取到版本号: {version}") return version self.log_message(f"无法从文件名中提取版本号: {filename}", "warning") return None def get_file_platform(self, filename: str) -> Tuple[Optional[str], Optional[str]]: """根据文件名判断平台和架构""" filename_lower = filename.lower() # 首先尝试精确匹配 for platform_key, (platform, arch) in self.platform_mapping.items(): if all(part in filename_lower for part in platform_key.split('-')): return platform, arch # 宽松匹配逻辑 if 'linux' in filename_lower: if 'arm64' in filename_lower: return 'linux', 'arm64' elif ('x64' in filename_lower or 'x86_64' in filename_lower or 'amd64' in filename_lower): return 'linux', 'x64' elif '.deb' in filename_lower: return 'linux', 'amd64' # 默认AMD64 elif '.tar.gz' in filename_lower: return 'linux', 'x64' # 默认x64 elif 'mac' in filename_lower or 'darwin' in filename_lower: if 'arm64' in filename_lower or ('apple' in filename_lower and 'silicon' in filename_lower): return 'mac', 'arm64' elif 'x64' in filename_lower or 'x86_64' in filename_lower or 'intel' in filename_lower: return 'mac', 'x64' elif '.dmg' in filename_lower: return 'mac', 'x64' # 默认Intel elif 'win' in filename_lower or filename.endswith(('.exe', '.msi')): if 'x64' in filename_lower or '64' in filename_lower or 'amd64' in filename_lower: return 'win', 'x64' elif 'x86' in filename_lower or '32' in filename_lower or 'ia32' in filename_lower: return 'win', 'x86' elif filename.endswith(('.exe', '.msi')): return 'win', 'x64' # 默认64位 self.log_message(f"无法识别平台: {filename}", "warning") return None, None def get_target_path(self, platform: str, arch: str, filename: str, version: Optional[str]) -> Optional[Path]: """获取目标文件路径""" # 确定目标目录 if platform == 'linux': target_dir = self.base_path / 'linux' elif platform == 'mac': if arch in ['arm64', 'x64', 'intel']: target_dir = self.base_path / 'mac' / arch else: target_dir = self.base_path / 'mac' elif platform == 'win': if arch in ['x64', 'x86']: target_dir = self.base_path / 'win' / arch else: target_dir = self.base_path / 'win' else: return None # 确保目标目录存在 try: target_dir.mkdir(parents=True, exist_ok=True) self.log_message(f"创建目标目录: {target_dir}") except Exception as e: self.log_message(f"创建目标目录失败: {e}", "error") return None # 获取文件扩展名 file_path = Path(filename) file_extension = ''.join(file_path.suffixes) # 支持多个扩展名如 .tar.gz # 生成包含版本号的文件名 if version: timestamp = datetime.now().strftime("%Y%m%d%H%M%S") # 计算文件哈希作为缓存破坏机制 target_filename = f"wukong-{version}-{timestamp}{file_extension}" else: timestamp = datetime.now().strftime("%Y%m%d%H%M%S") target_filename = f"wukong-latest-{timestamp}{file_extension}" return target_dir / target_filename def calculate_file_hash(self, file_path: Path) -> str: """计算文件的MD5哈希值""" try: with open(file_path, 'rb') as f: return hashlib.md5(f.read()).hexdigest()[:8] except Exception as e: self.log_message(f"计算文件哈希失败: {e}", "error") return "unknown" def update_version_file(self, version: str): """更新 update.txt 文件中的版本号""" try: # 创建备份 if self.update_file.exists(): backup_file = self.update_file.with_suffix('.txt.bak') shutil.copy2(self.update_file, backup_file) self.log_message(f"备份版本文件到: {backup_file}") # 写入新版本号 with open(self.update_file, 'w', encoding='utf-8') as f: version_info = { 'version': version, 'timestamp': datetime.now().isoformat(), 'files_processed': len(self.processing_results['processed_files']) } f.write(json.dumps(version_info, indent=2, ensure_ascii=False)) self.log_message(f"✅ 版本号已更新为: {version}") return True except Exception as e: self.log_message(f"❌ 更新版本文件失败: {e}", "error") self.processing_results['errors'].append(f"更新版本文件失败: {e}") return False def save_processing_log(self): """保存处理日志""" try: with open(self.log_file, 'w', encoding='utf-8') as f: json.dump(self.processing_results, f, indent=2, ensure_ascii=False) self.log_message(f"处理日志已保存到: {self.log_file}") except Exception as e: self.log_message(f"保存处理日志失败: {e}", "error") def process_files(self) -> bool: """处理所有文件""" self.log_message("🚀 开始处理文件") self.log_message(f"源文件夹: {self.original_folder}") # 检查源文件夹是否存在 if not self.original_folder.exists(): error_msg = f"❌ 找不到源文件夹: {self.original_folder}" self.log_message(error_msg, "error") self.processing_results['errors'].append(error_msg) return False # 获取所有文件 try: files = [f for f in self.original_folder.glob("*") if f.is_file()] except Exception as e: error_msg = f"读取源文件夹失败: {e}" self.log_message(error_msg, "error") self.processing_results['errors'].append(error_msg) return False if not files: self.log_message(f"⚠️ 源文件夹为空: {self.original_folder}", "warning") return False self.log_message(f"🔍 找到 {len(files)} 个文件待处理") for file_path in files: try: self._process_single_file(file_path) except Exception as e: error_msg = f"处理文件 {file_path.name} 时发生异常: {e}" self.log_message(error_msg, "error") self.processing_results['errors'].append(error_msg) self.processing_results['skip_count'] += 1 # 总结处理结果 self._log_processing_summary() # 更新版本号 if self.processing_results['latest_version']: self.update_version_file(self.processing_results['latest_version']) else: self.log_message("⚠️ 未找到版本号，update.txt 未更新", "warning") # 保存处理日志 self.save_processing_log() return self.processing_results['success_count'] > 0 def _process_single_file(self, file_path: Path): """处理单个文件""" filename = file_path.name # 跳过元数据文件 if any(suffix in filename for suffix in ['.blockmap', '.sig', '.asc']): self.log_message(f"⏭️ 跳过元数据文件: {filename}") self.processing_results['skipped_files'].append({ 'filename': filename, 'reason': '元数据文件' }) self.processing_results['skip_count'] += 1 return self.log_message(f"📄 处理文件: {filename}") # 提取版本号 version = self.extract_version(filename) if version: self.processing_results['latest_version'] = version # 获取平台和架构 platform, arch = self.get_file_platform(filename) if not platform: self.log_message(f" ⚠️ 无法识别平台，跳过") self.processing_results['skipped_files'].append({ 'filename': filename, 'reason': '无法识别平台' }) self.processing_results['skip_count'] += 1 return self.log_message(f" 平台: {platform}, 架构: {arch}, 版本: {version or '未知'}") # 获取目标路径 target_path = self.get_target_path(platform, arch, filename, version) if not target_path: self.log_message(f" ❌ 无法确定目标路径，跳过") self.processing_results['errors'].append(f"无法确定 {filename} 的目标路径") self.processing_results['skip_count'] += 1 return # 移动文件到目标位置 try: # 如果目标文件已存在，先删除 if target_path.exists(): target_path.unlink() self.log_message(f" 🗑️ 删除已存在的文件: {target_path}") # 计算文件哈希 file_hash = self.calculate_file_hash(file_path) # 移动文件 shutil.move(str(file_path), str(target_path)) self.log_message(f" ✅ 已移动到: {target_path}") # 记录处理结果 self.processing_results['processed_files'].append({ 'original_name': filename, 'new_name': target_path.name, 'platform': platform, 'arch': arch, 'version': version, 'file_hash': file_hash, 'target_path': str(target_path) }) self.processing_results['success_count'] += 1 except Exception as e: error_msg = f" ❌ 移动文件失败: {e}" self.log_message(error_msg, "error") self.processing_results['errors'].append(f"移动 {filename} 失败: {e}") self.processing_results['skip_count'] += 1 def _log_processing_summary(self): """记录处理总结""" self.log_message("\n📊 处理完成:") self.log_message(f" ✅ 成功处理: {self.processing_results['success_count']} 个文件") self.log_message(f" ⏭️ 跳过: {self.processing_results['skip_count']} 个文件") self.log_message(f" ❌ 错误: {len(self.processing_results['errors'])} 个") if self.processing_results['latest_version']: self.log_message(f" 📋 最新版本: {self.processing_results['latest_version']}") def main(): """主函数""" print("🚀 悟空公文文件自动归类和重命名工具 (改进版)") print("=" * 60) # 解析命令行参数 debug_mode = '--debug' in sys.argv if 'sys' in globals() else False renamer = WukongRenamerImproved(debug=debug_mode) success = renamer.process_files() if success: print("\n✨ 所有操作完成！") return 0 else: print("\n❌ 处理过程中遇到问题") return 1 if __name__ == "__main__": import sys exit(main())