#!/usr/bin/env python3 """ maker_run.py — Picks next ready_for_maker item, generates PNG via DALL-E, validates output. No AI agent needed — calls OpenAI API directly. """ import json import os import re import time import urllib.request import urllib.error from pathlib import Path WORKSPACE = Path("/Users/tonyclaw/.openclaw/workspace") ETSY_DIR = WORKSPACE / "Etsy" STATE_PATH = ETSY_DIR / "pipeline-state.json" PIPELINE_TOOL = ETSY_DIR / "tools" / "pipeline_state.py" DESIGNS_ROOT = Path("/Users/tonyclaw/Documents/Etsy Designs") SOURCE_PNG_DIR = DESIGNS_ROOT / "01-Source-PNG" AUTH_PROFILES = Path("/Users/tonyclaw/.openclaw/auth-profiles.json") OPENAI_API_URL = "https://api.openai.com/v1/images/generations" # DALL-E config DALLE_MODEL = "dall-e-3" DALLE_SIZE = "1024x1024" DALLE_QUALITY = "standard" def load_openai_key() -> str: try: profiles = json.loads(AUTH_PROFILES.read_text(encoding="utf-8")) # Try openai:manual profile first for name, profile in profiles.get("profiles", {}).items(): if "openai" in name.lower() and profile.get("apiKey"): return profile["apiKey"] # Try env fallback return os.environ.get("OPENAI_API_KEY", "") except Exception: return os.environ.get("OPENAI_API_KEY", "") def run_pipeline_tool(args: list) -> tuple[int, str, str]: import subprocess r = subprocess.run( ["python3", str(PIPELINE_TOOL)] + args, check=False, text=True, capture_output=True ) return r.returncode, r.stdout.strip(), r.stderr.strip() def load_state() -> dict: with open(STATE_PATH, "r", encoding="utf-8") as f: return json.load(f) def pick_next_item(state: dict) -> dict | None: # Only pick API-generated items (skip manual uploads) candidates = [it for it in state.get("items", []) if it.get("status") == "ready_for_maker" and it.get("source") != "manual"] if not candidates: return None candidates.sort(key=lambda x: (-int(x.get("priority", 0)), x.get("created_at", ""))) return candidates[0] def ps_set(item_id, stage, status, message, design_name=None, paths=None): args = [ "set", "--path", str(STATE_PATH), "--id", item_id, "--stage", stage, "--status", status, "--by", "etsy-maker", "--message", message ] if design_name: args += ["--design_name", design_name] if paths: for k, v in paths.items(): args += ["--path_kv", f"{k}={v}"] rc, out, err = run_pipeline_tool(args) if rc != 0: raise RuntimeError(err or "pipeline_state set failed") def ps_error(item_id, message): args = [ "error", "--path", str(STATE_PATH), "--id", item_id, "--by", "etsy-maker", "--message", message ] run_pipeline_tool(args) def slugify(theme: str) -> str: s = theme.lower() s = re.sub(r"[^a-z0-9\s-]", "", s) s = re.sub(r"\s+", "-", s).strip("-") s = re.sub(r"-+", "-", s) return s[:50] if len(s) > 50 else s def build_dalle_prompt(item: dict) -> str: """ Build a DALL-E prompt from pipeline item. Following prompt-guidelines.md: black silhouette, clean lines, no SVG/vector language. """ theme = item.get("theme", "") niche = item.get("niche", "") keywords = item.get("keywords", []) or [] avoid = item.get("avoidList", []) or [] # Strip calendar prefix like "Easter (Upload by 2026-02-22) " core_theme = theme if ") " in theme: core_theme = theme.split(") ", 1)[1].strip() # Remove "(no text)" hints — we encode that in the prompt core_theme = re.sub(r"\(no text\)", "", core_theme).strip() niche_style = { "USMC/Military": "bold black silhouette of a {subject}, clean simple lines, high contrast black on pure white background, suitable for vinyl cutting or laser engraving, no gradients, no shading, no text", "Reformed Christian": "clean black silhouette of {subject}, elegant simple lines, high contrast black on pure white background, suitable for vinyl cutting, no gradients, no text, no scripture text", "Patriotic": "bold black silhouette of {subject}, clean simple lines, high contrast black on pure white background, suitable for vinyl cutting, no gradients, no text", "Nature/Outdoor": "clean black silhouette of {subject}, simple nature illustration style, high contrast black on pure white background, suitable for vinyl cutting, no gradients, no text", } style_template = niche_style.get(niche, "bold black silhouette of {subject}, clean simple lines, high contrast black on pure white background, no gradients, no text") prompt = style_template.format(subject=core_theme) # Add avoid list guidance if avoid: clean_avoid = [a for a in avoid if a not in ("trademarked", "phrases", "bible", "verse", "text")] if clean_avoid: prompt += f". Avoid: {', '.join(clean_avoid[:3])}" return prompt def generate_dalle_image(prompt: str, api_key: str) -> bytes: """Call DALL-E 3 API, return PNG bytes.""" payload = json.dumps({ "model": DALLE_MODEL, "prompt": prompt, "n": 1, "size": DALLE_SIZE, "quality": DALLE_QUALITY, "response_format": "url" }).encode("utf-8") req = urllib.request.Request( OPENAI_API_URL, data=payload, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {api_key}" }, method="POST" ) with urllib.request.urlopen(req, timeout=60) as resp: result = json.loads(resp.read().decode("utf-8")) image_url = result["data"][0]["url"] # Download the PNG img_req = urllib.request.Request(image_url) with urllib.request.urlopen(img_req, timeout=60) as resp: return resp.read() def main(): api_key = load_openai_key() if not api_key: print("Maker: no OpenAI API key found — check auth-profiles.json or OPENAI_API_KEY env") return state = load_state() item = pick_next_item(state) if not item: print("Maker: no items ready_for_maker") return item_id = item["id"] theme = item.get("theme", item_id) design_name = item.get("design_name") or slugify(theme) # Claim ps_set(item_id, stage="maker_in_progress", status="blocked", message="Claimed for DALL-E generation") try: # Build prompt prompt = build_dalle_prompt(item) print(f"Maker: generating '{design_name}' | niche={item.get('niche')}") print(f"Maker: prompt={prompt[:120]}...") # Create output folder folder = SOURCE_PNG_DIR / design_name folder.mkdir(parents=True, exist_ok=True) # Generate via DALL-E png_bytes = generate_dalle_image(prompt, api_key) # Save PNG png_path = folder / f"{design_name}.png" png_path.write_bytes(png_bytes) # Save prompt for reference prompt_path = folder / "prompt.txt" prompt_path.write_text( f"Theme: {theme}\nNiche: {item.get('niche')}\nPrompt: {prompt}\n", encoding="utf-8" ) print(f"Maker: saved {png_path} ({len(png_bytes):,} bytes)") # Validate if not png_path.exists() or png_path.stat().st_size < 10000: ps_error(item_id, f"PNG too small or missing: {png_path}") return # Mark complete ps_set( item_id, stage="maker_complete", status="ready_for_critic", message=f"DALL-E PNG generated ({len(png_bytes):,} bytes).", design_name=design_name, paths={ "source_folder": str(folder), "png": str(png_path) } ) print(f"Maker complete: {item_id} -> ready_for_critic ({design_name})") except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") ps_error(item_id, f"DALL-E HTTP {e.code}: {body[:200]}") except Exception as e: ps_error(item_id, f"Unhandled exception: {e}") if __name__ == "__main__": main()