#11 – Junção datasets Twitch + IGDB

IGDB

O website IGDB (Internet Game Database) é um database online de jogos que foi adquirida pela Twitch em 2019. Lá tem as informações básicas de jogos como Desenvolvedora, Avaliações, Resumo, Classificação Indicativa, Datas de Lançamento e Plataformas compatíveis.

Link IGDB

No caso, essas informações são relevantes ao Projeto para entender as relações entre Nota de avaliação com número de visualizações, ou desenvolvedora, classificação indicativa, entre outras informações valiosas que podem ser relacionadas ou não.

Por ter sido adquirida pela Twitch, há um relacionamento entre ID_Twitch e ID_IGDB no qual facilita a consulta via API apenas pelo ID.

No entanto, para a consulta de Viewers da Twitch, não é recebido na resposta o ID_IGDB, portanto há alguns Streams que não possuem esse ID. Para tal, foi desenvolvido o código abaixo para fazer a coleta via API desses ID faltantes.

Para os que já possuem o ID_IGDB, não há necessidade de realizar a consulta e acabar sobrecarregando o request, portanto é feito uma separação de dataframe com apenas os ID que precisam ser consultados.

#API IGDB - Database IGDB - Consulta ID IGDB

#Leitura do arquivo Twitch Database e IGDB Database
df_base = pd.read_csv(twitch_database_igdb, sep=",", encoding='utf-8-sig')
df_atualizado = pd.read_csv(twitch_database, sep=",", encoding='utf-8-sig')

# Remove colunas desnecessárias do df_atualizado e renomeia IDdoJogo para ID_Twitch
df_atualizado = df_atualizado.rename(columns={'IDdoJogo': 'ID_Twitch'})

# Converte para string para evitar problemas
df_base['ID_Twitch'] = df_base['ID_Twitch'].astype(int).astype(str)

#Para o df_base, pode ser que tenham ID_IGDB vazios (já consultados e salvos previamente).
#É aplicado a função lambda para não necessitar reconsulta de API, preenchendo o vazio por 'NA'
df_base['ID_IGDB'] = df_base['ID_IGDB'].apply(lambda x: str(x) if pd.notna(x) else 'NA')

#Transforma o ID Twitch em string para a consulta API
df_atualizado['ID_Twitch'] = df_atualizado['ID_Twitch'].astype(int).astype(str)

# Remove duplicados de base e atualizado
df_base = df_base.drop_duplicates(subset='ID_Twitch')
df_atualizado = df_atualizado.drop_duplicates(subset='ID_Twitch')

# Cria o dataframe combinado juntando base + atualizado
df_combinado = pd.concat([df_base, df_atualizado], ignore_index=True)
df_combinado = df_combinado.drop_duplicates(subset='ID_Twitch', keep='first')

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


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

# Define a URL base da API (imagem)
url_base = url_base_twitch

# 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_igdb_id = {}

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

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

    #Definição da URL com a ID
    url = f"{url_base}{game_id}"

    #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.get(url, headers=headers)

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

            #Resposta padrão da API em JSON: "data": [{"id":, "name":, "box_art_url":, "igdb_id"}]
            #Verificação da lista "data" no dicionário "game_data"
            if game_data['data']:
                #Salva o igdb_id ao acessar o primeiro item da lista ([0])
                #Caso não encontre, salva como None (Vazio)
                igdb_id = game_data['data'][0].get('igdb_id', None)

                #Checa o resultado do igdb_id
                if igdb_id:
                    print(f"ID encontrado para {game_id}: {igdb_id}")

                    #Salva no dicionário cache
                    cache_igdb_id[game_id] = igdb_id

                    #Retorno da função
                    return igdb_id
                else:
                    print(f"[Aviso] Nenhum igdb_id encontrado para o ID {game_id}")
                    cache_igdb_id[game_id] = 'NA'  # Armazenar 'NA' se não houver igdb_id
                    return 'NA'

            #Caso não tenha nada (resposta vazia) em "data"
            else:
                print(f"[Aviso] Nenhum dado retornado pela API para o ID {game_id}")
                cache_igdb_id[game_id] = 'NA'  # Armazenar 'NA' se não houver dados
                return 'NA'

        #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}")
            cache_igdb_id[game_id] = 'Erro'  # Armazenar 'Erro' se a API falhar
            return 'Erro'

    #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}")
        cache_igdb_id[game_id] = 'Erro'  # Armazenar 'Erro' se houver erro na consulta
        return 'Erro'

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

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

    #Se o novo df estiver vazio, não há informações para consultar
    if df_faltando_id.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 informações 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_id), batch_size):
        #Define o batch conforme posição das linhas por indexação (0 até 5 - sem incluir o 6)
        batch = df_faltando_id.iloc[i:i + batch_size]

        # Para cada linha do batch, pega o ID, e salva o retorno da Função acima
        for idx, row in batch.iterrows():
            game_id = row['ID_Twitch']
            img = get_igdb_id(game_id)
            df.loc[idx, 'ID_IGDB'] = img

        #Informa usuário qual lote (inteiro) está sendo processado.
        print(f"Processando o lote {i // batch_size + 1} de {(len(df_faltando_id) - 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_igdb_ids(df_combinado)

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

df_combinado

No próximo post irei detalhar um pouco mais sobre a conexão via API da IGDB, por ser um pouco diferente comparado ao da Twitch, e desenvolvido em Python.

Até o próximo Post!