[PYTHON] [Abgeschlossen] Bot, der eine Benachrichtigung über den Start der Live-Übertragung von Nico Nico an Discord auf discord.py sendet

zunaechst

Dieser Artikel ist eine Fortsetzung der Concept Edition, daher ist es leicht zu verstehen, wenn Sie ihn auch lesen. [Konzept] Bot, der die Startbenachrichtigung von Nico Nico Live-Übertragung an Discord auf discord.py veröffentlicht

Zu lösende Probleme

• Sie müssen eine Datenbank mit PostgreSQL erstellen und dort Daten haben
• Wenn Sie die Einstellungswerte in der Datenbank haben, ist es nicht notwendig, den Code neu zu schreiben und Heroku neu zu starten? ――Ich möchte nicht nur Community Broadcasting, sondern auch Channel Broadcasting unterstützen

DB-Tabellenstruktur

Einstelltabelle

--token: DiscordBot-Token --channel_id: ID des Kanals, an den der Bot die Benachrichtigung sendet

Zieltabelle

--community: Die Community-Nummer, die Sie überwachen möchten --comment: Für Memo (wird im System nicht verwendet)

Protokolltabelle

--live: Broadcast-ID benachrichtigt

Code

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()

#Bekomme Token
TOKEN = result[0]
#Kanal-ID abrufen
CHANNEL_ID = result[1]

#Holen Sie sich die Community, die Sie überprüfen möchten, aus der Zieltabelle
def getTarget():
    targetCommunitys = []
    sql = "select community from target"
    cur.execute(sql)
    for row in cur:
        targetCommunitys.append(row[0])
    return targetCommunitys

#Broadcast-ID von der Broadcast-URL(lvXXXXXXX)Extraktion
def liveIdExtraction(liveURL):
    repatter = re.compile('lv[0-9]+')
    return repatter.search(liveURL).group()

#Ruft den Broadcast-Titel von der Broadcast-URL ab
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')

#Rufen Sie den Namen des Senders von der Broadcast-URL ab
def getLiveName(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    return soup.find("span",{"class":"name"}).text

#logs Gibt False zurück, wenn in der Tabelle True keine Suche vorhanden ist
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

#Fügen Sie der Protokolltabelle die Broadcast-ID hinzu
def addList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "insert into logs(live) values('"+ liveLV + "');"
    cur.execute(sql)
    connection.commit()

#Generieren Sie die für die Verbindung benötigten Objekte
client = discord.Client()

#Verarbeitung, die beim Start ausgeführt wird
@client.event
async def on_ready():

    while(True):
        #Wiederholen Sie diesen Vorgang für die Anzahl der Zielgemeinschaften
        targetCommunitys = getTarget()
        for targetCommunity in targetCommunitys:
            #URL festlegen
            r = requests.get("https://live.nicovideo.jp/watch/" + targetCommunity)

            #Überprüfen Sie die Community-TOP-Seite
            soup = BeautifulSoup(r.content, "html.parser")
            result = soup.find('meta', attrs={'property': 'og:url', 'content': True})
            #Broadcast-URL abrufen
            liveURL = result['content']

            #Durchsuchen Sie die Liste und verarbeiten Sie sie nicht, wenn die Broadcast-ID bereits verarbeitet wurde
            if searchList(liveURL) is False:
                #Holen Sie sich den Titel der Sendung
                liveTitle = getLiveTitle(liveURL)
                #Holen Sie sich den Namen des Senders
                liveName = getLiveName(liveURL)

                #An Zwietracht senden
                channel = client.get_channel(int(CHANNEL_ID))
                await channel.send(liveName + 'Hat mit der Lieferung begonnen\n\n' + liveTitle + '\n' + liveURL)

                #Broadcast ID Postscript
                addList(liveURL)

        #Kanalsuche
        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'
        }
        #Anfrage
        res = requests.get(url, headers=headers, params=params)
        #Speichern Sie den erfassten JSON in der Variablen Variablen
        lists = json.loads(res.text)

        if lists['meta']['totalCount'] > 0:
            for data in lists['data']:
                if searchList(data['contentId']) is False:
                    #An Zwietracht senden
                    channel = client.get_channel(int(CHANNEL_ID))
                    await channel.send('Verteilung auf dem Kanal gestartet\n\n' + data['title'] + '\nhttps://nico.ms/' + data['contentId'])

                    #Broadcast ID Postscript
                    addList(data['contentId'])

        #Warte 1 Minute
        await asyncio.sleep(60)

#Stellen Sie eine Verbindung zu Discord her
client.run(TOKEN)

Was machst du

• Beziehen Sie verschiedene Einstellungen wie die Zielgemeinschaft von DB --Scraping der Community-Seite von Nico Nico Live Broadcast, um die Broadcast-URL zu erhalten ――Drücken Sie außerdem die API der Live-Kanalübertragung, um zu suchen, ob die Übertragung gestartet wurde.
• Zwietracht melden
• Lassen Sie die benachrichtigte Sendung in der Datenbank, damit sie nicht erneut benachrichtigt wird

Verschiedene Einstellwerte für die Verbindung zur 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)

Rufen Sie das Token und die Discord-Kanal-ID aus der Einstellungstabelle ab

discordbot.py

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

#Bekomme Token
TOKEN = result[0]
#Kanal-ID abrufen
CHANNEL_ID = result[1]

Holen Sie sich die Community, die Sie überprüfen möchten, aus der Zieltabelle und geben Sie sie als Array zurück

discordbot.py

#Holen Sie sich die Community, die Sie überprüfen möchten, aus der Zieltabelle
def getTarget():
    targetCommunitys = []
    sql = "select community from target"
    cur.execute(sql)
    for row in cur:
        targetCommunitys.append(row[0])
    return targetCommunitys

Verschiedene Daten werden durch Scraping von der Broadcast-URL erfasst

discordbot.py

#Broadcast-ID von der Broadcast-URL(lvXXXXXXX)Extraktion
def liveIdExtraction(liveURL):
    repatter = re.compile('lv[0-9]+')
    return repatter.search(liveURL).group()

#Ruft den Broadcast-Titel von der Broadcast-URL ab
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')

#Rufen Sie den Namen des Senders von der Broadcast-URL ab
def getLiveName(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    return soup.find("span",{"class":"name"}).text

Überprüfen Sie die Protokolltabelle, um festzustellen, ob Discord bereits benachrichtigt wurde

discordbot.py

#logs Gibt False zurück, wenn in der Tabelle True keine Suche vorhanden ist
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

Notieren Sie die Benachrichtigung in der Protokolltabelle

discordbot.py

#Fügen Sie der Protokolltabelle die Broadcast-ID hinzu
def addList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "insert into logs(live) values('"+ liveLV + "');"
    cur.execute(sql)
    connection.commit()

Es ist eine Bestätigung der Live-Übertragung des Senders, die nicht die Concept Edition war. Ich verwende die API von Nico Nico Live Broadcasting API-Handbuch zur Inhaltssuche von niconico

discordbot.py

        #Kanalsuche
        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'
        }
        #Anfrage
        res = requests.get(url, headers=headers, params=params)
        #Speichern Sie den erfassten JSON in der Variablen Variablen
        lists = json.loads(res.text)

        if lists['meta']['totalCount'] > 0:
            for data in lists['data']:
                if searchList(data['contentId']) is False:
                    #An Zwietracht senden
                    channel = client.get_channel(int(CHANNEL_ID))
                    await channel.send('Verteilung auf dem Kanal gestartet\n\n' + data['title'] + '\nhttps://nico.ms/' + data['contentId'])

                    #Broadcast ID Postscript
                    addList(data['contentId'])