はじめに
「この作業、毎日やるの面倒だな…」
そんなときこそPythonの出番。この記事では、実際に動作確認済みの自動化スクリプト集を紹介するよ。
1. ファイル整理(拡張子ごとにフォルダ分け)
ダウンロードフォルダがカオスになっていませんか?
import shutil
from pathlib import Path
def organize_files(source_dir: Path, dry_run: bool = True):
"""拡張子ごとにファイルを整理"""
extension_folders = {
'.jpg': 'Images', '.jpeg': 'Images', '.png': 'Images', '.gif': 'Images',
'.pdf': 'Documents', '.doc': 'Documents', '.docx': 'Documents', '.txt': 'Documents',
'.mp3': 'Music', '.wav': 'Music', '.flac': 'Music',
'.mp4': 'Videos', '.avi': 'Videos', '.mkv': 'Videos',
'.py': 'Code', '.js': 'Code', '.html': 'Code', '.css': 'Code',
'.zip': 'Archives', '.rar': 'Archives', '.7z': 'Archives',
}
for file_path in source_dir.iterdir():
if file_path.is_file():
ext = file_path.suffix.lower()
folder_name = extension_folders.get(ext, 'Others')
dest_folder = source_dir / folder_name
if dry_run:
print(f" {file_path.name} → {folder_name}/")
else:
dest_folder.mkdir(exist_ok=True)
shutil.move(str(file_path), str(dest_folder / file_path.name))
# 使用例(まず dry_run=True で確認!)
organize_files(Path("~/Downloads").expanduser(), dry_run=True)
実行結果:
archive.zip → Archives/
photo.jpg → Images/
report.pdf → Documents/
script.py → Code/
song.mp3 → Music/
2. 重複ファイル検出
ディスク容量を圧迫している重複ファイルを見つけます。
import hashlib
from pathlib import Path
def find_duplicates(directory: Path) -> dict[str, list[Path]]:
"""MD5ハッシュで重複ファイルを検出"""
hash_map: dict[str, list[Path]] = {}
for file_path in directory.rglob("*"): # 再帰的に検索
if file_path.is_file():
try:
file_hash = hashlib.md5(file_path.read_bytes()).hexdigest()
hash_map.setdefault(file_hash, []).append(file_path)
except PermissionError:
continue
# 重複(2つ以上)のみ返す
return {h: paths for h, paths in hash_map.items() if len(paths) > 1}
# 使用例
duplicates = find_duplicates(Path("./"))
for hash_val, paths in duplicates.items():
print(f"重複発見 (hash: {hash_val[:8]}...):")
for p in paths:
print(f" - {p}")
実行結果:
重複発見 (hash: 793953ee...):
- dup_demo\file1.txt
- dup_demo\file2.txt
3. ログファイル解析
大量のログからエラーだけ抽出したい。
def analyze_log(log_path: Path) -> dict:
"""ログレベル別にカウント"""
levels = {"ERROR": 0, "WARNING": 0, "INFO": 0, "DEBUG": 0}
errors = []
for line in log_path.read_text().split("\n"):
for level in levels:
if level in line:
levels[level] += 1
if level == "ERROR":
errors.append(line.strip())
break
return {"counts": levels, "errors": errors}
# 使用例
result = analyze_log(Path("server.log"))
print(f"ログレベル別件数: {result['counts']}")
# {'ERROR': 2, 'WARNING': 1, 'INFO': 2, 'DEBUG': 1}
応用:Slackに通知
import requests
def notify_slack(webhook_url: str, errors: list[str]):
if errors:
message = "🚨 エラー検出:\n" + "\n".join(errors[:5])
requests.post(webhook_url, json={"text": message})
4. 定期バックアップ(ローテーション付き)
古いバックアップを自動削除しながらバックアップを取る。
from datetime import datetime
import shutil
from pathlib import Path
def backup_files(source: Path, backup_dir: Path, max_backups: int = 5):
"""タイムスタンプ付きバックアップを作成"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_path = backup_dir / f"backup_{timestamp}"
# バックアップ作成
if source.is_dir():
shutil.copytree(source, backup_path)
else:
backup_path.mkdir(parents=True)
shutil.copy2(source, backup_path)
# 古いバックアップを削除(ローテーション)
backups = sorted(backup_dir.glob("backup_*"), reverse=True)
for old_backup in backups[max_backups:]:
shutil.rmtree(old_backup)
print(f"古いバックアップ削除: {old_backup.name}")
return backup_path
# 毎日実行(cron / タスクスケジューラで)
backup_path = backup_files(
Path("./important_data"),
Path("./backups"),
max_backups=7 # 1週間分保持
)
5. CSVデータ処理
import csv
from pathlib import Path
from collections import Counter
def analyze_csv(csv_path: Path) -> dict:
"""CSVを分析"""
with open(csv_path, newline='', encoding='utf-8') as f:
reader = csv.DictReader(f)
rows = list(reader)
# 都市別カウント
cities = Counter(row['city'] for row in rows)
ages = [int(row['age']) for row in rows]
return {
"total_rows": len(rows),
"city_distribution": dict(cities),
"avg_age": sum(ages) / len(ages),
}
# 結果
# {'total_rows': 4, 'city_distribution': {'Tokyo': 2, 'Osaka': 1, 'Nagoya': 1}, 'avg_age': 29.5}
6. JSON設定ファイル管理
ドット記法でネストした設定にアクセス。
import json
from pathlib import Path
class ConfigManager:
def __init__(self, config_path: Path):
self.path = config_path
self.config = json.loads(config_path.read_text()) if config_path.exists() else {}
def get(self, key: str, default=None):
"""ドット記法でアクセス: 'database.host'"""
value = self.config
for k in key.split("."):
if isinstance(value, dict):
value = value.get(k)
else:
return default
return value if value is not None else default
def set(self, key: str, value):
keys = key.split(".")
config = self.config
for k in keys[:-1]:
config = config.setdefault(k, {})
config[keys[-1]] = value
def save(self):
self.path.write_text(json.dumps(self.config, indent=2, ensure_ascii=False))
# 使用例
config = ConfigManager(Path("config.json"))
print(config.get("database.host")) # localhost
print(config.get("database.user", "admin")) # admin(デフォルト値)
config.set("database.user", "myuser")
config.save()
7. 一括リネーム
from pathlib import Path
import re
def batch_rename(directory: Path, pattern: str, replacement: str, dry_run: bool = True):
"""正規表現でファイル名を一括変更"""
for file_path in directory.iterdir():
if file_path.is_file():
new_name = re.sub(pattern, replacement, file_path.name)
if new_name != file_path.name:
new_path = file_path.parent / new_name
if dry_run:
print(f" {file_path.name} → {new_name}")
else:
file_path.rename(new_path)
# 使用例:IMG_20240115_001.jpg → 2025-01-15_001.jpg
batch_rename(
Path("./photos"),
r"IMG_(\d{4})(\d{2})(\d{2})_(\d+)",
r"\1-\2-\3_\4",
dry_run=True
)
8. タスクスケジューラ連携
Windows(タスクスケジューラ):
schtasks /create /tn "DailyBackup" /tr "python C:\scripts\backup.py" /sc daily /st 03:00
Linux(cron):
# crontab -e
0 3 * * * /usr/bin/python3 /home/user/scripts/backup.py
Python内でスケジュール(schedule):
import schedule
import time
def job():
print("バックアップ実行中...")
# backup_files(...)
schedule.every().day.at("03:00").do(job)
schedule.every(30).minutes.do(job) # 30分ごと
while True:
schedule.run_pending()
time.sleep(60)
まとめ
| スクリプト | 用途 |
|---|---|
| ファイル整理 | ダウンロードフォルダのカオス解消 |
| 重複検出 | ディスク容量の節約 |
| ログ解析 | エラー監視・アラート |
| バックアップ | データ保全 |
| CSV処理 | レポート作成 |
| 設定管理 | 環境ごとの設定切り替え |
ポイント:
- 必ず
dry_run=Trueで確認してから実行 -
pathlibを使えばOS問わず動作 - 大量ファイルは
tqdmで進捗表示
from tqdm import tqdm
for file in tqdm(list(directory.rglob("*")), desc="処理中"):
process(file)
面倒な作業は自動化して、クリエイティブな仕事に時間を使いましょう!