0
respostas

[Sugestão] Desafio: utilizando parâmetros do json_normalize

import pandas as pd
import json
import requests
import os
import logging
from requests.exceptions import RequestException

# Configuração de logging
logging.basicConfig(
    level=logging.INFO,
    format='[%(levelname)s] %(asctime)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

# --- Configurações principais ---
ARQUIVO_JSON = 'informacoes.json'
URL = 'https://cdn3.gnarususercontent.com.br/2929-pandas/informacoes.json'

def obter_dados_json(url, arquivo_local):
    """
    Verifica e baixa o arquivo JSON, se necessário.
    Retorna o conteúdo do JSON ou None em caso de falha.
    """
    if os.path.exists(arquivo_local):
        logging.info(f'O arquivo "{arquivo_local}" já existe.')
        atualizar = input('Deseja atualizar o arquivo? (s/n): ').strip().lower()
        if atualizar == 's':
            os.remove(arquivo_local)
            logging.info('Baixando novo conteúdo...')
        else:
            logging.info('Usando o arquivo local existente.')
            with open(arquivo_local, 'r', encoding='utf-8') as f:
                return json.load(f)
    
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        dados = response.json()
        with open(arquivo_local, 'w', encoding='utf-8') as f:
            json.dump(dados, f, ensure_ascii=False, indent=4)
        logging.info(f'Arquivo "{arquivo_local}" salvo com sucesso.')
        return dados
    except (RequestException, json.JSONDecodeError, IOError) as e:
        logging.critical(f'Falha ao obter ou salvar o arquivo JSON. Erro: {e}')
        return None

def normalizar_automaticamente(df):
    """Normaliza colunas aninhadas (listas de dicts ou dicts)."""
    while True:
        colunas_aninhadas = [
            col for col in df.columns 
            if any(
                isinstance(v, list) and v and isinstance(v[0], dict) or isinstance(v, dict)
                for v in df[col].dropna().iloc[:5]
            )
        ]
        
        if not colunas_aninhadas:
            break
            
        for col_aninhada in colunas_aninhadas:
            logging.info(f'Normalizando coluna: "{col_aninhada}"')
            
            s_aninhada = df[col_aninhada].apply(lambda x: x[0] if isinstance(x, list) and x else x)
            # A linha abaixo foi alterada para remover o .add_prefix()
            df_aninhado = pd.json_normalize(s_aninhada, sep='_')
            
            df = df.drop(columns=[col_aninhada]).join(df_aninhado)

    return df

# --- Execução principal ---
dados_json = obter_dados_json(URL, ARQUIVO_JSON)

if dados_json:
    try:
        df = pd.json_normalize(dados_json, sep='_')
        df_final = normalizar_automaticamente(df)
        
        # Exibe os resultados
        print('\n' + '-'*50)
        print(f"Processamento concluído. DataFrame com {df_final.shape[0]} linhas e {df_final.shape[1]} colunas.")
        print("\n### Colunas do DataFrame Final ###")
        print(df_final.columns.tolist())
        print("\n" + '-'*50)
        df_final.info()
        print("\n" + '-'*50)
        print("\n### Amostra do DataFrame ###")
        print(df_final.head())
        print('\n' + '-'*50)

    except (pd.errors.ParserError, KeyError) as e:
        logging.critical(f'Erro ao processar o JSON com pandas. Erro: {e}')