feat(products): CSV codec — file-level parse + INV-18 caps (PUC-5a)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-06-11 14:58:53 -07:00
parent c39bbd4728
commit f64c3fddf9
2 changed files with 134 additions and 0 deletions
@@ -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
+    )