← back to Reid Witlin Onboarding
build_recovery_manifest.py
184 lines
#!/usr/bin/env python3
"""Build a read-only, immutable manifest for TK-10066 recovery.
This script performs no Shopify or PostgreSQL writes. It snapshots the live
Shopify products and the corresponding canonical rows, classifies duplicate
SKUs, and proposes deterministic fresh SKUs for cross-identity collisions.
"""
import collections
import csv
import datetime as dt
import json
import os
import re
import subprocess
import create2
HERE = os.path.dirname(os.path.abspath(__file__))
OUT_DIR = os.path.join(HERE, "verification", "recovery-manifests")
VENDOR = "Architectural Fabrics"
PRODUCTS = """
query($query: String!, $after: String) {
products(first: 250, after: $after, query: $query) {
nodes {
id handle title status vendor createdAt updatedAt
variants(first: 10) { nodes { id sku price } }
}
pageInfo { hasNextPage endCursor }
}
}"""
def sql_rows(sql):
result = subprocess.run(
["psql", "host=/tmp dbname=dw_unified", "-X", "-qAt", "-F", "\t", "-c", sql],
capture_output=True, text=True, check=True,
)
return [line.split("\t") for line in result.stdout.splitlines() if line]
def fetch_products():
products, after = [], None
while True:
data = create2.gql(PRODUCTS, {"query": f"vendor:'{VENDOR}'", "after": after})
if data.get("errors"):
raise RuntimeError(data["errors"])
conn = data["data"]["products"]
products.extend(conn["nodes"])
if not conn["pageInfo"]["hasNextPage"]:
return products
after = conn["pageInfo"]["endCursor"]
def numeric_suffix(sku):
match = re.fullmatch(r"DWKR-(\d+)", sku or "")
return int(match.group(1)) if match else -1
def main():
targets = list(csv.DictReader(open(os.path.join(HERE, "targets_ready_v2.csv"))))
target_by_mfr = {r["mfr_sku"].strip().upper(): r for r in targets}
products = fetch_products()
by_sku = collections.defaultdict(list)
for product in products:
for variant in product.get("variants", {}).get("nodes", []):
sku = variant.get("sku") or ""
if sku.startswith("DWKR-"):
by_sku[sku].append({
"product_id": product["id"], "variant_id": variant["id"],
"handle": product["handle"], "title": product["title"],
"status": product["status"], "created_at": product["createdAt"],
"updated_at": product["updatedAt"], "price": variant.get("price"),
})
duplicate_skus = sorted(s for s, nodes in by_sku.items() if len(nodes) > 1)
bases = sorted({s.removesuffix("-Sample") for s in duplicate_skus})
quoted = ",".join("'%s'" % s.replace("'", "''") for s in bases) or "''"
catalog = sql_rows(f"""
SELECT dw_sku, COALESCE(mfr_sku,''), COALESCE(shopify_product_id,''),
COALESCE(shopify_handle,''), id::text
FROM rwltd_catalog WHERE dw_sku IN ({quoted}) ORDER BY dw_sku,mfr_sku,id;
""")
registry = sql_rows(f"""
SELECT dw_sku, COALESCE(mfr_sku,''), COALESCE(shopify_product_id,''),
COALESCE(shopify_handle,''), id::text
FROM dw_sku_registry WHERE dw_sku IN ({quoted}) ORDER BY dw_sku;
""")
catalog_by_sku = collections.defaultdict(list)
for sku, mfr, pid, handle, row_id in catalog:
catalog_by_sku[sku].append({"row_id": int(row_id), "mfr_sku": mfr,
"product_id": pid, "handle": handle})
registry_by_sku = {r[0]: {"mfr_sku": r[1], "product_id": r[2],
"handle": r[3], "row_id": int(r[4])} for r in registry}
all_base_skus = {s.removesuffix("-Sample") for s in by_sku}
all_base_skus.update(r[0] for r in sql_rows(
"SELECT dw_sku FROM dw_sku_registry WHERE vendor_prefix='DWKR';"))
next_number = max(map(numeric_suffix, all_base_skus)) + 1
same_identity, cross_identity = [], []
for variant_sku in duplicate_skus:
base = variant_sku.removesuffix("-Sample")
rows = catalog_by_sku[base]
identities = sorted({r["mfr_sku"].strip().upper() for r in rows if r["mfr_sku"].strip()})
shopify = by_sku[variant_sku]
registry_owner = registry_by_sku.get(base)
if len(identities) <= 1:
retain_pid = next((r["product_id"] for r in rows if r["product_id"]), "")
retain_gid = retain_pid if retain_pid.startswith("gid://") else (
f"gid://shopify/Product/{retain_pid}" if retain_pid else "")
same_identity.append({
"sku": variant_sku, "mfr_skus": identities, "catalog_rows": rows,
"shopify_products": shopify, "retain_product_id": retain_gid,
"archive_product_ids": [p["product_id"] for p in shopify
if p["product_id"] != retain_gid],
})
continue
# The collision was introduced by Pool B minting into SKUs already held
# by Pool A. Only the first 12 pre-existing owners had registry rows;
# for the remaining 43, the immutable batch provenance is authoritative.
move_candidates = [r for r in rows
if target_by_mfr.get(r["mfr_sku"].strip().upper(), {}).get("pool") == "B-new-gap"]
if len(move_candidates) != 1:
raise RuntimeError(f"Cannot identify exactly one Pool-B mover for {base}")
move_row = move_candidates[0]
retain_row = next((r for r in rows if r is not move_row), None)
owner_mfr = (registry_owner or retain_row or {}).get("mfr_sku", "").strip().upper()
if not owner_mfr or not move_row["product_id"] or not retain_row:
raise RuntimeError(f"Cannot deterministically assign original owner for {base}")
move_gid = move_row["product_id"] if move_row["product_id"].startswith("gid://") else \
f"gid://shopify/Product/{move_row['product_id']}"
move_product = next((p for p in shopify if p["product_id"] == move_gid), None)
if not move_product:
raise RuntimeError(f"Catalog product link not present in Shopify duplicate set for {base}")
fresh = f"DWKR-{next_number:06d}"
next_number += 1
cross_identity.append({
"old_sku": base, "new_sku": fresh, "owner_source":
"registry" if registry_owner else "pool-provenance",
"registry_owner": registry_owner, "retain_identity": retain_row,
"move_identity": move_row, "move_product": move_product,
"catalog_rows": rows, "shopify_products": shopify,
})
handle_repairs = []
for handle in ("chilhowie-alaska-architectural-fabrics",
"haymarket-obsidian-architectural-fabrics"):
target = next((r for r in targets if r.get("handle") == handle), None)
product = next((p for p in products if p.get("handle") == handle), None)
handle_repairs.append({"handle": handle, "target": target, "shopify_product": product})
now = dt.datetime.now(dt.timezone.utc)
manifest = {
"schema": "tk-10066-recovery-manifest/v1", "generated_at": now.isoformat(),
"mode": "READ_ONLY", "vendor": VENDOR, "shopify_product_count": len(products),
"duplicate_sku_count": len(duplicate_skus),
"same_identity_count": len(same_identity), "cross_identity_count": len(cross_identity),
"same_identity": same_identity, "cross_identity": cross_identity,
"handle_repairs": handle_repairs,
"assertions": {
"expected_89_duplicates": len(duplicate_skus) == 89,
"expected_34_same_identity": len(same_identity) == 34,
"expected_55_cross_identity": len(cross_identity) == 55,
"all_archive_targets_are_draft": all(
p["status"] == "DRAFT" for item in same_identity
for p in item["shopify_products"] if p["product_id"] in item["archive_product_ids"]),
"fresh_skus_unique": len({x["new_sku"] for x in cross_identity}) == len(cross_identity),
"handle_repairs_resolved": all(x["target"] and x["shopify_product"] for x in handle_repairs),
},
}
if not all(manifest["assertions"].values()):
raise RuntimeError(json.dumps(manifest["assertions"], sort_keys=True))
os.makedirs(OUT_DIR, exist_ok=True)
path = os.path.join(OUT_DIR, now.strftime("%Y%m%dT%H%M%SZ") + ".json")
with open(path, "x") as fh:
json.dump(manifest, fh, indent=2, sort_keys=True)
fh.write("\n")
print(json.dumps({"manifest": path, **manifest["assertions"]}, indent=2))
if __name__ == "__main__":
main()