ben.stull
af72299a57
A stored variant position is a CatalogVariant attribute, not a fields{} entry,
so the serializer emitted an empty Variant Position cell — which re-imports as
'reset to file order'. A non-sequential stored position (a merchant can import
explicit positions) then round-tripped to a spurious update, violating INV-12.
Emit str(variant.position) explicitly (like _write_image). The property-test
generator now assigns non-sequential positions to lock the regression, plus a
targeted test. Also wires isExportEnabled into ProductsPage (was dead code).
Both caught by the final code review.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
"""Canonical serializer — CatalogProduct snapshot → canonical CSV (SD-0002 §6.5.5).
|
|
|
|
The export half of "one codec, two directions": this writes exactly the columns
|
|
codec.py/validate.py parse, in a row grammar (§6.5.1) the validator regroups
|
|
identically — so re-importing an unmodified export diffs to nothing (INV-12).
|
|
DB-free, like diff.py: it consumes the same CatalogProduct snapshot the diff
|
|
engine reads (repo.load_catalog), and the service streams the result.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
from collections.abc import Iterable, Iterator
|
|
from decimal import Decimal
|
|
|
|
from .diff import CatalogProduct
|
|
from .models import (
|
|
IMAGE_COLUMNS,
|
|
OPTION_VALUE_COLUMNS,
|
|
PRODUCT_COLUMNS,
|
|
VARIANT_COLUMNS,
|
|
)
|
|
|
|
# Canonical column order: Handle, then product, option-value, variant, image
|
|
# columns. A superset of everything the parser knows (models.KNOWN_COLUMNS minus
|
|
# the #15-reserved Component columns, which export never emits).
|
|
HEADER: list[str] = [
|
|
"Handle",
|
|
*PRODUCT_COLUMNS,
|
|
*OPTION_VALUE_COLUMNS,
|
|
*VARIANT_COLUMNS,
|
|
*IMAGE_COLUMNS,
|
|
]
|
|
|
|
# field name -> column name, inverting the model registry (the parser reads
|
|
# column->field; the serializer writes field->column).
|
|
_PRODUCT_FIELD_TO_COL = {field: col for col, field in PRODUCT_COLUMNS.items()}
|
|
_VARIANT_FIELD_TO_COL = {field: col for col, field in VARIANT_COLUMNS.items()}
|
|
|
|
|
|
def _cell(value: object) -> str:
|
|
"""Serialize one value to its canonical cell text (the parse inverse)."""
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, bool):
|
|
return "TRUE" if value else "FALSE"
|
|
if isinstance(value, Decimal):
|
|
return str(value)
|
|
if isinstance(value, (list, tuple)):
|
|
return ", ".join(str(v) for v in value)
|
|
return str(value)
|
|
|
|
|
|
def catalog_to_csv(products: Iterable[CatalogProduct]) -> Iterator[str]:
|
|
"""Stream canonical CSV text, header first, one product block at a time."""
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf)
|
|
writer.writerow(HEADER)
|
|
yield _drain(buf)
|
|
for product in products:
|
|
for row in _product_rows(product):
|
|
writer.writerow([row.get(col, "") for col in HEADER])
|
|
yield _drain(buf)
|
|
|
|
|
|
def _drain(buf: io.StringIO) -> str:
|
|
text = buf.getvalue()
|
|
buf.seek(0)
|
|
buf.truncate(0)
|
|
return text
|
|
|
|
|
|
def _product_rows(product: CatalogProduct) -> list[dict[str, str]]:
|
|
"""The product's CSV rows (§6.5.1 grammar): product fields + option names on
|
|
the first row; one variant per row; images interleaved; image-only rows when
|
|
a product has more images than variants."""
|
|
base = {"Handle": product.handle, "Title": product.title}
|
|
for field, col in _PRODUCT_FIELD_TO_COL.items():
|
|
if field in product.fields:
|
|
base[col] = _cell(product.fields[field])
|
|
for slot, name in enumerate(product.option_names, start=1):
|
|
if name:
|
|
base[f"Option{slot} Name"] = name
|
|
|
|
count = max(len(product.variants), len(product.images))
|
|
rows: list[dict[str, str]] = []
|
|
for i in range(count):
|
|
# Row 0 carries the product-level fields; later rows carry only Handle.
|
|
row = dict(base) if i == 0 else {"Handle": product.handle}
|
|
if i < len(product.variants):
|
|
_write_variant(row, product, product.variants[i])
|
|
if i < len(product.images):
|
|
_write_image(row, product.images[i])
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def _write_image(row: dict[str, str], image) -> None:
|
|
row["Image Src"] = image.source_url
|
|
row["Image Position"] = str(image.position)
|
|
if image.alt_text is not None:
|
|
row["Image Alt Text"] = image.alt_text
|
|
|
|
|
|
def _write_variant(row: dict[str, str], product: CatalogProduct, variant) -> None:
|
|
"""Fill a row's option-value + variant columns for one variant."""
|
|
for slot, value in enumerate(variant.options, start=1):
|
|
# Only emit an option value when the product actually has that option
|
|
# (a no-option product's single variant carries all-NULL options).
|
|
if product.option_names[slot - 1] and value is not None:
|
|
row[f"Option{slot} Value"] = value
|
|
# position is a CatalogVariant attribute, not a fields{} entry — emit it
|
|
# explicitly. An empty Variant Position cell re-imports as "reset to file
|
|
# order", so a non-sequential stored position would round-trip to an update
|
|
# (INV-12). (Like _write_image, which emits its position attribute.)
|
|
row["Variant Position"] = str(variant.position)
|
|
for field, col in _VARIANT_FIELD_TO_COL.items():
|
|
if field in variant.fields:
|
|
row[col] = _cell(variant.fields[field])
|