← back to Shopify Room Mockup
shopify-room-mockup.py
372 lines
#!/usr/bin/env python3
"""
shopify-room-mockup — add photoreal room-setting mockups to Shopify products.
For each selected product it: pulls the featured image (the pattern/swatch),
renders a photoreal room LOCALLY via the room-engines $0 compose_real.py
patch-quilt compositor (no network dependency, no API key), and uploads the
result as a product image (Shopify staged-upload + productCreateMedia).
REPOINTED 2026-09-23 (TK-12096, Lane G of TK-12090) — the original :3075
wallco room-setting service on Kamatera is DEAD: the port is now bound by
tailscaled, and the actual `room-setting-app` pm2 process moved to :3076 and
is crash-looping (50 restarts / 2h uptime). This script now shells out to
room-engines' compose_real.py (the "real-swatch quilt" engine — one of the
engines room-engines' own judge/apply_winners pipeline already selects
per-SKU) instead of SSHing to the dead service. room-engines' heavier
FLUX-Kontext / SDXL engines need their model weights re-cached before they
can run again; compose_real.py needs none and is deterministic + fast.
room-engines lives at:
~/Projects/Designer-Wallcoverings/scripts/wallquest-refresh/carl-robinson-viewer/room-engines/
This script's distinct value vs. room-engines' own pipeline is unchanged:
generic Shopify-side product SELECTION (--ids/--query/--collection across the
WHOLE store) pulling each product's OWN live featured image — room-engines'
scripts are wired to a single book's local catalog table + a pre-built
manifest.tsv, not to arbitrary Shopify products.
Built 2026-05-29 from the Crowley's Crocodile batch (designer-laboratory-sandbox).
SELECT (choose one):
--ids 7797773500467,6811742175283 explicit product IDs (or full GIDs)
--query "title:Crowley*" Shopify product search query
--collection 123456789 all products in a collection
OPTIONS:
--store designer-laboratory-sandbox store handle (default) or full *.myshopify.com
--room-type living_room kept for CLI compat / alt-text only — the
one preserved base-room asset is a living
room, so all renders use it regardless
--pattern-width 39 pattern repeat width in inches (scale; default 39)
--pattern-height 39
--tile-px N compose_real.py patch size on the wall in px
(default derived from --pattern-width)
--room-engines-dir PATH room-engines/ directory (default: the DW
carl-robinson-viewer/room-engines checkout)
--box root@45.61.58.125 DEPRECATED / no-op (kept for CLI compat —
the :3075 SSH room service is retired)
--skip-existing skip products that already have a room mockup
(detected by media alt containing "in a room")
--force push even if out/room-mockup-ledger.jsonl already has a live
upload for that product+ticket (default: refuse — double-run guard)
--alt-suffix "shown in a living room" alt-text suffix for the new image
--limit N cap number of products processed
--dry-run list targets + render, but DON'T upload
--token-env SHOPIFY_ADMIN_TOKEN env var (falls back to secrets-manager/.env)
Examples:
python3 shopify-room-mockup.py --query "title:Crowley*" --skip-existing
python3 shopify-room-mockup.py --ids 7797773500467 --room-type bedroom
python3 shopify-room-mockup.py --collection 480239649 --dry-run
"""
import argparse, base64, json, os, re, shutil, subprocess, sys, tempfile, urllib.request
from datetime import datetime, timezone
API_VERSION = '2024-10'
DEFAULT_ROOM_ENGINES_DIR = os.path.expanduser(
'~/Projects/Designer-Wallcoverings/scripts/wallquest-refresh/carl-robinson-viewer/room-engines')
# Bootstrap dir compose_real.py's hardcoded BASE/MASK paths expect (its own
# /tmp scratch convention from the original Carl Robinson bake-off). The real,
# permanently-preserved assets live one level up in the viewer's bakeoff-assets/.
BAKEOFF_SCRATCH = '/tmp/cr-room-bakeoff'
def load_token(env_name):
if os.environ.get(env_name):
return os.environ[env_name]
envp = os.path.expanduser('~/Projects/secrets-manager/.env')
if os.path.isfile(envp):
for line in open(envp):
if line.startswith(env_name + '='):
return line.split('=', 1)[1].strip()
sys.exit(f'No token: set ${env_name} or add it to ~/Projects/secrets-manager/.env')
def store_host(store):
return store if store.endswith('.myshopify.com') else f'{store}.myshopify.com'
class Shopify:
def __init__(self, host, token):
self.url = f'https://{host}/admin/api/{API_VERSION}/graphql.json'
self.token = token
def gql(self, query, variables=None):
body = json.dumps({'query': query, 'variables': variables or {}}).encode()
req = urllib.request.Request(self.url, data=body,
headers={'X-Shopify-Access-Token': self.token, 'Content-Type': 'application/json'})
d = json.load(urllib.request.urlopen(req, timeout=60))
if d.get('errors'):
raise RuntimeError('GraphQL: ' + json.dumps(d['errors'])[:300])
return d['data']
def gid(x):
return x if str(x).startswith('gid://') else f'gid://shopify/Product/{x}'
def select_products(sh, args):
"""Return [{id, title, image_url, has_room}]."""
out = []
if args.ids:
gids = [gid(i.strip()) for i in args.ids.split(',') if i.strip()]
q = '{ nodes(ids:%s){ ... on Product { id title featuredImage{url} media(first:25){edges{node{... on MediaImage{alt}}}} } } }' % json.dumps(gids)
nodes = sh.gql(q)['nodes']
else:
if args.collection:
qfilter = ''
base = 'collection(id:"%s"){ products(first:250%s){ edges{ node{ id title featuredImage{url} media(first:25){edges{node{... on MediaImage{alt}}}} } } } }' % (gid(args.collection).replace('Product', 'Collection'), '')
data = sh.gql('{ %s }' % base)
nodes = [e['node'] for e in data['collection']['products']['edges']]
else:
q = '{ products(first:250, query:%s){ edges{ node{ id title featuredImage{url} media(first:25){edges{node{... on MediaImage{alt}}}} } } } }' % json.dumps(args.query)
nodes = [e['node'] for e in sh.gql(q)['products']['edges']]
for n in nodes:
if not n:
continue
alts = [(e['node'] or {}).get('alt') or '' for e in n.get('media', {}).get('edges', [])]
# marker matches "shown in a <roomtype>" (living room / bedroom / …) AND
# the literal "in a room" — robust to room type. The plain "in a room"
# check alone missed "in a living room" (the word 'living' breaks it).
has_room = any(('shown in a' in s) or ('in a room' in s)
for s in ((a or '').lower() for a in alts))
out.append({'id': n['id'], 'title': n['title'],
'image_url': (n.get('featuredImage') or {}).get('url'), 'has_room': has_room})
return out
def _bootstrap_bakeoff_assets(room_engines_dir):
"""compose_real.py hardcodes BASE=/tmp/cr-room-bakeoff/00-empty-room.png and
MASK=/tmp/cr-room-bakeoff/wall-mask.png (its own scratch convention from the
original bake-off run). Those two source assets are permanently preserved one
level up at <viewer>/bakeoff-assets/{empty-room.png,wall-mask.png} — copy them
into the expected scratch path once (idempotent, $0, purely local)."""
os.makedirs(BAKEOFF_SCRATCH, exist_ok=True)
dst_room = os.path.join(BAKEOFF_SCRATCH, '00-empty-room.png')
dst_mask = os.path.join(BAKEOFF_SCRATCH, 'wall-mask.png')
viewer_dir = os.path.dirname(os.path.normpath(room_engines_dir))
src_room = os.path.join(viewer_dir, 'bakeoff-assets', 'empty-room.png')
src_mask = os.path.join(viewer_dir, 'bakeoff-assets', 'wall-mask.png')
if not os.path.exists(dst_room) and os.path.exists(src_room):
shutil.copy2(src_room, dst_room)
if not os.path.exists(dst_mask) and os.path.exists(src_mask):
shutil.copy2(src_mask, dst_mask)
return os.path.exists(dst_room) and os.path.exists(dst_mask)
def render_room(image_url, room_engines_dir, tile_px):
"""Render LOCALLY via room-engines' compose_real.py $0 patch-quilt compositor
(no network service, no API key). Downloads the product's own pattern/swatch
image, then shells out to compose_real.py <swatch> <tile_px> <out>.
Returns raw PNG bytes (or None)."""
compose_script = os.path.join(room_engines_dir, 'compose_real.py')
if not os.path.isfile(compose_script):
print(f' ✗ room-engines compose_real.py not found at {compose_script}', file=sys.stderr)
return None
if not _bootstrap_bakeoff_assets(room_engines_dir):
print(' ✗ base-room/wall-mask assets missing (bakeoff-assets/) — cannot compose', file=sys.stderr)
return None
with tempfile.TemporaryDirectory() as td:
swatch = os.path.join(td, 'swatch.img')
try:
urllib.request.urlretrieve(image_url, swatch)
except Exception as e:
print(f' ✗ swatch download failed: {e}', file=sys.stderr)
return None
out = os.path.join(td, 'room.png')
p = subprocess.run([sys.executable, compose_script, swatch, str(tile_px), out],
capture_output=True, text=True, timeout=120)
if p.returncode != 0 or not os.path.exists(out):
print(f' ✗ compose_real.py failed: {(p.stderr or p.stdout or "").strip()[:200]}', file=sys.stderr)
return None
with open(out, 'rb') as f:
buf = f.read()
return buf if buf[:8] == bytes([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]) else None
def append_ledger(ledger_path, row):
"""Append one JSONL row and fsync before returning, so the row is durable on
disk the instant this call returns — a crash immediately after (e.g. before
the caller prints its success line) cannot lose the media id. Raises on
failure; callers MUST treat a ledger-write failure as loud, not swallow it —
a media object that exists live with no recorded undo is unrecoverable."""
os.makedirs(os.path.dirname(ledger_path) or '.', exist_ok=True)
with open(ledger_path, 'a') as f:
f.write(json.dumps(row) + '\n')
f.flush()
os.fsync(f.fileno())
def ledgered_live(ledger_path, ticket):
"""Product gids that already have a LIVE ledgered room upload for this ticket.
A row with a media_id and no `action` (or action=='upload') is an upload; an
action=='deleted' row for the same media_id cancels it. This is the
double-run guard (TK-12096: two runs 3 min apart pushed duplicate images to
5 live products). A missing ledger means nothing is ledgered yet; an
UNREADABLE ledger raises, so the guard can never fail open on a bad file.
"""
if not os.path.exists(ledger_path):
return set()
live = {}
with open(ledger_path) as f:
for line in f:
if not line.strip():
continue
r = json.loads(line)
if r.get('ticket') != ticket or not r.get('media_id'):
continue
if r.get('action', 'upload') == 'deleted':
live.pop(r['media_id'], None)
else:
live[r['media_id']] = r['product_gid']
return set(live.values())
def undo_cmd_for(store, token_env, media_id, product_gid):
return (f'python3 shopify-room-mockup.py --store {store} --token-env {token_env} '
f'--delete-media {media_id} --product {product_gid}')
def upload_media(sh, product_gid, png_bytes, alt):
fn = 'room-mockup.png'
su = sh.gql('mutation($i:[StagedUploadInput!]!){stagedUploadsCreate(input:$i){stagedTargets{url resourceUrl parameters{name value}} userErrors{field message}}}',
{'i': [{'resource': 'IMAGE', 'filename': fn, 'mimeType': 'image/png', 'httpMethod': 'POST'}]})['stagedUploadsCreate']
if su['userErrors']:
return False, su['userErrors']
t = su['stagedTargets'][0]
with tempfile.NamedTemporaryFile(suffix='.png', delete=False) as f:
f.write(png_bytes); tmp = f.name
try:
args = ['curl', '-s', '-o', '/dev/null', '-w', '%{http_code}', '-X', 'POST', t['url']]
for p in t['parameters']:
args += ['-F', f"{p['name']}={p['value']}"]
args += ['-F', f'file=@{tmp};type=image/png']
code = subprocess.run(args, capture_output=True, text=True).stdout.strip()
finally:
os.unlink(tmp)
if code not in ('200', '201', '204'):
return False, f'staged upload HTTP {code}'
pc = sh.gql('mutation($p:ID!,$m:[CreateMediaInput!]!){productCreateMedia(productId:$p,media:$m){media{... on MediaImage{id status}} mediaUserErrors{field message}}}',
{'p': product_gid, 'm': [{'originalSource': t['resourceUrl'], 'alt': alt, 'mediaContentType': 'IMAGE'}]})['productCreateMedia']
if pc['mediaUserErrors']:
return False, pc['mediaUserErrors']
return True, pc['media']
def main():
ap = argparse.ArgumentParser(description='Add photoreal room mockups to Shopify products.')
g = ap.add_mutually_exclusive_group(required=False)
g.add_argument('--ids'); g.add_argument('--query'); g.add_argument('--collection')
ap.add_argument('--store', default='designer-laboratory-sandbox')
ap.add_argument('--room-type', default='living_room')
ap.add_argument('--pattern-width', type=int, default=39)
ap.add_argument('--pattern-height', type=int, default=39)
ap.add_argument('--tile-px', type=int, default=0,
help='compose_real.py wall-patch size in px (0 = derive from --pattern-width)')
ap.add_argument('--room-engines-dir', default=DEFAULT_ROOM_ENGINES_DIR)
ap.add_argument('--box', default=None, help=argparse.SUPPRESS) # deprecated no-op, kept for CLI compat
ap.add_argument('--skip-existing', action='store_true')
ap.add_argument('--alt-suffix', default='shown in a room')
ap.add_argument('--limit', type=int, default=0)
ap.add_argument('--dry-run', action='store_true')
ap.add_argument('--token-env', default='SHOPIFY_ADMIN_TOKEN')
ap.add_argument('--ticket', default='TK-12096', help='ticket id stamped into ledger rows')
ap.add_argument('--ledger', default=os.path.join(os.path.dirname(os.path.abspath(__file__)), 'out', 'room-mockup-ledger.jsonl'),
help='JSONL path recording {media_id, undo_cmd, ...} for every uploaded media (TK-12090)')
ap.add_argument('--delete-media', default=None,
help='UNDO utility mode: productDeleteMedia on this media id/gid (requires --product). '
'This is exactly the invocation the ledger writes as undo_cmd.')
ap.add_argument('--product', default=None, help='product id/gid, used with --delete-media')
ap.add_argument('--force', action='store_true',
help='push even when the ledger already has a live room upload for this product+ticket')
args = ap.parse_args()
if args.delete_media:
if not args.product:
ap.error('--delete-media requires --product <product id or gid>')
sh = Shopify(store_host(args.store), load_token(args.token_env))
media_gid = args.delete_media if str(args.delete_media).startswith('gid://') else f'gid://shopify/MediaImage/{args.delete_media}'
pgid = gid(args.product)
d = sh.gql('mutation($ids:[ID!]!,$pid:ID!){productDeleteMedia(mediaIds:$ids,productId:$pid){deletedMediaIds mediaUserErrors{field message}}}',
{'ids': [media_gid], 'pid': pgid})['productDeleteMedia']
errs = d.get('mediaUserErrors') or []
if errs:
sys.exit(f'productDeleteMedia errors: {errs}')
print(f' ↩ deleted media {media_gid} from {pgid}: {d.get("deletedMediaIds")}')
return
if not (args.ids or args.query or args.collection):
ap.error('one of --ids, --query, --collection is required (or --delete-media + --product for undo mode)')
if args.box:
print(' (note) --box is deprecated/no-op — the :3075 SSH room service is retired; '
'rendering now happens locally via room-engines/compose_real.py', file=sys.stderr)
tile_px = args.tile_px or max(20, min(120, args.pattern_width))
sh = Shopify(store_host(args.store), load_token(args.token_env))
prods = select_products(sh, args)
if args.limit:
prods = prods[:args.limit]
print(f'{len(prods)} product(s) selected on {args.store}.')
already = set() if args.force else ledgered_live(args.ledger, args.ticket)
done = skipped = failed = 0
for p in prods:
tag = p['title']
if p['id'] in already:
print(f' ⏭ SKIP (already ledgered for {args.ticket}; --force to override): {tag}')
skipped += 1; continue
if args.skip_existing and p['has_room']:
print(f' ⏭ SKIP (has room): {tag}'); skipped += 1; continue
if not p['image_url']:
print(f' ⚠ no featured image: {tag}'); failed += 1; continue
buf = render_room(p['image_url'], args.room_engines_dir, tile_px)
if not buf:
print(f' ✗ render failed: {tag}'); failed += 1; continue
if args.dry_run:
print(f' ◌ DRY-RUN rendered {len(buf)//1024}KB (not uploaded): {tag}'); done += 1; continue
alt = f"{p['title']} — {args.alt_suffix}"
ok, info = upload_media(sh, p['id'], buf, alt)
if ok:
media_id = None
if isinstance(info, list) and info and isinstance(info[0], dict):
media_id = info[0].get('id')
if not media_id:
# The mutation reported success but returned no media id we can
# act on — that means a live media object may exist with NO
# recoverable undo. Loud failure, not a silent success line.
print(f' ✗ uploaded but NO media id in response — cannot ledger, refusing to report success: {tag} — {info}',
file=sys.stderr)
failed += 1
continue
row = {
'ts': datetime.now(timezone.utc).isoformat(),
'ticket': args.ticket,
'store': args.store,
'product_gid': p['id'],
'product_title': p['title'],
'media_id': media_id,
'alt': alt,
'undo_cmd': undo_cmd_for(args.store, args.token_env, media_id, p['id']),
}
try:
# Write BEFORE printing success — a crash between write and
# print can never lose the id (TK-12090 liveness-artifact rule).
append_ledger(args.ledger, row)
except Exception as e:
print(f' ✗ LEDGER WRITE FAILED for {tag} (media {media_id} now LIVE with NO recorded undo!): {e}',
file=sys.stderr)
failed += 1
continue
print(f' ✓ added room: {tag} (media {media_id} ledgered -> {args.ledger})')
done += 1
else:
print(f' ✗ upload failed: {tag} — {info}'); failed += 1
print(f'\n[done] added={done} skipped={skipped} failed={failed}'
+ (' (dry-run — nothing uploaded)' if args.dry_run else ''))
if __name__ == '__main__':
main()