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