ComfyUI 全自動画像量産パイプライン
効率化・API運用 完全ガイド 2026年版

import json, time, uuid, urllib.parse, urllib.request, threading
import websocket  # pip install websocket-client

class ComfyClient:
    def __init__(self, host="127.0.0.1:8188"):
        self.host = host
        self.cid  = str(uuid.uuid4())          # client_id は固定で使い回す
        self.http_timeout = 10                 # /prompt 投入は短く [10]

    # ---- HTTP ヘルパ ----
    def _get(self, path):
        with urllib.request.urlopen(f"http://{self.host}{path}", timeout=self.http_timeout) as r:
            return json.loads(r.read())

    def _post(self, path, obj):
        data = json.dumps(obj).encode()
        req  = urllib.request.Request(f"http://{self.host}{path}", data=data,
                                      headers={"Content-Type": "application/json"})
        with urllib.request.urlopen(req, timeout=self.http_timeout) as r:
            return json.loads(r.read())

    def queue_prompt(self, workflow):
        """API形式workflowを投入し prompt_id を返す"""
        res = self._post("/prompt", {"prompt": workflow, "client_id": self.cid})
        if "error" in res or res.get("node_errors"):
            # validationエラー → 二度と成功しない。即raiseしリトライさせない [10]
            raise ValidationError(res.get("node_errors") or res.get("error"))
        return res["prompt_id"]

    def get_history(self, pid):
        return self._get(f"/history/{pid}").get(pid)

    def get_image(self, filename, subfolder, ftype):
        q = urllib.parse.urlencode({"filename": filename,
                                    "subfolder": subfolder, "type": ftype})
        with urllib.request.urlopen(f"http://{self.host}/view?{q}", timeout=60) as r:
            return r.read()

    def free(self, unload_models=False):
        # メモリ番人と同じbody形式 [3]
        self._post("/free", {"unload_models": unload_models, "free_memory": True})

    # ---- 1ジョブ完了まで待つ（WS監視 + 再接続 + /history cross-check）----
    def run_and_wait(self, workflow, exec_timeout=900, max_reconnect=5):
        pid = self.queue_prompt(workflow)
        done = threading.Event()
        last_seen = {"t": time.time()}
        reconnects = {"n": 0}

        def on_message(ws, msg):
            if not isinstance(msg, str):       # バイナリ(プレビュー)は無視 [1]
                return
            d = json.loads(msg)
            last_seen["t"] = time.time()
            t = d.get("type")
            if t == "progress":
                p = d["data"]
                print(f"\r  {p['value']}/{p['max']}", end="", flush=True)
            elif t == "executing" and d["data"].get("node") is None \
                 and d["data"].get("prompt_id") == pid:
                done.set(); ws.close()          # node=null = 完了
            elif t == "execution_error" and d["data"].get("prompt_id") == pid:
                done.set(); ws.close()

        def on_error(ws, err):
            pass

        deadline = time.time() + exec_timeout
        while not done.is_set() and time.time() < deadline:
            ws = websocket.WebSocketApp(
                f"ws://{self.host}/ws?clientId={self.cid}",
                on_message=on_message, on_error=on_error)
            # ping_interval=30 で heartbeat、無音切断を検知 [10]
            t = threading.Thread(target=ws.run_forever,
                                 kwargs={"ping_interval": 30, "ping_timeout": 10},
                                 daemon=True)
            t.start()
            t.join(timeout=exec_timeout)
            if done.is_set():
                break
            # --- WSが切れた: ジョブが生きているか /history と /queue で確認 ---
            hist = self.get_history(pid)
            if hist and hist.get("status", {}).get("completed"):
                done.set(); break              # 実は完走していた → 失敗扱いしない [10]
            reconnects["n"] += 1
            if reconnects["n"] > max_reconnect:
                raise WSLost(f"prompt {pid} WS再接続上限超過")
            time.sleep(2)                       # 少し待って再接続

        hist = self.get_history(pid)
        if not hist:
            raise JobTimeout(f"prompt {pid} 実行タイムアウト")
        # outputs から画像filenameを集めて回収
        imgs = []
        for node_out in hist["outputs"].values():
            for im in node_out.get("images", []):
                imgs.append(self.get_image(im["filename"], im["subfolder"], im["type"]))
        return pid, imgs

class ValidationError(Exception): pass
class WSLost(Exception): pass
class JobTimeout(Exception): pass

import time, random
from comfy_client import ComfyClient, ValidationError, WSLost, JobTimeout

cli = ComfyClient("127.0.0.1:8188")

def set_seed(workflow, seed_node_id):
    s = random.randint(0, 2**31 - 1)
    workflow[seed_node_id]["inputs"]["seed"] = s      # 毎回必ず更新 [10]
    return s

def run_one(workflow, seed_node, max_retry=3):
    """1ジョブを賢くリトライ。validationエラーはリトライしない。"""
    attempt = 0
    while True:
        attempt += 1
        try:
            return cli.run_and_wait(workflow, exec_timeout=900)
        except ValidationError as e:
            # node_errors → 何度やっても同じ。即あきらめ人間に通知 [10]
            raise RuntimeError(f"VALIDATION(リトライ不可): {e}")
        except (WSLost, JobTimeout, OSError) as e:
            # ネット/WS/タイムアウト → 再試行OK
            if attempt > max_retry:
                raise
            print(f"  retry {attempt} ({type(e).__name__})")
            time.sleep(3)
        except Exception as e:
            # CUDA OOM等 → モデルをunloadしてから再試行 [3][9]
            msg = str(e).lower()
            if ("out of memory" in msg or "oom" in msg) and attempt <= max_retry:
                print("  OOM → /free(unload) して再試行")
                cli.free(unload_models=True)          # 断片化解消 [3]
                set_seed(workflow, seed_node)         # seedも変えておく
                time.sleep(5)
                continue
            raise

import io, torch, clip                     # pip install git+https://github.com/openai/CLIP
from PIL import Image
import torch.nn as nn

# LAION aesthetic predictor: CLIP ViT-L/14 embedding 上の線形回帰 [11]
device = "cuda" if torch.cuda.is_available() else "cpu"
clip_model, preprocess = clip.load("ViT-L/14", device=device)

class AestheticHead(nn.Module):           # sa_0_4_vit_l_14_linear.pth をロード [11]
    def __init__(self):
        super().__init__()
        self.fc = nn.Linear(768, 1)
    def forward(self, x): return self.fc(x)

head = AestheticHead().to(device)
head.load_state_dict(torch.load("sa_0_4_vit_l_14_linear.pth", map_location=device))
head.eval()

@torch.no_grad()
def aesthetic_score(png_bytes):
    img = preprocess(Image.open(io.BytesIO(png_bytes)).convert("RGB")).unsqueeze(0).to(device)
    feat = clip_model.encode_image(img).float()
    feat = feat / feat.norm(dim=-1, keepdim=True)   # 正規化が必須
    return head(feat).item()                        # 1.0〜10.0

# ---- 合格するまでseedを変えてリジェネ（最悪N回で打ち切り＝原価管理）----
def gen_until_pass(workflow, seed_node, min_aesthetic=5.5, max_tries=6):
    from job_runner import run_one, set_seed
    best = (-1.0, None)
    for i in range(max_tries):
        set_seed(workflow, seed_node)               # 毎回seed変更 [10]
        pid, imgs = run_one(workflow, seed_node)
        if not imgs:
            continue
        score = aesthetic_score(imgs[0])
        print(f"  try{i+1} aesthetic={score:.2f}")
        if score >= min_aesthetic:
            return pid, imgs[0], score              # 合格
        if score > best[0]:
            best = (score, imgs[0])
    # 規定回数で合格せず → ベストを返す（無限ループ防止＝原価上限） [#1]
    return None, best[1], best[0]

import csv, json, copy, sqlite3, datetime, pathlib
from aesthetic_gate import gen_until_pass
from comfy_client import ComfyClient

cli = ComfyClient("127.0.0.1:8188")
BASE = json.load(open("golden_api.json", encoding="utf-8"))  # GOLDEN workflow [17]
SEED_NODE, POS_NODE, CKPT_NODE = "3", "6", "4"               # 自環境のノードIDに合わせる
OUT = pathlib.Path(r"D:\ComfyUI\output\night")               # 出力先 [17]
OUT.mkdir(parents=True, exist_ok=True)

db = sqlite3.connect("night.db")
db.execute("""CREATE TABLE IF NOT EXISTS gen(
    ts TEXT, row INT, prompt TEXT, model TEXT, aesthetic REAL, file TEXT, passed INT)""")

def build(row):
    wf = copy.deepcopy(BASE)
    wf[POS_NODE]["inputs"]["text"]  = row["prompt"]
    wf[CKPT_NODE]["inputs"]["ckpt_name"] = row.get("model", "waiIllustriousSDXL_v160.safetensors")
    return wf

with open("queue.csv", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))           # 列: prompt, model, count

n = 0
for i, row in enumerate(rows):
    for _ in range(int(row.get("count", 1))):
        wf = build(row)
        pid, img, score = gen_until_pass(wf, SEED_NODE, min_aesthetic=5.5, max_tries=6)
        passed = 1 if pid else 0
        fname = OUT / f"{datetime.datetime.now():%Y%m%d_%H%M%S}_{i}_{score:.1f}.png"
        if img:
            fname.write_bytes(img)
        db.execute("INSERT INTO gen VALUES(?,?,?,?,?,?,?)",
                   (datetime.datetime.now().isoformat(), i, row["prompt"],
                    row.get("model",""), score, str(fname), passed))
        db.commit(); n += 1
        # 適宜メモリ回収（メモリ番人と二重でも害なし）[3]
        if n % 20 == 0:
            cli.free(unload_models=False)

print(f"夜間バッチ完了: {n}枚処理 / DB=night.db")

schtasks /Create /TN "ComfyNightBatch" /SC DAILY /ST 02:00 ^
  /TR "cmd /c set PYTORCH_CUDA_ALLOC_CONF=expandable_segments:True&& python D:\proj\night_batch.py" ^
  /RL HIGHEST /F