#13 – Consulta de dados IGDB

Informações Gerais IGDB

O IGDB (Internet Game Database) possui várias informações gerais dos jogos como coleção (collection), empresa (company), datas de lançamento (date), franquia (franchise), gênero (genre), plataformas (platforms), classificação indicativa (rating) e resumo (summary).

Foi decidido para o projeto coletar todas essas informações de cada jogo justamente para a tela de detalhes de cada jogo.

Para a consulta API de cada informação, cada um possui sua URL e seus campos, mas a lógica e sequência de consulta seria a mesma para cada um.

Portanto, decidi criar aqui um menu no qual você escolhe qual informação coletar que é definido automaticamente, buscando agilidade no código e redução do tamanho.

#Tabelas IGDB

# 1 - Collection
# 2 - Company
# 3 - Date
# 4 - Franchise
# 5 - Genre
# 6 - Platform
# 7 - Rating
# 8 - Summary

# Configurações das tabelas
tabelas_config = {
    1: {
        "name": "Collection",
        "base": "Base/base_collection.CSV",
        "base2": "Não disponível",
        "base3": "Não disponível",
        "dim": "Dim/dim_collection.CSV",
        "dim2": "Não disponível",
        "rel": "Rel/rel_collection.CSV",
        "colunas": "collection_id",
        "colunas_dim": "collection_id",
        "api_fields": "collections", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/collections", # Defina a URL base da API
        "url_base_dim_2": "Não disponível",
        "url_base_dim_3": "Não disponível",
    },
    2: {
        "name": "Company",
        "base": "Base/base_company.CSV",
        "base2": "Base/base_company_2api.CSV",
        "base3": "Não disponível",
        "dim": "Dim/dim_company.CSV",
        "dim2": "Não disponível",
        "rel": "Rel/rel_company.CSV",
        "colunas": "company_id",
        "colunas_dim": "Não disponível",
        "api_fields": "involved_companies", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/involved_companies",
        "url_base_dim_2": "https://api.igdb.com/v4/companies",
        "url_base_dim_3": "Não disponível",
    },
    3: {
        "name": "Date",
        "base": "Base/base_date.CSV",
        "base2": "Não disponível",
        "base3": "Não disponível",
        "dim": "Dim/dim_date.CSV",
        "dim2": "Não disponível",
        "rel": "Rel/rel_date.CSV",
        "colunas": "date_id",
        "colunas_dim": "Não disponível",
        "api_fields": "release_dates", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/release_dates",
        "url_base_dim_2": "Não disponível",
        "url_base_dim_3": "Não disponível",
    },
    4: {
        "name": "Franchise",
        "base": "Base/base_franchise.CSV",
        "base2": "Não disponível",
        "base3": "Não disponível",
        "dim": "Dim/dim_franchise.CSV",
        "dim2": "Não disponível",
        "rel": "Rel/rel_franchise.CSV",
        "colunas": "franchise_id",
        "colunas_dim": "franchise_id",
        "api_fields": "franchises", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/franchises", # Defina a URL base da API
        "url_base_dim_2": "Não disponível",
        "url_base_dim_3": "Não disponível",
    },
    5: {
        "name": "Genre",
        "base": "Base/base_genre.CSV",
        "base2": "Não disponível",
        "base3": "Não disponível",
        "dim": "Dim/dim_genre.CSV",
        "dim2": "Não disponível",
        "rel": "Rel/rel_genre.CSV",
        "colunas": "genre_id",
        "colunas_dim": "genre_id",
        "api_fields": "genres", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/genres", # Defina a URL base da API
        "url_base_dim_2": "Não disponível",
        "url_base_dim_3": "Não disponível",
    },
    6: {
        "name": "Platform",
        "base": "Base/base_platform.CSV",
        "base2": "Não disponível",
        "base3": "Não disponível",
        "dim": "Dim/dim_platform.CSV",
        "dim2": "Não disponível",
        "rel": "Rel/rel_platform.CSV",
        "colunas": "platform_id",
        "colunas_dim": "platform_id",
        "api_fields": "platforms", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/platforms", # Defina a URL base da API
        "url_base_dim_2": "Não disponível",
        "url_base_dim_3": "Não disponível",
    },
    7: {
        "name": "Rating",
        "base": "Base/base_rating.CSV",
        "base2": "Base/base_rating_2api.CSV",
        "base3": "Base/base_rating_3api.CSV",
        "dim": "Dim/dim_rating.CSV",
        "dim2": "Dim/dim_rating_2api.CSV",
        "rel": "Rel/rel_rating.CSV",
        "colunas": "rating_id",
        "colunas_dim": "Não disponível",
        "api_fields": "age_ratings", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/age_ratings",
        "url_base_dim_2": "https://api.igdb.com/v4/age_rating_categories",
        "url_base_dim_3": "https://api.igdb.com/v4/age_rating_content_descriptions_v2",
    },
    8: {
        "name": "Summary - Não disponível",
        "base": "Não disponível",
        "base2": "Não disponível",
        "base3": "Não disponível",
        "dim": "Dim/dim_summary.CSV",
        "dim2": "Não disponível",
        "rel": "Não disponível",
        "colunas": "rating_id",
        "colunas_dim": "Não disponível",
        "api_fields": "age_ratings", #texto que será enviado para API (sem separação em aspas)
        "url_base_dim": "https://api.igdb.com/v4/games",
        "url_base_dim_2": "Não disponível",
        "url_base_dim_3": "Não disponível",
    },
}

Portanto o código desenvolvido para acesso à API e coleta dessas informações

#API IGDB - IGDB - Todas as tabelas

#Opção de selecionar Tabela

#Auxiliar para o While
aux = 0

#Roda função até usuário selecionar tabela disponível.
while aux == 0:

    # Exibir opções para o usuário
    print("Escolha uma das tabelas abaixo:")

    #Exibe os valores no name de cada tabela
    for key, value in tabelas_config.items():
        print(f"{key}: {value['name']}")

    #Recebe valor selecionado pelo usuário
    tabela_id  = int(input("Digite o número da tabela desejada: "))

    #Verifica condição, Tabela 8 não está disponível
    #Se for 8, pede para usuário selecionar outra
    if (tabela_id == 8):
        print("Tabela não disponível. Selecione outra.")

    #Ao escolher uma das opções disponíveis
    else:
        #Salva em config
        config = tabelas_config[tabela_id]
        #Altera valor de aux para sair do loop
        aux = 1

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

# Leitura dos arquivos IGDB Database e da Tabela selecionada
df_atualizado = pd.read_csv(igdb_database, sep=",", encoding='utf-8-sig')
df_base = pd.read_csv(os.path.join(caminho_tabelas, config["base"]), sep=",", encoding='utf-8-sig')

# Remove colunas desnecessárias do df_atualizado
df_atualizado = df_atualizado.drop(columns=[
    'name', 'rating', 'rating_count', 'aggregated_rating',
    'aggregated_rating_count', 'total_rating', 'total_rating_count'
])

# Converte para string para evitar problemas
df_base['ID_IGDB'] = df_base['ID_IGDB'].apply(lambda x: str(x) if pd.notna(x) else 'NA')

#Somente na primeira vez que salva
#df_base[config["colunas"]] = ''

#Para o df_base, pode ser que tenham IDs 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[config["colunas"]] = df_base[config["colunas"]].apply(lambda x: str(x) if pd.notna(x) else 'NA')

#Transforma o ID_IGDB em string para a consulta API
df_atualizado['ID_IGDB'] = df_atualizado['ID_IGDB'].apply(lambda x: str(x) if pd.notna(x) else 'NA')

# Adiciona colunas vazias para consulta depois
df_atualizado[config["colunas"]] = ''

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

# Junta dados antigos + novos no df_combinado sem duplicatas
df_combinado = pd.concat([df_base, 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()

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

# Define a URL base da API (imagem)
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_api_info(game_id, api_fields):
    #Checa se o ID já tem informação 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 da URL com a ID
    body = f"""
    where id = {game_id};
    fields {api_fields};
    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()

            # Verificando se 'data' é uma lista não vazia
            if isinstance(game_data, list) and len(game_data) > 0:
                #Salva a informação ao acessar o primeiro item da lista ([0])
                #Caso não encontre, salva como 'NA'
                info = str(game_data[0].get(api_fields) or 'NA')

                #Imprime resultado
                print(f"{api_fields} encontrado para {game_id}: {info}")

                #Salva no dicionário cache
                cache_info[game_id] = info

                #Retorno da função
                return info

            #Caso não tenha nada (resposta vazia) em "data"
            else:
                print(f"Nenhum dado retornado para {game_id}")
                return 'NA' # Armazenar 'NA' se não houver informação

        #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 ao acessar API para {game_id}: {response.status_code}")
            return 'Erro'  # Armazenar 'Erro' se a API falhar

    #Falha na consulta da API por outro motivo
    except Exception as e:
        print(f"Exceção ao consultar API para {game_id}: {e}")
        return 'Erro'  # Armazenar 'Erro' se houver erro na consulta

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

# Função para atualizar o DataFrame
def update_df_colunas(df, coluna_df, coluna_api):
    #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_to_update = df[df[coluna_df].isna() | (df[coluna_df].astype(str).str.strip() == '')].copy()

    #Se o novo df estiver vazio, não há informações para consultar
    if df_to_update.empty:
        print("Todos os jogos já têm id. Nenhuma chamada de API necessária.")
        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
    #len(df_to_update)

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

        #Chama função da API, a partir do game id, para os campos na api_fields e armazena na respectiva coluna
        batch[coluna_df] = batch['ID_IGDB'].apply(lambda game_id: get_api_info(game_id, coluna_api))

        #Atualiza o DataFrame original com os novos valores
        df.loc[batch.index, coluna_df] = batch[coluna_df]

        #Informa usuário qual lote (inteiro) está sendo processado
        print(f"Processando o lote {i // batch_size + 1} de {(len(df_to_update) - 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_df_colunas(df_combinado, config["colunas"], config["api_fields"])

#Salvar df para csv
df_combinado.to_csv(os.path.join(caminho_tabelas, config["base"]), index=False, encoding='utf-8-sig')

df_combinado

No próximo post irei explicar em mais detalhes nos casos em que os campos possuem mais de uma informação, conforme demonstrado no Post 12, mais especificamente de Datas de Lançamento, Empresas, Classificação Indicativa e Resumo.

Até o próximo post!