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" 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", "")}] 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) -> Provider | None: configured = providers() requested = configured.get(requested_provider_id) admitted = admission_ids(configured) if requested is None: return None 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) -> Provider | None: async with provider_selection_lock: provider = await select_provider(requested_provider_id) 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) 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"]))