基礎から学ぶPython入門 90日コース | 業務自動化 - Day 42:定期実行

Python 90日で身につけるPython
スポンサーリンク
スポンサーリンク

Day 42:定期実行で「自動で動き続ける仕組み」を考える

Day 42では、業務自動化の中でもとても重要な 「定期実行」 をテーマにします。 これまで作ってきた「CSV変換ツール」「集計ツール」「PDF操作ツール」などは、 手動で python xxx.py を実行していましたが、 本当に業務を楽にするには「毎日決まった時間に勝手に動いてくれる」仕組みが欲しくなります。

ここでは、次のキーワードを軸に進めていきます。

  • スケジュール処理
  • バッチ処理
  • 定期実行の考え方

Pythonだけで完結する「簡易的な定期実行」と、 OSの機能(cronやタスクスケジューラ)と組み合わせる「本格的な定期実行」の考え方を、 初心者向けにかみ砕いて説明していきます。

定期実行とは何かを言葉で整理する

「人が押す実行ボタン」を「時間」が押してくれるイメージ

これまでのスクリプトは、次のような流れでした。

  1. 人がコマンドを打つ(例:python day39_csv_transform_tool.py
  2. スクリプトが1回だけ動く
  3. 終わったら止まる

定期実行 は、これを次のように変えるイメージです。

  • 「毎日朝9時に自動で動く」
  • 「1時間ごとにログを集計する」
  • 「毎週月曜日にレポートを作る」

つまり、「時間をトリガーにしてスクリプトを動かす」 という考え方です。

バッチ処理とは何か

「まとめて一気に処理する」スタイル

バッチ処理 という言葉は、業務システムでよく出てきます。

  • 昼間はシステムを使っている人が多いので、 夜間に「1日のデータをまとめて集計する」
  • 月末に「1ヶ月分の請求書をまとめて発行する」

こうした「あるタイミングで、まとめて一気に処理する」スタイルがバッチ処理です。

Pythonで作る「CSV変換ツール」「集計ツール」などは、 まさにバッチ処理の一種と言えます。

スケジュール処理の基本的な考え方

1. 「何を」定期実行したいかを決める

まずは、定期実行したい処理を1つの関数にまとめます。

例:前日に作られたCSVを集計してレポートを作る処理。

# day42_batch_job.py
def run_daily_job():
    """毎日実行したいバッチ処理の例です。"""
    print("=== 日次バッチ処理を開始します ===")
    # ここに「CSV読み込み → データ加工 → 集計 → 出力」などの処理を書きます。
    # 今はサンプルとしてメッセージだけにしておきます。
    print("CSVを読み込みます...")
    print("データを加工します...")
    print("集計結果を出力します...")
    print("=== 日次バッチ処理が完了しました ===")
Python

2. 「いつ」「どのくらいの頻度で」実行するかを決める

  • 毎日1回
  • 1時間ごと
  • 5分ごと

など、頻度を決めます。

3. その頻度で関数を呼び出す仕組みを作る

ここからが「スケジュール処理」の話になります。

Pythonだけで作る簡易的な定期実行

time.sleep() を使った「簡易ループ」

Python標準ライブラリの time モジュールを使うと、 「一定時間待ってから処理を実行する」ということができます。

# day42_simple_scheduler.py
import time
from datetime import datetime


def run_daily_job():
    """毎回実行したい処理(バッチ処理)の例です。"""
    print("=== バッチ処理開始 ===")
    print("現在時刻:", datetime.now())
    print("ここでCSV集計やレポート生成などを行います。")
    print("=== バッチ処理終了 ===\n")


def main():
    # ここでは「60秒ごとに実行する」簡易スケジューラを作ります。
    interval_seconds = 60

    print("=== 簡易スケジューラを開始します ===")
    print(f"{interval_seconds}秒ごとにバッチ処理を実行します。")

    while True:
        run_daily_job()          # バッチ処理を実行
        time.sleep(interval_seconds)  # 指定秒数だけ待機


if __name__ == "__main__":
    main()
Python

ポイント:

  • while True: で「永遠に繰り返す」ループを作っている。
  • time.sleep(秒数) で「その秒数だけ待つ」。
  • この例では「60秒ごと」に run_daily_job() が呼ばれます。

これは「本番用のスケジューラ」というより、 「定期実行の感覚をつかむための練習用」 として捉えると良いです。

「毎日決まった時間に実行する」考え方

現在時刻を見て「次の実行時刻まで待つ」

もう少し踏み込んで、「毎日9時に実行する」イメージを考えてみます。

  1. 現在時刻を取得する
  2. 今日の9時を表す datetime を作る
  3. 「今」が9時より前なら、その差分だけ sleep する
  4. 「今」が9時を過ぎているなら、翌日の9時までの差分を計算して sleep する

これをコードにすると、少し長くなりますが、考え方はシンプルです。

# day42_daily_scheduler.py
import time
from datetime import datetime, timedelta


def run_daily_job():
    """毎日決まった時間に実行したい処理の例です。"""
    print("=== 日次バッチ処理開始 ===")
    print("現在時刻:", datetime.now())
    print("ここで日次レポート生成などを行います。")
    print("=== 日次バッチ処理終了 ===\n")


def get_next_run_time(hour=9, minute=0):
    """次に実行すべき日時(今日または明日の指定時刻)を計算する関数です。"""
    now = datetime.now()

    # 今日の指定時刻(例:9:00)を作ります。
    today_run = datetime(now.year, now.month, now.day, hour, minute)

    if now < today_run:
        # まだ今日の指定時刻前なら、今日の指定時刻が次の実行時刻です。
        return today_run
    else:
        # すでに今日の指定時刻を過ぎているなら、翌日の指定時刻が次の実行時刻です。
        next_day = now + timedelta(days=1)
        return datetime(next_day.year, next_day.month, next_day.day, hour, minute)


def main():
    print("=== 日次スケジューラを開始します ===")
    print("毎日9:00にバッチ処理を実行します。")

    while True:
        next_run = get_next_run_time(hour=9, minute=0)
        now = datetime.now()

        # 次の実行時刻までの秒数を計算します。
        wait_seconds = (next_run - now).total_seconds()

        print("現在時刻:", now)
        print("次の実行時刻:", next_run)
        print("待機秒数:", wait_seconds)

        # 次の実行時刻まで待機します。
        time.sleep(wait_seconds)

        # 待機が終わったらバッチ処理を実行します。
        run_daily_job()


if __name__ == "__main__":
    main()
Python

重要ポイント:

  • 「次にいつ実行すべきか」を get_next_run_time() で計算している。
  • next_run - now で「待つべき時間(timedelta)」を求めている。
  • total_seconds() で「待つべき秒数」を取り出し、sleep に渡している。

このように、日時処理(Day 41)で学んだ datetime timedelta が、定期実行のロジックの土台になる ことが分かります。

実務でよく使われる「OS側のスケジューラ」との役割分担

ここまでの例は「Pythonだけで定期実行を頑張る」形でしたが、 実務では次のような役割分担をすることが多いです。

  • Pythonスクリプト
    • 「1回分のバッチ処理」をきちんと書く(CSV集計、PDF生成など)
  • OSのスケジューラ
    • そのスクリプトを「いつ」「どの頻度で」実行するかを管理する

代表的なスケジューラ:

  • Linux / macOS:cron
  • Windows:タスクスケジューラ

イメージとしては、

  • Python側:python day39_csv_transform_tool.py を1回実行すると、ちゃんと仕事をするように作る
  • OS側:そのコマンドを「毎日9時に実行するように設定する」

という分担です。

Day 42では、OSの設定方法の細かい手順までは踏み込みませんが、 「定期実行は、Pythonだけで完結させることもできるが、実務ではOSのスケジューラと組み合わせるのが一般的」 という考え方を持っておくと、後で応用しやすくなります。

定期実行の設計で意識したいポイント

1. 「1回分の処理」を必ず関数にまとめる

def run_daily_job():
    # ここに処理を書く
    ...
Python

こうしておくと、

  • 手動でテストするとき:run_daily_job() を直接呼べる
  • スケジューラから呼ぶとき:main() の中で run_daily_job() を呼ぶだけでよい

という形になり、コードが整理されます。

2. エラーが起きたときの振る舞いを考える

定期実行では、「たまたまその日だけデータが壊れていた」 などのケースも起こり得ます。

  • エラーが起きたらログを残す
  • エラーが起きてもスクリプト全体が落ちないように try / except を使う
  • 次回の実行に影響が出ないようにする

などを意識すると、安定したバッチ処理になります。

簡単な例:

def run_daily_job():
    print("=== 日次バッチ処理開始 ===")
    try:
        print("CSVを読み込みます...")
        # ここで実際の処理を書く(例:ファイル読み込み・集計など)
        print("処理が正常に完了しました。")
    except Exception as e:
        print("エラーが発生しました:", e)
        # ここでログファイルに書き出すなども検討できます。
    print("=== 日次バッチ処理終了 ===\n")
Python

Day 42ミニテンプレート:定期実行の「型」

最後に、Day 42の内容をまとめた「定期実行の型」を示します。

# day42_batch_template.py
import time
from datetime import datetime, timedelta


def run_batch():
    """定期的に実行したいバッチ処理をまとめた関数です。"""
    print("=== バッチ処理開始 ===")
    print("現在時刻:", datetime.now())
    # ここに実際の処理を書く(CSV集計、PDF生成など)
    print("ここで業務ロジックを実行します。")
    print("=== バッチ処理終了 ===\n")


def get_next_run_time(hour=9, minute=0):
    """次の実行時刻を計算する関数です。"""
    now = datetime.now()
    today_run = datetime(now.year, now.month, now.day, hour, minute)

    if now < today_run:
        return today_run
    else:
        next_day = now + timedelta(days=1)
        return datetime(next_day.year, next_day.month, next_day.day, hour, minute)


def scheduler_loop():
    """毎日指定時刻にバッチ処理を実行する簡易スケジューラです。"""
    print("=== スケジューラ開始 ===")
    print("毎日9:00にバッチ処理を実行します。")

    while True:
        next_run = get_next_run_time(hour=9, minute=0)
        now = datetime.now()
        wait_seconds = (next_run - now).total_seconds()

        print("現在時刻:", now)
        print("次の実行時刻:", next_run)
        print("待機秒数:", wait_seconds)

        time.sleep(wait_seconds)
        run_batch()


def main():
    scheduler_loop()


if __name__ == "__main__":
    main()
Python

このテンプレートは、

  • 「バッチ処理を1つの関数にまとめる」
  • 「次の実行時刻を計算する」
  • 「その時刻まで待ってから処理を実行する」

という、定期実行の基本的な考え方をそのままコードにしたものです。

Day 42のまとめ

Day 42では、定期実行として、

  • バッチ処理とは「あるタイミングでまとめて一気に処理する」スタイルであること
  • 定期実行とは「人が押す実行ボタンを、時間が押してくれる仕組み」であること
  • time.sleep()while True を使った簡易的なスケジューラの作り方
  • datetimetimedelta を使って「次の実行時刻」を計算する考え方
  • 実務では、PythonスクリプトとOSのスケジューラ(cronやタスクスケジューラ)を組み合わせるのが一般的であること
  • バッチ処理を関数にまとめ、エラー処理やログ出力を意識することで、安定した定期実行が実現できること

をステップバイステップで体験していただきました。

ここまで来ると、 「毎日・毎週・毎月の定型業務を、Pythonスクリプト+スケジューラで自動化する」 というイメージがかなり具体的になっているはずです。

次のステップでは、こうした定期実行の仕組みと、 これまで作ってきたCSV・Excel・PDF・pandasのツールを組み合わせて、 より実務に近い「業務自動化システム」を形にしていくことができます。

タイトルとURLをコピーしました