【Amazon Connect Customer】オペレーション時間をカレンダーとして AWS Lambda で配送可能日を判定させてみた
Amazon Connect Customer のオペレーション時間と AWS Lambda を活用して、配送可能日を自動判定する仕組みを実装してみました。顧客からの入力日付に対して営業日判定を行い、適切な処理分岐を行う方法について紹介します。
2026.07.24
はじめに
こんにちは、フニです。
本記事では、Amazon Connect Customer インスタンスのオペレーション時間と AWS Lambda を活用し、配送可能日の判定を実装する方法について解説します。
やってみた
前提
- Amazon Connect Customer インスタンス構築済み
- インスタンス内に利用可能な電話番号が設定されていること
- AWS Lambda は Python 3.14 を利用
フローの全体像
今回の検証には、以下のフローを利用します。
- ①：お客様から電話番号を入力してもらい、属性として保存します
- ②：AWS Lambda にパラメータとして渡し、オペレーション時間の営業日判定によって処理を分岐します
- ③：
statusの値が OPEN か CLOSED かによって異なるアナウンスを再生します
フローの補足
配送希望日で例えると、7月25日の場合 0725 を入力してもらいます。
それを $.Attributes.Date 属性に保存しておきます。
AWS Lambda には、以下のように $.Attributes.Date の値を inputDate パラメータとして渡します。
オペレーション時間の判定により $.External.status に値が保存されます。
OPEN か CLOSED かによって処理を分岐します。
- OPEN：オペレーション時間が営業日の場合
- CLOSED：オペレーション時間が休業日の場合
オペレーション時間設定
以下のように、日本時間の月曜日から金曜日を営業日として指定しました。
テストのため、もともと営業日だった7/27(月)はオーバーライド設定で休業日にしてます。
カレンダービューで確認すると、営業日と休業日が一覧として確認できます。
AWS Lambda 設定
環境変数には、CONNECT_INSTANCE_ID と HOURS_OF_OPERATION_ID を登録しました。
作成したオペレーション時間画面からどちらも確認いただけます。
オペレーション時間を確認する権限が必要になるため、AWS Lambda の実行ロール側には以下のポリシーを付与しました。
本番環境等では、必要に応じて * を最小権限に変更することを推奨します。
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "AllowHoursOfOperationRead",
"Effect": "Allow",
"Action": [
"connect:DescribeHoursOfOperation",
"connect:ListHoursOfOperationOverrides"
],
"Resource": "arn:aws:connect:*:*:instance/*/operating-hours/*"
}
]
}
AWS Lambda コードは以下のサンプルコードを利用しました。
必要に応じてご変更の上ご利用ください。
import os
from datetime import date, datetime, timedelta
from zoneinfo import ZoneInfo
import boto3
# クライアントはモジュールレベルで生成し、実行環境間で再利用する
connect_client = boto3.client("connect")
INSTANCE_ID = os.environ["CONNECT_INSTANCE_ID"]
HOURS_OF_OPERATION_ID = os.environ["HOURS_OF_OPERATION_ID"]
# Connect API の曜日文字列 -> Python の weekday() 値(月曜=0)の対応表
DAY_TO_WEEKDAY = {
"MONDAY": 0,
"TUESDAY": 1,
"WEDNESDAY": 2,
"THURSDAY": 3,
"FRIDAY": 4,
"SATURDAY": 5,
"SUNDAY": 6,
}
# 1日を分単位で表現する際の上限(24:00)
MINUTES_PER_DAY = 1440
def lambda_handler(event: dict, context) -> dict:
"""Connect フローのエントリーポイント。
フロー側は「AWS Lambda 関数を呼び出す」ブロックのパラメータに
inputDate (MMDD 形式) を設定して呼び出すこと。
戻り値はフラットな文字列マップ(Connect の制約)で、
status / targetDate / errorMessage を含む。
"""
params = event.get("Details", {}).get("Parameters", {})
input_date = str(params.get("inputDate", "")).strip()
try:
hoop = describe_hours_of_operation()
tz = ZoneInfo(hoop["TimeZone"])
target = parse_mmdd(input_date, today=datetime.now(tz).date())
overrides = list_effective_overrides(target)
open_intervals = resolve_open_intervals(target, hoop["Config"], overrides)
is_open = len(open_intervals) > 0
return {
"status": "OPEN" if is_open else "CLOSED",
"targetDate": target.isoformat(),
"errorMessage": "",
}
except ValueError as e:
# 入力日付が不正な場合はフロー側でエラー分岐できるように返す
return {
"status": "INVALID_INPUT",
"targetDate": "",
"errorMessage": str(e),
}
def parse_mmdd(input_date: str, today: date) -> date:
"""MMDD 文字列を日付に変換する。
今日より過去の日付が指定された場合は翌年の日付として解釈する
(配送日は未来日である前提のため)。
"""
if len(input_date) != 4 or not input_date.isdigit():
raise ValueError(f"日付は MMDD の4桁で入力してください: '{input_date}'")
month = int(input_date[:2])
day = int(input_date[2:])
try:
target = date(today.year, month, day)
except ValueError:
raise ValueError(f"存在しない日付です: 月={month} 日={day}")
if target < today:
# 年をまたぐ入力(例: 12月に 0115 と入力)は翌年扱い
try:
target = date(today.year + 1, month, day)
except ValueError:
# 2/29 など翌年に存在しない日付
raise ValueError(f"翌年に存在しない日付です: 月={month} 日={day}")
return target
def describe_hours_of_operation() -> dict:
"""Hours of Operation の標準設定(タイムゾーン・曜日別時間帯)を取得する。"""
response = connect_client.describe_hours_of_operation(
InstanceId=INSTANCE_ID,
HoursOfOperationId=HOURS_OF_OPERATION_ID,
)
return response["HoursOfOperation"]
def list_effective_overrides(target: date) -> list[dict]:
"""対象日に有効期間が重なるオーバーライドのみを抽出して返す。"""
paginator = connect_client.get_paginator("list_hours_of_operation_overrides")
effective = []
for page in paginator.paginate(
InstanceId=INSTANCE_ID,
HoursOfOperationId=HOURS_OF_OPERATION_ID,
):
for override in page.get("HoursOfOperationOverrideList", []):
if is_within_effective_range(override, target):
effective.append(override)
return effective
def is_within_effective_range(override: dict, target: date) -> bool:
"""EffectiveFrom / EffectiveTill (YYYY-MM-DD) の範囲内かを判定する。"""
effective_from = override.get("EffectiveFrom")
effective_till = override.get("EffectiveTill")
if effective_from and target < date.fromisoformat(effective_from):
return False
if effective_till and target > date.fromisoformat(effective_till):
return False
return True
def resolve_open_intervals(
target: date, standard_config: list[dict], overrides: list[dict]
) -> list[tuple[int, int]]:
"""対象日の最終的な営業時間帯(分単位の区間リスト)を算出する。
Connect の優先順位に従い、ベースとなる時間帯を組み立ててから
OPEN 繰り返しを加算し、CLOSED 繰り返しを減算する。
空リストが返れば終日休業(CLOSED)を意味する。
"""
weekday = target.weekday()
# ベース: 標準の曜日別営業時間
intervals = config_intervals_for_weekday(standard_config, weekday)
# STANDARD(臨時営業時間)が対象日に有効なら、標準スケジュールを置き換える
# Config が空の場合は「その日は営業時間帯なし = 終日休業」を意味する
for override in overrides:
if override_type_of(override) == "STANDARD":
intervals = config_intervals_for_weekday(
override.get("Config", []), weekday
)
# OPEN 繰り返しオーバーライド: 該当パターンに一致すれば時間帯を追加
for override in overrides:
if override_type_of(override) == "OPEN" and recurrence_matches(
override, target
):
intervals = merge_intervals(
intervals + config_intervals_for_weekday(
override.get("Config", []), weekday, match_any_day=True
)
)
# CLOSED 繰り返しオーバーライド: 最優先。該当時間帯を除外する
for override in overrides:
if override_type_of(override) == "CLOSED" and recurrence_matches(
override, target
):
closed = config_intervals_for_weekday(
override.get("Config", []), weekday, match_any_day=True
)
# Config が空の CLOSED は終日休業とみなす
if not closed:
closed = [(0, MINUTES_PER_DAY)]
intervals = subtract_intervals(intervals, closed)
return intervals
def override_type_of(override: dict) -> str:
"""OverrideType を返す。未設定の場合は STANDARD(臨時営業時間)として扱う。
コンソールの「臨時営業時間を作成」で作成したオーバーライドは
API レスポンスに OverrideType が含まれないことがあるため、
デフォルトを STANDARD とする。
"""
return override.get("OverrideType") or "STANDARD"
def config_intervals_for_weekday(
config: list[dict], weekday: int, match_any_day: bool = False
) -> list[tuple[int, int]]:
"""Config 配列から対象曜日の時間帯を (開始分, 終了分) のリストで返す。
EndTime が 00:00 の場合は「翌日の深夜0時」= 24:00 として扱う
(管理者ガイドの midnight 指定ルールに準拠)。
match_any_day=True の場合は曜日を問わず全エントリを対象とする
(繰り返しオーバーライドは recurrence 側で日付一致を判定済みのため)。
"""
intervals = []
for entry in config:
if not match_any_day and DAY_TO_WEEKDAY.get(entry.get("Day", "")) != weekday:
continue
start = entry["StartTime"]["Hours"] * 60 + entry["StartTime"]["Minutes"]
end = entry["EndTime"]["Hours"] * 60 + entry["EndTime"]["Minutes"]
if end == 0:
end = MINUTES_PER_DAY
if start < end:
intervals.append((start, end))
return merge_intervals(intervals)
def recurrence_matches(override: dict, target: date) -> bool:
"""繰り返しオーバーライド(OPEN/CLOSED)が対象日に該当するかを判定する。
RecurrencePattern の仕様:
Frequency: WEEKLY | MONTHLY | YEARLY
Interval: 1〜6 (頻度単位の間隔)
ByMonth: 対象月 (YEARLY 用、単一値)
ByMonthDay: 対象日 (1〜31、-1 は月末)
ByWeekdayOccurrence: 月内の第 n 曜日 (-1 は最終)、曜日は Config の Day で指定
"""
recurrence = override.get("RecurrenceConfig", {}).get("RecurrencePattern")
if not recurrence:
# 繰り返し設定がない OPEN/CLOSED は有効期間内なら常に該当とみなす
return True
frequency = recurrence["Frequency"]
interval = recurrence.get("Interval", 1)
anchor = date.fromisoformat(override["EffectiveFrom"])
if frequency == "WEEKLY":
return weekly_matches(recurrence, override, target, anchor, interval)
if frequency == "MONTHLY":
return monthly_matches(recurrence, override, target, anchor, interval)
if frequency == "YEARLY":
return yearly_matches(recurrence, override, target, anchor, interval)
return False
def weekly_matches(
recurrence: dict, override: dict, target: date, anchor: date, interval: int
) -> bool:
"""WEEKLY: Config の曜日に一致し、かつ週間隔が Interval の倍数であること。"""
if not day_in_config(override, target):
return False
# EffectiveFrom を含む週(月曜起点)からの経過週数で間隔を判定する
anchor_week_start = anchor - timedelta(days=anchor.weekday())
target_week_start = target - timedelta(days=target.weekday())
weeks_elapsed = (target_week_start - anchor_week_start).days // 7
return weeks_elapsed % interval == 0
def monthly_matches(
recurrence: dict, override: dict, target: date, anchor: date, interval: int
) -> bool:
"""MONTHLY: 月間隔が Interval の倍数で、日付条件に一致すること。"""
months_elapsed = (target.year - anchor.year) * 12 + (target.month - anchor.month)
if months_elapsed % interval != 0:
return False
return day_condition_matches(recurrence, override, target)
def yearly_matches(
recurrence: dict, override: dict, target: date, anchor: date, interval: int
) -> bool:
"""YEARLY: 年間隔が Interval の倍数で、月・日条件に一致すること。"""
if (target.year - anchor.year) % interval != 0:
return False
by_month = recurrence.get("ByMonth")
if by_month and target.month not in by_month:
return False
return day_condition_matches(recurrence, override, target)
def day_condition_matches(recurrence: dict, override: dict, target: date) -> bool:
"""ByMonthDay または ByWeekdayOccurrence + Day の日付条件を評価する。"""
by_month_day = recurrence.get("ByMonthDay")
if by_month_day:
return month_day_matches(by_month_day, target)
by_weekday_occurrence = recurrence.get("ByWeekdayOccurrence")
if by_weekday_occurrence:
return weekday_occurrence_matches(by_weekday_occurrence, override, target)
# 日付条件の指定がなければ月・年の一致のみで該当とする
return True
def month_day_matches(by_month_day: list[int], target: date) -> bool:
"""ByMonthDay の判定。-1 は月末日を意味する。"""
for month_day in by_month_day:
if month_day == -1:
if target.day == last_day_of_month(target).day:
return True
elif target.day == month_day:
return True
return False
def weekday_occurrence_matches(
by_weekday_occurrence: list[int], override: dict, target: date
) -> bool:
"""第 n 曜日の判定。曜日は Config の Day、n は -1(最終) または 1〜4。"""
if not day_in_config(override, target):
return False
occurrence = by_weekday_occurrence[0]
if occurrence == -1:
# 最終出現: 対象日の7日後が翌月になるか
return (target + timedelta(days=7)).month != target.month
# 第 n 出現: 月内で何回目のその曜日か
return (target.day - 1) // 7 + 1 == occurrence
def day_in_config(override: dict, target: date) -> bool:
"""対象日の曜日が override の Config の Day に含まれるかを判定する。"""
weekdays = {
DAY_TO_WEEKDAY[entry["Day"]]
for entry in override.get("Config", [])
if entry.get("Day") in DAY_TO_WEEKDAY
}
return target.weekday() in weekdays
def last_day_of_month(target: date) -> date:
"""対象月の月末日を返す。"""
next_month = date(target.year + (target.month // 12), target.month % 12 + 1, 1)
return next_month - timedelta(days=1)
def merge_intervals(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]:
"""重複・隣接する時間帯を統合する。"""
if not intervals:
return []
merged = []
for start, end in sorted(intervals):
if merged and start <= merged[-1][1]:
merged[-1] = (merged[-1][0], max(merged[-1][1], end))
else:
merged.append((start, end))
return merged
def subtract_intervals(
intervals: list[tuple[int, int]], closed: list[tuple[int, int]]
) -> list[tuple[int, int]]:
"""営業時間帯から休業時間帯を差し引く。"""
result = intervals
for closed_start, closed_end in closed:
next_result = []
for start, end in result:
# 休業帯と重ならない部分だけを残す
if end <= closed_start or start >= closed_end:
next_result.append((start, end))
continue
if start < closed_start:
next_result.append((start, closed_start))
if end > closed_end:
next_result.append((closed_end, end))
result = next_result
return result
結果
オペレーション時間が営業日の場合
営業日の場合、status が OPEN になるため 配送可能日です とアナウンスが流れてきました。
{
"ExternalResults": {
"status": "OPEN",
"targetDate": "2026-07-28",
}
}
{
"ContactFlowModuleType": "PlayPrompt",
"Parameters": {
"Text": "配送可能日です",
}
}
オペレーション時間が休業日の場合
休業日の場合、status が CLOSED になるため 配送できない日です とアナウンスが流れてきました。
{
"ExternalResults": {
"status": "CLOSED",
"targetDate": "2026-07-25",
}
}
{
"ContactFlowModuleType": "PlayPrompt",
"Parameters": {
"Text": "配送できない日です",
}
}
オーバーライド設定による例外的な終日休業日の場合
休業日と同じく、status が CLOSED になり 配送できない日です とアナウンスが流れてきました。
{
"ExternalResults": {
"status": "CLOSED",
"targetDate": "2026-07-27",
}
}
{
"ContactFlowModuleType": "PlayPrompt",
"Parameters": {
"Text": "配送できない日です",
}
}
さいごに
今回は、Amazon Connect Customer インスタンスのオペレーション時間と AWS Lambda を活用し、配送可能日の判定を実装する方法について解説しました。
DB 側でこのような処理を実装するケースもあるかと思いますが、このように Amazon Connect Customer インスタンスだけで完結できるのは嬉しいです。
この記事が、誰かの助けになれば幸いです。
参考
