""" title: Stable Horde Image Generator author_url: https://yourwebsite.com author: yourname version: 0.0.1 license: MIT """ import json import httpx import asyncio from typing import List, AsyncGenerator, Callable, Awaitable from pydantic import BaseModel, Field class Pipe: class Valves(BaseModel): API_KEY: str = Field(default="", description="Stable Horde API key") def __init__(self): self.type = "manifold" self.name = "Stable Horde" self.valves = self.Valves() self.emitter = None async def emit_status(self, message: str = "", done: bool = False): if self.emitter: await self.emitter( { "type": "status", "data": { "description": message, "done": done, }, } ) def pipes(self) -> List[dict]: return [ {"id": "stablehorde-kandinsky", "name": "Kandinsky v2.2"}, {"id": "stablehorde-deliberate", "name": "Deliberate v2"}, {"id": "stablehorde-realistic", "name": "Realistic Vision v5.1"}, ] async def pipe( self, body: dict, __event_emitter__: Callable[[dict], Awaitable[None]] = None, ) -> AsyncGenerator[str, None]: self.emitter = __event_emitter__ api_key = self.valves.API_KEY.strip() if not api_key: yield "❌ Error: API_KEY 未设置" return model_id = body.get("model", "stablehorde-kandinsky").replace( "stablehorde-", "" ) prompt = body["messages"][-1]["content"] await self.emit_status(f"🎨 使用模型 {model_id} 生成图像...") payload = { "prompt": prompt, "params": { "n": 1, "width": 512, "height": 512, "models": [model_id], }, "nsfw": False, } headers = {"apikey": api_key, "Client-Agent": "OpenWebUI-StableHorde/0.2"} try: async with httpx.AsyncClient() as client: submit = await client.post( "https://stablehorde.net/api/v2/generate/async", json=payload, headers=headers, timeout=30, ) if submit.status_code != 202: yield f"❌ 提交失败: HTTP {submit.status_code} - {submit.text}" return job_id = submit.json()["id"] yield f"🕓 已提交任务，任务ID: {job_id}，等待生成..." # 轮询直到完成 for attempt in range(30): # 最多等待150秒 await asyncio.sleep(5) status = await client.get( f"https://stablehorde.net/api/v2/generate/status/{job_id}", headers=headers, ) status_data = status.json() if status_data.get("done", False): break generations = status_data.get("generations", []) if not generations: yield "⚠️ 没有生成图像" return for i, img in enumerate(generations, 1): img_url_or_data = img.get("img", "") mime = img.get("mime_type", "image/png") if img_url_or_data.startswith("http"): # 直接展示外部托管图像 yield f"图像 {i}：\n\n\n" elif img_url_or_data: # 是 base64 图像数据 yield f"图像 {i}：\n\n\n" else: yield f"⚠️ 图像 {i} 无数据" await self.emit_status("✅ 图像生成完成", done=True) except Exception as e: yield f"❌ 出错: {str(e)}" await self.emit_status("❌ 生成失败", done=True)