YouTubeの新着通知をDiscordで受け取りたい！

はじめに

この記事は某鯖アドベントカレンダー2025のために執筆されたものです
※テック系の鯖ではありませんのでご注意ください
某鯖01, 某鯖02

何がしたい？

・特定チャンネルのYouTube通知をDiscordのメッセージで受け取りたい！
・今まで通知された動画はElasticsearchで管理したい！

概要

1. YouTubeに新着動画がアップされる
2. Elasticsearchに動画の情報を格納する
3. Discordにメッセージを送信する

やってみた

前提

もともと似たようなことを自分のメインPC (Windows) で行っていたのですが、Elasticsearchで管理していなかったこと、タスクスケジューラとかいうクソ使いにくいものを使う必要があったこと、そして何より、スリープ中だと機能しないことから、新たにLinuxサーバを建て、そこで管理することにしました

環境構築

1　適当なミニPCを買います

2　最初から入っていたWindows11 Proを吹き飛ばし、Rocky Linux10をインストールします
3　ここでなんか物理メモリが足りないことに気づきましたが、もう遅いので気にしないこととします
4　PythonやElasticsearch、Kibana等々をインストール&設定すると同時に、ローカルIPの固定もやっておきます (この辺はChat GPTと雑談しながら進めました)
5　4の手順と同時に必要なポートを開けます。今回は5601 (Kibana), 8000 (FastAPI), 9200 (Elasticsearch), 10050 (Zabbix)のポートを解放しました
6　実装します

実装

実装もChat GPTと雑談しながら進めました
現状のコード例は以下となります

config.conf

[elasticsearch]
host = 172.31.11.1
port = 9200
scheme = http
user = elastic
password = ***************

youtube.conf

[hasunosora]
name = 蓮ノ空女学院スクールアイドルクラブ
id = UCxUgvwrVfqVpyak4cuKcevQ
avatar = https://yt3.googleusercontent.com/r97NQTyqVlW_fTKpRFbOv89DorN4rue0u0sBGGVdRmHq-tjZSFHn-TR26VWLb469umCO5dNO=s72-c-k-c0x00ffffff-no-rj
url = https://canary.discord.com/api/webhooks/XXXXXXXXXXXX/XXXXXXXXXXXXXXXXXXXXXX_YYYYYYYYYYYYY_**********************************
role_id = 1448253**********7202

__main__.py

import argparse
import configparser
import json
import feedparser
import requests

from datetime import datetime
from elasticsearch import Elasticsearch


CONFIG_PATH = '/home/{user名}/Documents/develop/YouTube/config/config.conf'
YOUTUBE_CONFIG_PATH = '/home/{user名}/Documents/develop/YouTube/config/youtube.conf'
INDEX_NAME = 'youtube_videos'


# =========================================================
# Utility
# =========================================================
def load_config(path: str) -> configparser.ConfigParser:
    config = configparser.ConfigParser()
    config.read(path)
    return config


def parse_args():
    parser = argparse.ArgumentParser(description='YouTube RSS Webhook Script')
    parser.add_argument(
        '--channel', '-c', required=True,
        help='YouTube channel name (shigureui, hasunosora)'
    )
    return parser.parse_args()


# =========================================================
# Elasticsearch
# =========================================================
def init_es(config: configparser.ConfigParser) -> Elasticsearch:
    es_conf = config['elasticsearch']
    return Elasticsearch(
        hosts=[{
            "host": es_conf['host'],
            "port": int(es_conf['port']),
            "scheme": es_conf['scheme']
        }],
        basic_auth=(es_conf['user'], es_conf['password']),
    )


def save_video(es: Elasticsearch, video_id: str, doc: dict) -> bool:
    res = es.index(index=INDEX_NAME, id=video_id, document=doc)
    return res.get("result") == "created"


# =========================================================
# YouTube
# =========================================================
def load_youtube_config(channel_key: str) -> dict:
    yt_conf = load_config(YOUTUBE_CONFIG_PATH)
    
    if channel_key not in yt_conf:
        raise ValueError(f"Channel '{channel_key}' not found in youtube.conf")
    
    section = yt_conf[channel_key]
    
    return {
        "channel_name": section['name'],
        "channel_id": section['id'],
        "avatar": section['avatar'],
        "webhook_url": section['url'],
        "role_id": str(section['role_id'])
    }


def fetch_rss_entries(channel_id: str):
    rss_url = f'https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}'
    feed = feedparser.parse(rss_url)
    return feed.entries


def to_doc(entry, channel_id: str, channel_name: str) -> dict:
    published = (
        datetime(*entry.published_parsed[:6]).isoformat()
        if 'published_parsed' in entry else None
    )
    
    return {
        "channel_id": channel_id,
        "channel_name": channel_name,
        "video_id": entry.yt_videoid,
        "video_title": entry.title,
        "video_link": entry.link,
        "published_dt": published
    }


# =========================================================
# Discord
# =========================================================
def send_discord_notification(webhook_url: str, channel_name: str, avatar: str, role_id: str, doc: dict):
    body = {
        "content": f'<@&{role_id}>\n{doc["video_title"]}\n{doc["video_link"]}',
        "username": channel_name,
        "avatar_url": avatar,
        "allowed_mentions": {
            "parse": [],
            "roles": [role_id]
        }
    }
    
    requests.post(
        url=webhook_url,
        headers={"Content-Type": "application/json"},
        data=json.dumps(body)
    )
    
    print(f'[INFO] {json.dumps(body, ensure_ascii=False)}')


# =========================================================
# Main Logic
# =========================================================
def main():
    args = parse_args()
    channel_key = args.channel
    
    config = load_config(CONFIG_PATH)
    es = init_es(config)
    
    yt = load_youtube_config(channel_key)
    print(f'[INFO] name : {yt["channel_name"]}')
    
    # save videos to elasticsearch
    entries = fetch_rss_entries(yt['channel_id'])
    new_docs = []
    save_video_flg = False
    for entry in entries:
        doc = to_doc(entry, yt['channel_id'], yt['channel_name'])
        if save_video(es, doc['video_id'], doc):
            new_docs.append(doc)
            save_video_flg = True
            print(f'[INFO] {json.dumps(doc, ensure_ascii=False)}')   
    if not save_video_flg:
        print(f'[INFO] No New Videos')
    
    # Discord Webhooks
    send_message_flg = False
    for doc in new_docs:
        send_discord_notification(
            yt['webhook_url'], yt['channel_name'], yt['avatar'], yt['role_id'], doc
        )
        send_message_flg = True
    if not send_message_flg:
        print(f'[INFO] 0 messages sent to Discord.')


# =========================================================
# Entry Point
# =========================================================
if __name__ == '__main__':
    start = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f'[INFO] start: {start}')
    main()
    end = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    print(f'[INFO] end  : {end}')
    print('|-----------------------------------------------|')

きったねえコードとかパスワード平文で保存するなとか思ってもコメントに書かないでください
コンフィグは別にjsonでもよかったのですが、confファイルを使ったら開発してる感が出て気分が良かったので採用しました

Discordのwebhook URLは、Discord側でサーバ設定→連携サービス→ウェブフック→新しいウェブフックと進み、適当な名前を付けてメッセージ送信先のチャンネルを選択した後、ウェブフックURLをコピーで取得できます

手動で実行したい場合、下記コマンドで実行します
(下記実行結果は新着動画が無かった場合)

[{user名}@{ホスト名} ~]$ /usr/bin/python3 /home/{user名}/Documents/develop/YouTube/sourse/__main__.py -c hasunosora
[INFO] start: 2025-12-16 23:07:53
[INFO] name : 蓮ノ空女学院スクールアイドルクラブ
[INFO] No New Videos
[INFO] 0 messages sent to Discord.
[INFO] end  : 2025-12-16 23:07:53
|-----------------------------------------------|

私は毎分バッチ実行させたかったので、cron実行させることにしました
ログはlogrotateで管理します (毎日圧縮、30日保存)

crontab -e
* * * * * /usr/bin/python3 /home/{user名}/Documents/develop/YouTube/sourse/__main__.py -c hasunosora >> /var/log/YouTube/YouTube.log 2>&1

/var/log/YouTube/YouTube.log {
    daily
    rotate 30
    compress
    missingok
    notifempty
    create 644 {user名} {user名}
    copytruncate
}

解説

-c のオプションで指定したチャンネルの新着動画をDiscordに通知します

youtube.conf内の情報を使用し、YouTube RSSから対象チャンネルのfeedを取得、Pythonライブラリのfeedparserを利用し、feed.entriesをパースします
feedparserを使うことで、dictとしてPython内で扱うことが可能となります

feedの情報をもとに、下記をElasticsearchに登録します
・チャンネルID (youtube.confに記載)
・チャンネル名 (youtube.confに記載)
・動画ID (entry.yt_videoid)
・動画タイトル (entry.title)
・動画リンク (entry.link)
・公開日 (entry.published_parsed, isoフォーマット)

Kibanaで見るとこんな感じ

Elasticsearchのレスポンスを解析し、res["result"]が"create"となった動画が、DiscordにWebhooksを通じて通知されます
通知先はロールとなっているため、Discord上で通知が欲しい人、いらない人を分けることができます
実際に通知されるとこのようになります

さいごに

セキュリティとか諸々できる人ならぶっちゃけGoogle WebSub使ったほうが早いし楽だと思います
今回はその辺が面倒だったのと自宅サーバをインターネットに公開しないことを前提に設計したので、RSS feed取得→諸々の処理というかなり遠回りな方法をとることになりました

現在もいろいろ改良中ですが、一旦運用軌道に乗ったのでこの記事を公開しようと思った次第です

今後もこのツールは (個人的用途ですが) 改良されていく予定ですので、気が向いたらこの記事も更新されるか新しい記事が投稿されると思います