Spaces:
Running on Zero
Running on Zero
| """高层 API 图片参数兼容层；顺序与能力校验复用 UI 的执行计划。""" | |
| import re | |
| from PIL import Image | |
| from core.execution_plan import ( | |
| MODE_MODEL_PK, MODE_MULTI_MODEL_GRID, MODE_MULTI_REFERENCE, MODE_SINGLE, | |
| available_run_modes, build_execution_plan, | |
| ) | |
| from core.reference_inputs import ImageInputError, REFERENCE_CHAIN_SPECS, SOURCE_IMAGE_KEYS, reference_choices | |
| from core.runtime_config import CONFIG | |
| NATIVE_CHAIN_ALIASES = { | |
| "reference_edit": "reference_latent", | |
| "mage_flow_reference_edit": "reference_image", | |
| "joyai_reference_edit": "joyai_image", | |
| "boogu_edit": "boogu_image_edit", | |
| } | |
| _IMAGE_KEY = re.compile(r"(?:图|img|image)([0-9０-９]+)", re.IGNORECASE) | |
| def normalize_image_request(params: dict) -> None: | |
| """只修改 handle_run 的副本。显式混用入口时拒绝，不猜优先级。""" | |
| numbered = {} | |
| for key in list(params): | |
| match = _IMAGE_KEY.fullmatch(key) if isinstance(key, str) else None | |
| if not match: | |
| continue | |
| index = int(match.group(1)) | |
| if index in numbered: | |
| raise ImageInputError(f"图片编号 {index} 重复；图{index} / img{index} / image{index} 只能提供一个字段。") | |
| numbered[index] = params.pop(key) | |
| if numbered: | |
| if "images" in params or params.get("image") is not None: | |
| raise ImageInputError("编号字段、images 列表和 image 源图不能混用，请只保留一个入口。") | |
| if sorted(numbered) != list(range(1, len(numbered) + 1)): | |
| raise ImageInputError("图片编号必须从 1 开始且连续，不能缺少 img1 或跳号。") | |
| params["images"] = [numbered[i] for i in range(1, len(numbered) + 1)] | |
| if "images" in params: | |
| images = params["images"] | |
| if not isinstance(images, list) or not images: | |
| raise ImageInputError("images 必须是非空、有序图片列表。") | |
| if any(not isinstance(image, (str, Image.Image)) or (isinstance(image, str) and not image.strip()) for image in images): | |
| raise ImageInputError("images 不允许空图片；删除空项后重新按顺序编号。") | |
| if params.get("image") is not None: | |
| raise ImageInputError("image 与 images 不能混用。") | |
| if params["task_type"] == "reference": | |
| params.setdefault("run_mode", MODE_MULTI_REFERENCE) | |
| if params.get("image") is not None: | |
| if not isinstance(params["image"], (str, Image.Image)): | |
| raise ImageInputError("image 必须是图片 URL 或 Base64 Data URI。") | |
| params["images"] = [params.pop("image")] | |
| elif params["task_type"] == "txt2img" and params.get("image") is not None: | |
| raise ImageInputError("txt2img 不使用 image 源图；参考图编辑请用 task_type=reference 和 images。") | |
| if params.get("image") is not None and not isinstance(params["image"], (str, Image.Image)): | |
| raise ImageInputError("image 必须是图片 URL 或 Base64 Data URI。") | |
| for field in ("run_mode", "reference_role"): | |
| if field in params and not isinstance(params[field], str): | |
| raise ImageInputError(f"{field} 必须是字符串。") | |
| if "extra_models" in params and ( | |
| not isinstance(params["extra_models"], list) | |
| or not all(isinstance(model, str) for model in params["extra_models"]) | |
| ): | |
| raise ImageInputError("extra_models 必须是模型名称列表。") | |
| if "use_model_defaults" in params and not isinstance(params["use_model_defaults"], bool): | |
| raise ImageInputError("use_model_defaults 必须是 boolean。") | |
| def validate_image_plan(params: dict) -> None: | |
| """不下载图片、不分配 GPU，提交队列前校验完整调用语义。""" | |
| values = { | |
| "task_type": params["task_type"], "model_display_name": params["model"], | |
| "positive_prompt": params["prompt"], "negative_prompt": params.get("negative_prompt") or "", | |
| "seed": params.get("seed", -1), "batch_size": params.get("batch_size", 1), | |
| } | |
| task = params["task_type"] | |
| if task in SOURCE_IMAGE_KEYS and params.get("image") is not None: | |
| values[SOURCE_IMAGE_KEYS[task]] = params["image"] | |
| for item in params.get("chain", []): | |
| chain = NATIVE_CHAIN_ALIASES.get(item["injector_type"], item["injector_type"]) | |
| if chain in REFERENCE_CHAIN_SPECS: | |
| if params.get("images"): | |
| raise ImageInputError("images 与原生参考 chain 不能混用；自动路由用 images，旧调用只用 chain。") | |
| if item.get("image") is None: | |
| raise ImageInputError(f"{chain} 缺少 image。") | |
| if not isinstance(item["image"], (str, Image.Image)) or (isinstance(item["image"], str) and not item["image"].strip()): | |
| raise ImageInputError(f"{chain}.image 必须是图片 URL 或 Base64 Data URI。") | |
| values.setdefault(REFERENCE_CHAIN_SPECS[chain][0], []).append(item["image"]) | |
| mode = params.get("run_mode", MODE_SINGLE) | |
| if mode not in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID} and params.get("extra_models"): | |
| raise ImageInputError("extra_models 只用于 model_pk / multi_model_grid。") | |
| if mode != MODE_MULTI_REFERENCE and params.get("reference_role", "auto") != "auto": | |
| raise ImageInputError("reference_role 只用于 reference / multi_reference。") | |
| if mode in {MODE_MODEL_PK, MODE_MULTI_MODEL_GRID} and ( | |
| params.get("chain") or params.get("vae_id") or params.get("pid") or params.get("pid_settings") | |
| ): | |
| raise ImageInputError("模型对比仅使用基础模型，不支持 chain / VAE / PiD 覆盖。") | |
| build_execution_plan( | |
| values, mode=mode, images=params.get("images"), | |
| extra_models=params.get("extra_models"), reference_role=params.get("reference_role", "auto"), | |
| use_model_defaults=params.get("use_model_defaults", True), | |
| ) | |
| def image_workflow_contract(model: str) -> dict: | |
| """在现有 discovery 上增量公开高层图片契约，不要求 AI 猜 injector。""" | |
| roles = reference_choices(model) | |
| return { | |
| "reference_supported": bool(roles), | |
| "reference_roles": { | |
| role: {"chain": spec["chain"], "max_images": spec["max_images"]} | |
| for role, spec in roles.items() | |
| }, | |
| "run_modes_by_task": { | |
| task: [mode for _, mode in available_run_modes(task, model)] | |
| for task in ["txt2img", *SOURCE_IMAGE_KEYS, "reference"] | |
| }, | |
| "parameters_schema": { | |
| "type": "object", | |
| "properties": { | |
| "images": {"type": "array", "minItems": 1, "maxItems": CONFIG.max_multi_images, | |
| "items": {"type": "string", "description": "Public HTTP(S) image URL or Base64 Data URI; not a local path."}}, | |
| "run_mode": {"type": "string", "enum": ["single", "model_pk", "multi_independent", "multi_model_grid", "multi_reference"], | |
| "description": "Default single; task_type=reference defaults to multi_reference. Use run_modes_by_task."}, | |
| "reference_role": {"type": "string", "enum": list(roles) or ["auto"], "default": "auto"}, | |
| "extra_models": {"type": "array", "items": {"type": "string"}, "description": "Only for model_pk / multi_model_grid; current model is included automatically."}, | |
| "use_model_defaults": {"type": "boolean", "default": True}, | |
| }, | |
| "patternProperties": {r"^(图|img|image)[1-9][0-9]*$": {"type": "string", "description": "Alternative to images; indices must start at 1, be contiguous and unique."}}, | |
| }, | |
| "rules": [ | |
| "Use task_type=reference with images, prompt, width, height for instruction editing / fusion; no native chain needed. Model must support reference.", | |
| "images order is 1-based: 图1 = img1 = image1. Numbered prompt references become image 1; quoted/backticked text is literal. No auto translation or prompt expansion.", | |
| "Do not mix images, numbered fields, image, or native reference chain. Legacy txt2img + chain and single-image image calls remain supported.", | |
| "multi_independent / multi_model_grid use one shared prompt per image. Say 当前图片 / the current image; cross-image numbered references are rejected.", | |
| "Leave steps/cfg/sampler/scheduler omitted to use model defaults. Negative prompts depend on model/CFG and may be unused at CFG=1.", | |
| "Prefer async_execution=true; poll the same task_id with get_task_status until completed/failed; do not resubmit on a polling timeout.", | |
| "Read result.image_references, effective_prompt, captions and jobs. Existing result.images remains a flat URL list; partial_success signals mixed results.", | |
| ], | |
| "limits": {"max_images": CONFIG.max_multi_images, "max_pk_models": CONFIG.max_pk_models, | |
| "max_jobs": CONFIG.max_plan_jobs, "max_outputs": CONFIG.max_plan_outputs}, | |
| } | |