← back to Fentucci Naturals
scripts/lib/weight_guard.py
298 lines
#!/usr/bin/env python3
"""weight_guard.py — TK-11471 PYTHON port of ~/Projects/designerwallcoverings/scripts/lib/weight-guard.mjs
Steve's rule (TK-11414): NO product may go ACTIVE with a missing/zero product WEIGHT
(zero weight collapses orders into the lowest weight tier / free-shipping band and
mis-costs DW freight).
Why a Python twin exists at all: this pipeline is PYTHON + REST and lives OUTSIDE
scripts/*-onboard/, so every .js/.mjs-only or GraphQL-only sweep of this lineage
structurally misses it. Constants are kept BYTE-IDENTICAL to the .mjs so the two
implementations cannot drift into disagreeing about what a compliant weight is.
Two call sites:
1. create payload: weight = resolve_weight_lb(variant, product) (unit POUNDS)
2. before activate: r = heal_and_verify_weights(gql, gid, product)
if not r["ok"]: HOLD as draft (never flip ACTIVE at zero weight)
SAMPLES COUNT. dw-active-weight-canary FAILs on ANY zero-weight ACTIVE variant —
its last live run split the offenders 43 sample / 40 sellable. The .mjs's
zeroWeightBlockers() filters samples OUT and is DEPRECATED there; that bug is
deliberately NOT ported. all_zero_weight_variants() is the only population view here.
"""
import math
SAMPLE_WEIGHT_LB = 0.25
FALLBACK_LB = 2.0
# product_type -> sellable default weight (POUNDS). Identical to the .mjs table.
TYPE_DEFAULT_LB = {
'Wallcovering': 3.0, 'Wallcoverings': 3.0, 'Wallpaper': 3.0,
'Metallic Wallcovering': 3.0, 'Commercial Wallcovering': 3.0,
'Mural': 4.0,
'Fabric': 1.0, 'Commercial Fabric': 1.0, 'Commercial Drapery': 1.0,
'Trim': 0.5, 'Acoustic Panel': 6.0, 'Pillow': 1.5,
'Upholstered Walls/Panels': 6.0, 'Tin Ceiling Tile': 2.0,
'Hardware': 1.0, 'Furniture': 15.0, 'Memo Sample': 0.25,
}
NAN = float('nan')
def _norm(t):
return ('' if t is None else str(t)).strip().lower()
def _num(v):
"""float(v), or NaN when unparseable — mirrors JS Number() -> NaN. NaN is
treated as zero-weight downstream, so a junk value FAILS SAFE."""
try:
f = float(v)
except (TypeError, ValueError):
return NAN
return f
def _d(x):
return x if isinstance(x, dict) else {}
def _gql_weight(variant):
"""inventoryItem.measurement.weight from a GraphQL node (camel or snake key)."""
iv = _d(variant).get('inventoryItem')
if iv is None:
iv = _d(variant).get('inventory_item')
return _d(_d(iv).get('measurement')).get('weight')
def is_sample_variant(variant=None):
"""Sample variant? (importers create `Sample` @ $4.25 + the real unit)."""
v = _d(variant)
label = _norm(v.get('title') if v.get('title') is not None else v.get('option1'))
sku = _norm(v.get('sku'))
if 'sample' in label or 'memo' in label:
return True
if sku.endswith('-sample') or 'sample' in sku:
return True
p = _num(v.get('price'))
return math.isfinite(p) and abs(p - 4.25) < 0.01
def weight_field_present(variant=None):
"""Did the payload we were handed CARRY a weight field at all?
TK-11431 amendment 1 — an UNMEASURED input is never a PASS. A response that
simply never contained weight must not be read as 'weight is zero' by luck,
nor as 'fine'. has_zero_weight() flags both, but a caller that needs to
distinguish 'measured 0' from 'never measured' asks here.
"""
v = _d(variant)
if _d(_gql_weight(v)).get('value') is not None:
return True
return v.get('grams') is not None or v.get('weight') is not None
def current_weight_lb(variant=None):
"""Current weight in POUNDS, or 0 if missing. Accepts variant.weight (+weight_unit,
REST), variant.grams, or inventoryItem.measurement.weight{value,unit} (GraphQL)."""
v = _d(variant)
g = _d(_gql_weight(v))
if g.get('value') is not None:
val = _num(g.get('value'))
u = _norm(g.get('unit')) # Shopify WeightUnit enum
if u == 'kilograms' or u.startswith('kg'):
return val * 2.20462
if u == 'grams' or u == 'g':
return val / 453.59237
if u == 'ounces' or u == 'oz':
return val / 16
return val # POUNDS (unit-less => assume lb)
if v.get('grams') is not None:
return _num(v.get('grams')) / 453.59237
if v.get('weight') is not None:
val = _num(v.get('weight'))
u = _norm(v.get('weight_unit') or 'lb')
if u.startswith('kg'):
return val * 2.20462
if u == 'g' or u.startswith('gram'):
return val / 453.59237
if u == 'oz':
return val / 16
return val # lb
return 0.0
def has_zero_weight(variant=None):
w = current_weight_lb(variant)
return (not math.isfinite(w)) or w <= 0
def default_weight_lb(variant=None, product=None):
"""The default weight (lb) to assign a variant that has none."""
if is_sample_variant(variant):
return SAMPLE_WEIGHT_LB
p = _d(product)
ptype = p.get('productType') if p.get('productType') is not None else p.get('product_type')
return TYPE_DEFAULT_LB.get(ptype, FALLBACK_LB)
def resolve_weight_lb(variant=None, product=None):
"""THE CREATE-SIDE GUARD: keep a real positive weight, else fill the default.
Drop-in for the REST variant `weight` field (returns POUNDS; pair with
weight_unit='lb')."""
w = current_weight_lb(variant)
return w if (math.isfinite(w) and w > 0) else default_weight_lb(variant, product)
def variants_of(product=None):
"""Normalize variants out of either a GraphQL connection or a plain REST array."""
p = _d(product)
edges = _d(p.get('variants')).get('edges')
if isinstance(edges, list):
return [e.get('node') for e in edges if isinstance(e, dict)]
v = p.get('variants')
return v if isinstance(v, list) else []
def all_zero_weight_variants(product=None):
"""THE ACTIVATE-SIDE GUARD: EVERY zero-weight variant, SAMPLE INCLUDED.
Matches dw-active-weight-canary exactly. Non-empty => DO NOT flip ACTIVE."""
return [v for v in variants_of(product) if has_zero_weight(v)]
def unmeasured_variants(product=None):
"""Variants whose payload carried NO weight field at all — the response never
measured them. Returned separately so a caller can report NOT-MEASURED rather
than claiming a clean zero reading."""
return [v for v in variants_of(product) if not weight_field_present(v)]
def variant_label(variant=None):
v = _d(variant)
return v.get('sku') or v.get('title') or v.get('id') or '?'
# ---------------------------------------------------------------- activate gate
# The re-query an activate site MUST run so the guard MEASURES something. A product
# query that omits inventoryItem{measurement{weight}} makes every variant look
# zero-weight to current_weight_lb, and one that omits productType silently defaults
# every heal to FALLBACK_LB. Both fields are required.
WEIGHT_REQUERY = (
"query($id:ID!){ product(id:$id){ productType variants(first:100){edges{node{ "
"id sku title price inventoryItem{ id measurement{ weight{ value unit } } } }}} } }"
)
M_WEIGHT_SET = (
"mutation($id:ID!,$w:Float!){ inventoryItemUpdate(id:$id, "
"input:{measurement:{weight:{value:$w, unit:POUNDS}}}){ userErrors{message} } }"
)
def response_carries_weight_field(product=None):
"""Did the re-query response ACTUALLY contain the weight field we asked for?
A gate that reads a response which never carried weight silently passes
everything (or, here, fails everything) for the wrong reason — the false-green
class CLAUDE.md warns about. Every variant node must carry an `inventoryItem`
key holding a `measurement` key. `measurement.weight = null` is a legitimate
'no weight set' reading and is NOT a shape failure.
"""
vs = variants_of(product)
if not vs:
return False
for v in vs:
d = _d(v)
iv = d.get('inventoryItem', d.get('inventory_item'))
if not isinstance(iv, dict) or 'measurement' not in iv:
return False
return True
def heal_and_verify_weights(gql, product_gid, product=None, requery=None, mutation=None,
rest_heal=None):
"""SELF-HEAL then VERIFY — the pattern Steve approved in
sanderson-onboard/scripts/create_sdg.mjs (8d09eed). Stranding product is worse
than assigning the already-approved default, but a heal that silently fails must
NEVER activate.
1. every zero-weight variant (SAMPLE INCLUDED) is written default_weight_lb() in POUNDS
2. the product is RE-READ and re-checked — the mutation's own 200 is not evidence
3. ok is False => caller must HOLD the product as draft and name `weight>0`
Idempotent and a NO-OP (zero network calls) when every weight is already positive.
Fails CLOSED: an unreadable re-query, a response that never carried the weight
field, a missing inventoryItem id, or a userError all yield ok=False.
gql -- the call site's own gql(query, variables) -> parsed JSON
rest_heal -- optional fallback heal fn (variant_dict, lb) -> True/False, used when
the GraphQL inventoryItemUpdate path is unavailable (e.g. a token
without write_inventory). Verification stays GraphQL-side either way.
"""
requery = requery or WEIGHT_REQUERY
mutation = mutation or M_WEIGHT_SET
errs, healed, still_zero = [], [], []
heal_failures = 0
p = _d(product)
product_type = p.get('productType') if p.get('productType') is not None else p.get('product_type')
zero = all_zero_weight_variants(p)
if not zero:
return {"ok": True, "healed": healed, "stillZero": [], "errs": errs} # no-op
for v in zero:
label = variant_label(v)
lb = default_weight_lb(v, {"productType": product_type})
iid = _d(_d(v).get('inventoryItem') or _d(v).get('inventory_item')).get('id')
wrote = False
if iid:
try:
r = gql(mutation, {"id": iid, "w": lb})
ue = (((r or {}).get('data') or {}).get('inventoryItemUpdate') or {}).get('userErrors') or []
top = (r or {}).get('errors') or []
for e in ue:
errs.append("weight:%s:%s" % (label, e.get('message')))
for e in top:
errs.append("weight:%s:%s" % (label, str(e.get('message'))[:80]))
wrote = not ue and not top
except Exception as e: # noqa: BLE001 - fail closed
errs.append("weight:%s:%s" % (label, str(e)[:80]))
else:
errs.append("weight:no-inventory-item:%s" % label)
if not wrote and rest_heal is not None:
try:
wrote = bool(rest_heal(v, lb))
except Exception as e: # noqa: BLE001 - fail closed
errs.append("weight:rest:%s:%s" % (label, str(e)[:80]))
if wrote:
healed.append({"sku": label, "lb": lb})
else:
heal_failures += 1
# RE-VERIFY against the live record. Never trust the write.
fresh = None
try:
resp = gql(requery, {"id": product_gid})
if (resp or {}).get('errors'):
errs.append("weight:reverify:%s" % str(resp['errors'])[:80])
fresh = ((resp or {}).get('data') or {}).get('product')
except Exception as e: # noqa: BLE001 - fail closed
errs.append("weight:reverify:%s" % str(e)[:80])
if not fresh:
errs.append("weight:reverify-failed")
return {"ok": False, "healed": healed, "stillZero": [], "errs": errs}
if not response_carries_weight_field(fresh):
# NOT MEASURED. The re-query came back without the field it asked for, so a
# clean-looking read proves nothing. Never a PASS.
errs.append("weight:reverify-response-carried-no-weight-field")
return {"ok": False, "healed": healed, "stillZero": [], "errs": errs}
still_zero = [variant_label(v) for v in all_zero_weight_variants(fresh)]
# FAIL CLOSED on an UNHEALED variant even when the re-verify comes back clean: a
# variant we could not write is UNMEASURED with respect to our own action, and a
# clean re-verify that happens to disagree is not licence to activate. Holding is
# reversible and the next run is a no-op, so the conservative branch costs nothing.
return {"ok": (not still_zero) and heal_failures == 0,
"healed": healed, "stillZero": still_zero, "errs": errs}