#!/usr/bin/env python3 """ comfy-web — web GUI + payment frontend for the ComfyUI app registry. Enumerates comfy_apps.APPS, takes payment in sats via BTCPay, runs the app against nightmare's ComfyUI (GPU), and delivers the image. Run: python3 app.py (dev, port 5188) """ import os, json, time, sqlite3, threading, uuid, uuid as _uuid import urllib.request, urllib.parse from flask import Flask, request, render_template, redirect, url_for, jsonify, send_from_directory, abort import sys sys.path.insert(0, os.path.expanduser("~/auto-publisher/core")) import comfy_apps as CA import comfy_workflows_lib as CL # --------------------------------------------------------------------------- # config # --------------------------------------------------------------------------- BASE = os.path.expanduser("~/comfy-web") DB = os.path.join(BASE, "jobs.db") UPLOAD_DIR = os.path.join(BASE, "uploads") RESULT_DIR = os.path.join(BASE, "results") os.makedirs(UPLOAD_DIR, exist_ok=True) os.makedirs(RESULT_DIR, exist_ok=True) # BTCPay (CT133). API calls use the LAN IP (public domain has CF 1010 block for non-browser); # checkout redirects use the public domain. Empty => MOCK mode. BTCPAY_API = "https://10.30.20.140" # server-to-server API (self-signed, ssl unverified) BTCPAY_PUBLIC = "https://btcpay.thetempleofdoom.com" # customer-facing checkout BTCPAY_STORE_ID = "6AZoSSdADJ8NndwLDhhh3geg6BgS487ZCkW3msX2mT1" BTCPAY_API_KEY = "18ebeb23d41e971e7a133479a7c0644f78a337fc" WEBHOOK_URL = "https://nebula.thetempleofdoom.com" MOCK_PAYMENTS = not (BTCPAY_API and BTCPAY_STORE_ID and BTCPAY_API_KEY) import ssl _CTX = ssl.create_default_context() _CTX.check_hostname = False _CTX.verify_mode = ssl.CERT_NONE app = Flask(__name__) # --------------------------------------------------------------------------- # db # --------------------------------------------------------------------------- def db(): c = sqlite3.connect(DB) c.row_factory = sqlite3.Row return c def init_db(): with db() as c: c.execute("""CREATE TABLE IF NOT EXISTS jobs ( id TEXT PRIMARY KEY, app TEXT, prompt TEXT, params TEXT, status TEXT, -- queued | running | done | failed | awaiting_payment btcpay_invoice TEXT, result_file TEXT, error TEXT, created_at REAL, paid_at REAL, finished_at REAL )""") init_db() # --------------------------------------------------------------------------- # payment (BTCPay Greenfield API, or mock) # --------------------------------------------------------------------------- def create_invoice(job_id, app_name, sats): price = CA.APPS[app_name]["price_sats"] amount = f"{sats / 100_000_000:.8f}" # sats -> BTC if MOCK_PAYMENTS: return f"mock-{job_id}" body = json.dumps({ "amount": amount, "currency": "BTC", "metadata": {"job_id": job_id, "app": app_name}, "checkout": {"redirectURL": f"{WEBHOOK_URL}/job/{job_id}", "requiresRefundEmail": False}, }).encode() req = urllib.request.Request( f"{BTCPAY_API}/api/v1/stores/{BTCPAY_STORE_ID}/invoices", data=body, headers={"Authorization": f"token {BTCPAY_API_KEY}", "Content-Type": "application/json"}, ) r = urllib.request.urlopen(req, timeout=30, context=_CTX) inv = json.loads(r.read()) return inv["id"] def settle_job(job_id): """Mark paid and queue for execution (idempotent — no double-credit).""" with db() as c: cur = c.execute("UPDATE jobs SET status='queued', paid_at=? WHERE id=? AND status='awaiting_payment'", (time.time(), job_id)) if cur.rowcount == 0: return # already settled (replay guard) with db() as c: row = c.execute("SELECT id, app, prompt, params FROM jobs WHERE id=?", (job_id,)).fetchone() if row: _queue.append(row) # --------------------------------------------------------------------------- # job worker (serializes the GPU) # --------------------------------------------------------------------------- _queue = [] def _worker(): while True: if not _queue: time.sleep(1) continue row = _queue.pop(0) job_id, app_name, prompt, params = row["id"], row["app"], row["prompt"], json.loads(row["params"] or "{}") try: with db() as c: c.execute("UPDATE jobs SET status='running' WHERE id=?", (job_id,)) # free the GPU (unload Ollama + switch to ComfyUI OC) before rendering CL.free_gpu() # run the app fn = CA.APPS[app_name]["fn"] params = dict(params) if prompt: params["prompt"] = prompt wf = fn(**params) pid, err = CL.submit(wf) if err: raise RuntimeError(str(err)[:400]) paths = CL.fetch_result(pid, timeout=900) if not paths: raise RuntimeError("job timed out") # move result into results dir under job id src = paths[0] ext = os.path.splitext(src)[1] or ".png" dest = os.path.join(RESULT_DIR, job_id + ext) os.rename(src, dest) with db() as c: c.execute("UPDATE jobs SET status='done', result_file=?, finished_at=? WHERE id=?", (os.path.basename(dest), time.time(), job_id)) CL.restore_gpu() except Exception as e: CL.restore_gpu() with db() as c: c.execute("UPDATE jobs SET status='failed', error=?, finished_at=? WHERE id=?", (str(e)[:500], time.time(), job_id)) threading.Thread(target=_worker, daemon=True).start() # --------------------------------------------------------------------------- # routes # --------------------------------------------------------------------------- @app.route("/") def index(): apps = CA.list_apps() return render_template("index.html", apps=apps) @app.route("/app/") def app_detail(name): if name not in CA.APPS: abort(404) meta = CA.list_apps()[name] return render_template("app.html", name=name, meta=meta) @app.route("/run", methods=["POST"]) def run_job(): app_name = request.form.get("app") if app_name not in CA.APPS: abort(400) prompt = request.form.get("prompt", "") # gather params + handle uploaded source image params = {} img = request.files.get("image") if img and img.filename: local = os.path.join(UPLOAD_DIR, uuid.uuid4().hex + os.path.splitext(img.filename)[1]) img.save(local) remote = os.path.basename(local) CL.upload_image(local, name=remote) params["image_name"] = remote # for inpaint, a mask may come too mask = request.files.get("mask") if mask and mask.filename: mlocal = os.path.join(UPLOAD_DIR, uuid.uuid4().hex + os.path.splitext(mask.filename)[1]) mask.save(mlocal) CL.upload_image(mlocal, name=os.path.basename(mlocal)) params["mask_name"] = os.path.basename(mlocal) # collect numeric/select params for key in ("seed", "steps", "denoise", "upscaler", "control_type", "weight", "left", "right", "top", "bottom", "w", "h", "cfg", "strength"): if key in request.form and request.form[key] != "": v = request.form[key] params[key] = int(v) if v.lstrip("-").isdigit() else v if request.form.get("negative", "").strip(): params["negative"] = request.form["negative"].strip() job_id = uuid.uuid4().hex with db() as c: c.execute("INSERT INTO jobs (id, app, prompt, params, status, created_at) VALUES (?,?,?,?,?,?)", (job_id, app_name, prompt, json.dumps(params), "awaiting_payment", time.time())) # payment sats = CA.APPS[app_name]["price_sats"] if MOCK_PAYMENTS: settle_job(job_id) return redirect(url_for("job", job_id=job_id)) inv = create_invoice(job_id, app_name, sats) with db() as c: c.execute("UPDATE jobs SET btcpay_invoice=? WHERE id=?", (inv, job_id)) return redirect(f"{BTCPAY_PUBLIC}/i/{inv}") @app.route("/webhook/btcpay", methods=["POST"]) def btcpay_webhook(): data = request.get_json(force=True) # metadata is TOP-LEVEL in the BTCPay webhook payload, and job_id is nested inside it if data.get("type") in ("InvoiceSettled", "InvoiceProcessing"): metadata = (data.get("metadata") or {}) job_id = metadata.get("job_id") or data.get("job_id") if job_id: settle_job(job_id) return "", 200 @app.route("/job/") def job(job_id): with db() as c: row = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone() if not row: abort(404) return render_template("job.html", job=row) @app.route("/jobs") def jobs(): with db() as c: rows = c.execute("SELECT * FROM jobs ORDER BY created_at DESC LIMIT 100").fetchall() return render_template("jobs.html", jobs=rows) @app.route("/image/") def image(filename): return send_from_directory(RESULT_DIR, filename) @app.route("/health") def health(): return jsonify({"ok": True, "apps": len(CA.APPS), "mock_payments": MOCK_PAYMENTS}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5188, debug=False)