Python 自動化批次任務

簡介

在日常開發與維運工作中，大量重複性、時間敏感的工作往往是最容易出錯、最耗費人力的部份。透過 Python 撰寫批次腳本，我們可以把手動操作自動化，讓工作流程更可靠、更快速，同時也減少人為失誤的機會。

自動化批次任務不僅適用於系統管理員的例行維護、資料科學家的資料前處理，甚至是一般開發者在 CI/CD 流程中需要的檔案搬移、測試執行等工作。掌握這些技巧，能讓 「寫程式」 變成 「解決問題」 的利器，而不是單純的工具使用者。

核心概念

1. 什麼是批次任務？

批次任務（Batch Job）指的是在無需使用者即時互動的情況下，一次性或定時執行的一系列指令或程式碼。常見的觸發方式有：

觸發方式	說明
手動執行 (`python script.py`)	適合一次性或測試用
排程工具 (`cron`, `Task Scheduler`)	讓腳本在固定時間自動跑
事件驅動 (`inotify`, `watchdog`)	檔案變動或訊息到達時觸發

2. 基本檔案與目錄操作

自動化的第一步往往是讀寫檔案、遍歷目錄。Python 標準庫 os、pathlib、shutil 已提供完整支援。

# 例 1：使用 pathlib 列出目錄下的所有 .log 檔案
from pathlib import Path

log_dir = Path("/var/log/myapp")
for log_file in log_dir.glob("*.log"):
    print(log_file.name)          # 印出檔名

# 例 2：搬移舊檔至備份資料夾，若目標不存在則自動建立
import shutil
from pathlib import Path

src = Path("data/raw")
dst = Path("data/archive")
dst.mkdir(parents=True, exist_ok=True)

for file in src.iterdir():
    if file.suffix == ".csv":
        shutil.move(str(file), dst / file.name)

3. 子程序與平行執行

大量檔案或網路請求時，單執行緒往往太慢。concurrent.futures 提供簡潔的執行緒池或工作池介面。

# 例 3：使用 ThreadPoolExecutor 同時壓縮多個檔案
import gzip
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor

def compress_file(file_path: Path):
    with file_path.open('rb') as f_in, gzip.open(str(file_path) + '.gz', 'wb') as f_out:
        f_out.writelines(f_in)

files = list(Path("logs").glob("*.log"))
with ThreadPoolExecutor(max_workers=4) as executor:
    executor.map(compress_file, files)

4. 讀寫 CSV / JSON / Excel

自動化常需要處理結構化資料。csv、json、pandas（對於較大或複雜的資料）都是好幫手。

# 例 4：使用 pandas 讀取 CSV、過濾資料、寫回新檔
import pandas as pd

df = pd.read_csv("sales_2024.csv")
# 只保留 2024 年第一季的記錄
q1 = df[(df["date"] >= "2024-01-01") & (df["date"] <= "2024-03-31")]
q1.to_csv("sales_Q1_2024.csv", index=False)

5. 與外部指令結合

有時候需要呼叫系統指令或其他程式，subprocess 能夠安全地執行並取得結果。

# 例 5：呼叫 rsync 同步遠端備份，捕捉錯誤訊息
import subprocess

cmd = ["rsync", "-avz", "/data/", "user@remote:/backup/"]
result = subprocess.run(cmd, capture_output=True, text=True)

if result.returncode == 0:
    print("同步完成")
else:
    print("發生錯誤:", result.stderr)

常見陷阱與最佳實踐

陷阱	為什麼會發生	解決方式
硬編碼路徑	程式在不同環境會找不到檔案	使用 `pathlib.Path` 並透過環境變數或設定檔注入路徑
未捕捉例外	批次任務在單一失敗時會整個中斷	用 `try/except` 包住關鍵區塊，並寫入錯誤日誌
同步 I/O 阻塞	大量檔案操作時速度極慢	采用 `ThreadPoolExecutor` 或 `asyncio`（對 I/O 密集型）
日誌不完整	難以追蹤哪一步失敗	使用 `logging` 模組，設定檔案與 console 兩種輸出
排程時間衝突	多個 cron 任務同時執行互相干擾	為每個任務加上鎖檔 (`flock`) 或使用 `systemd` 的 `OnFailure` 機制

最佳實踐

保持腳本可參數化：使用 argparse 讓使用者自行指定來源、目的地或日期。
統一日誌格式：logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')。
測試與回滾：在正式排程前，先在測試環境跑一次；若失敗，確保有備份或回滾機制。
使用虛擬環境：venv 或 conda 能避免套件版本衝突。
文件化腳本：在檔案開頭寫明功能、參數說明與執行範例，方便後續維護。

實際應用場景

場景	需求	可能的腳本結構
每日資料匯入	早上 02:00 從 SFTP 拉下 CSV、清洗、寫入資料庫	`download.py → clean.py → load.py`（使用 `subprocess` 串接）
系統日誌輪替	每週壓縮舊 log、搬至遠端備份、刪除本地 30 天前檔案	`rotate_logs.py`（`glob` + `gzip` + `shutil`）
CI/CD 測試	每次 push 後自動跑單元測試、產生測試報告、發送 Slack 通知	`run_tests.py`（`unittest` + `subprocess` + `requests`）
圖片批次處理	把大量 PNG 轉成 JPEG、加上水印、上傳至雲端	`process_images.py`（`Pillow` + `ThreadPoolExecutor` + `boto3`）
報表自動寄送	每月第一個工作天產生 PDF 報表，寄給主管	`generate_report.py`（`pandas` + `matplotlib` + `smtplib`）

總結

自動化批次任務是 提升工作效率、降低錯誤率 的關鍵技術。透過 Python 豐富的標準庫與第三方套件，我們可以輕鬆完成檔案操作、平行處理、資料清洗與系統整合等工作。

在實作時，務必要注意 參數化、例外處理、日誌紀錄，並配合排程工具或容器化部署，以確保腳本在不同環境下都能穩定執行。只要掌握本文的核心概念與最佳實踐，從簡單的檔案搬移到複雜的資料管道，都能以 Python 迅速構建可靠的自動化解決方案。祝你寫腳本順利、批次任務無憂！