from pathlib import Path import pandas as pd import requests import shutil from datetime import datetime import logging import webbrowser import tempfile import numpy as np import re # Configurações iniciais URL_HOSPEDAGEM = "https://cdn3.gnarususercontent.com.br/2928-transformacao-manipulacao-dados/dados_hospedagem.json" ARQUIVO_PARQUET = Path("df_dados_hospedagem.parquet") PASTA_BACKUP = Path("backup") LINHAS_FIDELIZACAO = 10 # Logger logging.basicConfig(level=logging.INFO, format='[%(asctime)s] %(levelname)s: %(message)s') def baixar_json_para_dataframe(url): logging.info("Baixando JSON da URL...") response = requests.get(url) response.raise_for_status() json_data = response.json() return extrair_dataframe(json_data) def extrair_dataframe(json_data): """Extrai dataframe de estrutura JSON aninhada.""" if isinstance(json_data, list): return pd.json_normalize(json_data) elif isinstance(json_data, dict): for key in json_data: if isinstance(json_data[key], list): return pd.json_normalize(json_data[key]) return pd.json_normalize(json_data) else: raise ValueError("Estrutura JSON não suportada") def inferir_menor_float(series): """Infere o menor tipo float para economizar memória.""" numeric_series = pd.to_numeric(series, errors='coerce') numeric_series = numeric_series.dropna() if numeric_series.empty: return series.astype(np.float32) if not series.isnull().all() else series min_val = numeric_series.min() max_val = numeric_series.max() if np.can_cast(min_val, np.float16) and np.can_cast(max_val, np.float16): return series.astype(np.float16) elif np.can_cast(min_val, np.float32) and np.can_cast(max_val, np.float32): return series.astype(np.float32) else: return series.astype(np.float64) def inferir_menor_int(series): """Infere o menor tipo inteiro para economizar memória.""" numeric_series = pd.to_numeric(series, errors='coerce') numeric_series = numeric_series.dropna() if numeric_series.empty: return series.astype(np.int32) if not series.isnull().all() else series min_val = numeric_series.min() max_val = numeric_series.max() if np.can_cast(min_val, np.int8) and np.can_cast(max_val, np.int8): return series.astype(np.int8) elif np.can_cast(min_val, np.int16) and np.can_cast(max_val, np.int16): return series.astype(np.int16) elif np.can_cast(min_val, np.int32) and np.can_cast(max_val, np.int32): return series.astype(np.int32) else: return series.astype(np.int64) def tratar_dados_especificos(df): """Padroniza strings, converte datas e otimiza tipos numéricos e de string.""" df_tratado = df.copy() float_cols = ['avaliacao_geral', 'taxa_deposito', 'taxa_limpeza', 'preco'] int_cols = ['max_hospedes', 'quantidade_banheiros', 'quantidade_quartos', 'quantidade_camas'] string_cols = ['experiencia_local', 'descricao_local', 'descricao_vizinhanca', 'modelo_cama', 'comodidades'] for col in df_tratado.columns: if col in float_cols: if df_tratado[col].dtype == 'object': # Parseia todos os números e, se for uma lista, pega o primeiro válido. df_tratado[col] = df_tratado[col].astype(str).apply( lambda x: [ float(re.sub(r'[^\d.]', '', val).replace(',', '.')) for val in re.findall(r'[+-]?\d+(?:[\.,]\d+)?', x.replace('R$', '').replace('.', '').replace(',', '.')) ] if ( isinstance(x, str) and x.strip().startswith('[') and x.strip().endswith(']') and pd.notna(x) ) else ( float(re.sub(r'[^\d.]', '', x).replace(',', '.')) if pd.notna(x) and re.sub(r'[^\d.]', '', x).replace(',', '.').strip() != '' else np.nan ) ) # Para as colunas de preço/taxa, pegamos o primeiro valor da lista # Isso permite converter a coluna para um único tipo float, economizando RAM df_tratado[col] = df_tratado[col].apply( lambda x: x[0] if isinstance(x, list) and len(x) > 0 else np.nan ) # Aplica infer_menor_float após garantir que não há mais listas e é um único valor float df_tratado[col] = inferir_menor_float(df_tratado[col]) elif col in int_cols: if df_tratado[col].dtype == 'object': df_tratado[col] = pd.to_numeric(df_tratado[col], errors='coerce') # Preenche NaN com 0 antes de inferir o menor int para as colunas de quantidade if col in ['quantidade_banheiros', 'quantidade_quartos', 'quantidade_camas']: df_tratado[col] = df_tratado[col].fillna(0) df_tratado[col] = inferir_menor_int(df_tratado[col]) elif col in string_cols: if df_tratado[col].dtype == 'object': # Lida com estruturas aninhadas (listas) unindo-as com uma vírgula # Mantenha essa lógica para ter todo o texto em uma única célula df_tratado[col] = df_tratado[col].apply( lambda x: ','.join(map(str, x)) if isinstance(x, list) else str(x) ) # Aplica limpeza geral de strings df_tratado[col] = df_tratado[col].str.lower() df_tratado[col] = df_tratado[col].apply(lambda x: re.sub(r'[^\w\s\',\-\,\.]', '', x) if pd.notna(x) else x) df_tratado[col] = df_tratado[col].str.replace(r'\s+', ' ', regex=True).str.strip() # Converte para o tipo 'string' para eficiência de memória df_tratado[col] = df_tratado[col].astype('string') elif df_tratado[col].dtype == 'object': try: if df_tratado[col].dropna().astype(str).str.match(r'\d{4}-\d{2}-\d{2}').any(): df_tratado[col] = pd.to_datetime(df_tratado[col], errors='coerce') else: df_tratado[col] = df_tratado[col].astype(str).str.strip().str.title() except Exception as e: logging.warning(f"Ignorando coluna '{col}' durante o tratamento: {e}") return df_tratado def padronizar_colunas(df, n_linhas=10): """Garante a fidelização dos nomes de colunas com base nas primeiras n linhas e transforma em lista.""" df_limpo = df.head(n_linhas).copy() colunas_padronizadas = {col: col.strip().lower().replace(" ", "_") for col in df_limpo.columns} df.rename(columns=colunas_padronizadas, inplace=True) df.columns = [col.replace(' ', '_') for col in df.columns] logging.info(f"Nomes das colunas padronizados: {','.join(df.columns.tolist())}") return df def fazer_backup(arquivo: Path): if not PASTA_BACKUP.exists(): PASTA_BACKUP.mkdir() timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") destino = PASTA_BACKUP / f"{arquivo.stem}_{timestamp}{arquivo.suffix}" shutil.copy2(arquivo, destino) logging.info(f"Backup criado: {destino}") def salvar_em_parquet(df: pd.DataFrame, caminho: Path): df.to_parquet(caminho, index=False) logging.info(f"Arquivo salvo como: {caminho}") def mostrar_resumo(df): print("\nResumo do DataFrame:") print(f"Linhas: {df.shape[0]} | Colunas: {df.shape[1]}") print("\nInfo:") df.info() # Correção: Chamando df.info() sem o argumento 'buf' def exibir_df_html(df: pd.DataFrame): html = df.to_html(index=False, classes="tabela-preta") estilo = """ """ conteudo = f"{estilo}{html}" arquivo_temp = Path(tempfile.gettempdir()) / "visualizacao_df.html" with open(arquivo_temp, "w", encoding="utf-8") as f: f.write(conteudo) webbrowser.open(f"file://{arquivo_temp}") def processar_hospedagem(): if ARQUIVO_PARQUET.exists(): resposta = input("Arquivo já existe. Deseja atualizá-lo? (s/n): ").strip().lower() if resposta != 's': logging.info("Carregando arquivo existente...") df = pd.read_parquet(ARQUIVO_PARQUET) mostrar_resumo(df) exibir_df_html(df) return else: fazer_backup(ARQUIVO_PARQUET) df = baixar_json_para_dataframe(URL_HOSPEDAGEM) df = tratar_dados_especificos(df) df = padronizar_colunas(df, n_linhas=LINHAS_FIDELIZACAO) salvar_em_parquet(df, ARQUIVO_PARQUET) mostrar_resumo(df) exibir_df_html(df) # Executar if __name__ == "__main__": processar_hospedagem()