はじめに
本記事では、Google Cloud Platform (GCP) 上で Cloud Run と Cloud Scheduler を使って、自動的に Slack へ投稿するシステムの実装方法を解説します。
具体的には、Gemini API を利用して情報を検索・要約し、さらに参照サイト情報を抽出する処理を組み込み、9時と17時で異なるクエリに基づいて Slack へ投稿する仕組みを実現します。
この記事では、コードの設計思想や各機能の概要、実装例、そして動作確認の方法について説明します。
システム構成・アーキテクチャ
このシステムは以下のコンポーネントで構成されています。
-
Cloud Run アプリケーション
main.py（Flask アプリ）をデプロイし、Cloud Scheduler などからのリクエストを受け取ります。
/post_message エンドポイントで、subprocess を用いて auto_post.py および simple_post.py を実行します。 -
auto_post.py
現在の時刻をチェックし、9時の場合は「今日の天気を詳しく教えて下さい」、17時の場合は「今日の日本のニュースを教えて下さい」というクエリを Gemini に送信して応答内容を Slack に投稿します。 -
simple_post.py
通常の定期投稿処理（例：DOCX から情報を抽出して投稿）を実装しており、こちらはそのまま動作させます。 -
GeminiSlackPoster クラス
auto_post.py の中で、Gemini API を使った検索・要約・参照サイト抽出、そして Slack への投稿処理を一元管理するためのクラスです。
実装の詳細
GeminiSlackPoster クラス（auto_post.py）
以下は、GCP Cloud Run 上で動作する auto_post.py の主要コードです。
このクラスは、リトライ付きの HTTP GET、Gemini API での要約処理、参照サイトの抽出、そして Slack への投稿を一元管理しています。
import os
import random
import requests
from bs4 import BeautifulSoup
from datetime import datetime
from google import genai
from zoneinfo import ZoneInfo
class GeminiSlackPoster:
def __init__(self, gemini_api, slack_bot_token, slack_channel_id):
self.gemini_api = gemini_api
self.slack_bot_token = slack_bot_token
self.slack_channel_id = slack_channel_id
# Gemini クライアントの初期化
self.client = genai.Client(api_key=self.gemini_api, http_options={'api_version': 'v1alpha'})
self.search_client = self.client.chats.create(
model='gemini-2.0-flash-exp',
config={'tools': [{'google_search': {}}]}
)
# ----- リトライ処理付き GET 関数 -----
def robust_get(self, url, max_retries=3, timeout=20):
for attempt in range(max_retries):
try:
r = requests.get(url, allow_redirects=True, timeout=timeout)
return r
except Exception as e:
if attempt < max_retries - 1:
continue
else:
return None
# ----- リダイレクト先のURL取得 -----
def get_final_url(self, redirect_url):
r = self.robust_get(redirect_url)
return r.url if r else None
# ----- Gemini 応答の要約 -----
def summary_client(self, original_text):
summary_prompt = f"""
以下のテキストを要約してください。
可能であれば、文中の引用番号 ([1], [2], [10]など) は維持してください。
--- Original Text ---
{original_text}
---------------------
"""
summary_response = self.client.models.generate_content(
model="gemini-2.0-flash-exp",
contents=summary_prompt
)
return summary_response.text.strip()
# ----- 参照サイト一覧生成 -----
def serch_references(self, response):
if not response.candidates[0].grounding_metadata:
return "検索結果情報 (grounding_metadata) がありませんでした。"
else:
grounding_chunks = response.candidates[0].grounding_metadata.grounding_chunks
if not grounding_chunks:
return "URLが取得できませんでした。ご興味のある方はご自身でも調べてみて下さい。"
else:
ref_lines = []
ref_lines.append("*取得した参照サイト一覧:*")
for i, chunk in enumerate(grounding_chunks, start=1):
redirect_url = chunk.web.uri
final_url = self.get_final_url(redirect_url)
if final_url is None:
page_title = "（リダイレクト失敗）"
else:
try:
resp = self.robust_get(final_url)
if resp is None:
page_title = "（取得失敗）"
elif resp.status_code == 200:
soup = BeautifulSoup(resp.text, "html.parser")
page_title = soup.title.string if soup.title else "（タイトルなし）"
else:
page_title = f"（取得できませんでした: {resp.status_code}）"
except Exception as e:
page_title = f"（エラー: {str(e)}）"
ref_lines.append(f"{i}. <{final_url}|{page_title}>")
return "\n".join(ref_lines)
# ----- search_info 関数 -----
def search_info(self, user_query):
response = self.search_client.send_message(user_query)
original_text = ""
for part in response.candidates[0].content.parts:
if part.text:
original_text += part.text + "\n"
summary_text = self.summary_client(original_text)
references_text = self.serch_references(response)
return summary_text, references_text, response
# ----- Slack 投稿関数 -----
def post_message_to_slack(self, message):
headers = {
"Authorization": f"Bearer {self.slack_bot_token}",
"Content-Type": "application/json"
}
today_date = datetime.now(ZoneInfo("Asia/Tokyo")).strftime("%Y-%m-%d")
# self.slack_channel_id がリストの場合も想定
channels = self.slack_channel_id
for channel in channels:
payload = {
"channel": channel,
"text": message
}
response = requests.post("https://slack.com/api/chat.postMessage", headers=headers, json=payload)
data = response.json()
if data.get("ok"):
print(f"✅ {today_date} に Slack へメッセージを投稿しました！(チャンネル: {channel})")
else:
print(f"❌ Slack への投稿に失敗しました: {data.get('error')} (チャンネル: {channel})")
print("📌 詳細なレスポンス:", data)
# ----- search_info を利用して Slack 投稿する関数 -----
def post_search_result(self, query):
summary, references, response = self.search_info(query)
slack_message = f"*要約結果:*\n{summary}\n\n{references}"
print(slack_message)
self.post_message_to_slack(slack_message)
# ----------------------------------------
# メイン処理（auto_post.py）
# ----------------------------------------
if __name__ == "__main__":
# 環境変数などから各種値を取得
from google.colab import userdata # この行は GCP 上では不要
gemini_api = GEMINI_API
slack_bot_token = SLACK_BOT_TOKEN
# SLACK_CHANNEL_ID がリストの場合、そのまま利用、単一ならリスト化する
slack_channel_id = SLACK_CHANNEL_ID
poster = GeminiSlackPoster(gemini_api, slack_bot_token, slack_channel_id)
# 現在の時刻を JST で取得
from zoneinfo import ZoneInfo
current_hour = datetime.now(ZoneInfo("Asia/Tokyo")).hour
if current_hour == 9:
poster.post_search_result("今日の天気を詳しく教えて下さい")
elif current_hour == 17:
poster.post_search_result("今日の日本のニュースを教えて下さい")
else:
print("現在は9時または17時ではないため、投稿処理は行いません。")
7. 動作確認・デプロイ
-
ローカル/Cloud Shell でのテスト
コマンドライン引数でシミュレーションすることもできます。たとえば、Cloud Shell で以下のように実行して JST の 17 時処理を確認できます。python auto_post.py 17※タイムゾーンはコード内で ZoneInfo("Asia/Tokyo") により設定済みです。
-
Cloud Run / Cloud Scheduler との連携
main.py の /post_message エンドポイントから subprocess 経由で auto_post.py を実行するように設定します。
例：subprocess.run([sys.executable, "auto_post.py"], check=True)Cloud Scheduler から定期的にこのエンドポイントへ POST リクエストを送ることで、予定時刻に応じた自動投稿が実現します。