DR: ComfyUI API自動化・大量生成パイプライン 完全攻略（2026-04-28）

# comfyui_client.py import json, uuid, time, urllib.request, urllib.parse, pathlib import websocket # pip install websocket-client class ComfyUIClient: """ComfyUI HTTP+WebSocket クライアント基盤""" def __init__(self, host="127.0.0.1:8188"): self.host = host self.client_id = str(uuid.uuid4()) self.ws = None self._completed = set() # チェックポイント def _post(self, path, data): req = urllib.request.Request( f"http://{self.host}{path}", data=json.dumps(data).encode(), headers={"Content-Type": "application/json"}, ) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read()) def _get(self, path): with urllib.request.urlopen(f"http://{self.host}{path}", timeout=30) as r: return json.loads(r.read()) def queue_prompt(self, workflow: dict) -> str: """ワークフロー投入 → prompt_id 返却""" payload = {"prompt": workflow, "client_id": self.client_id} result = self._post("/prompt", payload) if result.get("node_errors"): raise ValueError(f"ノードエラー: {result['node_errors']}") return result["prompt_id"] def wait_for_completion(self, prompt_id: str, timeout: int = 600) -> bool: """WebSocketで完了まで待機。True=成功 False=エラー""" if self.ws is None: self.ws = websocket.WebSocket() self.ws.connect(f"ws://{self.host}/ws?clientId={self.client_id}") self.ws.settimeout(timeout) while True: msg = self.ws.recv() if not isinstance(msg, str): continue data = json.loads(msg) if data["type"] == "executing": node = data["data"].get("node") pid = data["data"].get("prompt_id") if node is None and pid == prompt_id: return True # 完了 elif data["type"] == "execution_error": print(f"[ERROR] {data['data']}") return False def get_output_images(self, prompt_id: str) -> list: """履歴から出力画像のパラメータリストを返す""" history = self._get(f"/history/{prompt_id}") outputs = [] for node_id, node_out in history.get(prompt_id, {}).get("outputs", {}).items(): for img in node_out.get("images", []): outputs.append(img) return outputs def download_image(self, img_info: dict, save_path: str): """画像をローカルに保存""" params = urllib.parse.urlencode({ "filename": img_info["filename"], "subfolder": img_info.get("subfolder", ""), "type": img_info.get("type", "output"), }) url = f"http://{self.host}/view?{params}" with urllib.request.urlopen(url, timeout=60) as r: pathlib.Path(save_path).write_bytes(r.read()) def free_vram(self): """VRAMキャッシュ解放（100枚ごとに呼ぶ）""" try: self._post("/free", {"unload_models": False, "free_memory": True}) except Exception: pass def health_check(self) -> bool: """サーバー生存確認""" try: self._get("/system_stats") return True except Exception: return False def close(self): if self.ws: self.ws.close()

# batch_generate.py import csv, json, time, pathlib from comfyui_client import ComfyUIClient # ===== 設定 ===== COMFYUI_HOST = "127.0.0.1:8188" WORKFLOW_JSON = "D:/workflows/pony_t2i_api.json" # API Formatで保存したJSON PROMPTS_CSV = "D:/prompts/prompts.csv" # プロンプトリスト OUTPUT_DIR = pathlib.Path("D:/generated") CHECKPOINT_FILE = pathlib.Path("D:/generated/checkpoint.json") VRAM_FREE_EVERY = 50 # N枚ごとにVRAM解放 RETRY_MAX = 3 # ===== CSV形式 ===== # id, prompt, negative_prompt, seed, cfg, steps, style # 例: 1,"1girl, masterpiece, solo","lowres, bad anatomy",42,7,20,fantasy def load_checkpoint(): if CHECKPOINT_FILE.exists(): return set(json.loads(CHECKPOINT_FILE.read_text())) return set() def save_checkpoint(done_ids): CHECKPOINT_FILE.write_text(json.dumps(list(done_ids))) def load_workflow(): return json.loads(pathlib.Path(WORKFLOW_JSON).read_text()) def apply_prompt_to_workflow(workflow, row): """ワークフロー内の特定ノードにCSV値を注入""" wf = json.loads(json.dumps(workflow)) # ディープコピー for node_id, node in wf.items(): cls = node.get("class_type", "") # CLIPTextEncode（ポジティブプロンプト）→ ノードタイトルで判断 if cls == "CLIPTextEncode": meta = node.get("_meta", {}) if meta.get("title", "").lower() in ["positive", "prompt", "clip text encode (prompt)"]: node["inputs"]["text"] = row["prompt"] elif meta.get("title", "").lower() in ["negative", "neg", "negative prompt"]: node["inputs"]["text"] = row.get("negative_prompt", "lowres, bad quality") # KSampler にseed/cfg/stepsを注入 if cls == "KSampler": if row.get("seed"): node["inputs"]["seed"] = int(row["seed"]) if row.get("cfg"): node["inputs"]["cfg"] = float(row["cfg"]) if row.get("steps"): node["inputs"]["steps"] = int(row["steps"]) return wf def main(): OUTPUT_DIR.mkdir(parents=True, exist_ok=True) done_ids = load_checkpoint() base_workflow = load_workflow() client = ComfyUIClient(COMFYUI_HOST) with open(PROMPTS_CSV, encoding="utf-8") as f: rows = list(csv.DictReader(f)) print(f"総プロンプト数: {len(rows)} / 完了済み: {len(done_ids)}") count = 0 for row in rows: row_id = row["id"] if row_id in done_ids: print(f"[SKIP] id={row_id}") continue # サーバー確認 if not client.health_check(): print("[WARN] ComfyUI応答なし。30秒待機...") time.sleep(30) if not client.health_check(): raise RuntimeError("ComfyUIサーバーが起動していません") workflow = apply_prompt_to_workflow(base_workflow, row) # リトライ付き生成 for attempt in range(RETRY_MAX): try: prompt_id = client.queue_prompt(workflow) success = client.wait_for_completion(prompt_id) if not success: raise RuntimeError("生成エラー") images = client.get_output_images(prompt_id) style_dir = OUTPUT_DIR / row.get("style", "default") style_dir.mkdir(parents=True, exist_ok=True) for i, img in enumerate(images): save_path = style_dir / f"{row_id}_{i:03d}.png" client.download_image(img, str(save_path)) print(f"[OK] {save_path}") done_ids.add(row_id) save_checkpoint(done_ids) count += 1 break except Exception as e: print(f"[RETRY {attempt+1}/{RETRY_MAX}] id={row_id}: {e}") time.sleep(10) # VRAM解放 if count % VRAM_FREE_EVERY == 0 and count > 0: client.free_vram() print(f"[VRAM解放] {count}枚完了") client.close() print(f"\n完了: 合計{count}枚生成") if __name__ == "__main__": main()

# quality_filter.py # 前提: pip install torch transformers pillow aesthetic-predictor import pathlib, shutil, json from PIL import Image # ----- 方法A: CLIP + aesthetic-predictor でスコア計算 ----- # pip install git+https://github.com/christophschuhmann/improved-aesthetic-predictor import torch, clip from aesthetics_predictor import AestheticsPredictor # improved-aesthetic-predictor class AestheticFilter: def __init__(self, threshold=5.0, device="cuda"): self.threshold = threshold self.device = device self.model, self.preprocess = clip.load("ViT-L/14", device=device) self.predictor = AestheticsPredictor().to(device) self.predictor.eval() def score(self, image_path: str) -> float: """0〜10のAestheticスコアを返す""" img = Image.open(image_path).convert("RGB") img_tensor = self.preprocess(img).unsqueeze(0).to(self.device) with torch.no_grad(): emb = self.model.encode_image(img_tensor) emb = emb / emb.norm(dim=-1, keepdim=True) score = self.predictor(emb.float()) return float(score.item()) def filter_directory(self, input_dir: str, accept_dir: str, reject_dir: str): """ディレクトリ内画像を採点→振り分け""" input_path = pathlib.Path(input_dir) accept_path = pathlib.Path(accept_dir) reject_path = pathlib.Path(reject_dir) accept_path.mkdir(parents=True, exist_ok=True) reject_path.mkdir(parents=True, exist_ok=True) results = [] for img_file in sorted(input_path.glob("*.png")): try: s = self.score(str(img_file)) dest = accept_path if s >= self.threshold else reject_path shutil.copy2(img_file, dest / img_file.name) results.append({"file": img_file.name, "score": round(s, 3), "result": "ACCEPT" if s >= self.threshold else "REJECT"}) print(f" {img_file.name}: {s:.2f} → {'ACCEPT' if s >= self.threshold else 'REJECT'}") except Exception as e: print(f" [ERROR] {img_file.name}: {e}") # レポート出力 report_path = accept_path.parent / "filter_report.json" report_path.write_text(json.dumps(results, indent=2, ensure_ascii=False)) accept_count = sum(1 for r in results if r["result"] == "ACCEPT") print(f"\nフィルタ結果: {accept_count}/{len(results)} 枚ACCEPT (threshold={self.threshold})") return results # ----- 方法B: ComfyUI内ワークフロー完結フィルタ（ノード構成）----- # ワークフロー内で以下のノードを連結： # [KSampler] → [VAEDecode] → [CalculateAestheticScore] → [ImageFilterByFloatScoreNode(min=5.0)] → [SaveImage] # ImageFilterByFloatScoreNodeのmin_scoreを5.0に設定すると自動リジェクト # これによりPython側は合格画像のみを受け取る設計が可能 if __name__ == "__main__": f = AestheticFilter(threshold=5.0) f.filter_directory( input_dir="D:/generated/raw", accept_dir="D:/generated/accepted", reject_dir="D:/generated/rejected" )

# multi_gpu_launcher.py # ComfyUIを2ポートで2プロセス起動し、Pythonスレッドから並列投入する import subprocess, threading, pathlib, csv, json, time from comfyui_client import ComfyUIClient COMFYUI_PATH = r"C:\ddrive\AI\ComfyUI_portable\ComfyUI" PYTHON_PATH = r"C:\ddrive\AI\ComfyUI_portable\python_embeded\python.exe" INSTANCES = [ {"port": 8188, "cuda_device": 0, "host": "127.0.0.1:8188"}, {"port": 8189, "cuda_device": 1, "host": "127.0.0.1:8189"}, ] def start_comfyui_instance(port: int, cuda_device: int): """ComfyUIをバックグラウンドプロセスで起動""" cmd = [ PYTHON_PATH, "-s", pathlib.Path(COMFYUI_PATH) / "main.py", "--port", str(port), "--cuda-device", str(cuda_device), "--preview-method", "none", # プレビューをOFFにして高速化 "--dont-print-server", ] proc = subprocess.Popen(cmd, cwd=COMFYUI_PATH) print(f"[起動] ComfyUI port={port} GPU={cuda_device} PID={proc.pid}") return proc def worker_thread(client: ComfyUIClient, job_queue: list, results: list, lock: threading.Lock): """1GPUのワーカースレッド""" base_workflow = json.loads(pathlib.Path("D:/workflows/pony_t2i_api.json").read_text()) while True: with lock: if not job_queue: break row = job_queue.pop(0) # ワークフローにプロンプト注入（apply_prompt_to_workflow関数を流用） wf = json.loads(json.dumps(base_workflow)) for nid, node in wf.items(): if node.get("class_type") == "CLIPTextEncode": meta = node.get("_meta", {}) if "positive" in meta.get("title", "").lower(): node["inputs"]["text"] = row["prompt"] try: pid = client.queue_prompt(wf) client.wait_for_completion(pid) images = client.get_output_images(pid) for i, img in enumerate(images): save_path = f"D:/generated/gpu{client.host[-1]}/{row['id']}_{i:03d}.png" pathlib.Path(save_path).parent.mkdir(parents=True, exist_ok=True) client.download_image(img, save_path) with lock: results.append({"id": row["id"], "status": "ok"}) except Exception as e: with lock: results.append({"id": row["id"], "status": f"error: {e}"}) def run_parallel(prompts_csv: str): """全プロンプトを2GPUに分散処理""" # ComfyUI 2インスタンス起動 procs = [start_comfyui_instance(inst["port"], inst["cuda_device"]) for inst in INSTANCES] print("ComfyUI起動待機 (20秒)...") time.sleep(20) # プロンプトロード with open(prompts_csv, encoding="utf-8") as f: job_queue = list(csv.DictReader(f)) print(f"総ジョブ数: {len(job_queue)}") results = [] lock = threading.Lock() clients = [ComfyUIClient(inst["host"]) for inst in INSTANCES] # ヘルスチェック for c in clients: if not c.health_check(): raise RuntimeError(f"{c.host} が応答しません") # スレッド起動 threads = [ threading.Thread(target=worker_thread, args=(clients[i], job_queue, results, lock)) for i in range(len(INSTANCES)) ] for t in threads: t.start() for t in threads: t.join() for c in clients: c.close() for p in procs: p.terminate() ok_count = sum(1 for r in results if r["status"] == "ok") print(f"\n完了: {ok_count}/{len(results)} 枚成功") return results if __name__ == "__main__": run_parallel("D:/prompts/prompts.csv")

# full_pipeline.py # pip install pillow piexif requests import zipfile, pathlib, json, datetime, requests import piexif, piexif.helper from PIL import Image # ===== 設定 ===== ACCEPTED_DIR = pathlib.Path("D:/generated/accepted") OUTPUT_DIR = pathlib.Path("D:/output") SET_NAME = "fantasy_girls_vol01" UPLOAD_URL = "https://your-upload-endpoint.com/api/upload" # FTP/SFTPも可 UPLOAD_TOKEN = "YOUR_UPLOAD_TOKEN" def add_exif_metadata(image_path: str, metadata: dict): """PNG/JPEGにEXIFメタデータを埋め込む""" img = Image.open(image_path) # UserComment にJSON文字列で埋め込み comment = json.dumps(metadata, ensure_ascii=False) exif_bytes = piexif.dump({ "Exif": { piexif.ExifIFD.UserComment: piexif.helper.UserComment.dump(comment, encoding="unicode"), piexif.ExifIFD.DateTimeOriginal: datetime.datetime.now().strftime("%Y:%m:%d %H:%M:%S"), }, "0th": { piexif.ImageIFD.Software: b"ComfyUI Auto Pipeline", piexif.ImageIFD.ImageDescription: metadata.get("prompt", "").encode("utf-8")[:200], } }) # PNG→JPEGに変換してEXIF付与（PNGはEXIF非対応なため） save_path = pathlib.Path(image_path).with_suffix(".jpg") img.convert("RGB").save(str(save_path), "JPEG", quality=95, exif=exif_bytes) return str(save_path) def create_zip(source_dir: pathlib.Path, set_name: str) -> pathlib.Path: """採点済み画像をZIPにまとめる""" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) today = datetime.datetime.now().strftime("%Y%m%d") zip_path = OUTPUT_DIR / f"{set_name}_{today}.zip" images = sorted(source_dir.glob("*.png")) + sorted(source_dir.glob("*.jpg")) with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf: for img in images: zf.write(img, img.name) size_mb = zip_path.stat().st_size / 1024 / 1024 print(f"[ZIP] {zip_path.name}: {len(images)}枚 / {size_mb:.1f}MB") return zip_path def upload_zip(zip_path: pathlib.Path) -> bool: """ZIPをサーバーにアップロード（HTTP POST）""" with open(zip_path, "rb") as f: resp = requests.post( UPLOAD_URL, files={"file": (zip_path.name, f, "application/zip")}, headers={"Authorization": f"Bearer {UPLOAD_TOKEN}"}, timeout=300, ) if resp.status_code == 200: print(f"[UPLOAD] 成功: {zip_path.name}") return True else: print(f"[UPLOAD] 失敗: {resp.status_code} {resp.text}") return False def generate_catalog_json(source_dir: pathlib.Path, set_name: str) -> pathlib.Path: """DLsite/FANZA提出用カタログJSON生成""" images = sorted(source_dir.glob("*.jpg")) + sorted(source_dir.glob("*.png")) catalog = { "set_name": set_name, "created_at": datetime.datetime.now().isoformat(), "total_images": len(images), "images": [{"filename": img.name, "index": i+1} for i, img in enumerate(images)] } catalog_path = OUTPUT_DIR / f"{set_name}_catalog.json" catalog_path.write_text(json.dumps(catalog, indent=2, ensure_ascii=False)) return catalog_path def run_full_pipeline(prompts_json: str = None): """フルパイプライン実行""" print("=== Full Pipeline Start ===") # プロンプトのメタデータをロード（オプション） prompt_map = {} if prompts_json: prompt_map = {p["id"]: p for p in json.loads(pathlib.Path(prompts_json).read_text())} # 1. EXIF付与 print("[1/4] EXIFメタデータ付与...") for img_path in sorted(ACCEPTED_DIR.glob("*.png")): img_id = img_path.stem.split("_")[0] meta = prompt_map.get(img_id, {}) add_exif_metadata(str(img_path), { "prompt": meta.get("prompt", ""), "model": "Pony Diffusion V6", "set": SET_NAME, "generated_by": "ComfyUI Auto Pipeline", }) # 2. ZIP作成 print("[2/4] ZIP作成...") zip_path = create_zip(ACCEPTED_DIR, SET_NAME) # 3. カタログ生成 print("[3/4] カタログJSON生成...") catalog_path = generate_catalog_json(ACCEPTED_DIR, SET_NAME) print(f" → {catalog_path}") # 4. アップロード（オプション） print("[4/4] アップロード...") if UPLOAD_URL and UPLOAD_URL != "https://your-upload-endpoint.com/api/upload": upload_zip(zip_path) else: print(f" → アップロードURL未設定。ZIPは {zip_path} に保存済み") print("=== Pipeline Complete ===") return zip_path if __name__ == "__main__": run_full_pipeline("D:/prompts/prompts.json")

// ecosystem.config.js module.exports = { apps: [ { name: "comfy-batch", script: "D:/scripts/batch_generate.py", interpreter: "C:/ddrive/AI/ComfyUI_portable/python_embeded/python.exe", cwd: "D:/scripts", autorestart: false, // バッチ完了後に自動停止 env: { PYTHONIOENCODING: "utf-8" } }, { name: "comfy-pipeline", script: "D:/scripts/full_pipeline.py", interpreter: "C:/ddrive/AI/ComfyUI_portable/python_embeded/python.exe", cwd: "D:/scripts", autorestart: false, } ] }