Clean Spreadsheet and Tabular Data Before AI Analysis

order_id,order_date,city,revenue_pkr,status,customer_phone
0001,2026-07-01,Lahore,"2,500",delivered,0300-0000001
0002,2026-07-02,lahore,1800,Delivered,0300-0000002
0002,2026-07-02,lahore,1800,Delivered,0300-0000002
0003,2026-07-03,ISB,,pending,0300-0000003
0004,not-a-date,Karachi,3200,returned,0300-0000004
0005,2026-07-05,KHI,-500,delivered,0300-0000005

from pathlib import Path
import json
import pandas as pd

SOURCE = Path("sales_raw.csv")
OUTPUT = Path("sales_clean.csv")
REPORT = Path("quality_report.json")

# Read as text first so identifiers and original spellings survive inspection.
raw = pd.read_csv(SOURCE, dtype=str, keep_default_na=False)
df = raw.copy()

required = {"order_id", "order_date", "city", "revenue_pkr", "status"}
missing_columns = sorted(required - set(df.columns))
if missing_columns:
    raise ValueError(f"Missing required columns: {missing_columns}")

df.columns = [c.strip().lower() for c in df.columns]
df["order_id"] = df["order_id"].str.strip()
df["city_original"] = df["city"]

city_map = {
    "lahore": "Lahore",
    "isb": "Islamabad",
    "islamabad": "Islamabad",
    "khi": "Karachi",
    "karachi": "Karachi",
}
df["city"] = df["city"].str.strip().str.casefold().map(city_map)
df["status"] = df["status"].str.strip().str.casefold()
allowed_statuses = {"delivered", "pending", "returned"}

df["order_date_parsed"] = pd.to_datetime(
    df["order_date"], format="%Y-%m-%d", errors="coerce"
)
df["revenue_pkr_parsed"] = pd.to_numeric(
    df["revenue_pkr"].str.replace(",", "", regex=False).str.strip(),
    errors="coerce",
)

# Remove only later copies of fully identical rows. If the same order ID has
# conflicting content, keep every version and quarantine the whole ID group.
duplicate_id_mask = df.duplicated(subset=["order_id"], keep=False)
full_row_duplicate_mask = df.duplicated(keep=False)
conflicting_ids = df.loc[
    duplicate_id_mask & ~full_row_duplicate_mask, "order_id"
].unique()
conflicting_id_mask = df["order_id"].isin(conflicting_ids)
exact_duplicate_drop_mask = df.duplicated(keep="first") & ~conflicting_id_mask
missing_order_id_mask = df["order_id"].eq("")
invalid_date_mask = df["order_date_parsed"].isna()
invalid_city_mask = df["city"].isna()
invalid_status_mask = ~df["status"].isin(allowed_statuses)
missing_revenue_mask = df["revenue_pkr_parsed"].isna()
negative_revenue_mask = df["revenue_pkr_parsed"].lt(0).fillna(False)

report = {
    "source_file": SOURCE.name,
    "rows_in": int(len(raw)),
    "exact_duplicate_rows_removed": int(exact_duplicate_drop_mask.sum()),
    "conflicting_duplicate_id_rows": int(conflicting_id_mask.sum()),
    "missing_order_ids": int(missing_order_id_mask.sum()),
    "invalid_dates": int(invalid_date_mask.sum()),
    "unmapped_cities": int(invalid_city_mask.sum()),
    "unmapped_statuses": int(invalid_status_mask.sum()),
    "missing_or_invalid_revenue": int(missing_revenue_mask.sum()),
    "negative_revenue_needing_review": int(negative_revenue_mask.sum()),
    "raw_parseable_revenue_total_pkr": float(df["revenue_pkr_parsed"].sum()),
}

# Deduplication is justified here only for fully identical copies. Conflicting
# versions remain visible and are marked for source-owner review.
clean = df.loc[~exact_duplicate_drop_mask].copy()

# Do not impute questionable business facts. Exclude unresolved rows from analysis.
review_mask = (
    clean["order_id"].eq("")
    | clean["order_date_parsed"].isna()
    | clean["city"].isna()
    | ~clean["status"].isin(allowed_statuses)
    | clean["revenue_pkr_parsed"].isna()
    | clean["revenue_pkr_parsed"].lt(0)
    | clean["order_id"].isin(conflicting_ids)
)
clean["review_status"] = review_mask.map({True: "needs_review", False: "analysis_ready"})

# Remove the unnecessary phone field from the analytic export.
clean = clean.drop(columns=["customer_phone"], errors="ignore")
clean.to_csv(OUTPUT, index=False)

report["rows_out"] = int(len(clean))
report["analysis_ready_rows"] = int((clean["review_status"] == "analysis_ready").sum())
REPORT.write_text(json.dumps(report, indent=2), encoding="utf-8")
print(json.dumps(report, indent=2))