← back to Shopify Room Mockup
shopify-room-mockup.py
206 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 via the wallco room-setting service (:3075 on the
Kamatera box, reached over SSH), and uploads the result as a product image
(Shopify staged-upload + productCreateMedia).
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 living_room|bedroom|dining_room|office (default living_room)
--pattern-width 39 pattern repeat width in inches (scale; default 39)
--pattern-height 39
--box root@45.61.58.125 SSH host running the :3075 room service
--skip-existing skip products that already have a room mockup
(detected by media alt containing "in a room")
--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, subprocess, sys, tempfile, urllib.request
API_VERSION = '2024-10'
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
REMOTE_RENDER = """import base64,json,urllib.request,sys
url=%(url)r
jpg='/tmp/_srm_pattern.img'
urllib.request.urlretrieve(url,jpg)
b64=base64.b64encode(open(jpg,'rb').read()).decode()
body=json.dumps({'patternBase64':b64,'roomType':%(room)r,'angle':'straight_on',
'cameraDistance':5,'patternWidth':%(pw)d,'patternHeight':%(ph)d}).encode()
req=urllib.request.Request('http://127.0.0.1:3075/api/generate-room',data=body,
headers={'Content-Type':'application/json'})
r=json.load(urllib.request.urlopen(req,timeout=180))
img=r.get('image') or r.get('imageBase64') or ''
if img.startswith('data:'): img=img.split(',',1)[1]
sys.stdout.write(img or '')
"""
def render_room(box, image_url, room_type, pw, ph):
"""Render on the box via SSH; returns raw PNG bytes (or None)."""
script = REMOTE_RENDER % {'url': image_url, 'room': room_type, 'pw': pw, 'ph': ph}
p = subprocess.run(['ssh', box, 'python3', '-'], input=script, capture_output=True, text=True, timeout=240)
b64 = (p.stdout or '').strip()
if not b64:
return None
try:
buf = base64.b64decode(b64)
return buf if buf[:8] == bytes([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]) else None
except Exception:
return None
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=True)
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('--box', default='root@45.61.58.125')
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')
args = ap.parse_args()
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}.')
done = skipped = failed = 0
for p in prods:
tag = p['title']
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(args.box, p['image_url'], args.room_type, args.pattern_width, args.pattern_height)
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:
print(f' ✓ added room: {tag}'); 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()