Mastering Data Collection with Python – Tools, Techniques, and Workflow

import pandas as pd
import numpy as np
from typing import Dict, List, Any
import logging

class DataValidator:
    def __init__(self):
        self.validation_errors = []
        self.logger = logging.getLogger(__name__)
    
    def validate_required_fields(self, df: pd.DataFrame, required_fields: List[str]) -> pd.DataFrame:
        """Check for missing required fields"""
        missing_fields = [field for field in required_fields if field not in df.columns]
        if missing_fields:
            raise ValueError(f"Missing required fields: {missing_fields}")
        
        # Remove rows with null values in required fields
        initial_count = len(df)
        df_clean = df.dropna(subset=required_fields)
        dropped_count = initial_count - len(df_clean)
        
        if dropped_count > 0:
            self.logger.warning(f"Dropped {dropped_count} rows with missing required data")
        
        return df_clean
    
    def validate_data_types(self, df: pd.DataFrame, type_mapping: Dict[str, str]) -> pd.DataFrame:
        """Validate and convert data types"""
        df_typed = df.copy()
        
        for column, expected_type in type_mapping.items():
            if column in df_typed.columns:
                try:
                    if expected_type == 'datetime':
                        df_typed[column] = pd.to_datetime(df_typed[column])
                    elif expected_type == 'numeric':
                        df_typed[column] = pd.to_numeric(df_typed[column], errors='coerce')
                    elif expected_type == 'string':
                        df_typed[column] = df_typed[column].astype(str)
                    elif expected_type == 'category':
                        df_typed[column] = df_typed[column].astype('category')
                        
                except Exception as e:
                    self.logger.error(f"Failed to convert {column} to {expected_type}: {e}")
        
        return df_typed
    
    def remove_duplicates(self, df: pd.DataFrame, subset: List[str] = None) -> pd.DataFrame:
        """Remove duplicate records"""
        initial_count = len(df)
        df_dedup = df.drop_duplicates(subset=subset)
        duplicate_count = initial_count - len(df_dedup)
        
        if duplicate_count > 0:
            self.logger.info(f"Removed {duplicate_count} duplicate records")
        
        return df_dedup
    
    def validate_ranges(self, df: pd.DataFrame, range_checks: Dict[str, Dict]) -> pd.DataFrame:
        """Validate numeric ranges"""
        df_validated = df.copy()
        
        for column, ranges in range_checks.items():
            if column in df_validated.columns:
                min_val = ranges.get('min')
                max_val = ranges.get('max')
                
                if min_val is not None:
                    invalid_mask = df_validated[column] < min_val
                    df_validated.loc[invalid_mask, column] = np.nan
                
                if max_val is not None:
                    invalid_mask = df_validated[column] > max_val
                    df_validated.loc[invalid_mask, column] = np.nan
        
        return df_validated

# Usage example
def clean_collected_data(raw_data: List[Dict[str, Any]]) -> pd.DataFrame:
    """Complete data cleaning pipeline"""
    
    # Convert to DataFrame
    df = pd.DataFrame(raw_data)
    
    # Initialize validator
    validator = DataValidator()
    
    # Define validation rules
    required_fields = ['id', 'timestamp', 'value']
    type_mapping = {
        'id': 'string',
        'timestamp': 'datetime',
        'value': 'numeric',
        'category': 'category'
    }
    range_checks = {
        'value': {'min': 0, 'max': 1000}
    }
    
    # Apply validation steps
    df = validator.validate_required_fields(df, required_fields)
    df = validator.validate_data_types(df, type_mapping)
    df = validator.remove_duplicates(df, subset=['id'])
    df = validator.validate_ranges(df, range_checks)
    
    # Final cleanup
    df = df.dropna()  # Remove any remaining invalid rows
    
    print("Data cleaning complete: {len(df)} valid records")
    return df