feat(products): canonical serializer skeleton — header + single product row (SD-0002 §6.5.5)

backend/app/domains/products/serialize.py

@@ -0,0 +1,97 @@
+"""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). Task 1: first variant only."""
+    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
+    variant = product.variants[0]
+    row = dict(base)
+    _write_variant(row, product, variant)
+    return [row]
+
+
+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
+    for field, col in _VARIANT_FIELD_TO_COL.items():
+        if field in variant.fields:
+            row[col] = _cell(variant.fields[field])