はじめに
当方はしがないバショウカジキなので、情報収集にSNSはあまり活用できていません。
代わりにslackにRSS feedを流して眺めています。
外国語のサイトのfeedも購読しているのですが、外国語を読むのは母国語を読むのとは頭の使い方が違うので、記事が大量に配信されるようなサイトだとつい読むのをサボりがちです。
読むかどうかの取捨選択がもう少し簡単になるよう、RSSで配信される内容を翻訳してからslackに投稿する処理を作ることにしました。
(実はZapierに同じことをやって翻訳feedを生成できるテンプレートがあるみたいなんですが(使ったことはない)、せっかくやる気になったので自分で作ります。)
どういうものができますか?
Before
slackにあるアプリを利用したRSS feed投稿はこういったものですが、
After
今回作るものでは、それをこういう形になるよう処理します。
設計
構成
とりあえずローカル開発環境での実行前提で構成を作成します。
RSS情報を取得してメッセージアプリのwebhookに投稿する処理部分については、先駆者様がいらっしゃったので、参考にさせていただき考えました。
[python,AWS]RSSをサーバーレスに自動取得してDiscordに送ってみる
考慮事項
- 処理は
Python3.9で実装する - 翻訳処理は
googletransで行う- API制限があるのでrequestが増えてきたら代替を考える必要はありそう
- 後々の展開を考えて、処理で利用するパラメータは
DynamoDBに格納する形とする- slackのwebhook URLとRSS feedのURLを格納
処理
処理の流れの概要は、
- DynamoDBで事前定義しておいたパラメータを取得
- パラメータのRSS feed URLで取得処理(feedparserを利用)
- 取得した内容のうち、titleとsummaryを翻訳にかける(googletransを利用)
- 取得した内容を整形して、パラメータのslackのwebhook URLに投稿
雑ですが図はこんな感じです
事前準備
0-1. slackでwebhook URLを発行しておく
詳細は割愛します。発行時に投稿先にしたいslackのチャンネルを指定します。
0-2. DynamoDBテーブルを作成し情報を格納しておく
詳細は割愛します。私の場合、データの型については現時点ではこだわりがないので、すべて文字列で格納しました。
処理の作成
1. DynamoDBで事前定義しておいたパラメータを取得する
詳細は割愛します。データ取得時のAWS認証情報の設定などは好みのやり方でよいでしょう。
2. パラメータのRSS feed URLで取得(feedparserを利用)
feedparserでRSS情報を取得し、必要な要素を切り出してListで返します。
関数の実行時に引数としてRSS feedのURLを渡すことでRSS情報を取得します。また、後々の処理でエラーになる要素(RSS情報にlink,summary要素がない場合)はこの時点で除外しておきます。
import feedparser
import time
from dataclasses import dataclass
from typing import List
@dataclass
class DataRssBody():
title: str
url: str
summary: str
published: str
def getRssBodyFromUrl(endpoint: str) -> List[DataRssBody]:
feed = feedparser.parse(endpoint)
rss_list: List[DataRssBody] = []
for entry in feed.entries:
if not entry.get("link"):
continue
elif not entry.get("summary"):
continue
rss_content = DataRssBody(
title=entry.title,
url=entry.link,
summary=entry.summary,
published=entry.published_parsed
)
rss_list.append(rss_content)
print('Article Count: ' + str(len(rss_list)))
return rss_list
3. 取得した内容のうち、titleとsummaryを翻訳にかける(googletransを利用)
取得したRSS情報のうち、title要素とsummary要素を翻訳APIに送り、返ってきた結果を格納し直してListで返します。
from dataclasses import dataclass
from googletrans import Translator
from typing import List
@dataclass
class DataRssBody():
title: str
url: str
summary: str
published: str
def translateRssContentsByGoogle(rss_list) -> List[DataRssBody]:
tr = Translator()
translated_contents: List[DataRssBody] = []
for entry in rss_list:
title_trans = tr.translate(entry.title, dest="ja").text
summary_trans = tr.translate(entry.summary, dest="ja").text
rss_content_trans = DataRssBody(
title=title_trans,
url=entry.url,
summary=summary_trans,
published=entry.published
)
translated_contents.append(rss_content_trans)
return translated_contents
4. 取得した内容を整形して、パラメータのslackのwebhook URLに投稿
翻訳済みデータを格納したListを使用して、slackのwebhook URLに情報を送信します。関数実行時の引数として、翻訳済み内容のListとslack webhookのURLを渡してあげます。
import json
import requests
def sendWebhookBody(contents,webhook_url):
if len(contents) > 0:
for rss in contents:
webhook_body = json.dumps(
{
'text': f"\n<{rss.url}|{rss.title}>\r{rss.summary}",
'unfurl_links': u'true'
})
requests.post(webhook_url, data = webhook_body, timeout=20)
else:
pass
return
これらの関数を組み合わせて実行することで、最初に載せた結果を得ることができます。
RSSの仕様はゆるい約束事なので、RSS Feedによって配信される内容に結構ばらつきがあります。summary要素にそれが顕著で、投稿される内容を綺麗に揃えようとこだわると、別途本文の整形処理が必要です。このあたりは結構興味深かったので、気が向いたら別で記事を書くかもしれません。
NextStep
(気が向いたら)この処理を定期実行する環境を構築する予定です。実装codeの全体はその時にでも。
おわりに(無関係)
SNSの長所って良く言えば積極的情報供給(悪く言えば情報の洪水攻め)だと思うんですが、RSSの情報供給は、それに比べると少し控えめですよね。カジキである私にとって、RSSは往年の専門雑誌購読の良さを彷彿とさせます。