GCP Cloud Run と Cloud Scheduler で実現する自動 Slack 投稿システムの構築

# はじめに
本記事では、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 への投稿を一元管理しています。

```python
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 時処理を確認できます。
  ```bash
  python auto_post.py 17
  ```
  ※タイムゾーンはコード内で ZoneInfo("Asia/Tokyo") により設定済みです。

- **Cloud Run / Cloud Scheduler との連携**  
  main.py の /post_message エンドポイントから subprocess 経由で auto_post.py を実行するように設定します。  
  例：
  ```python
  subprocess.run([sys.executable, "auto_post.py"], check=True)
  ```
  Cloud Scheduler から定期的にこのエンドポイントへ POST リクエストを送ることで、予定時刻に応じた自動投稿が実現します。

---

## 関連記事
1. [Ry03(2025.3), Slackボットによる知識データの定期投稿と質問応答の自動化](https://qiita.com/ry033rdqiita/items/e5da1732979375c87db2)