Skip to content

Instantly share code, notes, and snippets.

@pedrovasconcellos
Created January 27, 2025 20:33
Show Gist options
  • Save pedrovasconcellos/4341993f635392f6f8a512fc03725571 to your computer and use it in GitHub Desktop.
Save pedrovasconcellos/4341993f635392f6f8a512fc03725571 to your computer and use it in GitHub Desktop.
Check CSV
import os
import csv
# Gets the user's HOME directory
home_dir = os.path.expanduser("~")
# Dynamically builds the path for the input and output CSV
input_csv_path = os.path.join(home_dir, "Downloads", "test", "base.csv")
output_csv_path = os.path.join(home_dir, "Downloads", "test", "base_altered.csv")
# Opens the input file for reading and creates the output file for writing
with open(input_csv_path, 'r', newline='', encoding='utf-8') as input_file, \
open(output_csv_path, 'w', newline='', encoding='utf-8') as output_file:
reader = csv.DictReader(input_file)
column_names = reader.fieldnames # Gets the column names
writer = csv.DictWriter(output_file, fieldnames=column_names)
writer.writeheader() # Writes the header to the output file
for row in reader:
# Changes all values in the 'email' column
row['email'] = '[email protected]'
writer.writerow(row)
print("Modified file generated successfully!")
import os
import csv
# Obtém o diretório HOME do usuário
home_dir = os.path.expanduser("~")
# Constrói dinamicamente o caminho para o CSV de entrada e de saída
caminho_csv_entrada = os.path.join(home_dir, "Downloads", "test", "base.csv")
caminho_csv_saida = os.path.join(home_dir, "Downloads", "test", "base_alterada.csv")
# Abre o arquivo de entrada para leitura e cria o arquivo de saída para escrita
with open(caminho_csv_entrada, 'r', newline='', encoding='utf-8') as arquivo_entrada, \
open(caminho_csv_saida, 'w', newline='', encoding='utf-8') as arquivo_saida:
leitor = csv.DictReader(arquivo_entrada)
nomes_colunas = leitor.fieldnames # Pega os nomes das colunas
escritor = csv.DictWriter(arquivo_saida, fieldnames=nomes_colunas)
escritor.writeheader() # Escreve o cabeçalho no arquivo de saída
for linha in leitor:
# Altera todos os valores da coluna 'email'
linha['email'] = '[email protected]'
escritor.writerow(linha)
print("Arquivo alterado gerado com sucesso!")
import pandas as pd
import os
def check_csv():
# Additional configurations
# Specify the expected data types for each column (optional)
expected_dtypes = {
# 'column_name': 'expected_type',
# Example:
'name': 'object',
'email': 'object',
'document': 'int64',
}
# Specify the columns that must have unique values (optional)
unique_columns = [
'document',
]
# Specify the expected categories for categorical columns (optional)
expected_categories = {
# 'Status': ['Active', 'Inactive', 'Pending'],
}
# Specify the mandatory columns in the CSV
required_columns = [
'name',
'email',
'document',
]
# Gets the user's home directory
home_dir = os.path.expanduser("~")
# Defines the full path to the CSV file in the Downloads/test folder
csv_path = os.path.join(home_dir, "Downloads/test", "base.csv")
print(f"Checking the file: {csv_path}\n")
# Checks if the file exists
if not os.path.isfile(csv_path):
print("Error: The file 'file.csv' was not found in the Downloads folder.")
return
try:
# Attempts to read the CSV using pandas
df = pd.read_csv(csv_path)
print("CSV reading successful.\n")
# 1. Check if all required columns are present
print("Checking required columns...")
present_columns = df.columns.tolist()
missing_columns = [col for col in required_columns if col not in present_columns]
if missing_columns:
print(f"Error: The following required columns are missing: {missing_columns}\n")
else:
print("All required columns are present.\n")
# 2. Check for missing values
print("Checking for missing values...")
if df.isnull().values.any():
print("Warning: The CSV file contains missing values.")
print(df.isnull().sum(), "\n")
else:
print("No missing values found in the CSV.\n")
# 3. Check if all rows have the same number of columns
print("Checking consistency in the number of columns per row...")
with open(csv_path, 'r', encoding='utf-8') as file:
num_columns = None
line_num = 0
inconsistencies = []
for line in file:
line_num += 1
columns = line.strip().split(',')
if num_columns is None:
num_columns = len(columns)
elif len(columns) != num_columns:
inconsistencies.append((line_num, len(columns)))
if inconsistencies:
for ln, num in inconsistencies:
print(f"Error: Inconsistency in the number of columns on line {ln}. Found {num} columns.")
print("\n")
else:
print("All rows have the same number of columns.\n")
# 4. Check for duplicate rows
print("Checking for duplicate rows...")
num_duplicates = df.duplicated().sum()
if num_duplicates > 0:
print(f"Attention: There are {num_duplicates} duplicate rows in the CSV.\n")
else:
print("No duplicate rows found in the CSV.\n")
# 5. Check for unique values in specific columns
if unique_columns:
print("Checking for unique values in specific columns...")
for column in unique_columns:
if column in df.columns:
num_duplicated_col = df[column].duplicated().sum()
if num_duplicated_col > 0:
print(f"Attention: The column '{column}' contains {num_duplicated_col} duplicated values.")
else:
print(f"All values in the column '{column}' are unique.")
else:
print(f"Error: The column '{column}' does not exist in the CSV.")
print("\n")
# 6. Check column data types
print("Checking column data types...")
real_dtypes = df.dtypes
inconsistent_dtypes = {}
for column, expected_type in expected_dtypes.items():
if column in df.columns:
real_type = real_dtypes[column]
# Simplified mapping for comparison
type_map = {
'int': 'int64',
'float': 'float64',
'str': 'object',
'datetime': 'datetime64[ns]'
}
expected_type_real = type_map.get(expected_type, expected_type)
if real_type != expected_type_real:
inconsistent_dtypes[column] = {'expected': expected_type, 'found': real_type}
else:
print(f"Error: The column '{column}' does not exist in the CSV.")
if inconsistent_dtypes:
print("Error: Inconsistencies in the data types of the following columns:")
for column, types in inconsistent_dtypes.items():
print(f" - Column '{column}': Expected {types['expected']}, Found {types['found']}")
print("\n")
else:
print("All column data types match the expected values.\n")
# 7. Check for consistent categories in categorical columns
if expected_categories:
print("Checking for consistent categories in categorical columns...")
for column, categories in expected_categories.items():
if column in df.columns:
unique_values = df[column].dropna().unique()
unexpected_values = [val for val in unique_values if val not in categories]
if unexpected_values:
print(f"Attention: The column '{column}' contains unexpected values: {unexpected_values}")
else:
print(f"All categories in the column '{column}' match the expected values.")
else:
print(f"Error: The column '{column}' does not exist in the CSV.")
print("\n")
# 8. (Optional) Check for extra columns not expected
# print("Checking for extra columns in the CSV...")
# extra_columns = [col for col in df.columns if col not in required_columns]
# if extra_columns:
# print(f"Attention: There are extra columns in the CSV that are not expected: {extra_columns}\n")
# else:
# print("There are no extra columns in the CSV.\n")
# Final summary
print("Verification Summary:")
if (not missing_columns and
num_duplicates == 0 and
not inconsistent_dtypes and
(not unique_columns or all(df[col].duplicated().sum() == 0 for col in unique_columns)) and
(not expected_categories or all(
not any(val not in expected_categories[col] for val in df[col].dropna())
for col in expected_categories
))):
print("✔️ The CSV file is 100% correct.")
else:
print("⚠️ The CSV file has some inconsistencies. Please review the messages above.")
except pd.errors.EmptyDataError:
print("Error: The CSV file is empty.")
except pd.errors.ParserError as e:
print(f"CSV parsing error: {e}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
if __name__ == "__main__":
check_csv()
import pandas as pd
import os
def verificar_csv():
# Configurações adicionais
# Especifique os tipos de dados esperados para cada coluna (opcional)
tipos_esperados = {
# 'nome_da_coluna': 'tipo_esperado',
# Exemplo:
'nome': 'object',
'email': 'object',
'documento': 'int64',
}
# Especifique as colunas que devem ter valores únicos (opcional)
colunas_unicas = [
'documento',
]
# Especifique as categorias esperadas para colunas categóricas (opcional)
categorias_esperadas = {
# 'Status': ['Ativo', 'Inativo', 'Pendente'],
}
# Especifique as colunas que são obrigatórias no CSV
colunas_obrigatorias = [
'nome',
'email',
'documento',
]
# Obtém o diretório home do usuário
home_dir = os.path.expanduser("~")
# Define o caminho completo para o arquivo CSV na pasta Downloads
caminho_csv = os.path.join(home_dir, "Downloads/test", "base.csv")
print(f"Verificando o arquivo: {caminho_csv}\n")
# Verifica se o arquivo existe
if not os.path.isfile(caminho_csv):
print("Erro: O arquivo 'file.csv' não foi encontrado na pasta Downloads.")
return
try:
# Tenta ler o CSV usando pandas
df = pd.read_csv(caminho_csv)
print("Leitura do CSV bem-sucedida.\n")
# 1. Verificar se todas as colunas obrigatórias estão presentes
print("Verificando colunas obrigatórias...")
colunas_presentes = df.columns.tolist()
colunas_faltantes = [col for col in colunas_obrigatorias if col not in colunas_presentes]
if colunas_faltantes:
print(f"Erro: As seguintes colunas obrigatórias estão faltando: {colunas_faltantes}\n")
else:
print("Todas as colunas obrigatórias estão presentes.\n")
# 2. Verificar se há valores ausentes
print("Verificando valores ausentes...")
if df.isnull().values.any():
print("Aviso: O arquivo CSV contém valores ausentes.")
print(df.isnull().sum(), "\n")
else:
print("Nenhum valor ausente encontrado no CSV.\n")
# 3. Verificar se todas as linhas têm o mesmo número de colunas
print("Verificando consistência no número de colunas por linha...")
with open(caminho_csv, 'r', encoding='utf-8') as file:
num_colunas = None
linha_num = 0
inconsistencias = []
for linha in file:
linha_num += 1
colunas = linha.strip().split(',')
if num_colunas is None:
num_colunas = len(colunas)
elif len(colunas) != num_colunas:
inconsistencias.append((linha_num, len(colunas)))
if inconsistencias:
for linha_num, num in inconsistencias:
print(f"Erro: Inconsistência no número de colunas na linha {linha_num}. Encontrado {num} colunas.")
print("\n")
else:
print("Todas as linhas têm o mesmo número de colunas.\n")
# 4. Verificar linhas duplicadas
print("Verificando linhas duplicadas...")
num_duplicados = df.duplicated().sum()
if num_duplicados > 0:
print(f"Atenção: Existem {num_duplicados} linhas duplicadas no CSV.\n")
else:
print("Nenhuma linha duplicada encontrada no CSV.\n")
# 5. Verificar valores únicos em colunas específicas
if colunas_unicas:
print("Verificando valores únicos em colunas específicas...")
for coluna in colunas_unicas:
if coluna in df.columns:
num_duplicados_col = df[coluna].duplicated().sum()
if num_duplicados_col > 0:
print(f"Atenção: A coluna '{coluna}' contém {num_duplicados_col} valores duplicados.")
else:
print(f"Todos os valores na coluna '{coluna}' são únicos.")
else:
print(f"Erro: A coluna '{coluna}' não existe no CSV.")
print("\n")
# 6. Verificar tipos de dados das colunas
print("Verificando tipos de dados das colunas...")
tipos_reais = df.dtypes
tipos_inconsistentes = {}
for coluna, tipo_esperado in tipos_esperados.items():
if coluna in df.columns:
tipo_real = tipos_reais[coluna]
# Mapeamento simplificado para comparação
tipo_map = {
'int': 'int64',
'float': 'float64',
'str': 'object',
'datetime': 'datetime64[ns]'
}
tipo_esperado_real = tipo_map.get(tipo_esperado, tipo_esperado)
if tipo_real != tipo_esperado_real:
tipos_inconsistentes[coluna] = {'esperado': tipo_esperado, 'encontrado': tipo_real}
else:
print(f"Erro: A coluna '{coluna}' não existe no CSV.")
if tipos_inconsistentes:
print("Erro: Inconsistências nos tipos de dados das seguintes colunas:")
for coluna, tipos in tipos_inconsistentes.items():
print(f" - Coluna '{coluna}': Esperado {tipos['esperado']}, Encontrado {tipos['encontrado']}")
print("\n")
else:
print("Todos os tipos de dados das colunas estão conforme o esperado.\n")
# 7. Verificar categorias consistentes em colunas categóricas
if categorias_esperadas:
print("Verificando categorias consistentes em colunas categóricas...")
for coluna, categorias in categorias_esperadas.items():
if coluna in df.columns:
valores_unicos = df[coluna].dropna().unique()
valores_inesperados = [val for val in valores_unicos if val not in categorias]
if valores_inesperados:
print(f"Atenção: A coluna '{coluna}' contém valores inesperados: {valores_inesperados}")
else:
print(f"Todas as categorias na coluna '{coluna}' estão conforme o esperado.")
else:
print(f"Erro: A coluna '{coluna}' não existe no CSV.")
print("\n")
# 8. Verificar se há colunas extras não esperadas (opcional)
# Você pode descomentar e definir 'colunas_obrigatorias' para verificar isso
# print("Verificando colunas extras no CSV...")
# colunas_extras = [col for col in df.columns if col not in colunas_obrigatorias]
# if colunas_extras:
# print(f"Atenção: Existem colunas extras no CSV que não são esperadas: {colunas_extras}\n")
# else:
# print("Não há colunas extras no CSV.\n")
# Resumo final
print("Resumo da Verificação:")
if (not colunas_faltantes and
num_duplicados == 0 and
not tipos_inconsistentes and
(not colunas_unicas or all(df[col].duplicated().sum() == 0 for col in colunas_unicas)) and
(not categorias_esperadas or all(
not any(val not in categorias_esperadas[col] for val in df[col].dropna())
for col in categorias_esperadas
))):
print("✔️ O arquivo CSV está 100% correto.")
else:
print("⚠️ O arquivo CSV possui algumas inconsistências. Por favor, revise as mensagens acima.")
except pd.errors.EmptyDataError:
print("Erro: O arquivo CSV está vazio.")
except pd.errors.ParserError as e:
print(f"Erro de parsing no CSV: {e}")
except Exception as e:
print(f"Ocorreu um erro inesperado: {e}")
if __name__ == "__main__":
verificar_csv()
**Creating the Virtual Environment:**
Run the following command to create a virtual environment named `venv`:
```bash
python3 -m venv venv
```
After creating the virtual environment, you need to activate it in order to start using it:
```bash
source venv/bin/activate
```
Before installing packages, it's a good practice to upgrade pip to the latest version within the virtual environment:
```bash
pip install --upgrade pip
```
With the virtual environment active, install pandas using pip:
```bash
pip install pandas
```
---
**Verifying the Installation**
To ensure pandas was installed correctly, you can run:
```bash
python -c "import pandas as pd; print(pd.__version__)"
```
---
**Deactivating the Virtual Environment**
After you finish working in the virtual environment, you can deactivate it with:
```bash
deactivate
```
**Criando o Ambiente Virtual:**
Execute o seguinte comando para criar um ambiente virtual chamado `venv`:
```bash
python3 -m venv venv
```
Após criar o ambiente virtual, você precisa ativá-lo para começar a usá-lo:
```bash
source venv/bin/activate
```
Antes de instalar pacotes, é boa prática atualizar o pip para a versão mais recente dentro do ambiente virtual:
```bash
pip install --upgrade pip
```
Com o ambiente virtual ativo, instale o pandas usando o pip:
```bash
pip install pandas
```
---
**Verificando a Instalação**
Para garantir que o pandas foi instalado corretamente, você pode executar:
```bash
python -c "import pandas as pd; print(pd.__version__)"
```
---
**Desativando o Ambiente Virtual**
Depois de terminar de trabalhar no ambiente virtual, você pode desativá-lo com:
```bash
deactivate
```
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment