import json
from pathlib import Path
from datetime import datetime
import requests
import pandas as pd
from typing import List, Dict, Union
# Função para obter e validar JSON da URL
def obter_json(url: str) -> Union[dict, list]:
resp = requests.get(url, timeout=10)
resp.raise_for_status()
if 'application/json' not in resp.headers.get('Content-Type', ''):
raise ValueError('Conteúdo não é JSON.')
data = resp.json()
if not isinstance(data, (dict, list)):
raise ValueError('JSON não é lista ou dicionário.')
return data
# Função para normalizar JSON aninhado em DataFrame
def normalizar_json(data: Union[dict, list]) -> pd.DataFrame:
return pd.json_normalize(data, sep='_', max_level=10)
# Função para converter tipos específicos em DataFrame
def converter_tipos(df: pd.DataFrame, tipos: Dict[str, str]) -> pd.DataFrame:
for col, tipo in tipos.items():
if col not in df.columns:
continue
try:
if tipo == 'datetime':
df[col] = pd.to_datetime(df[col], errors='raise')
elif tipo == 'numeric':
df[col] = pd.to_numeric(df[col], errors='raise')
except Exception:
print(f'Aviso: Não foi possível converter coluna "{col}" para {tipo}.')
return df
# Função para controlar o número máximo de versões do arquivo CSV
def gerenciar_versoes(pasta: Path, prefixo: str, max_versoes: int = 3):
arquivos = sorted(pasta.glob(f'{prefixo}_*.csv'), reverse=True)
for arquivo in arquivos[max_versoes:]:
try:
arquivo.unlink()
print(f'Arquivo removido: {arquivo.name}')
except Exception as e:
print(f'Erro ao remover arquivo {arquivo.name}: {e}')
# Função para salvar DataFrame em CSV com controle de versões
def salvar_csv(df: pd.DataFrame, pasta: str, prefixo: str, max_versoes: int = 3):
pasta_path = Path(pasta)
pasta_path.mkdir(exist_ok=True)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
arquivo_path = pasta_path / f'{prefixo}_{timestamp}.csv'
df.to_csv(arquivo_path, index=False, encoding='utf-8')
print(f'Arquivo CSV salvo em: {arquivo_path}')
gerenciar_versoes(pasta_path, prefixo, max_versoes)
# Função para listar colunas e seus tipos de dados de forma legível
def listar_colunas_e_tipos(df: pd.DataFrame):
tipos_map = {
'object': 'string',
'int64': 'inteiro',
'float64': 'float',
'bool': 'booleano',
'datetime64[ns]': 'datetime',
'timedelta[ns]': 'timedelta'
}
print('\nColunas e tipos de dados:')
for col, dtype in df.dtypes.items():
tipo_legivel = tipos_map.get(str(dtype), str(dtype))
print(f'- {col:<30} : {tipo_legivel}')
# Função principal que orquestra o fluxo de obtenção, processamento e salvamento dos dados
def main():
url = 'https://jsonplaceholder.typicode.com/users'
print('Obtendo e validando dados JSON...')
data = obter_json(url)
print('Normalizando dados...')
df_dados = normalizar_json(data)
tipos_esperados = {
'address_geo_lat': 'numeric',
'address_geo_lng': 'numeric',
# 'data_exemplo': 'datetime', # incluir se houver
}
df_dados = converter_tipos(df_dados, tipos_esperados)
print('\nVisualização das 10 primeiras linhas dos dados normalizados:')
print(df_dados.head(10))
listar_colunas_e_tipos(df_dados)
salvar_csv(df_dados, 'output', 'dados')
if __name__ == '__main__':
main()