anidiscover/anilist-api.py

import pprint
import numpy as np
import pandas as pd
from graphql_client import GraphQLClient

pp = pprint.PrettyPrinter(indent=2)

def myprint(data, show = True):
    if show:
        pp.pprint(data)

class AnilistApi:
    # Le client graphQL pour effectuer les requetes
    graphql = GraphQLClient('https://graphql.anilist.co')
    # La derniere query graphql
    query = None
    # Le derniere utilisateur
    user = None
    
    def findUser(self, username, returnSelf = True):
        """Cherche un utilisateur passé en paramètre pour les futures appels à l'API"""
        request = '''
            query ($name: String) {
              User(name: $name) {
                    id
                    mediaListOptions {
                        scoreFormat
                    }
              }
            }
        '''
        variables = {
            'name': username
        }
        # Création query
        self.query = self.graphql.query(request, variables)
        # Execution query
        result = self.query.execute().get()
        if(self.graphql.isError(result)):
            result = None
        self.user = result
        # Retour
        if returnSelf:
            return self
        return result != None
    
    def getUserAnimeList(self):
        """Retourne la liste d'anime complétés de l'utilisateur (ou None en cas de problème)"""
        if(self.user == None):
            return None
        request = '''
            query ($id: Int) {
              MediaListCollection (userId: $id, type: ANIME, status: COMPLETED) {
                lists {
                    entries {
                        score
                        media {
                          id
                          title {
                            userPreferred
                          }
                          averageScore
                          format
                          episodes
                          duration
                          popularity
                          source
                          tags {
                            name
                          }
                          genres
                        }
                    }
                }
            }
        }
        '''
        variables = {
            'id': self.user['User']['id']
        }
        # Création query
        self.query = self.graphql.query(request, variables)
        # Execution query
        result = self.query.execute().get()
        if(self.graphql.isError(result)):
            return None
        return UserAnimeList(result['MediaListCollection']['lists'][0]['entries'], self.user['User']['mediaListOptions']['scoreFormat'])
    
    def fetchAnimeList(self, query, page = 0, element = 50):
        if query.object != 'Media':
            raise ValueError('bad query type: ' + query.object)
        # Requete graphql
        request = '''
        query ($page: Int, $element: Int, ${def}) {
            Page(page: $page, perPage: $element) {
                media(${param}) {
                    id
                    title {
                        userPreferred
                    }
                    averageScore
                    format
                    episodes
                    duration
                    popularity
                    source
                    tags {
                        name
                    }
                    genres
                }
            }
        }
        '''
        # Recupération info query et application
        definition, parameter, variable = query.get()
        request = request.replace('${def}', definition)
        request = request.replace('${param}', parameter)
        variable['page'] = page
        variable['element'] = element
        # Création query
        self.query = self.graphql.query(request, variable)
        # Execution query
        result = self.query.execute().get()
        if(self.graphql.isError(result)):
            return None
        # Préparation pour creation objet AnimeList
        data = []
        for i in range(0, len(result['Page']['media'])):
            data.append({'media': result['Page']['media'][i]})
        return UserAnimeList(data)
    
    def iterateAnimeListFetch(self, query, nb, element):
        for i in range (0, nb):
            yield self.fetchAnimeList(query, i, element)
    
class AnilistQuery:
    __declaration = ''
    __parameter = ''
    __variable = {}
    
    # Constructeur
    def __init__(self, obj = 'Media'):
        self.object = obj
        self.reset()
        
    def scoreGreaterThan(self, value):
        if(self.__variable.get('sgt') == None):
            self.__declaration += ',$sgt: Int'
            self.__parameter += ',averageScore_greater: $sgt'
            self.__variable['sgt'] = round(value)
        return self
    
    def popularityGreaterThan(self, value):
        if(self.__variable.get('pgt') == None):
            self.__declaration += ',$pgt: Int'
            self.__parameter += ',popularity_greater: $pgt'
            self.__variable['pgt'] = round(value)
        return self
    
    def episodeBetween(self, minValue, maxValue):
        if(self.__variable.get('egt') == None):
            self.__declaration += ',$egt: Int'
            self.__parameter += ',episodes_greater: $egt'
            self.__variable['egt'] = round(minValue)
        if(self.__variable.get('elt') == None):
            self.__declaration += ',$elt: Int'
            self.__parameter += ',episodes_lesser: $elt'
            self.__variable['elt'] = round(maxValue)
        return self
    
    def durationBetween(self, minValue, maxValue):
        if(self.__variable.get('dgt') == None):
            self.__declaration += ',$dgt: Int'
            self.__parameter += ',duration_greater: $dgt'
            self.__variable['dgt'] = round(minValue)
        if(self.__variable.get('dlt') == None):
            self.__declaration += ',$dlt: Int'
            self.__parameter += ',duration_lesser: $dlt'
            self.__variable['dlt'] = round(maxValue)
        return self
    
    def formatIn(self, value):
        if(self.__variable.get('fin') == None):
            self.__declaration += ',$fin: [MediaFormat]'
            self.__parameter += ',format_in: $fin'
            self.__variable['fin'] = value
        return self
    
    def sourceIn(self, value):
        if(self.__variable.get('sin') == None):
            self.__declaration += ',$sin: [MediaSource]'
            self.__parameter += ',source_in: $sin'
            self.__variable['sin'] = value
        return self
    
    def tagIn(self, value):
        if(self.__variable.get('tin') == None):
            self.__declaration += ',$tin: [String]'
            self.__parameter += ',tag_in: $tin'
            self.__variable['tin'] = value
        return self
    
    def genreIn(self, value):
        if(self.__variable.get('gin') == None):
            self.__declaration += ',$gin: [String]'
            self.__parameter += ',genre_in: $gin'
            self.__variable['gin'] = value
        return self
    
    def get(self):
        return self.__declaration.strip(','), self.__parameter.strip(','), self.__variable
    
    def reset(self):
        self.__declaration = ''
        self.__parameter = ''
        self.__variable = {}
    
    def __str__(self):
        return 'Declaration: ' + self.__declaration.strip(',') + '\nParameter: ' + self.__parameter.strip(',') + '\nVariable: ' + pp.pformat(self.__variable)

class UserAnimeList:
    data = []
    
    # Constructeur, met en forme les données
    def __init__(self, entries, userScoreType = None):
        self.data = []
        for entry in entries:
            media = entry['media']
            d = {
                'id': media['id'],
                'title': media['title']['userPreferred'],
                'score': media['averageScore'],
                'popularity': media['popularity'],
                'format': media['format'],
                'episode': media['episodes'],
                'duration': media['duration'],
                'source': media['source']
            }
            # Ajout score de l'utilisateur si present
            if userScoreType != None and entry.get('score') != None:
                d['userScore'] = self.__formatScore(entry['score'], userScoreType)
            # Ajout tag (3 max)
            end = 3 if len(media['tags']) > 3 else len(media['tags'])
            for i in range(0, 3):
                if(i < end):
                    d['tag' + str(i + 1)] = media['tags'][i]['name']
                else:
                    d['tag' + str(i + 1)] = 'None'
            # Ajout genre (5 max)
            end = 5 if len(media['genres']) > 5 else len(media['genres'])
            for i in range(0, 5):
                if(i < end):
                    d['genre' + str(i + 1)] = media['genres'][i]
                else:
                    d['genre' + str(i + 1)] = 'None'
            # Ajoute la ligne aux données
            self.data.append(d)
            
    def toDataFrame(self, dummies = False):
        df = pd.DataFrame(self.data)
        if dummies and len(self.data) > 0:
            df = self.__multipleColDummies(df, 'tag', 3)
            df = self.__multipleColDummies(df, 'genre', 5)
            df = pd.get_dummies(df, columns=['format', 'source'])
        return df
        
    def __formatScore(self, score, scoreType):
        if scoreType == 'POINT_10_DECIMAL' or scoreType == 'POINT_10':
            return 10 * score
        elif scoreType == 'POINT_5':
            return 20 * score
        elif scoreType == 'POINT_3':
            return (score / 3) * 100
        else:
            return score
        
    def __multipleColDummies(self, df, colname, num):
        # Recherche de toutes les valeurs possibles
        values = []
        for i in range(1, num + 1):
            d = df[colname + str(i)]
            for item in d.iteritems():
                if not(item[1] in values and item[1] != 'None'):
                    values.append(item[1])
        # Creation colonne dans la dataframe
        for val in values:
            df[colname + '_' + val.replace(' ', '_')] = 0
        # Remplissage des colonnes
        newDf = []
        for row in df.iterrows():
            row = pd.Series(row[1])
            for i in range(1, num + 1):
                if row[colname + str(i)] != None:
                    name = row[colname + str(i)].replace(' ', '_')
                    row[colname + '_' + name] = 1
            newDf.append(row)
        df = pd.DataFrame(newDf)
        # Suppr ancienne colonne
        for i in range(1, num + 1):
            df = df.drop(colname + str(i), axis = 1)
        return df