ラジオで今流れている曲を Slack に投稿する

現在の職場でラジオを流しているのですが、たま～に気に入った曲があって、調べたりします。
ですが、いちいちラジオ局のサイトを開いて流れている曲を検索するのが実に面倒です。
なので、定期的にラジオ局のサイトを見に行って、流れている曲を見つけて投稿してくれるアプリを作りましょう💪

例として、ここでは TOKYO FM で流れている曲を毎分調べて Slack に投稿するというシチュエーションを設定します。

今回使用する AWS のサービスは以下の通りです。

• Lambda
• Secrets Manager
• IAM
• CloudWatch Logs
• EventBridge (CloudWatch Events)

前提

• AWS アカウントを持っている
• Python をインストールしている
• Slack のアカウントとチャンネルを作成している

オンエア曲のサイトを見てみる

TOKYO FM のオンエア曲一覧はこちらで確認することができます。

開発者ツールで確認する

今流れている曲は一覧の一番上にある曲なので、開発者ツールのコンソールで以下のコマンドを実行して確認してみましょう。

document.querySelector('#searchResult .entry');

querySelectorAll ではなく querySelector を使用することで最初に見つかった要素、すなわち一番上にある曲が取得できます。

また、アルバムジャケット・曲名・アーティスト名は以下のように取得できます。

// アルバムジャケット
document.querySelector('#searchResult .entry .entryThum img').src;

// 曲名
document.querySelector('#searchResult .entry .entryTxt a').innerText;

// アーティスト名
document.querySelector('#searchResult .entry .entryArtist').innerText;

手順

Slack のアプリを作成する

https://api.slack.com/apps にアクセスし、「Create App」をクリックします。

「From scratch」をクリックします。

アプリ名を入力し、ワークスペースを選択します。

アプリの作成が完了すると、アプリの概要が表示されるので、「Edit Manifest」をクリックします。

Manifest の編集画面が表示されたら、以下のように入力して「Save Changes」をクリックします。

_metadata:
  major_version: 1
  minor_version: 1
display_information:
  name: Now Playing
settings:
  org_deploy_enabled: false
  socket_mode_enabled: false
  token_rotation_enabled: false
features:
  bot_user:
    display_name: Now Playing Bot
oauth_config:
  scopes:
    bot:
      - chat:write
      - chat:write.customize

左のメニューから「OAuth & Permissions」をクリックし、OAuth Tokens for Your Workspace の中の「Install to workspace」をクリックします。

権限リクエストの画面が表示されるので「Allow」をクリックすると、先ほどの画面にもどり今度は Bot User OAuth Token が表示されているはずです。
その値をテキストエディタに貼り付けておいてください。

作成したアプリを Slack チャンネルに追加する

アプリを追加したい Slack チャンネルを選び、右クリックして「Open channel details」をクリックします。

「Integrations」を選択し、Apps のところの「Add an app」をクリックします。

作成したアプリを選択し、「Add」をクリックします。

トークンを Secrets Manager に格納する

AWS コンソールから Secrets Manager を開き、「Store a new secret」をクリックします。

「Other type of secret」を選択し、画面下の方でトークンを入力します。
今回はキーを token, 値を先ほどメモしておいたトークンにします。
入力したら「Next」をクリックします。

次の画面で Secret name を入力します。ここでは slack-bot-user-oauth-token とします。
入力したら「Next」をクリックします。

残りの画面はそのままで進み、登録します。

すると、一覧に先ほど登録したものが出てくるようになります。

これでトークンの登録は完了です。

Lambda 実行用の IAM Role を作成する

AWS コンソールから IAM を開き、左のメニューから「Roles」を選択、その後「Create role」をクリックします。

Choose a use case のところで「Lambda」を選択し、「Next: Permissions」をクリックします。

次の画面ではアタッチするロールを選択する画面になります。フィルターで検索するなどして以下のロールを選択します。

• AWSLambdaBasicExecutionRole
• IAMFullAccess
• SecretsManagerReadWrite

選択したら、「Next: Tags」「Next: Review」の順にクリックします。

Role name, Role description を入力し、Policies が先ほど選択した 3 つになっていることを確認して、「Create role」をクリックします。
今回は Role name を slack-post-now-playing-role にしました。

Lambda function を作成する

方針としては以下のようになります：

• オンエア曲のサイトにアクセスして曲情報を取得する
• 前回取得した情報と異なるとき、Slack に投稿する
• 投稿した曲情報を格納する

AWS コンソールから Lambda を開き、「Create function」をクリックします。

「Author from scratch」を選択し、Basic information の中身は以下のようにします

• Function name: slack-post-now-playing
• Runtime: Python 3.9

Basic information の下にある「Change default」をクリックし、Execution role は「Use an existing role」を選択します。

Existing role には前の項で作成した slack-post-now-playing-role を選択します。

選択したら「Create function」をクリックします。

ローカルで Lambda function を作成してアップロードする

今回は Python の外部ライブラリを使用するため、コードは AWS コンソール上ではなく自分の PC 上で作成して zip にしたものをアップロードします。

まず、作業用フォルダを作成し、その中に移動します。

mkdir -p now-playing
cd now-playing/

lambda_function.py というファイルを作成し、中身は以下のようにします。

lambda_function.py

import base64
import hashlib
import json
import re

import boto3
import requests
from botocore.exceptions import ClientError
from bs4 import BeautifulSoup


def get_song_info() -> dict:
    """オンエア曲のサイトからスクレイピング

    Notes:
        - BeautifulSoup を使用して URL から取得したソースのパースを行っています。

    Returns:
        dict: 曲情報

    """
    url = 'https://www.keitai.fm/search/view/au/'
    r = requests.get(url)
    soup = BeautifulSoup(r.content, 'html.parser')

    if soup:
        jacket_url = soup.select('#searchResult .entry .entryThum img')[0]['src']
        if not re.match('https?://', jacket_url):
            jacket_url = url + jacket_url
        title_obj = soup.select('#searchResult .entry .entryTxt a') or soup.select('#searchResult .entry .entryTxt')
        title = re.sub(r'\r\n\s+', '', title_obj[0].string)
        artist = soup.select('#searchResult .entry .entryArtist')[0].string
        return {
            'jacket_url': jacket_url,
            'title': title,
            'artist': artist,
        }
    else:
        return {}


def get_hash(title: str, artist: str) -> str:
    """曲名・アーティスト名からハッシュを生成する。この値をもとに現在流れている曲が変わったことを検知する。

    Args:
        title:
        artist:

    Returns:
        str: hash.

    """
    return hashlib.sha512(f'{title}{artist}'.encode('utf-8')).hexdigest()


def song_is_updated(hash512: str) -> bool:
    """曲が変わった？

    Notes:
        - 最終的にスクリプトは毎分実行になるので、曲情報が更新されていない＝ハッシュが以前のものと同じであることを検知します。

    Args:
        hash512:

    Returns:
        bool: True if 変わった, False otherwise.

    """
    client = boto3.client('iam')
    user_tags = client.list_user_tags(UserName='slack-post-now-playing-tags')
    for tag in user_tags['Tags']:
        if tag['Key'] == 'hash' and tag['Value'] == hash512:
            return False
    return True


def update_tag(hash512: str) -> None:
    """曲情報のハッシュを更新する。

    Notes:
        - IAM ユーザのタグを使用してお金をケチります。

    Args:
        hash512:

    Returns:

    """
    client = boto3.client('iam')
    client.tag_user(
        UserName='slack-post-now-playing-tags',
        Tags=[
            {
                'Key': 'hash',
                'Value': hash512,
            }
        ],
    )


def get_secret() -> dict:
    """トークンを取得する。

    例外が発生した場合は例外の内容を取得する。

    Notes:
        - トークンの内容は秘密なので Secrets Manager で管理します。
        - せっかく曲情報の格納で IAM ユーザのタグを使ってお金をケチったのに、ここでお金がかかります。

    Returns:

    """
    secret_name = "arn:aws:secretsmanager:us-east-1:567877680446:secret:slack-bot-user-oauth-token-cDkd36"
    region_name = "us-east-1"

    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )

    try:
        get_secret_value_response = client.get_secret_value(
            SecretId=secret_name
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'DecryptionFailureException':
            return {'exception': 'DecryptionFailureException'}
        elif e.response['Error']['Code'] == 'InternalServiceErrorException':
            return {'exception': 'InternalServiceErrorException'}
        elif e.response['Error']['Code'] == 'InvalidParameterException':
            return {'exception': 'InvalidParameterException'}
        elif e.response['Error']['Code'] == 'InvalidRequestException':
            return {'exception': 'InvalidRequestException'}
        elif e.response['Error']['Code'] == 'ResourceNotFoundException':
            return {'exception': 'ResourceNotFoundException'}
    else:
        if 'SecretString' in get_secret_value_response:
            secret = get_secret_value_response['SecretString']
        else:
            secret = base64.b64decode(get_secret_value_response['SecretBinary'])
        return json.loads(secret)


def post(channel: str, song_info: dict) -> dict:
    """Slack に投稿する。

    Args:
        channel:
        song_info:

    Returns:

    See Also:
        - https://api.slack.com/methods/chat.postMessage

    """
    secret = get_secret()
    if 'exception' in secret:
        return {'exception': secret['exception']}
    elif 'token' not in secret:
        return {}

    data = {
        'channel': channel,
        'blocks': [
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*{song_info['title']}*\n{song_info['artist']}"
                },
                "accessory": {
                    "type": "image",
                    "image_url": song_info['jacket_url'],
                    "alt_text": "cd_image"
                }
            }
        ],
        'icon_emoji': ':radio:',
        'text': f"{song_info['artist']} - {song_info['title']}",
        'username': 'Now Playing Bot',
    }
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f"Bearer {secret['token']}",
    }

    response = requests.post(
        'https://slack.com/api/chat.postMessage',
        data=json.dumps(data),
        headers=headers,
    )
    return response.json()


def lambda_handler(event, context):
    song_info = get_song_info()
    hash512 = get_hash(song_info['title'], song_info['artist'])
    if song_is_updated(hash512):
        update_tag(hash512)
        res = post(event['channel'], song_info)
    else:
        res = {}

    return {
        "statusCode": 200,
        "body": json.dumps(res),
    }

pip を使ってライブラリのインストールを行います。

pip install --target ./package requests beautifulsoup4 boto3

ライブラリを zip に圧縮します。

cd package/
zip -r ../my-deployment-package.zip .

lambda_function.py を zip に追加します。

cd ..
zip -g my-deployment-package.zip lambda_function.py

Lambda function のタイムアウト時間を設定する

関数の画面で「Configuration」を選択します。

左のメニューから「General configuration」、「Edit」と選択し、タイムアウトを「0 min 15 sec」に設定します。

Lambda function に zip をアップロードする

AWS コンソールから Lambda を選択し、作成していた slack-post-now-playing をクリックします。

Code source のあたりで右側の「Upload from」をクリックし、「.zip file」を選択します。

「Upload」をクリックして先ほどの my-deployment-package.zip を選択したら、「Save」をクリックします。

ウィンドウが閉じたら「Test」を選択します。
Test event が表示されたら Template に hello-world を選択し、Name は slack-post-now-playing とします。
イベントを入力し、「Save changes」をクリックします。

今回は sandbox というチャンネルに投稿します。

{
  "channel": "sandbox"
}

---

テスト

Test event の右側にある「Test」を押してテストしてみましょう。

すると・・・ﾌﾟｼｭ、ｺｺｺｯ🎵

通知が来ました！チャンネルを見てみましょう！

うおおおめでとうございます！
Now Playing Bot さんがラジオのアイコンで今流れている曲を投稿してくれました！

試しに何回か続けて「Test」を押してみましょう。

すると、違う曲が投稿される or 何も起こらないはずです。
今流れている曲が前回と変わらない場合は投稿されないようになっていれば成功です！

実際の運用へ向けて

無事投稿のテストが成功したところで、
最後に、流れている曲を毎分調べて更新があったら自動的に投稿してくれる仕組みを作ります。

設定

AWS コンソールから Amazon EventBridge を開きます。

※ CloudWatch Events は Amazon EventBridge に生まれ変わったようです。

左のメニューから「Rules」を選び、「Create rule」をクリックします。

Name and description のところで Name を slack-post-now-playing-rule とします。
Define pattern のところで「Schedule」を選択し、Fixed rate every を 1 Minutes に設定します。

Select targets のところで Target を Lambda function に指定し、Function に作成したものを指定します。
Configure input のところは「Constant (JSON text)」を選択し、{"channel":"sandbox"} (sandbox のところはチャンネル名) と入力して、下の「Create」をクリックします。

しばらくして、以下のように曲が自動で投稿されるようになったら成功です！

---

まとめ

今回は AWS のサービスを使って、ラジオで今流れている曲を Slack に自動投稿するアプリを作ってきました。

なお、今回と同じ仕組みでサービスを 1 か月間運用した場合、料金は以下のようになります。

合計: 0.616 USD (2021/11/24 現在、1 ドル 115 円換算で 70.84 円)

• Secrets Manager
  • シークレットを 1 つ保存: 0.40 USD
  • API コール: 43,200 回 x 0.05 USD/10,000 コール = 0.216 USD
    • API コール数: 60 分 x 24 時間 x 30 日 = 43,200 回

その他のサービスは基本的に無料枠の範囲内で利用できます。

• Lambda: 月 100 万リクエストまで無料なので問題なし
• CloudWatch Logs: 5 GB まで無料だが、ログの有効期限を 1 day など短めにすれば無料枠を超過することはない

また、Secrets Manager を使わなければ 0 円で運用することも一応可能です。

みなさんも機会がありましたら是非やってみてください。
私はこれを作ったばかりのとき、投稿されるたびに Slack の画面を確認していたので仕事が進みませんでしたので気を付けてください😅

ありがとうございました。