Go to Qiita Advent Calendar 2024 Top
LoginSignup
2
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

@katsuki_ono

AWSにSpotifyを搭載したChatGPT APIを用いたSlack botを移してみた。

Last updated at Posted at 2023-04-16

Spotifyを搭載したChatGPT APIを用いたSlack botをAWSのインスタンスに乗せてみました。

Google colabで動いたコードをそのまま、AWSのインスタンスに載せて動かすことは簡単なのですが、UIが悪いので、以下の機能をつけています。

・SlackからSpotifyを認証できるようにする
・Spotifyのアクセストークンをリフレッシュすることで、一度Spotifyの認証をしたらずっと使えるようにする(Spotifyのアクセストークンは1時間で期限が切れます。)
・Spotifyからログアウトできるようにする

今回の記事では、上記の機能のコードの書き方をメインに記載します。

AWSのインスタンスの立ち上げ方やChatGPT APIを用いたSlack botを使う為にインスタンスのコンソールで何をしたら良いのかは、こちらの記事を参考にして下さい。

今回、ChatGPT APIを用いたSlack botから追加でインストールが必要なのは、Spotifyのみです。

この記事では、以下を記載します。
⒈Spotifyを搭載したChatGPT APIを用いたSlack botをAWSに実装する方法(そのまま使えるコード)
⒉Spotifyを搭載したChatGPT APIを用いたSlack botのコード説明

この記事のインスタンスのインスタンスイメージはubuntuです。

⒈Spotifyを搭載したChatGPT APIを用いたSlack botをAWSに実装する方法(そのまま使えるコード)

まず、AWSのインスタンスで、ChatGPT APIを用いたSlack botを使えるようにして下さい。

手順は、こちらの記事を参考にして下さい。
※こちらの記事では、iPadからインスタンスにSSH接続していますが、インスタンスにSSH接続出来れば、何でも良いです。
※「⒊インスタンスにコードを置き実行中にする」の「次にpythonのファイルを保存するディレクトリを作成します。」まで完了して下さい。

上記まで完了したら、Spotifyをインストールします。

sudo pip install spotipy

作成したディレクトリに移動します。

cd ディレクトリ名

作成したディレクトリでファイルを作成します。

nano ファイル名.py

作成したファイルの中に下記のbotのコードを入れて、ctrl oで保存してエンターを押してから、ctrl xでファイルを閉じて下さい。
※Slack、Spotify、openaiのキーやトークン、リダイレクトURLをあらかじめ入れておいてください。詳しくはこちらの記事をご覧ください。

※このSlack botにはSlack APIの設定も必要になります。詳しくはこちらの記事をご覧ください。

# インポート
import openai
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
import spotipy
from spotipy.oauth2 import SpotifyOAuth
import requests
import base64
import sched
import time
import concurrent.futures
import threading
import os

# Slackのトークン
os.environ["SLACK_BOT_TOKEN"] =  ""
os.environ["SLACK_SIGNING_SECRET"] = ""
os.environ["SLACK_APP_TOKEN"] =  ""

# openaiキー
openai_api_key = ""

# Spotify APIのクライアントIDとシークレットとリダイレクトURL
client_id = ''
client_secret = ''
redirect_uri = ''


app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
)

openai.api_key = openai_api_key

# Spotifyに必要なスコープを指定
scope = 'user-read-recently-played playlist-read-private playlist-read-collaborative app-remote-control user-read-playback-state user-library-read user-modify-playback-state'

# Spotifyの認証情報を取得するためのオブジェクトを作成
sp_oauth = SpotifyOAuth(client_id=client_id, client_secret=client_secret, redirect_uri=redirect_uri, scope=scope)

# グローバル変数用
sp = None
refresh_token = None
assist1 = ""

@app.event("message")
def handle_message(event, say):
    global assist1
    global sp
    global sp_oauth
    global refresh_token

    text = event["text"]
    channel = event["channel"]

    if not ("AQ" in text or "logout" in text or "ログアウト" in text):
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "任意の設定"},
                {"role": "assistant", "content": assist1},
                {"role": "user", "content": text},
            ]
        )
        assist1 = response["choices"][0]["message"]["content"]  # assist1を更新
        say(assist1)

    if "おすすめして" in text:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "以下を音楽のジャンル、アーティスト名、ムードのどれか一つだけを一単語で表す。アーティスト名が含まれる場合には、アーティスト名だけ一単語で返す。"},
                {"role": "user", "content": text},
            ]
        )
        assist2 = response["choices"][0]["message"]["content"]  

        # ジャンルを指定してプレイリストを検索
        genre = assist2
        results = sp.search(q=genre, type='playlist')
        items = results['playlists']['items']

        # 最初のプレイリストを選択
        playlist_uri = items[0]['uri']
        playlist_name = items[0]['name']

        # プレイリストのトラックリストを取得
        playlist_tracks = sp.playlist_tracks(playlist_uri, fields='items(track(name,id,uri))')

        # 最初のトラックのURIを取得
        track_uri = playlist_tracks['items'][0]['track']['uri']

        # 現在のユーザーのデバイスを取得
        devices = sp.devices()
        device_id = devices['devices'][0]['id']

        # 取得したデバイスで再生を開始する
        sp.start_playback(device_id=device_id, context_uri=playlist_uri, offset={"uri": track_uri})

    if "聞きたい" in text or "聴きたい" in text or "ききたい"in text:
        if sp is None:
            # ユーザーに認証を求めるために認証ページのURLを表示
            auth_url = sp_oauth.get_authorize_url()
            auth_url = auth_url.replace("&", "&")

            message = {
                "text": f"ブラウザにこちらのURLの「文字列」をコピペして開き、開いたページのURLをSlackに送ってください。※URLにリンクされたURLは無効のため、URLの「文字列」をコピーしてください。: <https://informationaboutthis.wordpress.com/2023/04/08/slack_spotify/|{auth_url}>"
            }
            response = app.client.chat_postMessage(channel=channel, **message)

        else:
            keywords = ["聴きたい", "聞きたい", "ききたい"]
            filtered_text = text
            for keyword in keywords:
                filtered_text = filtered_text.replace(keyword, "")
            assist3 = filtered_text

            # ジャンルを指定してプレイリストを検索
            genre = assist3
            results = sp.search(q=genre, type='playlist')
            items = results['playlists']['items']

            # 最初のプレイリストを選択
            playlist_uri = items[0]['uri']
            playlist_name = items[0]['name']

            # プレイリストのトラックリストを取得
            playlist_tracks = sp.playlist_tracks(playlist_uri, fields='items(track(name,id,uri))')

            # 最初のトラックのURIを取得
            track_uri = playlist_tracks['items'][0]['track']['uri']

            # 現在のユーザーのデバイスを取得
            devices = sp.devices()
            device_id = devices['devices'][0]['id']

            # 取得したデバイスで再生を開始する
            sp.start_playback(device_id=device_id, context_uri=playlist_uri, offset={"uri": track_uri})

    if "AQ" in text:
        # 認証コードの取得
        code_idx = text.find("?code=")
        bird_idx = text.find("|https://")
        url = text[code_idx + 6 : bird_idx]

        code = sp_oauth.parse_response_code(url)
        token_info = sp_oauth.get_access_token(code)
        # アクセストークンを取得し、認証オブジェクトを作成
        token = token_info['access_token']
        refresh_token = token_info['refresh_token']
        sp = spotipy.Spotify(auth=token)

    if "logout" in text or "ログアウト"in text:
        refresh_token = None
        sp = None
        os.system("rm .cache")

def refresh_spotify_token(sc):
    global refresh_token
    global client_id
    global client_secret
    global sp

    if refresh_token is not None:
        # Base64エンコードされたクライアントIDとシークレットを生成
        client_credentials = f"{client_id}:{client_secret}"
        client_credentials_b64 = base64.b64encode(client_credentials.encode()).decode()

        # POSTリクエストのパラメータを設定
        token_url = 'https://accounts.spotify.com/api/token'
        token_data = {
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token
        }

        # POSTリクエストを送信して、アクセストークンを取得
        token_headers = {
            'Authorization': f'Basic {client_credentials_b64}'
        }
        response = requests.post(token_url, data=token_data, headers=token_headers)
        response_data = response.json()

        # 新しいアクセストークンを取得
        token = response_data['access_token']
        sp = spotipy.Spotify(auth=token)

        # 次回の実行を30分後にスケジュール
        s.enter(1800, 1, refresh_spotify_token, (s,))

    else:
        s.enter(0, 1, refresh_spotify_token, (s,))

# スケジュール
s = sched.scheduler(time.time, time.sleep)
# 最初の実行をスケジュール
s.enter(0, 1, refresh_spotify_token, (s,))

def function1():
    handler = SocketModeHandler(app, os.environ.get("SLACK_APP_TOKEN"))
    handler.start()

def function2():
    s.run()

if __name__ == "__main__":
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future1 = executor.submit(function1)
        future2 = executor.submit(function2)

        # スレッド1~2の両方が終了するまで待機する
        concurrent.futures.wait([future1, future2])

一度、上手くファイルが作成されているか下記のコマンドで確認します。

find ~

上手くファイルが作成されていたら、screenを作成します。

screen -S スクリーン名

screen内でpythonで作成したファイルを実行します。

python3 ファイル名.py

bolt run！的なやつが出力されたら上手くファイルを実行できています。

実行できていたら、「聴きたい」とbotに言ってみてください。するとbotがコメントと合わせてSpotifyの認証のURLをSlackに送ってきます。
認証のURLが届いたら、URLの文字列をコピーしてブラウザで開き、認証を行なってください。(SlackにURLをboxから送る際、URLが勝手にエンコードされて認証用のURLが無効になってしまう為、認証の説明ページを文字列にリンクさせています。)
認証が完了したら、リダイレクトされて開くページのURLをコピーしてSlackからbotにそのURLを送ってください。これでSpotifyの認証が完了します。
Spotifyの認証が完了したら、Spotifyで音楽を再生してから、 botに「〇〇聴きたい」と言ってみてください。
Slack botからSpotifyの認証URLが送られて来ず、Spotifyから再生される曲が〇〇に関するプレイリストの曲に変更されたら成功です！
botにログアウトと言ってSpotifyからログアウトも出来ます。

⒉Spotifyを搭載したChatGPT APIを用いたSlack botのコード説明

コードが長いので、コードの上から区切りながらコードの意味を説明します。

必要なインポート

上から、openai、Slack、Spotify、その他をインポートしています。
その他の用途は主に以下です。
requests、base64はアクセストークンのリフレッシュ、
sched、timeはスケジューラー、
concurrent.futures、threadingは、スケジューラーとSlack botを同時に実行中にするスレッド、
osは環境変数とキャッシュクリア

# インポート
import openai

from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

import spotipy
from spotipy.oauth2 import SpotifyOAuth

import requests
import base64
import sched
import time
import concurrent.futures
import threading
import os

トークンやキーの設定

必要なトークンやキー、リダイレクトURLを設定します。

# Slackのトークン
os.environ["SLACK_BOT_TOKEN"] =  ""
os.environ["SLACK_SIGNING_SECRET"] = ""
os.environ["SLACK_APP_TOKEN"] =  ""

# openaiキー
openai_api_key = ""

# Spotify APIのクライアントIDとシークレットとリダイレクトURL
client_id = ''
client_secret = ''
redirect_uri = ''


app = App(
    token=os.environ.get("SLACK_BOT_TOKEN"),
    signing_secret=os.environ.get("SLACK_SIGNING_SECRET")
)

openai.api_key = openai_api_key

Spotifyの設定

必要なスコープを設定して、Spotifyの認証オブジェクトを作成します。

# Spotifyに必要なスコープを指定
scope = 'user-read-recently-played playlist-read-private playlist-read-collaborative app-remote-control user-read-playback-state user-library-read user-modify-playback-state'

# Spotifyの認証情報を取得するためのオブジェクトを作成
sp_oauth = SpotifyOAuth(client_id=client_id, client_secret=client_secret, redirect_uri=redirect_uri, scope=scope)

グローバル変数を設定

主にSpotifyの認証に必要な変数をグローバル変数にして、関数内外で使い回せるようにします。
assist1だけ、ChatGPT用です。

# グローバル変数用
sp = None
refresh_token = None
assist1 = ""

Slack botの設定 - グローバル変数と変数

グローバル変数を関数内で使えるようにしています。またよく使うSlackのイベントに起因する事項を変数にしています。

@app.event("message")
def handle_message(event, say):
    global assist1
    global sp
    global sp_oauth
    global refresh_token

    text = event["text"]
    channel = event["channel"]

Slack botの設定 - ChatGPTとテキストのレスポンス

Spotifyの認証のトークンがAQで始まるため、AQが含まれる場合と、ログアウトが含まれる場合を除き、Slackからのメッセージを botで取得して、ChatGPTで処理してSlackに botから返すようにしています。

    if not ("AQ" in text or "logout" in text or "ログアウト" in text):
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "任意の設定"},
                {"role": "assistant", "content": assist1},
                {"role": "user", "content": text},
            ]
        )
        assist1 = response["choices"][0]["message"]["content"]  # assist1を更新
        say(assist1)

Slack botの設定 - ChatGPTがおすすめの曲をかける

おすすめしてがメッセージに含まれる場合、そのメッセージをChatGPTが処理して、Spotifyで処理したテキストで、プレイリストを検索をして、検索したプレイリストを再生します。
遊び要素です。

    if "おすすめして" in text:
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=[
                {"role": "system", "content": "以下を音楽のジャンル、アーティスト名、ムードのどれか一つだけを一単語で表す。アーティスト名が含まれる場合には、アーティスト名だけ一単語で返す。"},
                {"role": "user", "content": text},
            ]
        )
        assist2 = response["choices"][0]["message"]["content"]  

        # ジャンルを指定してプレイリストを検索
        genre = assist2
        results = sp.search(q=genre, type='playlist')
        items = results['playlists']['items']

        # 最初のプレイリストを選択
        playlist_uri = items[0]['uri']
        playlist_name = items[0]['name']

        # プレイリストのトラックリストを取得
        playlist_tracks = sp.playlist_tracks(playlist_uri, fields='items(track(name,id,uri))')

        # 最初のトラックのURIを取得
        track_uri = playlist_tracks['items'][0]['track']['uri']

        # 現在のユーザーのデバイスを取得
        devices = sp.devices()
        device_id = devices['devices'][0]['id']

        # 取得したデバイスで再生を開始する
        sp.start_playback(device_id=device_id, context_uri=playlist_uri, offset={"uri": track_uri})

Slack botの設定 - 認証用URLの送付と音楽の再生

聴きたいが含まれる場合に処理を行います。
spに認証情報が含まれない場合に、認証用のURLを発行して、 botからSlackに認証用のURLを送ります。Slack botからURLを送付する際にエンコードを防げないので、ユーザーのミスを防ぐためにURLの文字列に別のリンクを貼っています。
spに認証情報が含まれている場合、聴きたいを排除したテキストで、Spotifyでプレイリストを検索して、検索したプレイリストを再生します。

    if "聞きたい" in text or "聴きたい" in text or "ききたい"in text:
        if sp is None:
            # ユーザーに認証を求めるために認証ページのURLを表示
            auth_url = sp_oauth.get_authorize_url()
            auth_url = auth_url.replace("&amp;", "&")

            message = {
                "text": f"ブラウザにこちらのURLの「文字列」をコピペして開き、開いたページのURLをSlackに送ってください。※URLにリンクされたURLは無効のため、URLの「文字列」をコピーしてください。: <https://informationaboutthis.wordpress.com/2023/04/08/slack_spotify/|{auth_url}>"
            }
            response = app.client.chat_postMessage(channel=channel, **message)

        else:
            keywords = ["聴きたい", "聞きたい", "ききたい"]
            filtered_text = text
            for keyword in keywords:
                filtered_text = filtered_text.replace(keyword, "")
            assist3 = filtered_text

            # ジャンルを指定してプレイリストを検索
            genre = assist3
            results = sp.search(q=genre, type='playlist')
            items = results['playlists']['items']

            # 最初のプレイリストを選択
            playlist_uri = items[0]['uri']
            playlist_name = items[0]['name']

            # プレイリストのトラックリストを取得
            playlist_tracks = sp.playlist_tracks(playlist_uri, fields='items(track(name,id,uri))')

            # 最初のトラックのURIを取得
            track_uri = playlist_tracks['items'][0]['track']['uri']

            # 現在のユーザーのデバイスを取得
            devices = sp.devices()
            device_id = devices['devices'][0]['id']

            # 取得したデバイスで再生を開始する
            sp.start_playback(device_id=device_id, context_uri=playlist_uri, offset={"uri": track_uri})

Slack botの設定 - Spotifyの認証オブジェクトを作成

ユーザーが認証情報が含まれるURLを送付してきた場合に、処理を行います。認証コードはAQから始まるため、メッセージにAQが含まれる場合に処理します。
まずURLからコードを取得するための処理をして、コードを取得します。その後、認証コードからアクセストークンとリフレッシュトークンを取得して、認証オブジェクトのspを作成します。

    if "AQ" in text:
        # 認証コードの取得
        code_idx = text.find("?code=")
        bird_idx = text.find("|https://")
        url = text[code_idx + 6 : bird_idx]

        code = sp_oauth.parse_response_code(url)
        token_info = sp_oauth.get_access_token(code)
        # アクセストークンを取得し、認証オブジェクトを作成
        token = token_info['access_token']
        refresh_token = token_info['refresh_token']
        sp = spotipy.Spotify(auth=token)

Slack botの設定 - Spotifyからログアウト

認証情報をクリアしてログアウトします。キャッシュが悪さすることがあるので、キャッシュもクリアします。

    if "logout" in text or "ログアウト"in text:
        refresh_token = None
        sp = None
        os.system("rm .cache")

Spotifyのアクセストークンをリフレッシュ

Spotifyのアクセストークンは1時間で期限切れになるため、30分毎に、リフレッシュトークンでアクセストークンを更新しています。
スケジューラーで30分毎に実行するようにしています。

def refresh_spotify_token(sc):
    global refresh_token
    global client_id
    global client_secret
    global sp

    if refresh_token is not None:
        # Base64エンコードされたクライアントIDとシークレットを生成
        client_credentials = f"{client_id}:{client_secret}"
        client_credentials_b64 = base64.b64encode(client_credentials.encode()).decode()

        # POSTリクエストのパラメータを設定
        token_url = 'https://accounts.spotify.com/api/token'
        token_data = {
            'grant_type': 'refresh_token',
            'refresh_token': refresh_token
        }

        # POSTリクエストを送信して、アクセストークンを取得
        token_headers = {
            'Authorization': f'Basic {client_credentials_b64}'
        }
        response = requests.post(token_url, data=token_data, headers=token_headers)
        response_data = response.json()

        # 新しいアクセストークンを取得
        token = response_data['access_token']
        sp = spotipy.Spotify(auth=token)

        # 次回の実行を30分後にスケジュール
        s.enter(1800, 1, refresh_spotify_token, (s,))

    else:
        s.enter(0, 1, refresh_spotify_token, (s,))

スケジューラーを設定

スケジューラーを設定しています。

# スケジュール
s = sched.scheduler(time.time, time.sleep)
# 最初の実行をスケジュール
s.enter(0, 1, refresh_spotify_token, (s,))

Slack botとスケジューラーを同時に実行

Slack botとスケジューラーを常に実行中にするため、スレッドを作成して同時に実行中にしています。

def function1():
    handler = SocketModeHandler(app, os.environ.get("SLACK_APP_TOKEN"))
    handler.start()

def function2():
    s.run()

if __name__ == "__main__":
    with concurrent.futures.ThreadPoolExecutor() as executor:
        future1 = executor.submit(function1)
        future2 = executor.submit(function2)

        # スレッド1~2の両方が終了するまで待機する
        concurrent.futures.wait([future1, future2])

最後に

今回は常にSpotifyも使えるChatGPTを用いたSlack botを使えるようにしました。
色々機能付けてUIを高めるのが楽しかったです。
また何か作ってみます。

2
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?