批量重命名与文件分类整理

import os import shutil from pathlib import Path # 方法一：os.rename() 基础重命名 os.rename('old_name.txt', 'new_name.txt') # 批量重命名目录下的所有 .txt 文件 folder = './documents' for i, filename in enumerate(os.listdir(folder)): if filename.endswith('.txt'): old_path = os.path.join(folder, filename) new_path = os.path.join(folder, f'file_{i+1:03d}.txt') os.rename(old_path, new_path)

# 方法二：shutil.move() 重命名并移动 shutil.move('./temp/report.txt', './archive/final_report.txt') # 批量移动到分类目录 for filename in os.listdir('./downloads'): if filename.endswith('.pdf'): src = f'./downloads/{filename}' dst = f'./pdf_files/{filename}' shutil.move(src, dst)

# 方法三：pathlib 方式 (推荐) folder = Path('./documents') for file in folder.iterdir(): if file.suffix == '.txt': new_name = f"document_{file.stem}.txt" file.rename(folder / new_name) # 结合 glob 模式匹配更简洁 for file in Path('./docs').glob('*.log'): new_path = file.with_suffix('.txt') # 仅修改扩展名 file.rename(new_path)

# 健壮的批量重命名函数 import os def safe_rename(src, dst): """安全重命名，自动处理冲突""" if not os.path.exists(src): raise FileNotFoundError(f"源文件不存在: {src}") if src == dst: return # 相同路径则跳过 # 如果目标已存在则加序号 base, ext = os.path.splitext(dst) counter = 1 while os.path.exists(dst): dst = f"{base}_{counter}{ext}" counter += 1 os.rename(src, dst) return dst # 批量应用 for file in os.listdir('./photos'): if file.lower().endswith(('.jpg', '.png')): new_name = file.lower().replace(' ', '_') safe_rename(f'./photos/{file}', f'./photos/{new_name}')

import re # 基本替换：去除文件名中的空格和特殊字符 def clean_filename(filename): """清洗文件名，只保留中文、字母、数字、下划线和连字符""" name, ext = os.path.splitext(filename) # 保留中文字符、字母、数字，其余替换为下划线 clean = re.sub(r'[^一-龥\w-]', '_', name) # 合并连续下划线 clean = re.sub(r'_+', '_', clean) return clean + ext # 示例 print(clean_filename("项目报告 (final) v2.0!!.pdf")) # 输出: 项目报告_final_v2_0_.pdf

# 分组替换：提取并重组文件名中的信息 def extract_and_rename(filename): """ 从 "2023_Report_财务部_Q4.xlsx" 中提取信息 重命名为 "Q4-财务部-2023_Report.xlsx" """ pattern = r'(\d{4})_Report_(\w+)_(Q[1-4])\.(\w+)' match = re.search(pattern, filename) if match: year, dept, quarter, ext = match.groups() return f"{quarter}-{dept}-{year}_Report.{ext}" return filename # 批量处理一批报告文件 files = ['2023_Report_财务部_Q4.xlsx', '2023_Report_技术部_Q3.xlsx'] renamed = [extract_and_rename(f) for f in files] # 结果: ['Q4-财务部-2023_Report.xlsx', 'Q3-技术部-2023_Report.xlsx']

# 格式转换：蛇形命名与驼峰命名互转 def snake_to_camel(filename): """蛇形命名转驼峰命名：my_file_name -> MyFileName""" name, ext = os.path.splitext(filename) parts = name.split('_') camel = ''.join(p.capitalize() for p in parts if p) return camel + ext def camel_to_snake(filename): """驼峰命名转蛇形命名：MyFileName -> my_file_name""" name, ext = os.path.splitext(filename) # 在字母边界插入下划线并转小写 snake = re.sub(r'(?<=[a-z])(?=[A-Z])', '_', name).lower() return snake + ext print(snake_to_camel("project_data_report.py")) # 输出: ProjectDataReport.py print(camel_to_snake("ProjectDataReport.py")) # 输出: project_data_report.py # 批量匹配特定模式的文件 def filter_and_rename(folder, pattern, replacement): """按正则模式匹配并重命名""" for file in os.listdir(folder): old_name = os.path.join(folder, file) if os.path.isfile(old_name): new_name = re.sub(pattern, replacement, file) if new_name != file: os.rename(old_name, os.path.join(folder, new_name)) # 将所有 "IMG_2023xxxx.jpg" 改为 "Photo_2023xxxx.jpg" filter_and_rename('./photos', r'^IMG_', 'Photo_')

import os from pathlib import Path # 基础序号命名：前缀 + 序号 def basic_rename(folder, prefix='file_', digits=3): files = sorted([f for f in os.listdir(folder) if os.path.isfile(os.path.join(folder, f))]) total = len(files) for i, filename in enumerate(files, 1): _, ext = os.path.splitext(filename) new_name = f"{prefix}{i:0{digits}d}{ext}" os.rename( os.path.join(folder, filename), os.path.join(folder, new_name) ) print(f"共重命名 {total} 个文件") # 使用示例 basic_rename('./course_materials', '第', digits=3) # 文件变为: 第001.pdf, 第002.pdf, ... 第012.pdf

# 按创建时间排序命名 import time def rename_by_date(folder, prefix='photo_', digits=4): items = [] for file in Path(folder).iterdir(): if file.is_file(): # 获取创建时间 ctime = file.stat().st_ctime items.append((ctime, file)) # 按创建时间排序 items.sort(key=lambda x: x[0]) for i, (_, file) in enumerate(items, 1): new_name = f"{prefix}{i:0{digits}d}{file.suffix}" file.rename(Path(folder) / new_name) print(f"按时间排序完成，共处理 {len(items)} 个文件") # 按文件大小排序命名（从大到小） def rename_by_size(folder, prefix='file_'): items = [] for file in Path(folder).iterdir(): if file.is_file(): items.append((file.stat().st_size, file)) # 从大到小排序 items.sort(key=lambda x: x[0], reverse=True) for i, (size, file) in enumerate(items, 1): new_name = f"{prefix}{i:03d}{file.suffix}" file.rename(Path(folder) / new_name) print(f"按大小排序完成，最大文件: {items[0][0]/1024:.1f}KB")

# 自定义序号范围与多级序号 def custom_seq_rename(folder, start=1, step=1, prefix='img_', suffixes=None): """ 自定义序号范围、步长和后缀 suffixes: 可选列表，对应每个文件添加的不同后缀标记 """ files = sorted(Path(folder).iterdir(), key=lambda f: f.name) current = start for i, file in enumerate(files): if file.is_file(): suffix_tag = f"_{suffixes[i]}" if suffixes and i < len(suffixes) else "" new_name = f"{prefix}{current:03d}{suffix_tag}{file.suffix}" file.rename(Path(folder) / new_name) current += step # 示例：照片按组命名 custom_seq_rename('./photos', start=1, prefix='vacation_', suffixes=['beach', 'mountain', 'city', 'beach']) # 生成: vacation_001_beach.jpg, vacation_002_mountain.jpg, ...

import os from datetime import datetime from pathlib import Path # 基础日期命名 def date_rename(folder, date_format='%Y%m%d'): """按文件创建时间添加日期前缀""" for file in Path(folder).iterdir(): if file.is_file(): ctime = datetime.fromtimestamp(file.stat().st_ctime) date_str = ctime.strftime(date_format) new_name = f"{date_str}_{file.name}" file.rename(Path(folder) / new_name) # 精确到秒的命名 def precise_timestamp(folder): for file in Path(folder).iterdir(): if file.is_file(): mtime = datetime.fromtimestamp(file.stat().st_mtime) ts = mtime.strftime('%Y%m%d_%H%M%S') new_name = f"{ts}{file.suffix}" file.rename(Path(folder) / new_name) # 示例 date_rename('./reports', '%Y-%m-%d') # 文件: 2026-05-05_季度报告.xlsx, 2026-05-05_月度总结.docx

# 模板命名引擎 from string import Template class TemplateRenamer: """基于模板的文件重命名器""" def __init__(self, template_str): self.template = template_str def rename(self, folder, seq_start=1): files = sorted(Path(folder).iterdir(), key=lambda f: f.stat().st_ctime) for i, file in enumerate(files, seq_start): if file.is_file(): stat = file.stat() # 准备模板变量 vars_dict = { 'date': datetime.fromtimestamp(stat.st_ctime).strftime('%Y%m%d'), 'datetime': datetime.fromtimestamp(stat.st_ctime).strftime('%Y%m%d_%H%M%S'), 'year': datetime.fromtimestamp(stat.st_ctime).strftime('%Y'), 'month': datetime.fromtimestamp(stat.st_ctime).strftime('%m'), 'day': datetime.fromtimestamp(stat.st_ctime).strftime('%d'), 'seq': f"{i:03d}", 'name': file.stem, 'ext': file.suffix, 'size': f"{stat.st_size // 1024}KB", } # 替换模板变量 new_name = Template(self.template).safe_substitute(vars_dict) file.rename(Path(folder) / new_name) print(f" {file.name} -> {new_name}") # 使用模板 renamer = TemplateRenamer('${date}_${seq}_${name}${ext}') renamer.rename('./docs') # 生成: 20260505_001_项目报告.docx, 20260505_002_会议纪要.docx # 更多模板示例 renamer2 = TemplateRenamer('${year}年${month}月_${name}_${seq}${ext}') renamer2.rename('./reports')

# 从图片 Exif 提取拍摄日期命名 try: from PIL import Image from PIL.ExifTags import TAGS except ImportError: Image = None def rename_photos_by_exif(folder): """根据照片 Exif 中的拍摄日期重命名""" if Image is None: print("请安装 Pillow: pip install Pillow") return for file in Path(folder).iterdir(): if file.suffix.lower() in ('.jpg', '.jpeg', '.png', '.tiff'): try: img = Image.open(file) exif_data = img._getexif() if exif_data: for tag_id, value in exif_data.items(): tag_name = TAGS.get(tag_id, tag_id) if tag_name == 'DateTimeOriginal': # Exif 时间格式: '2026:05:05 14:30:00' dt = datetime.strptime(value, '%Y:%m:%d %H:%M:%S') new_name = f"photo_{dt.strftime('%Y%m%d_%H%M%S')}{file.suffix}" file.rename(Path(folder) / new_name) print(f" {file.name} -> {new_name}") break except Exception as e: print(f"处理 {file.name} 时出错: {e}") # 通用元数据提取模板 def meta_template_rename(folder, template='${date}_${author}_${title}${ext}'): """从文档属性中提取元数据用于命名（需安装 python-docx 等库）""" # 示例逻辑：根据扩展名选择不同的元数据提取方式 for file in Path(folder).iterdir(): if file.is_file() and file.suffix == '.docx': try: from docx import Document doc = Document(file) props = doc.core_properties author = props.author or 'unknown' title = props.title or file.stem created = props.created or datetime.now() date_str = created.strftime('%Y%m%d') new_name = f"{date_str}_{author}_{title}{file.suffix}" # 清理不合法的文件名字符 new_name = re.sub(r'[\\/:*?"<>|]', '_', new_name) file.rename(Path(folder) / new_name) except ImportError: print("请安装 python-docx: pip install python-docx") break

import os import shutil from pathlib import Path # 按扩展名自动分类 def classify_by_extension(folder): """按文件扩展名分类到对应子目录""" # 扩展名 -> 目录名 映射 type_map = { '文档': ['.pdf', '.doc', '.docx', '.xls', '.xlsx', '.ppt', '.pptx', '.txt', '.md'], '图片': ['.jpg', '.jpeg', '.png', '.gif', '.bmp', '.svg', '.webp', '.ico'], '视频': ['.mp4', '.avi', '.mkv', '.mov', '.wmv', '.flv', '.m4v'], '音频': ['.mp3', '.wav', '.flac', '.aac', '.ogg', '.wma', '.m4a'], '压缩包': ['.zip', '.rar', '.7z', '.tar', '.gz', '.bz2'], '程序': ['.exe', '.msi', '.sh', '.bat', '.py', '.js', '.html', '.css'], } files_moved = 0 for file in Path(folder).iterdir(): if file.is_file(): suffix = file.suffix.lower() target_dir = '其他' for dir_name, extensions in type_map.items(): if suffix in extensions: target_dir = dir_name break target_path = Path(folder) / target_dir target_path.mkdir(exist_ok=True) shutil.move(str(file), str(target_path / file.name)) files_moved += 1 print(f"分类完成，共移动 {files_moved} 个文件") classify_by_extension('./downloads') # 下载目录下的文件将被分类到 文档/ 图片/ 视频/ 等子目录

# 按日期自动分类 from datetime import datetime def classify_by_date(folder, date_source='ctime', depth='month'): """按时间分类：年 / 年-月 / 年-月-日""" depth_map = { 'year': lambda dt: f"{dt.year}", 'month': lambda dt: f"{dt.year}-{dt.month:02d}", 'day': lambda dt: f"{dt.year}-{dt.month:02d}-{dt.day:02d}", } time_func = depth_map.get(depth, depth_map['month']) ts_func = { 'ctime': lambda s: s.st_ctime, 'mtime': lambda s: s.st_mtime, }.get(date_source, lambda s: s.st_ctime) for file in Path(folder).iterdir(): if file.is_file(): dt = datetime.fromtimestamp(ts_func(file.stat())) subdir = time_func(dt) target = Path(folder) / subdir target.mkdir(parents=True, exist_ok=True) shutil.move(str(file), str(target / file.name)) print(f"{file.name} -> {subdir}/") # 按修改时间分类，精确到月深度的目录 classify_by_date('./backup', date_source='mtime', depth='month') # 文件被归入: 2026-05/, 2026-04/, ... 等目录

# 按大小分类 + 自定义规则 def classify_by_size(folder): """按文件大小分类：小(<1MB) 中(1-100MB) 大(>100MB)""" size_categories = { '小文件': (0, 1 * 1024 * 1024), # 0 ~ 1MB '中文件': (1 * 1024 * 1024, 100 * 1024 * 1024), # 1MB ~ 100MB '大文件': (100 * 1024 * 1024, float('inf')), # >100MB } for file in Path(folder).iterdir(): if file.is_file(): size = file.stat().st_size for dir_name, (low, high) in size_categories.items(): if low <= size < high: target = Path(folder) / dir_name target.mkdir(exist_ok=True) shutil.move(str(file), str(target / file.name)) break # 组合分类规则 def smart_classify(folder): """智能分类：先按类型，再按日期""" type_map = { '文档': ['.pdf', '.docx', '.xlsx', '.pptx', '.txt'], '图片': ['.jpg', '.png', '.gif'], '视频': ['.mp4', '.mkv'], } for file in Path(folder).iterdir(): if file.is_file(): # 确定类型 suffix = file.suffix.lower() file_type = '其他' for t, exts in type_map.items(): if suffix in exts: file_type = t break # 确定日期 dt = datetime.fromtimestamp(file.stat().st_ctime) date_dir = f"{dt.year}-{dt.month:02d}" # 类型/日期 层级目录 target = Path(folder) / file_type / date_dir target.mkdir(parents=True, exist_ok=True) shutil.move(str(file), str(target / file.name))

import os import re import unicodedata from pathlib import Path # 通用文件名清洗函数 def sanitize_filename(filename, replacement='_'): """ 清洗文件名：去除非法字符、控制字符、多余空格 参数： filename: 原始文件名 replacement: 非法字符替换为的字符串 返回：清洗后的合法文件名 """ name, ext = os.path.splitext(filename) # 1. 去除控制字符和不可见字符 name = ''.join(c if unicodedata.category(c)[0] != 'C' else replacement for c in name) # 2. 替换 Windows 非法字符 illegal = r'[\\/:*?"<>|]' name = re.sub(illegal, replacement, name) # 3. 合并连续替换字符 name = re.sub(f'{re.escape(replacement)}+', replacement, name) # 4. 去除首尾替换字符和空格 name = name.strip(f'{replacement} ') # 5. 处理空名称 if not name: name = 'unnamed' # 6. 限制总长度 (Windows MAX_PATH 减去目录路径) max_len = 200 - len(ext) if len(name) > max_len: name = name[:max_len] return name + ext # 测试 test_names = [ 'file:report*2026?.docx', ' 有很多空格 的文件 .txt', 'a\tb\nc.txt', 'CON.txt', # Windows 保留设备名 ] for name in test_names: print(f"'{name}' -> '{sanitize_filename(name)}'")

# 统一大小写和替换空格 def normalize_case(folder, mode='lower'): """统一文件名大小写""" mode_funcs = { 'lower': lambda s: s.lower(), 'upper': lambda s: s.upper(), 'title': lambda s: s.title(), } case_func = mode_funcs.get(mode, mode_funcs['lower']) for file in Path(folder).iterdir(): if file.is_file(): name, ext = os.path.splitext(file.name) # 扩展名通常保持小写 ext = ext.lower() new_name = case_func(name) + ext if new_name != file.name: file.rename(Path(folder) / new_name) def replace_spaces(folder, replacement='_'): """将文件名中的空格替换为指定字符""" for file in Path(folder).iterdir(): if file.is_file() and ' ' in file.name: new_name = file.name.replace(' ', replacement) file.rename(Path(folder) / new_name) # 全角转半角 def fullwidth_to_halfwidth(filename): """全角字符转半角""" result = [] for char in filename: code = ord(char) # 全角字母数字转半角 if 0xFF01 <= code <= 0xFF5E: result.append(chr(code - 0xFEE0)) elif code == 0x3000: # 全角空格 result.append(chr(0x0020)) else: result.append(char) return ''.join(result) # 综合清洗管线 def pipeline_clean(folder): """执行完整清洗流程""" for file in Path(folder).iterdir(): if file.is_file(): name = file.name name = fullwidth_to_halfwidth(name) # 全角转半角 name = sanitize_filename(name) # 去非法字符 name = re.sub(r'\s+', '_', name) # 空格替换为下划线 name = re.sub(r'_{2,}', '_', name) # 合并连续下划线 name, ext = os.path.splitext(name) name = name.lower() # 统一小写 ext = ext.lower() new_name = name + ext if new_name != file.name: file.rename(Path(folder) / new_name)

# 中文乱码修复（常见场景） def fix_garbled_text(text): """尝试修复常见编码错误导致的中文乱码""" # 场景1: UTF-8 被解析为 Latin-1 try: fixed = text.encode('latin-1').decode('utf-8') if any('一' <= c <= '鿿' for c in fixed): return fixed except: pass # 场景2: GBK 被解析为 Latin-1 try: fixed = text.encode('latin-1').decode('gbk') if any('一' <= c <= '鿿' for c in fixed): return fixed except: pass return text # 批量清洗并重命名 def clean_rename_batch(folder, rules=None): """ 批量清洗重命名 rules: 需要应用的规则列表 ['lower', 'spaces', 'illegal', 'fullwidth', 'truncate'] """ default_rules = ['fullwidth', 'illegal', 'spaces', 'lower'] rules = rules or default_rules rule_funcs = { 'fullwidth': fullwidth_to_halfwidth, 'illegal': lambda s: sanitize_filename(s), 'spaces': lambda s: re.sub(r'\s+', '_', s), 'lower': lambda s: s.lower(), 'truncate': lambda s: s[:200] if len(s) > 200 else s, } for file in Path(folder).iterdir(): if file.is_file(): name = file.name for rule in rules: if rule in rule_funcs: name = rule_funcs[rule](name) if name != file.name: file.rename(Path(folder) / name) print(f" {file.name} -> {name}") clean_rename_batch('./downloads', rules=['fullwidth', 'illegal', 'spaces'])

import os import hashlib from pathlib import Path from collections import defaultdict # 单文件哈希计算 def file_hash(filepath, algorithm='md5', chunk_size=8192): """计算文件的哈希值，支持 MD5/SHA1/SHA256""" h = hashlib.new(algorithm) with open(filepath, 'rb') as f: while True: chunk = f.read(chunk_size) if not chunk: break h.update(chunk) return h.hexdigest() # 完整重复文件检测 def find_duplicates(folder, algorithm='md5'): """ 检测文件夹中的重复文件 策略：先按大小分组，再对同组文件计算哈希 """ # 第一步：按大小分组 size_map = defaultdict(list) for file in Path(folder).rglob('*'): # 递归所有子目录 if file.is_file(): size_map[file.stat().st_size].append(file) # 第二步：对大小相同的文件计算哈希 hash_map = defaultdict(list) for size, files in size_map.items(): if len(files) < 2: # 没有重复可能 continue for file in files: h = file_hash(file, algorithm) hash_map[h].append(file) # 提取真正的重复组 duplicates = {h: files for h, files in hash_map.items() if len(files) > 1} return duplicates # 使用示例 dups = find_duplicates('./downloads') if dups: print(f"发现 {len(dups)} 组重复文件：") for h, files in dups.items(): print(f"\n哈希: {h[:12]}...") for f in files: print(f" - {f} ({f.stat().st_size / 1024:.1f}KB)") else: print("未发现重复文件")

# 生成重复文件报告 def duplicate_report(folder, output='duplicate_report.txt'): """生成重复文件报告并估算可释放空间""" dups = find_duplicates(folder) total_wasted = 0 with open(output, 'w', encoding='utf-8') as f: f.write("=" * 60 + "\n") f.write("重复文件检测报告\n") f.write(f"扫描目录: {folder}\n") f.write("=" * 60 + "\n\n") for h, files in dups.items(): size = files[0].stat().st_size wasted = size * (len(files) - 1) total_wasted += wasted f.write(f"重复组 (大小: {size/1024:.1f}KB, 浪费: {wasted/1024:.1f}KB):\n") for file in files: f.write(f" {file}\n") f.write("\n") f.write("=" * 60 + "\n") f.write(f"总计: {len(dups)} 组重复, 可释放空间: {total_wasted/1024/1024:.2f}MB\n") print(f"报告已生成: {output}") print(f"可释放空间: {total_wasted/1024/1024:.2f}MB") return dups # 去重策略：保留一个，其他移动到回收区域 def remove_duplicates(folder, action='move', backup_dir='./duplicates_backup'): """执行去重操作 action: 'delete' 直接删除, 'move' 移动到备份目录, 'report' 仅报告 """ dups = duplicate_report(folder) if action == 'report': return dups backup_path = Path(backup_dir) for h, files in dups.items(): original = files[0] # 保留第一个 for dup in files[1:]: if action == 'delete': dup.unlink() print(f" 删除: {dup}") elif action == 'move': backup_path.mkdir(parents=True, exist_ok=True) import shutil shutil.move(str(dup), str(backup_path / dup.name)) print(f" 移动: {dup} -> backup/") # 安全去重（交互确认） def safe_deduplicate(folder): """交互式去重，让用户选择保留哪些""" dups = duplicate_report(folder) for h, files in dups.items(): print(f"\n重复组 ({files[0].stat().st_size/1024:.1f}KB):") for i, f in enumerate(files): print(f" [{i}] {f}") choice = input("保留哪个文件? (输入编号，默认0): ").strip() if not choice: choice = 0 else: choice = int(choice) for i, f in enumerate(files): if i != choice: f.unlink() print(f" 已删除: {f}")

# 高效的增量检测（使用缓存） import json class DuplicateCache: """使用缓存文件加速重复检测""" def __init__(self, cache_file='.dup_cache.json'): self.cache_file = cache_file self.cache = self._load_cache() def _load_cache(self): try: with open(self.cache_file, 'r') as f: return json.load(f) except: return {} def _save_cache(self): with open(self.cache_file, 'w') as f: json.dump(self.cache, f) def scan(self, folder): """增量扫描，只计算新文件和被修改的文件""" results = [] for file in Path(folder).rglob('*'): if file.is_file(): stat = file.stat() key = str(file.absolute()) cached = self.cache.get(key) # 检查文件是否被修改过 if cached and cached['mtime'] == stat.st_mtime and cached['size'] == stat.st_size: results.append((key, cached['hash'])) else: h = file_hash(file) self.cache[key] = { 'hash': h, 'mtime': stat.st_mtime, 'size': stat.st_size, } results.append((key, h)) self._save_cache() # 分析重复 hash_groups = defaultdict(list) for path, h in results: hash_groups[h].append(path) return {h: paths for h, paths in hash_groups.items() if len(paths) > 1} # 模糊相似度检测（文件名层级） def find_name_similar_duplicates(folder, threshold=0.85): """检测文件名相似的重复文件（比如不同版本的同一文件）""" from difflib import SequenceMatcher files = list(Path(folder).rglob('*')) similar_pairs = [] for i in range(len(files)): if not files[i].is_file(): continue for j in range(i + 1, len(files)): if not files[j].is_file(): continue name1, name2 = files[i].stem, files[j].stem ratio = SequenceMatcher(None, name1, name2).ratio() if ratio >= threshold: similar_pairs.append((files[i], files[j], ratio)) return similar_pairs

# 目录监控 + 自动整理 try: from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler except ImportError: print("请安装 watchdog: pip install watchdog") Observer = None class AutoOrganizeHandler(FileSystemEventHandler): """文件变化事件处理器""" def __init__(self, rules): self.rules = rules self.log = [] def on_created(self, event): if event.is_directory: return self.process_file(event.src_path) def on_modified(self, event): if event.is_directory: return self.process_file(event.src_path) def process_file(self, filepath): path = Path(filepath) if not path.exists(): return # 文件可能已被移动 # 应用整理规则 for rule in self.rules: if self.match_rule(path, rule): self.apply_rule(path, rule) break def match_rule(self, path, rule): """检查文件是否匹配规则条件""" import fnmatch # 匹配文件名模式 patterns = rule.get('patterns', ['*']) matched = any(fnmatch.fnmatch(path.name, p) for p in patterns) # 匹配大小条件 if matched and 'max_size' in rule: matched = path.stat().st_size <= rule['max_size'] return matched def apply_rule(self, path, rule): target_dir = Path(rule['target']) target_dir.mkdir(parents=True, exist_ok=True) target = target_dir / path.name import shutil shutil.move(str(path), str(target)) self.log.append({ 'time': datetime.now().isoformat(), 'source': str(path), 'target': str(target), 'rule': rule.get('name', 'unknown'), }) print(f"已整理: {path.name} -> {target_dir.name}/") # 启动监控 def start_monitor(watch_dir, rules, recursive=True): if Observer is None: return event_handler = AutoOrganizeHandler(rules) observer = Observer() observer.schedule(event_handler, watch_dir, recursive=recursive) observer.start() print(f"监控已启动: {watch_dir}") print("按 Ctrl+C 停止监控...") try: observer.join() except KeyboardInterrupt: observer.stop() print("\n监控已停止") observer.join() return event_handler # 示例规则 rules = [ {'name': '文档', 'patterns': ['*.pdf', '*.docx', '*.xlsx'], 'target': './organized/文档'}, {'name': '图片', 'patterns': ['*.jpg', '*.png', '*.gif'], 'target': './organized/图片'}, {'name': '小文件', 'patterns': ['*.zip', '*.rar'], 'max_size': 10*1024*1024, 'target': './organized/小包'}, ] # start_monitor('./watch_folder', rules)

# 配置文件驱动的整理系统 (config.json) """ config.json 示例： { "watch_dir": "./downloads", "recursive": false, "auto_clean": true, "classification_rules": [ { "name": "文档", "patterns": ["*.pdf", "*.docx", "*.txt"], "target": "./sorted/文档", "rename_template": "${date}_${name}${ext}" }, { "name": "代码", "patterns": ["*.py", "*.js", "*.html", "*.css"], "target": "./sorted/代码", "rename_template": "${name}${ext}" } ], "ignore_patterns": ["*.tmp", "*.swp", "~$*"], "log_file": "./organize_log.json", "max_log_entries": 1000 } """ import json class ConfigDrivenOrganizer: """基于配置文件的自动整理器""" def __init__(self, config_path): with open(config_path, 'r', encoding='utf-8') as f: self.config = json.load(f) self.log = [] self.undo_stack = [] def run_once(self): """执行一次整理""" watch_dir = self.config['watch_dir'] rules = self.config['classification_rules'] ignores = self.config.get('ignore_patterns', []) import fnmatch for file in Path(watch_dir).iterdir(): if file.is_file(): name = file.name # 检查忽略规则 if any(fnmatch.fnmatch(name, p) for p in ignores): continue # 查找匹配规则 for rule in rules: if any(fnmatch.fnmatch(name, p) for p in rule['patterns']): target_dir = Path(rule['target']) target_dir.mkdir(parents=True, exist_ok=True) # 应用重命名模板 template = rule.get('rename_template', '${name}${ext}') new_name = Template(template).safe_substitute({ 'date': datetime.now().strftime('%Y%m%d'), 'name': file.stem, 'ext': file.suffix, }) target = target_dir / new_name shutil.move(str(file), str(target)) # 记录撤销信息 self.undo_stack.append({ 'source': str(target), 'original': str(file), }) break def undo(self): """撤销上次操作""" for action in reversed(self.undo_stack): shutil.move(action['source'], action['original']) self.undo_stack.clear() print(f"已撤销，还原了 {len(self.undo_stack)} 个文件")

# 整理日志与撤销机制 import json from datetime import datetime class OrganizeManager: """完整的整理管理器，包含日志和撤销""" def __init__(self, log_file='organize_history.json'): self.log_file = log_file self.history = self._load_history() def _load_history(self): try: with open(self.log_file, 'r', encoding='utf-8') as f: return json.load(f) except: return {'sessions': []} def _save_history(self): with open(self.log_file, 'w', encoding='utf-8') as f: json.dump(self.history, f, ensure_ascii=False, indent=2) def organize(self, folder, rules): """执行一次整理操作并记录日志""" session = { 'id': datetime.now().strftime('%Y%m%d_%H%M%S'), 'timestamp': datetime.now().isoformat(), 'folder': folder, 'actions': [] } for rule in rules: for file in Path(folder).iterdir(): if file.is_file(): import fnmatch if any(fnmatch.fnmatch(file.name, p) for p in rule['patterns']): target = Path(rule['target']) target.mkdir(parents=True, exist_ok=True) dst = target / file.name shutil.move(str(file), str(dst)) session['actions'].append({ 'file': file.name, 'from': str(file), 'to': str(dst), 'rule': rule['name'], }) self.history['sessions'].append(session) self._save_history() print(f"整理完成，操作 {len(session['actions'])} 个文件") return session def undo_session(self, session_id): """撤销指定次整理操作""" for session in self.history['sessions']: if session['id'] == session_id: # 逆序还原 for action in reversed(session['actions']): if Path(action['to']).exists(): shutil.move(action['to'], action['from']) self.history['sessions'].remove(session) self._save_history() print(f"已撤销整理会话: {session_id}") return True print(f"未找到会话: {session_id}") return False def list_history(self): """查看整理历史""" for session in self.history['sessions']: print(f"[{session['id']}] {session['timestamp']} - {len(session['actions'])} 个文件") for a in session['actions'][:5]: print(f" {a['file']} -> {a['to']}") if len(session['actions']) > 5: print(f" ... 还有 {len(session['actions'])-5} 个文件") def schedule(self, folder, rules, interval_hours=24): """定时执行整理（运行在后台线程）""" import threading def run_loop(): while True: self.organize(folder, rules) time.sleep(interval_hours * 3600) thread = threading.Thread(target=run_loop, daemon=True) thread.start() print(f"定时整理已启动，每 {interval_hours} 小时执行一次")

# 案例一：照片整理工具 """ 功能： 1. 读取照片 Exif 信息获取拍摄日期和 GPS 2. 按年/月/日创建目录结构 3. 可选择按地点添加子分类 4. 自动去除重复照片 5. 生成整理报告 """ from pathlib import Path import shutil import hashlib from datetime import datetime class PhotoOrganizer: def __init__(self, source, dest=None): self.source = Path(source) self.dest = Path(dest) if dest else self.source / 'organized' self.stats = {'total': 0, 'organized': 0, 'duplicates': 0, 'errors': 0} self.photo_exts = {'.jpg', '.jpeg', '.png', '.tiff', '.tif', '.bmp', '.gif', '.webp', '.heic', '.raw', '.cr2'} def get_photo_date(self, photo_path): """从 Exif 或文件系统获取拍摄日期""" try: from PIL import Image from PIL.ExifTags import TAGS img = Image.open(photo_path) exif = img._getexif() if exif: for tag_id, value in exif.items(): tag = TAGS.get(tag_id) if tag == 'DateTimeOriginal': return datetime.strptime(value, '%Y:%m:%d %H:%M:%S') except: pass # 降级使用文件修改时间 mtime = photo_path.stat().st_mtime return datetime.fromtimestamp(mtime) def get_gps_location(self, photo_path): """从 Exif 提取 GPS 信息（简化版）""" try: from PIL import Image from PIL.ExifTags import TAGS, GPSTAGS img = Image.open(photo_path) exif = img._getexif() if not exif: return None gps_info = {} for tag_id, value in exif.items(): tag = TAGS.get(tag_id) if tag == 'GPSInfo': for gps_tag_id, gps_value in value.items(): gps_tag = GPSTAGS.get(gps_tag_id, gps_tag_id) gps_info[gps_tag] = gps_value if 'GPSLatitude' in gps_info and 'GPSLongitude' in gps_info: lat = sum(v[0]/v[1] for v in gps_info['GPSLatitude']) / len(gps_info['GPSLatitude']) lon = sum(v[0]/v[1] for v in gps_info['GPSLongitude']) / len(gps_info['GPSLongitude']) if gps_info.get('GPSLatitudeRef') == 'S': lat = -lat if gps_info.get('GPSLongitudeRef') == 'W': lon = -lon return f"{lat:.4f}_{lon:.4f}" except: pass return None def file_hash(self, path): h = hashlib.md5() with open(path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): h.update(chunk) return h.hexdigest() def run(self): """执行照片整理主流程""" print("开始整理照片...") hashes_seen = set() for file in self.source.iterdir(): if file.suffix.lower() not in self.photo_exts: continue self.stats['total'] += 1 try: # 去重检测 fhash = self.file_hash(file) if fhash in hashes_seen: self.stats['duplicates'] += 1 duplicate_dir = self.dest / '_重复文件' duplicate_dir.mkdir(parents=True, exist_ok=True) shutil.move(str(file), str(duplicate_dir / file.name)) continue hashes_seen.add(fhash) # 获取日期 date = self.get_photo_date(file) date_dir = f"{date.year}/{date.month:02d}/{date.day:02d}" # 获取位置 location = self.get_gps_location(file) if location: date_dir += f"_{location}" # 创建目录并移动 target_dir = self.dest / date_dir target_dir.mkdir(parents=True, exist_ok=True) shutil.move(str(file), str(target_dir / file.name)) self.stats['organized'] += 1 print(f" ✓ {file.name} -> {date_dir}/") except Exception as e: self.stats['errors'] += 1 print(f" ✗ {file.name}: {e}") self.report() def report(self): print("\n" + "=" * 40) print("照片整理完成报告") print("=" * 40) print(f"总计处理: {self.stats['total']} 张") print(f"已整理: {self.stats['organized']} 张") print(f"重复文件: {self.stats['duplicates']} 张") print(f"错误: {self.stats['errors']} 个") # 使用 org = PhotoOrganizer('./photos', './photo_library') org.run()

# 案例二：下载目录自动分类 """ 功能： 1. 监控下载目录的文件变化 2. 按文件类型自动分类到对应子目录 3. 自动清理临时文件 4. 支持自定义规则扩展 """ import time import json from pathlib import Path from collections import defaultdict class DownloadOrganizer: def __init__(self, download_dir='./downloads', config=None): self.download_dir = Path(download_dir) self.config = config or self.default_config() self.type_stats = defaultdict(int) def default_config(self): return { '分类规则': { '文档': { '扩展名': ['.pdf', '.docx', '.xlsx', '.pptx', '.txt', '.md'], '子分类': { '报表': ['*报告*', '*报表*', '*总结*'], '合同': ['*合同*', '*协议*', '*契约*'], '资料': ['*手册*', '*指南*', '*教程*'], } }, '图片': { '扩展名': ['.jpg', '.jpeg', '.png', '.gif', '.svg'], }, '视频': { '扩展名': ['.mp4', '.avi', '.mkv', '.mov'], }, '音频': { '扩展名': ['.mp3', '.wav', '.flac'], }, '压缩包': { '扩展名': ['.zip', '.rar', '.7z', '.tar', '.gz'], }, '安装程序': { '扩展名': ['.exe', '.msi', '.dmg', '.AppImage'], }, '代码': { '扩展名': ['.py', '.js', '.html', '.css', '.java', '.cpp', '.json', '.xml'], }, }, '临时文件': ['*.tmp', '*.temp', '*.swp', '~$*', '*.crdownload', '*.part'], '忽略文件': ['desktop.ini', 'thumbs.db', '.ds_store'], '自动清理临时文件': True, '按时间子分类': False, # 若启用则按日期再分子目录 } def organize(self): """执行一次整理""" import fnmatch files = [f for f in self.download_dir.iterdir() if f.is_file()] if not files: print("没有需要整理的文件") return print(f"开始整理 {len(files)} 个文件...") for file in files: # 跳过忽略文件 if any(fnmatch.fnmatch(file.name.lower(), ig.lower()) for ig in self.config['忽略文件']): continue # 检查是否为临时文件 if self.config['自动清理临时文件']: if any(fnmatch.fnmatch(file.name, p) for p in self.config['临时文件']): file.unlink() print(f" 删除临时文件: {file.name}") continue # 分类移动 suffix = file.suffix.lower() target_category = '其他' for category, rules in self.config['分类规则'].items(): if suffix in rules['扩展名']: target_category = category # 检查子分类 if '子分类' in rules: for sub_cat, patterns in rules['子分类'].items(): if any(fnmatch.fnmatch(file.stem, p) for p in patterns): target_category = f"{category}/{sub_cat}" break break target_path = self.download_dir / target_category target_path.mkdir(parents=True, exist_ok=True) # 处理同名冲突 dest = target_path / file.name counter = 1 while dest.exists(): stem = file.stem dest = target_path / f"{stem}_{counter}{suffix}" counter += 1 shutil.move(str(file), str(dest)) self.type_stats[target_category] += 1 self.print_summary() def print_summary(self): print("\n整理结果：") for category, count in sorted(self.type_stats.items()): print(f" {category}: {count} 个文件") def start_watch(self, interval=10): """持续监控模式""" print(f"启动下载目录监控（扫描间隔: {interval}秒）...") print("按 Ctrl+C 停止") try: while True: self.organize() time.sleep(interval) except KeyboardInterrupt: print("\n监控已停止") # 使用 org = DownloadOrganizer('./downloads') org.organize() # 持续监控模式 # org.start_watch(interval=30)

# 案例三：项目文件规范化工具 """ 功能： 1. 扫描项目目录中所有文件 2. 统一为蛇形命名（snake_case） 3. 按文件类型整理到规范目录结构 4. 生成项目文件清单 """ import os import re from pathlib import Path class ProjectNormalizer: """项目文件规范化工具""" # 推荐的 Python 项目目录结构 PROJECT_STRUCTURE = { 'src': ['.py', '.js', '.ts', '.go', '.rs'], 'tests': ['test_*.py', '*_test.py', 'test_*.js'], 'docs': ['.md', '.rst', '.txt', '.pdf', '.docx'], 'config': ['.json', '.yaml', '.yml', '.toml', '.ini', '.cfg'], 'data': ['.csv', '.xlsx', '.json', '.xml', '.db', '.sqlite'], 'scripts': ['.sh', '.bat', '.ps1'], 'assets': ['.png', '.jpg', '.svg', '.ico', '.css', '.scss'], } def __init__(self, project_root, dry_run=True): self.root = Path(project_root) self.dry_run = dry_run # 默认为模拟模式 self.log = [] def to_snake_case(self, name): """文件名转蛇形命名""" name_no_ext, ext = os.path.splitext(name) # 驼峰转蛇形 s1 = re.sub(r'(.)([A-Z][a-z]+)', r'\1_\2', name_no_ext) s2 = re.sub(r'([a-z0-9])([A-Z])', r'\1_\2', s1) # 替换所有非字母数字字符为下划线 s3 = re.sub(r'[^\w]', '_', s2) # 合并下划线并转小写 s4 = re.sub(r'_+', '_', s3).lower() # 去除首尾下划线 s4 = s4.strip('_') return s4 + ext.lower() def normalize(self): """执行规范化""" print(f"{'[模拟] ' if self.dry_run else ''}开始规范化项目: {self.root}") all_files = [f for f in self.root.rglob('*') if f.is_file()] for file in all_files: # 跳过 .git 等隐藏目录 rel = file.relative_to(self.root) if any(p.startswith('.') for p in rel.parts): continue # 标准化文件名 new_name = self.to_snake_case(file.name) if new_name != file.name: new_path = file.with_name(new_name) if not self.dry_run: file.rename(new_path) self.log.append(f" {file.name} -> {new_name}") file = new_path # 更新路径用于后续操作 # 生成文件清单报告 self.generate_inventory() def organize_to_structure(self): """按标准目录结构整理（可选）""" if self.dry_run: print("[模拟] 创建标准目录结构并移动文件...") else: for dir_name in self.PROJECT_STRUCTURE: (self.root / dir_name).mkdir(exist_ok=True) for file in self.root.iterdir(): if not file.is_file(): continue suffix = file.suffix.lower() target_dir = None for dir_name, exts in self.PROJECT_STRUCTURE.items(): if suffix in exts: target_dir = dir_name break if target_dir and target_dir != 'tests': # 测试文件特殊处理 target = self.root / target_dir / file.name if not self.dry_run: shutil.move(str(file), str(target)) self.log.append(f" {file.name} -> {target_dir}/") def generate_inventory(self): """生成项目文件清单""" inventory_file = self.root / 'FILE_INVENTORY.md' if self.dry_run: print(f"[模拟] 生成文件清单: {inventory_file}") return lines = ["# 项目文件清单", "", f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", ""] tree = {} for file in sorted(self.root.rglob('*')): if file.is_file(): rel = file.relative_to(self.root) parts = rel.parts current = tree for part in parts[:-1]: current = current.setdefault(part, {}) current[parts[-1]] = file.stat().st_size def write_tree(d, indent=0): for name, value in sorted(d.items()): if isinstance(value, dict): lines.append(" " * indent + f"- **{name}/**") write_tree(value, indent + 1) else: lines.append(" " * indent + f"- {name} ({value/1024:.1f}KB)") write_tree(tree) with open(inventory_file, 'w', encoding='utf-8') as f: f.write('\n'.join(lines)) print(f"文件清单已生成: {inventory_file}") def undo(self): """撤销重命名操作""" if not self.log: print("没有可撤销的操作") return for entry in reversed(self.log): match = re.match(r'\s*(\S+) -> (\S+)', entry) if match: old, new = match.groups() old_path = self.root / old new_path = self.root / new if new_path.exists(): if not self.dry_run: new_path.rename(old_path) print(f" {new} -> {old}") # 使用 normalizer = ProjectNormalizer('./my_project', dry_run=True) normalizer.normalize() print(f"\n计划执行 {len(normalizer.log)} 项修改") print("确认后设置 dry_run=False 执行实际操作")