前回の復習と今回の目標

前回の記事では、Pythonを使用して定期的に都市の天気情報を取得し、指定の微信（WeChat）友人に送信する方法について解説しました。その際、以下の2つの課題が残りました。

• Windowsサービスとして定期タスクを実行する
• WeChatではなく、メールで通知を行う

本記事では、これらの課題を解決するために、既存のコードをモジュール化し、Windowsサービスとしての実行とメール送信機能の追加について解説します。

モジュール構成の再設計

コードの再利用性と保守性を向上させるために、以下のモジュールを追加・再構成しました。

• my_job.py：タスク処理を記述するモジュール
• util.py：共通ユーティリティ関数を記述するモジュール
• weather_service.py：Windowsサービスとして実行するメインモジュール
• timing_task.py：定期的に天気情報を取得・メール送信する処理を記述するモジュール

Windowsサービスとしての実行

定期タスクをWindowsサービスとして実行するには、pywin32ライブラリが必要です。以下のコマンドでインストールできます。

pip install pywin32

サービスの操作コマンド

• サービスのインストール：python PythonService.py install
• 自動起動設定：python PythonService.py --startup auto install
• サービスの起動：python PythonService.py start
• サービスの再起動：python PythonService.py restart
• サービスの停止：python PythonService.py stop
• サービスの削除：python PythonService.py remove

権限エラーの対処

サービスの起動時に「アクセス拒否」というエラーが発生した場合、以下の対応が必要です。

• Pythonの環境変数設定をユーザー変数からシステム変数に変更
• コマンドラインを管理者権限で実行

サービスの実装コード

Windowsサービスとして実行するクラスは、win32serviceutil.ServiceFrameworkを継承し、SvcDoRunメソッドに処理を記述します。

import win32serviceutil
import win32service
import win32event
import time

class WeatherService(win32serviceutil.ServiceFramework):
    _svc_name_ = "WeatherNotificationService"
    _svc_display_name_ = "Weather Notification Service"
    _svc_description_ = "指定された時間に天気情報を取得し、メールで通知します"

    def __init__(self, args):
        win32serviceutil.ServiceFramework.__init__(self, args)
        self.hWaitStop = win32event.CreateEvent(None, 0, 0, None)
        self.is_running = True

    def SvcStop(self):
        self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
        win32event.SetEvent(self.hWaitStop)
        self.is_running = False

    def SvcDoRun(self):
        while self.is_running:
            execute_job()
            time.sleep(5)

if __name__ == '__main__':
    win32serviceutil.HandleCommandLine(WeatherService)

定期タスクの実装

execute_job()関数では、現在時刻とタスクの実行時刻を比較し、一致した場合に天気情報を取得し、メールを送信します。

import time
from weather_service import get_weather
from util import dict_to_string
from sendmail import send_email

my_jobs = [
    {
        "receivers": ["user1@example.com"],
        "city": "Shanghai",
        "time": "7.00,18.30"
    },
    {
        "receivers": ["user2@example.com"],
        "city": "Beijing",
        "time": "6.30,19.00"
    }
]

def execute_job():
    current_time = time.localtime(time.time())
    current_hour = current_time.tm_hour
    current_minute = current_time.tm_min

    for job in my_jobs:
        for t in job["time"].split(","):
            hour, minute = t.split(".")
            if int(hour) == current_hour and int(minute) == current_minute:
                city_code = get_city_code(job["city"])
                weather_info = get_weather(city_code)
                weather_str = dict_to_string(weather_info)
                subject = f"{job['city']} 天気情報"
                send_email(job["receivers"], subject, weather_str)

メール送信機能の実装

今回はQQメールを使用して通知を行います。SMTPサーバーはsmtp.qq.com、ポート番号は465を使用します。

メール送信処理

import smtplib
from email.mime.text import MIMEText
from email.utils import formataddr
from email.header import Header

mail_host = "smtp.qq.com"
mail_port = 465
mail_user = "your@qq.com"
mail_pass = "your_authorization_code"
sender = mail_user

def send_email(receivers, title, text):
    message = MIMEText(text, 'plain', 'utf-8')
    message['From'] = formataddr(["朝十晚八", mail_user])
    message['To'] = Header(','.join(receivers), 'utf-8')
    message['Subject'] = Header(title, 'utf-8')

    try:
        smtp = smtplib.SMTP_SSL(mail_host, mail_port)
        smtp.login(mail_user, mail_pass)
        smtp.sendmail(sender, receivers, message.as_string())
        return True
    except Exception as e:
        print(f"メール送信失敗: {e}")
        return False

テスト送信コード

send_email(["test@example.com"], "天気情報", "今日は晴れです。")

動作確認とログ

サービスが正しく動作しない場合、Windowsイベントビューアーでログを確認できます。

• コマンド：eventvwr.exe

ログファイルを出力する機能も追加し、メール送信状況を記録します。