はじめに

当方はしがないバショウカジキなので、情報収集に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は往年の専門雑誌購読の良さを彷彿とさせます。

参考文献

[python,AWS]RSSをサーバーレスに自動取得してDiscordに送ってみる