はじめに
以前は以下の記事の中で紹介されている RSS Bot がありました。
残念ながらこの Bot は 2021 年ごろに retire してしまい、復活の気配もないので、RSS の情報を取得して、定期的に通知するような Bot を作りたいと思います。RSS Reader のような感覚で使えると思います。
一人で気軽に使うパターン(簡単)と、ある程度の人数の組織で使うようなパターン(少し面倒)の 2 種類の方法を紹介します。
一人で気軽に使うパターン(簡単)
ここでは一人で気軽に使うことを想定した簡単な方法を示します。
必要なものは以下の 2 つです。
- Webex Bot
- Python と cron(Python を定期的に実行するため) を使える環境(Linux 等)
以下に実装手順を示します。
Step 1
以下のページから Webex Bot を作成します。
以下のように適当に埋める。
- Bot name: rytakao-rss-bot
- Bot username: rytakao-rss-bot
- Icon: 用意されている 3 つの画像からどれか選ぶか適当な画像を upload
- App Hub Description: rytakao-RSS-bot (Bot の説明なので本当はもう少しきちんと書いたほうが良いとは思います。)
上記を記入/選択したら Add Bot -> Bot Access token が発行されるので、発行された Bot Access token をどこかに控えておきましょう。
Step 2
作成した Bot とのやりとりで使用する roomId を確認します。
- Step 2-1: Webex の検索バーから先程作成した Bot username(上記の例の場合であれば rytakao-rss-bot@webex.bot) を検索します。
- Step 2-2: Bot とのチャット画面が選択された状態で Webex 上部の Help > Copy Space Details を選択して Bot の情報をコピーします。
- Step 2-3: Step 2-2 でコピーした内容を適当なテキストエディタに貼り付けて Space ID の情報を確認します。これが Bot とのやりとりで使用される roomId です。
もし Space で Bot を使いたければ以下のようにしてください。
- Step 2-1: Space に Bot を追加
- Step 2-2: スペースが選択された状態で Webex 上部の Help > Copy Space Details を選択して Bot の情報をコピーします。
- Step 2-3: Step 2-2 でコピーした内容を適当なテキストエディタに貼り付けて Space ID の情報を確認します。
確認した roomId(Space ID) をどこかに控えておいてください。
Step 3
生成AI で適当に python のコードを生成します。
プロンプトは一旦は以下のような感じで良いと思います。
- Python で 24時間以内の RSS の情報を取得して Webex スペースに取得した情報を送るコードを作成してください。
生成AIが作成したコードを試験して問題なく動作するようなら以下のような条件を1つずつ加えていきます。
- Webex は一回に送れるメッセージの上限があるので、それを考慮して、上限を超えるようならメッセージを分割してください。
- 様々なエラーを考慮してエラーハンドリングのコードを加えてください。
最初から全ての条件を盛り込んでも良いと思いますが、最初から色々命令を与えると期待したものが出来ないことが多い印象があるので、まずはシンプルなものを作り、そこに色々機能を加えていった方が良いと思います。
適当に生成AIとやりとりして私の場合は以下のようなコードが出来上がりました。
rss-simple.py
import requests
import feedparser
from datetime import datetime, timedelta, timezone
import sys
from urllib.parse import quote
WEBEX_ROOM_ID = 'ここに Webex roomId'
RSS_FEED_URLS = [
'https://news.yahoo.co.jp/rss/topics/top-picks.xml',
'https://rss.itmedia.co.jp/rss/2.0/itmedia_all.xml',
'https://newsroom.cisco.com/c/services/i/servlets/newsroom/rssfeed.json'
]
WEBEX_BOT_TOKEN = 'ここに Webex Bot Access Token'
# Webex APIのメッセージ長制限 (バイト)
MAX_WEBEX_MESSAGE_LENGTH = 7439
# デバッグ用: 各フィードから取得するエントリ数を制限します。
# None に設定するか、大きな数を設定すると制限を無効にします。
DEBUG_MAX_ENTRIES_PER_FEED = 5 # 各フィードから最大5つのエントリのみ処理します。
def send_message_to_webex(room_id, message, webex_bot_token):
"""
Webexスペースにメッセージを送信するヘルパー関数。
Args:
room_id (str): WebexスペースID
message (str): 送信メッセージ
token (str): Webex Botのアクセストークン
"""
url = "https://webexapis.com/v1/messages"
headers = {
"Authorization": f"Bearer {webex_bot_token}",
"Content-Type": "application/json"
}
payload = {
"roomId": room_id,
"markdown": message
}
print(f"Debug: Sending payload to room {room_id}: {payload}", file=sys.stderr)
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # HTTPエラーが発生した場合、例外を発生させる
print(f"Message sent to Webex room {room_id} successfully!")
except requests.exceptions.HTTPError as e:
# HTTPエラーが発生しても、ここでは例外を再発生させずに処理を続行します。
# これにより、メッセージ送信が失敗しても、次のRSSフィードの処理に進むことができます。
print(f"Error sending message to Webex room {room_id}: {e}", file=sys.stderr)
if e.response is not None:
print(f"Webex API Response (Error): {e.response.text}", file=sys.stderr)
# raise # この行がコメントアウトされているため、エラーが発生してもプログラムは停止しません。
def notify_webex_rss_updates(webex_room_id, rss_feed_urls, webex_bot_token):
"""
指定された複数のRSSフィードURLから24時間以内の新しいエントリを取得し、
Webexスペースに通知する関数。メッセージが長い場合は複数に分割して送信します。
Args:
webex_room_id (str): 通知先のWebexスペースID
rss_feed_urls (list): RSSフィードのURLリスト
webex_bot_token (str): Webex Botのアクセストークン
"""
new_entries_text = []
now = datetime.now(timezone.utc)
time_threshold = now - timedelta(days=1) # 24時間前
for url in rss_feed_urls:
print(f"Processing RSS feed: {url} for Room ID: {webex_room_id}", file=sys.stderr)
try:
feed = feedparser.parse(url)
if feed.bozo: # フィードのパースエラーをチェック
print(f"Warning: RSS Feed '{url}' parsing error: {feed.bozo_exception}", file=sys.stderr)
continue
entry_count = 0
for entry in feed.entries:
# デバッグ用: エントリ数を制限
if DEBUG_MAX_ENTRIES_PER_FEED is not None and entry_count >= DEBUG_MAX_ENTRIES_PER_FEED:
print(f"Debug: Limiting entries for {url} to {DEBUG_MAX_ENTRIES_PER_FEED}", file=sys.stderr)
break # このフィードのエントリ処理を停止
# entry.published_parsedが存在しない場合もあるためチェック
if hasattr(entry, 'published_parsed'):
entry_time = datetime(*entry.published_parsed[:6], tzinfo=timezone.utc)
if entry_time > time_threshold:
# タイトルとリンクが存在するか確認
title = getattr(entry, 'title', 'No Title')
link = getattr(entry, 'link', 'No Link')
# 全角スペースを半角スペースに置換
title = title.replace('\u3000', ' ')
new_entries_text.append(f"- {title}\n {link}")
entry_count += 1
elif hasattr(entry, 'published'):
try:
# published_parsed が利用できない場合、published 文字列をそのまま利用し、
# 時間ベースのフィルタリングは行わないことを示すメッセージを追加
title = getattr(entry, 'title', 'No Title')
link = getattr(entry, 'link', 'No Link')
title = title.replace('\u3000', ' ')
new_entries_text.append(f"- {title}\n {link} (Published time not parsed for thresholding)")
entry_count += 1
except Exception as e:
print(f"Warning: Could not process entry from {url} due to published time issue: {e}", file=sys.stderr)
except Exception as e:
print(f"Error processing RSS feed '{url}': {e}", file=sys.stderr)
if new_entries_text:
# メッセージの先頭にヘッダーを追加
full_message_content = "【New RSS info】\n" + "\n".join(new_entries_text)
# 継続を示すサフィックス (日本語で「続く」)
continuation_suffix = "\n... (続く)"
suffix_byte_length = len(continuation_suffix.encode('utf-8'))
messages_to_send_final = []
current_message_lines = []
current_message_byte_length = 0
# メッセージの末尾に改行があることを保証し、一貫した行処理を行う
if not full_message_content.endswith('\n'):
full_message_content += '\n'
lines = full_message_content.split('\n')
for i, line in enumerate(lines):
line_with_newline = line + '\n'
line_byte_length = len(line_with_newline.encode('utf-8'))
# 元のメッセージの最後の行かどうかを判断
is_last_original_line = (i == len(lines) - 1)
# 現在のメッセージパートに許容される最大コンテンツ長を決定
# 最後のパートでない場合、サフィックス分のスペースを確保する
max_allowed_content_length = MAX_WEBEX_MESSAGE_LENGTH
if not is_last_original_line:
max_allowed_content_length -= suffix_byte_length
# 単一の行がメッセージパートの最大長を超える場合
# この行はそれ自体で一つのメッセージとして送信される(Webex側で切り詰められる可能性あり)
if line_byte_length > MAX_WEBEX_MESSAGE_LENGTH:
if current_message_lines: # 蓄積された行がある場合、まずそれを確定して送信
# 現在確定するパートは、全体から見て最後のパートではないため、サフィックスを追加
messages_to_send_final.append("".join(current_message_lines).strip() + continuation_suffix)
current_message_lines = []
current_message_byte_length = 0
# 長すぎる行を単独のメッセージとして追加(サフィックスは付けない)
messages_to_send_final.append(line.strip())
continue # 次の行へ
# 現在のメッセージパートに行を追加しようとする
if current_message_byte_length == 0:
# 新しいメッセージパートを開始
current_message_lines.append(line_with_newline)
current_message_byte_length += line_byte_length
else:
# この行を追加すると最大許容コンテンツ長を超えるかチェック
if (current_message_byte_length + line_byte_length) > max_allowed_content_length:
# 現在のメッセージパートが満杯。確定して送信リストに追加。
# 元のメッセージの最後の行でない場合、サフィックスを追加。
if not is_last_original_line:
messages_to_send_final.append("".join(current_message_lines).strip() + continuation_suffix)
else:
# 最後のパートなのでサフィックスは不要
messages_to_send_final.append("".join(current_message_lines).strip())
# 新しいメッセージパートを現在の行で開始
current_message_lines = [line_with_newline]
current_message_byte_length = line_byte_length
else:
# 行が収まるので、現在のメッセージパートに追加
current_message_lines.append(line_with_newline)
current_message_byte_length += line_byte_length
# 残っている行があれば、それを最後のメッセージパートとして追加(サフィックスは不要)
if current_message_lines:
messages_to_send_final.append("".join(current_message_lines).strip())
# 空のパートを除外
messages_to_send_final = [p for p in messages_to_send_final if p.strip()]
# 分割された各メッセージをWebexに送信
for i, message_part in enumerate(messages_to_send_final):
print(f"Debug: Sending message part {i+1}/{len(messages_to_send_final)}. Length: {len(message_part.encode('utf-8'))} bytes.", file=sys.stderr)
print(message_part) # デバッグ用にメッセージ内容も出力
send_message_to_webex(webex_room_id, message_part, webex_bot_token)
else:
print(f"No new RSS entries found in the last 24 hours for Room ID: {webex_room_id} and URL(s): {rss_feed_urls}.", file=sys.stderr)
notify_webex_rss_updates(WEBEX_ROOM_ID, RSS_FEED_URLS, WEBEX_BOT_TOKEN)
以下の項目についてはさきほど控えた roomId, Bot Access Token, 取得したい RSS の URL に変えてください。
WEBEX_ROOM_ID = 'ここに Webex roomId'
RSS_FEED_URLS = [
'https://news.yahoo.co.jp/rss/topics/top-picks.xml',
'https://rss.itmedia.co.jp/rss/2.0/itmedia_all.xml',
'https://newsroom.cisco.com/c/services/i/servlets/newsroom/rssfeed.json'
]
WEBEX_BOT_TOKEN = 'ここに Webex Bot Access Token'
また、WEBEX_BOT_TOKEN をハードコードするのはセキュリティー上あまりよろしくはないので、必要に応じて、入力方法は工夫しましょう。
Step 4
作成した python の実行環境の作成
- Step 4-1: 適当な directory を作成します。ここでは simple-rss を作成しています。
rytakao@RYTAKAO-M-X0JK ~ % pwd
/Users/rytakao
rytakao@RYTAKAO-M-X0JK ~ % mkdir simple-rss
rytakao@RYTAKAO-M-X0JK ~ %
- Step 4-2: python の仮想環境を作成します。
rytakao@RYTAKAO-M-X0JK ~ % python3 -m venv simple-rss
rytakao@RYTAKAO-M-X0JK ~ %
- Step 4-3: 仮想環境を Activate します。
rytakao@RYTAKAO-M-X0JK ~ % source simple-rss/bin/activate
(simple-rss) rytakao@RYTAKAO-M-X0JK ~ %
- Step 4-4: 必要なモジュールを適当に install します。
(simple-rss) rytakao@RYTAKAO-M-X0JK ~ % pip install quote requests feedparser
- Step 4-5: Step 3 で作成した python を ~/simple-rss 配下に配置して動作確認します。
(simple-rss) rytakao@RYTAKAO-M-X0JK ~ % python3 ~/simple-rss/simple-rss.py
Step 5
cron で Step 3 で作成した python を定期的に実行するようにします。
crontab -e を実行し以下のように記載する。ご自身の環境に合わせて適当に変えてください。
0 9 * * * cd /Users/rytakao/simple-rss; /Users/rytakao/simple-rss/bin/python simple-rss.py
0 9 * * * の部分は左から順に 分、時、日、月、曜日 です。0 9 * * * だと毎日朝9 時に、その右のコマンドを実行します。
cd /Users/rytakao/simple-rss で python file があるディレクトリに移動しています。
/Users/rytakao/simple-rss/bin/python simple-rss.py で venv で simple-rss.py を実行しています。
Step 6
指定した時間になると作成した bot から RSS の情報が送られてきます。
rytakao-rss-bot 09:00
【New RSS info】
• 黄砂 2日にかけ東京に飛ぶ可能性
https://news.yahoo.co.jp/pickup/6560621?source=rss
• 「困難な船出」1カ月 政権の今後
https://news.yahoo.co.jp/pickup/6560625?source=rss
• 全日空95便欠航 約1万3200人影響
https://news.yahoo.co.jp/pickup/6560615?source=rss
• 小6が裸の写真を送信 巧妙な手口
https://news.yahoo.co.jp/pickup/6560627?source=rss
• 浅草寺横40年「占拠」裁判の背景
https://news.yahoo.co.jp/pickup/6560626?source=rss
• [ITmedia News] コンデジ復活の象徴、キヤノン「IXY 650m」に触れて改めて思う"スマホでは味わえない楽しさ"とは?
https://www.itmedia.co.jp/news/articles/2511/29/news006.html
• [ITmedia News] 暗黒物質(ダークマター)がついに"見えた"? 天の川銀河で「よく似たガンマ線」、東大が発見
https://www.itmedia.co.jp/news/articles/2511/29/news038.html
• [ITmedia Mobile] 人は見かけではないのと同様に、モバイルオーダーも見かけではない
https://www.itmedia.co.jp/mobile/articles/2511/29/news037.html
• [ITmedia News] 健康保険証は12月1日で"期限切れ"に マイナ保険証がまだない場合はどうすべき?
https://www.itmedia.co.jp/news/articles/2511/29/news028.html
• [ITmedia News] "通常価格"に実態なし ツルハグループの通販サイトに景表法違反 原因は「メンテナンス不備」
https://www.itmedia.co.jp/news/articles/2511/29/news036.html
以上となります。
ある程度の人数の組織で使うようなパターン(少し面倒)
ここではある程度の人数の組織で使うようなことを想定した方法を示します。上で紹介した方法より少し面倒です。
以下のような機能を持たせます。
- Bot に命令を送って RSS の URL の登録や削除を行えるようにします。
- 新着の RSS 情報を通知する時間を指定できるようにします。
- Space ごとにこれらの情報を管理して、複数のユーザーで Bot を使えるようにします。
必要なものは以下の 4 つです。
- Webex Bot
- AWS Lambda
- AWS EventBridge
- AWS DynamoDB
AWS Lambda, AWS EventBridge, AWS DynamoDB については他の Webhook サービス、DataBase 等でも良いと思いますが、比較的気軽に使えて利用者が多く情報が多いのでこれらのサービスにしました。
以下に実装手順を示します。Step 2 までは上のものと同様です。
Step 1
以下のページから Webex Bot を作成します。
以下のように適当に埋める。
- Bot name: rytakao-rss-bot
- Bot username: rytakao-rss-bot
- Icon: 用意されている 3 つの画像からどれか選ぶか適当な画像を upload
- App Hub Description: rytakao-RSS-bot (Bot の説明なので本当はもう少しきちんと書いたほうが良いとは思います。)
上記を記入/選択したら Add Bot -> Bot Access token が発行されるので、発行された Bot Access token をどこかに控えておきましょう。
Step 2
作成した Bot とのやりとりで使用する roomId を確認します。
- Step 2-1: Webex の検索バーから先程作成した Bot username(上記の例の場合であれば rytakao-rss-bot@webex.bot) を検索します。
- Step 2-2: Bot とのチャット画面が選択された状態で Webex 上部の Help > Copy Space Details を選択して Bot の情報をコピーします。
- Step 2-3: Step 2-2 でコピーした内容を適当なテキストエディタに貼り付けて Space ID の情報を確認します。これが Bot とのやりとりで使用される roomId です。
もし Space で Bot を使いたければ以下のようにしてください。
- Step 2-1: Space に Bot を追加
- Step 2-2: スペースが選択された状態で Webex 上部の Help > Copy Space Details を選択して Bot の情報をコピーします。
- Step 2-3: Step 2-2 でコピーした内容を適当なテキストエディタに貼り付けて Space ID の情報を確認します。
確認した roomId(Space ID) をどこかに控えておいてください。
Step 3
AWS のサービスを使用するためのアカウントがない方はアカウントを作成してください。
https://aws.amazon.com/jp/ にアクセスして「アカウントの作成」からアカウントを作成します。
Step 4
Bot が受け取ったメッセージの情報を保存しておくための DataBase を用意します。今回は DynamoDB を使用します。
- Step 4-1: https://console.aws.amazon.com/console/home/?nc2=h_si&src=header-signin から AWS サービスコンソールにアクセスします。
- Step 4-2: 上の検索バーから DynamoDB を検索して DynamoDB のサービスコンソール画面にアクセスします。
- Step 4-3: テーブルの作成をクリック
- Step 4-4: 以下のようにテーブルを作成
- テーブル名: RssSubscriptions
- パーティションキー: roomId → 文字列(default)を選択
- ソートキー - オプション: rssUrl → 文字列(default)を選択
- 他の項目は default のまま「テーブルの作成」をクリック
- Step 4-5: 再度テーブルの作成をクリック
- Step 4-6: 以下のようにテーブルを作成
- テーブル名: RssNotificationSchedules
- パーティションキー: roomId → 文字列(default)を選択
- ソートキー - オプション: 空のままで OK
- 他の項目は default のまま「テーブルの作成」をクリック
Step 5
Bot が受け取ったメッセージを処理するための Webhook のサービスを用意します。
今回は AWS Lambda を利用します。
AWS Lambda で 2 つの関数を作成します。
| 関数名 | 役割 |
|---|---|
| webex-rss-webhook-handler | Webhook 用。ユーザーの指示を受けて設定をデータベースに保存/変更 |
| webex-rss-scheduler | データベースに保存された情報を見て定期的に実行&通知する |
ここではまず Webhook 用に webex-rss-webhook-handler を作成します。
- Step 5-1: https://console.aws.amazon.com/console/home/?nc2=h_si&src=header-signin から AWS サービスコンソールにアクセスします。
- Step 5-2: 上の検索バーから Lambda を検索して Lambda のサービスコンソール画面ににアクセスします。
- Step 5-3: 「関数の作成」をクリック
- Step 5-4: 「一から作成」を選択して以下のように入力します。
- 関数名: webex-rss-webhook-handler
- ランタイム: Python 3.14
- アーキテクチャ: どちらでも良いですが今回は x86_64 を選択します。
- デフォルトの実行ロールの変更: 基本的な Lambda アクセス権限で新しいロールを作成
以上を入力したら「関数の作成」
- Step 5-5: 環境変数を設定します。設定 > 環境変数 > 編集 > 環境変数の追加
以下のように環境変数を設定します。
| キー | 値 |
|---|---|
| WEBEX_BOT_TOKEN | Step 1 で控えた Bot Access token |
| DDB_SUBSCRIPTIONS_TABLE | RssSubscriptions (先ほど作成したテーブル名) |
| DDB_SCHEDULES_TABLE | RssNotificationSchedules (先ほど作成したテーブル名) |
| DEFAULT_TIMEZONE | Asia/Tokyo |
設定が終わったら「保存」をクリック
- Step 5-6: アクセス権限を設定します。設定 > アクセス権限 > ロール名 webex-rss-webhook-handler-role-xxxxxxxx をクリック > IAM のロール画面へ遷移。
- Step 5-7: 許可を追加 > ポリシーをアタッチ > AmazonDynamoDBFullAccess を検索欄に入力し AmazonDynamoDBFullAccess にチェックをつける。※AmazonDynamoDBFullAccess_v2 には付けなくて大丈夫です。 > 許可を追加
- Step 5-8: lambda_function.py を編集する。
default では以下のコードが設定されていると思います。
import json
def lambda_handler(event, context):
# TODO implement
return {
'statusCode': 200,
'body': json.dumps('Hello from Lambda!')
}
コードの中身を以下に入れ替えてください。
import os
import json
import boto3
import urllib.request
from boto3.dynamodb.conditions import Key
DDB_SUBSCRIPTIONS_TABLE = os.environ["DDB_SUBSCRIPTIONS_TABLE"]
DDB_SCHEDULES_TABLE = os.environ["DDB_SCHEDULES_TABLE"]
WEBEX_BOT_TOKEN = os.environ["WEBEX_BOT_TOKEN"]
dynamodb = boto3.resource("dynamodb")
subs_table = dynamodb.Table(DDB_SUBSCRIPTIONS_TABLE)
schedules_table = dynamodb.Table(DDB_SCHEDULES_TABLE)
WEBEX_API_BASE = "https://webexapis.com/v1"
def lambda_handler(event, context):
"""
API Gateway からの HTTP リクエストを想定したハンドラ。
event["body"] に Webex Webhook からの JSON が入る。
"""
# Webex からのヘルスチェックや検証リクエストに応じる場合は、
# ヘッダや body の中身を見て分岐させることもありますが、
# ここでは「messages / created」のみを扱う前提で実装します。
try:
body = json.loads(event.get("body", "{}"))
except json.JSONDecodeError:
return _response(400, {"message": "Invalid JSON"})
resource = body.get("resource")
event_type = body.get("event")
data = body.get("data", {})
# メッセージ作成イベントのみ処理
if resource == "messages" and event_type == "created":
message_id = data.get("id")
room_id = data.get("roomId")
person_id = data.get("personId")
# Bot 自身のメッセージは無視(無限ループ防止)
bot_person_id = _get_bot_person_id()
if person_id == bot_person_id:
return _response(200, {"message": "Ignored bot's own message"})
# メッセージ内容取得
message = _get_message(message_id)
text = message.get("text", "").strip()
# コマンド解析 & 実行
reply_text = handle_command(room_id, text)
# Webex に返信
_post_message(room_id, reply_text)
# 何が来ても 200 を返す(Webhook の再送防止)
return _response(200, {"message": "OK"})
# --------------------------------------------------
# コマンド処理部分
# --------------------------------------------------
def handle_command(room_id, text):
"""
Bot に送られてきたメッセージ (text) を解析し、
RSS 登録/削除/一覧/スケジュール設定などを行う。
"""
tokens = text.split()
if not tokens:
return "コマンドを認識できませんでした。`help` と入力するとヘルプが表示されます。"
cmd = tokens[0].lower()
if cmd == "help":
return get_help_text()
# add rss <URL>
if cmd == "add" and len(tokens) >= 3 and tokens[1].lower() == "rss":
rss_url = tokens[2]
subs_table.put_item(
Item={
"roomId": room_id,
"rssUrl": rss_url,
"createdAt": _now_iso(),
"lastFetchedAt": None,
}
)
return f"RSS を登録しました: {rss_url}"
# remove rss <URL>
if cmd == "remove" and len(tokens) >= 3 and tokens[1].lower() == "rss":
rss_url = tokens[2]
subs_table.delete_item(
Key={
"roomId": room_id,
"rssUrl": rss_url,
}
)
return f"RSS を削除しました: {rss_url}"
# list rss
if cmd == "list" and len(tokens) >= 2 and tokens[1].lower() == "rss":
resp = subs_table.query(
KeyConditionExpression=Key("roomId").eq(room_id)
)
items = resp.get("Items", [])
if not items:
return "登録されている RSS はありません。`add rss <URL>` で追加できます。"
lines = ["登録済みの RSS 一覧:"]
for it in items:
lines.append(f"- {it['rssUrl']}")
return "\n".join(lines)
# set schedule daily HH:MM
if cmd == "set" and len(tokens) >= 4 and tokens[1].lower() == "schedule":
sched_type = tokens[2].lower() # 例: daily
time_str = tokens[3] # 例: 09:00
if not _validate_time_str(time_str):
return "時間の形式が不正です。`HH:MM` 形式で指定してください(例: `set schedule daily 09:00`)。"
schedules_table.put_item(
Item={
"roomId": room_id,
"type": sched_type,
"time": time_str,
"timezone": os.environ.get("DEFAULT_TIMEZONE", "Asia/Tokyo"),
"enabled": True,
}
)
return f"通知スケジュールを設定しました: {sched_type} {time_str}"
# disable schedule
if cmd == "disable" and len(tokens) >= 2 and tokens[1].lower() == "schedule":
schedules_table.update_item(
Key={"roomId": room_id},
UpdateExpression="SET #enabled = :false",
ExpressionAttributeNames={"#enabled": "enabled"},
ExpressionAttributeValues={":false": False},
)
return "通知スケジュールを無効化しました。"
# check now(今回は「受け付けました」だけ返す簡易版)
if cmd == "check" and len(tokens) >= 2 and tokens[1].lower() == "now":
return "RSS のチェックをリクエストしました。(定期実行 Lambda 側の実装に応じて処理を追加してください)"
# 想定外コマンド
return "コマンドを認識できませんでした。`help` と入力するとヘルプが表示されます。"
def get_help_text():
return (
"利用可能なコマンド:\n"
"- `add rss <URL>`: RSS フィードを登録します。\n"
"- `remove rss <URL>`: 登録済みの RSS フィードを削除します。\n"
"- `list rss`: 登録済みの RSS フィード一覧を表示します。\n"
"- `set schedule daily HH:MM`: 毎日指定時刻に通知するよう設定します。\n"
"- `disable schedule`: RSS 通知を停止します。\n"
"- `check now`: 直近 24 時間以内の RSS 記事をすぐに確認します(要実装)。\n"
"- `help`: このヘルプを表示します。"
)
# --------------------------------------------------
# Webex API 呼び出し部分
# --------------------------------------------------
def _get_bot_person_id():
"""
Bot 自身の personId を Webex API から取得。
毎回取っているが、環境変数やメモリにキャッシュしてもよい。
"""
url = f"{WEBEX_API_BASE}/people/me"
req = urllib.request.Request(
url,
headers={"Authorization": f"Bearer {WEBEX_BOT_TOKEN}"},
)
with urllib.request.urlopen(req) as res:
data = json.loads(res.read().decode())
return data["id"]
def _get_message(message_id):
"""
message_id から Webex メッセージ詳細を取得。
"""
url = f"{WEBEX_API_BASE}/messages/{message_id}"
req = urllib.request.Request(
url,
headers={"Authorization": f"Bearer {WEBEX_BOT_TOKEN}"},
)
with urllib.request.urlopen(req) as res:
data = json.loads(res.read().decode())
return data
def _post_message(room_id, text):
"""
指定 roomId に Bot としてメッセージを送信。
markdown フィールドを使うことでリンクや箇条書きが利用可能。
"""
url = f"{WEBEX_API_BASE}/messages"
body = json.dumps({"roomId": room_id, "markdown": text}).encode("utf-8")
req = urllib.request.Request(
url,
data=body,
headers={
"Authorization": f"Bearer {WEBEX_BOT_TOKEN}",
"Content-Type": "application/json",
},
method="POST",
)
with urllib.request.urlopen(req) as res:
res.read()
# --------------------------------------------------
# ユーティリティ
# --------------------------------------------------
def _now_iso():
from datetime import datetime, timezone
return datetime.now(timezone.utc).isoformat()
def _validate_time_str(time_str):
"""
HH:MM 形式かどうかの簡易チェック。
"""
try:
parts = time_str.split(":")
if len(parts) != 2:
return False
h = int(parts[0])
m = int(parts[1])
return 0 <= h <= 23 and 0 <= m <= 59
except Exception:
return False
def _response(status_code, body):
"""
API Gateway 用の HTTP レスポンスを返す。
"""
return {
"statusCode": status_code,
"headers": {"Content-Type": "application/json"},
"body": json.dumps(body),
}
左の「Deploy」ボタンをクリックします。
Step 6
Lambda 単体では Webex から直接呼べないので、API Gateway を間に挟みます。
- Step 6-1: https://console.aws.amazon.com/console/home/?nc2=h_si&src=header-signin から AWS サービスコンソールにアクセスします。
- Step 6-2: 上の検索バーから HTTP API を検索して HTTP API のサービスコンソール画面ににアクセスします。
- Step 6-2: API の作成 > API タイプを選択: HTTP API > 構築 > API 名: HTTP-API-RSS, IP アドレスのタイプ: IPv4 > 統合を追加 > Lambda > Lambda 関数 に webex-rss-webhook-handler 関数を指定 > 次へ > ルートを追加 > メソッド: POST, リソースパス: /webex, 統合ターゲット: webex-rss-webhook-handler > 次へ > 次へ > 作成
- Step 6-3: 左のタブから Deploy > Stages > 「HTTP-API-RSS のステージ」の横の「作成」をクリック > 名前: prod, 他の項目は default のまま作成 > 右上のデプロイ > ステージに prod を選択 > デプロイ
- Step 6-4: 「URL を呼び出す」の部分の URL にリソースパス(/webex)を追加したものが webhook のための URL です。webhook のための URL を控えておきましょう。
例: https://y0ap8fgs43.execute-api.ap-northeast-1.amazonaws.com/prod/webex
Step 7
Webex 側の Webhook 設定を行います。
https://developer.webex.com/meeting/docs/api/v1/webhooks/create-a-webhook にアクセスして Webhook を作成します。
右側の Parameters の部分の項目を以下のように埋め流。
- name: webhook-RSS
- targetUrl: Step 6-4 で控えた URL
- resource: messages
- event: created
- Filter: 空欄のまま
- Secret: 空欄のまま
以上の項目を埋めたら「Run」をクリックして Run の下に「Response: 200」が表示されれば OK です。
Step 8
データベースに保存された情報を見て定期的に実行&通知するために Lambda 関数 webex-rss-scheduler を作成します。
- Step 8-1: https://console.aws.amazon.com/console/home/?nc2=h_si&src=header-signin から AWS サービスコンソールにアクセスします。
- Step 8-2: 上の検索バーから Lambda を検索して Lambda のサービスコンソール画面ににアクセスします。
- Step 8-3: 「関数の作成」をクリック
- Step 8-4: 「一から作成」を選択して以下のように入力します。
- 関数名: webex-rss-scheduler
- ランタイム: Python 3.14
- アーキテクチャ: どちらでも良いですが今回は x86_64 を選択します。
- デフォルトの実行ロールの変更: 基本的な Lambda アクセス権限で新しいロールを作成
以上を入力したら「関数の作成」
- Step 8-5: 環境変数を設定します。設定 > 環境変数 > 編集 > 環境変数の追加
以下のように環境変数を設定します。
| キー | 値 |
|---|---|
| WEBEX_BOT_TOKEN | Step 1 で控えた Bot Access token |
| DDB_SUBSCRIPTIONS_TABLE | RssSubscriptions (先ほど作成したテーブル名) |
| DDB_SCHEDULES_TABLE | RssNotificationSchedules (先ほど作成したテーブル名) |
| DEFAULT_TIMEZONE | Asia/Tokyo |
設定が終わったら「保存」をクリック
- Step 8-6: アクセス権限を設定します。設定 > アクセス権限 > ロール名 webex-rss-scheduler-role-xxxxxxxx をクリック > IAM のロール画面へ遷移。
- Step 8-7: 許可を追加 > ポリシーをアタッチ > AmazonDynamoDBFullAccess を検索欄に入力し AmazonDynamoDBFullAccess にチェックをつける。※AmazonDynamoDBFullAccess_v2 には付けなくて大丈夫です。 > 許可を追加
- Step 8-8: ローカルPC で以下のコマンドを実行して python で import する feedparser を zip 化します。
mkdir feedparser_layer
cd feedparser_layer
mkdir python
pip install --upgrade feedparser -t python/
zip -r feedparser_layer.zip python
- Step 8-9: 新しいレイヤーの作成
Lambda の top ページにアクセス > 左側のメニューの「レイヤー」 > レイヤーを作成 > 名前: feedparser_layer > .zip ファイルをアップロード > ファイルを選択 > Step 8-8 で作成したファイルを選択 > 互換性のあるランタイム - オプション: Python 3.14 > 作成
- Step 8-10: レイヤーをアタッチ
Lambda の top ページにアクセス > 関数 > webex-rss-scheduler > コード > 一番下のレイヤーのところで「レイヤーの追加」 > カスタムレイヤー > カスタムレイヤー: 先ほど作成した feedparser_layer を選択 > バージョン: 1 > 追加
- Step 8-11: webex-rss-scheduler の lambda_function.py を更新
コードタブを選択し lambda_function.py の内容を更新します。
default では lambda_function.py の内容は以下になっています。
import json
def lambda_handler(event, context):
# TODO implement
return {
'statusCode': 200,
'body': json.dumps('Hello from Lambda!')
}
この内容を以下に変更します。
import os
import json
import boto3
import urllib.request
from datetime import datetime, timedelta, timezone
import feedparser
from boto3.dynamodb.conditions import Key
DDB_SUBSCRIPTIONS_TABLE = os.environ["DDB_SUBSCRIPTIONS_TABLE"]
DDB_SCHEDULES_TABLE = os.environ["DDB_SCHEDULES_TABLE"]
WEBEX_BOT_TOKEN = os.environ["WEBEX_BOT_TOKEN"]
DEFAULT_TIMEZONE = os.environ.get("DEFAULT_TIMEZONE", "Asia/Tokyo")
dynamodb = boto3.resource("dynamodb")
subs_table = dynamodb.Table(DDB_SUBSCRIPTIONS_TABLE)
schedules_table = dynamodb.Table(DDB_SCHEDULES_TABLE)
WEBEX_API_BASE = "https://webexapis.com/v1"
def lambda_handler(event, context):
"""
EventBridge からの定期実行を想定。
5分ごとなどで起動し、「現在時刻に通知すべき roomId があれば処理する」。
"""
now_utc = datetime.now(timezone.utc)
# 簡易的に JST 固定
jst = now_utc + timedelta(hours=9)
current_hhmm = jst.strftime("%H:%M")
print(f"[DEBUG] now_utc={now_utc.isoformat()}, jst={jst.isoformat()}, current_hhmm={current_hhmm}")
try:
resp = schedules_table.scan()
except Exception as e:
print(f"[ERROR] failed to scan schedules table: {e}")
return
schedules = resp.get("Items", [])
print(f"[DEBUG] schedules raw count={len(schedules)}")
for sched in schedules:
print(f"[DEBUG] check schedule item={sched}")
if not sched.get("enabled", False):
print("[DEBUG] skip: enabled is False")
continue
if sched.get("type") != "daily":
print(f"[DEBUG] skip: type is not daily (type={sched.get('type')})")
continue
sched_time = sched.get("time")
if sched_time != current_hhmm:
print(f"[DEBUG] skip: time mismatch (sched={sched_time}, now={current_hhmm})")
continue
room_id = sched["roomId"]
print(f"[DEBUG] process room {room_id}")
process_room(room_id, now_utc)
def process_room(room_id, now_utc):
"""
特定の roomId について、登録済み RSS を取得し、
直近24時間以内の新着記事を Webex に通知する。
"""
print(f"[DEBUG] process_room start room_id={room_id}")
try:
resp = subs_table.query(
KeyConditionExpression=Key("roomId").eq(room_id)
)
except Exception as e:
print(f"[ERROR] failed to query subscriptions for room_id={room_id}: {e}")
return
items = resp.get("Items", [])
print(f"[DEBUG] subscriptions count for room={len(items)}")
if not items:
print("[DEBUG] no subscriptions, return")
return
since = now_utc - timedelta(hours=24)
print(f"[DEBUG] since={since.isoformat()}")
new_entries = []
for it in items:
rss_url = it["rssUrl"]
print(f"[DEBUG] fetching RSS {rss_url}")
try:
feed = feedparser.parse(rss_url)
except Exception as e:
print(f"[ERROR] Failed to fetch RSS from {rss_url}: {e}")
continue
entries_count = len(getattr(feed, "entries", []))
print(f"[DEBUG] feed entries count={entries_count} for {rss_url}")
for entry in feed.entries:
published = getattr(entry, "published_parsed", None) or getattr(
entry, "updated_parsed", None
)
if not published:
continue
published_dt = datetime(
published.tm_year,
published.tm_mon,
published.tm_mday,
published.tm_hour,
published.tm_min,
published.tm_sec,
tzinfo=timezone.utc,
)
if published_dt >= since:
title = getattr(entry, "title", "No title")
link = getattr(entry, "link", "")
print(f"[DEBUG] new entry: {title} {link} (published={published_dt.isoformat()})")
new_entries.append((title, link))
try:
subs_table.update_item(
Key={"roomId": room_id, "rssUrl": rss_url},
UpdateExpression="SET lastFetchedAt = :t",
ExpressionAttributeValues={":t": now_utc.isoformat()},
)
except Exception as e:
print(f"[ERROR] failed to update lastFetchedAt for {rss_url}: {e}")
print(f"[DEBUG] new_entries count={len(new_entries)}")
if not new_entries:
print("[DEBUG] no new entries, return")
return
MAX_ENTRIES = 50
if len(new_entries) > MAX_ENTRIES:
print(f"[DEBUG] new_entries > {MAX_ENTRIES}, truncate")
new_entries = new_entries[:MAX_ENTRIES]
lines = ["直近24時間以内の新着 RSS 記事:"]
for title, link in new_entries:
if link:
lines.append(f"- [{title}]({link})")
else:
lines.append(f"- {title}")
text = "\n".join(lines)
_post_message(room_id, text)
def _post_message(room_id, text):
"""
指定 roomId に Webex メッセージを送信。
"""
url = f"{WEBEX_API_BASE}/messages"
body_dict = {"roomId": room_id, "markdown": text}
body = json.dumps(body_dict).encode("utf-8")
print(f"[DEBUG] posting to Webex room_id={room_id}, body={body_dict}")
req = urllib.request.Request(
url,
data=body,
headers={
"Authorization": f"Bearer {WEBEX_BOT_TOKEN}",
"Content-Type": "application/json",
},
method="POST",
)
try:
with urllib.request.urlopen(req) as res:
resp_body = res.read().decode()
status = getattr(res, "status", None)
print(f"[DEBUG] Webex response status={status}, body={resp_body}")
except Exception as e:
print(f"[ERROR] failed to post message to Webex: {e}")
- Step 8-12: Deploy をクリック
Step 9
webex-rss-scheduler を定期的に呼び出すために EventBridge(CloudWatch Events)の設定を行います。
- Step 9-1: https://console.aws.amazon.com/console/home/?nc2=h_si&src=header-signin から AWS サービスコンソールにアクセスします。
- Step 9-2: 上の検索バーから EventBridge を検索して EventBridge のサービスコンソール画面ににアクセスします。
- Step 9-3: ルールを作成します。
左メニューのルール > ルールを作成の右側の ▼ をクリック > スケジュールされたルールを作成 > 名前: webex-rss-scheduler-every-1-min > そのほかの項目は default のまま「次へ」 > 「特定の時刻 (毎月第 1 月曜日の午前 8 時 (PST) など) に実行されるきめ細かいスケジュール。」を選択 > cron 式を 0/1 分, * 時間, * 日付, * 月, ? 曜日, * 年 で設定。これで 1 分ごとに動作します。 > 次へ > ターゲットタイプ: AWS のサービス: Lambda 関数 > ターゲットの場所: このアカウントのターゲット > webex-rss-scheduler を選択 > そのほかの項目は default のまま「次へ」 > 「次へ」 > ルールの作成
少し長かったですが以上で設定は完了です。
Step 10
作成した bot で利用可能なコマンド:
• add rss : RSS フィードを登録します。
• remove rss : 登録済みの RSS フィードを削除します。
• list rss: 登録済みの RSS フィード一覧を表示します。
• set schedule daily HH:MM: 毎日指定時刻に通知するよう設定します。
• disable schedule: RSS 通知を停止します。
• help: このヘルプを表示します。
動作確認のために適当な URL を add rss <URL> で追加します。
add rss https://news.yahoo.co.jp/rss/topics/top-picks.xml
正常に追加できると以下のようなメッセージが返ってきます。
RSS を登録しました: https://news.yahoo.co.jp/rss/topics/top-picks.xml
RSS の URL は複数登録可能です。
登録されている URL を確認するには list rss を実行します。
数秒待つと登録されている URL が返ってきます。
登録済みの RSS 一覧:
• https://news.yahoo.co.jp/rss/topics/top-picks.xml
• https://rss.itmedia.co.jp/rss/2.0/itmedia_all.xml
set schedule daily HH:MM でRSS の通知を受ける時間をセットします。
例:
set schedule daily 09:00
この設定の場合は毎朝 09:00 に RSS のサイトの 24 時間以内の新着情報を通知します。
スケジュールされた時間になると以下のような通知があります。※実際にはそれぞれ News への Link が張られています。
直近24時間以内の新着 RSS 記事:
• タイとカンボジア 双方に食い違い
• 紀州資産家二審 無罪は誤りと検察
• 政治資金でキャバクラ 吉村氏苦言
• 小学生3人が車にはねられる 逮捕
• ポッキーなど20品目600万個を回収
• サンリオ 大分の施設リゾート化へ
• 異例のドラ1不在 SB新入団会見
• 日テレ 松岡昌宏&城島茂への思い
• https://www.itmedia.co.jp/pcuser/articles/2512/08/news093.html
• [ITmedia PC USER] Desktop Mateの新DLC「仕事猫」が12月15日発売 Ctrl+Sで指さし確認「ヨシ!」
• [ITmedia PC USER] Ryzen 5 PRO 6650Hを搭載したGMKtecのミニPC「M8」が25%オフで5万円切りに
• [ITmedia PC USER] ラトック、RS-232CポートをWi-Fi化できる変換アダプター
• [ITmedia エンタープライズ] NECはなぜ「AIによるDXの推進」を強調するのか? 2026年のIT業界の注目ポイントとともに考察
• [ITmedia News] ドコモ、「ひかりTV」を値上げ、3月から 基本プランは1100円→1210円に
• [ITmedia エンタープライズ] 「コーディングはAI任せ」でエンジニアは何をする? AWSが示す、開発の新たな“主戦場”
• [ITmedia PC USER] エージェントAI時代のWindowsはどうなる? Microsoftの苦悩
• [ITmedia ビジネスオンライン] サンリオ、大分「ハーモニーランド」を大規模刷新 “滞在型”リゾートへ
• [ITmedia PC USER] YouTube Premiumサブスクリプションを割引利用できるGoogle One利用者向けアドオンが国内でも提供開始
• [ITmedia Mobile] 3COINSで330円の「ガイド付きガラスフィルム」を試す iPhone 13~16シリーズに対応、100円ショップよりもハイレベル
• [ITmedia News] 令和に蘇った「まんが日本昔ばなし」が人気に 公式YouTubeの登録者数14万人超え
• [ITmedia Mobile] 中古スマホやゲーム機がおトクになる「GEO クリスマスSALE 2025」12月13日から 抽選でカニやPS5も当たる
• [ITmedia PC USER] Swann、ソーラーパネル充電パネルを備えた屋外対応のセキュリティカメラ
• [ITmedia News] メルカリモバイル、複数回線契約に対応 最大5回線まで
• [ITmedia News] 映画「ズートピア2」を見たマンガ家が感じた、王道だからこその魅力と不満 「メタルギアソリッド」の小島秀夫監督もカメオ出演
• [ITmedia News] The New York Times、Perplexityを提訴 有料コンテンツの「盗用」を主張
• [ITmedia PC USER] 10年以上ぶりにSapphire製マザーボードが売り場に並ぶ
• [ITmedia Mobile] 手のひらサイズのポータブル電源「MIRENEX NEX430」、Makuakeで先行販売 6万mAhのバッテリー搭載
• https://www.itmedia.co.jp/pcuser/articles/2512/08/news067.html
• [ITmedia ビジネスオンライン] なぜフジテレビは失敗し、アイリスオーヤマは成功したのか 危機対応で見えた「会社の本性」
• [ITmedia Mobile] 「UGREEN USB Type-Cケーブル 100W PD対応 1m」が41%オフの708円で販売中
• [ITmedia Mobile] 「ゼンハイザー HD 280 PRO MK2」が11%オフの1万円で販売中
• [ITmedia エグゼクティブ] リモート相談やドローン配送 ローソン、地域創生拠点のコンビニを全国100カ所に展開
• [ITmedia Mobile] 「Anker 521 Power Bank」が44%オフの4990円で販売中
• https://www.itmedia.co.jp/mobile/articles/2512/08/news032.html
• [ITmedia Mobile] あなたの街の「スマホ決済」キャンペーンまとめ【2025年12月版】~PayPay、d払い、au PAY、楽天ペイ
• [ITmedia ビジネスオンライン] 26年卒の「内定承諾の決め手」 昨年2位の「社員や社風」は5位に、上がった項目は?
• [ITmedia ビジネスオンライン] M&Aの売却理由 「後継者不足」が半減、どんな理由が増えた?
• [ITmedia ビジネスオンライン] 厳しい競争と人口減に苦しんだ「ひらパー」が、なぜ再び100万人台を達成できたのか
• [ITmedia ビジネスオンライン] 「ウソでしょ?」過去最高益なのに冬のボーナス減…… 知らないと損する、評価制度の3つの「共通点」
• [ITmedia エンタープライズ] Accentureが“ChatGPT精通集団”に コンサル業務含め幅広く活用
• [ITmedia ビジネスオンライン] Suicaや駅ビルの購買データがあるのになぜ? JR東がLINEヤフーと連携するワケ
• [ITmedia エンタープライズ] Nintendo Switchのネットサービスさえ止めた「AWS障害」 CIOが得るべき教訓は?
• [ITmedia News] こっちはEnter、こっちはCtrl+Enterで送信? 漫画「ITお嬢様の当惑 送信コマンド編」(1/3)
• [ITmedia News] EUがXにDSA違反で約217億円の制裁金、イーロン・マスク氏は「bullshit」と反発
• [ITmedia ビジネスオンライン] 忘年会の繁忙期でも「回る店」は何が違う? 串カツ田中の人材育成を変えたテクノロジー活用
• [ITmedia Mobile] 薄型スマホは中国で不人気? 「iPhone Air」が売れていない理由
• [ITmedia エンタープライズ] NEC 業務ノウハウを自動抽出、資産化するAIエージェントを提供開始 属人業務の行方は
• [ITmedia ビジネスオンライン] 「1年以内に自律型AIが自社業務を進化させる」69% 先進5カ国の経営幹部に調査
なお、Step 9-3 の cron 式を 0/10 分, * 時間, * 日付, * 月, ? 曜日, * 年 で設定すると 10 分に一度ポーリングする形になるので料金を節約できるので必要に応じて変更頂ければと思いますが、その場合、schdule できる時間は 10 の倍数の時間だけになります。10 の倍数ではない時間でも schedule 自体は可能ですが、時間が一致しないので動作しません。
同様に cron 式を 0 分, * 時間, * 日付, * 月, ? 曜日, * 年 で設定すると1時間に一度ポーリングする形になるのでさらに料金を節約できるので必要に応じて変更頂ければと思いますが、その場合、schdule できる時間は1時間毎の時間だけになります。
こちらの手順は少し面倒だったと思いますが、bot に対して命令を送ってコードを実行するような様々な bot アプリ作成に応用できると思うので覚えておくと良いと思います。
RSS URL
RSS の URL は以下などにまとめられています。
- Cisco: The Newsroom RSS Feeds
- Cisco Security RSS Feeds
- BeRSS
- (随時更新)AI系のニュースのRSSリスト
- AI, LLM, VR/ARの情報を得るためのRSSリスト(たまに更新)
以下なども人気です。
その他
Cisco 社内向けには社内インフラを利用して別途作成したものがあるので、社内の検索ページで「RSS Reader for Webex」で検索して見てください。社内向けに作成した Bot の使い方について説明したページがあります。
免責事項
本サイトおよび対応するコメントにおいて表明される意見は、投稿者本人の個人的意見であり、シスコの意見ではありません。本サイトの内容は、情報の提供のみを目的として掲載されており、シスコや他の関係者による推奨や表明を目的としたものではありません。各利用者は、本Webサイトへの掲載により、投稿、リンクその他の方法でアップロードした全ての情報の内容に対して全責任を負い、本Web サイトの利用に関するあらゆる責任からシスコを免責することに同意したものとします。