Python自動化マスター第8編：スケジューリングと実践プロジェクト
Python Automation Master Part 8: Scheduling and Real-World Projects
はじめに：自動化の完成、スケジューリング
これまで学んだファイル処理、Webスクレイピング、API連携などの技術を実際に有用にする核心は「スケジューリング」です。決まった時間に自動的に実行されるスクリプトは、手作業を完全に代替し、24時間休まず作業を遂行します。
今回の最終編では、Pythonでタスクを予約して実行する様々な方法とともに、安定した自動化システム構築のためのロギング、エラー処理、設定管理技法を学習します。最後にこのシリーズで学んだすべての内容を総合した実践プロジェクトを構築してみましょう。
1. 時間処理：timeとdatetimeモジュール
1.1 timeモジュールの基礎
import time
# 現在時刻（Unixタイムスタンプ）
timestamp = time.time()
print(f"現在のタイムスタンプ: {timestamp}") # 例: 1737529200.123456
# プログラムの一時停止
print("3秒間待機中...")
time.sleep(3)
print("待機完了！")
# 実行時間の測定
start_time = time.time()
# 作業の実行
for i in range(1000000):
pass
end_time = time.time()
print(f"実行時間: {end_time - start_time:.4f}秒")
# より精密な時間測定
start = time.perf_counter()
# 作業の実行
time.sleep(0.1)
end = time.perf_counter()
print(f"精密測定: {end - start:.6f}秒")
# 構造化された時間
local_time = time.localtime()
print(f"現在時刻: {local_time.tm_year}年 {local_time.tm_mon}月 {local_time.tm_mday}日")
print(f"時間: {local_time.tm_hour}:{local_time.tm_min}:{local_time.tm_sec}")
# 文字列に変換
formatted = time.strftime("%Y-%m-%d %H:%M:%S", local_time)
print(f"フォーマット済み時間: {formatted}")
1.2 datetimeモジュールの活用
from datetime import datetime, date, time, timedelta
import pytz # pip install pytz
# 現在の日付と時間
now = datetime.now()
print(f"現在: {now}")
print(f"日付: {now.date()}")
print(f"時間: {now.time()}")
# 特定の日付/時間を作成
specific_date = datetime(2026, 1, 22, 9, 30, 0)
print(f"特定の時間: {specific_date}")
# 日付のフォーマット
formatted = now.strftime("%Y年 %m月 %d日 %H時 %M分 %S秒")
print(f"日本語フォーマット: {formatted}")
# 文字列のパース
date_str = "2026-01-22 14:30:00"
parsed = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")
print(f"パースされた日付: {parsed}")
# 日付演算
tomorrow = now + timedelta(days=1)
next_week = now + timedelta(weeks=1)
two_hours_later = now + timedelta(hours=2)
print(f"明日: {tomorrow.date()}")
print(f"来週: {next_week.date()}")
print(f"2時間後: {two_hours_later.time()}")
# 2つの日付の差
date1 = datetime(2026, 12, 31)
date2 = datetime.now()
diff = date1 - date2
print(f"2026年末まで{diff.days}日残り")
# タイムゾーン処理
jst = pytz.timezone('Asia/Tokyo')
utc = pytz.UTC
now_jst = datetime.now(jst)
now_utc = datetime.now(utc)
print(f"日本時間: {now_jst}")
print(f"UTC時間: {now_utc}")
2. scheduleライブラリ
2.1 基本的な使い方
import schedule
import time
# pip install schedule
def job():
print(f"[{time.strftime('%H:%M:%S')}] タスク実行！")
def morning_report():
print("朝のレポートを生成します。")
def backup_data():
print("データをバックアップします。")
# 様々なスケジューリングパターン
schedule.every(10).seconds.do(job) # 10秒ごと
schedule.every(5).minutes.do(job) # 5分ごと
schedule.every().hour.do(job) # 毎時
schedule.every().day.at("09:00").do(morning_report) # 毎日午前9時
schedule.every().monday.do(job) # 毎週月曜日
schedule.every().wednesday.at("13:15").do(job) # 毎週水曜日13:15
# 特定の時間帯にのみ実行
schedule.every().day.at("00:00").do(backup_data) # 深夜にバックアップ
# メインループ
print("スケジューラ開始...")
while True:
schedule.run_pending()
time.sleep(1)
2.2 高度なscheduleパターン
import schedule
import time
from functools import partial
# 引数のあるタスク
def greet(name):
print(f"こんにちは、{name}さん！")
# partialを使用した引数の渡し方
schedule.every().day.at("08:00").do(partial(greet, "田中"))
# タグを使用したタスク管理
def data_sync():
print("データ同期中...")
def send_notification():
print("通知送信中...")
schedule.every(30).minutes.do(data_sync).tag('sync', 'database')
schedule.every().hour.do(send_notification).tag('notification')
# 特定タグのタスクのみキャンセル
# schedule.clear('sync')
# タスクのキャンセル（一度だけ実行）
def run_once():
print("一度だけ実行されるタスク")
return schedule.CancelJob
schedule.every().second.do(run_once)
# 次の実行時間を確認
next_run = schedule.next_run()
print(f"次のタスク実行時間: {next_run}")
# スケジューラ実行
while True:
schedule.run_pending()
time.sleep(1)
3. APSchedulerライブラリ
3.1 APSchedulerの紹介とインストール
# pip install apscheduler
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.date import DateTrigger
from datetime import datetime, timedelta
def my_job(message="デフォルトメッセージ"):
print(f"[{datetime.now().strftime('%H:%M:%S')}] {message}")
# BlockingScheduler - メインスレッドで実行（ブロッキング）
scheduler = BlockingScheduler()
# Intervalトリガー - 一定間隔で実行
scheduler.add_job(
my_job,
IntervalTrigger(seconds=10),
args=["10秒ごとに実行"],
id='interval_job'
)
# Cronトリガー - cron式を使用
scheduler.add_job(
my_job,
CronTrigger(hour=9, minute=0),
args=["毎日午前9時"],
id='daily_job'
)
# Dateトリガー - 特定の日時に一度だけ実行
scheduler.add_job(
my_job,
DateTrigger(run_date=datetime.now() + timedelta(seconds=5)),
args=["5秒後に一度だけ実行"],
id='one_time_job'
)
# スケジューラ開始
try:
print("APScheduler開始...")
scheduler.start()
except KeyboardInterrupt:
print("スケジューラ終了")
scheduler.shutdown()
3.2 Cron式の活用
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime
scheduler = BlockingScheduler()
def job(name):
print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {name} 実行")
# Cron式パターン
# ┌───────────── 分 (0 - 59)
# │ ┌───────────── 時 (0 - 23)
# │ │ ┌───────────── 日 (1 - 31)
# │ │ │ ┌───────────── 月 (1 - 12)
# │ │ │ │ ┌───────────── 曜日 (0 - 6, 0=月曜日)
# │ │ │ │ │
# * * * * *
# 毎分実行
scheduler.add_job(job, CronTrigger(minute='*'), args=["毎分"])
# 毎時30分に実行
scheduler.add_job(job, CronTrigger(minute=30), args=["毎時30分"])
# 平日午前9時に実行（月-金）
scheduler.add_job(job, CronTrigger(day_of_week='mon-fri', hour=9, minute=0), args=["平日9時"])
# 毎月1日の深夜0時に実行
scheduler.add_job(job, CronTrigger(day=1, hour=0, minute=0), args=["毎月1日"])
# 15分ごとに実行
scheduler.add_job(job, CronTrigger(minute='*/15'), args=["15分ごと"])
scheduler.start()
4. ロギング（loggingモジュール）
4.1 基本的なロギング設定
import logging
# 基本設定
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# ロガーの作成
logger = logging.getLogger(__name__)
# ログレベル別メッセージ
logger.debug("デバッグメッセージ - 詳細な診断情報")
logger.info("情報メッセージ - 正常動作の確認")
logger.warning("警告メッセージ - 注意が必要な状況")
logger.error("エラーメッセージ - 機能の失敗")
logger.critical("致命的エラー - プログラム実行不可")
# 例外ロギング
try:
result = 1 / 0
except Exception as e:
logger.exception("例外発生！") # トレースバック付き
4.2 ファイルロギングとローテーション
import logging
from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler
from pathlib import Path
def setup_logger(name, log_dir="logs", level=logging.DEBUG):
"""ロガー設定関数"""
log_path = Path(log_dir)
log_path.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger(name)
logger.setLevel(level)
# フォーマッター
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
# コンソールハンドラー
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)
console_handler.setFormatter(formatter)
# ファイルハンドラー（サイズベースローテーション）
file_handler = RotatingFileHandler(
log_path / f"{name}.log",
maxBytes=10*1024*1024, # 10MB
backupCount=5,
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
# エラー専用ファイルハンドラー
error_handler = RotatingFileHandler(
log_path / f"{name}_error.log",
maxBytes=10*1024*1024,
backupCount=5,
encoding='utf-8'
)
error_handler.setLevel(logging.ERROR)
error_handler.setFormatter(formatter)
# ハンドラーを追加
logger.addHandler(console_handler)
logger.addHandler(file_handler)
logger.addHandler(error_handler)
return logger
# 使用例
logger = setup_logger("automation")
logger.debug("デバッグメッセージ")
logger.info("情報メッセージ")
logger.warning("警告メッセージ")
logger.error("エラーメッセージ")
5. エラー処理とリトライロジック
5.1 リトライデコレーター
import time
import logging
from functools import wraps
import random
logger = logging.getLogger(__name__)
def retry(max_attempts=3, delay=1, backoff=2, exceptions=(Exception,)):
"""
リトライデコレーター
Args:
max_attempts: 最大試行回数
delay: 初期待機時間（秒）
backoff: 待機時間の増加倍率
exceptions: リトライする例外のタプル
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
current_delay = delay
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except exceptions as e:
if attempt == max_attempts:
logger.error(f"{func.__name__} 最終失敗: {e}")
raise
logger.warning(
f"{func.__name__} 失敗 (試行 {attempt}/{max_attempts}): {e}"
)
logger.info(f"{current_delay}秒後にリトライ...")
time.sleep(current_delay)
current_delay *= backoff
return None
return wrapper
return decorator
# 使用例
@retry(max_attempts=3, delay=1, backoff=2, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
"""ネットワークリクエスト（失敗する可能性あり）"""
if random.random() < 0.7: # 70%の確率で失敗
raise ConnectionError("接続失敗")
return "データ"
# テスト
try:
result = fetch_data("https://example.com")
print(f"結果: {result}")
except ConnectionError:
print("すべてのリトライ失敗")
6. 設定ファイル管理
6.1 python-dotenvの使用
# pip install python-dotenv
from dotenv import load_dotenv
import os
from pathlib import Path
# .envファイルの例:
"""
# データベース設定
DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydb
DB_USER=admin
DB_PASSWORD=secret123
# APIキー
OPENAI_API_KEY=sk-xxxxxxxxxxxxx
# 環境
ENVIRONMENT=development
DEBUG=true
"""
class EnvConfig:
"""環境変数ベースの設定管理"""
def __init__(self, env_file=".env"):
# .envファイルを読み込み
env_path = Path(env_file)
load_dotenv(env_path)
@staticmethod
def get(key, default=None):
"""環境変数を取得"""
return os.getenv(key, default)
@staticmethod
def get_bool(key, default=False):
"""ブール環境変数"""
value = os.getenv(key, str(default)).lower()
return value in ('true', '1', 'yes', 'on')
@staticmethod
def get_int(key, default=0):
"""整数環境変数"""
try:
return int(os.getenv(key, default))
except ValueError:
return default
@property
def is_debug(self):
return self.get_bool('DEBUG')
@property
def environment(self):
return self.get('ENVIRONMENT', 'development')
# 使用例
config = EnvConfig()
print(f"環境: {config.environment}")
print(f"デバッグ: {config.is_debug}")
print(f"OpenAI API Key: {config.get('OPENAI_API_KEY', 'Not set')[:20]}...")
7. 実践総合プロジェクト：日次業務自動化システム
"""
日次業務自動化システム
- ニュース収集と要約
- 天気情報の取得
- 日次レポート生成
- メール送信
"""
import os
import json
import logging
from datetime import datetime
from pathlib import Path
from dotenv import load_dotenv
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
import requests
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from functools import wraps
import time
# 環境変数の読み込み
load_dotenv()
# ディレクトリ設定
BASE_DIR = Path(__file__).parent
LOG_DIR = BASE_DIR / "logs"
DATA_DIR = BASE_DIR / "data"
REPORT_DIR = BASE_DIR / "reports"
for dir_path in [LOG_DIR, DATA_DIR, REPORT_DIR]:
dir_path.mkdir(parents=True, exist_ok=True)
# ロギング設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOG_DIR / "automation.log", encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
def retry(max_attempts=3, delay=5):
"""リトライデコレーター"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
logger.error(f"{func.__name__} 最終失敗: {e}")
raise
logger.warning(f"{func.__name__} 失敗、リトライ {attempt + 1}/{max_attempts}")
time.sleep(delay)
return wrapper
return decorator
class WeatherCollector:
"""天気情報収集クラス"""
def __init__(self):
self.api_key = os.getenv("OPENWEATHERMAP_API_KEY")
@retry(max_attempts=3)
def collect(self, city="Tokyo"):
"""天気情報を取得"""
if not self.api_key:
logger.warning("OpenWeatherMap APIキーが設定されていません。")
return None
url = "https://api.openweathermap.org/data/2.5/weather"
params = {
"q": city,
"appid": self.api_key,
"units": "metric",
"lang": "ja"
}
response = requests.get(url, params=params, timeout=10)
response.raise_for_status()
data = response.json()
weather_info = {
"city": city,
"temperature": data["main"]["temp"],
"humidity": data["main"]["humidity"],
"description": data["weather"][0]["description"],
"wind_speed": data["wind"]["speed"]
}
logger.info(f"天気情報収集完了: {city}")
return weather_info
class ReportGenerator:
"""レポート生成クラス"""
def __init__(self):
self.weather_collector = WeatherCollector()
def generate_daily_report(self):
"""日次レポート生成"""
logger.info("日次レポート生成開始")
report_data = {
"date": datetime.now().strftime("%Y-%m-%d"),
"generated_at": datetime.now().isoformat(),
"weather": None
}
# 天気情報
try:
report_data["weather"] = self.weather_collector.collect("Tokyo")
except Exception as e:
logger.error(f"天気収集失敗: {e}")
# レポート保存
report_path = REPORT_DIR / f"report_{datetime.now().strftime('%Y%m%d')}.json"
with open(report_path, 'w', encoding='utf-8') as f:
json.dump(report_data, f, ensure_ascii=False, indent=2)
logger.info(f"レポート保存: {report_path}")
return report_data
class DailyAutomation:
"""日次自動化メインクラス"""
def __init__(self):
self.report_generator = ReportGenerator()
self.scheduler = BlockingScheduler(timezone='Asia/Tokyo')
def morning_routine(self):
"""朝のルーティン"""
logger.info("=" * 50)
logger.info("朝のルーティン開始")
try:
report_data = self.report_generator.generate_daily_report()
logger.info("朝のルーティン完了")
except Exception as e:
logger.error(f"朝のルーティン失敗: {e}")
def setup_schedule(self):
"""スケジュール設定"""
# 毎日午前8時に朝のルーティン
self.scheduler.add_job(
self.morning_routine,
CronTrigger(hour=8, minute=0),
id='morning_routine',
replace_existing=True
)
logger.info("スケジュール設定完了")
logger.info("登録されたタスク:")
for job in self.scheduler.get_jobs():
logger.info(f" - {job.id}: 次回実行 {job.next_run_time}")
def run(self):
"""自動化システム実行"""
logger.info("日次業務自動化システム開始")
self.setup_schedule()
try:
self.scheduler.start()
except KeyboardInterrupt:
logger.info("システム終了リクエスト")
self.scheduler.shutdown()
logger.info("システム終了完了")
# メイン実行
if __name__ == "__main__":
import sys
automation = DailyAutomation()
if len(sys.argv) > 1:
if sys.argv[1] == "test":
print("テストモード: 朝のルーティンを即時実行")
automation.morning_routine()
elif sys.argv[1] == "schedule":
automation.run()
else:
print("使用方法:")
print(" python automation.py test # テスト実行")
print(" python automation.py schedule # スケジューラ開始")
まとめ
今回の編でPython自動化マスターシリーズを終了します。8編にわたってPythonを活用した様々な自動化技法を学習しました。
このシリーズで扱った内容をまとめると：
- 第1-2編：Python基礎とファイル/フォルダ自動化
- 第3-4編：Excel自動化とWebスクレイピング
- 第5-6編：ExcelとCSV処理、メール自動化
- 第7編：API活用とデータ収集
- 第8編：スケジューリングと実践プロジェクト
自動化の核心は、繰り返し作業をコードで代替して時間を節約し、人的ミスを減らし、より価値のある仕事に集中できるようにすることです。このシリーズで学んだ技術を組み合わせれば、ほぼすべての種類の業務を自動化できます。
実践で自動化システムを構築する際は、次のことを常に覚えておいてください：
自動化システム構築の核心原則
1. 段階的構築：小さなことから始めて徐々に拡張してください。
2. 徹底したロギング：問題発生時に原因を把握できるようにログを残してください。
3. エラー処理：予期しないエラーでもシステムが停止しないようにしてください。
4. モニタリング：システムが正常に動作しているか定期的に確認してください。
5. 設定分離：環境別の設定をコードから分離してください。
Python自動化は無限の可能性を持っています。このシリーズで学んだ内容をもとに、皆さん独自の自動化システムを構築してみてください。繰り返し業務から解放されて、より創造的で価値のある仕事に時間を投資できることを願っています。