仕事をしていると無限にカフェインを摂ってしまうので、Slackbotにカフェイン摂取量を計測してもらい、一定の摂取量を超えたら注意してもらうことにしました。
chat_postEphemeralで自分だけに表示
chat_postEphemeral は、Slack APIのメソッドの一つで、特定のユーザーに対してのみ一時的に表示されるメッセージを送信するために使われます。
エフェメラルメッセージは特定のチャンネルの会話の流れの中で、実行者だけに表示させるための機能です。
今回は便利なエフェメラルメッセージとスラッシュコマンドを利用して簡易的なSlackBotを作ってみました。
メソッドの使い方
chat_postEphemeralを呼び出す際に、最低限必要となる主要なパラメーターは以下の3つです。
| パラメーター | 説明 | 例 |
|---|---|---|
channel |
メッセージを表示するチャンネルID。 エフェメラルメッセージを「どこに」表示するかという場所のコンテキストを指定します。 |
C012345ABC |
user |
メッセージを受け取るユーザーID。 チャンネル内にいるメンバーのうち、「誰に」メッセージを見せるかという個人を特定します。 |
U012345ABC |
text |
ユーザーに対して一時的に表示されるメッセージの本文。MarkdownやSlackの書式設定(絵文字、太字など)を含めることができます。 | ログを記録しました。 |
ちなみに、DM内でもDから始まるchannelIDのようなものを渡せばエフェメラル表示することができます。
作ってみた
ディレクトリ構成はとてもシンプルです。
slack-bot/
├── .env # 環境変数ファイル
├── .venv/ # Pythonの仮想環境
├── app.py # メインアプリケーション
└── history.json # カフェイン摂取履歴ログ←これをDBの代わりにしてます
それではSlackBotの核となるapp.pyに以下の処理を書いていきます。
# .envファイルを読み込む
load_dotenv()
# アプリを初期化
app = App(token=os.environ.get("SLACK_BOT_TOKEN"))
# ログを保存するファイル名
LOG_FILE = "history.json"
# 警告ラインの設定 (mg)
CAFFEINE_LIMIT_WARNING = 300 # そろそろ注意
CAFFEINE_LIMIT_DANGER = 400 # これ以上は危険
健康な大人だとして、1日当たり400mg(コーヒーで5杯程度)までであれば、カフェインによる健康への危険な悪影響はないとしているらしいので、400mgを警告のラインとしました。
次にスラッシュコマンドで入力するドリンクの種類を決めます。
私はコーヒー(カフェラテ)かRedBullしか飲まないので2種類にしました。
DRINK_MENU = {
"/redbull": {
"caffeine": 80,
"message": "🥤 Red Bull摂取"
},
"/coffee": {
"caffeine": 60,
"message": "☕️ コーヒー摂取"
},
}
次に、実行されたコマンドの記録を、history.jsonの末尾に追加保存する処理を書きます。
# コマンドの実行履歴をJSONファイルに保存する
def save_log(user_id, command_name):
log_entry = {
"user_id": user_id,
"command": command_name,
"timestamp": time.time(),
"date_str": str(date.today())
}
# ログファイルの読み込み
if os.path.exists(LOG_FILE):
with open(LOG_FILE, "r", encoding="utf-8") as f:
try:
history = json.load(f)
except json.JSONDecodeError:
history = []
else:
history = []
# 新しいログを追加
history.append(log_entry)
# ログをJSONファイルに書き込み
with open(LOG_FILE, "w", encoding="utf-8") as f:
json.dump(history, f, indent=2, ensure_ascii=False)
そして1日分のデータを取得し、摂取量や回数を集計する関数の処理を書きます。
def get_report_data(user_id):
# ログファイルが存在しない場合はエラーメッセージを返す
if not os.path.exists(LOG_FILE):
return None, "データがまだありません。"
# ログファイルを読み込む
with open(LOG_FILE, "r", encoding="utf-8") as f:
try:
history = json.load(f)
except:
return None, "データ読み込みエラー"
today = date.today()
# 期間は今日に設定
start_date = today
title_text = str(today)
start_timestamp = datetime.combine(start_date, datetime.min.time()).timestamp()
# ユーザーと期間でログをフィルタリング
target_logs = [
log for log in history
if log["user_id"] == user_id and log["timestamp"] >= start_timestamp
]
# 該当期間のログがない場合はメッセージを返す
if not target_logs:
return None, f"{title_text} の期間にはまだ何も飲んでいません。"
counts = {}
total_caffeine = 0
# フィルタリングされたログから集計
for log in target_logs:
cmd = log["command"]
menu = DRINK_MENU.get(cmd)
if menu:
counts[cmd] = counts.get(cmd, 0) + 1
# 合計カフェイン量を算出
total_caffeine += menu.get("caffeine", 0)
# ログリストと集計結果を返す
return target_logs, (counts, total_caffeine, title_text)
1日分のカフェイン摂取合計レポートを作成し、警告文を追加します。
def get_today_report(user_id):
# 今日の集計データを取得
target_logs, result = get_report_data(user_id)
if target_logs is None:
return result
counts, total_caffeine, today_str = result
# 残りのカフェイン量の計算
current_time = time.time()
estimated_remaining_caffeine = 0
for log in target_logs:
# 経過時間基づき、残存量を計算し合算
caffeine_amount = DRINK_MENU.get(log["command"], {}).get("caffeine", 0)
elapsed_min = (current_time - log["timestamp"]) / 60
if elapsed_min > 0 and caffeine_amount > 0:
# 計算式: 残存率 = (0.5) ^ (経過時間 / 半減期)
remaining_rate = (0.5) ** (elapsed_min / CAFFEINE_HALF_LIFE_MIN)
estimated_remaining_caffeine += caffeine_amount * remaining_rate
# レポート本文の作成
report_lines = [f"*{today_str} のカフェインレポート*"]
report_lines.append("----------------")
# ドリンクごとの摂取回数と合計をリストに追加
for cmd, count in counts.items():
menu = DRINK_MENU.get(cmd)
if menu:
caffeine_per_drink = menu.get('caffeine', 0)
report_lines.append(f"• `{cmd}` : {count}回 ({caffeine_per_drink * count}mg)")
report_lines.append("----------------")
report_lines.append(f"合計摂取量: {total_caffeine} mg")
report_lines.append(f"推定残存量: {estimated_remaining_caffeine:.1f} mg")
# 警告判定と追記
if total_caffeine > CAFFEINE_LIMIT_DANGER:
report_lines.append(f"\n🚫 警告: 1日の上限({CAFFEINE_LIMIT_DANGER}mg)を超えています! これ以上飲むのは控えましょう。")
elif total_caffeine > CAFFEINE_LIMIT_WARNING:
report_lines.append(f"\n⚠️ 注意: 上限({CAFFEINE_LIMIT_DANGER}mg)に近づいています。")
return "\n".join(report_lines)
今回のメインである、ドリンク摂取を記録し、エフェメラルでフィードバック(摂取量と警告)を返す処理を書きます!!
@app.command("/report")
def record_drink(user_id, command_name, client, channel_id):
menu = DRINK_MENU.get(command_name)
if not menu: return
# ログ保存
save_log(user_id, command_name)
#今日のデータ集計
# get_report_dataを呼び出し、最新の合計値とログ件数を取得
target_logs, result = get_report_data(user_id)
daily_count = 0
daily_total = 0
if target_logs:
_, daily_total, _ = result
daily_count = len(target_logs) # 今日のログ件数
# メッセージ作成
current_caffeine = menu["caffeine"]
# 基本メッセージ
msg = f"{menu['message']} (+{current_caffeine}mg)"
# 2回目以降なら合計値を追記
if daily_count > 1:
msg += f"\n本日の合計: {daily_total} mg"
# ★ 警告ロジックの追加 ★
# 上限(400mg)を超えた、または上限(300mg)に近づいているかを判定し、警告メッセージを追記
if daily_total > CAFFEINE_LIMIT_DANGER:
msg += f"\n\n🚨 警告: {CAFFEINE_LIMIT_DANGER}mgを超えました!\nこれ以上の摂取は危険です。控えてください。"
elif daily_total > CAFFEINE_LIMIT_WARNING:
msg += f"\n\n⚠️ 注意: もうすぐ上限です (現在 {daily_total}/{CAFFEINE_LIMIT_DANGER} mg)"
# 通知
try:
# chat_postEphemeralで実行者本人にだけメッセージを表示
client.chat_postEphemeral(
channel=channel_id,
user=user_id,
text=msg
)
except SlackApiError as e:
print(f"Error: {e}")
/reportコマンドハンドラの処理を書きます。
@app.command("/report")
def handle_report(ack, body, client):
# /report コマンドの処理。常に今日の詳細レポートを表示
ack()
user_id = body["user_id"]
channel_id = body["channel_id"]
text = body.get("text", "").strip().lower()
should_clear_history = False
history_cleared_message = ""
# 履歴初期化の設定
if text == "clear" or text == "reset":
should_clear_history = True
# レポート作成
report_text = get_today_report(user_id)
# 履歴削除処理
if should_clear_history:
if os.path.exists(LOG_FILE):
try:
os.remove(LOG_FILE)
history_cleared_message = "\n\n 履歴を初期化しました。"
except Exception as e:
history_cleared_message = f"\n\n 初期化失敗: {e}"
# エフェメラルで実行者本人に結果を送信
client.chat_postEphemeral(
channel=channel_id,
user=user_id,
text=report_text + history_cleared_message
)
ドリンクコマンドを一括で処理し、記録関数(record_drink)へ渡します。
@app.command("/redbull")
@app.command("/coffee")
def handle_drinks(ack, body, client):
ack()
record_drink(body["user_id"], body["command"], client, body["channel_id"])
最後に、ターミナルでpythonファイルを実行した時にbotを起動させるようにします。
if __name__ == "__main__":
SocketModeHandler(app, os.environ.get("SLACK_APP_TOKEN")).start()
SlackApp設定
Appの設定項目は以下です。簡潔に書きます。
・Socket Modeを有効化しxapp-で始まるApp-Level Tokenトークンを.envに貼る
・OAuth & PermissionsのBot Token Scopesにcommandsとchat:writeを追加し、
xoxb-から始まるBot User OAuth Tokenを.envに貼る
・Slash Commandsの設定をする
→Create New Commandで/redbull,/coffee,/reportを追加
Slackで計測してみる
pythonファイルを実行し、チャンネルで試してみましょう。
/coffeeとチャット欄に入力し送信すると...
エフェメラルメッセージでカフェイン摂取表示されました!
何回かコマンドを打ってみると...
しっかり警告を出して注意してくれました。
最後に/reportコマンドを入力してレポートを出してみます。
カフェイン摂取量レポートを出してくれました。
まとめ
今回はエフェメラルメッセージとスラッシュコマンドを利用して簡易Slakbotを作ってみました。皆さんもぜひ作ってみてください。


