nicolive discord webhook notification

#nicoliveDiscordNotifiHooker#

Dockerでニコ生のユーザーページから最新の配信枠を取得してDiscordのWebhookで通知します。

config.ini

[config]
NotificationUser = 0
DISCORD_WEBHOOK_URL = https://discord.com/api/webhooks/
API_URL = https://live.nicovideo.jp/front/api/v1/user-broadcast-history?providerId={NotificationUser}&providerType=user&limit=1
CHECK_INTERVAL = 60
MentionRole = 0

docker-compose.yml

ersion: '3'

services:
  app:
    build: .
    container_name: nicoliveNotifiHooker
    restart: always
    volumes:
      - .:/app

Dockerfile

FROM python:3.10-slim
LABEL maintainer="Winding"

WORKDIR /app
COPY . ./app

COPY requirements.txt requirements.txt
RUN pip install --upgrade pip && pip install --no-cache-dir -r requirements.txt

CMD ["python", "nicoliveDiscordNotifiHooker.py"]

nicoliveDiscordNotifiHooker.py

import requests
import json
import time
from bs4 import BeautifulSoup
import configparser

def load_config(filepath):
    config = configparser.ConfigParser()
    config.read(filepath)
    return config['config']

def send_discord_notificationemb(message,embTitle,embDesc,embUrl,embThumb):
    payload = {
        'content': f"<@&{MentionRoleID}> {message}",
        "embeds": [
            {
                "title": embTitle,
                "url": embUrl,
                "description": embDesc,
                "color": 13843226,  # Optional: color in decimal (you can use any color)
                "image": {
                    "url": embThumb
                }
            }
        ]
    }
    try:
        response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
        response.raise_for_status()
        print('Message Sent Successfully.')
    except requests.RequestException as e:
        print(f"Failed to send notification: {e}")

def send_discord_notification(message):
    payload = {
        'content': message
    }
    try:
        response = requests.post(DISCORD_WEBHOOK_URL, json=payload)
        response.raise_for_status()
        print('Message Sent Successfully.')
    except requests.RequestException as e:
        print(f"Failed to send notification: {e}")

def desc_tagconvert(text):
    soup = BeautifulSoup(text, 'html.parser') # BeautifulSoupオブジェクトを作成
    for br_tag in soup.find_all('br'): # <br> タグを改行に変換
        br_tag.replace_with('\n')
    return soup.get_text() # 残りのHTMLタグを削除してテキストを取得


def check_program_status():
    global PROGRAMID
    global POSTSTATUS
    try:
        response = requests.get(API_URL)
        response.raise_for_status()
        data = response.json()

        status = data.get('data', {}).get('programsList', [{}])[0].get('program', {}).get('schedule', {}).get('status', '')
        lvid = data.get('data', {}).get('programsList', [{}])[0].get('id', {}).get('value', '')
        title = data.get('data', {}).get('programsList', [{}])[0].get('program', {}).get('title', '')
        #description = data.get('data', {}).get('programsList', [{}])[0].get('program', {}).get('description', '')
        description = desc_tagconvert( data.get('data', {}).get('programsList', [{}])[0].get('program', {}).get('description', '') )
        url = "https://live.nicovideo.jp/watch/" + data.get('data', {}).get('programsList', [{}])[0].get('id', {}).get('value', '')
        thumbnail = data.get('data', {}).get('programsList', [{}])[0].get('thumbnail', {}).get('screenshot', {}).get('large', '')

        if status == 'ON_AIR':
            if PROGRAMID != lvid: #lvが違う、別の枠であることと投稿をしたことがあるか
                send_discord_notificationemb(f"__A program is currently ON AIR！__\r## {title}\r## {url}",title,description,url,thumbnail)
                PROGRAMID = lvid #lvを記録
                POSTSTATUS = False
            else:
                PROGRAMID = lvid #lvを記録
        elif status == 'ENDED':
            if PROGRAMID == lvid and POSTSTATUS == False: #lvが同じ*一度配信開始投稿を済ませていること、かつ一度も投稿していない。
                send_discord_notification(f"__A program is currently EnD！__\r## {title}\r## {url}")
                PROGRAMID = lvid #lvを記録
                POSTSTATUS = True
            else:
                PROGRAMID = lvid #lvを記録
    except requests.RequestException as e:
        print(f"Error fetching or processing data: {e}")

if __name__ == "__main__":
    config = load_config('config.ini')
    NotificationUser = config.get('NotificationUser')
    DISCORD_WEBHOOK_URL = config.get('DISCORD_WEBHOOK_URL')
    API_URL_BASE = config.get('API_URL')
    CHECK_INTERVAL = config.getint('CHECK_INTERVAL')
    MentionRoleID = config.getint('MentionRole')
    PROGRAMID = ""
    POSTSTATUS = False

    API_URL = API_URL_BASE.replace('{NotificationUser}', NotificationUser)

    print("\n--------------------------------------------------------")
    print("　　Settings")
    print("--------------------------------------------------------")
    print(f"NotificationUser: {NotificationUser}")
    print(f"MentionRoleID: {MentionRoleID}")
    print(f"DISCORD_WEBHOOK_URL: {DISCORD_WEBHOOK_URL}")
    print(f"API_URL: {API_URL}")
    print(f"CHECK_INTERVAL: {CHECK_INTERVAL}")
    print("--------------------------------------------------------")

    print("\n:-: nicolive discord notification running...:-: \n")
    send_discord_notification(':-: 常駐監視を開始します... :-:')
    while True:
        check_program_status()
        time.sleep(CHECK_INTERVAL)

requirements.txt

beautifulsoup4
requests