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