Go to Qiita Advent Calendar 2024 Top
LoginSignup
4
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 3 years have passed since last update.

@fctokyo1016

【完成編】discord.pyでニコニコ生放送の放送開始通知をDiscordに投稿するbot

Posted at

初めに

この記事は構想編の続きなので併せてそちらをご覧いただくとわかりやすいです
【構想編】discord.pyでニコニコ生放送の放送開始通知をDiscordに投稿するbot

解決すべき課題

  • PostgreSQLでDBを作成してそこにデータを持つ必要がある
  • 設定値もDBで持てばいちいちコード書き直してHeroku再起動とかする必要無くない？
  • コミュニティ放送だけじゃなくチャンネル放送も対応したい

DBのテーブル構成

settingテーブル

  • token：DiscordBotのトークン
  • channel_id：Botが通知を投稿するチャンネルのID
token channel_id
XXXXXXXXXXXXXXXXXXXXXXXXXXXXX 0000000000000000

targetテーブル

  • community：監視対象にしたいコミュニティ番号
  • comment：メモ用（システム上は不使用）
id community comment
1 co1520016 BotTV
2 co117683 myCommunity

logsテーブル

  • live：通知した放送ID
id live
1 lv0000000
2 lv2222222

コード

discordbot.py
import requests
import os.path
import re
import discord
import asyncio
import os, psycopg2
import json
from bs4 import BeautifulSoup

path = "PATH"
port = "5432"
dbname = "DB_NAME"
user = "USER"
password = "PASSWORD"
conText = "host={} port={} dbname={} user={} password={}"
conText = conText.format(path,port,dbname,user,password)
connection = psycopg2.connect(conText)

cur = connection.cursor()
sql = "select token,channel_id from settings"
cur.execute(sql)
result = cur.fetchone()

# トークン取得
TOKEN = result[0]
# チャンネルID取得
CHANNEL_ID = result[1]

# targetテーブルから確認したいコミュニティを取得
def getTarget():
    targetCommunitys = []
    sql = "select community from target"
    cur.execute(sql)
    for row in cur:
        targetCommunitys.append(row[0])
    return targetCommunitys

# 放送URLから放送ID(lvXXXXXXX)抽出
def liveIdExtraction(liveURL):
    repatter = re.compile('lv[0-9]+')
    return repatter.search(liveURL).group()

# 放送URLから放送タイトル取得
def getLiveTitle(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    for meta_tag in soup.find_all('meta', attrs={'property': 'og:title'}):
        return meta_tag.get('content')

# 放送URLから放送者名取得
def getLiveName(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    return soup.find("span",{"class":"name"}).text

# logsテーブル内検索 あればTrue 無ければFalseを返す
def searchList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "SELECT count(*)  FROM logs WHERE live = '" + liveLV + "'"
    cur.execute(sql)
    result = cur.fetchone()
    if int(result[0]) > 0:
        return True
    else:
        return False

# logsテーブルに放送ID追記
def addList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "insert into logs(live) values('"+ liveLV + "');"
    cur.execute(sql)
    connection.commit()

# 接続に必要なオブジェクトを生成
client = discord.Client()

# 起動時に動作する処理
@client.event
async def on_ready():

    while(True):
        # ターゲットコミュニティの数だけ繰り返す
        targetCommunitys = getTarget()
        for targetCommunity in targetCommunitys:
            # URLを設定
            r = requests.get("https://live.nicovideo.jp/watch/" + targetCommunity)

            # コミュニティTOPページを確認
            soup = BeautifulSoup(r.content, "html.parser")
            result = soup.find('meta', attrs={'property': 'og:url', 'content': True})
            # 放送URL取得
            liveURL = result['content']

            # リスト内を検索してすでに処理済みの放送IDであれば処理しない
            if searchList(liveURL) is False:
                # 放送タイトル取得
                liveTitle = getLiveTitle(liveURL)
                # 放送者名取得
                liveName = getLiveName(liveURL)

                # Discordへ送信
                channel = client.get_channel(int(CHANNEL_ID))
                await channel.send(liveName + 'さんが配信を開始しました\n\n' + liveTitle + '\n' + liveURL)

                # 放送ID追記
                addList(liveURL)

        # チャンネル検索
        url = 'https://api.search.nicovideo.jp/api/v2/live/contents/search'
        ua = 'Twitter @fctokyo1016'
        headers = {'User-Agent': ua}
        params = {
            'q': 'BoxTV',
            'targets': 'title,description,tags',
            '_sort': '-openTime',
            '_context': ua,
            'fields': 'contentId,channelId,liveStatus,title',
            'filters[channelId][0]': '2640777',
            'filters[liveStatus][0]': 'onair'
        }
        # リクエスト
        res = requests.get(url, headers=headers, params=params)
        # 取得したjsonをlists変数に格納
        lists = json.loads(res.text)

        if lists['meta']['totalCount'] > 0:
            for data in lists['data']:
                if searchList(data['contentId']) is False:
                    # Discordへ送信
                    channel = client.get_channel(int(CHANNEL_ID))
                    await channel.send('チャンネルで配信を開始しました\n\n' + data['title'] + '\nhttps://nico.ms/' + data['contentId'])

                    # 放送ID追記
                    addList(data['contentId'])

        # 1分待つ
        await asyncio.sleep(60)

# Discordに接続
client.run(TOKEN)

やってること

  • DBから対象コミュニティほか、諸々の設定を取得
  • ニコニコ生放送のコミュニティページをスクレイピングして放送URLを取得
  • ついでにチャンネル生放送のAPIも叩いて放送開始していないか検索する
  • Discordに通知
  • 通知した放送は再通知しないようにDBに残す

DBに接続するための諸々の設定値です

discordbot.py
path = "PATH"
port = "5432"
dbname = "DB_NAME"
user = "USER"
password = "PASSWORD"
conText = "host={} port={} dbname={} user={} password={}"
conText = conText.format(path,port,dbname,user,password)
connection = psycopg2.connect(conText)

settingsテーブルからトークンとDiscordのチャンネルIDを取得します

discordbot.py
cur = connection.cursor()
sql = "select token,channel_id from settings"
cur.execute(sql)
result = cur.fetchone()

# トークン取得
TOKEN = result[0]
# チャンネルID取得
CHANNEL_ID = result[1]

targetテーブルから確認したいコミュニティを取得して配列にして返します

discordbot.py
# targetテーブルから確認したいコミュニティを取得
def getTarget():
    targetCommunitys = []
    sql = "select community from target"
    cur.execute(sql)
    for row in cur:
        targetCommunitys.append(row[0])
    return targetCommunitys

放送URLからスクレイピングで各種データを取得しています

discordbot.py
# 放送URLから放送ID(lvXXXXXXX)抽出
def liveIdExtraction(liveURL):
    repatter = re.compile('lv[0-9]+')
    return repatter.search(liveURL).group()

# 放送URLから放送タイトル取得
def getLiveTitle(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    for meta_tag in soup.find_all('meta', attrs={'property': 'og:title'}):
        return meta_tag.get('content')

# 放送URLから放送者名取得
def getLiveName(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    return soup.find("span",{"class":"name"}).text

既にDiscordに通知済みかどうかlogsテーブルを確認します

discordbot.py
# logsテーブル内検索 あればTrue 無ければFalseを返す
def searchList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "SELECT count(*)  FROM logs WHERE live = '" + liveLV + "'"
    cur.execute(sql)
    result = cur.fetchone()
    if int(result[0]) > 0:
        return True
    else:
        return False

通知したことをlogsテーブルに記録します

discordbot.py
# logsテーブルに放送ID追記
def addList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "insert into logs(live) values('"+ liveLV + "');"
    cur.execute(sql)
    connection.commit()

構想編ではなかったチャンネル生放送の確認です。
ニコニコ生放送のAPIを利用しています
niconico コンテンツ検索APIガイド

discordbot.py
        # チャンネル検索
        url = 'https://api.search.nicovideo.jp/api/v2/live/contents/search'
        ua = 'Twitter @fctokyo1016'
        headers = {'User-Agent': ua}
        params = {
            'q': 'BoxTV',
            'targets': 'title,description,tags',
            '_sort': '-openTime',
            '_context': ua,
            'fields': 'contentId,channelId,liveStatus,title',
            'filters[channelId][0]': '2640777',
            'filters[liveStatus][0]': 'onair'
        }
        # リクエスト
        res = requests.get(url, headers=headers, params=params)
        # 取得したjsonをlists変数に格納
        lists = json.loads(res.text)

        if lists['meta']['totalCount'] > 0:
            for data in lists['data']:
                if searchList(data['contentId']) is False:
                    # Discordへ送信
                    channel = client.get_channel(int(CHANNEL_ID))
                    await channel.send('チャンネルで配信を開始しました\n\n' + data['title'] + '\nhttps://nico.ms/' + data['contentId'])

                    # 放送ID追記
                    addList(data['contentId'])
4
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
4
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?