flow-extension/flowclient.py
type-two 501431bf59 Flow rinse: browser downloads, fast-fail, autoStart, launcher + self-healing launchd
- background.js: chrome.downloads via page session (no CSP/base64)
- content.js: fast-fail on Flow error banner, image vs video timeouts, autoStart
- local_server_example.py: /enqueue + /save, downloads/ output, resume-skip
- launch.sh: idempotent server-up + caffeinate + single-tab focus
- install.sh + plist: self-healing agent every 10 min (TCC-safe App Support copy)
- flowclient.py: enqueue/wait_for helper for game projects
- README: full setup + macOS TCC gotcha

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-06 01:20:32 +10:00

50 lines
1.9 KiB
Python

#!/usr/bin/env python3
"""Tiny client so OpShopGame / 90sDJsim can REQUEST Flow assets and RETRIEVE them.
The running Flow extension is the bridge to Google — the games never touch it.
from flowclient import enqueue, wait_for
r = enqueue("Low-poly vinyl crate icon, transparent bg, 90s UI",
kind="image", category="djsim_ui",
dest="/Users/johnking/Documents/90sDJsim/assets/ui")
path = wait_for(r["dest"], r["id"]) # blocks until the file lands
Request -> appended to the queue; the extension drains it, generates, downloads.
Retrieve -> the file appears in <dest> (and in flowext/downloads/<category>/).
"""
import json, os, sys, time, urllib.request
SERVER = os.environ.get("FLOW_SERVER", "http://localhost:8017")
def enqueue(prompt, kind="image", model=None, category="request", dest=None, server=SERVER):
body = json.dumps({"prompt": prompt, "kind": kind, "model": model,
"category": category, "dest": dest}).encode()
req = urllib.request.Request(server + "/enqueue", body, {"Content-Type": "application/json"})
out = json.load(urllib.request.urlopen(req, timeout=10)) # {"ok":True,"id":...}
out["dest"] = dest
return out
def wait_for(dest, task_id, timeout=1800, poll=10):
"""Block until a file for task_id lands in <dest>; return its path (or None)."""
if not dest:
return None
end = time.time() + timeout
while time.time() < end:
if os.path.isdir(dest):
for f in sorted(os.listdir(dest)):
if f.startswith(task_id):
return os.path.join(dest, f)
time.sleep(poll)
return None
if __name__ == "__main__":
if "--demo" in sys.argv:
print(enqueue("Low-poly flat-shaded vinyl record crate icon, transparent background, 90s UI",
kind="image", category="djsim_ui"))
else:
print(__doc__)