def safe_batch_rename(directory, pattern, replacement, backup_dir='rename_backup'):
"""
安全批量重命名：带备份和回滚

Args:
directory: 目标文件夹
pattern: 正则表达式匹配模式
replacement: 新文件名模板（支持{date}等占位符）
backup_dir: 备份文件夹名
"""
dir_path = Path(directory)
backup_path = dir_path / backup_dir

# 步骤1：创建备份
print(f"[1/4] 创建备份到 {backup_path}")
shutil.copytree(dir_path, backup_path, dirs_exist_ok=True)

# 步骤2：收集所有要改的文件
rename_map = {} # 原路径 -> 新路径
errors = []

for file_path in dir_path.glob('*'):
if not file_path.is_file() or file_path.name.startswith('.'):
continue

old_name = file_path.name
# 支持简单替换和日期占位符
if '{date}' in replacement:
# 尝试从文件名提取日期
date_match = re.search(r'(\d{4}[-_]\d{2}[-_]\d{2})', old_name)
if date_match:
date_str = date_match.group(1).replace('_', '-')
else:
date_str = datetime.now().strftime('%Y-%m-%d')
new_name = replacement.replace('{date}', date_str)
else:
new_name = re.sub(pattern, replacement, old_name)

if new_name == old_name:
continue

new_path = dir_path / new_name
rename_map[file_path] = new_path

# 步骤3：检查冲突
print(f"[2/4] 发现 {len(rename_map)} 个文件需要改名")
new_names = [p.name for p in rename_map.values()]
duplicates = set([n for n in new_names if new_names.count(n) > 1])
if duplicates:
print(f"⚠️ 发现命名冲突: {duplicates}")
print("正在自动添加序号...")
# 冲突处理：自动添加(1),(2)等
counter = {}
for old_path, new_path in rename_map.items():
if new_path.name in duplicates:
counter[new_path.name] = counter.get(new_path.name, 0) + 1
stem = new_path.stem
suffix = new_path.suffix
rename_map[old_path] = new_path.with_name(f"{stem}_{counter[new_path.name]}{suffix}")

# 步骤4：执行改名（带异常回滚）
print(f"[3/4] 正在执行改名...")
renamed_count = 0
try:
for old_path, new_path in rename_map.items():
if new_path.exists():
print(f"⚠️ 跳过 {new_path.name}（已存在）")
continue
old_path.rename(new_path)
renamed_count += 1
print(f"[4/4] 成功改名 {renamed_count} 个文件")
except Exception as e:
print(f"❌ 出错: {e}")
print("正在回滚...")
# 回滚：从备份恢复
for old_path, new_path in rename_map.items():
if new_path.exists():
new_path.unlink()
shutil.copytree(backup_path, dir_path, dirs_exist_ok=True)
print("已回滚到原始状态")

使用示例

这个设计真的反人类？不，是现实太反人类。你永远猜不到文件名里会出现什么奇葩字符——空格、中文括号、甚至emoji。上面代码里的正则已经处理了大部分情况，但如果你遇到文件名带换行符的，建议先手动清理一下。

另一个坑：批量格式转换。

def convert_csv_to_excel_chunked(csv_dir, output_dir, chunk_size=10000):
"""
分块转换CSV到Excel，带进度条和错误日志

Args:
csv_dir: CSV文件夹路径
output_dir: 输出Excel文件夹
chunk_size: 每块行数（根据内存调整）
"""
csv_path = Path(csv_dir)
output_path = Path(output_dir)
output_path.mkdir(exist_ok=True)

error_log = []
start_time = time.time()

for csv_file in csv_path.glob('*.csv'):
print(f"正在处理: {csv_file.name}")
try:
excel_file = output_path / f"{csv_file.stem}.xlsx"

# 分块读取 + 流式写入
reader = pd.read_csv(csv_file, chunksize=chunk_size)
first_chunk = True

for chunk in reader:
if first_chunk:
# 创建Excel文件，写入表头
with pd.ExcelWriter(excel_file, engine='openpyxl', mode='w') as writer:
chunk.to_excel(writer, sheet_name='Sheet1', index=False)
first_chunk = False
else:
# 追加数据
with pd.ExcelWriter(excel_file, engine='openpyxl', mode='a') as writer:
writer.book = openpyxl.load_workbook(excel_file)
start_row = writer.sheets['Sheet1'].max_row + 1
chunk.to_excel(writer, sheet_name='Sheet1',
index=False, startrow=start_row, header=False)

# 格式化：自动调整列宽
wb = openpyxl.load_workbook(excel_file)
ws = wb.active
for column in ws.columns:
max_length = 0
column_letter = column[0].column_letter
for cell in column:
try:
if len(str(cell.value)) > max_length:
max_length = len(str(cell.value))
except:
pass
adjusted_width = min(max_length + 2, 50)
ws.column_dimensions[column_letter].width = adjusted_width
wb.save(excel_file)

elapsed = time.time() - start_time
print(f" ✓ 完成 (耗时 {elapsed:.1f}秒)")

except Exception as e:
error_log.append(f"{csv_file.name}: {str(e)}")
print(f" ✗ 失败: {e}")

# 输出错误日志
if error_log:
log_file = output_path / 'conversion_errors.txt'
with open(log_file, 'w', encoding='utf-8') as f:
f.write('\n'.join(error_log))
print(f"\n⚠️ 有 {len(error_log)} 个文件转换失败，详情见 {log_file}")

total_time = time.time() - start_time
print(f"\n全部完成！总耗时 {total_time:.1f}秒")

使用示例

还有个技巧：批量数据提取。

为什么要写这个？ 老板扔给我200个日志文件，说"把里面所有的IP地址和错误码提取出来"。手动复制粘贴？我只能说"老板你认真的吗？"

这个需求的核心是：不打开每个文件，直接用正则扫描。我优化后的版本支持多线程，从原来遍历200个文件需要5分钟，降到15秒。

def extract_pattern_from_files(directory, patterns, file_pattern='*.log', max_workers=8):
"""
多线程批量提取文件中的数据

Args:
directory: 目标文件夹
patterns: 正则表达式字典，如 {'ip': r'\d+\.\d+\.\d+\.\d+', 'error': r'ERROR:\s*\w+'}
file_pattern: 文件匹配模式，如 '*.log', '*.txt'
max_workers: 线程数
Returns:
dict: 按文件分组的提取结果
"""
dir_path = Path(directory)
files = list(dir_path.glob(file_pattern))

if not files:
print(f"没有找到匹配 {file_pattern} 的文件")
return {}

print(f"找到 {len(files)} 个文件，使用 {max_workers} 个线程处理...")

results = defaultdict(lambda: defaultdict(list))

def process_file(file_path):
"""处理单个文件，返回文件路径和提取结果"""
file_results = {}
try:
content = file_path.read_text(encoding='utf-8', errors='ignore')
for key, pattern in patterns.items():
matches = re.findall(pattern, content)
if matches:
file_results[key] = matches
return str(file_path), file_results
except Exception as e:
return str(file_path), {'error': str(e)}

# 多线程执行
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {executor.submit(process_file, f): f for f in files}

for i, future in enumerate(as_completed(futures), 1):
file_path, file_results = future.result()
if file_results:
results[file_path] = file_results

# 简单进度显示
if i % 20 == 0:
print(f" 进度: {i}/{len(files)}")

return dict(results)

使用示例

patterns = {
'ip': r'\b(?:\d{1,3}\.){3}\d{1,3}\b',
'error_code': r'ERROR:\s*([A-Z]+\d+)',
'timestamp': r'\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}',
'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
}

extracted_data = extract_pattern_from_files(
directory='./logs',
patterns=patterns,
file_pattern='*.txt',
max_workers=12
)

保存结果

output_file = Path('./extracted_results.json')
output_file.write_text(json.dumps(extracted_data, indent=2, ensure_ascii=False))
print(f"\n结果已保存到 {output_file}")

统计信息

（核心配图：一张对比表格，左边是"手动处理"——200个文件、5小时、疲劳值爆表；右边是"Python处理"——15秒、一键搞定、咖啡在手。对比数据极具冲击力）

说到多线程，有个细节：如果你处理的文件都在同一个硬盘上，线程数太多反而会变慢，因为硬盘读写是瓶颈。我测试过，SSD下8-12个线程最优，机械硬盘4-6个就够了。上面的代码我默认设了8，你可以根据实际情况调。

总结一下，你可以立刻用的三个点：

（总结前配图：一张Python代码截图，旁边是一杯咖啡和一台笔记本电脑，桌面上有"自动化"标签的文件夹，体现"高效工作"的氛围）

这些技巧看起来简单，但每一个都藏着血泪教训。比如那个回滚机制——我第一次写批量重命名时，中途有个文件被占用报错，结果50个文件改了30个就停了，剩下20个名字对不上号，手动了一个小时才修好。