Transforming Data Cleaning

# python
import os
import pandas as pd
import shutil
from datetime import datetime

# Define folders and paths
data_folder = "data"
archive_folder = os.path.join(data_folder, "archive")
insights_folder = "insights"
logs_folder = "logs"
log_path = os.path.join(logs_folder, "ingest_log.csv")

# Create folders if they don't exist
for folder in [data_folder, archive_folder, insights_folder, logs_folder]:
    os.makedirs(folder, exist_ok=True)

print("✓ Folder structure and paths set up.")

# Find the orders file in data_folder (any filename containing 'orders')
files = os.listdir(data_folder)
file_name = next((f for f in files if "orders" in f and f.lower().endswith(".csv")), None)

if not file_name:
    raise FileNotFoundError("No orders file found in the data folder.")
else:
    file_path = os.path.join(data_folder, file_name)
    file_id = os.path.splitext(file_name)[0]
    print(f"✅ Found file: {file_name}")

# Load existing ingest log if it exists
if os.path.exists(log_path):
    log = pd.read_csv(log_path)
else:
    log = pd.DataFrame()

# python
# Load data from the located file and lookup tables
orders_path = file_path  # the file we found above
products_path = os.path.join("data", "products.csv")
customers_path = os.path.join("data", "customers.csv")

orders = pd.read_csv(orders_path)
products = pd.read_csv(products_path)
customers = pd.read_csv(customers_path)

# Keep a raw copy for logging or later re-ingestion
orders_raw = orders.copy()

# python
required_columns = ["order_id", "customer_id", "product_id", "quantity", "order_date"]

# Validate required columns exist
missing_columns = [c for c in required_columns if c not in orders.columns]
if missing_columns:
    raise KeyError(f"Missing required columns: {missing_columns}")

# Identify rows with missing required fields
missing_mask = orders[required_columns].isnull().any(axis=1)
dropped_missing_ids = orders.loc[missing_mask, "order_id"].tolist()

if dropped_missing_ids:
    print(f"🗑️ Removed {len(dropped_missing_ids)} rows with missing required fields: {dropped_missing_ids}")

# Drop them
orders = orders[~missing_mask]

# python
# Convert order_date to datetime, invalid parsing will be NaT
orders["order_date_parsed"] = pd.to_datetime(orders["order_date"], errors="coerce")
invalid_dates_mask = orders["order_date_parsed"].isna()
dropped_date_ids = orders.loc[invalid_dates_mask, "order_id"].tolist()

if dropped_date_ids:
    print(f"🗑️ Removed {len(dropped_date_ids)} rows with invalid order_date: {dropped_date_ids}")

orders = orders[~invalid_dates_mask]

# python
numeric_fields = ["customer_id", "product_id", "quantity"]
invalid_numeric_mask = pd.Series(False, index=orders.index)

for field in numeric_fields:
    # Coerce to numeric; invalid parsing becomes NaN
    orders[f"{field}_checked"] = pd.to_numeric(orders[field], errors="coerce")

    # Invalid if NaN or negative (for quantity require > 0; for IDs require >= 0)
    if field == "quantity":
        invalids = orders[f"{field}_checked"].isna() | (orders[f"{field}_checked"] <= 0)
    else:
        invalids = orders[f"{field}_checked"].isna() | (orders[f"{field}_checked"] < 0)

    if invalids.any():
        dropped = orders.loc[invalids, "order_id"].tolist()
        print(f"🗑️ Removed {len(dropped)} rows with invalid {field}: {dropped}")
        invalid_numeric_mask |= invalids

# Drop all rows that failed any numeric checks
orders = orders[~invalid_numeric_mask]

# Convert checked numeric columns to integers (safe now)
orders["customer_id"] = orders["customer_id_checked"].astype(int)
orders["product_id"] = orders["product_id_checked"].astype(int)
orders["quantity"] = orders["quantity_checked"].astype(int)

# Drop helper checked columns
orders.drop(columns=[f"{field}_checked" for field in numeric_fields], inplace=True)

# python
duplicates_mask = orders.duplicated(keep="first")
duplicate_ids = orders.loc[duplicates_mask, "order_id"].tolist()

if duplicate_ids:
    print(f"🗑️ Removed {len(duplicate_ids)} duplicate rows: {duplicate_ids}")

orders = orders[~duplicates_mask]

# python
# Customers
valid_customer_ids = set(customers["customer_id"])
invalid_customer_mask = ~orders["customer_id"].isin(valid_customer_ids)
dropped_customer_ids = orders.loc[invalid_customer_mask, "order_id"].tolist()

if dropped_customer_ids:
    print(f"❌ Removed {len(dropped_customer_ids)} rows with invalid customer_id: {dropped_customer_ids}")

orders = orders[~invalid_customer_mask]

# Products
valid_product_ids = set(products["product_id"])
invalid_product_mask = ~orders["product_id"].isin(valid_product_ids)
dropped_product_ids = orders.loc[invalid_product_mask, "order_id"].tolist()

if dropped_product_ids:
    print(f"❌ Removed {len(dropped_product_ids)} rows with invalid product_id: {dropped_product_ids}")

orders = orders[~invalid_product_mask]

# python
dropped_order_ids = set(orders_raw["order_id"].tolist()) - set(orders["order_id"].tolist())
if dropped_order_ids:
    dropped_rows = orders_raw[orders_raw["order_id"].isin(dropped_order_ids)]
    dropped_path = os.path.join(insights_folder, "ordersDropped.csv")
    dropped_rows.to_csv(dropped_path, index=False)
    print(f"Saved dropped rows ({len(dropped_rows)}) to {dropped_path}")
else:
    print("No rows were dropped during cleaning.")

# python
# Remove helper columns and reset index
if "order_date_parsed" in orders.columns:
    orders.drop(columns=["order_date_parsed"], inplace=True)

orders = orders.reset_index(drop=True)

# Save cleaned orders
cleaned_path = os.path.join(insights_folder, f"{file_id}_cleaned.csv")
orders.to_csv(cleaned_path, index=False)
print(f"Saved cleaned data to {cleaned_path}")

# Archive the raw source file
os.makedirs(archive_folder, exist_ok=True)
shutil.move(file_path, os.path.join(archive_folder, file_name))
print(f"Moved raw file to {archive_folder}/{file_name}")

# Update ingest log
status = "cleaned"
row_count = len(orders)
log_entry = pd.DataFrame([{
    "file_name": file_name,
    "status": status,
    "rows": row_count,
    "timestamp": datetime.now().replace(microsecond=0).isoformat()
}])

if os.path.exists(log_path):
    log = pd.read_csv(log_path)
    log = pd.concat([log, log_entry], ignore_index=True)
else:
    log = log_entry

os.makedirs(logs_folder, exist_ok=True)
log.to_csv(log_path, index=False)
print(f"Logged ingestion to {log_path}")