Pythonサービス化して処理を常駐

windowsサービス化[pywin32]

Pythonプログラムをbatで実行してもいいが、何を起動させているかわからなくなるので起動しているプログラムをサービス化できるとユーザーにやさしい。

Windows サービスはバックグラウンド(コマンドプロンプトや画面などには出ない)で実行されるプログラムです。
例えるなら、ウイルス対策ソフトなど起動した覚えはないけど、PC起動したら動いているし、異常があったら報告してくれますよね。同じようにPythonのプログラムをバックグランドで動くサービスに登録したり削除したりするのがこの記事の目的です。

こんな方に

Pythonで自動処理など作ったが、

• この黒い画面はどの処理をしているんだ？状態になってる
• エンジニア以外の人でも自動処理しているPythonプログラムの状態を確認したい

という人向け。

ついでに、サービスは
「Windows管理ツール」⇒「サービス」
で確認できる。

設定方法

1. システムの環境変数にpathを通す

コントロール パネル\システムとセキュリティ\システム
システムの詳細設定 → 環境変数

システムの環境変数のPathに以下があるか確認

• ユーザーの環境変数でないので注意
• Python36は使用しているバージョンで変わります

C:\Users\〇〇〇〇\AppData\Local\Programs\Python\Python36\
C:\Users\〇〇〇〇\AppData\Local\Programs\Python\Python36\Scripts\

また、以下を追加

C:\Users\〇〇〇〇\AppData\Local\Programs\Python\Python36\Lib\site-packages\pywin32_system32
C:\Users\〇〇〇〇\AppData\Local\Programs\Python\Python36\Lib\site-packages\win32

※下のシステム環境変数のPathに追加

2.管理者権限でコマンドプロンプトから実行

メモ：管理者から実行の方法
検索でcmd → コマンドプロンプトアプリ → 右クリックから管理者から実行

pip install pywin32

3.サービスの登録（自動 遅延開始）

システム化するPythonファイルがあるディレクトリに移動後、管理者権限でコマンドプロンプトを開き、下記のコマンドを実行します。

$ python test.py --startup delayed install

4.サービスの実行

管理者権限でコマンドプロンプトを開き、下記のコマンドを実行します。

$ python test.py start

5.サービスの停止

管理者権限でコマンドプロンプトを開き、下記のコマンドを実行します。

$ python test.py stop

6.サービスの削除

管理者権限でコマンドプロンプトを開き、下記のコマンドを実行します。

$ python test.py remove

サンプルコード1

# -*- coding:utf-8 -*-

import win32serviceutil
import win32service
import win32event
import servicemanager
import socket

import time
import threading
import logging
import random

logging.basicConfig(
    filename = 'c:\\work\\test-service.log',
    level = logging.DEBUG, 
    format="%(asctime)s:LINE[%(lineno)s] %(levelname)s %(message)s"
)

INTERVAL = 10

class MySvc (win32serviceutil.ServiceFramework):
    _svc_name_ = "test-service"
    _svc_display_name_ = "test service"

    # Classの初期化
    def __init__(self,args):
        win32serviceutil.ServiceFramework.__init__(self,args)
        self.stop_event = win32event.CreateEvent(None,0,0,None)
        socket.setdefaulttimeout(60)
        self.stop_requested = False

    # サービス停止
    def SvcStop(self):
        self.ReportServiceStatus(win32service.SERVICE_STOP_PENDING)
        win32event.SetEvent(self.stop_event)
        logging.info('Request to Stop Service...')
        self.stop_requested = True

    # サービス開始
    def SvcDoRun(self):
        servicemanager.LogMsg(
            servicemanager.EVENTLOG_INFORMATION_TYPE,
            servicemanager.PYS_SERVICE_STARTED,
            (self._svc_name_,'')
        )

        # メインループ関数呼び出し
        self.main_loop()

    # メイン処理関数
    def main_task(self):
        logging.debug('mainTask Start...55sec sleep')
        # time.sleep(55)
        logging.debug('Do Something...after 55sec')

    # メインループ関数
    def main_loop(self):
        logging.info('サービス開始')
        exec_time = time.time()

        # ループ処理
        while True:
            # 終了要求確認
            if self.stop_requested:
                logging.info('サービスの停止します')
                break

            try: 
                # 実行日時を超えている場合
                if exec_time <= time.time():

                    # Main処理呼び出し
                    self.main_task()

                    # 次に実行する時刻を設定
                    exec_time = exec_time + INTERVAL

            except Exception as e:
                logging.error("Error occured.")

            # 0.1秒スリープ
            time.sleep(0.1)

        logging.info("サービス停止")
        return

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

サンプルコード2

servicetest.py

import os
import win32service 
import win32serviceutil 
import win32event 
import datetime 

class SmallestPythonService(win32serviceutil.ServiceFramework): 
    # サービス名 
    _svc_name_ = "TESTServise" 
    # 表示名(サービス画面にはこれを表示) 
    _svc_display_name_ = "TEST Service" 
    # サービスの説明 
    _svc_description_='一定周期でデータをテーブルに保存する' 
    # シグナルまでの待ち用タイムアウト時間(今回は10秒 = 10,000ミリ秒) 
    _timeout_Milliseconds = 10000 

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

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

    def SvcDoRun(self): 
        print('サービス開始') 
        while 1: 
            # イベントのシグナル化を10秒待つ 
            ret = win32event.WaitForSingleObject( 
                    self.hWaitStop, 
                    self._timeout_Milliseconds 
                    ) 
                # サービス停止(イベントがシグナル化)ならば、処理を中止 
            if ret == win32event.WAIT_OBJECT_0: 
                break 

            self.main_loop() 

    # 実際のサービスの処理 
    def main_loop(self): 

        # print('サービス開始') 
        # ↓↓ test用のため削除　実行ディレクトリにtest.txt作成されたらテストOK
        FILEADDR = os.path.dirname(os.path.abspath(__file__)) + '/test.txt'
        with open(FILEADDR, "a") as f:
            f.write("[Test] %s \n" % (datetime.datetime.now())) 

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