# 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))