Python 業務自動化 | ファイル・フォルダ自動化：基本操作 － ディレクトリ監視

Python Python
スポンサーリンク

ディレクトリ監視は「フォルダ内の変化を自動で検知し、即座に処理を実行する」ための強力な仕組み

業務では、次のような「フォルダの変化をトリガーにした自動処理」が非常に多くあります。

  • 新しいファイルが届いたら自動で読み込む
  • ログファイルが更新されたら解析を走らせる
  • 監視フォルダに ZIP が置かれたら自動解凍する
  • 画像が追加されたらサムネイルを生成する

Python では watchdog というライブラリを使うことで、フォルダの変化（作成・更新・削除・移動）をリアルタイムに検知できます。
ここでは、初心者でも理解しやすいように、基本から実務テンプレートまで丁寧に解説します。

監視の基本：watchdog を使ってフォルダの変化を検知する

監視に必要なインストール

pip install watchdog

最小構成：フォルダの変化を検知して表示する

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

class Handler(FileSystemEventHandler):
    def on_created(self, event):
        print("作成:", event.src_path)

    def on_modified(self, event):
        print("更新:", event.src_path)

    def on_deleted(self, event):
        print("削除:", event.src_path)

def watch(folder):
    observer = Observer()
    observer.schedule(Handler(), folder, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

watch("watch_folder")
Python

深掘りポイント

  • on_created / on_modified / on_deleted でイベントを受け取る
  • recursive=True でサブフォルダも監視
  • Observer がバックグラウンドで監視し続ける
  • while True で常時監視する仕組み

新規ファイルが来たら自動処理する（最もよく使うパターン）

例：新しい CSV が来たら自動で読み込む

import pandas as pd
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class CSVHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.src_path.endswith(".csv"):
            print("新規CSV:", event.src_path)
            df = pd.read_csv(event.src_path)
            print(df.head())

def watch_csv(folder):
    observer = Observer()
    observer.schedule(CSVHandler(), folder, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

watch_csv("inbox")
Python

深掘りポイント

  • 新規ファイルだけを対象にしたい場合は on_created を使う
  • 拡張子フィルタで不要な処理を避ける
  • 自動 ETL（データ取り込み）に最適

例題①：監視フォルダに ZIP が置かれたら自動解凍する

シナリオ

外部システムから ZIP が届く → 自動で解凍 → 中身を処理したい。

import zipfile
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
import os

class ZipHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.src_path.endswith(".zip"):
            print("ZIP検出:", event.src_path)
            with zipfile.ZipFile(event.src_path, "r") as z:
                z.extractall("extracted")
            print("解凍完了")

def watch_zip(folder):
    observer = Observer()
    observer.schedule(ZipHandler(), folder, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

watch_zip("incoming")
Python

深掘りポイント

  • ZIP が届いた瞬間に自動解凍できる
  • 手動作業ゼロでファイル処理が進む
  • 業務フローの自動化に直結する

例題②：ログファイルが更新されたら自動で解析する

シナリオ

ログが追記されたら ERROR 行だけ抽出したい。

class LogHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if event.src_path.endswith(".log"):
            print("ログ更新:", event.src_path)
            with open(event.src_path, "r", encoding="utf-8") as f:
                lines = f.readlines()
                errors = [line for line in lines if "ERROR" in line]
                if errors:
                    print("エラー検出:")
                    for e in errors:
                        print(e.strip())

def watch_log(folder):
    observer = Observer()
    observer.schedule(LogHandler(), folder, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

watch_log("logs")
Python

深掘りポイント

  • on_modified はログ監視に最適
  • ERROR 行をリアルタイムに検出できる
  • 監視 → 通知 → 自動処理の基礎になる

例題③：監視フォルダに画像が来たらサムネイルを自動生成する

シナリオ

画像が届いたら自動で縮小版を作りたい。

from PIL import Image
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class ImageHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.src_path.lower().endswith((".jpg", ".png")):
            print("画像検出:", event.src_path)
            img = Image.open(event.src_path)
            img.thumbnail((200, 200))
            img.save(event.src_path + "_thumb.jpg")
            print("サムネイル作成")

def watch_images(folder):
    observer = Observer()
    observer.schedule(ImageHandler(), folder, recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

watch_images("images")
Python

深掘りポイント

  • 画像処理と組み合わせると業務の幅が広がる
  • EC サイトの商品画像処理などでよく使われる

pathlib を使った読みやすい監視コード

Path オブジェクトで直感的に書ける

from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class PathHandler(FileSystemEventHandler):
    def on_created(self, event):
        p = Path(event.src_path)
        print("作成:", p.name)

def watch_path(folder: Path):
    observer = Observer()
    observer.schedule(PathHandler(), str(folder), recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()

    observer.join()

watch_path(Path("watch_folder"))
Python

メリット

  • Path によるパス操作が読みやすい
  • 大規模コードでも保守性が高い

ディレクトリ監視を業務で設計するときの視点

  • どのイベント（作成・更新・削除）を監視するか明確にする
  • 不要なファイルを無視するために拡張子フィルタを使う
  • 自動処理（解凍・解析・コピー）と組み合わせると強力
  • 長時間動かす場合はログ出力や例外処理が重要
  • ネットワークフォルダ監視は遅延があるため注意

スポンサーリンク
タイトルとURLをコピーしました