#10 – Limpeza e normalização Twitch

Limpeza dos dados

A limpeza dos dados é uma etapa importante para a área de dados no geral, pois é uma das etapas de padronização para conseguir se obter um insight com fundamentação.

Correção dos Nomes

Conforme explicado no post anterior, a correção feita nas tabelas coletadas seriam a limpeza no nome dos Streams. Alguns Streams coletados no API da Twitch possuem a vírgula “,” no nome e para arquivos .CSV o padrão é que o delimitador normalmente é a vírgula.

Portanto, ao abrir o arquivo no Excel por exemplo, a vírgula nos nomes dos Streams são entendidos como separadores, ou seja uma nova coluna. Para evitar esse tipo de erro, é feito a correção no nome dos Streams.

df_game = pd.read_csv(top_100_twitch, sep=",", encoding='latin-1')

df_game['NomedoJogo'] = df_game['NomedoJogo'].replace({
    'Animals, Aquariums, and Zoos': 'Animals Aquariums and Zoos',
    'Pools, Hot Tubs, and Beaches': 'Pools Hot Tubs and Beaches',
    'Yes, Your Grace': 'Yes Your Grace',
    'Warhammer 40,000: Darktide': 'Warhammer 40000: Darktide',
    'Warhammer 40,000: Space Marine II': 'Warhammer 40000: Space Marine II',
    'Yes, Your Grace: Snowfall': 'Yes Your Grace: Snowfall',
    'Ship, Inc.': 'Ship Inc.',
    "No, I'm Not A Human": "No I'm Not A Human"
})

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

df_game

Criação Database Twitch

A Database Twitch seria basicamente um arquivo de consulta com o ID_Twitch, Nome e Imagem, com entrada única por ID. O código faz uma verificação se há um ID que está ou não no Database, caso não esteja, é separado num dataframe separado que faz a “adição” no Database.

#Leitura do arquivo Top 100 Twitch e do Database
df_games = pd.read_csv(top_100_twitch_corrected, sep=",", encoding='utf-8-sig')
df_database = pd.read_csv(twitch_database, sep=",", encoding='utf-8-sig')

#Exclui colunas de data e Index - irrelevantes para o Database prévio
df_games = df_games.drop(columns=['Date_GET_Games', 'Index'])

#Remove duplicatas
df_games = df_games.drop_duplicates()

#Para garantir a duplicata, agrupa os jogos por Id, Nome e Image
df_games = df_games.groupby(['IDdoJogo', 'NomedoJogo', 'ImagedoJogo']).size().reset_index(name='Count')

#Remove coluna Count criada no agrupamento
df_games = df_games.drop(columns=['Count'])

#Substitui o texto pelo tamanho das imagens a serem geradas posteriormente
df_games['ImagedoJogo'] = df_games['ImagedoJogo'].str.replace('{width}', '300')
df_games['ImagedoJogo'] = df_games['ImagedoJogo'].str.replace('{height}', '400')

#Para garantir a duplicata, agrupa os jogos por Id e Nome, depois remove coluna gerada
df_database = df_database.groupby(['IDdoJogo', 'NomedoJogo', 'ImagedoJogo']).size().reset_index(name='Count')
df_database = df_database.drop(columns=['Count'])

# Converte IDs para string para evitar problemas nas chamadas das APIs
df_games['IDdoJogo'] = df_games['IDdoJogo'].astype(int).astype(str)
df_database['IDdoJogo'] = df_database['IDdoJogo'].astype(int).astype(str)

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

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

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

df_combinado

Como o código acima só faz a verificação pelo arquivo de Top 100 Streams, é necessário verificar pelo Top 100 Views, no entanto pelo Views não tem a informação da Imagem do Stream. Portanto, além de checar os ID novos, é necessário a consulta API para pegar as informações da Imagem.

#Atualização da entrada de jogos novos no Database prévio pelo Top 100 Viewers
#Top 100 Games e Top 100 Viewers são diferentes.

# Leitura dos CSVs
df_atualizado = pd.read_csv(top_100_viewers_corrected, sep=",", encoding='utf-8-sig')
df_viewers = pd.read_csv(twitch_games_database, sep=",", encoding='utf-8-sig')

# Remove colunas desnecessárias
df_atualizado = df_atualizado.drop(columns=['Date_GET_Games', 'Index','Count_views'])

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

# Adiciona coluna de imagem vazia para consulta depois
df_atualizado['ImagedoJogo'] = ''

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

# Junta dados antigos + novos no df_combinado sem duplicatas
df_combinado = pd.concat([df_viewers, df_atualizado], ignore_index=True)
df_combinado = df_combinado.drop_duplicates(subset='IDdoJogo', 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_imagens = {}

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

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

    #Definição da URL com a ID da Stream
    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 box_art_url ao acessar o primeiro item da lista ([0])
                #Caso não encontre, salva como None (Vazio)
                box_art_url = game_data['data'][0].get('box_art_url', None)

                #Checa o resultado do box_art_url
                if box_art_url:
                    print(f"Box Art encontrado para {game_id}: {box_art_url}")

                    #Salva no dicionário cache
                    cache_imagens[game_id] = box_art_url

                    #Retorno da função
                    return box_art_url
                else:
                    print(f"[Aviso] Nenhum box_art_url encontrado para o ID {game_id}")
                    cache_imagens[game_id] = 'NA'  # Armazenar 'NA' se não houver box_art_url
                    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_imagens[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_imagens[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_imagens[game_id] = 'Erro'  # Armazenar 'Erro' se houver erro na consulta
        return 'Erro'

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

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

    #Se o novo df estiver vazio, não há informações para consultar
    if df_faltando_imagem.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_imagem), batch_size):
        #Define o batch conforme posição das linhas por indexação (0 até 5 - sem incluir o 6)
        batch = df_faltando_imagem.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['IDdoJogo']
            img = get_game_image(game_id)
            df.loc[idx, 'ImagedoJogo'] = img

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

#Substitui o texto pelo tamanho das imagens a serem geradas posteriormente
df_combinado['ImagedoJogo'] = df_combinado['ImagedoJogo'].str.replace('{width}', '300')
df_combinado['ImagedoJogo'] = df_combinado['ImagedoJogo'].str.replace('{height}', '400')

# Salva o DataFrame atualizado
df_combinado.to_csv(twitch_database, index=False, encoding='utf-8-sig')

# Exibe resultado final
df_combinado

Desse modo, o Database da Twitch é atualizado quando há Streams novo e pode ser utilizado para relacionar o ID com o Nome e Imagem do Stream.

No próximo post detalharei a consulta dos dados relacionados ao IGDB.

Nos vemos no próximo post!