sakuraisan.py

# -*- coding: utf-8 -*-

import random, requests, json, sys, time
import urllib.request
import mysql.connector as mydb
import pandas as pd

from bs4 import BeautifulSoup

ARTIST_ID = 1 #Identifiant d'artiste enregistré à l'avance dans DB
AGENT_NAME = 'Sakurai' #Le nom de l'agent qui répond

#Classe API COTOHA
class CotohaApi():
    def __init__(self):
        self.COTOHA_ACCESS_INFO = {
            "grantType": "client_credentials",
            "clientId": "<Votre identifiant client>",
            "clientSecret": "<Votre propre secret client>"
        }
        self.ACCESS_TOKEN_PUBLISH_URL = '<URL de publication de votre jeton d'accès>'
        self.BASE_URL = '<Votre URL de base API'

        self.ACCESS_TOKEN = self.get_access_token()

    #Obtenez un jeton d'accès
    def get_access_token(self):
        headers = {
            "Content-Type": "application/json;charset=UTF-8"
        }
        access_data = json.dumps(self.COTOHA_ACCESS_INFO).encode()
        request_data = urllib.request.Request(self.ACCESS_TOKEN_PUBLISH_URL, access_data, headers)
        token_body = urllib.request.urlopen(request_data)
        token_body = json.loads(token_body.read())
        self.access_token = token_body["access_token"]
        self.headers = {
            'Content-Type': 'application/json;charset=UTF-8',
            'Authorization': 'Bearer {}'.format(self.access_token)
        }

    #Implémentation de l'API d'analyse des émotions et retour des résultats d'analyse
    def sentiment_analysis(self, text):
        request_body = {
            'sentence': text
        }
        url = self.BASE_URL + 'nlp/v1/sentiment'
        text_data = json.dumps(request_body).encode()
        request_data = urllib.request.Request(url, text_data, headers=self.headers, method='POST')
        sentiment_result = urllib.request.urlopen(request_data)
        sentiment_result = json.loads(sentiment_result.read())
        return sentiment_result

    # Positive:1, Negative:-1, Neutral:Convertir en 0
    def convert_sentiment(self, sentiment_in_word):
        if sentiment_in_word == 'Positive':
            return 1
        elif sentiment_in_word == 'Neutral':
            return 0
        elif sentiment_in_word == 'Negative':
            return -1

#Classe d'opération DB
class DBHandler():
    def __init__(self):
        self.conn = mydb.connect(
            host = '<Nom d'hôte de la base de données>',
            port = '<Numéro de port DB>',
            user = '<Nom d'utilisateur DB>',
            password = '<Mot de passe DB>',
            database = '<Nom de la base de données>',
            charset='utf8'
        )

        self.conn.ping(reconnect=True)
        self.cur = self.conn.cursor()

#Classe de construction de données
class Learn():
    def __init__(self):
        self.FILE_NAME = 'list.csv'
        self.ARTIST_NUMBER = '684' #Uta Net Artist Non.（Mr.Enfants 684)
        self.MAX_PAGE = 2 #Nombre de pages de liste de chansons des artistes du réseau de chansons (Mr.Les enfants ont 2 pages)

    #Recueillir les paroles de Song Net
    def gather_lyric(self):
        #Créer une table pour contenir les données récupérées
        list_df = pd.DataFrame(columns=['Titre de la chanson', 'Paroles'])

        for page in range(1, self.MAX_PAGE + 1):
            #Adresse du haut de la page de la chanson
            base_url = 'https://www.uta-net.com'

            #Page de la liste des paroles
            url = 'https://www.uta-net.com/artist/' + self.ARTIST_NUMBER + '/0/' + str(page) + '/'
            response = requests.get(url)
            soup = BeautifulSoup(response.text, 'lxml')
            links = soup.find_all('td', class_='side td1')

            for link in links:
                a = base_url + (link.a.get('href'))

                #Page de détail des paroles
                response = requests.get(a)
                soup = BeautifulSoup(response.text, 'lxml')
                title = soup.find('h2').text
                print(title)
                song_lyrics = soup.find('div', itemprop='text')
                
                for lyric in song_lyrics.find_all("br"):
                    lyric.replace_with('\n')
                song_lyric = song_lyrics.text

                #Attendez 1 seconde pour ne pas charger le serveur
                time.sleep(1)

                #Ajouter les paroles acquises au tableau
                tmp_se = pd.DataFrame([title, song_lyric], index=list_df.columns).T
                list_df = list_df.append(tmp_se)

        #csv enregistrer
        list_df.to_csv(self.FILE_NAME, mode = 'a', encoding='utf8')

    #Divisez les paroles en phrases et enregistrez les données, y compris les résultats de l'analyse des émotions dans la base de données
    def add_lyric(self):
        db = DBHandler()
        df_file = pd.read_csv(self.FILE_NAME, encoding='utf8')
        song_titles = df_file['Titre de la chanson'].tolist()
        song_lyrics = df_file['Paroles'].tolist()
        
        #Remarque: s'il y a beaucoup de chansons, la limite supérieure d'API pouvant être exécutée dans COTOHA par jour sera atteinte (environ 100 chansons par jour sont un guide).
        for i in range(len(song_titles)):

            #Ajouter un titre
            title = song_titles[i]

            print("Info: Saving {}...".format(title), end="")
            db.cur.execute(
                """
                insert into title (title, artist_id)
                values (%s, %s);
                """,
                (title, ARTIST_ID)
            )
            db.conn.commit()
            db.cur.execute(
                """
                select title_id from title
                where title= %s
                and artist_id = %s;
                """,
                (title, ARTIST_ID)
            )
            title_id = db.cur.fetchall()[-1][0]

            #Enregistrer le résultat de l'analyse des émotions de la phrase des paroles
            #Délimiteur de phrase lorsque deux sauts de ligne apparaissent
            lyric = song_lyrics[i]
            lyric_phrases = lyric.split('\n\n')
            lyric_phrases = [lyric.replace('\u3000', ' ').replace('\n', ' ') for lyric in lyric_phrases]
            
            #Utilisez l'API d'analyse des émotions pour chaque phrase et enregistrez les résultats de l'analyse des émotions dans la base de données
            cotoha_api= CotohaApi()
            for phrase in lyric_phrases:
                sentiment_result = cotoha_api.sentiment_analysis(phrase)['result']
                sentiment = cotoha_api.convert_sentiment(sentiment_result['sentiment'])
                score = sentiment_result['score']
                
                db.cur.execute(
                    """
                    insert into lyric (title_id, score, sentiment, phrase)
                    values (%s, %s, %s, %s);
                    """,
                    (title_id, score, sentiment, phrase)
                )
                db.conn.commit()

            print("Done")
                
        db.conn.close()
        if db.conn.is_connected() == False:
            print("Info: DB Disonnected")

    def execute(self):
        print("Info:Collectionner les paroles...")
        self.gather_lyric()
        print("Info:Ajout de paroles à DB...")
        self.add_lyric()

#Classe de recherche d'expression
class Search():
    def __init__(self):
        self.SEARCH_SCOPE = [0.01, 0.1, 0.3] #Largeur du score à rechercher SCORE ± SEARCH_Rechercher dans l'ordre de la liste dans la plage de SCOPE

    def execute(self):
        print("tu:", end="")
        input_data = input()
        print("{}：".format(AGENT_NAME), end="")
        
        cotoha_api= CotohaApi()
        sentiment_result = cotoha_api.sentiment_analysis(input_data)['result']
        sentiment = cotoha_api.convert_sentiment(sentiment_result['sentiment'])
        score = sentiment_result['score']
        
        db = DBHandler()

        find_flag = 0
        #Recherchez des expressions avec des scores similaires tout en élargissant progressivement la plage de recherche
        for scope in self.SEARCH_SCOPE:

            #Confirmez qu'il y en a au moins un
            db.cur.execute(
                """
                select count(phrase_id) from lyric
                join title on lyric.title_id = title.title_id
                where sentiment = %s
                and score between %s and %s
                and artist_id = %s;
                """,
                (sentiment, score-scope, score+scope, ARTIST_ID)
            )
            hit_num = db.cur.fetchall()[-1][0]
            if hit_num > 0:
                find_flag = 1
                break
        
        #S'il y a même un résultat de recherche, obtenez le résultat de la recherche et répondez au client
        if find_flag == 1:
            db.cur.execute(
                """
                select phrase,title from lyric
                join title on lyric.title_id = title.title_id
                where sentiment = %s
                and score between %s and %s
                and artist_id = %s;
                """,
                (sentiment, score-scope, score+scope, ARTIST_ID)
            )
            search_result = db.cur.fetchall()
            phrase_chosen = random.choice(search_result)
            print("{} [{}]".format(phrase_chosen[0], phrase_chosen[1]))
        else:
            print("Je n'ai pas trouvé de bonnes paroles.")
        
        db.conn.close()
        

if __name__ == "__main__":
    args = sys.argv
    if len(args) == 2:
        process = args[1] #Argument de ligne de commande apprendre:Enregistrer les informations sur les paroles dans DB, rechercher:Extraire des phrases émotionnelles similaires de DB
        if process == 'search':
            searcher = Search()
            searcher.execute()
        elif process == 'learn':
            learner = Learn()
            learner.execute()
        else:
            print("Error:Spécifiez un argument de ligne de commande[learn/search]")
    else:
        print("Error:Spécifiez un argument de ligne de commande[learn/search]")

Il existe deux manières de l'exécuter.

• Lors de la création de données

python sakuraisan.py learn

• Lors de la recherche de phrases

python sakuraisan.py search

Recommended Posts