Cleaning Messy CSV Data with Pandas

# Dependencies: pip install pandas
# Usage: python clean_encoding.py ./data/raw_export.csv

import pandas as pd
import sys

def load_robust_csv(filepath: str) -> pd.DataFrame:
 """Ingest CSV with automatic delimiter detection and encoding fallback."""
 try:
 # Attempt UTF-8 with BOM support and auto-separator detection
 df = pd.read_csv(
 filepath, 
 encoding='utf-8-sig', 
 sep=None, 
 engine='python'
 )
 print(f"[OK] Loaded with UTF-8-SIG encoding.")
 except UnicodeDecodeError:
 # Fallback to Latin-1 for legacy Windows/ISO-8859-1 exports
 df = pd.read_csv(
 filepath, 
 encoding='latin-1', 
 sep=None, 
 engine='python'
 )
 print("[WARN] Fallback to Latin-1 encoding applied.")
 except Exception as e:
 print(f"[ERROR] Ingestion failed: {e}")
 sys.exit(1)
 
 return df

if __name__ == "__main__":
 if len(sys.argv) < 2:
 print("Usage: python script.py <relative/path/to/file.csv>")
 sys.exit(1)
 raw_df = load_robust_csv(sys.argv[1])
 print(f"Shape: {raw_df.shape} | Columns: {list(raw_df.columns)}")

# Dependencies: pip install pandas
# Assumes raw_df is loaded from the previous step

def standardize_schema(df: pd.DataFrame) -> pd.DataFrame:
 """Clean headers and enforce explicit dtypes."""
 try:
 # Normalize headers: strip whitespace, lowercase, replace spaces with underscores
 df.columns = df.columns.str.strip().str.lower().str.replace(r'\s+', '_', regex=True)
 
 # Explicit type mapping to prevent inference overhead
 type_map = {
 'order_id': 'string',
 'quantity': 'Int64', # Nullable integer
 'unit_price': 'float64',
 'created_at': 'datetime64[ns]'
 }
 
 # Apply mapping safely (ignores missing columns)
 existing_cols = [c for c in type_map.keys() if c in df.columns]
 df = df.astype({col: type_map[col] for col in existing_cols})
 
 # Parse dates if not caught by astype
 if 'created_at' in df.columns:
 df['created_at'] = pd.to_datetime(df['created_at'], errors='coerce')
 
 return df
 except Exception as e:
 print(f"[ERROR] Schema standardization failed: {e}")
 raise

# df_clean = standardize_schema(raw_df)

# Dependencies: pip install pandas
# Assumes df_clean is loaded

def remediate_records(df: pd.DataFrame) -> pd.DataFrame:
 """Impute placeholders, forward-fill categorical gaps, and deduplicate."""
 try:
 # Standardize common placeholders to pandas NA
 placeholder_cols = ['status', 'shipping_method']
 for col in placeholder_cols:
 if col in df.columns:
 df[col] = df[col].replace(['', 'N/A', 'unknown', '-'], pd.NA)
 
 # Forward-fill categorical gaps where business logic allows
 if 'status' in df.columns:
 df['status'] = df['status'].ffill()
 
 # Drop rows missing critical identifiers
 df = df.dropna(subset=['order_id', 'quantity'], how='any')
 
 # Enforce unique record constraint, keeping the most recent entry
 if 'created_at' in df.columns:
 df = df.sort_values('created_at')
 
 df = df.drop_duplicates(subset=['order_id'], keep='last')
 
 return df.reset_index(drop=True)
 except Exception as e:
 print(f"[ERROR] Record remediation failed: {e}")
 raise

# df_final = remediate_records(df_clean)

# Dependencies: pip install pandas
# Usage: python process_large.py ./data/large_export.csv

import pandas as pd
import os

def process_large_csv(filepath: str, chunk_size: int = 50000) -> pd.DataFrame:
 """Memory-efficient chunked processing for large CSVs."""
 if not os.path.exists(filepath):
 raise FileNotFoundError(f"File not found: {filepath}")
 
 cleaned_chunks = []
 try:
 # Define categorical columns upfront to save memory during iteration
 dtype_map = {'region': 'category', 'product_sku': 'category'}
 
 # Initialize chunk iterator
 chunks = pd.read_csv(
 filepath, 
 chunksize=chunk_size, 
 dtype=dtype_map,
 encoding='utf-8-sig',
 sep=None,
 engine='python'
 )
 
 for i, chunk in enumerate(chunks):
 # Apply lightweight cleaning per chunk
 chunk.columns = chunk.columns.str.strip().str.lower().str.replace(' ', '_')
 chunk = chunk.dropna(subset=['email', 'user_id'])
 cleaned_chunks.append(chunk)
 print(f"[PROGRESS] Processed chunk {i+1}")
 
 # Concatenate once outside the loop to avoid fragmentation
 return pd.concat(cleaned_chunks, ignore_index=True)
 except Exception as e:
 print(f"[ERROR] Chunked processing failed: {e}")
 raise

# df_large = process_large_csv('./data/large_export.csv')

# Dependencies: pip install pandas
# Assumes df_validated is loaded

def validate_and_export(df: pd.DataFrame, output_path: str) -> None:
 """Run integrity checks and serialize cleaned DataFrame."""
 try:
 # 1. Type & Null Validation
 null_threshold = 0.05
 null_pct = df.isnull().mean()
 if (null_pct > null_threshold).any():
 cols_exceed = null_pct[null_pct > null_threshold].index.tolist()
 print(f"[WARN] Columns exceeding {null_threshold*100}% null threshold: {cols_exceed}")
 
 # 2. Row Count Assertion (example: expect > 1000 records)
 if len(df) < 1000:
 print("[ALERT] Row count below expected minimum. Review ingestion filters.")
 
 # 3. Export
 df.to_csv(output_path, index=False, encoding='utf-8')
 print(f"[SUCCESS] Cleaned data exported to {output_path}")
 print(f"Final Shape: {df.shape} | Memory: {df.memory_usage(deep=True).sum() / 1024**2:.2f} MB")
 except Exception as e:
 print(f"[ERROR] Validation/Export failed: {e}")
 raise

# validate_and_export(df_large, './output/cleaned_export.csv')