Van/攻城狮鹿鸣君
92a1b50f91
Replace provider-switching task nodes with provider-specific config nodes and stable generic image/video execution contracts. Add Agnes Image 2.1 and Video 2.0 using behavior verified against the live API. Constraint: Old workflows are intentionally unsupported; Agnes Image 2.0 remains documentation-only. Rejected: Single task nodes with provider dropdowns | provider-specific union schemas create serialization risk and cross-provider coupling. Confidence: high Scope-risk: broad Directive: New providers must register isolated operation adapters and reuse generic execution nodes without modifying existing provider packages. Tested: 120 pytest tests; Python compileall; JavaScript syntax; 16 workflow JSON validations; restarted ComfyUI smoke; live Agnes Image 2.1 T2I/I2I; live Agnes Video 2.0 T2V completion and I2V Data URI submission. Not-tested: User-driven visual inspection of every imported UI workflow and full completed Agnes I2V download.
86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Explicit paid/live Agnes API probes. Never prints credentials."""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from mediahub.providers.agnes.image_21 import build_i2i_payload, build_t2i_payload, generate_images
|
|
from mediahub.providers.agnes.video_20 import build_video_payload, final_url, query_video, submit_video, video_state
|
|
|
|
|
|
def load_key() -> str:
|
|
for line in (ROOT / ".api_env").read_text().splitlines():
|
|
if line.startswith("AGNES_API_KEY="):
|
|
return line.split("=", 1)[1].strip().strip("'\"")
|
|
raise RuntimeError("AGNES_API_KEY is missing from .api_env")
|
|
|
|
|
|
def probe_image(key: str) -> None:
|
|
sources, raw = generate_images(key, build_t2i_payload("A red cube on white", 1024, 768, "url"))
|
|
url = sources[0][1]
|
|
image = requests.get(url, timeout=120)
|
|
image.raise_for_status()
|
|
import base64
|
|
data_uri = "data:image/png;base64," + base64.b64encode(image.content).decode()
|
|
image_location = "extra_body.image"
|
|
payload = build_i2i_payload("Make the cube blue", 1024, 768, [data_uri], "url")
|
|
try:
|
|
i2i_sources, i2i_raw = generate_images(key, payload)
|
|
except requests.HTTPError:
|
|
image_location = "top-level image"
|
|
payload["image"] = payload.pop("extra_body")["image"]
|
|
payload["extra_body"] = {"response_format": "url"}
|
|
i2i_sources, i2i_raw = generate_images(key, payload)
|
|
print(json.dumps({
|
|
"t2i": {"data_items": len(raw.get("data") or []), "source_type": sources[0][0]},
|
|
"i2i": {"data_items": len(i2i_raw.get("data") or []), "source_type": i2i_sources[0][0], "image_location": image_location},
|
|
}, indent=2))
|
|
|
|
|
|
def probe_video(key: str) -> None:
|
|
import time
|
|
payload = build_video_payload("A red cube slowly rotates", 1152, 768, 81, 24, 30, False, 0, "", [], "t2v")
|
|
job = submit_video(key, payload)
|
|
deadline = time.monotonic() + 600
|
|
while True:
|
|
raw = query_video(key, job)
|
|
state = video_state(raw)
|
|
if state == "completed":
|
|
break
|
|
if state == "failed" or time.monotonic() >= deadline:
|
|
raise RuntimeError(f"Agnes video probe ended with state={state}")
|
|
time.sleep(5)
|
|
print(json.dumps({
|
|
"task_id_present": bool(job.task_id),
|
|
"video_id_present": bool(job.metadata.get("video_id")),
|
|
"final_query_status": raw.get("status"),
|
|
"final_url_field": next(key for key in ("video_url", "remixed_from_video_id", "url") if raw.get(key) == final_url("", job, raw)),
|
|
"response_fields": sorted(raw),
|
|
}, indent=2))
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--image", action="store_true")
|
|
parser.add_argument("--video", action="store_true")
|
|
args = parser.parse_args()
|
|
if not args.image and not args.video:
|
|
parser.error("choose --image and/or --video")
|
|
key = load_key()
|
|
if args.image:
|
|
probe_image(key)
|
|
if args.video:
|
|
probe_video(key)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|