66 lines
2.5 KiB
Python
66 lines
2.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 .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
|
|
)
|