📊 Python Data Workflows – 📸 Data Versioning 🐍

from datetime import datetime
from pathlib import Path

import pandas as pd


SOURCE_FILE = "sales_data.csv"
SNAPSHOT_DIR = Path("snapshots")
REPORT_DIR = Path("version_reports")
PRIMARY_KEY = "order_id"


def load_dataset(file_path: str) -> pd.DataFrame:
    """Load the current dataset from a CSV file."""
    df = pd.read_csv(file_path)

    if PRIMARY_KEY not in df.columns:
        raise ValueError(
            f"Required primary-key column '{PRIMARY_KEY}' was not found."
        )

    if df[PRIMARY_KEY].duplicated().any():
        raise ValueError(
            f"Duplicate values found in primary-key column '{PRIMARY_KEY}'."
        )

    print(f"✅ Loaded {len(df)} records from {file_path}\n")
    return df


def save_snapshot(df: pd.DataFrame) -> Path:
    """Save the current dataset as a timestamped snapshot."""
    SNAPSHOT_DIR.mkdir(exist_ok=True)

    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    snapshot_path = (
        SNAPSHOT_DIR / f"sales_snapshot_{timestamp}.csv"
    )

    df.to_csv(snapshot_path, index=False)

    print(f"📸 Snapshot saved: {snapshot_path}\n")
    return snapshot_path


def get_snapshots() -> list[Path]:
    """Return all saved snapshots in chronological order."""
    return sorted(
        SNAPSHOT_DIR.glob("sales_snapshot_*.csv")
    )


def compare_snapshots(
    previous_path: Path,
    current_path: Path
) -> dict[str, pd.DataFrame]:
    """Compare two snapshots using the primary-key column."""
    previous_df = pd.read_csv(previous_path).set_index(PRIMARY_KEY)
    current_df = pd.read_csv(current_path).set_index(PRIMARY_KEY)

    all_columns = sorted(
        set(previous_df.columns) | set(current_df.columns)
    )

    previous_df = previous_df.reindex(columns=all_columns)
    current_df = current_df.reindex(columns=all_columns)

    added_ids = current_df.index.difference(previous_df.index)
    removed_ids = previous_df.index.difference(current_df.index)

    added_records = current_df.loc[added_ids].copy()
    removed_records = previous_df.loc[removed_ids].copy()

    common_ids = previous_df.index.intersection(current_df.index)

    previous_common = previous_df.loc[common_ids]
    current_common = current_df.loc[common_ids]

    changed_mask = (
        previous_common.fillna("")
        .ne(current_common.fillna(""))
        .any(axis=1)
    )

    modified_ids = common_ids[changed_mask]

    modified_before = previous_common.loc[modified_ids].copy()
    modified_after = current_common.loc[modified_ids].copy()

    return {
        "added": added_records,
        "removed": removed_records,
        "modified_before": modified_before,
        "modified_after": modified_after,
    }


def save_comparison_report(
    comparison: dict[str, pd.DataFrame]
) -> None:
    """Export version-comparison results."""
    REPORT_DIR.mkdir(exist_ok=True)

    comparison["added"].to_csv(
        REPORT_DIR / "added_records.csv"
    )

    comparison["removed"].to_csv(
        REPORT_DIR / "removed_records.csv"
    )

    comparison["modified_before"].to_csv(
        REPORT_DIR / "modified_before.csv"
    )

    comparison["modified_after"].to_csv(
        REPORT_DIR / "modified_after.csv"
    )

    print("💾 Version reports exported successfully\n")


def display_summary(
    comparison: dict[str, pd.DataFrame]
) -> None:
    """Print a summary of detected changes."""
    print("📊 Version Comparison Summary")
    print(f"➕ Added records: {len(comparison['added'])}")
    print(f"➖ Removed records: {len(comparison['removed'])}")
    print(
        "✏️ Modified records: "
        f"{len(comparison['modified_after'])}"
    )

    if not comparison["added"].empty:
        print("\n➕ Added Records:")
        print(comparison["added"])

    if not comparison["removed"].empty:
        print("\n➖ Removed Records:")
        print(comparison["removed"])

    if not comparison["modified_before"].empty:
        print("\n✏️ Modified Records — Before:")
        print(comparison["modified_before"])

        print("\n✏️ Modified Records — After:")
        print(comparison["modified_after"])


def main() -> None:
    current_df = load_dataset(SOURCE_FILE)
    save_snapshot(current_df)

    snapshots = get_snapshots()

    print(f"📁 Available snapshots: {len(snapshots)}\n")

    if len(snapshots) < 2:
        print(
            "ℹ️ A first snapshot has been created.\n"
            "Update sales_data.csv and run the script again "
            "to compare changes."
        )
        return

    previous_path = snapshots[-2]
    current_path = snapshots[-1]

    print(f"Previous version: {previous_path.name}")
    print(f"Current version:  {current_path.name}\n")

    comparison = compare_snapshots(
        previous_path,
        current_path
    )

    display_summary(comparison)
    save_comparison_report(comparison)


if __name__ == "__main__":
    main()