diff --git a/README.md b/README.md index 1781818..4638be9 100644 --- a/README.md +++ b/README.md @@ -35,6 +35,13 @@ pip install -r comfyui-PD_comfy-api-node/requirements.txt ## 📝 更新日志 +### 2026-10-11 - Grok Video +- 新增 `PD- grok video comfyapi`（`PDGrokVideoAPIKey`），位于 `PD_Tools/Video_Generation`，手动填写 Comfy.org API Key。 +- 支持 `grok-imagine-video` / `grok-imagine-video-1.5` / `grok-imagine-video-1.5-lite`，文生视频或单张起始图生视频，1–15 秒；参数与当前官方 Grok Video 节点一致。普通版不支持 1080p，1.5 系列有起始图时可省略提示词，seed 不保证输出复现。 +- 输出 `VIDEO` 可直接连接 Save Video，`info` 包含任务 ID、模型、服务返回的 usage，以及响应提供的实际 credits。费用角标明确标为估算。 +- 一次执行只提交一次生成请求；轮询、下载期间每秒显示耗时，每 0.1 秒检查取消，整次执行限时 600 秒，不自动重试付费请求。取消不保证退还已受理任务费用。 +- 验证：`python custom_nodes/comfyui-PD_comfy-api-node/tests/test_grok_video.py`。离线覆盖三个模型文/图生视频、MP4 输出可解码、任务/费用信息、提交/轮询/下载取消、超时不重试及请求前校验。本次没有真实付费生成。 + ### 2026-10-11 - Grok 文生图 - 新增 `PDGrokImageAPIKey` 文生图节点，复用现有 Grok 模型列表；API Key 由节点输入提供。同步更新 Grok 编辑节点的取消检查与失败状态。 diff --git a/py/VIDEO_Grok_apikey.py b/py/VIDEO_Grok_apikey.py new file mode 100644 index 0000000..edfc2b8 --- /dev/null +++ b/py/VIDEO_Grok_apikey.py @@ -0,0 +1,79 @@ +import io +import json + +from comfy_api.latest import IO, VideoFromFile +from comfy_api_nodes.nodes_grok import GrokVideoNode, tensor_to_base64_string + +from .OPENAI_latest_apikey import BASE_URL, _Request + + +class PDGrokVideoAPIKey(IO.ComfyNode): + @classmethod + def define_schema(cls): + schema = GrokVideoNode.define_schema() + schema.node_id = "PDGrokVideoAPIKey" + schema.display_name = "PD- grok video comfyapi" + schema.category = "PD_Tools/Video_Generation" + schema.inputs.insert(0, IO.String.Input("comfy_api_key", default="", tooltip="Comfy.org API Key")) + schema.outputs.append(IO.String.Output(display_name="info")) + schema.hidden = [IO.Hidden.unique_id] + schema.is_api_node = False + schema.price_badge.expr = schema.price_badge.expr.replace( + '"usd": $is15 ? $total * 1.43 : $total}', + '"usd": $is15 ? $total * 1.43 : $total, "format": {"approximate": true}}', + ) + return schema + + @classmethod + def execute(cls, comfy_api_key, model, prompt, resolution, aspect_ratio, duration, seed, image=None): + if model not in ("grok-imagine-video", "grok-imagine-video-1.5", "grok-imagine-video-1.5-lite"): + raise ValueError(f"Unknown Grok video model: {model}") + if resolution == "1080p" and model == "grok-imagine-video": + raise ValueError("grok-imagine-video 不支持 1080p。") + if not 1 <= duration <= 15: + raise ValueError("视频时长须为 1–15 秒。") + if image is not None and image.shape[0] != 1: + raise ValueError("Grok Video 只支持一张起始图。") + if not prompt.strip() and (image is None or model == "grok-imagine-video"): + raise ValueError("请填写提示词。") + payload = {"model": model, "prompt": prompt, "resolution": resolution, + "duration": duration, "seed": seed} + if aspect_ratio != "auto": + payload["aspect_ratio"] = aspect_ratio + with _Request(cls.hidden.unique_id, comfy_api_key) as api: + api.check() + if image is not None: + payload["image"] = {"url": "data:image/png;base64," + tensor_to_base64_string(image)} + created = api.request("POST", BASE_URL + "/proxy/xai/v1/videos/generations", json=payload) + api.response_id = created.json().get("request_id", "") + if not api.response_id: + raise ValueError("Grok 响应缺少 request_id。") + while True: + response = api.request("GET", BASE_URL + "/proxy/xai/v1/videos/" + api.response_id) + result = response.json() + status = result.get("status") + if status in ("failed", "error", "cancelled", "canceled", "expired") or result.get("error"): + raise ValueError(f"Grok video {status}: {result.get('error', '')}") + video = result.get("video") or {} + if video.get("url") and status in (None, "done", "complete", "completed", "success", "succeeded"): + break + api.sleep(2) + api.usage = result.get("usage") or {} + info = {"request_id": api.response_id, "model": model, "usage": api.usage} + credits = response.headers.get("X-Comfy-Credits-Used") or created.headers.get("X-Comfy-Credits-Used") + if credits is not None: + info["credits_used"] = credits + url = video["url"] + if url.startswith("/proxy/"): + url = BASE_URL + url + downloaded = api.request("GET", url) + api.check() + output = VideoFromFile(io.BytesIO(downloaded.content)) + # Reject invalid downloads before reporting success. + output.get_dimensions() + api.check() + return IO.NodeOutput(output, json.dumps(info, ensure_ascii=False)) + + +NODE_CLASS_MAPPINGS = {"PDGrokVideoAPIKey": PDGrokVideoAPIKey} +NODE_DISPLAY_NAME_MAPPINGS = {"PDGrokVideoAPIKey": "PD- grok video comfyapi"} diff --git a/tests/test_grok_video.py b/tests/test_grok_video.py new file mode 100644 index 0000000..280e5f3 --- /dev/null +++ b/tests/test_grok_video.py @@ -0,0 +1,168 @@ +import importlib +import io +import json +import sys +import threading +import time +import types +import unittest +from pathlib import Path +from unittest.mock import patch + +import av +import numpy as np +import requests +import torch + +sys.path.insert(0, str(Path(__file__).resolve().parents[3])) +sys.argv = [sys.argv[0], "--cpu"] +package = types.ModuleType("pd_grok_video_test") +package.__path__ = [str(Path(__file__).resolve().parents[1])] +sys.modules[package.__name__] = package +module = importlib.import_module("pd_grok_video_test.py.VIDEO_Grok_apikey") +common = importlib.import_module("pd_grok_video_test.py.OPENAI_latest_apikey") +Node = module.PDGrokVideoAPIKey +Node.hidden = types.SimpleNamespace(unique_id="test-node") + + +def make_video(): + buffer = io.BytesIO() + with av.open(buffer, "w", format="mp4") as container: + stream = container.add_stream("mpeg4", rate=24) + stream.width = stream.height = 16 + stream.pix_fmt = "yuv420p" + frame = av.VideoFrame.from_ndarray(np.zeros((16, 16, 3), dtype=np.uint8), format="rgb24") + for packet in stream.encode(frame): + container.mux(packet) + for packet in stream.encode(): + container.mux(packet) + return buffer.getvalue() + + +class Response: + def __init__(self, result=None): + self.result = result or {} + self.headers = {"X-Comfy-Credits-Used": "12.3"} + self.content = make_video() + + def json(self): + return self.result + + def raise_for_status(self): + pass + + +class Session: + def __init__(self, slow=None, timeout=False, failed=False): + self.calls = [] + self.closed = threading.Event() + self.slow, self.timeout, self.failed = slow, timeout, failed + + def request(self, method, url, **kwargs): + self.calls.append((method, url, kwargs)) + stage = "submit" if method == "POST" else ("download" if "example.test" in url else "poll") + if self.timeout: + raise requests.Timeout("simulated") + if stage == self.slow: + self.closed.wait(5) + if stage == "submit": + return Response({"request_id": "test-task"}) + if stage == "poll": + if self.failed: + return Response({"status": "failed", "error": "simulated failure"}) + return Response({"status": "done", "video": {"url": "https://example.test/video.mp4"}, + "usage": {"cost_in_usd_ticks": 1000}}) + return Response() + + def close(self): + self.closed.set() + + +class GrokVideoTests(unittest.TestCase): + def setUp(self): + self.cancel = threading.Event() + self.events = [] + self.patches = [ + patch.object(common.PromptServer, "instance", types.SimpleNamespace(send_sync=lambda event, data: self.events.append(data)), create=True), + patch.object(common.comfy.model_management, "throw_exception_if_processing_interrupted", self.check), + ] + for item in self.patches: + item.start() + + def tearDown(self): + for item in reversed(self.patches): + item.stop() + + def check(self): + if self.cancel.is_set(): + raise common.comfy.model_management.InterruptProcessingException() + + def run_node(self, **kwargs): + values = dict(comfy_api_key="test-key", model="grok-imagine-video-1.5-lite", prompt="test prompt", + resolution="720p", aspect_ratio="auto", duration=6, seed=42) + values.update(kwargs) + return Node.execute(**values) + + def test_models_text_and_image_video_output(self): + for model in ("grok-imagine-video", "grok-imagine-video-1.5", "grok-imagine-video-1.5-lite"): + for image in (None, torch.zeros((1, 8, 8, 3))): + session = Session() + with patch.object(common.requests, "Session", return_value=session): + result = self.run_node(model=model, image=image) + self.assertEqual(result.result[0].get_dimensions(), (16, 16)) + info = json.loads(result.result[1]) + self.assertEqual(info["request_id"], "test-task") + self.assertEqual(info["credits_used"], "12.3") + self.assertEqual([item[0] for item in session.calls], ["POST", "GET", "GET"]) + payload = session.calls[0][2]["json"] + self.assertEqual(payload["model"], model) + self.assertEqual(payload["seed"], 42) + self.assertEqual("image" in payload, image is not None) + self.assertNotIn("aspect_ratio", payload) + self.assertEqual(session.calls[-1][2]["headers"], {}) + + def test_cancel_submit_poll_download(self): + for stage in ("submit", "poll", "download"): + self.cancel.clear() + self.events.clear() + errors = [] + session = Session(slow=stage) + def run(): + try: + self.run_node() + except common.comfy.model_management.InterruptProcessingException as exc: + errors.append(exc) + with patch.object(common.requests, "Session", return_value=session): + worker = threading.Thread(target=run) + worker.start() + time.sleep(1.25) + started = time.monotonic() + self.cancel.set() + worker.join(0.5) + self.assertFalse(worker.is_alive()) + self.assertLess(time.monotonic() - started, 0.5) + self.assertEqual(len(errors), 1) + self.assertTrue(session.closed.is_set()) + self.assertEqual(sum(item[0] == "POST" for item in session.calls), 1) + self.assertTrue(self.events[-1]["text"].startswith("Cancelled")) + self.assertGreaterEqual(sum(item["text"].startswith("Running") for item in self.events), 2) + + def test_timeout_failure_and_validation(self): + for session, error in ((Session(timeout=True), requests.Timeout), (Session(failed=True), ValueError)): + with patch.object(common.requests, "Session", return_value=session), self.assertRaises(error): + self.run_node() + self.assertEqual(sum(item[0] == "POST" for item in session.calls), 1) + session = Session() + with patch.object(common.requests, "Session", return_value=session): + for kwargs in ({"model": "grok-imagine-video", "resolution": "1080p"}, + {"image": torch.zeros((2, 8, 8, 3))}, {"prompt": ""}, {"duration": 16}): + with self.assertRaises(ValueError): + self.run_node(**kwargs) + self.cancel.set() + with self.assertRaises(common.comfy.model_management.InterruptProcessingException): + self.run_node() + self.assertEqual(session.calls, []) + + +if __name__ == "__main__": + unittest.main(argv=[sys.argv[0]])