[PYTHON] [Terminé] bot qui publie une notification du début de la diffusion en direct de Nico Nico sur Discord sur discord.py

en premier

Cet article est une suite de l'édition du concept, il est donc facile à comprendre si vous le lisez également. [Concept] bot qui publie la notification de démarrage de la diffusion en direct de Nico Nico sur Discord sur discord.py

Problèmes à résoudre

• Vous devez créer une base de données avec PostgreSQL et y avoir des données
• Si vous avez les valeurs de réglage dans la base de données, n'est-il pas nécessaire de réécrire le code et de redémarrer Heroku? ――Je veux soutenir non seulement la diffusion communautaire mais aussi la diffusion de chaînes

Structure de la table DB

table de réglage

--token: jeton DiscordBot --channel_id: ID du canal sur lequel le Bot publie la notification

table cible

--community: le numéro de communauté que vous souhaitez surveiller --comment: pour mémo (non utilisé sur le système)

table des journaux

--live: ID de diffusion notifié

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

#Obtenez un jeton
TOKEN = result[0]
#Obtenir l'identifiant de la chaîne
CHANNEL_ID = result[1]

#Obtenez la communauté que vous souhaitez vérifier à partir de la table cible
def getTarget():
    targetCommunitys = []
    sql = "select community from target"
    cur.execute(sql)
    for row in cur:
        targetCommunitys.append(row[0])
    return targetCommunitys

#Identifiant de diffusion de l'URL de diffusion(lvXXXXXXX)Extraction
def liveIdExtraction(liveURL):
    repatter = re.compile('lv[0-9]+')
    return repatter.search(liveURL).group()

#Obtenir le titre de diffusion à partir de l'URL de diffusion
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')

#Obtenez le nom du diffuseur à partir de l'URL de diffusion
def getLiveName(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    return soup.find("span",{"class":"name"}).text

#logs Renvoie False s'il n'y a pas de recherche dans la table True
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

#Ajouter un ID de diffusion à la table des journaux
def addList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "insert into logs(live) values('"+ liveLV + "');"
    cur.execute(sql)
    connection.commit()

#Générer les objets nécessaires à la connexion
client = discord.Client()

#Traitement qui fonctionne au démarrage
@client.event
async def on_ready():

    while(True):
        #Répétez pour le nombre de communautés cibles
        targetCommunitys = getTarget()
        for targetCommunity in targetCommunitys:
            #Définir l'URL
            r = requests.get("https://live.nicovideo.jp/watch/" + targetCommunity)

            #Consultez la page TOP de la communauté
            soup = BeautifulSoup(r.content, "html.parser")
            result = soup.find('meta', attrs={'property': 'og:url', 'content': True})
            #Obtenir l'URL de diffusion
            liveURL = result['content']

            #Rechercher dans la liste et ne pas traiter si l'ID de diffusion a déjà été traité
            if searchList(liveURL) is False:
                #Obtenir le titre de la diffusion
                liveTitle = getLiveTitle(liveURL)
                #Obtenez le nom du diffuseur
                liveName = getLiveName(liveURL)

                #Envoyer à Discord
                channel = client.get_channel(int(CHANNEL_ID))
                await channel.send(liveName + 'A commencé la livraison\n\n' + liveTitle + '\n' + liveURL)

                #ID de diffusion postscript
                addList(liveURL)

        #Recherche de chaînes
        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'
        }
        #demande
        res = requests.get(url, headers=headers, params=params)
        #Stockez le json acquis dans la variable lists
        lists = json.loads(res.text)

        if lists['meta']['totalCount'] > 0:
            for data in lists['data']:
                if searchList(data['contentId']) is False:
                    #Envoyer à Discord
                    channel = client.get_channel(int(CHANNEL_ID))
                    await channel.send('Distribution lancée sur la chaîne\n\n' + data['title'] + '\nhttps://nico.ms/' + data['contentId'])

                    #ID de diffusion postscript
                    addList(data['contentId'])

        #Attendez 1 minute
        await asyncio.sleep(60)

#Connectez-vous à Discord
client.run(TOKEN)

Que fais tu

• Obtenez divers paramètres tels que la communauté cible de DB
• Gratter la page de la communauté de Nico Nico Live Broadcast pour obtenir l'URL de diffusion
• De plus, appuyez sur l'API de la chaîne en direct pour rechercher si la diffusion a commencé. --Notifier Discord --Laisser la diffusion notifiée dans la base de données afin qu'elle ne soit plus notifiée

Différentes valeurs de réglage pour la connexion au 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)

Obtenez le jeton et l'ID du canal Discord dans le tableau des paramètres

discordbot.py

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

#Obtenez un jeton
TOKEN = result[0]
#Obtenir l'identifiant de la chaîne
CHANNEL_ID = result[1]

Récupérez la communauté que vous souhaitez vérifier à partir de la table cible et renvoyez-la sous forme de tableau

discordbot.py

#Obtenez la communauté que vous souhaitez vérifier à partir de la table cible
def getTarget():
    targetCommunitys = []
    sql = "select community from target"
    cur.execute(sql)
    for row in cur:
        targetCommunitys.append(row[0])
    return targetCommunitys

Diverses données sont acquises en grattant à partir de l'URL de diffusion

discordbot.py

#Identifiant de diffusion de l'URL de diffusion(lvXXXXXXX)Extraction
def liveIdExtraction(liveURL):
    repatter = re.compile('lv[0-9]+')
    return repatter.search(liveURL).group()

#Obtenir le titre de diffusion à partir de l'URL de diffusion
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')

#Obtenez le nom du diffuseur à partir de l'URL de diffusion
def getLiveName(liveURL):
    r = requests.get(liveURL)
    soup = BeautifulSoup(r.content, "html.parser")
    return soup.find("span",{"class":"name"}).text

Consultez le tableau des journaux pour voir si Discord a déjà été notifié

discordbot.py

#logs Renvoie False s'il n'y a pas de recherche dans la table True
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

Enregistrez la notification dans le tableau des journaux

discordbot.py

#Ajouter un ID de diffusion à la table des journaux
def addList(liveURL):
    liveLV = liveIdExtraction(liveURL)
    cur = connection.cursor()
    sql = "insert into logs(live) values('"+ liveLV + "');"
    cur.execute(sql)
    connection.commit()

C'est une confirmation de la diffusion en direct de la chaîne qui n'était pas l'édition conceptuelle. J'utilise l'API de Nico Nico Live Broadcasting Guide de l'API de recherche de contenu niconico

discordbot.py

        #Recherche de chaînes
        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'
        }
        #demande
        res = requests.get(url, headers=headers, params=params)
        #Stockez le json acquis dans la variable lists
        lists = json.loads(res.text)

        if lists['meta']['totalCount'] > 0:
            for data in lists['data']:
                if searchList(data['contentId']) is False:
                    #Envoyer à Discord
                    channel = client.get_channel(int(CHANNEL_ID))
                    await channel.send('Distribution lancée sur la chaîne\n\n' + data['title'] + '\nhttps://nico.ms/' + data['contentId'])

                    #ID de diffusion postscript
                    addList(data['contentId'])