"""Guided capstone starter. Edit the FIVE named TODO stages below.
Read the bridge checkpoints in lesson 10 first. The ingestion helper is supplied.
Public checks run against YOUR clean_orders; no reference solution is imported.
"""
from io import StringIO
import csv
import pandas as pd
REQUIRED = ["order_id", "date", "category", "units", "unit_price_cents"]
CATEGORIES = {"notebooks", "pens", "markers"}
# Start at 1; move through 5 as each stage works. Set 6 for the full project.
CHECKPOINT_STAGE = 1
STAGE_CSV = 'order_id,date,category,units,unit_price_cents\nQ01,2026-06-01,pens,2,80\nQ01,2026-06-01,pens,2,80\n Q01 ,2026-06-01,pens,3,80\nQ02,bad-date,markers,unknown,150\nQ03,2026-06-03, NOTEBOOKS ,4,250\n'
def prepare_work(raw):
"""Stage 1. Return (work, duplicates, conflicts).
Split exact extra copies BEFORE trimming IDs/categories. Preserve indexes.
conflicts is a Boolean Series on work.index marking every repeated ID.
Checkpoint: full-schema STAGE_CSV -> duplicate label [1], conflicts [0, 2].
"""
raise NotImplementedError("Stage 1: split duplicates and flag conflicts")
def validate_work(work, conflicts):
"""Stage 2. Return (converted, reasons), preserving work.index.
Copy work; check date/numeric source spelling before conversion, then check
parsed calendar/range and numeric bounds. Include ID/category/conflict rules.
reasons is a string Series: empty for valid rows, semicolon-separated names
for failures. Preserve ALL failure reasons on a row. See bridge checkpoint 3.
"""
raise NotImplementedError("Stage 2: create aligned validation reasons")
def split_evidence(raw, converted, reasons):
"""Stage 3. Return (accepted, rejected).
Select accepted converted rows where reasons is empty. Select rejected rows
from raw using the corresponding source labels, then attach their reasons.
Convert accepted unit/price columns to int64; add revenue_cents.
Sort accepted rows by date/order_id; only now may their index be reset.
"""
raise NotImplementedError("Stage 3: retain original rejected evidence")
def summarize_accepted(accepted):
"""Stage 4. One row per category, columns category,units,revenue_cents.
Sum numeric columns and sort categories. Empty accepted input stays empty.
"""
raise NotImplementedError("Stage 4: compute category summary")
def make_audit(raw, duplicates, rejected, accepted):
"""Stage 5. Return a dict with raw_rows, duplicate_rows, rejected_rows,
accepted_rows and accepted_revenue_cents. Sum cents with Python integers.
Verify raw = duplicates + rejected + accepted, even for empty input.
"""
raise NotImplementedError("Stage 5: reconcile counts and subtotal")
def clean_orders(csv_text):
# The orchestration is supplied; the five data decisions above are yours.
raw = read_order_records(csv_text)
work, duplicates, conflicts = prepare_work(raw)
converted, reasons = validate_work(work, conflicts)
accepted, rejected = split_evidence(raw, converted, reasons)
summary = summarize_accepted(accepted)
audit = make_audit(raw, duplicates, rejected, accepted)
return accepted, rejected, duplicates, summary, audit
def preview_stage(stage):
"""Supplied driver for the full-schema five-row checkpoint; no answers."""
raw = read_order_records(STAGE_CSV)
work, duplicates, conflicts = prepare_work(raw)
print("Stage 1 working labels:", work.index.tolist())
print("Stage 1 duplicate labels:", duplicates.index.tolist())
print("Stage 1 conflicts:", conflicts.to_dict())
if stage == 1:
return
converted, reasons = validate_work(work, conflicts)
print("Stage 2 reasons:", reasons.to_dict())
if stage == 2:
return
accepted, rejected = split_evidence(raw, converted, reasons)
print("Stage 3 accepted:\n" + accepted.to_string(index=False))
print("Stage 3 rejected evidence:\n" + rejected.to_string(index=False))
if stage == 3:
return
summary = summarize_accepted(accepted)
print("Stage 4 summary:\n" + summary.to_string(index=False))
if stage == 4:
return
print("Stage 5 audit:", make_audit(raw, duplicates, rejected, accepted))
# SUPPLIED INGESTION SUPPORT. Read the structural checkpoint, then reuse this.
def read_order_records(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")
return raw
if __name__ == "__main__":
if 1 <= CHECKPOINT_STAGE <= 5:
preview_stage(CHECKPOINT_STAGE)
elif CHECKPOINT_STAGE == 6:
from pathlib import Path
root = Path(__file__).resolve().parents[1]
text = (root / "datasets" / "workshop_orders.csv").read_text(encoding="utf-8")
accepted, rejected, duplicates, summary, audit = clean_orders(text)
print("AUDIT", audit)
for label, frame in [("ACCEPTED", accepted), ("REJECTED", rejected),
("DUPLICATES", duplicates), ("SUMMARY", summary)]:
print("\n" + label + " CSV")
print(frame.to_csv(index=False))
else:
raise ValueError("CHECKPOINT_STAGE must be 1 through 6")
