From f64c3fddf96e5066dda821887b0ff4695203aee1 Mon Sep 17 00:00:00 2001 From: Ben Stull Date: Thu, 11 Jun 2026 14:58:53 -0700 Subject: [PATCH] =?UTF-8?q?feat(products):=20CSV=20codec=20=E2=80=94=20fil?= =?UTF-8?q?e-level=20parse=20+=20INV-18=20caps=20(PUC-5a)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5 --- backend/app/domains/products/codec.py | 65 +++++++++++++++++++++++++ backend/tests/test_products_codec.py | 69 +++++++++++++++++++++++++++ 2 files changed, 134 insertions(+) create mode 100644 backend/app/domains/products/codec.py create mode 100644 backend/tests/test_products_codec.py diff --git a/backend/app/domains/products/codec.py b/backend/app/domains/products/codec.py new file mode 100644 index 0000000..52ba6ec --- /dev/null +++ b/backend/app/domains/products/codec.py @@ -0,0 +1,65 @@ +"""CSV codec — bytes → ParsedFile (SD-0002 §6.5.1). File-level gates only (PUC-5a); +row semantics live in validate.py. Dialect detection is the INV-17 seam (SLICE-8 +adds Shopify).""" +from __future__ import annotations + +import csv +import io + +from .errors import FileRejected +from .models import KNOWN_COLUMNS, MAX_DATA_ROWS, MAX_FILE_BYTES, ParsedFile, Row + +_REQUIRED_HEADER_COLUMNS = ("Handle", "Title") + + +def detect_dialect(header: list[str]) -> str: + """The INV-17 seam: SLICE-8 recognizes Shopify's exact header set here.""" + return "canonical" + + +def parse_csv(data: bytes) -> ParsedFile: + if len(data) > MAX_FILE_BYTES: + raise FileRejected("file_too_large", "This file is larger than 10 MB.") + try: + text = data.decode("utf-8-sig") + except UnicodeDecodeError: + raise FileRejected("not_csv", "This file isn't readable as CSV.") from None + reader = csv.reader(io.StringIO(text)) + try: + try: + raw_header = next(reader) + except StopIteration: + raise FileRejected("not_csv", "This file isn't readable as CSV.") from None + header = [h.strip() for h in raw_header] + for col in _REQUIRED_HEADER_COLUMNS: + if col not in header: + raise FileRejected( + "missing_required_column", + f"This file is missing the required column '{col}'.", + ) + # First occurrence of a duplicated column wins. + col_index: dict[str, int] = {} + for i, name in enumerate(header): + if name and name not in col_index: + col_index[name] = i + known_present = [c for c in col_index if c in KNOWN_COLUMNS] + unknown = [c for c in col_index if c not in KNOWN_COLUMNS] + rows: list[Row] = [] + for raw in reader: + if not any(cell.strip() for cell in raw): + continue + if len(rows) >= MAX_DATA_ROWS: + raise FileRejected( + "too_many_rows", + f"This file has more than {MAX_DATA_ROWS:,} rows — split it and import in parts.", + ) + cells = { + c: (raw[col_index[c]].strip() if col_index[c] < len(raw) else "") + for c in known_present + } + rows.append(Row(line_number=reader.line_num, cells=cells)) + except csv.Error: + raise FileRejected("not_csv", "This file isn't readable as CSV.") from None + return ParsedFile( + dialect=detect_dialect(header), header=header, unknown_columns=unknown, rows=rows + ) diff --git a/backend/tests/test_products_codec.py b/backend/tests/test_products_codec.py new file mode 100644 index 0000000..4880f5c --- /dev/null +++ b/backend/tests/test_products_codec.py @@ -0,0 +1,69 @@ +"""§6.5.1 file-level codec — parse, caps, required columns (PUC-5a fixtures).""" +import pytest + +from app.domains.products import FileRejected +from app.domains.products.codec import parse_csv + + +def _csv(*lines: str) -> bytes: + return ("\n".join(lines) + "\n").encode() + + +GOOD = _csv( + "Handle,Title,Variant Price,Bogus Column", + "moon-mug,Moon Mug,18.00,x", + "star-tee,Star Tee,24.00,y", +) + + +def test_parses_header_rows_and_unknown_columns(): + parsed = parse_csv(GOOD) + assert parsed.dialect == "canonical" + assert parsed.unknown_columns == ["Bogus Column"] + assert [r.line_number for r in parsed.rows] == [2, 3] + assert parsed.rows[0].cells["Handle"] == "moon-mug" + assert parsed.rows[0].cells["Variant Price"] == "18.00" + assert "Bogus Column" not in parsed.rows[0].cells + + +def test_bom_tolerated(): + parsed = parse_csv(b"\xef\xbb\xbf" + GOOD) + assert parsed.rows[0].cells["Handle"] == "moon-mug" + + +def test_quoted_cells_rfc4180(): + parsed = parse_csv(_csv("Handle,Title,Tags", 'mug,"The ""Best"" Mug","a, b"')) + assert parsed.rows[0].cells["Title"] == 'The "Best" Mug' + assert parsed.rows[0].cells["Tags"] == "a, b" + + +def test_empty_rows_skipped_short_rows_padded(): + parsed = parse_csv(_csv("Handle,Title,Vendor", "mug,Mug", "", ",,", "tee,Tee,Acme")) + assert [r.cells["Handle"] for r in parsed.rows] == ["mug", "tee"] + assert parsed.rows[0].cells["Vendor"] == "" + + +@pytest.mark.parametrize( + "data,code", + [ + (b"\xff\xfe\x00garbage\x00", "not_csv"), + (b"", "not_csv"), + (_csv("Title,Vendor", "Mug,Acme"), "missing_required_column"), + (_csv("Handle,Vendor", "mug,Acme"), "missing_required_column"), + ( + _csv("Handle,Title", *(f"h{i},T{i}" for i in range(5001))), + "too_many_rows", + ), + (b"Handle,Title\n" + b"x" * (10 * 1024 * 1024), "file_too_large"), + ], +) +def test_file_level_rejections(data, code): + with pytest.raises(FileRejected) as exc: + parse_csv(data) + assert exc.value.code == code + + +def test_missing_column_message_names_the_column(): + with pytest.raises(FileRejected) as exc: + parse_csv(_csv("Handle,Vendor", "mug,Acme")) + assert "'Title'" in exc.value.message