ソムリエ協会イベント投稿bot
ソムリエ協会は、例会セミナーという協会員が無料で参加できるイベントを行っています。しかし、例会セミナーの情報はソムリエ協会サイトでしか公開されておらず、また東京開催のイベントとなるとすぐに満席になってしまうので、定期的にソムリエ協会のサイトを閲覧していない方だと中々参加することができません。
そんなわけで今回は、ソムリエ協会のサイトを巡回してイベント情報をいち早くキャッチし、twitterで発信するbotを作成しようと思います。
ソースコード一式はGithubで掲載していますので、こちらも御覧ください。
環境
- Windows 10
- Python 3.8
- AWS Lambda
今回は自分の好きなpythonでコードを組んでいきます。関数の実行はAWSのLambdaというサービスを使用します。
データ収集
まず、ソムリエ協会のホームページからイベントを集めてくるプログラムを作成します。
ソムリエ協会は残念ながらAPIが作成されていませんので、愚直にhttpリクエストを送って、返ってきたhtmlを解析することでイベント情報を集めます。ソース解析にはBeautiful Soupを使用します。
まず、ソムリエ協会のサーバにリクエストを送ってイベントページのデータを取得します。1ページずつ取得するしかなさそうなので、ページが存在しないエラーが出るページまで1ページずつデータを取得します。
import datetime
import requests
import urllib.parse
from bs4 import BeautifulSoup
today = datetime.date.today()
scheme = "https"
netloc = "www.sommelier.jp"
path = "/event"
def get_event_page(page):
params = fragment = None
query_dict = {
"viewType" : "l",
"calenderYear" : today.year,
"calenderMonth" : today.month,
"page" : page,
}
query = urllib.parse.urlencode(query_dict)
url = urllib.parse.urlunparse((scheme, netloc, path, params, query, fragment))
response = requests.get(url)
if response.ok:
soup = BeautifulSoup(response.text, "html.parser")
return soup
else:
return False次にデータを整形します。各イベントの開催日、会場、名前、URLを抽出します。
ソムリエ協会のイベント一覧ページは、そのイベントが開催される月日は表示してくれるのですが、そのイベントが開催される年を表示してくれないので今日の日付から算出しています。
import os
def parse_event_page(soup):
event_list = []
event_li_list = soup.find(id="e_list_area").find("ul", class_="event_list").findAll("li")
for event_li in event_li_list:
event_month, event_day = map(int, event_li.find(class_="eve_data").text.split("/"))
event_date = datetime.date(today.year, int(event_month), int(event_day))
# 年を超す場合
if (event_date - today).days < 0:
event_date = datetime.date(today.year + 1, int(event_month), int(event_day))
event_venue = event_li.find(class_="eve_name").text
event_name = event_li.find(class_="eve_txt").find("a").text.replace("\u3000", "")
event_path = event_li.find(class_="eve_txt").find("a").get("href")
event_id = os.path.basename(event_path)
event_dict = {
"id" : event_id,
"date" : event_date,
"venue" : event_venue,
"name" : event_name,
"path" : event_path,
}
event_list.append(event_dict)
return event_listこれでソムリエ協会のイベント情報を取得することができました。
データ格納
次は取得してきたイベントをデータベースに格納します。
このプログラムのためだけにSQLサーバを立てるのはばかばかしかったため、今回はSpread Sheetにデータを格納します。イベントID(URLの最後)、日付、会場、イベント名、URLの順で格納します。新しいイベントが追加された際にSpread Sheetにイベントを追記します。
Spread Sheetをpythonで操作するために、予めGoogle Cloud Platformでcredential情報(credentials.json)を取得し、実行ファイルと同じディレクトリに置きます。また、作成したSpread Sheetのキー(SPREADSHEET_KEY)を環境変数に格納しておきます。
import gspread
from oauth2client.service_account import ServiceAccountCredentials
rootdir = os.path.abspath(os.path.dirname(__file__))
credential_json_path = os.path.join(rootdir, "credentials.json")
def write_spreadsheet(event_list):
# spreadsheetにログイン
scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"]
credentials = ServiceAccountCredentials.from_json_keyfile_name(credential_json_path, scope)
gc = gspread.authorize(credentials)
SPREADSHEET_KEY = os.environ.get("SPREADSHEET_KEY")
worksheet = gc.open_by_key(SPREADSHEET_KEY).sheet1
# 1行目のイベントIDを取得
registered_id_list = worksheet.col_values(1)
# url作成用
params = query = fragment = None
# 新しいイベントがあった際に格納する
new_event_list = []
for event in event_list:
event_id = event["id"]
if event_id not in registered_id_list:
event_date = event["date"].strftime("%Y/%m/%d")
event_venue = event["venue"]
event_name = event["name"]
event_path = os.path.join(path, event["path"])
event_url = urllib.parse.urlunparse((scheme, netloc, event_path, params, query, fragment))
# spreadsheetに行追加
event_row = [event_id, event_date, event_venue, event_name, event_url]
worksheet.append_row(event_row)
# 新しいイベントがあったのでflgをtrueに
new_event_list.append(event)
return new_event_listこれでソムリエ協会のイベント情報のデータベース化が完了しました。
Twitterで投稿
新しいイベントが追加された際に、そのイベントの日付、会場、イベント名をTwitterでアナウンスします。
今回は、Twitterをpythonから操作するためのライブラリであるtweepyを使用します。予めTwitter DeveloperでCONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_SECRETを取得しておきます。
Twitterで投稿する際、ツイートを140字に収めないといけないので、字数カウントのために専用のライブラリ(twitter-text-python)を使用しました。また、曜日取得はdatetimeで行いたかったのですが、localeを変更すると文字化けしてしまったため、泣く泣くこのような実装になっています。今回Twitterに投稿するのは、関東近辺のイベントが追加されたときのみにします。
import tweepy
from twitter_text import parse_tweet
week = ["月", "火", "水", "木", "金", "土", "日"]
target_venue = ["東京", "神奈川", "千葉", "埼玉", "本部"]
def make_tweet(new_event_list):
tweet_content = "[自動]ソムリエ協会のイベントが更新されました！"
tweet_list = []
for event in new_event_list:
event_venue = event["venue"]
if event_venue in target_venue:
event_name = event["name"]
event_date = event["date"]
# locale設定すると文字化けしてしまったので泣く泣く
event_date_str = "%s月%s日(%s)" % (event_date.month, event_date.day, week[event_date.weekday()])
tweet_sentence = "%s %s %s" % (event_venue, event_date_str, event_name)
tweet_content_new = "%s\n%s" % (tweet_content, tweet_sentence)
if parse_tweet(tweet_content_new).valid:
tweet_content = tweet_content_new
else:
tweet_list.append(tweet_content)
tweet_content = tweet_sentence
tweet_list.append(tweet_content)
return tweet_list
def tweet(tweet_list):
auth = tweepy.OAuthHandler(os.environ["CONSUMER_KEY"], os.environ["CONSUMER_SECRET"])
auth.set_access_token(os.environ["ACCESS_TOKEN"], os.environ["ACCESS_SECRET"])
api = tweepy.API(auth)
reply_id = None
for tweet in tweet_list:
response = api.update_status(tweet, in_reply_to_status_id=reply_id)
reply_id = response.idこれで一連の流れが完了です。
AWS Lambdaで定期実行
ソムリエ協会のサイトを1日1回閲覧して、新しいイベントが更新されていないかを確認します。今回はAWS Lambdaを使用します。
まず、policyを作成します。
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}
$ aws iam --profile [CHANGE YOUR PROFILE] create-role --role-name lambda-exec --assume-role-policy-document file://policy.json次に、Lambdaに関数を登録します。
上で紹介した関数を一つのファイル(sommelier-event.py)にまとめます。詳細はGithubを参照してください。
また、必要なライブラリをインストールします。Lambdaにプッシュするため、今回はプロジェクトルート直下にpackagesというディレクトリを作り、そこに必要なライブラリをインストールします。
$ pip install --target ./packages tweepy gspread beautifulsoup4 oauth2client python-dotenv twitter-text-parserLambdaにコードをプッシュします。インストールしたライブラリと、メインプログラム(sommelier-event.py)と、Spread Sheetの認証に使用するcredentials.jsonをzipにまとめます。また、プログラムに使用する環境変数も同時に格納します。
$ cd packages
$ zip -r9 ../function.zip .
$ cd ../
$ zip -g function.zip sommelier-event.py
$ zip -g function.zip credentials.json
$ aws lambda --profile [CHANGE YOUR PROFILE] create-function --function-name sommelier-event --zip-file fileb://function.zip --handler sommelier-event.main --runtime python3.8 --role arn:aws:iam::[CHANGE YOUR ACCOUNT ID]:role/lambda-exec
$ aws lambda --profile [CHANGE YOUR PROFILE] update-function-configuration --function-name sommelier-event --timeout 120 --environment "Variables={SPREADSHEET_KEY=[CHANGE YOUR SPREADSHEET_KEY],CONSUMER_KEY=[CHANGE YOUR CONSUMER_KEY],CONSUMER_SECRET=[CHANGE YOUR CONSUMER_SECRET],ACCESS_TOKEN=[CHANGE YOUR ACCESS_TOKEN],ACCESS_SECRET=[CHANGE YOUR ACCESS_SECRET]}"最後に、1日に1回実行するにようにAWS Event Bridgeを使用してスケジューリングします。今回はManagement Consoleから手動で設定しました。全部まとめてterraformで設定すればよかったですね。。
これで、AWS Lambdaで1日に1回ソムリエ協会のイベント情報を収集し、新しいイベントがあったら自動でツイートするbotが完成しました。
まとめ
以上で、ソムリエ協会のサイトを自動で巡回し、イベントの更新があった際にいち早く情報をキャッチすることができるようになりました。新着イベントは自分のTwitterで発信していますのでぜひフォローよろしくお願いします。
協会員であればせっかくなので色んなイベントに参加したいですよね。みなさんのイベント情報収集労力の削減になれば幸いです。
もし改善の要望がありましたらお気軽にお問い合わせください。
