Solvio株式会社Publicationへの投稿
🍣

TechCrunch記事を要約・翻訳してDiscord経由で自動投稿するBotを構築する

okawa_
に公開
AI
Python
Bot
ChatGPT
tech

はじめに

こんにちは:)
今回は、TechCrunchの最新記事を自動取得し、Dicsordに翻訳・要約付きでチャンネルに告知するDicordBotを構築します。
いきなりですが、みなさまは最新の毎朝ニュースサイトをチェックするの正直めんどくさくないですか？
今回つくったのは、海外テックメディア「TechCrunch」の最新記事をDiscordに自動で投稿し、さらに「この記事ってどういうこと？」みたいな質問にも答えてくれる、ちょっと賢いニュースBotです。
そんな仕組みを、今回も一から構築したので、このあと実装ステップ・サンプルコードも交えて紹介していきます！

背景と課題

海外のテックニュースを読む際に以下の課題に直面することはないですか？

  • 『英語で読まないといけないし、情報量も多くて気軽に追いづらい』
  • 『AIに「日本語で要約して」と作業がめんどくさい』
  • 『いちいちチームにDiscord(slack)で共有するのがめんどくさい』

私自身もエンジニアとして常に最新のテックニュースを確認するのですが、忙しい朝や仕事終わりにニュースを漁る気力がないです...ましてや、英語の記事だと余計厳しいですよね。

そこで今回、「特定のトピックに合った記事を自動で収集・翻訳・要約し、Discordへ投稿する」というBotを作成しました。
これにより、上記の課題を自動的に解決できます！

  • 海外の最新テックニュースを素早くキャッチアップ
  • 日本語で要約された情報をチームに即共有
  • 投稿される記事の重複や古い情報の投稿を阻止

アプリケーション概要

今回作成したBotの機能を全てさらっと並べました！

主な機能

  • TechCrunchからRSS経由で最新記事を取得
  • DynamoDB による投稿済み記事の重複チェック
  • ChatGPTによる日本語翻訳＋簡単要約
  • Discord Bot による自動投稿
  • Botに記事を要求すると自動的に欲しい記事を返却
  • Discord上で記事に関する質問をすると、内容をもとに回答
  • 一般的な質問（『富士山って何？』）の様な質問に対しても適切な回答

フロー図と構成

今回はユーザーからBotへの問い合わせ処理とLambdaをトリガーとする記事定期実行の２つの構成・フロー図のご紹介いたします！

定期実行の構成とフロー図

構成要素
  • AWS EventBridge・・・毎朝指定時間に Lambda を自動実行するスケジューラー
  • AWS Lambda・・・FastAPI の /trigger エンドポイントを HTTP 経由で叩く実行ユニット
  • FastAPI（AppRunner）・・・/trigger エンドポイントでリクエストを受け取り、記事取得処理を開始
  • DynamoDB・・・記事の重複チェック・URL記録先
  • ChatGPT・・・新着記事の翻訳・要約処理
  • Uvicorn・・・FastAPIを動かすPython製の非同期Webサーバー

フローの詳細

  1. EventBridgeが投稿させたい時間にLambdaをトリガー
    • cron(0 0 * * ? *) のようなルールで、毎日定時に Lambda を呼び出します。
  2. Lambdaが/triggerエンドポイントを叩く
    • POSTを送信し、AppRunner上のFastAPIに対して処理を依頼します。
  3. FastAPIでRSSフィードを取得
    • feedparserを用いてTechCrunchのRSS・BeutifulSoupで取得
  4. 24時間以内の記事かを判定
    • 24時間以内で未投稿のもののみ抽出され、古い記事はスキップ。
  5. DynamoDBで重複チェック
    • rss_url+rss_idをキーに、既に投稿済みかどうかを確認します。
    • 同じ記事が二度と投稿されないよう、記事IDを記録。
  6. 新しい記事があれば翻訳＋要約しDiscordに投稿
    • ChatGPTを使って日本語に翻訳＋簡単要約し、Botが投稿します。

フロー図（ユーザー入力回答）

ユーザーの入力に対しての回答処理の構成図とフロー図

構成要素

  • Discord・・・Bot にメンション（トリガー）
  • Discord Bot・・・メッセージを受信し、FastAPIに処理を振り分ける
  • Python ・・・ユーザーの意図分類、記事検索、Web回答処理などを実行
  • DynamoDB・・・投稿済み記事URLや直近記事を記録・検索する
  • Chat GPT・・・入力メッセージに対する翻訳・簡単要約・自然文応答を生成
  • TechCrunch RSS Feed・・・記事取得元。トピックに一致する記事をフィルタリング

フローの詳細

  1. ユーザーが Bot にメンション付きでメッセージを投稿
    • 例：「@NewsBot AI に関する記事教えて」
    • 例：「@NewsBot 富士山ってなんですか？」
  2. Intent（意図）を判定
    • detect_intent 関数で OpenAI に意図判別（fetch_article / article_question / web_question）
  3. Intentに応じた処理に分岐
    • fetch_article:トピックに関連する新しい記事をRSSから検索→見つかれば翻訳・要約してDiscordに投稿
    • article_question:リプライ元 or 直近記事の内容を取得→質問に応答
    • web_question:記事がない場合は一般的なWeb質問とみなして回答を生成（Web検索）
  4. 記事を取得した場合は DynamoDB に記録
    • rss_url + entry.linkをkeyに重複防止
  5. 生成した回答（記事 or 回答文）をDiscordに送信

フロー図（ユーザー入力回答）

実際の構築

1. Discord Botの作成と設定

  • Discord Developer Portal で Bot を作成
  • メッセージの読み取り／送信権限（MESSAGE_CONTENT など）を付与
  • .env ファイルにDISCORD_BOT_TOKENやDISCORD_CHANNEL_IDを設定

Discord Developer Portalの設定

DISCORD_BOT_TOKEN や DISCORD_CHANNEL_IDなどの環境変数設定（.env）

# Discord Botのトークン
DISCORD_BOT_TOKEN=xxxxxxxxxxxxxxx

# 通知を送るチャンネルのID
DISCORD_CHANNEL_ID=xxxxxxxxxxxxxx

2. FastAPI アプリの作成

定期投稿実行Bot

  • FastAPI と Uvicorn による非同期Webサーバーの構築
  • /trigger エンドポイントにより、Lambdaなど外部から起動可能にする
  • DyanamoDBに格納されていない記事（URL）を投稿

定期投稿実行Botのサンプルコード

# DynamoDB テーブルのリソース初期化
table = boto3.resource(
    "dynamodb",
    region_name="ap-northeast-1",
    aws_access_key_id="AWS_ACCESS_KEY_ID",
    aws_secret_access_key="AWS_SECRET_ACCESS_KEY",
).Table("posted_entries")

# FastAPI のエンドポイント `/trigger` に対応
t@app.post("/trigger")
async def trigger():
    # Discord のチャンネル取得（キャッシュ or API）
    channel_id = int(DISCORD_CHANNEL_ID)
    bot_channel = bot.get_channel(channel_id) or await bot.fetch_channel(channel_id)

    # RSS フィードの URL と、投稿対象の時間範囲（24時間以内）
    rss_url = "https://techcrunch.com/feed/"
    threshold = datetime.utcnow() - timedelta(hours=24)

    # DynamoDB テーブル接続
    table = boto3.resource(
        "dynamodb",
        region_name="ap-northeast-1",
        aws_access_key_id=AWS_ACCESS_KEY_ID,
        aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
    ).Table("posted_entries")

    results = []  # 投稿済み記事のログ格納

    # RSS フィードをパースして記事をループ処理
    for entry in feedparser.parse(rss_url).entries:
        # 記事に公開日が無い or 古すぎる場合はスキップ
        if (
            not hasattr(entry, "published_parsed")
            or datetime.fromtimestamp(time.mktime(entry.published_parsed)) < threshold
        ):
            continue

        entry_id = str(entry.id)

        # DynamoDB に存在する記事ならスキップ（重複チェック）
        if table.get_item(Key={"rss_url": rss_url, "rss_id": entry_id}).get("Item"):
            continue

        # 記事本文の抽出
        content = await fetch_article_text(entry.link)

        # 記事本文の翻訳と要約
        translated = await translate_and_summarize(content)

        # Discord に送信する投稿メッセージ作成
        message_content = f"**新しい記事**\n{entry.title}\n{entry.link}\n\n{translated}"

        # 非同期スレッドで Discord に投稿
        asyncio.run_coroutine_threadsafe(bot_channel.send(message_content), bot.loop)

        # DynamoDB に投稿済みとして記録
        table.put_item(Item={"rss_url": rss_url, "rss_id": entry_id})

        # ログとして記事タイトルを保持
        results.append({"title": entry.title})

    # 投稿がなければ status メッセージを返す
    return results or {"status": "no_new_entries"}


# 記事本文を取得する関数（html から <p> タグを抽出）
async def fetch_article_text(url: str) -> str:
    async with httpx.AsyncClient(timeout=10) as client:
        res = await client.get(url)
    soup = BeautifulSoup(res.text, "html.parser")
    return " ".join(p.get_text() for p in soup.find_all("p"))


# 英文記事を翻訳し、短く要約する関数
async def translate_and_summarize(text: str) -> str:
    raw_output = await chat_completion(
        [
            {
                "role": "system",
                "content": "以下の英文記事を日本語に翻訳して要点をまとめた要約だけを返してください\n読みやすい文量と適切な改行を入れ、コンパクトな内容にまとめてください。非常に長い文章は避け、3行程度にまとめてください。\n日本人が読んでも理解できる文章にしてください。",
            },
            {"role": "user", "content": text},
        ],
        json_mode=False,
    )
    return f"\ud83d\udcdd **要約**\n\n{raw_output.strip()}"

ユーザー入力対応Bot

ユーザーが Discord 上でボットにメンションすると、Bot はそのメッセージの意図を判断し、以下の処理に分岐するようにします。
また、ユーザーが特定の記事を要求した場合に備えて、入力内容のワードをtopicとしてAIが処理し、元の記事でマッチさせて処理させます。

  • fetch_article：指定トピックに関連する記事を探して要約付きで提示
  • article_question：投稿済み記事に対しての追加質問に回答
  • web_question：記事と関係のない一般的な知識の質問に回答

意図判定のサンプルコード

async def detect_intent(text: str) -> dict:
    try:
        res = await chat_completion([
            {
                "role": "system",
                "content": (
                    "ユーザーの入力から意図を特定し、JSON形式で返してください。\n"
                    "intent: fetch_article（記事取得）, article_question（記事に関する質問）, web_question（一般知識）\n"
                    "該当する場合は topic を英語で含めてください"
                ),
            },
            {"role": "user", "content": text},
        ], json_mode=True)
        return json.loads(res)
    except:
        return {"intent": "unknown"}

fetch_articleのサンプルコード

if intent == "fetch_article":
    translated_topic = await translate_topic_to_english(topic)
    feed = feedparser.parse("https://techcrunch.com/feed/")
    for entry in feed.entries:
        if translated_topic.lower() in entry.title.lower():
            text = await fetch_article_text(entry.link)
            translated = await translate_and_summarize(text)
            await message.channel.send(f"**{entry.title}**\n{entry.link}\n\n{translated}")
            break

article_questionのサンプルコード

if intent == "article_question":
    # 記事URL取得（リプライ or 記憶）
    article_text = await fetch_article_text(url)
    user_question = message.content.strip()
    messages = [
        {"role": "system", "content": "以下の記事についての質問に日本語で答えてください。"},
        {"role": "user", "content": f"{article_text}\n\n質問: {user_question}"},
    ]
    reply = await chat_completion(messages)
    await message.channel.send(reply.strip())

web_questionのサンプルコード

if intent == "web_question":
    messages = [
        {"role": "system", "content": "質問に対して簡潔に日本語で答えてください。"},
        {"role": "user", "content": topic},
    ]
    reply = await chat_completion(messages)
    await message.channel.send(reply.strip())

3. RSS記事の取得と重複チェック

  • feedparser によるRSS取得
  • DynamoDBで過去の投稿を記録し、同じ記事をスキップ
    • rss_url(String): RSS フィードのURL→パーテーションキー
    • rss_id(String): 記事のID →ソートキー

重複チェックロジック

entry_id = entry.link.strip().lower()
if table.get_item(Key={"rss_url": rss_url, "rss_id": entry_id}).get("Item"):
    continue

このコードにより、すでに投稿済みかどうかを判定。もし、過去に投稿された記事と同じ rss_id（＝URL）を持つデータが DynamoDB に存在していれば、その記事はスキップされます。

この仕組みによって、定期的に/triggerが実行されても、Botが同じ記事を何度も投稿することなく、チャンネルが「同じネタで埋まる」ような事も回避できます。

また、記事の投稿が完了すると同時に、その記事の情報（rss_url と rss_id）は DynamoDB に書き込まれ、次回以降の重複チェックに活用される仕組みにしました。

4. ChatGPTによる翻訳＆要約

  • openai.AsyncOpenAI を使って英語記事を翻訳・要約
  • ChatGPTに「3行以内で要約して」と指示を出すことで短くまとめる
from openai import AsyncOpenAI

# OpenAIクライアントを初期化
openai_client = AsyncOpenAI(api_key=xxxxxxxxxxxxxxx))

#記事の本文を日本語に翻訳し、要約する関数
from openai import AsyncOpenAI

openai_client = AsyncOpenAI(api_key="xxxxxxxxxxxxxxx")

# 記事の本文を日本語に翻訳し、要約する関数
async def translate_and_summarize(text: str) -> str:
    # ChatGPT へ渡すメッセージ
    messages = [
        {
            "role": "system",
            "content": (
                "以下の英文記事を日本語に翻訳し、自然な日本語で3行以内に要約してください。"
            ),
        },
        {
            "role": "user",
            "content": text,
        },
    ]

    # ChatGPT への API リクエスト（非同期・Responses API対応）
    resp = await openai_client.chat.completions.create(
        model="gpt-4.1",
        messages=messages,
        temperature=0,
        response_format={"type": "text"},  # ResponsesAPI
    )

    return f"📝 **翻訳・要約結果**\n\n{resp.choices[0].message.content.strip()}"

5. AWSインフラとの統合

App RunnerへのFastAPIデプロイ

1.FastAPIアプリケーションをDockerイメージとして構築し、Amazon ECR（Elastic Container Registry）にプッシュ！

2.AWS AppRunnerを使用してECRに保存したDockerイメージをデプロイする。

3.App Runnerにより自動的にスケーリングされ、WebアプリケーションとしてFastAPIが公開させる

Lambda + EventBridgeによる定期実行

  • AWS EventBridgeで毎日特定の時間にLambda関数がトリガーされるようにスケジュール設定を行う。

  • Lambda関数ではPythonスクリプトを用いて、HTTPリクエストを発行し、App Runner上で稼働するFastAPIアプリの/triggerエンドポイントを呼び出す。

  • 処理結果は、DynamoDBに格納され、記事の重複チェックと新着記事の管理が行われる。

lambda_function.py

# lambda_function.py
import requests


def lambda_handler(event, context):
    url = "xxxxxxxxxxxxxxxxxxxxx"

    try:
        res = requests.post(url, timeout=10)
        res.raise_for_status()
        return {"statusCode": 200, "body": "App Runner triggered successfully"}
    except requests.RequestException as e:
        return {"statusCode": 500, "body": f"Failed to trigger App Runner: {str(e)}"}

AWS環境変数の設定

AWS_ACCESS_KEY_ID=xxxxxxxxxxxxxxx
AWS_SECRET_ACCESS_KEY=xxxxxxxxxxxxxxx
AWS_SESSION_TOKEN==xxxxxxxxxxxxxx

結果

記事取得に関する質問の処理（fetch_article）

例：『googleに関する記事をください』

  • input
    • 特定の記事をリクエストするユーザー入力
  • process
    • RSSを元にメディアから記事を抽出
    • ユーザーからの質問に対して、関連する記事が複数存在する場合でも、 RSS フィード（通常は最新の 20件前後の新着記事）でユーザーが入力したキーワードが類似している満たした1件のみを返答対象にする。
    • 記事内容を元に翻訳・要約を実施
  • output
    • ユーザーの入力したワードに関連する記事

例：

記事に関する質問の処理（article_question)

  • ユーザーがリプライで記事に質問した場合、元の投稿からURLを抽出
  • 対象記事を再取得して内容に基づき回答

例：『この記事に関して詳細に教えて』

  • input
    • 記事の詳細に関する質問
  • process
    • リプライされた記事のURLから記事の内容を取得
    • 記事情報の取得はBeautifulSoupで取得
  • output
    • ユーザーの質問に対して記事の内容を反映した回答

例：

一般質問の処理（web_question）

例：『富士山って何？』

  • input
    • ユーザーが知りたいことなどの質問
  • process
    • Webサーチ
  • output
    • ユーザーがの質問に対してWebサーチの結果を回答

例：

どうでしょう！！
質問の位置に対して、AIが意図判定を行い、適切な回答が返却されているのが確認できました！

まとめ

今回の記事では、英語のニュース記事を自動で日本語に翻訳・要約して、Discordに投稿するアプリケーションの作り方を紹介しました。

このBotのミソとしてユーザーが記事に対して質問すると、AIが内容を読み取って適切なレスポンスを返せれる柔軟性も大きなポイントです。
これによって、最新ニュースを定期的に投稿するだけでなく、ユーザーの質問に沿って適切な回答が実現することができました。

AWSを活用したので技術的な要素も多いですが、アイデア次第でさまざまな場面に応用できる構成になっているので、興味がある方はぜひ参考にしてみてください！

Solvio株式会社
Solvio株式会社Publication

生成AIを軸に企業の課題解決を行う会社です。

Discussion