Qiita Advent Calendar 2025
今年もこの季節がいよいよ始まりました 
誰よりもこの日を待ちわびていたと自負しております。
2024年12月26日から首を長くして楽しみにしておりました。
はじめに
この記事は、Works Human Intelligenceサンタ 様のプレゼント🎁カレンダー 
「「レガシー」を保守したり、刷新したりするにあたり得られた知見・ノウハウ・苦労話 by Works Human Intelligence Advent Calendar 2025」への応募記事です。
昨年も下記の記事を応募させていただきました。
今年はそのアップデートをお届けします。
それではまず、ベースとなる昨年の記事からの振り返りです。
問題点とその解決策
まず、問題点とその解決策を示します。
before（問題点）
私が所属するハウインターナショナルは、フルリモートワークを実現している会社です。本社は福岡県飯塚市にありますが、東京、愛知、さらには「闘魂の国」からもメンバーが働いています。この分散したチーム環境において、電話対応をどのように効率化するかが大きな課題でした。
電話対応の流れは次のようなものでした：
- 電話代行サービスが受けた電話の内容（誰から誰への電話だったのか）をメールで送信
- メールを受け取った担当者が内容を確認
- Slackで該当者にメンションを付けて連絡
このプロセスはシンプルに見えますが、図中の「担当者」の手動の作業が課題でした。電話はすぐに取り次ぐのが鉄則ですから、それまでしていた作業を一時中断して、該当者への連絡を優先せねばなりません。想像に難くない通り、「担当者」の負担は実は相当なものです。
after（解決策）
私はこの手動プロセスの「2番と3番」を自動化することに挑戦しました。その結果、新しいフローは次のように改善されました：
- 電話代行サービスが電話の内容をメールで送信
- ボットが、そのメールを解析し、Slackで該当する担当者にメンションを付けて電話があったことを自動通知
たったこれだけの改善ですが、業務効率は格段に向上しました。
担当者は、リンボーダンスをはじめました！
リンボーダンスは大げさですが、電話応対で作業を中断する必要がなくなり、その分他のタスクに集中できるようになったということを言いたいわけです。
昨年の記事の振り返りは以上です。それではいよいよ本題へと入ります。
今年は何を変えたの?
今年は以下を変えました。
- Ruby => Pythonへ置き換え
- EC2のcron実行 => AWS LambdaとAmazon EventBrigeで定期実行
- 手作業のAWSリソース構築をAWS CDKで構築
今年は、この仕組みが大変革を遂げた年になりました。
以下、ひとつひとつみていきます。
1. Ruby => Pythonへ置き換え
以前は存在した https://developers.google.com/gmail/api/quickstart/ruby がありません。アクセスすると、 https://developers.google.com/workspace/gmail/api/quickstart/js?hl=ja へリダイレクトされてしまいます。
私はなにせ、「Elixir実践入門 | 技術評論社」の著者陣の末席を汚すものなので、Elixirが一番得意だと自負しております。JavaScriptは雰囲気で気持ち、なんとなくわからないわけではないし、生成AIコーディングツールの力を借りればなんとかなりそうですが、JavaScriptよりは自信のあるPythonを選びました。公式ドキュメントを片手に実装しました。AWS Lambdaで動かしているコードを惜しげもなく公開しておきます。
1-1. Gmailを受信するコード
まずはGmailを受信するコードです。永続化しておきたい情報は、AWS Systems Manager Parameter Storeに保存しています。
awesome_gmail.py
import os
import base64
import json
import boto3
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
# Constants
ENVIRONMENT = os.getenv('ENVIRONMENT', 'local') # 'local' or 'aws'
CREDENTIALS_PATH = "credentials.json"
TOKEN_PATH = "token.json"
CREDENTIALS_PARAM = "<your credentials param name>"
TOKEN_PARAM = "<your token param name>"
SCOPES = ['https://www.googleapis.com/auth/gmail.readonly']
MAX_RESULTS = 20
class AwesomeGmail:
@staticmethod
def get_credentials():
"""Retrieve credentials.json content based on environment."""
if ENVIRONMENT == 'local':
with open(CREDENTIALS_PATH, 'r') as f:
return json.load(f)
elif ENVIRONMENT == 'aws':
ssm = boto3.client('ssm')
response = ssm.get_parameter(Name=CREDENTIALS_PARAM, WithDecryption=True)
return json.loads(response['Parameter']['Value'])
@staticmethod
def get_token():
"""Retrieve token.json content based on environment."""
if ENVIRONMENT == 'local':
if os.path.exists(TOKEN_PATH):
with open(TOKEN_PATH, 'r') as f:
return json.load(f)
return None
elif ENVIRONMENT == 'aws':
ssm = boto3.client('ssm')
try:
response = ssm.get_parameter(Name=TOKEN_PARAM, WithDecryption=True)
return json.loads(response['Parameter']['Value'])
except ssm.exceptions.ParameterNotFound:
return None
@staticmethod
def save_token(token):
"""Save token.json content based on environment."""
if ENVIRONMENT == 'local':
with open(TOKEN_PATH, 'w') as f:
json.dump(token, f)
elif ENVIRONMENT == 'aws':
ssm = boto3.client('ssm')
ssm.put_parameter(
Name=TOKEN_PARAM,
Value=json.dumps(token),
Type='SecureString',
Overwrite=True
)
@staticmethod
def authorize():
"""Authorize and return Gmail service credentials."""
creds = None
# Retrieve token
token_data = AwesomeGmail.get_token()
if token_data:
creds = Credentials.from_authorized_user_info(token_data, SCOPES)
# If there are no (valid) credentials available, let the user log in.
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
else:
credentials_data = AwesomeGmail.get_credentials()
flow = InstalledAppFlow.from_client_config(credentials_data, SCOPES)
creds = flow.run_local_server(port=0)
# Save the credentials for the next run
AwesomeGmail.save_token(json.loads(creds.to_json())) # Use to_json and convert to dict
return creds
@staticmethod
def list():
"""List Gmail message IDs."""
creds = AwesomeGmail.authorize()
service = build('gmail', 'v1', credentials=creds)
messages = []
next_page_token = None
query = f"from:{os.getenv('GMAIL_TO_SLACK_FROM', '')}"
while len(messages) < MAX_RESULTS:
result = service.users().messages().list(
userId='me',
maxResults=MAX_RESULTS,
pageToken=next_page_token,
q=query
).execute()
if 'messages' in result:
messages.extend(result['messages'])
next_page_token = result.get('nextPageToken')
if not next_page_token or len(messages) >= MAX_RESULTS:
break
return [msg['id'] for msg in messages]
@staticmethod
def get(message_id):
"""Get a specific Gmail message by ID."""
creds = AwesomeGmail.authorize()
service = build('gmail', 'v1', credentials=creds)
result = service.users().messages().get(
userId='me',
id=message_id,
format='full'
).execute()
payload = result.get('payload', {})
headers = payload.get('headers', [])
# Extract header values
def get_header_value(header_name):
for header in headers:
if header['name'].lower() == header_name.lower():
return header['value']
return ''
date = get_header_value('Date')
from_addr = get_header_value('From')
to_addr = get_header_value('To')
subject = get_header_value('Subject')
# Extract body
parts = payload.get('parts', [])
body_data = payload.get('body', {}).get('data')
if not body_data and parts:
body_data = ''.join(part.get('body', {}).get('data', '') for part in parts)
try:
body = base64.urlsafe_b64decode(body_data.encode()).decode('utf-8') if body_data else ''
except Exception:
body = ''
return {
'id': result['id'],
'date': date,
'from': from_addr,
'to': to_addr,
'subject': subject,
'body': body
}
if __name__ == '__main__':
ids = AwesomeGmail.list()
for id in ids:
message = AwesomeGmail.get(id)
print(message)
それでこの記事を書いている途中で思い出したことがあります。たしかRuby => Pythonの置き換えは、どれを利用したのかは忘れましたが、生成AIコーディングツールに大部分を作ってもらいました。たぶん、これは6月くらいの作業（思い出）なので私はそのころは、GitHub Copilotをよく利用させていただいていたと思います。いまは好みがかわって、Kiro CL((旧Amazon Q Developer CLI)をよく利用しています。余談です。
1-2. Slackにメッセージを投げ込むコード
Slackにメッセージを投げ込むコードを示します。Incoming Webhookを利用しています。
awesome_slack.py
import os
import json
import requests
URL = os.getenv('GMAIL_TO_SLACK_WEB_HOOK_URL')
CHANNEL = os.getenv('GMAIL_TO_SLACK_CHANNEL')
class AwesomeSlack:
@staticmethod
def post(text):
"""Send a message to Slack."""
payload = {
"text": text,
"channel": CHANNEL,
"username": "awesome_bot",
"icon_emoji": ":telephone:",
"link_names": 1
}
headers = {
"Content-Type": "application/json"
}
response = requests.post(URL, data=json.dumps(payload), headers=headers)
if response.status_code != 200:
raise Exception(f"Slack API error: {response.status_code}, {response.text}")
1-3. main処理
mainの処理を示します。 lambda_handler 関数が、AWS Lambdaのエントリーポイントです。
main.py
import os
import boto3
from datetime import datetime, timedelta, timezone
from awesome_gmail import AwesomeGmail
from awesome_slack import AwesomeSlack
from users import USERS
ENVIRONMENT = os.getenv('ENVIRONMENT', 'local') # 'local' or 'aws'
LAST_AT_LOCAL = 'last_at'
LAST_AT_PARAM = '<your last at param name>'
FROM = os.getenv('GMAIL_TO_SLACK_FROM')
def get_last_at():
"""Retrieve the last processed time."""
if ENVIRONMENT == 'local':
if os.path.exists(LAST_AT_LOCAL):
with open(LAST_AT_LOCAL, 'r') as f:
return datetime.strptime(f.read().strip(), '%a, %d %b %Y %H:%M:%S %z')
else:
return datetime.now(timezone.utc) - timedelta(minutes=5)
elif ENVIRONMENT == 'aws':
ssm = boto3.client('ssm')
try:
response = ssm.get_parameter(Name=LAST_AT_PARAM)
return datetime.strptime(response['Parameter']['Value'], '%a, %d %b %Y %H:%M:%S %z')
except ssm.exceptions.ParameterNotFound:
return datetime.now(timezone.utc) - timedelta(minutes=5)
def update_last_at(last_at):
"""Update the last processed time."""
if ENVIRONMENT == 'local':
with open(LAST_AT_LOCAL, 'w') as f:
f.write(last_at.strftime('%a, %d %b %Y %H:%M:%S %z'))
elif ENVIRONMENT == 'aws':
ssm = boto3.client('ssm')
ssm.put_parameter(
Name=LAST_AT_PARAM,
Value=last_at.strftime('%a, %d %b %Y %H:%M:%S %z'),
Type='String',
Overwrite=True
)
def lambda_handler(event, context):
"""AWS Lambda entry point."""
# Read the last processed time
last_at = get_last_at()
# Fetch and filter emails
ids = AwesomeGmail.list()
filtered = []
for msg_id in ids:
mail = AwesomeGmail.get(msg_id)
if FROM in mail.get('from', ''):
try:
# Remove "(UTC)" from the date string
mail_date_str = mail.get('date', '').replace(' (UTC)', '')
mail_date = datetime.strptime(mail_date_str, '%a, %d %b %Y %H:%M:%S %z')
if mail_date > last_at:
filtered.append(mail)
except ValueError as e:
print(f"[WARN] Could not parse date: {mail.get('date')} | From: {mail.get('from')} | Subject: {mail.get('subject')}")
continue
# Process filtered emails
for mail in filtered:
body = mail['body']
usernames = {username for key, username in USERS.items() if key in body}
text = ' '.join(usernames) + "\n" + body
AwesomeSlack.post(text)
print(f"Posted to Slack: {mail['date']}, {mail['from']}, {mail['subject']}, {text.replace('\n', ' ')}")
# Update the last processed time
if filtered:
last_mail = max(filtered, key=lambda mail: datetime.strptime(mail['date'], '%a, %d %b %Y %H:%M:%S %z'))
update_last_at(datetime.strptime(last_mail['date'], '%a, %d %b %Y %H:%M:%S %z'))
return {
"statusCode": 200,
"body": f"Processed {len(filtered)} emails."
}
if __name__ == '__main__':
lambda_handler(None, None)
ここでも、永続化しておきたい情報は、AWS Systems Manager Parameter Storeに保存しています。
USERSは、以下のような辞書型のデータです。
USERS = {
"山内": "@awesome",
"やまうち": "@awesome",
"ヤマウチ": "@awesome"
}
2. EC2のcron実行 => AWS LambdaとAmazon EventBrigeで定期実行
以前はEC2で動かしていました。サーバーレスに目覚めたので、AWS LambdaとAmazon EventBrigeで定期実行をするようにしました。これにより利用料金は大幅に抑えられるようになりました。
3. 手作業のAWSリソース構築をAWS CDKで構築
これまで、AWSでリソース構築をする際にはもっぱら手作業がメインでした。私は。自慢できることではなく、お恥ずかしい限りです。今年は突如、AWS CDKに目覚めました。この記事を書いた時点では、GitHub Copilotをなだめすかしてがんばって書いてもらいました。いまは、Kiro CL((旧Amazon Q Developer CLI)をよく利用しています。開眼しました。
この一大変化は、教育機関や企業、社会人にAWSを教えさせていただくという経験が続いたこととは無関係ではないと思います。一番学ばせていただいたのは私だったのだと思います。その機会を得られたことに感謝しかありません。
隠すほどの内容はなにもないので、ソースコードをお披露目しておきます。2の内容（「AWS LambdaとAmazon EventBrigeで定期実行」）を包含しています。
gmail-to-slack-stack.ts
import * as cdk from 'aws-cdk-lib';
import { Construct } from 'constructs';
import { PythonFunction } from '@aws-cdk/aws-lambda-python-alpha';
// import * as sqs from 'aws-cdk-lib/aws-sqs';
import * as lambda from 'aws-cdk-lib/aws-lambda';
import * as logs from 'aws-cdk-lib/aws-logs';
import * as events from 'aws-cdk-lib/aws-events';
import * as targets from 'aws-cdk-lib/aws-events-targets';
import * as iam from 'aws-cdk-lib/aws-iam';
import * as path from 'path';
import * as sns from 'aws-cdk-lib/aws-sns';
import * as subscriptions from 'aws-cdk-lib/aws-sns-subscriptions';
import * as cloudwatch from 'aws-cdk-lib/aws-cloudwatch';
import * as cloudwatch_actions from 'aws-cdk-lib/aws-cloudwatch-actions';
export class GmailToSlackStack extends cdk.Stack {
constructor(scope: Construct, id: string, props?: cdk.StackProps) {
super(scope, id, props);
// The code that defines your stack goes here
// example resource
// const queue = new sqs.Queue(this, 'GmailToSlackQueue', {
// visibilityTimeout: cdk.Duration.seconds(300)
// });
// Lambda関数の定義
const gmailToSlackLambda = new PythonFunction(this, 'GmailToSlackLambda', {
runtime: lambda.Runtime.PYTHON_3_13,
index: 'main.py',
handler: 'lambda_handler',
entry: path.join(__dirname, '../lambda'), // main.pyがあるディレクトリを指定
logRetention: logs.RetentionDays.ONE_WEEK, // CloudWatch Logsの保存期間を7日間に設定
timeout: cdk.Duration.seconds(60),
memorySize: 160,
environment: {
GMAIL_TO_SLACK_FROM: process.env.GMAIL_TO_SLACK_FROM || 'ひみつ',
GMAIL_TO_SLACK_WEB_HOOK_URL: process.env.GMAIL_TO_SLACK_WEB_HOOK_URL || '',
GMAIL_TO_SLACK_CHANNEL: process.env.GMAIL_TO_SLACK_CHANNEL || '#ひみつ',
ENVIRONMENT: process.env.ENVIRONMENT || 'aws',
},
});
// SSMへのアクセス権限を付与
const ssmRunCmdPolicy = new iam.PolicyStatement({
actions: ['ssm:GetParameter', 'ssm:PutParameter'],
effect: iam.Effect.ALLOW,
resources: [
'arn:aws:ssm:*:*:parameter/<your last at param name>',
'arn:aws:ssm:*:*:parameter/<your credentials param name>',
'arn:aws:ssm:*:*:parameter/<your token param name>',
],
});
const lambdaRole = gmailToSlackLambda.role as iam.Role;
lambdaRole.addToPolicy(ssmRunCmdPolicy);
// 月曜から金曜の8:00-19:00(JST)の間は1分ごとに実行
const weekdayRule = new events.Rule(this, 'WeekdayScheduleRule', {
schedule: events.Schedule.cron({
minute: '0-59', // 毎分
hour: '23-9', // UTCで23:00〜09:59 → JSTで8:00〜18:59
weekDay: 'MON-FRI', // JSTの月〜金 = UTCの月〜金
}),
});
weekdayRule.addTarget(new targets.LambdaFunction(gmailToSlackLambda));
// それ以外の時間帯では5分ごとに実行
const offHoursRule = new events.Rule(this, 'OffHoursScheduleRule', {
schedule: events.Schedule.cron({
minute: '0/5', // 5分ごと
hour: '0-22,10-23', // JSTの19:00〜7:59 に対応するUTC時間
weekDay: '*', // 毎日
}),
});
offHoursRule.addTarget(new targets.LambdaFunction(gmailToSlackLambda));
// SNSトピックの作成
const errorNotificationTopic = new sns.Topic(this, 'ErrorNotificationTopic', {
displayName: 'Lambda Error Notifications',
});
// SNSトピックにメールサブスクリプションを追加
errorNotificationTopic.addSubscription(
new subscriptions.EmailSubscription('ひみつ')
);
// CloudWatchアラームの作成
const lambdaErrorAlarm = new cloudwatch.Alarm(this, 'LambdaErrorAlarm', {
metric: gmailToSlackLambda.metricErrors(), // Lambda関数のエラーメトリクス
threshold: 1, // エラーが1回以上発生した場合にアラームを発動
evaluationPeriods: 1, // 1つの評価期間でアラームを発動
alarmDescription: 'Alarm for Lambda function errors',
});
lambdaErrorAlarm.addAlarmAction(new cloudwatch_actions.SnsAction(errorNotificationTopic));
}
}
さいごに
電話代行サービスの会社に「Slack コネクト」でSlackに参加してもらえばいいのではないか?
そ、そッ、そっ、そうかもしれません 
それができるのなら一番効率的です。きっと、地球レベルでみたときには電力消費をもっとも少なくする解決策だと思いますので、いずれは限りある資源を有効活用するという地球レベルの観点ではそうすべきだと思います。
ただ、電話代行をする会社のサービス内容にかかわることなので、私があれこれ言えないこともありますし、担当者の業務負荷を軽減したことも事実です。そして、モダンにAWS CDKで構築する機会を得られたことは感謝しかありません。
限りある資源を有効活用するという地球レベルの観点からの解決策を模索しつつ、自分にできることを「今すぐやる」という、この姿勢で私は生きて参りたいと思います。これが、プレゼント🎁 カレンダー 「「レガシー」を保守したり、刷新したりするにあたり得られた知見・ノウハウ・苦労話 by Works Human Intelligence Advent Calendar 2025」への私なりのアンサーです。
このような形で今年もその進化をご報告できたことをうれしく思います。その機会をご提供くださった、Works Human Intelligenceサンタ 様と @Qiita 様には感謝しかありません。
そして、各所で私の拙いAWSの授業を受けていただいたみなさまに感謝です。よりよい授業をお届けできるように精進を重ねて参りたいと思います。