Lambdaによる営業日判定

2022.08.01
こんにちは。磯野です。
 
AWSを用いた案件の中で、Stepfunctionsを「営業日のみ自動実行させたい」という要件がありました。
営業日判定では、祝日や年末年始などの会社休日を考慮する必要があるため、Eventbridgeだけでは実装できません。
そこで今回は、営業日判定ができるLambdaを作成して、本要件を達成することとしました。
※本LambdaはStepfunctions以外にも、EC2の起動停止等にも応用できます。
 
なお、会社休日の判定が不要で、曜日判定のみで良い場合は、Lambdaの実装は不要となります。
Eventbridgeで実装できるため、以下の記事を参考にしていただければと思います。
 
AWS Cron式で「第○○曜日」を設定する方法 - Qiita
AWSでCloudWatch Eventsを使おうとして、クーロン式で「第2日曜日」、「第3土曜日」みたく月次で曜日を指定して動かしたくなったので、調べてみた。設定方法少し調べた結果、SUN#4等でいけるっぽいです…
qiita.com

本Lambdaを使用するケース

Stepfunctionsを自動で定期実行させる際は、一般的にEventbridgeでスケジュール設定を行います。
Eventbridgeでは曜日判定や日付判定は可能ですが、祝日や年末年始などを考慮する営業日判定はできません。
そこで、今回作成するLambdaの出番となります。
 
今回作成するLambdaは、以下のようなケースの解決に役立ちます。
  • 毎日△時にStepfunctionsを実行したい。ただし、土日祝や年末年始などの会社休日は実行したくない（＝営業日のみ実行したい）
    ※Eventbridgeだけでは、その日が会社休日かに関わらず、月曜日～金曜日に実行されてしまう。
  • 毎週○曜日△時にStepfunctionsを実行したい。ただし、○曜日が会社休日の場合は翌営業日に実行させたい。
    ※Eventbridgeだけでは、その日が会社休日かに関わらず、毎週○曜日に実行されてしまう。
  • 毎月第x番目の○曜日△時にStepfunctionsを実行したい。ただし、○曜日が会社休日の場合は翌営業日に実行させたい。
    ※Eventbridgeだけでは、その日が会社休日かに関わらず、毎月第x番目の○曜日に実行されてしまう。
    ※xを複数指定することも可能

なお、他の方法としてAWS Systems Manager Change Calendarでも実装可能です。ただし、毎年カレンダーの更新が発生するため、手動運用を極力避けたい場合はLambdaによる実装がおすすめです。

AWS Systems Manager Change Calendarで祝日を設定したジョブ実行が可能になりました | DevelopersIO
AWS 内でジョブ実行する際に祝日設定したいと以前から感じていました
dev.classmethod.jp

構成図

以下は、毎月第x番目の○曜日（会社休日なら翌営業日）に実行する場合の構成図ですが、「毎日」もしくは「毎週○曜日」に実行する場合も同様の構成図となります。

 

Lambdaのソースコード

本記事では、会社休日を「日本の祝日」+「指定した日付（年末年始など）」として定義しています。

「日本の祝日」はGoogleカレンダーから取得しております。その他の日付で休日がある場合は、company_holidays.pyで定義しております。要件に合わせて修正してください。

なお、EventbridgeやStepfunctionsのテンプレートは省略します。

毎週○曜日（会社休日なら翌営業日）を判定するソースコード

lambda_function.py

"""dow曜日かの判定を行う。会社休日なら翌営業日

Returns:
    string: 
        Nextなら「本日が、今週のdow曜日(会社休日なら翌営業日)」
        Endなら「本日は、今週のdow曜日(会社休日なら翌営業日)ではない」
        Failedなら、Lambdaの実行に失敗。
"""
from curses import wrapper
import sys
import os
import json
import logging
import datetime
from datetime import date

import requests
import workdays
from workdays import *
from dateutil.relativedelta import *
from icalendar import Calendar, Event
import calendar

import company_holidays

# ログ設定
Level = str(os.environ["LOG_LEVEL"])
logger = logging.getLogger()
logger.setLevel(Level)

# 変数設定
REGION_NAME = str(os.environ["REGION_NAME"])
CALENDAR_URL = str(os.environ["CALENDAR_URL"])
DAY_OF_WEEK = str(os.environ["DAY_OF_WEEK"])

#今週のdow曜日の日付(dd)を取得
def get_target_date(date, dow):
    '''dow: Monday(0) - Sunday(6)'''
    # 曜日を数値型で取得
    weekday = date.weekday()
    # dateから指定した曜日までの加算日数を計算
    add_days = dow - weekday
    # dateに加算
    target_date = date + datetime.timedelta(days = add_days)
    return target_date

# メイン処理
def lambda_handler(event, context):

    try:
        # 実行日付取得
        today_datetime = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
        today = today_datetime.date()
        print("Today: " + str(today))

        # Googleカレンダー祝日取得
        cal_request = requests.get(CALENDAR_URL)
        cal = Calendar.from_ical(cal_request.text)
        holidays = []

        for ev in cal.walk():
            if ev.name == "VEVENT":
                holiday = ev.decoded("dtstart")
                holidays.append(holiday)

        # 会社休日取得
        for ch in company_holidays.company_holiday_list:
            company_holiday = datetime.datetime.strptime(ch, '%Y-%m-%d')
            holidays.append(company_holiday.date())

        print("Holidays: " + str(holidays))

        #今週のdow曜日取得
        num_week_dow = get_target_date(today,int(DAY_OF_WEEK))
        print("ThisWeekDowDay: " + str(num_week_dow))

        #今週のdow曜日が会社休日なら翌営業日を取得(今週のdow曜日の前日から1日後営業日を取得)
        work_day = workdays.workday(num_week_dow+timedelta(days=-1), days=1, holidays=holidays)
        print("WorkDay: " + str(work_day))

        # 本日が、今週のdow曜日(会社休日なら翌営業日)かを確認
        if str(today) == str(work_day):
            logger.info("==========Start job==========")
            return "Next"
        else:
            logger.info("Execution date and working day do not match...")
            print(str(work_day) + " Execution date and working day do not match...")
            return "End"

    except Exception as e:
        logger.error("Working Days Check Error : %s", e)
        return "Failed"

company_holidays.py

# 会社休日設定
company_holiday_list = [
    '2020-12-30', '2020-12-31', '2021-01-02', '2021-01-03'
    , '2021-12-29', '2021-12-30', '2021-12-31', '2022-01-02', '2022-01-03'
    , '2022-12-29', '2022-12-30', '2022-12-31', '2023-01-02', '2023-01-03'
    , '2023-12-29', '2023-12-30', '2023-12-31', '2024-01-02', '2024-01-03'
    ]

CloudFormationテンプレート

以下は、「毎週月曜（会社休日なら翌営業日）」を判定するテンプレートですが、“DAY_OF_WEEK”をカスタマイズすることで、他の曜日へ変更可能です。

  WorkdayCheckLambdaEveryWeekDowCheck:
    Type: 'AWS::Lambda::Function'
    Properties:
      FunctionName: workday-check-lambda-every-week-dow-check
      Handler: lambda_function.lambda_handler
      Role: !GetAtt IamRoleWorkdayCheck.Arn
      Runtime: "python3.7"
      Timeout: "10"
      Environment:
        Variables:
          "LOG_LEVEL": "INFO"
          "CALENDAR_URL": "https://calendar.google.com/calendar/ical/ja.japanese%23holiday%40group.v.calendar.google.com/public/basic.ics"
          "REGION_NAME": "ap-northeast-1"
          "DAY_OF_WEEK": "0"
      Tags:
        - Key: Name
          Value: workday-check-lambda-every-week-dow-check
      Code:
        S3Bucket: my-s3-bucket
        S3Key: lambda/functions/workday-check-lambda-every-week-dow-check.zip

毎月第x番目の○曜日（会社休日なら翌営業日）を判定するソースコード

lambda_function.py

"""第num_list週目のdow_list曜日かの判定を行う。会社休日なら翌営業日

Returns:
    string: 
        Nextなら「本日が、第num_list週目のdow_list曜日」
        Endなら「本日は、第num_list週目のdow_list曜日ではない」
        Failedなら、Lambdaの実行に失敗。
"""
from curses import wrapper
import sys
import os
import json
import logging
import datetime
from datetime import date

import requests
import workdays
from workdays import *
from dateutil.relativedelta import *
from icalendar import Calendar, Event
import calendar

import company_holidays

# ログ設定
Level = str(os.environ["LOG_LEVEL"])
logger = logging.getLogger()
logger.setLevel(Level)

# 変数設定
REGION_NAME = str(os.environ["REGION_NAME"])
CALENDAR_URL = str(os.environ["CALENDAR_URL"])
WEEK_NUM_LIST =  [int(num) for num in str(os.environ["WEEK_NUM_LIST"]).split()] #ex)"1 2"→[1, 2]
DAY_OF_WEEK_LIST =  [int(day) for day in str(os.environ["DAY_OF_WEEK_LIST"]).split()] #ex)"1 2"→[1, 2]

#第num_list週目のdow_list曜日の日付(dd)を取得(num_listに複数の値が入っている場合、複数の日付を返す)
def get_day_of_nth_dow(year, month, num_list, dow_list):
    '''dow: Monday(0) - Sunday(6)'''
    day_list = []
    for num in num_list:
        for dow in dow_list:
            if num < 1 or dow < 0 or dow > 6:
                return None
            first_dow, n = calendar.monthrange(year, month)
            day = 7 * (num - 1) + (dow - first_dow) % 7 + 1
            if day > n:
                return None
            day_list.append(day)

    return day_list

#第num_list週目のdow_list曜日の日付(yyyy-mm-dd)を取得
def get_date_of_nth_dow(year, month, num_list, dow_list):
    day_list = get_day_of_nth_dow(year, month, num_list, dow_list)
    day_list_f = []
    for day in day_list:
        day_f = datetime.date(year, month, day)
        if day_f:
            day_list_f.append(day_f)
        else:
            return None
    return day_list_f

# メイン処理
def lambda_handler(event, context):

    try:
        # 実行日付取得
        today_datetime = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9)))
        today = today_datetime.date()
        print("Today: " + str(today))

        # Googleカレンダー祝日取得
        cal_request = requests.get(CALENDAR_URL)
        cal = Calendar.from_ical(cal_request.text)
        holidays = []

        for ev in cal.walk():
            if ev.name == "VEVENT":
                holiday = ev.decoded("dtstart")
                holidays.append(holiday)

        # 会社休日取得
        for ch in company_holidays.company_holiday_list:
            company_holiday = datetime.datetime.strptime(ch, '%Y-%m-%d')
            holidays.append(company_holiday.date())

        print("Holidays: " + str(holidays))

        # 月初日取得
        first_day = date(today.year, today.month, 1)
        print("FirstDay: " + str(first_day))

        #第num週目のdow曜日取得
        num_week_dow_list = get_date_of_nth_dow(today.year,today.month,WEEK_NUM_LIST,DAY_OF_WEEK_LIST)
        print("NumWeekDow: " + str(num_week_dow_list))

        #第num週目のdow曜日が会社休日なら翌営業日を取得(第num週目のdow曜日の前日から1日後営業日を取得)
        workday_list = []
        for num_week_dow in num_week_dow_list:
            work_day = workdays.workday(num_week_dow+timedelta(days=-1), days=1, holidays=holidays)
            workday_list.append(work_day)
        print("WorkDay: " + str(workday_list))

        # 本日が、第num週目のdow曜日(会社休日なら翌営業日)かを確認
        check_flg = False
        for work_day in workday_list:
            if str(today) == str(work_day):
                logger.info("==========Start job==========")
                check_flg = True
                return "Next"
            else:
                pass

        if check_flg == False:
            logger.info("Execution date and working day do not match...")
            print(str(workday_list) + " Execution date and working day do not match...")
            return "End"

    except Exception as e:
        logger.error("Working Days Check Error : %s", e)
        return "Failed"

company_holidays.py

前述のものと同一のため省略します。

CloudFormationテンプレート

以下は、「毎月第2月曜・第4月曜（会社休日なら翌営業日）」を判定するテンプレートですが、”WEEK_NUM_LIST”や“DAY_OF_WEEK_LIST”をカスタマイズすることで、他の週数や曜日へ変更可能です。

  WorkdayCheckLambdaNumWeekDowCheck:
    Type: 'AWS::Lambda::Function'
    Properties:
      FunctionName: workday-check-lambda-num-week-dow-check
      Handler: lambda_function.lambda_handler
      Role: !GetAtt IamRoleWorkdayCheck.Arn
      Runtime: "python3.7"
      Timeout: "10"
      Environment:
        Variables:
          "LOG_LEVEL": "INFO"
          "CALENDAR_URL": "https://calendar.google.com/calendar/ical/ja.japanese%23holiday%40group.v.calendar.google.com/public/basic.ics"
          "REGION_NAME": "ap-northeast-1"
          "WEEK_NUM_LIST": "2 4"
          "DAY_OF_WEEK_LIST": "0"
      Tags:
        - Key: Name
          Value: workday-check-lambda-num-week-dow-check
      Code:
        S3Bucket: my-s3-bucket
        S3Key: lambda/functions/workday-check-lambda-num-week-dow-check.zip

まとめ

いかがだったでしょうか。

今回は、営業日判定を行うLambdaについてご紹介しました。

本記事が皆様のお役に立てれば幸いです。

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