Python文件批量處理(重命名/備份/同步運維)的實戰(zhàn)指南
?老王在一家小公司管服務(wù)器。每天最煩的事,就是開發(fā)同事丟來一堆日志文件,文件名亂七八糟——有的叫log1.txt,有的叫1212.log,還有的直接叫新建文本文檔(1).log。更糟的是,每周五要手動備份一遍配置文件,還得確保備份目錄和源目錄同步。老王決定寫幾個Python腳本把這些破事自動化。
場景一:批量重命名——讓文件名規(guī)規(guī)矩矩
先拿日志文件開刀。老王想把所有.log文件改成2024-12-01_001.log這種格式,按修改時間排序。
import os
from datetime import datetime
def rename_logs(folder_path):
log_files = [f for f in os.listdir(folder_path) if f.endswith('.log')]
for index, filename in enumerate(log_files, 1):
old_path = os.path.join(folder_path, filename)
# 獲取修改時間
mtime = os.path.getmtime(old_path)
date_str = datetime.fromtimestamp(mtime).strftime('%Y-%m-%d')
new_name = f"{date_str}_{index:03d}.log"
new_path = os.path.join(folder_path, new_name)
os.rename(old_path, new_path)
print(f"{filename} -> {new_name}")
rename_logs('/var/logs')
運行一次,新建文本文檔(1).log變成了2024-12-15_005.log,一目了然。
但os.rename有個坑:目標文件已存在時會報錯。加上覆蓋檢查更穩(wěn)妥:
def safe_rename(old_path, new_path):
if os.path.exists(new_path):
base, ext = os.path.splitext(new_path)
counter = 1
while os.path.exists(f"{base}_{counter}{ext}"):
counter += 1
new_path = f"{base}_{counter}{ext}"
os.rename(old_path, new_path)
return new_path
更靈活的模式匹配:用glob模塊按規(guī)則篩選文件,比如只改report_*.xlsx:
import glob
for filepath in glob.glob('/data/report_*.xlsx'):
dirname, filename = os.path.split(filepath)
# 提取月份,比如 report_202412.xlsx -> 202412
month = filename.split('_')[1].split('.')[0]
new_name = f"財務(wù)報告_{month}.xlsx"
os.rename(filepath, os.path.join(dirname, new_name))
正則表達式批量替換:某些文件命名有規(guī)律但混亂,比如IMG_001 (1).jpg、IMG_001 (2).jpg這種括號空格讓人頭疼。
import re
def clean_filenames(folder, pattern, replacement):
for filename in os.listdir(folder):
new_name = re.sub(pattern, replacement, filename)
if new_name != filename:
old_path = os.path.join(folder, filename)
new_path = os.path.join(folder, new_name)
os.rename(old_path, new_path)
print(f"修正: {filename} -> {new_name}")
# 去掉文件名中的空格和括號
clean_filenames('./photos', r'[(\s)]', '')
# IMG_001 (1).jpg -> IMG_0011.jpg (雖然不太完美,但至少沒空格了)
場景二:智能備份——別每次都全量復(fù)制
老王之前用cp -r全量備份,幾十G的配置文件越來越慢。他需要增量備份——只復(fù)制改動過的文件。
import shutil
import hashlib
def md5_file(filepath):
"""計算文件MD5,用于判斷內(nèi)容是否變化"""
hash_md5 = hashlib.md5()
with open(filepath, 'rb') as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
def incremental_backup(src_dir, dst_dir, state_file='backup_state.json'):
import json
# 加載上次備份的狀態(tài)
if os.path.exists(state_file):
with open(state_file, 'r') as f:
last_state = json.load(f)
else:
last_state = {}
new_state = {}
copied_count = 0
for root, dirs, files in os.walk(src_dir):
rel_path = os.path.relpath(root, src_dir)
dst_root = os.path.join(dst_dir, rel_path)
os.makedirs(dst_root, exist_ok=True)
for file in files:
src_file = os.path.join(root, file)
dst_file = os.path.join(dst_root, file)
file_md5 = md5_file(src_file)
file_key = os.path.relpath(src_file, src_dir)
new_state[file_key] = file_md5
# 檢查是否需要備份
if file_key not in last_state or last_state[file_key] != file_md5:
shutil.copy2(src_file, dst_file) # copy2保留元數(shù)據(jù)
copied_count += 1
print(f"備份: {file_key}")
# 保存新狀態(tài)
with open(state_file, 'w') as f:
json.dump(new_state, f, indent=2)
print(f"完成,共備份 {copied_count} 個文件")
incremental_backup('/etc/nginx', '/backup/nginx')
第一次運行會全量備份,第二次只復(fù)制改動過的文件。狀態(tài)文件backup_state.json記錄了每個文件的MD5,拿掉它就能強制全量。
定時自動備份:結(jié)合schedule庫設(shè)置每天凌晨2點執(zhí)行。
import schedule
import time
def auto_backup_job():
print(f"{datetime.now()} 開始自動備份")
incremental_backup('/data/mysql', '/backup/mysql')
print("備份完成")
schedule.every().day.at("02:00").do(auto_backup_job)
while True:
schedule.run_pending()
time.sleep(60)
場景三:目錄同步——像rsync一樣工作
老王還負責(zé)維護兩臺服務(wù)器之間的文件同步。rsync很好用,但某些環(huán)境不允許裝額外軟件,自己寫一個輕量版。
import filecmp
import shutil
def sync_dirs(src, dst, delete_extra=False):
"""
同步目錄,dst會變成和src一模一樣
delete_extra=True時刪除dst中多余的文件
"""
if not os.path.exists(dst):
os.makedirs(dst)
# 比較兩個目錄
comparison = filecmp.dircmp(src, dst)
# 復(fù)制src有但dst沒有的
for file in comparison.left_only:
src_path = os.path.join(src, file)
dst_path = os.path.join(dst, file)
if os.path.isdir(src_path):
shutil.copytree(src_path, dst_path)
else:
shutil.copy2(src_path, dst_path)
print(f"新增: {file}")
# 復(fù)制有更新的
for file in comparison.diff_files:
src_path = os.path.join(src, file)
dst_path = os.path.join(dst, file)
shutil.copy2(src_path, dst_path)
print(f"更新: {file}")
# 刪除dst多余的
if delete_extra:
for file in comparison.right_only:
dst_path = os.path.join(dst, file)
if os.path.isdir(dst_path):
shutil.rmtree(dst_path)
else:
os.remove(dst_path)
print(f"刪除: {file}")
# 遞歸處理子目錄
for common_dir in comparison.common_dirs:
sync_dirs(
os.path.join(src, common_dir),
os.path.join(dst, common_dir),
delete_extra
)
sync_dirs('/home/user/project', '/backup/project', delete_extra=True)
filecmp.dircmp一次性告訴你三類差異:left_only(源獨有)、right_only(目標獨有)、diff_files(內(nèi)容不同)。遞歸調(diào)用就能同步整個目錄樹。
實時同步:監(jiān)聽文件系統(tǒng)事件,文件一改動立刻同步。
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SyncHandler(FileSystemEventHandler):
def __init__(self, src, dst):
self.src = src
self.dst = dst
def on_modified(self, event):
if not event.is_directory:
rel_path = os.path.relpath(event.src_path, self.src)
dst_path = os.path.join(self.dst, rel_path)
os.makedirs(os.path.dirname(dst_path), exist_ok=True)
shutil.copy2(event.src_path, dst_path)
print(f"實時同步: {rel_path}")
observer = Observer()
handler = SyncHandler('/watch/src', '/watch/dst')
observer.schedule(handler, '/watch/src', recursive=True)
observer.start()
運行后,在/watch/src里新建或修改任何文件,都會瞬間復(fù)制到/watch/dst。適合配置文件熱同步場景。
場景四:批量清理舊文件
日志文件占滿磁盤是家常便飯。寫個腳本自動刪除30天前的文件。
import time
def clean_old_files(folder, days=30, pattern='*'):
now = time.time()
cutoff = now - (days * 86400)
for root, dirs, files in os.walk(folder):
for file in files:
if not glob.fnmatch.fnmatch(file, pattern):
continue
filepath = os.path.join(root, file)
mtime = os.path.getmtime(filepath)
if mtime < cutoff:
os.remove(filepath)
print(f"刪除舊文件: {filepath} ({datetime.fromtimestamp(mtime).date()})")
# 刪除空目錄
for dir in dirs:
dirpath = os.path.join(root, dir)
try:
os.rmdir(dirpath)
print(f"刪除空目錄: {dirpath}")
except OSError:
pass
clean_old_files('/var/log/nginx', days=7, pattern='*.log*')
更安全的干跑模式:先預(yù)覽不執(zhí)行,確認無誤再真正刪除。
def dry_run_clean(folder, days=30):
now = time.time()
cutoff = now - (days * 86400)
to_delete = []
for root, dirs, files in os.walk(folder):
for file in files:
filepath = os.path.join(root, file)
if os.path.getmtime(filepath) < cutoff:
to_delete.append(filepath)
print(f"將刪除 {len(to_delete)} 個文件:")
for path in to_delete[:10]: # 只顯示前10個
print(f" {path}")
if input("確認刪除?(y/n): ").lower() == 'y':
for path in to_delete:
os.remove(path)
print("刪除完成")
踩坑與進階技巧
文件名編碼問題:Windows和Linux的文件名編碼不同??缙脚_時用os.fsencode和os.fsdecode處理。
def safe_listdir(path):
try:
return os.listdir(path)
except UnicodeEncodeError:
return [os.fsdecode(f) for f in os.listdir(os.fsencode(path))]
大量文件性能優(yōu)化:os.listdir在幾十萬文件的目錄里很慢,用scandir代替。
with os.scandir('/big_folder') as entries:
for entry in entries:
if entry.is_file() and entry.name.endswith('.log'):
print(entry.name, entry.stat().st_size)
scandir一次性返回文件屬性,避免額外調(diào)用stat,速度提升2-5倍。
移動而不是復(fù)制:備份到同一磁盤時,移動文件比復(fù)制快得多。
def move_to_archive(src, dst):
os.makedirs(dst, exist_ok=True)
for filename in os.listdir(src):
shutil.move(os.path.join(src, filename), os.path.join(dst, filename))
日志記錄:所有操作記下來,出問題能追溯。
import logging
logging.basicConfig(
filename='file_ops.log',
level=logging.INFO,
format='%(asctime)s - %(message)s'
)
def log_op(op, src, dst=None):
msg = f"{op}: {src}"
if dst:
msg += f" -> {dst}"
logging.info(msg)
print(msg)
一鍵搞定:綜合運維腳本
把常用功能打包成命令行工具:
import argparse
def main():
parser = argparse.ArgumentParser(description='文件批量處理工具')
subparsers = parser.add_subparsers(dest='command')
rename_parser = subparsers.add_parser('rename')
rename_parser.add_argument('folder')
rename_parser.add_argument('--pattern', default='*')
rename_parser.add_argument('--template', default='{date}_{index}')
backup_parser = subparsers.add_parser('backup')
backup_parser.add_argument('src')
backup_parser.add_argument('dst')
backup_parser.add_argument('--incremental', action='store_true')
sync_parser = subparsers.add_parser('sync')
sync_parser.add_argument('src')
sync_parser.add_argument('dst')
sync_parser.add_argument('--delete', action='store_true')
clean_parser = subparsers.add_parser('clean')
clean_parser.add_argument('folder')
clean_parser.add_argument('--days', type=int, default=30)
args = parser.parse_args()
if args.command == 'rename':
rename_files(args.folder, args.pattern, args.template)
elif args.command == 'backup':
if args.incremental:
incremental_backup(args.src, args.dst)
else:
shutil.copytree(args.src, args.dst)
elif args.command == 'sync':
sync_dirs(args.src, args.dst, args.delete)
elif args.command == 'clean':
clean_old_files(args.folder, args.days)
# 使用示例:
# python file_tool.py rename ./logs --pattern "*.log" --template "{date}_{index}"
# python file_tool.py backup /data /backup --incremental
# python file_tool.py sync /project /backup/project --delete
老王把這些腳本整合起來,設(shè)置好定時任務(wù)?,F(xiàn)在每天早上一到公司,檢查一下日志文件有沒有按規(guī)范命名,備份有沒有成功同步。之前每周五加班備份的日子一去不返,他泡杯茶就能開始一天的工作。
?以上就是Python文件批量處理(重命名/備份/同步運維)的實戰(zhàn)指南的詳細內(nèi)容,更多關(guān)于Python文件批量處理的資料請關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
解決pycharm中opencv-python導(dǎo)入cv2后無法自動補全的問題(不用作任何文件上的修改)
這篇文章主要介紹了解決pycharm中opencv-python導(dǎo)入cv2后無法自動補全的問題(不用作任何文件上的修改),本文給大家介紹的非常詳細,對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-03-03
Python將Markdown轉(zhuǎn)換為HTML的主流實現(xiàn)方法對比
在日常的技術(shù)寫作和文檔管理中,Markdown 憑借其簡潔的語法成為眾多開發(fā)者的首選,本文將介紹使用 Python 將 Markdown 轉(zhuǎn)換為 HTML 的常用方法,分別適用于不同的使用場景,大家可以根據(jù)需要進行選擇2026-04-04
Python數(shù)據(jù)分析神器DuckDB保姆級使用入門指南
DuckDB被譽為數(shù)據(jù)分析界的SQLite,以安裝極其簡單、運行速度飛快、直接查詢文件而著稱,本文將帶你從零開始,掌握這個?Python?辦公自動化領(lǐng)域的新晉頂流吧2026-01-01
python獲取響應(yīng)某個字段值的3種實現(xiàn)方法
這篇文章主要介紹了python獲取響應(yīng)某個字段值的3種實現(xiàn)方法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2020-04-04

