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 re
from decimal import Decimal
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()
df.columns = [c.strip().lower() for c in df.columns]
if df.columns.duplicated().any():
    raise ValueError("Column names collide after normalization")

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["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"
)
def parse_paisa(value):
    value = value.strip()
    if not re.fullmatch(r"-?(?:[0-9]{1,9}|[1-9][0-9]{0,2}(?:,[0-9]{3}){1,2})(?:\.[0-9]{1,2})?", value):
        return pd.NA
    return int(Decimal(value.replace(",", "")) * 100)

df["revenue_paisa"] = pd.array(df["revenue_pkr"].map(parse_paisa), dtype="Int64")

# 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.
distinct_rows = df.loc[~raw.duplicated(keep="first")]
conflicting_ids = distinct_rows.loc[
    distinct_rows.duplicated(subset=["order_id"], keep=False), "order_id"
].unique()
conflicting_id_mask = df["order_id"].isin(conflicting_ids)
exact_duplicate_drop_mask = raw.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_paisa"].isna()
negative_revenue_mask = df["revenue_paisa"].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_paisa": int(df["revenue_paisa"].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_paisa"].isna()
    | clean["revenue_paisa"].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["analysis_ready_total_paisa"] = int(clean.loc[clean["review_status"] == "analysis_ready", "revenue_paisa"].sum())
report["output_parseable_total_paisa"] = int(clean["revenue_paisa"].sum())
REPORT.write_text(json.dumps(report, indent=2), encoding="utf-8")
print(json.dumps(report, indent=2))