587 lines
31 KiB
Python
587 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import asyncio
|
|
import ipaddress
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
import secrets
|
|
import socket
|
|
from pathlib import Path
|
|
from job_cancellation import dispatched as cancellation_dispatched, record as record_cancellation
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from urllib.parse import quote, urlencode, urlsplit
|
|
|
|
import httpx
|
|
from fastapi import Depends, FastAPI, Header, HTTPException, Response
|
|
from fastapi.responses import StreamingResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from workflow import MAX_VIDEO_DURATION_SECONDS, compile_workflow, load_registry, public_model, public_workflow, validate_h3_references
|
|
from h3_audio_intent import plan_h3_audio
|
|
from h3_audio_validation import AudioOutputValidator, attach_audio_policy
|
|
from h3_visual_guard import VISUAL_REVISION, validate_visual_dimensions
|
|
from h3_audio_conditioning import AudioConditioningError, prepare_audio_conditioning
|
|
from h3_audio_output_guard import AudioQualityError
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Provider:
|
|
id: str
|
|
name: str
|
|
base_url: str
|
|
api_key: str = ""
|
|
control_token_file: str = ""
|
|
|
|
|
|
class JobRequest(BaseModel):
|
|
provider_id: str = "default"
|
|
strict_provider: bool = False
|
|
workflow_key: str
|
|
prompt: str
|
|
negative_prompt: str = ""
|
|
input_images: list[str] = Field(default_factory=list, max_length=9)
|
|
width: int | None = Field(default=None, ge=64, le=8192)
|
|
height: int | None = Field(default=None, ge=64, le=8192)
|
|
duration: float | None = Field(default=None, gt=0, le=MAX_VIDEO_DURATION_SECONDS)
|
|
generate_audio: bool | None = None
|
|
seed: int | None = Field(default=None, ge=0)
|
|
batch_size: int = Field(default=1, ge=1, le=4)
|
|
denoise: float | None = Field(default=None, ge=0, le=1)
|
|
metadata: dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
app = FastAPI(title="Canvas ComfyUI Workflow Adapter", version="1.0.0")
|
|
audio_validator = AudioOutputValidator()
|
|
provider_selection_lock = asyncio.Lock()
|
|
provider_reservations: dict[str, int] = {}
|
|
|
|
|
|
def auto_balance_enabled() -> bool:
|
|
return os.getenv("COMFY_AUTO_BALANCE", "").strip().lower() in {"1", "true", "yes", "on"}
|
|
|
|
|
|
def providers() -> dict[str, Provider]:
|
|
raw = os.getenv("COMFY_PROVIDERS_JSON", "").strip()
|
|
rows: list[dict[str, Any]] = []
|
|
if raw:
|
|
parsed = json.loads(raw)
|
|
rows = parsed if isinstance(parsed, list) else []
|
|
elif os.getenv("COMFY_URL", "").strip():
|
|
rows = [{"id": "default", "name": "ComfyUI", "base_url": os.getenv("COMFY_URL"), "api_key": os.getenv("COMFY_API_KEY", "")}]
|
|
extra_path = Path('/app/huobao-extra-providers.json')
|
|
if extra_path.exists():
|
|
extra = json.loads(extra_path.read_text())
|
|
if not isinstance(extra, list) or any(not isinstance(row, dict) or row.get('id') != 'gpu1' for row in extra):
|
|
raise ValueError('Invalid Huobao extra provider configuration')
|
|
existing_ids = {row.get('id') for row in rows}
|
|
rows.extend(row for row in extra if row['id'] not in existing_ids)
|
|
result: dict[str, Provider] = {}
|
|
for row in rows:
|
|
provider = Provider(id=str(row.get("id") or "").strip(), name=str(row.get("name") or row.get("id") or "ComfyUI"), base_url=str(row.get("base_url") or "").strip().rstrip("/"), api_key=str(row.get("api_key") or ""), control_token_file=str(row.get("control_token_file") or ""))
|
|
if provider.id and provider.base_url.startswith(("http://", "https://")):
|
|
result[provider.id] = provider
|
|
return result
|
|
|
|
|
|
def require_token(authorization: str | None = Header(default=None)) -> None:
|
|
expected = os.getenv("COMFY_ADAPTER_TOKEN", "").strip()
|
|
if expected and not secrets.compare_digest(authorization or "", f"Bearer {expected}"):
|
|
raise HTTPException(status_code=401, detail="invalid adapter token")
|
|
|
|
|
|
def provider_headers(provider: Provider) -> dict[str, str]:
|
|
headers = {"Authorization": f"Bearer {provider.api_key}"} if provider.api_key else {}
|
|
if provider.control_token_file:
|
|
path = Path(provider.control_token_file)
|
|
try:
|
|
# Operator-mounted token only; it is never accepted from JobRequest
|
|
# or exposed by the public provider catalogue.
|
|
if path.parent != Path('/run/h3-workers') or path.name != provider.id + '.token':
|
|
raise ValueError('invalid worker control token path')
|
|
token = path.read_text().strip()
|
|
if not 16 <= len(token) <= 512 or any(ord(x) < 33 or ord(x) > 126 for x in token):
|
|
raise ValueError('invalid worker control token')
|
|
headers['X-H3-Canary-Token'] = token
|
|
except (OSError, ValueError):
|
|
raise HTTPException(status_code=503, detail="生成节点控制配置未就绪，任务未提交")
|
|
return headers
|
|
|
|
|
|
async def provider_queue_depth(provider: Provider) -> int | None:
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
|
response = await client.get(f"{provider.base_url}/queue", headers=provider_headers(provider))
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
except (httpx.HTTPError, ValueError, TypeError):
|
|
return None
|
|
return len(payload.get("queue_running") or []) + len(payload.get("queue_pending") or [])
|
|
|
|
|
|
def admission_ids(configured: dict[str, Provider]) -> set[str]:
|
|
path = os.getenv("COMFY_ADMISSION_FILE", "").strip()
|
|
if not path:
|
|
return set(configured)
|
|
try:
|
|
raw = Path(path).read_bytes()
|
|
if len(raw) > 4096:
|
|
raise ValueError("oversized admission configuration")
|
|
value = json.loads(raw)
|
|
if not isinstance(value, dict):
|
|
raise ValueError("invalid admission configuration")
|
|
allowed = value.get("admit")
|
|
if type(value.get("schema")) is not int or value.get("schema") != 1 or not isinstance(allowed, list) or any(not isinstance(x, str) for x in allowed):
|
|
raise ValueError("invalid admission configuration")
|
|
if len(set(allowed)) != len(allowed) or not set(allowed) <= set(configured):
|
|
raise ValueError("unknown or duplicate admission provider")
|
|
return set(allowed)
|
|
except (OSError, ValueError, TypeError):
|
|
# Drain changes affect only new submissions; job lookup always retains
|
|
# both stable provider IDs and their existing history/output routes.
|
|
raise HTTPException(status_code=503, detail="生成服务正在切换，任务尚未提交，请稍后重试")
|
|
|
|
|
|
async def select_provider(requested_provider_id: str, strict_provider: bool = False) -> Provider | None:
|
|
configured = providers()
|
|
requested = configured.get(requested_provider_id)
|
|
admitted = admission_ids(configured)
|
|
if requested is None:
|
|
return None
|
|
if strict_provider:
|
|
if requested_provider_id not in admitted:
|
|
raise HTTPException(status_code=503, detail="指定显卡正在维护或排空，任务未提交；请稍后重试或选择自动分配")
|
|
if await provider_queue_depth(requested) is None:
|
|
raise HTTPException(status_code=503, detail="指定显卡不可用或缺少所需工作流，任务未提交")
|
|
return requested
|
|
if requested is None or requested_provider_id != "default" or not auto_balance_enabled() or len(configured) < 2:
|
|
if requested_provider_id not in admitted:
|
|
raise HTTPException(status_code=503, detail="该生成节点正在排空，现有任务继续执行，新任务请稍后提交")
|
|
return requested
|
|
|
|
candidates = [provider for provider in configured.values() if provider.id in admitted]
|
|
if not candidates:
|
|
raise HTTPException(status_code=503, detail="生成节点正在排空，现有任务和结果保持不变")
|
|
depths = await asyncio.gather(*(provider_queue_depth(provider) for provider in candidates))
|
|
available = [
|
|
(depth + provider_reservations.get(provider.id, 0), 0 if provider.id == requested_provider_id else 1, provider)
|
|
for provider, depth in zip(candidates, depths)
|
|
if depth is not None
|
|
]
|
|
if not available:
|
|
if os.getenv("COMFY_ADMISSION_FILE", "").strip():
|
|
raise HTTPException(status_code=503, detail="生成节点尚未就绪，任务未提交，请稍后重试")
|
|
return requested
|
|
return min(available, key=lambda item: (item[0], item[1]))[2]
|
|
|
|
|
|
async def reserve_provider(requested_provider_id: str, strict_provider: bool = False) -> Provider | None:
|
|
async with provider_selection_lock:
|
|
provider = await select_provider(requested_provider_id, strict_provider)
|
|
if provider is not None:
|
|
provider_reservations[provider.id] = provider_reservations.get(provider.id, 0) + 1
|
|
return provider
|
|
|
|
|
|
async def release_provider(provider: Provider) -> None:
|
|
async with provider_selection_lock:
|
|
remaining = provider_reservations.get(provider.id, 0) - 1
|
|
if remaining > 0:
|
|
provider_reservations[provider.id] = remaining
|
|
else:
|
|
provider_reservations.pop(provider.id, None)
|
|
|
|
|
|
def input_host_allowlist() -> set[str]:
|
|
return {value.strip().lower().rstrip(".") for value in os.getenv("COMFY_INPUT_HOST_ALLOWLIST", "").split(",") if value.strip()}
|
|
|
|
|
|
async def validate_input_url(value: str) -> None:
|
|
parsed = urlsplit(value)
|
|
host = (parsed.hostname or "").lower().rstrip(".")
|
|
if parsed.scheme not in {"http", "https"} or not host or parsed.username or parsed.password:
|
|
raise HTTPException(status_code=400, detail="reference image URL is invalid")
|
|
if host in input_host_allowlist():
|
|
return
|
|
try:
|
|
infos = await asyncio.to_thread(socket.getaddrinfo, host, parsed.port or (443 if parsed.scheme == "https" else 80), type=socket.SOCK_STREAM)
|
|
except OSError as exc:
|
|
raise HTTPException(status_code=400, detail="reference image host cannot be resolved") from exc
|
|
if not infos:
|
|
raise HTTPException(status_code=400, detail="reference image host cannot be resolved")
|
|
for info in infos:
|
|
address = ipaddress.ip_address(info[4][0])
|
|
if not address.is_global:
|
|
raise HTTPException(status_code=400, detail="reference image URL cannot target a private address")
|
|
|
|
|
|
async def download_input_image(client: httpx.AsyncClient, value: str) -> tuple[bytes, str]:
|
|
await validate_input_url(value)
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
async with client.stream("GET", value, follow_redirects=False) as response:
|
|
response.raise_for_status()
|
|
if 300 <= response.status_code < 400:
|
|
raise HTTPException(status_code=400, detail="reference image redirects are not accepted")
|
|
mime_type = response.headers.get("content-type", "image/png").split(";", 1)[0]
|
|
if not mime_type.startswith("image/"):
|
|
raise HTTPException(status_code=400, detail="reference image URL did not return an image")
|
|
async for chunk in response.aiter_bytes():
|
|
total += len(chunk)
|
|
if total > 30 * 1024 * 1024:
|
|
raise HTTPException(status_code=413, detail="reference image exceeds 30 MB")
|
|
chunks.append(chunk)
|
|
return b"".join(chunks), mime_type
|
|
|
|
|
|
def encode_job(provider_id: str, prompt_id: str) -> str:
|
|
return base64.urlsafe_b64encode(f"{provider_id}:{prompt_id}".encode()).decode().rstrip("=")
|
|
|
|
|
|
def decode_job(job_id: str) -> tuple[Provider, str]:
|
|
if job_id.startswith(("h3a1.", "h3v2.")):
|
|
job_id = job_id.split(".", 1)[1]
|
|
try:
|
|
raw = base64.urlsafe_b64decode(job_id + "=" * (-len(job_id) % 4)).decode()
|
|
provider_id, prompt_id = raw.split(":", 1)
|
|
provider = providers()[provider_id]
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=404, detail="job not found") from exc
|
|
return provider, prompt_id
|
|
|
|
|
|
def output_validation_mode() -> str:
|
|
mode = os.getenv("H3_OUTPUT_VALIDATION_MODE", "guarded").strip()
|
|
if mode not in {"guarded", "prompt-only"}:
|
|
raise HTTPException(status_code=503, detail="视频处理模式配置无效，任务未提交")
|
|
return mode
|
|
|
|
|
|
@app.get("/health")
|
|
async def health() -> dict[str, Any]:
|
|
return {
|
|
"status": "ok",
|
|
"providers": len(providers()),
|
|
"workflows": len(load_registry()),
|
|
"autoBalance": auto_balance_enabled(),
|
|
"outputValidationMode": output_validation_mode(),
|
|
}
|
|
|
|
|
|
@app.get("/v1/providers", dependencies=[Depends(require_token)])
|
|
async def list_providers() -> dict[str, Any]:
|
|
return {"providers": [{"id": item.id, "name": item.name, "baseUrl": item.base_url} for item in providers().values()]}
|
|
|
|
|
|
@app.get("/v1/workflows", dependencies=[Depends(require_token)])
|
|
async def list_workflows() -> dict[str, Any]:
|
|
return {"workflows": [public_workflow(item) for item in load_registry().values()]}
|
|
|
|
|
|
@app.get("/v1/models", dependencies=[Depends(require_token)])
|
|
async def list_models() -> dict[str, Any]:
|
|
return {"object": "list", "data": [public_model(item) for item in load_registry().values()]}
|
|
|
|
|
|
async def upload_image(client: httpx.AsyncClient, provider: Provider, value: str, index: int, client_id: str) -> str:
|
|
value = value.strip()
|
|
if value.startswith("comfy://"):
|
|
return value.removeprefix("comfy://")
|
|
if value.startswith("data:"):
|
|
header, encoded = value.split(",", 1)
|
|
mime_type = header.split(";", 1)[0].removeprefix("data:") or "image/png"
|
|
if not mime_type.startswith("image/"):
|
|
raise HTTPException(status_code=400, detail="reference data URL must contain an image")
|
|
try:
|
|
data = base64.b64decode(encoded, validate=True)
|
|
except (ValueError, base64.binascii.Error) as exc:
|
|
raise HTTPException(status_code=400, detail="reference image data URL is invalid") from exc
|
|
elif value.startswith(("http://", "https://")):
|
|
data, mime_type = await download_input_image(client, value)
|
|
else:
|
|
raise HTTPException(status_code=400, detail="input_images must contain http(s), data, or comfy URLs")
|
|
if len(data) > 30 * 1024 * 1024:
|
|
raise HTTPException(status_code=413, detail="reference image exceeds 30 MB")
|
|
extension = mimetypes.guess_extension(mime_type) or ".png"
|
|
filename = f"{client_id}-reference-{index}{extension}"
|
|
response = await client.post(f"{provider.base_url}/upload/image", files={"image": (filename, data, mime_type)}, data={"overwrite": "true"}, headers=provider_headers(provider))
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
return str(payload.get("name") or filename)
|
|
|
|
|
|
def request_audio_plan(spec, request: JobRequest) -> dict:
|
|
if not (spec.raw.get("speech_prompt_policy") == "minimax_h3" and spec.raw.get("native_audio_control")):
|
|
return {"supported": False}
|
|
if request.generate_audio is False:
|
|
return {"supported": True, "status": "ready", "normalized_prompt": request.prompt,
|
|
"dialogue_base_prompt": request.prompt, "dialogue": [], "sound_effects": [],
|
|
"music": [], "audio_mode": "silent", "silent": True, "speech_requested": False,
|
|
"expected_transcript": "", "required_seconds": 0, "max_duration_seconds": 15, "issues": []}
|
|
plan = plan_h3_audio(request.prompt, float(request.duration or 5))
|
|
if output_validation_mode() == "prompt-only" and not plan["silent"] and plan["audio_mode"] == "silent":
|
|
plan["audio_mode"] = "effects"
|
|
return {"supported": True, **plan}
|
|
|
|
|
|
@app.post("/v1/audio-plan", dependencies=[Depends(require_token)])
|
|
async def preview_audio_plan(request: JobRequest) -> dict:
|
|
spec = load_registry().get(request.workflow_key)
|
|
if spec is None:
|
|
raise HTTPException(status_code=400, detail="workflow is not registered")
|
|
return request_audio_plan(spec, request)
|
|
|
|
|
|
@app.post("/v1/jobs", dependencies=[Depends(require_token)])
|
|
async def create_job(request: JobRequest) -> dict[str, Any]:
|
|
registry = load_registry()
|
|
spec = registry.get(request.workflow_key)
|
|
if spec is None:
|
|
raise HTTPException(status_code=400, detail="workflow is not registered")
|
|
try:
|
|
validate_h3_references(spec, len(request.input_images))
|
|
except ValueError as error:
|
|
raise HTTPException(status_code=400, detail=str(error)) from error
|
|
audio_plan = request_audio_plan(spec, request)
|
|
guarded_output = bool(audio_plan.get("supported")) and output_validation_mode() == "guarded"
|
|
if audio_plan.get("supported"):
|
|
if audio_plan.get("status") != "ready":
|
|
raise HTTPException(status_code=400, detail="H3_AUDIO_CONFIRMATION_REQUIRED: 请先确认台词及视频时长，再提交视频生成")
|
|
if not guarded_output:
|
|
request = request.model_copy(update={"prompt": audio_plan["normalized_prompt"]})
|
|
try:
|
|
if guarded_output:
|
|
validate_visual_dimensions(request.width, request.height)
|
|
await audio_validator.ready(visual_required=True, audio_required=audio_plan.get("audio_mode") in {"dialogue", "effects"})
|
|
if audio_plan.get("audio_mode") in {"dialogue", "effects"}:
|
|
prompt = await prepare_audio_conditioning(audio_plan, request.prompt, float(request.duration or 5), len(request.input_images))
|
|
request = request.model_copy(update={"prompt": prompt})
|
|
except AudioQualityError as error:
|
|
raise HTTPException(status_code=503 if error.code.endswith("_UNAVAILABLE") else 400, detail=str(error)) from error
|
|
except AudioConditioningError as error:
|
|
raise HTTPException(status_code=503 if error.code == "H3_AUDIO_NORMALIZER_UNAVAILABLE" else 400, detail=str(error)) from error
|
|
provider = await reserve_provider(request.provider_id, request.strict_provider)
|
|
if provider is None:
|
|
raise HTTPException(status_code=400, detail="provider is not configured")
|
|
try:
|
|
client_id = ("canvas-h3v2-" if guarded_output else "canvas-") + secrets.token_hex(8)
|
|
async with httpx.AsyncClient(timeout=60) as client:
|
|
uploaded = [await upload_image(client, provider, value, index, client_id) for index, value in enumerate(request.input_images, start=1)]
|
|
try:
|
|
workflow = compile_workflow(spec, request.model_dump(), uploaded, client_id, preserve_requested_audio=not guarded_output and bool(audio_plan.get("supported")))
|
|
if guarded_output:
|
|
attach_audio_policy(workflow, audio_plan, float(request.duration or 5), str(spec.raw.get("save_node")), visual_required=True)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
response = await client.post(f"{provider.base_url}/prompt", json={"prompt": workflow, "client_id": client_id}, headers=provider_headers(provider))
|
|
if response.status_code == 400 and spec.raw.get("speech_prompt_policy") == "minimax_h3":
|
|
try:
|
|
rejected = response.json()
|
|
except ValueError:
|
|
rejected = {}
|
|
if "h3_caption_pipeline" in (rejected.get("node_errors") or {}) or "H3CaptionPipeline" in str((rejected.get("error") or {}).get("details") or ""):
|
|
raise HTTPException(status_code=503, detail="字幕工作流预检未通过，处理组件、模型或字体尚未就绪；任务未进入 GPU 生成，请联系管理员。")
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
finally:
|
|
await release_provider(provider)
|
|
prompt_id = str(payload.get("prompt_id") or "").strip()
|
|
if not prompt_id:
|
|
raise HTTPException(status_code=502, detail="ComfyUI did not return prompt_id")
|
|
job_id = encode_job(provider.id, prompt_id)
|
|
persisted = True
|
|
if guarded_output:
|
|
job_id = "h3v2." + job_id
|
|
try:
|
|
audio_validator.register_policy(job_id, audio_plan, float(request.duration or 5), visual_required=True)
|
|
except AudioQualityError as error:
|
|
raise HTTPException(status_code=502, detail=str(error)) from error
|
|
except OSError:
|
|
# The provider already accepted this exact job. Return its ID rather
|
|
# than an ambiguous 500 inviting another GPU submission. Its graph
|
|
# carries the same policy; missing evidence still fails closed.
|
|
persisted = False
|
|
return {"id": job_id, "promptId": prompt_id, "providerId": provider.id, "workflowKey": spec.key, "workflowRevision": spec.revision, "status": "submitted", "policyPersisted": persisted}
|
|
|
|
|
|
def extract_outputs(entry: dict[str, Any], job_id: str) -> list[dict[str, Any]]:
|
|
result: list[dict[str, Any]] = []
|
|
for node in (entry.get("outputs") or {}).values():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
for field in ("images", "gifs", "audio"):
|
|
for item in node.get(field) or []:
|
|
if not isinstance(item, dict) or not item.get("filename"):
|
|
continue
|
|
filename = str(item["filename"])
|
|
mime_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
|
|
kind = "video" if mime_type.startswith("video/") or filename.lower().endswith((".gif", ".webp")) and field == "gifs" else "audio" if mime_type.startswith("audio/") else "image"
|
|
index = len(result)
|
|
result.append({"index": index, "kind": kind, "mimeType": mime_type, "filename": filename, "url": f"/jobs/{quote(job_id, safe='')}/outputs/{index}"})
|
|
return result
|
|
|
|
|
|
async def job_state(job_id: str) -> tuple[Provider, str, dict[str, Any], list[dict[str, Any]]]:
|
|
provider, prompt_id = decode_job(job_id)
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.get(f"{provider.base_url}/history/{quote(prompt_id, safe='')}", headers=provider_headers(provider))
|
|
response.raise_for_status()
|
|
history = response.json()
|
|
entry = history.get(prompt_id) or {}
|
|
outputs = extract_outputs(entry, job_id)
|
|
return provider, prompt_id, entry, outputs
|
|
|
|
|
|
@app.get("/v1/jobs/{job_id}", dependencies=[Depends(require_token)])
|
|
async def get_job(job_id: str) -> dict[str, Any]:
|
|
try:
|
|
provider, prompt_id, entry, outputs = await job_state(job_id)
|
|
except httpx.TimeoutException:
|
|
provider, prompt_id = decode_job(job_id)
|
|
return {
|
|
"id": job_id,
|
|
"promptId": prompt_id,
|
|
"providerId": provider.id,
|
|
"status": "running",
|
|
"outputs": [],
|
|
"error": "",
|
|
"pollDeferred": True,
|
|
}
|
|
status_payload = entry.get("status") or {}
|
|
if status_payload.get("status_str") == "error":
|
|
status = "failed"
|
|
elif outputs and not (job_id.startswith("h3v2.") and not status_payload.get("completed")):
|
|
status = "succeeded"
|
|
elif status_payload.get("completed"):
|
|
status = "failed"
|
|
else:
|
|
status = "running"
|
|
if job_id.startswith("h3v2.") and status != "succeeded":
|
|
outputs = []
|
|
if status != "succeeded" and cancellation_dispatched(job_id):
|
|
# An acknowledged interrupt is not terminal until this exact prompt
|
|
# disappears from the worker queue. Never infer it from HTTP 200 alone.
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
response = await client.get(f"{provider.base_url}/queue", headers=provider_headers(provider))
|
|
response.raise_for_status()
|
|
queue = response.json()
|
|
if not isinstance(queue, dict) or not isinstance(queue.get("queue_running"), list) or not isinstance(queue.get("queue_pending"), list):
|
|
raise ValueError("invalid queue response")
|
|
remaining = {str(item[1]) for item in queue["queue_running"] + queue["queue_pending"] if isinstance(item, list) and len(item) > 1}
|
|
if prompt_id not in remaining:
|
|
status, outputs = "cancelled", []
|
|
else:
|
|
return {"id": job_id, "promptId": prompt_id, "providerId": provider.id, "status": "running", "outputs": [], "error": "", "stage": "正在取消生成", "cancelRequested": True}
|
|
except (httpx.HTTPError, ValueError):
|
|
return {"id": job_id, "promptId": prompt_id, "providerId": provider.id, "status": "running", "outputs": [], "error": "", "stage": "正在确认取消", "cancelRequested": True}
|
|
error = "ComfyUI execution failed" if status == "failed" else ""
|
|
if status == "failed":
|
|
for event in status_payload.get("messages") or []:
|
|
if not isinstance(event, (list, tuple)) or len(event) < 2 or event[0] != "execution_error" or not isinstance(event[1], dict):
|
|
continue
|
|
# 只映射已识别的内核错误，不向用户返回节点输入、提示词或 traceback。
|
|
if "tensor strides exceed int32 range" in str(event[1].get("exception_message") or ""):
|
|
error = "当前时长与分辨率组合超出采样内核范围，请缩短时长后重试（最多 15 秒）"
|
|
break
|
|
if status == "failed":
|
|
hints = {
|
|
"H3_SUBTITLE_MASK_UNCERTAIN": "检测到疑似字幕，但无法安全区分字幕和产品文字，任务已停止，未返回未清理视频。",
|
|
"H3_SUBTITLE_RESIDUAL": "字幕清理后仍检测到残留，任务已停止，避免生成双层字幕。",
|
|
"H3_SPEECH_MISMATCH": "实际口播与原台词不一致，未生成误导字幕；请缩短台词或调整后重试。",
|
|
"H3_SUBTITLE_ALIGNMENT": "口播时间轴校验未通过，无法准确烧录字幕；请缩短单句台词后重试。",
|
|
"H3_SUBTITLE_FONT": "字幕含字体不支持的字符，已停止以避免乱码；请使用常见文字。",
|
|
"H3_SUBTITLE_DEPENDENCY": "字幕处理服务依赖未就绪，请联系管理员；原任务未自动重试。",
|
|
"H3_SUBTITLE_FRAME_COUNT": "字幕处理前后帧数或尺寸不一致，已停止输出，请联系管理员。",
|
|
}
|
|
for event in status_payload.get("messages") or []:
|
|
if not isinstance(event, (list, tuple)) or len(event) < 2 or event[0] != "execution_error" or not isinstance(event[1], dict):
|
|
continue
|
|
message = str(event[1].get("exception_message") or "")
|
|
for code, hint in hints.items():
|
|
if message.startswith(code + ":"):
|
|
error = hint
|
|
break
|
|
# Never expose a partial/raw output from a failed cleanup/quality gate.
|
|
outputs = []
|
|
if status == "succeeded":
|
|
quality = await audio_validator.poll(job_id, entry, provider.base_url, provider_headers(provider))
|
|
if quality["status"] in {"pending", "unavailable"}:
|
|
return {"id": job_id, "promptId": prompt_id, "providerId": provider.id, "status": "running",
|
|
"outputs": [], "error": "", "stage": "正在校验字幕与音轨" if job_id.startswith("h3v2.") or quality.get("visual_required") else "正在校验视频音轨", "pollDeferred": True}
|
|
if quality["status"] == "failed":
|
|
status, outputs, error = "failed", [], quality["error"]
|
|
return {"id": job_id, "promptId": prompt_id, "providerId": provider.id, "status": status, "outputs": outputs, "error": error}
|
|
|
|
|
|
@app.post("/v1/jobs/{job_id}/cancel", dependencies=[Depends(require_token)])
|
|
async def cancel_job(job_id: str) -> dict[str, Any]:
|
|
provider, prompt_id = decode_job(job_id)
|
|
# Write intent before sending; disk failure must not produce an untracked
|
|
# cancellation. The worker API atomically compares the running prompt ID.
|
|
record_cancellation(job_id, False)
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
response = await client.post(f"{provider.base_url}/api/jobs/{quote(prompt_id, safe='')}/cancel", json={}, headers=provider_headers(provider))
|
|
if response.status_code in (404, 405):
|
|
raise HTTPException(status_code=409, detail="生成节点未提供按任务取消接口，未中断其他任务")
|
|
response.raise_for_status()
|
|
acknowledged = response.json().get("cancelled")
|
|
if type(acknowledged) is not bool:
|
|
raise HTTPException(status_code=502, detail="生成节点取消回执无效，请核对原任务状态")
|
|
if acknowledged:
|
|
record_cancellation(job_id, True)
|
|
return await get_job(job_id)
|
|
|
|
|
|
@app.get("/v1/jobs/{job_id}/outputs/{index}", dependencies=[Depends(require_token)])
|
|
async def get_output(job_id: str, index: int) -> Response:
|
|
provider, _, entry, outputs = await job_state(job_id)
|
|
if (entry.get("status") or {}).get("status_str") == "error":
|
|
raise HTTPException(status_code=409, detail="任务未成功，未返回部分输出")
|
|
if not outputs or job_id.startswith("h3v2.") and not (entry.get("status") or {}).get("completed"):
|
|
raise HTTPException(status_code=409, detail="视频生成尚未完成，未返回部分输出")
|
|
quality = await audio_validator.poll(job_id, entry, provider.base_url, provider_headers(provider))
|
|
if quality["status"] != "verified":
|
|
raise HTTPException(status_code=422 if quality["status"] == "failed" else 409,
|
|
detail=quality.get("error", "视频核验尚未完成，请稍后查询原任务"))
|
|
if index < 0 or index >= len(outputs):
|
|
raise HTTPException(status_code=404, detail="output not found")
|
|
wanted = outputs[index]
|
|
locator: dict[str, Any] | None = None
|
|
for node in (entry.get("outputs") or {}).values():
|
|
if not isinstance(node, dict):
|
|
continue
|
|
for field in ("images", "gifs", "audio"):
|
|
for item in node.get(field) or []:
|
|
if isinstance(item, dict) and str(item.get("filename")) == wanted["filename"]:
|
|
locator = item
|
|
break
|
|
if locator: break
|
|
if locator: break
|
|
if locator is None:
|
|
raise HTTPException(status_code=404, detail="output locator not found")
|
|
if quality.get("visual_revision") == VISUAL_REVISION:
|
|
bound_locator = {key: locator.get(key, "output" if key == "type" else "") for key in ("filename", "subfolder", "type")}
|
|
try:
|
|
media = await audio_validator.verified_download(quality, bound_locator, provider.base_url, provider_headers(provider))
|
|
except AudioQualityError as error:
|
|
raise HTTPException(status_code=422, detail=str(error)) from error
|
|
except (httpx.HTTPError, OSError, asyncio.TimeoutError):
|
|
raise HTTPException(status_code=503, detail="H3_VISUAL_VALIDATION_UNAVAILABLE: 已核验视频暂不可读取，请查询原任务") from None
|
|
return Response(content=media, media_type=wanted["mimeType"], headers={"Cache-Control": "private, no-store"})
|
|
query = urlencode({"filename": locator.get("filename", ""), "subfolder": locator.get("subfolder", ""), "type": locator.get("type", "output")})
|
|
client = httpx.AsyncClient(timeout=120)
|
|
response = await client.send(client.build_request("GET", f"{provider.base_url}/view?{query}", headers=provider_headers(provider)), stream=True)
|
|
if response.status_code >= 400:
|
|
await response.aclose(); await client.aclose()
|
|
raise HTTPException(status_code=502, detail="failed to read ComfyUI output")
|
|
async def stream():
|
|
try:
|
|
async for chunk in response.aiter_bytes():
|
|
yield chunk
|
|
finally:
|
|
await response.aclose(); await client.aclose()
|
|
return StreamingResponse(stream(), media_type=response.headers.get("content-type", wanted["mimeType"]))
|