【Python】フォルダに入れたファイルを自動処理！watchdogの使い方

import shutil
import time
from datetime import datetime
from pathlib import Path

from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer


# =============================================
# 設定
# =============================================
WATCH_DIR = Path.cwd() / "watch_folder"
OUTPUT_DIR = Path.cwd() / "organized"

# 同じ名前のファイルが来たときに重複しないよう、時刻まで付けます。
FILENAME_FORMAT = "%Y%m%d_%H%M%S"

# コピー途中のファイルを処理しないための待機設定
CHECK_INTERVAL = 0.5
STABLE_CHECK_COUNT = 3
MAX_WAIT_SECONDS = 30

# 拡張子ごとの分類名
CATEGORY_MAP = {
    ".jpg": "images",
    ".jpeg": "images",
    ".png": "images",
    ".webp": "images",
    ".gif": "images",
    ".bmp": "images",
    ".pdf": "pdf",
    ".csv": "csv",
    ".xlsx": "excel",
    ".xls": "excel",
    ".xlsm": "excel",
    ".docx": "word",
    ".doc": "word",
    ".txt": "text",
    ".mp4": "video",
    ".mov": "video",
    ".webm": "video",
    ".mkv": "video",
    ".mp3": "audio",
    ".wav": "audio",
    ".zip": "archive",
    ".7z": "archive",
    ".rar": "archive",
}


# =============================================
# 補助関数
# =============================================
def ensure_directories() -> None:
    WATCH_DIR.mkdir(parents=True, exist_ok=True)
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)


def get_category(path: Path) -> str:
    return CATEGORY_MAP.get(path.suffix.lower(), "other")


def wait_until_file_is_stable(path: Path) -> bool:
    """コピー中のファイルを触らないよう、サイズが一定になるまで待つ。"""
    stable_count = 0
    last_size = -1
    started = time.time()

    while time.time() - started < MAX_WAIT_SECONDS:
        if not path.exists():
            return False

        try:
            current_size = path.stat().st_size
        except OSError:
            time.sleep(CHECK_INTERVAL)
            continue

        if current_size == last_size:
            stable_count += 1
            if stable_count >= STABLE_CHECK_COUNT:
                return True
        else:
            stable_count = 0
            last_size = current_size

        time.sleep(CHECK_INTERVAL)

    return False


def build_destination(source: Path) -> Path:
    now = datetime.now()
    category = get_category(source)

    # organized / 種類 / YYYY-MM-DD /
    target_dir = OUTPUT_DIR / category / now.strftime("%Y-%m-%d")
    target_dir.mkdir(parents=True, exist_ok=True)

    base_name = now.strftime(FILENAME_FORMAT)
    suffix = source.suffix.lower()
    destination = target_dir / f"{base_name}{suffix}"

    # 同じ秒に複数ファイルが来ても上書きしない
    counter = 1
    while destination.exists():
        destination = target_dir / f"{base_name}_{counter:02d}{suffix}"
        counter += 1

    return destination


def organize_file(path: Path) -> None:
    if not path.is_file():
        return

    # 一時ファイルや隠しファイルを簡易除外
    if path.name.startswith("~$") or path.name.startswith("."):
        return

    print(f"[検知] {path.name}")

    if not wait_until_file_is_stable(path):
        print(f"[スキップ] ファイルが安定しませんでした: {path}")
        return

    destination = build_destination(path)

    try:
        shutil.move(str(path), str(destination))
        print(f"[整理完了] {path.name} -> {destination.relative_to(Path.cwd())}")
    except Exception as exc:
        print(f"[エラー] {path.name}: {exc}")


# =============================================
# watchdogイベント
# =============================================
class FileCreatedHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        organize_file(Path(event.src_path))

    def on_moved(self, event):
        # OSやアプリによっては一時ファイルからリネームされて配置されることがあるため対応
        if event.is_directory:
            return
        organize_file(Path(event.dest_path))


# =============================================
# メイン
# =============================================
def main() -> None:
    ensure_directories()

    print("=" * 60)
    print("ファイル自動整理ツール")
    print("=" * 60)
    print(f"監視フォルダ : {WATCH_DIR}")
    print(f"整理先       : {OUTPUT_DIR}")
    print()
    print("watch_folder にファイルを置くと、自動でリネームして整理します。")
    print("終了する場合は Ctrl + C を押してください。")
    print()

    observer = Observer()
    observer.schedule(FileCreatedHandler(),
                      str(WATCH_DIR),
                      recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("\n終了します...")
        observer.stop()

    observer.join()


if __name__ == "__main__":
    main()

def wait_until_file_is_stable(path: Path) -> bool:
    """コピー中のファイルを触らないよう、サイズが一定になるまで待つ。"""
    stable_count = 0
    last_size = -1
    started = time.time()

    while time.time() - started < MAX_WAIT_SECONDS:
        if not path.exists():
            return False

        try:
            current_size = path.stat().st_size
        except OSError:
            time.sleep(CHECK_INTERVAL)
            continue

        if current_size == last_size:
            stable_count += 1
            if stable_count >= STABLE_CHECK_COUNT:
                return True
        else:
            stable_count = 0
            last_size = current_size

        time.sleep(CHECK_INTERVAL)

    return False

# =============================================
# watchdogイベント
# =============================================
class FileCreatedHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        organize_file(Path(event.src_path))

    def on_moved(self, event):
        # OSやアプリによっては一時ファイルからリネームされて配置されることがあるため対応
        if event.is_directory:
            return
        organize_file(Path(event.dest_path))