📊 Python Data Workflows – ⚙️ Configuration-Driven Pipelines 🐍

import json
from pathlib import Path

import pandas as pd


CONFIG_FILE = Path("config.json")


def load_config(config_file: Path) -> dict:
    """Load pipeline configuration from JSON."""
    if not config_file.exists():
        raise FileNotFoundError(f"Configuration file not found: {config_file}")

    with open(config_file, "r", encoding="utf-8") as file:
        config = json.load(file)

    print(f"✅ Configuration loaded: {config_file.name}")
    return config


def validate_config(config: dict) -> None:
    """Validate required configuration keys."""
    required_keys = {
        "input_file",
        "output_file",
        "report_file",
        "required_columns",
        "numeric_columns",
        "drop_duplicates_on",
        "minimum_sales",
        "minimum_quantity",
        "group_by",
    }

    missing_keys = required_keys - set(config.keys())

    if missing_keys:
        raise ValueError(f"Missing configuration keys: {sorted(missing_keys)}")

    print("✅ Configuration validated")


def extract_data(config: dict) -> pd.DataFrame:
    """Load data using the configured input path."""
    input_file = Path(config["input_file"])

    if not input_file.exists():
        raise FileNotFoundError(f"Input file not found: {input_file}")

    df = pd.read_csv(input_file)

    print(f"✅ Loaded {len(df)} records from {input_file.name}")
    return df


def validate_data(df: pd.DataFrame, config: dict) -> None:
    """Validate dataset using configured rules."""
    required_columns = set(config["required_columns"])
    missing_columns = required_columns - set(df.columns)

    if missing_columns:
        raise ValueError(f"Missing required columns: {sorted(missing_columns)}")

    duplicate_column = config["drop_duplicates_on"]

    if duplicate_column not in df.columns:
        raise ValueError(f"Duplicate-check column '{duplicate_column}' was not found.")

    group_by = config["group_by"]

    if group_by not in df.columns:
        raise ValueError(f"Grouping column '{group_by}' was not found.")

    print("✅ Dataset validation passed")


def transform_data(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Apply configurable cleaning rules."""
    transformed_df = df.copy()

    for column in config["numeric_columns"]:
        if column not in transformed_df.columns:
            raise ValueError(f"Numeric column '{column}' was not found.")

        transformed_df[column] = pd.to_numeric(transformed_df[column], errors="coerce")

    transformed_df = transformed_df.drop_duplicates(subset=[config["drop_duplicates_on"]], keep="first")
    transformed_df = transformed_df.dropna(subset=config["numeric_columns"])

    transformed_df = transformed_df[
        (transformed_df["sales"] >= config["minimum_sales"])
        & (transformed_df["quantity"] >= config["minimum_quantity"])
    ].copy()

    transformed_df["revenue_per_item"] = (transformed_df["sales"] / transformed_df["quantity"]).round(2)

    print(f"✅ Transformation completed | records={len(transformed_df)}")
    return transformed_df


def generate_report(df: pd.DataFrame, config: dict) -> pd.DataFrame:
    """Generate a summary using configured grouping."""
    group_by = config["group_by"]

    summary = (
        df.groupby(group_by, observed=True)
        .agg(
            total_sales=("sales", "sum"),
            average_sales=("sales", "mean"),
            total_quantity=("quantity", "sum"),
            total_orders=("order_id", "count"),
        )
        .reset_index()
        .sort_values("total_sales", ascending=False)
    )

    summary["average_sales"] = summary["average_sales"].round(2)

    print(f"✅ Report generated | group_by={group_by}")
    return summary


def load_outputs(df: pd.DataFrame, summary: pd.DataFrame, config: dict) -> None:
    """Save configured pipeline outputs."""
    output_file = Path(config["output_file"])
    report_file = Path(config["report_file"])

    output_file.parent.mkdir(parents=True, exist_ok=True)
    report_file.parent.mkdir(parents=True, exist_ok=True)

    df.to_csv(output_file, index=False)
    summary.to_csv(report_file, index=False)

    print(f"💾 Processed data saved: {output_file}")
    print(f"💾 Summary saved: {report_file}")


def display_summary(df: pd.DataFrame, config: dict) -> None:
    """Display a concise workflow summary."""
    print("\n📊 Pipeline Summary")
    print(f"Processed records: {len(df)}")
    print(f"Total sales: {df['sales'].sum():,.2f}")
    print(f"Grouping field: {config['group_by']}")


def run_pipeline() -> None:
    """Run the configuration-driven pipeline."""
    print("🚀 Starting Configuration-Driven Pipeline\n")

    try:
        config = load_config(CONFIG_FILE)

        validate_config(config)

        raw_df = extract_data(config)

        validate_data(raw_df, config)

        processed_df = transform_data(raw_df, config)

        if processed_df.empty:
            raise ValueError("No valid records remain after transformation.")

        summary = generate_report(processed_df, config)

        load_outputs(processed_df, summary, config)

        display_summary(processed_df, config)

        print("\n✅ Configuration-driven pipeline completed successfully")

    except (FileNotFoundError, ValueError, KeyError, json.JSONDecodeError, pd.errors.ParserError) as error:
        print(f"\n❌ Pipeline failed: {error}")


if __name__ == "__main__":
    run_pipeline()