# Profile: Data · NumPy and pandas. No local files required.
from io import StringIO
import csv
import pandas as pd
REQUIRED = ["order_id", "date", "category", "units", "unit_price_cents"]
CATEGORIES = {"notebooks", "pens", "markers"}
CSV_TEXT = 'order_id,date,category,units,unit_price_cents\nO001,2026-04-01,notebooks,3,250\nO002,2026-04-01,pens,10,80\nO003,2026-04-02,notebooks,2,250\nO004,2026-04-02,markers,4,150\nO004,2026-04-02,markers,4,150\nO005,2026-04-03,pens,,80\nO006,2026-04-03,markers,-2,150\nO007,2026-04-04,notebooks,1,unknown\nO008,not-a-date,pens,5,80\nO009,2026-04-04, PENS ,6,80\nO010,2026-04-05,,2,150\nO011,2026-04-05,notebooks,5,250\nO012,2026-04-06,markers,3,150\nO013,2026-04-06,pens,7,80\nO014,2026-04-07,notebooks,4,250\nO015,2026-04-07,markers,2,150\n'
def clean_orders(csv_text):
# Parse strictly before constructing a DataFrame. Some CSV readers can
# reinterpret an extra field as an index; that would lose source evidence.
try:
records = list(csv.reader(StringIO(csv_text), strict=True))
except csv.Error as error:
raise ValueError(f"malformed CSV quoting: {error}") from error
if not records:
raise ValueError("empty CSV: required header is missing")
header = records[0]
missing = set(REQUIRED) - set(header)
if missing:
raise ValueError(f"missing columns: {sorted(missing)}")
if len(header) != len(REQUIRED) or set(header) != set(REQUIRED):
raise ValueError("unexpected or duplicate columns")
if len(records) - 1 > 1000:
raise ValueError("this teaching pipeline supports at most 1000 rows")
for position, record in enumerate(records[1:], start=2):
if len(record) != len(header):
raise ValueError(
f"CSV record {position} has {len(record)} fields; "
f"expected {len(header)}. Original fields: {record!r}"
)
# No indexing inference: every validated source field becomes a table cell.
raw = pd.DataFrame(records[1:], columns=header, dtype="string")
# Exact duplicates compare source field values, before normalization.
duplicate_mask = raw.duplicated(keep="first")
duplicates = raw.loc[duplicate_mask].copy()
work = raw.loc[~duplicate_mask].copy()
work["order_id"] = work["order_id"].str.strip()
work["category"] = work["category"].str.strip().str.lower()
# A same-ID disagreement is not permission to choose a winner.
conflict_mask = work["order_id"].duplicated(keep=False)
iso_date_shape = work["date"].str.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}")
work["date"] = pd.to_datetime(work["date"], format="%Y-%m-%d", errors="coerce")
integer_syntax = {}
for column in ["units", "unit_price_cents"]:
numeric_text = work[column].str.strip()
integer_syntax[column] = numeric_text.str.fullmatch(r"[0-9]+")
work[column] = pd.to_numeric(numeric_text, errors="coerce")
valid_units = (integer_syntax["units"] & work["units"].notna() & work["units"].gt(0)
& work["units"].le(1_000_000) & work["units"].mod(1).eq(0))
valid_price = (integer_syntax["unit_price_cents"] & work["unit_price_cents"].notna() & work["unit_price_cents"].ge(0)
& work["unit_price_cents"].le(1_000_000_000)
& work["unit_price_cents"].mod(1).eq(0))
# Upper bounds are an explicit teaching contract, keeping int64 products safe.
checks = {
"missing_order_id": work["order_id"].eq(""),
"conflicting_order_id": conflict_mask,
"invalid_date": (work["date"].isna() | ~iso_date_shape
| ~work["date"].between("2000-01-01", "2099-12-31")),
"invalid_category": ~work["category"].isin(CATEGORIES),
"invalid_units": ~valid_units,
"invalid_price": ~valid_price,
}
reasons = pd.Series("", index=work.index, dtype="string")
for reason, mask in checks.items():
reasons.loc[mask] = reasons.loc[mask] + reason + ";"
valid = reasons.eq("")
rejected = raw.loc[work.index[~valid]].copy()
rejected["reason"] = reasons.loc[~valid].str.rstrip(";")
accepted = work.loc[valid].copy()
for column in ["units", "unit_price_cents"]:
accepted[column] = accepted[column].astype("int64")
accepted["revenue_cents"] = accepted["units"] * accepted["unit_price_cents"]
accepted = accepted.sort_values(["date", "order_id"]).reset_index(drop=True)
summary = (accepted.groupby("category", as_index=False)[["units", "revenue_cents"]]
.sum().sort_values("category").reset_index(drop=True))
# Python ints avoid fixed-width overflow in the overall audit subtotal.
subtotal = sum(int(value) for value in accepted["revenue_cents"])
audit = {"raw_rows": len(raw), "duplicate_rows": len(duplicates),
"rejected_rows": len(rejected), "accepted_rows": len(accepted),
"accepted_revenue_cents": subtotal}
return accepted, rejected, duplicates, summary, audit
accepted, rejected, duplicates, summary, audit = clean_orders(CSV_TEXT)
assert audit == {"raw_rows":16,"duplicate_rows":1,"rejected_rows":5,"accepted_rows":10,"accepted_revenue_cents":6690}
print("AUDIT", audit)
for name, frame in [("ACCEPTED", accepted), ("REJECTED", rejected), ("DUPLICATES", duplicates), ("SUMMARY", summary)]:
print("\n" + name + " CSV")
print(frame.to_csv(index=False))
