#12 – Conexão API IGDB

API IGDB (Internet Game Database)

Pelo fato da IGDB ser do mesmo grupo da Twitch, as credenciais de autorização para as consultas APIs são as mesmas da Twitch, porém o que muda é a forma de consulta.

Link Documentação IGDB API

Na Twitch, conforme detalhei no Post #6, é pelo método GET, porém para IGDB utiliza-se o método POST que é um pouco diferente pois faz-se o envio de dados no corpo da requisição.

Por exemplo para a requisição principal que retorna as informações gerais dos Jogos o link utilizado foi:

https://api.igdb.com/v4/games

Fiz o teste de conexão pelo Postman e é necessário preencher no corpo da requisição as informações de ID a ser consultado e os campos a serem retornados, conforme imagem abaixo.

A resposta em JSON dessa consulta seria:

[
    {
        "id": 241,
        "aggregated_rating": 70.0,
        "aggregated_rating_count": 2,
        "franchises": [
            425
        ],
        "genres": [
            5
        ],
        "involved_companies": [
            5830,
            5831,
            211860
        ],
        "name": "Counter-Strike",
        "rating": 83.20654117083305,
        "rating_count": 682,
        "release_dates": [
            377412,
            296958,
            296959
        ],
        "summary": "Play the world's number 1 online action game. Engage in an incredibly realistic brand of terrorist warfare in this wildly popular team-based game. Ally with teammates to complete strategic missions. Take out enemy sites. Rescue hostages. Your role affects your team's success. Your team's success affects your role.",
        "total_rating": 76.60327058541652,
        "total_rating_count": 684,
        "websites": [
            64118,
            547038,
            15348,
            42734,
            521342,
            648366,
            521343
        ]
    }
]

Então, é necessário salvar as informações de cada campo separadamente.

Para esse projeto, fiz um Database principal com o ID, nome, e as informações de nota (rating, rating_count, total_rating, total_rating_count). Os outros campos são um pouco mais complexos pois envolvem várias informações de um campo só.

#API IGDB - Database IGDB - Consulta informações IGDB

#Leitura do arquivo IGDB Database e do arquivo relacionado com ID Twitch
df_atualizado = pd.read_csv(twitch_database_igdb, sep=",", encoding='utf-8-sig')
df_igdb = pd.read_csv(igdb_database, sep=",", encoding='utf-8-sig')

# Remove colunas desnecessárias do df_atualizado
df_atualizado = df_atualizado.drop(columns=['ID_Twitch', 'NomedoJogo'])

#Para garantir a duplicata, agrupa os jogos por Id, depois remove coluna gerada
df_atualizado = df_atualizado.groupby(['ID_IGDB']).size().reset_index(name='Count')
df_atualizado = df_atualizado.drop(columns=['Count'])

# Adiciona colunas vazias para consulta depois
cols_to_add = ['name', 'rating', 'rating_count', 'aggregated_rating', 'aggregated_rating_count',
               'total_rating', 'total_rating_count']
df_atualizado.loc[:, cols_to_add] = None

# Converte IDs para string para evitar problemas nas chamadas das APIs
df_atualizado['ID_IGDB'] = df_atualizado['ID_IGDB'].astype(int).astype(str)
df_igdb['ID_IGDB'] = df_igdb['ID_IGDB'].astype(int).astype(str)

# Junta dados antigos + novos no df_combinado sem duplicatas
df_combinado = pd.concat([df_igdb, df_atualizado], ignore_index=True)
df_combinado = df_combinado.drop_duplicates(subset='ID_IGDB', keep='first')

#Faz a cópia para evitar perda de informações no df
df_combinado = df_combinado.copy()

# Garante os tipos corretos nas colunas após a junção
df_combinado['name'] = df_combinado['name'].astype('string')
df_combinado['rating'] = df_combinado['rating'].astype('float64')
df_combinado['rating_count'] = df_combinado['rating_count'].astype('Int64')
df_combinado['aggregated_rating'] = df_combinado['aggregated_rating'].astype('float64')
df_combinado['aggregated_rating_count'] = df_combinado['aggregated_rating_count'].astype('Int64')
df_combinado['total_rating'] = df_combinado['total_rating'].astype('float64')
df_combinado['total_rating_count'] = df_combinado['total_rating_count'].astype('Int64')

#----------------------------------------------------------------------------------#

# Defina a URL base da API
url_base = url_base_igdb

# Client-ID e Token de Autenticação na definição dos headers
headers = {
    'Client-ID': client_id,
    'Authorization': f'Bearer {access_token}'
}

#Dicionário vazio para salvar a resposta da API
cache_info = {}

#----------------------------------------------------------------------------------#

# Função para pegar as informações da API
def get_game_info(game_id):
    #Checa se o ID já tem imagem ou não (certeza do funcionamento para remoção de IDs duplicados, principalmente os consultados pela API), utilizando o dicionário
    if game_id in cache_info:
        return cache_info[game_id]

    #Definição do body com a ID
    body = f"""
    where id = {game_id};
    fields name, rating, rating_count, aggregated_rating, aggregated_rating_count, total_rating, total_rating_count;
    limit 1;
    """

    #Informa o usuário se está fazendo a chamada da API para o ID solicitado
    print(f"Chamando API para o ID: {game_id}")

    #Dependendo da resposta da API:
    try:
        #Salva resposta de status da API para conferência
        response = requests.post(url_base, headers=headers, data=body)

        #Checa qual status da resposta
        if response.status_code == 200:
            #200 deu certo, salva resposta JSON na variável game_data (dicionário)
            game_data = response.json()

            #Verificação da lista "data" no dicionário "game_data"
            if game_data:
                #Dicionário Game_info para coletar toda a resposta JSON
                game_info = game_data[0]
                game_info_dict = {
                    'name': game_info.get('name') if game_info.get('name') else 'NA',
                    'rating': game_info.get('rating'),
                    'rating_count': game_info.get('rating_count'),
                    'aggregated_rating': game_info.get('aggregated_rating'),
                    'aggregated_rating_count': game_info.get('aggregated_rating_count'),
                    'total_rating': game_info.get('total_rating'),
                    'total_rating_count': game_info.get('total_rating_count'),
                }
                #Salva no dicionário cache
                cache_info[game_id] = game_info_dict

                #Retorno da função
                return game_info_dict

            #Caso não tenha nada (resposta vazia) em "data"
            else:
                print(f"[Aviso] Resposta 200, mas sem dados para o ID {game_id}")
                game_info_dict = {
                    'name': 'NA',
                    'rating': None,  # Usando None para valores ausentes numéricos
                    'rating_count': None,
                    'aggregated_rating': None,
                    'aggregated_rating_count': None,
                    'total_rating': None,
                    'total_rating_count': None
                }
                cache_info[game_id] = game_info_dict
                return game_info_dict

        #Falha no acesso da API (401 - Token OAuth não válido)
        #Sai do Programa sem sobreescrever no database
        elif response.status_code == 401:
            try:
                erro_json = response.json()
                mensagem = erro_json.get("message", "Token inválido ou expirado.")
            except ValueError:
                mensagem = "Erro 401: Não autorizado. (Resposta não é JSON)"
            print(f"[Erro 401] Acesso não autorizado")
            print(f"Mensagem da API: {mensagem}")
            print("Encerrando o programa. Verifique se o token OAuth está correto ou ainda válido.")
            sys.exit(1)

        #Falha na consulta API, com o status para verificação
        else:
            print(f"[Erro] API falhou para o ID {game_id} - Status: {response.status_code}")
            game_info_dict = {
                'name': 'Erro',
                'rating': None,  # Usando None para valores ausentes numéricos
                'rating_count': None,
                'aggregated_rating': None,
                'aggregated_rating_count': None,
                'total_rating': None,
                'total_rating_count': None
            }
            cache_info[game_id] = game_info_dict
            return game_info_dict

    #Falha na consulta da API por outro motivo
    except Exception as e:
        print(f"[Erro de exceção] Falha ao consultar API para o ID {game_id}: {e}")
        game_info_dict = {
            'name': 'Erro',
            'rating': None,  # Usando None para valores ausentes numéricos
            'rating_count': None,
            'aggregated_rating': None,
            'aggregated_rating_count': None,
            'total_rating': None,
            'total_rating_count': None
        }
        cache_info[game_id] = game_info_dict
        return game_info_dict

#----------------------------------------------------------------------------------#

# Função para atualizar o DataFrame
def update_game_info(df):
    #Copia em um novo df apenas os ID que possuem name vazia para consulta na API
    #Confere se é um valor ausente ou se está vazio (mesmo sem espaços)
    df_faltando_info = df[df['name'].isna() | (df['name'].astype(str).str.strip() == '')].copy()

    #Se o novo df estiver vazio, não há informações para consultar
    if df_faltando_info.empty:
        print("Nenhum jogo novo para consultar na API.")
        return df

    #Definição de "lotes" para consulta à API, para não sobrecarregar as consultas
        #Quantidade de Streams processados por iteração
    batch_size = 5

    #Processamento das consultas por lotes de todo o df a ser consultado
    for i in range(0, len(df_faltando_info), batch_size):
        #Define o batch conforme posição das linhas por indexação (0 até 5 - sem incluir o 6)
        batch = df_faltando_info.iloc[i:i + batch_size]

        # Para cada linha do batch, pega o ID do Stream, e salva o retorno da Função acima
        for idx, row in batch.iterrows():
            game_id = row['ID_IGDB']
            info = get_game_info(game_id)
            for col in cols_to_add:
                df.loc[idx, col] = info.get(col, None)

        #Informa usuário qual lote (inteiro) está sendo processado
        print(f"Processando o lote {i // batch_size + 1} de {(len(df_faltando_info) - 1) // batch_size + 1}")

        #Pausa de 5s para evitar sobrecarregar a API e evitar limites de taxa (rate limits)
        time.sleep(5)

    return df

#----------------------------------------------------------------------------------#

# Chamada da principal função
df_combinado = update_game_info(df_combinado)

#Salvar df para csv
df_combinado.to_csv(igdb_database, index=False, encoding='utf-8-sig')


df_combinado

No próximo post irei detalhar sobre a coleta desses campos mais complexos.

Nos vemos no próximo post!