OpenAIを使ってDevelopersIOの新着記事を3行で要約してSlackに投稿してみた

DevelopersIOの新着記事をOpenAIを使って要約して、Slackに投稿してみました。DevelopersIOの記事を手軽にチェックできればトレンディになれるかも。

#Python

#OpenAI

#ChatGPT

#Slack

鈴木純

2023.03.19 00:50

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

はじめに

最近ChatGPTが熱いですね。DevelopersIOにもたくさんの記事が流れてきて、わくわくしながら過ごしています。

そんな中以下ブログを読んでいた時にふと思いついたことがありました。

「DevelopersIOを全部読むのは難しいけど、同じように新着記事が3行で要約されていればトレンディになれるのでは…？」

ということでOpenAI APIの練習がてらスクリプトを作ってみました。

最終的なイメージ

最低限以下の動作をすることを目指しました。

RSSフィードで新着記事を取得
全てまとめてだと記事数が多すぎるので、1時間以内のものにフィルター
OpenAIのAPIで３行要約を作成
指定したSlackチャンネルへ投稿

Slackへ投稿されるとこんな感じになります。

事前準備

OpenAI

以下のブログが完結にまとまっているので、まずはここからOpenAIの準備をしてみてください。ここで環境変数にシークレットキーが設定されている前提で進めます。

Slack(Webhook URL)

WebhookはSlack Appを使って作成します。以下ブログを参考に用意してください。

必要なライブラリのインストール

スクリプトはPythonで実装していて、openaiの他に2つ今回のスクリプトではライブラリを使用します。以下コマンドを実行するかpoetry等の環境でインストールしてください。

❯ pip install feedparser
❯ pip install requests

やってみる

スクリプトをローカルから動作させてSlackに通知させてみます。

動作環境

❯ python --version            
Python 3.8.13

❯ pip list
Package            Version
------------------ ---------
feedparser         6.0.10
openai             0.27.2
requests           2.28.2

依存関係があるパッケージは省略しています。

Pythonスクリプト

Pythonで以下のように実装しました。通知先のWebhook URLは事前準備で作成したものに置き換えてください。

import openai
import feedparser
import json
import requests
from typing import List, Dict
from datetime import datetime, timedelta, timezone
import os

# タイムゾーンを日本時間に設定
JST = timezone(timedelta(hours=+9))

# APIキーの設定
openai.api_key = os.environ["OPENAI_API_KEY"]

# Slack Webhook URL
WEBHOOK_URL = "{通知先のWebhook URL}"

# OpenAI API モデル名
OPENAI_MODEL: str = "gpt-3.5-turbo"

# RSSフィードURL(DevelopersIO)
FEED_URL = "https://dev.classmethod.jp/feed/"

SYSTEM_PARAMETER = """```
与えられたフィードの情報を、以下の制約条件をもとに要約を出力してください。

制約条件:
・文章は簡潔にわかりやすく。
・箇条書きで3行で出力。
・要約した文章は日本語へ翻訳。
・最終的な結論を含めること。

期待する出力フォーマット:
1.
2.
3.
```"""

def get_feed_entries() -> List[Dict]:
    """
    RSSフィードの取得し、過去1時間以内の更新のみを取得する
    """
    updated_since = datetime.now(JST) - timedelta(hours=1)
    # RSSフィードの取得
    feed = feedparser.parse(FEED_URL)
    # 更新日時の閾値よりも新しいエントリーのみを取得
    new_entries: List[Dict] = [
        entry for entry in feed.entries
        if datetime(*entry.updated_parsed[:6], tzinfo=timezone.utc)
        .astimezone(JST) > updated_since
    ]

    return new_entries

def generate_summary(feed) -> str:
    """
    RSSフィードを元に、OpenAIで要約を実行
    """
    text = f"{feed.title}>\n{feed.summary}"
    response = openai.ChatCompletion.create(
        model=OPENAI_MODEL,
        messages=[
            {"role": "system", "content": SYSTEM_PARAMETER},
            {"role": "user", "content": text},
        ],
        temperature=0.25,
    )
    summary: str = response.choices[0]["message"]["content"].strip()
    return summary

def post_to_slack(message: str, link_url: str, title: str) -> None:
    """
    RSSフィードとOpenAIの要約SlackのWebhookURLへPOST
    """
    data = {
        "blocks": [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"<{link_url}|{title}>",
                },
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": message,
                },
            },
            {"type": "divider"},
        ],
        "unfurl_links": False,
    }
    requests.post(WEBHOOK_URL, data=json.dumps(data))

def handler() -> None:

    # RSSフィードから記事を取得し、要約を生成してSlackに投稿する
    for entry in get_feed_entries():
        summary: str = generate_summary(entry)

        post_to_slack(summary, entry.link, entry.title)

if __name__ == "__main__":
    handler()