feat(products): row validation — §6.5.1 rules + INV-15 sanitization
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,297 @@
|
||||
"""Row validation — ParsedFile rows → canonical products + row errors (SD-0002 §6.5.1).
|
||||
|
||||
The codec (codec.py) handles file-level gates; this module is the row-semantics
|
||||
half of the PUC-5 import spine. It groups consecutive rows sharing a Handle into
|
||||
product blocks (Shopify's grammar), normalizes product/variant/image fields, and
|
||||
records every rule violation as a merchant-language RowError. Errors never raise:
|
||||
an error poisons its whole product block (the product previews as kind="error"
|
||||
and is excluded from apply) while parsing continues so the merchant gets a
|
||||
complete accounting in one pass (BUC-1a). Description HTML is sanitized with nh3
|
||||
on the way in (INV-15).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from decimal import Decimal, InvalidOperation
|
||||
|
||||
import nh3
|
||||
|
||||
from .models import (
|
||||
COMPONENT_COLUMNS,
|
||||
OPTION_VALUE_COLUMNS,
|
||||
PRODUCT_COLUMNS,
|
||||
VARIANT_COLUMNS,
|
||||
CanonicalImage,
|
||||
CanonicalProduct,
|
||||
CanonicalVariant,
|
||||
ParsedFile,
|
||||
Row,
|
||||
RowError,
|
||||
)
|
||||
|
||||
_HANDLE_RE = re.compile(r"^[a-z0-9-]+$")
|
||||
_STATUSES = {"draft", "active", "archived"}
|
||||
# Product columns handled as attributes (title, option_names), not via fields{}.
|
||||
_ATTRIBUTE_COLUMNS = {"Title", "Option1 Name", "Option2 Name", "Option3 Name"}
|
||||
# Sentinel: the cell failed normalization (the error is already recorded).
|
||||
_INVALID = object()
|
||||
|
||||
|
||||
def build_products(parsed: ParsedFile) -> list[CanonicalProduct]:
|
||||
"""Group rows into product blocks and validate every §6.5.1 rule."""
|
||||
products: list[CanonicalProduct] = []
|
||||
closed_handles: set[str] = set()
|
||||
block: list[Row] = []
|
||||
|
||||
def flush() -> None:
|
||||
nonlocal block
|
||||
if block:
|
||||
closed_handles.add(block[0].cells["Handle"])
|
||||
products.append(_build_block(block))
|
||||
block = []
|
||||
|
||||
for row in parsed.rows:
|
||||
handle = row.cells.get("Handle", "")
|
||||
if block and handle == block[0].cells["Handle"]:
|
||||
block.append(row)
|
||||
continue
|
||||
flush()
|
||||
if not handle:
|
||||
products.append(
|
||||
_error_block(row, "(missing)", RowError(row.line_number, "Handle", "a row needs a Handle"))
|
||||
)
|
||||
elif not _HANDLE_RE.match(handle):
|
||||
products.append(
|
||||
_error_block(
|
||||
row,
|
||||
handle,
|
||||
RowError(
|
||||
row.line_number,
|
||||
"Handle",
|
||||
f"'{handle}' isn't a valid handle — lowercase letters, numbers, and dashes only",
|
||||
),
|
||||
)
|
||||
)
|
||||
elif handle in closed_handles:
|
||||
products.append(
|
||||
_error_block(
|
||||
row,
|
||||
handle,
|
||||
RowError(
|
||||
row.line_number,
|
||||
"Handle",
|
||||
f"rows for '{handle}' must be consecutive — it already appeared earlier in the file",
|
||||
),
|
||||
)
|
||||
)
|
||||
else:
|
||||
block = [row]
|
||||
flush()
|
||||
return products
|
||||
|
||||
|
||||
def all_errors(products: list[CanonicalProduct]) -> list[RowError]:
|
||||
return [error for product in products for error in product.errors]
|
||||
|
||||
|
||||
def _error_block(row: Row, handle: str, error: RowError) -> CanonicalProduct:
|
||||
return CanonicalProduct(first_line=row.line_number, handle=handle, title="", errors=[error])
|
||||
|
||||
|
||||
def _build_block(rows: list[Row]) -> CanonicalProduct:
|
||||
first = rows[0]
|
||||
handle = first.cells["Handle"]
|
||||
errors: list[RowError] = []
|
||||
|
||||
title = first.cells.get("Title", "")
|
||||
if not title:
|
||||
errors.append(RowError(first.line_number, "Title", f"'{handle}' is missing its Title"))
|
||||
|
||||
option_names = tuple(first.cells.get(f"Option{n} Name") or None for n in (1, 2, 3))
|
||||
has_options = any(option_names)
|
||||
|
||||
# Product-level fields come from the first row only; empty cell == clear (None).
|
||||
fields: dict[str, object] = {}
|
||||
for column, field_name in PRODUCT_COLUMNS.items():
|
||||
if column in _ATTRIBUTE_COLUMNS or column not in first.cells:
|
||||
continue
|
||||
cell = first.cells[column]
|
||||
if not cell:
|
||||
fields[field_name] = None
|
||||
continue
|
||||
value = _product_value(first.line_number, column, field_name, cell, errors)
|
||||
if value is not _INVALID:
|
||||
fields[field_name] = value
|
||||
|
||||
variants: list[CanonicalVariant] = []
|
||||
images: list[CanonicalImage] = []
|
||||
seen_combos: set[tuple[str | None, str | None, str | None]] = set()
|
||||
|
||||
for index, row in enumerate(rows):
|
||||
cells = row.cells
|
||||
line = row.line_number
|
||||
|
||||
for column in COMPONENT_COLUMNS:
|
||||
if cells.get(column):
|
||||
errors.append(
|
||||
RowError(line, column, "kits arrive in a coming release — leave the Component columns empty")
|
||||
)
|
||||
|
||||
has_image = bool(cells.get("Image Src"))
|
||||
if has_image:
|
||||
_collect_image(row, images, errors)
|
||||
|
||||
# A row carries a variant iff any option value / Variant-* cell is filled;
|
||||
# the first row of a no-option product always carries the single variant.
|
||||
carries_variant = (
|
||||
any(cells.get(c) for c in OPTION_VALUE_COLUMNS)
|
||||
or any(cells.get(c) for c in VARIANT_COLUMNS)
|
||||
or (index == 0 and not has_options)
|
||||
)
|
||||
if carries_variant:
|
||||
options = tuple(cells.get(c) or None for c in OPTION_VALUE_COLUMNS)
|
||||
for n in (1, 2, 3):
|
||||
value, name = options[n - 1], option_names[n - 1]
|
||||
if value and not name:
|
||||
errors.append(
|
||||
RowError(
|
||||
line,
|
||||
f"Option{n} Value",
|
||||
f"Option{n} Value given but the product has no Option{n} Name",
|
||||
)
|
||||
)
|
||||
elif name and not value:
|
||||
errors.append(
|
||||
RowError(
|
||||
line,
|
||||
f"Option{n} Value",
|
||||
f"this variant is missing its Option{n} Value ('{name}')",
|
||||
)
|
||||
)
|
||||
if not has_options:
|
||||
if variants:
|
||||
errors.append(RowError(line, None, "a product without options can have only one variant"))
|
||||
elif options in seen_combos:
|
||||
errors.append(
|
||||
RowError(line, None, f"duplicate variant — '{handle}' already has a variant with these options")
|
||||
)
|
||||
seen_combos.add(options)
|
||||
|
||||
variant_fields: dict[str, object] = {}
|
||||
for column, field_name in VARIANT_COLUMNS.items():
|
||||
if column not in cells:
|
||||
continue
|
||||
cell = cells[column]
|
||||
if not cell:
|
||||
variant_fields[field_name] = None
|
||||
continue
|
||||
value = _variant_value(line, column, field_name, cell, errors)
|
||||
if value is _INVALID:
|
||||
continue
|
||||
variant_fields[field_name] = value
|
||||
if field_name == "variant_image" and cell not in {i.source_url for i in images}:
|
||||
images.append(
|
||||
CanonicalImage(line_number=line, source_url=cell, position=len(images) + 1, alt_text=None)
|
||||
)
|
||||
variants.append(CanonicalVariant(line_number=line, options=options, fields=variant_fields))
|
||||
elif index > 0 and not has_image:
|
||||
errors.append(RowError(line, None, "this row has no variant or image data"))
|
||||
|
||||
return CanonicalProduct(
|
||||
first_line=first.line_number,
|
||||
handle=handle,
|
||||
title=title,
|
||||
option_names=option_names,
|
||||
fields=fields,
|
||||
variants=variants,
|
||||
images=images,
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
|
||||
def _collect_image(row: Row, images: list[CanonicalImage], errors: list[RowError]) -> None:
|
||||
cells = row.cells
|
||||
source_url = cells["Image Src"]
|
||||
position_cell = cells.get("Image Position", "")
|
||||
position: int | None = None
|
||||
if position_cell:
|
||||
try:
|
||||
position = int(position_cell)
|
||||
if position < 1:
|
||||
raise ValueError
|
||||
except ValueError:
|
||||
position = None
|
||||
errors.append(RowError(row.line_number, "Image Position", f"'{position_cell}' is not a position"))
|
||||
# Dedupe by source URL within the block — first occurrence wins.
|
||||
if source_url in {i.source_url for i in images}:
|
||||
return
|
||||
images.append(
|
||||
CanonicalImage(
|
||||
line_number=row.line_number,
|
||||
source_url=source_url,
|
||||
position=position if position is not None else len(images) + 1,
|
||||
alt_text=cells.get("Image Alt Text") or None,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def _product_value(line: int, column: str, field_name: str, cell: str, errors: list[RowError]) -> object:
|
||||
if field_name == "tags":
|
||||
return [tag.strip() for tag in cell.split(",") if tag.strip()]
|
||||
if field_name == "status":
|
||||
status = cell.lower()
|
||||
if status not in _STATUSES:
|
||||
errors.append(RowError(line, column, f"'{cell}' is not a status — use draft, active, or archived"))
|
||||
return _INVALID
|
||||
return status
|
||||
if field_name == "published":
|
||||
flag = cell.upper()
|
||||
if flag not in ("TRUE", "FALSE"):
|
||||
errors.append(RowError(line, column, f"'{cell}' is not TRUE or FALSE"))
|
||||
return _INVALID
|
||||
return flag == "TRUE"
|
||||
if field_name == "description_html":
|
||||
return nh3.clean(cell)
|
||||
if field_name == "product_type":
|
||||
if cell != "standalone":
|
||||
errors.append(RowError(line, column, "kits arrive in a coming release — Type must be 'standalone'"))
|
||||
return _INVALID
|
||||
return cell
|
||||
return cell
|
||||
|
||||
|
||||
def _variant_value(line: int, column: str, field_name: str, cell: str, errors: list[RowError]) -> object:
|
||||
if field_name in ("price", "cost"):
|
||||
return _decimal_or_error(line, column, cell, f"'{cell}' is not a price", errors)
|
||||
if field_name in ("weight", "volume"):
|
||||
return _decimal_or_error(line, column, cell, f"'{cell}' is not a number", errors)
|
||||
if field_name == "inventory_qty":
|
||||
try:
|
||||
quantity = int(cell)
|
||||
if quantity < 0:
|
||||
raise ValueError
|
||||
except ValueError:
|
||||
errors.append(RowError(line, column, f"'{cell}' is not a whole number"))
|
||||
return _INVALID
|
||||
return quantity
|
||||
if field_name == "position":
|
||||
try:
|
||||
position = int(cell)
|
||||
if position < 1:
|
||||
raise ValueError
|
||||
except ValueError:
|
||||
errors.append(RowError(line, column, f"'{cell}' is not a position"))
|
||||
return _INVALID
|
||||
return position
|
||||
return cell
|
||||
|
||||
|
||||
def _decimal_or_error(line: int, column: str, cell: str, message: str, errors: list[RowError]) -> object:
|
||||
try:
|
||||
value = Decimal(cell)
|
||||
if not value.is_finite() or value < 0:
|
||||
raise InvalidOperation
|
||||
except InvalidOperation:
|
||||
errors.append(RowError(line, column, message))
|
||||
return _INVALID
|
||||
return value
|
||||
Reference in New Issue
Block a user