83 lines
3.5 KiB
Python
83 lines
3.5 KiB
Python
"""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 .dialect_shopify import is_shopify_header, map_shopify_header
|
|
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: recognize Shopify's header set, else canonical (§6.5.1)."""
|
|
return "shopify" if is_shopify_header(header) else "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]
|
|
dialect = detect_dialect(header)
|
|
|
|
# INV-17: normalize the header to canonical names at the boundary. mapped[i]
|
|
# is the canonical name for header[i] (or None when that column has no
|
|
# canonical home); unknown is the not-imported warning list. Canonical files
|
|
# map to themselves; unknown columns are warned exactly as before.
|
|
if dialect == "shopify":
|
|
mapped, unknown = map_shopify_header(header)
|
|
else:
|
|
mapped = [c if c in KNOWN_COLUMNS else None for c in header]
|
|
unknown = [c for c in header if c and c not in KNOWN_COLUMNS]
|
|
|
|
for col in _REQUIRED_HEADER_COLUMNS:
|
|
if col not in mapped:
|
|
raise FileRejected(
|
|
"missing_required_column",
|
|
f"This file is missing the required column '{col}'.",
|
|
)
|
|
|
|
# First occurrence of a duplicated canonical column wins.
|
|
col_index: dict[str, int] = {}
|
|
for i, name in enumerate(mapped):
|
|
if name and name not in col_index:
|
|
col_index[name] = i
|
|
# De-dup the warning list, order-preserving.
|
|
seen: set[str] = set()
|
|
unknown = [c for c in unknown if not (c in seen or seen.add(c))]
|
|
|
|
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 col_index
|
|
}
|
|
# Shopify grams carry an implicit unit; canonical needs it explicit (§6.5.1).
|
|
if dialect == "shopify" and cells.get("Variant Weight"):
|
|
cells["Variant Weight Unit"] = "g"
|
|
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=dialect, header=header, unknown_columns=unknown, rows=rows)
|