Preprocessing Data With Automated Pipelines

# example_data.csv
# This CSV file provides sample data to demonstrate the functionality of the data_preprocessing_pipeline plugin.
#
# Column Descriptions:
#   - ID: Unique identifier for each record.
#   - Feature1: Numerical feature with some missing values.
#   - Feature2: Categorical feature with multiple categories and potential typos.
#   - Feature3: Date feature in string format.
#   - Target: Binary target variable (0 or 1).
#
# Placeholders:
#   - [MISSING_VALUE]: Represents a missing value to be handled by the pipeline.
#   - [TYPO_CATEGORY]: Represents a typo in a categorical value.
#
# Instructions:
#   - Feel free to modify this data to test different preprocessing scenarios.
#   - Ensure the data adheres to the expected format for each column.
#   - Use the `/preprocess` command to trigger the preprocessing pipeline on this data.

ID,Feature1,Feature2,Feature3,Target
1,10.5,CategoryA,2023-01-15,1
2,12.0,CategoryB,2023-02-20,0
3,[MISSING_VALUE],CategoryC,2023-03-25,1
4,15.2,CategoryA,2023-04-01,0
5,9.8,CateogryB,[MISSING_VALUE],1
6,11.3,CategoryC,2023-05-10,0
7,13.7,CategoryA,2023-06-15,1
8,[MISSING_VALUE],CategoryB,2023-07-20,0
9,16.1,CategoryC,2023-08-25,1
10,10.0,CategoryA,2023-09-01,0
11,12.5,[TYPO_CATEGORY],2023-10-10,1
12,14.9,CategoryB,2023-11-15,0
13,11.8,CategoryC,2023-12-20,1
14,13.2,CategoryA,2024-01-25,0
15,9.5,CategoryB,2024-02-01,1

#!/usr/bin/env python3
"""
Error handling script for data preprocessing pipeline.

Manages and logs errors during preprocessing including:
- Exception tracking
- Error categorization
- Logging to files
- Error statistics
- Recovery recommendations
"""

import argparse
import json
import sys
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List, Optional
import traceback


class ErrorHandler:
    """Handles and logs errors in preprocessing pipeline."""

    ERROR_CATEGORIES = {
        'validation': 'Data validation errors',
        'transformation': 'Data transformation errors',
        'io': 'File I/O errors',
        'type': 'Type conversion errors',
        'missing': 'Missing data errors',
        'duplicate': 'Duplicate data errors',
        'schema': 'Schema mismatch errors',
        'unknown': 'Unknown errors',
    }

    def __init__(self, log_file: Optional[str] = None):
        """
        Initialize error handler.

        Args:
            log_file: Path to log file (optional)
        """
        self.log_file = log_file
        self.errors = []
        self.error_stats = {cat: 0 for cat in self.ERROR_CATEGORIES.keys()}
        self.session_id = datetime.now().isoformat()

    def log_error(
        self,
        error: Exception,
        category: str = 'unknown',
        context: Optional[Dict[str, Any]] = None,
        stack_trace: bool = True
    ) -> Dict[str, Any]:
        """
        Log an error.

        Args:
            error: Exception object
            category: Error category
            context: Additional context data
            stack_trace: Include stack trace

        Returns:
            Error record dictionary
        """
        if category not in self.ERROR_CATEGORIES:
            category = 'unknown'

        error_record = {
            'timestamp': datetime.now().isoformat(),
            'category': category,
            'message': str(error),
            'type': type(error).__name__,
            'context': context or {},
        }

        if stack_trace:
            error_record['stack_trace'] = traceback.format_exc()

        self.errors.append(error_record)
        self.error_stats[category] += 1

        # Write to log file if configured
        if self.log_file:
            self._write_to_log(error_record)

        return error_record

    def _write_to_log(self, error_record: Dict[str, Any]) -> None:
        """Write error record to log file."""
        try:
            path = Path(self.log_file)
            path.parent.mkdir(parents=True, exist_ok=True)

            with open(self.log_file, 'a', encoding='utf-8') as f:
                f.write(json.dumps(error_record) + '\n')
        except Exception as e:
            print(f"Failed to write to log file: {str(e)}", file=sys.stderr)

    def validate_data_integrity(
        self,
        data: List[Dict[str, Any]],
        max_errors: int = 100
    ) -> Dict[str, Any]:
        """
        Validate data integrity and log issues.

        Args:
            data: List of data rows
            max_errors: Maximum errors to report

        Returns:
            Integrity report
        """
        integrity_issues = {
            'empty_rows': [],
            'null_values': [],
            'duplicates': [],
            'type_mismatches': [],
        }

        seen_rows = set()
        error_count = 0

        for idx, row in enumerate(data):
            if error_count >= max_errors:
                break

            # Check for empty rows
            if not row:
                integrity_issues['empty_rows'].append(idx)
                error_count += 1
                continue

            # Check for null values
            null_fields = [k for k, v in row.items() if v is None or v == '']
            if null_fields:
                if len(integrity_issues['null_values']) < max_errors:
                    integrity_issues['null_values'].append({
                        'row': idx,
                        'fields': null_fields
                    })
                    error_count += 1

            # Check for duplicates
            row_str = json.dumps(row, sort_keys=True, default=str)
            if row_str in seen_rows:
                if len(integrity_issues['duplicates']) < max_errors:
                    integrity_issues['duplicates'].append(idx)
                    error_count += 1
            else:
                seen_rows.add(row_str)

        return {
            'total_rows': len(data),
            'issues': integrity_issues,
            'issue_count': sum(len(v) for v in integrity_issues.values()),
        }

    def get_error_summary(self) -> Dict[str, Any]:
        """Get error summary."""
        return {
            'session_id': self.session_id,
            'total_errors': len(self.errors),
            'error_stats': self.error_stats,
            'errors': self.errors[-100:],  # Last 100 errors
        }

    def generate_recovery_recommendations(self) -> List[str]:
        """Generate recovery recommendations based on errors."""
        recommendations = []

        if self.error_stats['validation'] > 0:
            recommendations.append(
                "Validation errors detected. Review data against schema constraints."
            )

        if self.error_stats['missing'] > 0:
            recommendations.append(
                "Missing values detected. Consider imputation strategies (mean, median, forward-fill)."
            )

        if self.error_stats['duplicate'] > 0:
            recommendations.append(
                "Duplicate records detected. Remove duplicates or keep unique identifiers."
            )

        if self.error_stats['type'] > 0:
            recommendations.append(
                "Type conversion errors detected. Verify field types match expected schema."
            )

        if self.error_stats['io'] > 0:
            recommendations.append(
                "I/O errors detected. Check file permissions and storage availability."
            )

        if not recommendations:
            recommendations.append("No errors detected. Data appears clean.")

        return recommendations

    def save_error_report(self, output_file: str) -> None:
        """
        Save comprehensive error report.

        Args:
            output_file: Output file path
        """
        try:
            report = {
                'metadata': {
                    'session_id': self.session_id,
                    'generated_at': datetime.now().isoformat(),
                },
                'summary': self.get_error_summary(),
                'recommendations': self.generate_recovery_recommendations(),
                'categories': self.ERROR_CATEGORIES,
            }

            path = Path(output_file)
            path.parent.mkdir(parents=True, exist_ok=True)

            with open(output_file, 'w', encoding='utf-8') as f:
                json.dump(report, f, indent=2)

        except Exception as e:
            print(f"Failed to save error report: {str(e)}", file=sys.stderr)
            sys.exit(1)


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='Manage and log errors during data preprocessing pipeline'
    )
    parser.add_argument(
        'action',
        choices=['test', 'analyze', 'summary'],
        help='Action to perform'
    )
    parser.add_argument(
        '-l', '--log-file',
        help='Path to error log file',
        default=None
    )
    parser.add_argument(
        '-d', '--data-file',
        help='Path to data file for integrity checking'
    )
    parser.add_argument(
        '-o', '--output',
        help='Output file for error report'
    )
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Print detailed error information'
    )

    args = parser.parse_args()

    try:
        handler = ErrorHandler(log_file=args.log_file)

        if args.action == 'test':
            # Test error logging
            try:
                raise ValueError("Test validation error")
            except ValueError as e:
                handler.log_error(e, category='validation', context={'test': True})

            try:
                raise KeyError("Test missing field")
            except KeyError as e:
                handler.log_error(e, category='missing', context={'field': 'test'})

            print("Test errors logged successfully")

        elif args.action == 'analyze':
            # Analyze data file for integrity issues
            if not args.data_file:
                print("Error: --data-file required for analyze action", file=sys.stderr)
                sys.exit(1)

            path = Path(args.data_file)
            if not path.exists():
                print(f"Error: File not found: {args.data_file}", file=sys.stderr)
                sys.exit(1)

            # Load data
            data = []
            try:
                if path.suffix.lower() == '.json':
                    with open(args.data_file, 'r') as f:
                        content = json.load(f)
                        data = content if isinstance(content, list) else [content]
                else:
                    print(f"Error: Unsupported file format: {path.suffix}", file=sys.stderr)
                    sys.exit(1)
            except json.JSONDecodeError as e:
                handler.log_error(e, category='io')

            integrity_report = handler.validate_data_integrity(data)
            if args.verbose:
                print(json.dumps(integrity_report, indent=2))

        elif args.action == 'summary':
            # Generate error summary
            summary = handler.get_error_summary()
            recommendations = handler.generate_recovery_recommendations()

            report = {
                'summary': summary,
                'recommendations': recommendations,
            }

            if args.output:
                handler.save_error_report(args.output)
                print(f"Error report saved to: {args.output}")
            else:
                print(json.dumps(report, indent=2))

        sys.exit(0)

    except Exception as e:
        print(f"Error: {str(e)}", file=sys.stderr)
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Data preprocessing pipeline orchestrator.

Orchestrates the entire data preprocessing pipeline including:
- Data loading and validation
- Data transformation
- Error handling and recovery
- Pipeline execution and monitoring
- Report generation
"""

import argparse
import json
import sys
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List, Optional
import subprocess


class PreprocessingPipeline:
    """Orchestrates data preprocessing pipeline."""

    def __init__(self, config_file: Optional[str] = None, verbose: bool = False):
        """
        Initialize pipeline.

        Args:
            config_file: Path to configuration file (JSON)
            verbose: Enable verbose output
        """
        self.config = {}
        self.verbose = verbose
        self.steps = []
        self.execution_log = []
        self.start_time = None
        self.end_time = None

        if config_file:
            self._load_config(config_file)

    def _load_config(self, config_file: str) -> None:
        """Load pipeline configuration."""
        try:
            with open(config_file, 'r') as f:
                self.config = json.load(f)
            if self.verbose:
                print(f"Loaded configuration from {config_file}")
        except Exception as e:
            raise IOError(f"Failed to load config: {str(e)}")

    def _run_step(self, step_name: str, command: List[str]) -> bool:
        """
        Run a pipeline step.

        Args:
            step_name: Name of the step
            command: Command to execute

        Returns:
            True if successful, False otherwise
        """
        step_log = {
            'name': step_name,
            'timestamp': datetime.now().isoformat(),
            'command': ' '.join(command),
            'status': 'pending',
            'duration': 0,
        }

        start = datetime.now()
        try:
            if self.verbose:
                print(f"\nExecuting: {step_name}")
                print(f"Command: {' '.join(command)}")

            result = subprocess.run(
                command,
                capture_output=True,
                text=True,
                timeout=300
            )

            duration = (datetime.now() - start).total_seconds()
            step_log['duration'] = duration
            step_log['returncode'] = result.returncode

            if result.returncode == 0:
                step_log['status'] = 'success'
                if self.verbose and result.stdout:
                    print(f"Output: {result.stdout}")
            else:
                step_log['status'] = 'failed'
                step_log['stderr'] = result.stderr
                if self.verbose:
                    print(f"Error: {result.stderr}")

        except subprocess.TimeoutExpired:
            step_log['status'] = 'timeout'
            step_log['error'] = "Step execution timed out"
        except Exception as e:
            step_log['status'] = 'error'
            step_log['error'] = str(e)
            if self.verbose:
                print(f"Exception: {str(e)}")

        self.execution_log.append(step_log)
        return step_log['status'] == 'success'

    def add_validation_step(
        self,
        data_file: str,
        schema_file: Optional[str] = None
    ) -> None:
        """
        Add data validation step.

        Args:
            data_file: Path to data file
            schema_file: Path to schema file (optional)
        """
        command = ['python3', 'validate_data.py', data_file]
        if schema_file:
            command.extend(['-s', schema_file])

        self.steps.append({
            'name': 'validate_data',
            'command': command,
            'data_file': data_file,
            'schema_file': schema_file,
        })

    def add_transformation_step(
        self,
        input_file: str,
        output_file: str,
        transformations: Optional[Dict[str, Any]] = None
    ) -> None:
        """
        Add data transformation step.

        Args:
            input_file: Path to input file
            output_file: Path to output file
            transformations: Transformation configuration
        """
        command = ['python3', 'transform_data.py', input_file, '-o', output_file]

        if transformations:
            if 'normalize' in transformations:
                for field, method in transformations['normalize']:
                    command.extend(['-n', field, method])

            if 'encode' in transformations:
                for field, method in transformations['encode']:
                    command.extend(['-e', field, method])

            if 'impute' in transformations:
                for field, method in transformations['impute']:
                    command.extend(['-i', field, method])

        self.steps.append({
            'name': 'transform_data',
            'command': command,
            'input_file': input_file,
            'output_file': output_file,
            'transformations': transformations or {},
        })

    def add_error_handling_step(
        self,
        data_file: str,
        log_file: str,
        report_file: str
    ) -> None:
        """
        Add error handling step.

        Args:
            data_file: Path to data file for integrity check
            log_file: Path to error log file
            report_file: Path to error report file
        """
        command = [
            'python3', 'handle_errors.py', 'analyze',
            '-d', data_file,
            '-l', log_file,
            '-o', report_file,
        ]

        self.steps.append({
            'name': 'error_handling',
            'command': command,
            'data_file': data_file,
            'log_file': log_file,
            'report_file': report_file,
        })

    def execute(self, stop_on_error: bool = False) -> bool:
        """
        Execute pipeline.

        Args:
            stop_on_error: Stop execution on first error

        Returns:
            True if all steps successful, False otherwise
        """
        self.start_time = datetime.now()

        if self.verbose:
            print("\n" + "=" * 60)
            print("DATA PREPROCESSING PIPELINE")
            print("=" * 60)
            print(f"Total steps: {len(self.steps)}")

        all_successful = True

        for idx, step in enumerate(self.steps, 1):
            if self.verbose:
                print(f"\n[{idx}/{len(self.steps)}] {step['name']}")

            success = self._run_step(step['name'], step['command'])

            if not success:
                all_successful = False
                if stop_on_error:
                    if self.verbose:
                        print(f"Pipeline aborted due to failure in {step['name']}")
                    break

        self.end_time = datetime.now()

        return all_successful

    def get_summary(self) -> Dict[str, Any]:
        """Get execution summary."""
        total_duration = 0
        if self.start_time and self.end_time:
            total_duration = (self.end_time - self.start_time).total_seconds()

        successful_steps = sum(1 for log in self.execution_log if log['status'] == 'success')
        failed_steps = sum(1 for log in self.execution_log if log['status'] in ('failed', 'error', 'timeout'))

        return {
            'start_time': self.start_time.isoformat() if self.start_time else None,
            'end_time': self.end_time.isoformat() if self.end_time else None,
            'total_duration': total_duration,
            'total_steps': len(self.steps),
            'successful_steps': successful_steps,
            'failed_steps': failed_steps,
            'execution_log': self.execution_log,
        }

    def save_report(self, output_file: str) -> None:
        """
        Save pipeline execution report.

        Args:
            output_file: Output file path
        """
        try:
            report = {
                'metadata': {
                    'generated_at': datetime.now().isoformat(),
                    'pipeline_config': self.config,
                },
                'summary': self.get_summary(),
            }

            path = Path(output_file)
            path.parent.mkdir(parents=True, exist_ok=True)

            with open(output_file, 'w') as f:
                json.dump(report, f, indent=2)

            if self.verbose:
                print(f"\nReport saved to: {output_file}")

        except Exception as e:
            print(f"Failed to save report: {str(e)}", file=sys.stderr)


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='Orchestrate data preprocessing pipeline'
    )
    parser.add_argument(
        '-c', '--config',
        help='Path to configuration file (JSON)',
        default=None
    )
    parser.add_argument(
        '-i', '--input',
        required=True,
        help='Path to input data file'
    )
    parser.add_argument(
        '-o', '--output',
        required=True,
        help='Path to output data file'
    )
    parser.add_argument(
        '-s', '--schema',
        help='Path to schema file for validation'
    )
    parser.add_argument(
        '-n', '--normalize',
        nargs=2,
        metavar=('FIELD', 'METHOD'),
        action='append',
        help='Normalize field'
    )
    parser.add_argument(
        '-e', '--encode',
        nargs=2,
        metavar=('FIELD', 'METHOD'),
        action='append',
        help='Encode categorical field'
    )
    parser.add_argument(
        '-i-impute', '--impute',
        nargs=2,
        metavar=('FIELD', 'METHOD'),
        action='append',
        help='Impute missing values'
    )
    parser.add_argument(
        '--report',
        help='Save pipeline report to file'
    )
    parser.add_argument(
        '--stop-on-error',
        action='store_true',
        help='Stop pipeline on first error'
    )
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Print detailed execution information'
    )

    args = parser.parse_args()

    try:
        # Create pipeline
        pipeline = PreprocessingPipeline(config_file=args.config, verbose=args.verbose)

        # Add validation step
        pipeline.add_validation_step(args.input, schema_file=args.schema)

        # Add transformation step
        transformations = {}
        if args.normalize:
            transformations['normalize'] = args.normalize
        if args.encode:
            transformations['encode'] = args.encode
        if args.impute:
            transformations['impute'] = args.impute

        pipeline.add_transformation_step(args.input, args.output, transformations)

        # Add error handling step
        log_file = args.output.replace('.', '_errors.')
        report_file = args.output.replace('.', '_report.')
        pipeline.add_error_handling_step(args.output, log_file, report_file)

        # Execute pipeline
        success = pipeline.execute(stop_on_error=args.stop_on_error)

        # Save report if requested
        if args.report:
            pipeline.save_report(args.report)

        # Print summary
        summary = pipeline.get_summary()
        if args.verbose:
            print("\n" + "=" * 60)
            print("PIPELINE SUMMARY")
            print("=" * 60)
            print(f"Duration: {summary['total_duration']:.2f} seconds")
            print(f"Successful: {summary['successful_steps']}/{summary['total_steps']}")
            print(f"Failed: {summary['failed_steps']}/{summary['total_steps']}")

        sys.exit(0 if success else 1)

    except Exception as e:
        print(f"Error: {str(e)}", file=sys.stderr)
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Data transformation script for preprocessing pipeline.

Applies transformations to data including:
- Normalization (min-max, z-score)
- Scaling (standard, robust)
- Categorical encoding (one-hot, label)
- Missing value imputation
- Feature engineering
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Tuple
import csv
from statistics import mean, stdev


class DataTransformer:
    """Applies transformations to data."""

    def __init__(self):
        """Initialize transformer."""
        self.transformations = []
        self.statistics = {}

    def load_data(self, file_path: str) -> List[Dict[str, Any]]:
        """
        Load data from file.

        Args:
            file_path: Path to CSV or JSON file

        Returns:
            List of data rows as dictionaries
        """
        path = Path(file_path)
        if not path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")

        data = []
        try:
            if path.suffix.lower() == '.csv':
                with open(file_path, 'r', encoding='utf-8') as f:
                    reader = csv.DictReader(f)
                    data = list(reader)
            elif path.suffix.lower() == '.json':
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = json.load(f)
                    data = content if isinstance(content, list) else [content]
            else:
                raise ValueError(f"Unsupported file format: {path.suffix}")
        except Exception as e:
            raise IOError(f"Failed to load data: {str(e)}")

        return data

    def normalize(
        self,
        data: List[Dict[str, Any]],
        field: str,
        method: str = 'minmax'
    ) -> List[Dict[str, Any]]:
        """
        Normalize numeric field.

        Args:
            data: List of data rows
            field: Field name to normalize
            method: 'minmax' or 'zscore'

        Returns:
            Transformed data
        """
        try:
            values = []
            for row in data:
                if field in row and row[field] is not None:
                    try:
                        values.append(float(row[field]))
                    except (ValueError, TypeError):
                        continue

            if not values:
                raise ValueError(f"No valid numeric values for field: {field}")

            if method == 'minmax':
                min_val = min(values)
                max_val = max(values)
                if min_val == max_val:
                    for row in data:
                        if field in row:
                            row[f"{field}_normalized"] = 0.5
                else:
                    for row in data:
                        if field in row and row[field] is not None:
                            try:
                                val = float(row[field])
                                normalized = (val - min_val) / (max_val - min_val)
                                row[f"{field}_normalized"] = round(normalized, 4)
                            except (ValueError, TypeError):
                                row[f"{field}_normalized"] = None

                self.statistics[field] = {'method': 'minmax', 'min': min_val, 'max': max_val}

            elif method == 'zscore':
                if len(values) > 1:
                    mean_val = mean(values)
                    std_val = stdev(values)
                    if std_val == 0:
                        for row in data:
                            if field in row:
                                row[f"{field}_normalized"] = 0.0
                    else:
                        for row in data:
                            if field in row and row[field] is not None:
                                try:
                                    val = float(row[field])
                                    normalized = (val - mean_val) / std_val
                                    row[f"{field}_normalized"] = round(normalized, 4)
                                except (ValueError, TypeError):
                                    row[f"{field}_normalized"] = None

                    self.statistics[field] = {
                        'method': 'zscore',
                        'mean': round(mean_val, 4),
                        'stdev': round(std_val, 4)
                    }
            else:
                raise ValueError(f"Unknown normalization method: {method}")

            self.transformations.append(
                f"Normalized field '{field}' using {method}"
            )

        except Exception as e:
            raise ValueError(f"Normalization failed: {str(e)}")

        return data

    def encode_categorical(
        self,
        data: List[Dict[str, Any]],
        field: str,
        method: str = 'label'
    ) -> List[Dict[str, Any]]:
        """
        Encode categorical field.

        Args:
            data: List of data rows
            field: Field name to encode
            method: 'label' or 'onehot'

        Returns:
            Transformed data
        """
        try:
            categories = {}
            for row in data:
                if field in row and row[field] is not None:
                    val = str(row[field])
                    if val not in categories:
                        categories[val] = len(categories)

            if not categories:
                raise ValueError(f"No categorical values found for field: {field}")

            if method == 'label':
                for row in data:
                    if field in row and row[field] is not None:
                        val = str(row[field])
                        row[f"{field}_encoded"] = categories.get(val)
                    else:
                        row[f"{field}_encoded"] = None

                self.transformations.append(
                    f"Label encoded field '{field}' ({len(categories)} categories)"
                )

            elif method == 'onehot':
                for row in data:
                    for cat, code in categories.items():
                        col_name = f"{field}_{cat}"
                        if field in row and row[field] is not None:
                            val = str(row[field])
                            row[col_name] = 1 if val == cat else 0
                        else:
                            row[col_name] = 0

                self.transformations.append(
                    f"One-hot encoded field '{field}' ({len(categories)} categories)"
                )
            else:
                raise ValueError(f"Unknown encoding method: {method}")

            self.statistics[field] = {
                'method': method,
                'categories': categories
            }

        except Exception as e:
            raise ValueError(f"Categorical encoding failed: {str(e)}")

        return data

    def impute_missing(
        self,
        data: List[Dict[str, Any]],
        field: str,
        method: str = 'mean'
    ) -> List[Dict[str, Any]]:
        """
        Impute missing values.

        Args:
            data: List of data rows
            field: Field name for imputation
            method: 'mean', 'median', or 'forward_fill'

        Returns:
            Transformed data
        """
        try:
            if method in ('mean', 'median'):
                values = []
                for row in data:
                    if field in row and row[field] is not None:
                        try:
                            values.append(float(row[field]))
                        except (ValueError, TypeError):
                            continue

                if not values:
                    raise ValueError(f"No numeric values for imputation in field: {field}")

                if method == 'mean':
                    fill_value = mean(values)
                else:  # median
                    sorted_vals = sorted(values)
                    n = len(sorted_vals)
                    fill_value = (
                        sorted_vals[n // 2]
                        if n % 2 == 1
                        else (sorted_vals[n // 2 - 1] + sorted_vals[n // 2]) / 2
                    )

                for row in data:
                    if field not in row or row[field] is None or row[field] == '':
                        row[field] = round(fill_value, 4)

                self.statistics[field] = {'method': method, 'fill_value': fill_value}

            elif method == 'forward_fill':
                last_value = None
                for row in data:
                    if field in row and row[field] is not None:
                        last_value = row[field]
                    elif last_value is not None:
                        row[field] = last_value

                self.statistics[field] = {'method': 'forward_fill'}

            else:
                raise ValueError(f"Unknown imputation method: {method}")

            self.transformations.append(
                f"Imputed missing values in field '{field}' using {method}"
            )

        except Exception as e:
            raise ValueError(f"Imputation failed: {str(e)}")

        return data

    def save_data(self, data: List[Dict[str, Any]], output_file: str) -> None:
        """
        Save transformed data.

        Args:
            data: Transformed data
            output_file: Output file path
        """
        try:
            path = Path(output_file)
            path.parent.mkdir(parents=True, exist_ok=True)

            if path.suffix.lower() == '.csv':
                if data:
                    with open(output_file, 'w', newline='', encoding='utf-8') as f:
                        writer = csv.DictWriter(f, fieldnames=data[0].keys())
                        writer.writeheader()
                        writer.writerows(data)
            elif path.suffix.lower() == '.json':
                with open(output_file, 'w', encoding='utf-8') as f:
                    json.dump(data, f, indent=2)
            else:
                raise ValueError(f"Unsupported output format: {path.suffix}")

        except Exception as e:
            raise IOError(f"Failed to save data: {str(e)}")

    def get_summary(self) -> Dict[str, Any]:
        """Get transformation summary."""
        return {
            'transformations': self.transformations,
            'statistics': self.statistics,
            'transformation_count': len(self.transformations),
        }


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='Apply transformations to data (normalization, encoding, imputation)'
    )
    parser.add_argument(
        'input_file',
        help='Path to input data file (CSV or JSON)'
    )
    parser.add_argument(
        '-o', '--output',
        required=True,
        help='Path to output transformed data file'
    )
    parser.add_argument(
        '-n', '--normalize',
        nargs=2,
        metavar=('FIELD', 'METHOD'),
        action='append',
        help='Normalize field (minmax or zscore)'
    )
    parser.add_argument(
        '-e', '--encode',
        nargs=2,
        metavar=('FIELD', 'METHOD'),
        action='append',
        help='Encode categorical field (label or onehot)'
    )
    parser.add_argument(
        '-i', '--impute',
        nargs=2,
        metavar=('FIELD', 'METHOD'),
        action='append',
        help='Impute missing values (mean, median, or forward_fill)'
    )
    parser.add_argument(
        '-s', '--summary',
        help='Save transformation summary to JSON file'
    )
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Print transformation details'
    )

    args = parser.parse_args()

    try:
        transformer = DataTransformer()

        # Load data
        data = transformer.load_data(args.input_file)
        if args.verbose:
            print(f"Loaded {len(data)} rows")

        # Apply transformations
        if args.normalize:
            for field, method in args.normalize:
                transformer.normalize(data, field, method)
                if args.verbose:
                    print(f"Normalized '{field}' using {method}")

        if args.encode:
            for field, method in args.encode:
                transformer.encode_categorical(data, field, method)
                if args.verbose:
                    print(f"Encoded '{field}' using {method}")

        if args.impute:
            for field, method in args.impute:
                transformer.impute_missing(data, field, method)
                if args.verbose:
                    print(f"Imputed missing values in '{field}' using {method}")

        # Save output
        transformer.save_data(data, args.output)
        if args.verbose:
            print(f"Saved {len(data)} rows to {args.output}")

        # Save summary if requested
        if args.summary:
            summary = transformer.get_summary()
            with open(args.summary, 'w') as f:
                json.dump(summary, f, indent=2)
            if args.verbose:
                print(f"Saved summary to {args.summary}")

        sys.exit(0)

    except Exception as e:
        print(f"Error: {str(e)}", file=sys.stderr)
        sys.exit(1)


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
Data validation script for preprocessing pipeline.

Validates data against predefined schemas or rules including:
- Required fields presence
- Data type correctness
- Value ranges and constraints
- Missing value handling
- Duplicates detection
"""

import argparse
import json
import sys
from pathlib import Path
from typing import Any, Dict, List, Tuple
import csv


class DataValidator:
    """Validates data against schemas and rules."""

    def __init__(self, schema_file: str = None):
        """
        Initialize validator.

        Args:
            schema_file: Path to JSON schema file (optional)
        """
        self.schema = {}
        if schema_file and Path(schema_file).exists():
            with open(schema_file, 'r') as f:
                self.schema = json.load(f)
        self.errors = []
        self.warnings = []

    def validate_file(self, file_path: str) -> bool:
        """
        Validate data file.

        Args:
            file_path: Path to data file (CSV or JSON)

        Returns:
            True if valid, False otherwise
        """
        try:
            path = Path(file_path)
            if not path.exists():
                self.errors.append(f"File not found: {file_path}")
                return False

            if path.suffix.lower() == '.csv':
                return self._validate_csv(str(path))
            elif path.suffix.lower() == '.json':
                return self._validate_json(str(path))
            else:
                self.errors.append(f"Unsupported file format: {path.suffix}")
                return False
        except Exception as e:
            self.errors.append(f"Validation error: {str(e)}")
            return False

    def _validate_csv(self, file_path: str) -> bool:
        """Validate CSV file structure and content."""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                reader = csv.DictReader(f)
                if not reader.fieldnames:
                    self.errors.append("CSV file is empty or has no headers")
                    return False

                # Check schema fields if defined
                if self.schema:
                    required_fields = self.schema.get('required_fields', [])
                    for field in required_fields:
                        if field not in reader.fieldnames:
                            self.errors.append(f"Missing required field: {field}")
                            return False

                row_count = 0
                for row_count, row in enumerate(reader, start=1):
                    if not self._validate_row(row, reader.fieldnames):
                        self.errors.append(f"Invalid data at row {row_count}")
                        if len(self.errors) > 10:  # Limit errors
                            self.warnings.append("... (more errors truncated)")
                            break

                if row_count == 0:
                    self.errors.append("CSV file contains no data rows")
                    return False

                return len(self.errors) == 0
        except Exception as e:
            self.errors.append(f"CSV validation failed: {str(e)}")
            return False

    def _validate_json(self, file_path: str) -> bool:
        """Validate JSON file structure and content."""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)

            if isinstance(data, list):
                if not data:
                    self.errors.append("JSON array is empty")
                    return False

                for idx, item in enumerate(data):
                    if not isinstance(item, dict):
                        self.errors.append(f"Row {idx} is not a dictionary")
                        return False
                    if not self._validate_row(item, item.keys()):
                        self.errors.append(f"Invalid data at row {idx}")
                        if len(self.errors) > 10:
                            self.warnings.append("... (more errors truncated)")
                            break

            elif isinstance(data, dict):
                if not data:
                    self.warnings.append("JSON object is empty")
            else:
                self.errors.append("JSON must be an object or array")
                return False

            return len(self.errors) == 0
        except json.JSONDecodeError as e:
            self.errors.append(f"Invalid JSON: {str(e)}")
            return False
        except Exception as e:
            self.errors.append(f"JSON validation failed: {str(e)}")
            return False

    def _validate_row(self, row: Dict[str, Any], fields: List[str]) -> bool:
        """Validate individual row against schema."""
        if not self.schema:
            # No schema defined, just check for basic issues
            for key, value in row.items():
                if value is None or (isinstance(value, str) and not value.strip()):
                    self.warnings.append(f"Empty value for field: {key}")
            return True

        field_types = self.schema.get('field_types', {})
        for field, expected_type in field_types.items():
            if field in row:
                if not self._validate_type(row[field], expected_type):
                    self.errors.append(
                        f"Type mismatch for field '{field}': "
                        f"expected {expected_type}, got {type(row[field]).__name__}"
                    )
                    return False
        return True

    def _validate_type(self, value: Any, expected_type: str) -> bool:
        """Validate value type."""
        if value is None:
            return True

        type_map = {
            'string': str,
            'int': int,
            'float': (int, float),
            'bool': bool,
            'number': (int, float),
        }

        if expected_type not in type_map:
            return True

        expected = type_map[expected_type]
        return isinstance(value, expected)

    def get_report(self) -> Dict[str, Any]:
        """Get validation report."""
        return {
            'valid': len(self.errors) == 0,
            'errors': self.errors,
            'warnings': self.warnings,
            'error_count': len(self.errors),
            'warning_count': len(self.warnings),
        }


def main():
    """Main entry point."""
    parser = argparse.ArgumentParser(
        description='Validate data against predefined schemas or rules'
    )
    parser.add_argument(
        'data_file',
        help='Path to data file (CSV or JSON)'
    )
    parser.add_argument(
        '-s', '--schema',
        help='Path to JSON schema file for validation',
        default=None
    )
    parser.add_argument(
        '-v', '--verbose',
        action='store_true',
        help='Print detailed validation report'
    )
    parser.add_argument(
        '-o', '--output',
        help='Save validation report to JSON file',
        default=None
    )

    args = parser.parse_args()

    # Validate data file
    validator = DataValidator(schema_file=args.schema)
    is_valid = validator.validate_file(args.data_file)
    report = validator.get_report()

    # Output report
    if args.verbose or not is_valid:
        print(json.dumps(report, indent=2))

    # Save report if requested
    if args.output:
        with open(args.output, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"Validation report saved to: {args.output}")

    # Exit with appropriate code
    sys.exit(0 if is_valid else 1)


if __name__ == '__main__':
    main()