Capstone

"""An explicit small-data cleaning policy; never changes the raw input.

Accepted-record subtotal only. This is an instructional pipeline, not a
financial reporting system. Run `python -m solutions.capstone` from course root.
"""
from io import StringIO
import csv
from pathlib import Path
import pandas as pd
import matplotlib.pyplot as plt

REQUIRED = ["order_id", "date", "category", "units", "unit_price_cents"]
CATEGORIES = {"notebooks", "pens", "markers"}


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


def plot_summary(summary):
    fig, ax = plt.subplots(figsize=(7, 4.4))
    fig.subplots_adjust(bottom=0.24, top=0.86, left=0.12, right=0.97)
    ax.bar(summary["category"], summary["revenue_cents"] / 100, color="#0d9488")
    ax.set(title="Accepted-order subtotal by category (synthetic)",
           xlabel="Category", ylabel="Fictional currency units")
    largest = max((float(value) / 100 for value in summary["revenue_cents"]), default=0)
    ax.set_ylim(0, max(1.0, largest * 1.2))
    if summary.empty:
        ax.text(0.5, 0.5, "No accepted records", transform=ax.transAxes, ha="center")
    fig.text(0.02, 0.045, "Rejected records are excluded.\nThis accepted subtotal is not a complete sales total.", fontsize=8.5)
    return fig


def main():
    root = Path(__file__).resolve().parents[1]
    csv_text = (root / "datasets" / "workshop_orders.csv").read_text(encoding="utf-8")
    accepted, rejected, duplicates, summary, audit = clean_orders(csv_text)
    output = root / "outputs"
    output.mkdir(exist_ok=True)
    for name, frame in [("accepted_orders", accepted), ("rejected_orders", rejected),
                        ("duplicate_orders", duplicates), ("category_summary", summary)]:
        frame.to_csv(output / f"{name}.csv", index=False)
    fig = plot_summary(summary)
    fig.savefig(output / "accepted_subtotal.png", dpi=160)
    plt.close(fig)
    print(audit)
    print(summary.to_string(index=False))


if __name__ == "__main__":
    main()