"""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
# SUPPLIED PUBLIC CHECKS. These call your function with new toy inputs.
HEADER = "order_id,date,category,units,unit_price_cents\n"
def check_submission(clean_orders):
"""Call the function passed by the learner on fresh, public test fixtures."""
count = 0
def inspect(text):
result = clean_orders(text)
assert isinstance(result, tuple) and len(result) == 5, "Return the five-item tuple"
accepted, rejected, duplicates, summary, audit = result
assert all(isinstance(frame, pd.DataFrame) for frame in result[:4]), "First four results must be tables"
assert isinstance(audit, dict), "The last result must be an audit dictionary"
assert audit["raw_rows"] == len(accepted) + len(rejected) + len(duplicates), "Counts do not reconcile"
assert audit["accepted_rows"] == len(accepted)
assert audit["rejected_rows"] == len(rejected)
assert audit["duplicate_rows"] == len(duplicates)
assert audit["accepted_revenue_cents"] == sum(int(x) for x in accepted["revenue_cents"])
return result
accepted, rejected, duplicates, summary, audit = inspect(
HEADER+"T01,2026-05-01,pens,2,80\nT02,2026-05-02,markers,3,150\n")
assert audit["accepted_revenue_cents"] == 610 and len(accepted) == 2
paired = dict(zip(summary["category"], summary["revenue_cents"]))
assert paired == {"pens": 160, "markers": 450}, "Match category labels to totals"
count += 1
*_, audit = inspect(HEADER+"T03,2026-05-03,pens,9,90\n")
assert audit["accepted_revenue_cents"] == 810, "Changed data must change the total"
count += 1
row = "T04,2026-05-04,pens,1,80\n"
accepted, rejected, duplicates, _, audit = inspect(HEADER+row+row)
assert (len(accepted),len(rejected),len(duplicates)) == (1,0,1)
assert duplicates.iloc[0]["order_id"] == "T04"
count += 1
accepted, rejected, _, _, _ = inspect(HEADER+row+" T04 ,2026-05-04,pens,2,80\n")
assert accepted.empty and len(rejected) == 2
assert rejected["reason"].str.contains("conflicting_order_id").all()
count += 1
accepted, rejected, _, _, _ = inspect(HEADER+" T07 ,bad-date, PENS ,unknown,080\n")
assert accepted.empty and len(rejected) == 1
evidence = rejected.iloc[0]
assert evidence["order_id"] == " T07 " and evidence["category"] == " PENS "
assert evidence["units"] == "unknown" and evidence["unit_price_cents"] == "080"
assert {"invalid_date", "invalid_units"} <= set(evidence["reason"].split(";"))
count += 1
accepted, rejected, _, _, audit = inspect(HEADER)
assert accepted.empty and rejected.empty and audit["raw_rows"] == 0
count += 1
for row in ["extra,T08,2026-05-01,pens,1,80\n", 'T08,2026-05-01,pens,1,"80\n']:
try:
clean_orders(HEADER+row)
except ValueError:
pass
else:
raise AssertionError("Structurally malformed CSV must stop ingestion")
count += 1
for row in ["T09,2026-05-01,pens,1.0000000000000001,80\n", "T09,2500-05-01,pens,1,80\n"]:
accepted, rejected, _, _, _ = inspect(HEADER+row)
assert accepted.empty and len(rejected) == 1
count += 1
accepted, _, _, _, audit = inspect(HEADER+"T10,2000-01-01,pens,001,0\n")
assert len(accepted) == 1 and audit["accepted_revenue_cents"] == 0
count += 1
print(f"{count} public capstone contract checks passed for your supplied function")
print("Next: explain one new input, inspect exports and chart, and use the rubric.")
return count
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'
# Keep definitions plus this driver in one browser run.
if 1 <= CHECKPOINT_STAGE <= 5:
preview_stage(CHECKPOINT_STAGE)
elif CHECKPOINT_STAGE == 6:
check_submission(clean_orders)
accepted, rejected, duplicates, summary, audit = clean_orders(CSV_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")
