828 lines
44 KiB
Python
828 lines
44 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, replace
|
|
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
|
|
huobao_panels: 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') not in ('gpu1', 'gpu1_8189') 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:
|
|
if provider.id == 'gpu1_8189':
|
|
# Same physical card: finish old accepted jobs without moving or cancelling them.
|
|
legacy = providers().get('gpu1')
|
|
if legacy:
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
|
old = await client.get(f'{legacy.base_url}/queue', headers=provider_headers(legacy))
|
|
old.raise_for_status()
|
|
queue = old.json()
|
|
if queue.get('queue_running') or queue.get('queue_pending'):
|
|
return None
|
|
stats = await client.get(f'{legacy.base_url}/system_stats', headers=provider_headers(legacy))
|
|
stats.raise_for_status()
|
|
devices = stats.json().get('devices')
|
|
if not isinstance(devices, list) or not devices or any(
|
|
not isinstance(d, dict) or type(d.get('torch_vram_total')) not in (int, float)
|
|
or not 0 <= d['torch_vram_total'] < float('inf') for d in devices
|
|
):
|
|
return None
|
|
# An idle cudaMallocAsync worker retains ~2.2 GiB of persistent tensors.
|
|
# Allow at most 4 GiB within the new worker's existing 8 GiB reserve;
|
|
# active/queued legacy jobs still block above, and large caches drain below.
|
|
if sum(d['torch_vram_total'] for d in devices) > 4 * 1024**3:
|
|
# Release only idle model caches; queue contents and results stay intact.
|
|
released = await client.post(f'{legacy.base_url}/free', json={'unload_models': True, 'free_memory': True}, headers=provider_headers(legacy))
|
|
released.raise_for_status()
|
|
return None
|
|
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, huobao_panels: bool = False) -> Provider | None:
|
|
configured = providers()
|
|
admitted = admission_ids(configured)
|
|
if huobao_panels:
|
|
ports = {'default': 8188, 'gpu1_8189': 8189, 'gpu3': 8191}
|
|
configured = {key: replace(value, base_url=f'http://192.168.110.111:{ports[key]}') for key, value in configured.items() if key in ports}
|
|
else:
|
|
# Other applications retain their original worker selection.
|
|
configured = {key: value for key, value in configured.items() if key != 'gpu1_8189'}
|
|
requested = configured.get(requested_provider_id)
|
|
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=("GPU1 / 8189 暂不可用：旧 GPU1 任务或显存尚未释放，未取消原任务，也未提交新任务" if requested_provider_id == "gpu1_8189" else "指定显卡不可用或缺少所需工作流，任务未提交"))
|
|
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, huobao_panels: bool = False) -> Provider | None:
|
|
async with provider_selection_lock:
|
|
provider = await select_provider(requested_provider_id, strict_provider, huobao_panels)
|
|
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, request.huobao_panels)
|
|
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"]))
|
|
|
|
@app.post("/v1/huobao/route", dependencies=[Depends(require_token)])
|
|
async def huobao_route(request: JobRequest) -> dict[str, Any]:
|
|
provider = await select_provider(request.provider_id, request.strict_provider, True)
|
|
if provider is None:
|
|
raise HTTPException(status_code=400, detail="unknown requested worker")
|
|
return {"providerId": provider.id, "baseUrl": provider.base_url, "ready": True}
|
|
|
|
|
|
|
|
@app.get("/v1/huobao/jobs/{job_id}", dependencies=[Depends(require_token)])
|
|
async def huobao_job_status(job_id: str) -> dict[str, Any]:
|
|
# Keep the general API contract unchanged; Huobao needs actual queue state.
|
|
result = await get_job(job_id)
|
|
if result.get("status") != "running":
|
|
return {**result, "worker_state": result.get("status")}
|
|
if result.get("pollDeferred"):
|
|
return {**result, "worker_state": "checking"}
|
|
provider, prompt_id = decode_job(job_id)
|
|
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()
|
|
running = [row[1] for row in queue.get("queue_running", [])]
|
|
pending = [row[1] for row in queue.get("queue_pending", [])]
|
|
if prompt_id in running:
|
|
return {**result, "worker_state": "running", "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))}
|
|
if prompt_id in pending:
|
|
return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1, "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))}
|
|
# A task may finish between the first history read and the queue read.
|
|
fresh = await get_job(job_id)
|
|
if fresh.get("status") != "running":
|
|
return {**fresh, "worker_state": fresh.get("status")}
|
|
if fresh.get("pollDeferred"):
|
|
return {**fresh, "worker_state": "checking"}
|
|
# Do not invent success/failure or submit again when a worker has lost history.
|
|
return {**fresh, "worker_state": "missing"}
|
|
|
|
# BEGIN HUOBAO READ-ONLY PROGRESS
|
|
"""Read-only ComfyUI telemetry, embedded in the existing private normalizers.
|
|
|
|
Only the requesting Huobao job's client connection is observed. No queue mutation
|
|
or image previews are used. Sampling ETA is never presented as whole-video ETA.
|
|
"""
|
|
import asyncio as _hp_asyncio
|
|
import json as _hp_json
|
|
import time as _hp_time
|
|
import math as _hp_math
|
|
import statistics as _hp_stats
|
|
from urllib.parse import quote as _hp_quote
|
|
import httpx as _hp_httpx
|
|
import websockets as _hp_ws
|
|
|
|
_hp_watchers = {}
|
|
_hp_histories = {}
|
|
|
|
|
|
def _hp_phase(kind):
|
|
kind = str(kind).lower()
|
|
if 'sampler' in kind: return 'sampling'
|
|
if 'decode' in kind: return 'decoding'
|
|
if 'savevideo' in kind or 'combine' in kind or 'createvideo' in kind: return 'encoding'
|
|
if 'loader' in kind or kind == 'loadimage': return 'loading'
|
|
if 'caption' in kind: return 'postprocessing'
|
|
if 'encode' in kind or 'tovideo' in kind: return 'conditioning'
|
|
return 'running'
|
|
|
|
|
|
def _hp_signature(graph):
|
|
# Keep graph topology, model weights, image count, dimensions, FPS, duration,
|
|
# steps and sampler. Exclude content, random seeds and output filenames.
|
|
omitted = {'text', 'prompt', 'positive', 'negative', 'seed', 'noise_seed',
|
|
'filename_prefix', 'image', 'audio', 'video', 'file', 'path'}
|
|
result = {}
|
|
for node, value in graph.items():
|
|
inputs = {}
|
|
for key, item in value.get('inputs', {}).items():
|
|
if key in omitted and not isinstance(item, list): continue
|
|
if any(word in key.lower() for word in ['prompt', 'text', 'caption']) and not isinstance(item, list): continue
|
|
inputs[key] = item
|
|
result[node] = {'class_type': value.get('class_type'), 'inputs': inputs}
|
|
return _hp_json.dumps(result, sort_keys=True, separators=(',', ':'))
|
|
|
|
|
|
def _hp_history_estimate(graph, history):
|
|
signature = _hp_signature(graph)
|
|
durations = []
|
|
for entry in history.values():
|
|
status = entry.get('status', {})
|
|
prompt = entry.get('prompt', [])
|
|
if status.get('status_str') != 'success' or not status.get('completed') or len(prompt) < 3: continue
|
|
previous = prompt[2]
|
|
if not isinstance(previous, dict) or _hp_signature(previous) != signature: continue
|
|
times = {}; cached = []
|
|
for event in status.get('messages', []):
|
|
if len(event) < 2 or not isinstance(event[1], dict): continue
|
|
times[event[0]] = event[1].get('timestamp')
|
|
if event[0] == 'execution_cached': cached = event[1].get('nodes', [])
|
|
if any('sampler' in str(previous.get(str(n), {}).get('class_type', '')).lower() for n in cached): continue
|
|
start, end = times.get('execution_start'), times.get('execution_success')
|
|
if not isinstance(start, (int, float)) or not isinstance(end, (int, float)): continue
|
|
duration = (end - start) / 1000
|
|
if 2 <= duration <= 86400: durations.append(duration)
|
|
result = {'historySamples': len(durations)}
|
|
if len(durations) >= 2:
|
|
# A range, not a guarantee; model loading and shared GPU contention vary.
|
|
result.update(historyMinSeconds=round(min(durations) * .8), historyMaxSeconds=round(max(durations) * 1.2))
|
|
return result
|
|
|
|
|
|
def _hp_event(state, message, timestamp):
|
|
data = message.get('data', {})
|
|
if not isinstance(data, dict): return
|
|
event = message.get('type')
|
|
if data.get('prompt_id') not in (None, state['prompt']): return
|
|
if event == 'executing':
|
|
node = data.get('node')
|
|
if node is None: return
|
|
node = str(node)
|
|
if node != state.get('node'):
|
|
state.update(node=node, samples=[], value=None, total=None)
|
|
state['phase'] = _hp_phase(state['classes'].get(node))
|
|
state['updated'] = timestamp
|
|
elif event == 'progress' and data.get('prompt_id') == state['prompt']:
|
|
value, total = data.get('value'), data.get('max')
|
|
if not all(type(x) in (int, float) and _hp_math.isfinite(x) for x in (value, total)): return
|
|
if not 0 <= value <= total <= 1000000 or total <= 0: return
|
|
node = str(data.get('node', state.get('node', '')))
|
|
if node != state.get('node') or total != state.get('total') or value < (state.get('value') or 0):
|
|
state.update(node=node, samples=[])
|
|
if state.get('value') != value or not state['samples']:
|
|
state['samples'] = (state['samples'] + [(timestamp, value)])[-8:]
|
|
state.update(value=value, total=total, updated=timestamp, phase=_hp_phase(state['classes'].get(node)))
|
|
|
|
|
|
def _hp_snapshot(state, timestamp):
|
|
result = {'phase': state.get('phase', 'running')}
|
|
if not state.get('connected') or timestamp - state.get('updated', 0) > 600:
|
|
return {'phase': 'running'}
|
|
if state.get('value') is None: return result
|
|
result.update(step=state['value'], totalSteps=state['total'], stepUpdatedAt=round(state['updated'] * 1000))
|
|
samples = state['samples']
|
|
if state.get('phase') == 'sampling' and len(samples) >= 3:
|
|
seconds = samples[-1][0] - samples[0][0]; steps = samples[-1][1] - samples[0][1]
|
|
if seconds >= 5 and steps > 0 and state['value'] < state['total']:
|
|
speed = seconds / steps
|
|
if timestamp - state['updated'] <= max(30, speed * 2.5):
|
|
remaining = max(1, speed * (state['total'] - state['value']) - (timestamp - state['updated']))
|
|
result.update(samplingMinSeconds=round(remaining * .8), samplingMaxSeconds=round(remaining * 1.3))
|
|
return result
|
|
|
|
|
|
async def _hp_watch(state, base, headers, client_id):
|
|
uri = base.replace('https://', 'wss://').replace('http://', 'ws://') + '/ws?clientId=' + _hp_quote(client_id, safe='')
|
|
while _hp_time.time() - state['requested'] < 120:
|
|
try:
|
|
async with _hp_ws.connect(uri, additional_headers=headers, max_size=32 * 1024 * 1024, open_timeout=5) as ws:
|
|
state['connected'] = True
|
|
# This observer needs text progress only, never preview images.
|
|
await ws.send(_hp_json.dumps({'type':'feature_flags','data':{'supports_preview_metadata':True}}))
|
|
while _hp_time.time() - state['requested'] < 120:
|
|
try: message = await _hp_asyncio.wait_for(ws.recv(), 15)
|
|
except _hp_asyncio.TimeoutError: continue
|
|
if isinstance(message, str):
|
|
try: _hp_event(state, _hp_json.loads(message), _hp_time.time())
|
|
except (ValueError, TypeError, AttributeError): pass
|
|
except _hp_asyncio.CancelledError: raise
|
|
except Exception:
|
|
await _hp_asyncio.sleep(5)
|
|
finally: state['connected'] = False
|
|
|
|
|
|
async def huobao_progress(provider, prompt_id, queue, headers):
|
|
"""Failure of progress reporting must never fail or resubmit a generation."""
|
|
timestamp = _hp_time.time()
|
|
running = queue.get('queue_running', []); pending = queue.get('queue_pending', [])
|
|
row = next((x for x in running + pending if len(x) > 3 and x[1] == prompt_id), None)
|
|
if row is None: return None
|
|
result = {'observedAt': round(timestamp * 1000), 'phase': 'queued' if row in pending else 'running'}
|
|
if row in pending:
|
|
result.update(queuePosition=pending.index(row) + 1, queueAhead=len(running) + pending.index(row))
|
|
try:
|
|
previous = _hp_watchers.get(provider.id)
|
|
if previous and (previous['prompt'] != prompt_id and row in running or previous['task'].done()):
|
|
previous['task'].cancel(); _hp_watchers.pop(provider.id, None); previous = None
|
|
if row in running:
|
|
client_id = row[3].get('client_id')
|
|
# Huobao's normalizer generates unique canvas-* client IDs. Never
|
|
# replace a human's browser websocket or guess another job's state.
|
|
if isinstance(client_id, str) and client_id.startswith('canvas-'):
|
|
if previous is None:
|
|
previous = {'prompt':prompt_id, 'classes':{str(k):v.get('class_type') for k,v in row[2].items()},
|
|
'requested':timestamp, 'samples':[], 'connected':False}
|
|
previous['task'] = _hp_asyncio.create_task(_hp_watch(previous, provider.base_url, headers, client_id))
|
|
_hp_watchers[provider.id] = previous
|
|
previous['requested'] = timestamp
|
|
if previous['prompt'] == prompt_id: result.update(_hp_snapshot(previous, timestamp))
|
|
cached = _hp_histories.get(provider.id)
|
|
if cached is None or timestamp - cached[0] > 90:
|
|
async with _hp_httpx.AsyncClient(timeout=5) as client:
|
|
response = await client.get(provider.base_url + '/history', params={'max_items':80}, headers=headers)
|
|
response.raise_for_status(); history = response.json()
|
|
_hp_histories[provider.id] = (timestamp, history)
|
|
else: history = cached[1]
|
|
result.update(_hp_history_estimate(row[2], history))
|
|
except Exception:
|
|
pass
|
|
return result
|