feat(products): diff engine — add/update/unchanged/error + INV-11 fingerprint
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,207 @@
|
||||
"""Diff engine — catalog × canonical products → preview records (SD-0002 §6.5.2).
|
||||
|
||||
Classifies each canonical product against the storefront's current catalog as
|
||||
add / update / unchanged / error, producing JSON-ready preview records (stored
|
||||
as draft JSONB, served verbatim to the SPA) plus a summary and a deterministic
|
||||
fingerprint that detects catalog drift between preview and confirm (INV-11).
|
||||
|
||||
Deliberately DB-free: the catalog snapshot dataclasses are defined here and the
|
||||
repo layer builds them. Only fields present in the file's canonical fields{}
|
||||
participate in a comparison — an absent column is untouched, never a change
|
||||
(§6.5.1); a present-but-empty cell resolves to the field's CLEAR_DEFAULTS entry.
|
||||
Catalog variants/images absent from the file are likewise untouched (INV-10).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
|
||||
from .models import CLEAR_DEFAULTS, CanonicalProduct, CanonicalVariant
|
||||
|
||||
|
||||
@dataclass
|
||||
class CatalogVariant:
|
||||
id: int
|
||||
options: tuple[str | None, str | None, str | None]
|
||||
position: int
|
||||
# sku, barcode, price (Decimal|None), cost, weight, weight_unit, volume,
|
||||
# volume_unit, tax_id_1, tax_id_2, inventory_tracker, inventory_qty,
|
||||
# variant_image (linked image source_url or None)
|
||||
fields: dict[str, object]
|
||||
|
||||
|
||||
@dataclass
|
||||
class CatalogImage:
|
||||
id: int
|
||||
source_url: str
|
||||
position: int
|
||||
alt_text: str | None
|
||||
|
||||
|
||||
@dataclass
|
||||
class CatalogProduct:
|
||||
id: int
|
||||
handle: str
|
||||
title: str
|
||||
option_names: tuple[str | None, str | None, str | None]
|
||||
# title, description_html, vendor, product_type, google_product_category,
|
||||
# tags (list[str]), status, published (bool)
|
||||
fields: dict[str, object]
|
||||
variants: list[CatalogVariant]
|
||||
images: list[CatalogImage]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class DiffResult:
|
||||
records: list[dict]
|
||||
summary: dict
|
||||
fingerprint: str
|
||||
|
||||
|
||||
# Option-name presence markers in fields{} (see validate.py); the values are
|
||||
# compared via the option_names attribute, not through the generic field loop.
|
||||
_OPTION_NAME_FIELDS = ("option1_name", "option2_name", "option3_name")
|
||||
_SUMMARY_KEY = {"add": "adds", "update": "updates", "unchanged": "unchanged", "error": "errors"}
|
||||
|
||||
|
||||
def compute_diff(catalog: dict[str, CatalogProduct], products: list[CanonicalProduct]) -> DiffResult:
|
||||
records: list[dict] = []
|
||||
summary = {"adds": 0, "updates": 0, "unchanged": 0, "errors": 0}
|
||||
for product in products:
|
||||
current = catalog.get(product.handle)
|
||||
if product.errors:
|
||||
kind, detail = "error", {"errors": [e.as_json() for e in product.errors]}
|
||||
elif current is None:
|
||||
kind, detail = "add", _add_detail(product)
|
||||
else:
|
||||
detail = _update_detail(product, current)
|
||||
kind = "update" if detail else "unchanged"
|
||||
records.append(
|
||||
{
|
||||
"handle": product.handle,
|
||||
"title": product.title or (current.title if current else ""),
|
||||
"kind": kind,
|
||||
"variant_count": len(current.variants) if kind == "unchanged" else len(product.variants),
|
||||
"detail": detail,
|
||||
}
|
||||
)
|
||||
summary[_SUMMARY_KEY[kind]] += 1
|
||||
fingerprint = hashlib.sha256(
|
||||
json.dumps(records, sort_keys=True, separators=(",", ":"), default=str).encode()
|
||||
).hexdigest()
|
||||
return DiffResult(records=records, summary=summary, fingerprint=fingerprint)
|
||||
|
||||
|
||||
def _add_detail(product: CanonicalProduct) -> dict:
|
||||
# On add, absent fields fall back to their defaults for display where one
|
||||
# exists — the detail shows what will actually be set.
|
||||
set_fields = _resolved_product_fields(product)
|
||||
for field_name, default in CLEAR_DEFAULTS.items():
|
||||
set_fields.setdefault(field_name, default)
|
||||
return {
|
||||
"set": {f: _json_safe(v) for f, v in set_fields.items()},
|
||||
"option_names": list(product.option_names),
|
||||
"variants": [
|
||||
{"options": list(v.options), "set": _resolved_variant_fields(v)}
|
||||
for v in product.variants
|
||||
],
|
||||
"images": [
|
||||
{"src": i.source_url, "position": i.position, "alt_text": i.alt_text}
|
||||
for i in product.images
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _update_detail(product: CanonicalProduct, current: CatalogProduct) -> dict:
|
||||
changes: list[dict] = []
|
||||
# Title is always file-present (required header column); "" means the block
|
||||
# already carries an error and never reaches here.
|
||||
if product.title and product.title != current.title:
|
||||
changes.append(_change("title", current.title, product.title))
|
||||
for field_name, resolved in _resolved_product_fields(product).items():
|
||||
before = current.fields.get(field_name)
|
||||
if resolved != before:
|
||||
changes.append(_change(field_name, before, resolved))
|
||||
for slot in (1, 2, 3):
|
||||
if f"option{slot}_name" not in product.fields:
|
||||
continue
|
||||
before, after = current.option_names[slot - 1], product.option_names[slot - 1]
|
||||
if after != before:
|
||||
changes.append(_change(f"option{slot}_name", before, after))
|
||||
|
||||
variant_entries: list[dict] = []
|
||||
by_options = {v.options: v for v in current.variants}
|
||||
for variant in product.variants:
|
||||
match = by_options.get(variant.options)
|
||||
if match is None:
|
||||
variant_entries.append(
|
||||
{"options": list(variant.options), "kind": "add", "set": _resolved_variant_fields(variant)}
|
||||
)
|
||||
continue
|
||||
variant_changes = [
|
||||
_change(f, match.fields.get(f), resolved)
|
||||
for f, resolved in _resolved_fields(variant.fields).items()
|
||||
if resolved != match.fields.get(f)
|
||||
]
|
||||
if variant_changes:
|
||||
variant_entries.append(
|
||||
{"options": list(variant.options), "kind": "update", "changes": variant_changes}
|
||||
)
|
||||
|
||||
image_entries: list[dict] = []
|
||||
by_src = {i.source_url: i for i in current.images}
|
||||
for image in product.images:
|
||||
match = by_src.get(image.source_url)
|
||||
if match is None:
|
||||
image_entries.append(
|
||||
{"src": image.source_url, "kind": "add", "position": image.position, "alt_text": image.alt_text}
|
||||
)
|
||||
continue
|
||||
image_changes = [
|
||||
_change(f, before, after)
|
||||
for f, before, after in (
|
||||
("position", match.position, image.position),
|
||||
("alt_text", match.alt_text, image.alt_text),
|
||||
)
|
||||
if after != before
|
||||
]
|
||||
if image_changes:
|
||||
image_entries.append({"src": image.source_url, "kind": "update", "changes": image_changes})
|
||||
|
||||
detail: dict = {}
|
||||
if changes:
|
||||
detail["changes"] = changes
|
||||
if variant_entries:
|
||||
detail["variants"] = variant_entries
|
||||
if image_entries:
|
||||
detail["images"] = image_entries
|
||||
return detail
|
||||
|
||||
|
||||
def _resolved_fields(fields: dict[str, object]) -> dict[str, object]:
|
||||
"""File-present fields with None (an explicit clear) resolved to the default."""
|
||||
return {f: (v if v is not None else CLEAR_DEFAULTS.get(f)) for f, v in fields.items()}
|
||||
|
||||
|
||||
def _resolved_product_fields(product: CanonicalProduct) -> dict[str, object]:
|
||||
return {
|
||||
f: v for f, v in _resolved_fields(product.fields).items() if f not in _OPTION_NAME_FIELDS
|
||||
}
|
||||
|
||||
|
||||
def _resolved_variant_fields(variant: CanonicalVariant) -> dict[str, object]:
|
||||
return {f: _json_safe(v) for f, v in _resolved_fields(variant.fields).items()}
|
||||
|
||||
|
||||
def _change(field_name: str, before: object, after: object) -> dict:
|
||||
return {"field": field_name, "before": _json_safe(before), "after": _json_safe(after)}
|
||||
|
||||
|
||||
def _json_safe(value: object) -> object:
|
||||
if isinstance(value, Decimal):
|
||||
return str(value)
|
||||
if isinstance(value, (tuple, list)):
|
||||
return [_json_safe(v) for v in value]
|
||||
return value
|
||||
Reference in New Issue
Block a user