nodes.py

"""
Workflow1111 · the bound-function library
=========================================

Every function here becomes an `fn` operator node on the canvas. They are pure
local Python (Pillow + numpy) — no network, no token, no quota — which is what
makes the app's spine reliable: only the `model` and `space` nodes ever leave
the machine.

Two conventions matter, and both are load-bearing (see the module docstring in
`app.py` for why):

1. **Image inputs arrive in many shapes.** Depending on whether the value came
   from an upload, a `model` node, another `fn` node, or the browser executor,
   it can be a ``{"path"/"url"}`` dict, a ``data:`` URI, an ``http(s)`` URL, a
   ``/gradio_api/file=`` reference, or a plain path. `_load_image` normalizes
   all of them.
2. **Image outputs are ``{"path": <file>, "url": <data: URI>}``** — see `_emit`.
   The REST endpoint needs the real file, the canvas and any chained `model`
   node need the URI, so the value carries both.
3. **Structured data travels as JSON *text*, never on a ``json`` port.** The
   canvas stringifies a `json` port value with JavaScript's ``String(obj)``
   rather than ``JSON.stringify``, so the receiving node gets the literal text
   ``"[object Object]"`` and the data is gone. `detect_objects`,
   `classify_image`, `top_labels` and `png_info` therefore emit JSON strings,
   and `_as_list` parses them back.

Everything returned must be JSON-serializable — that is the contract for
`bind=` functions.
"""

import base64
import io
import json
import os
import random
import re
import tempfile
import urllib.parse
import urllib.request
from typing import Optional

import numpy as np
# Imported (not string-annotated) so `get_type_hints` can resolve it: gradio
# detects injected parameters by their resolved type, and an unresolvable
# annotation would turn the token into a visible input port.
from gradio.oauth import OAuthToken
from huggingface_hub import get_token as _saved_hf_token
from PIL import (
    Image,
    ImageDraw,
    ImageEnhance,
    ImageFilter,
    ImageFont,
    ImageOps,
    PngImagePlugin,
)

Image.MAX_IMAGE_PIXELS = 200_000_000

_RNG = random.SystemRandom()
_MAX_SEED = 2**31 - 1


# ─────────────────────────────────────────────────────────────────────────────
# Coercion helpers — the canvas can hand us strings where we want numbers
# ─────────────────────────────────────────────────────────────────────────────


def _text(value, default=""):
    if value is None:
        return default
    if isinstance(value, (list, tuple)):
        value = " ".join(str(v) for v in value)
    s = str(value).strip()
    return s if s else default


def _num(value, default, lo=None, hi=None, integer=False):
    """Best-effort numeric coercion with clamping. Blank/garbage → default."""
    if isinstance(value, bool):
        value = int(value)
    try:
        n = float(str(value).strip())
        if n != n or n in (float("inf"), float("-inf")):
            raise ValueError
    except (TypeError, ValueError, AttributeError):
        n = float(default)
    if lo is not None:
        n = max(lo, n)
    if hi is not None:
        n = min(hi, n)
    return int(round(n)) if integer else float(n)


def _flag(value, default=False):
    if value is None or value == "":
        return bool(default)
    if isinstance(value, bool):
        return value
    return str(value).strip().lower() in ("1", "true", "yes", "y", "on", "enable", "enabled")


def _choice(value, options, default):
    """Match a dropdown value against `options` leniently (case/space/punct)."""
    def norm(s):
        return re.sub(r"[^a-z0-9]", "", str(s).lower())

    v = norm(value)
    if not v:
        return default
    for o in options:
        if norm(o) == v:
            return o
    for o in options:
        if v and (norm(o).startswith(v) or v.startswith(norm(o))):
            return o
    return default


def _as_list(value):
    """Detection/label payloads arrive as a list, or as a JSON string of one."""
    if value is None or value == "":
        return []
    if isinstance(value, str):
        try:
            value = json.loads(value)
        except (ValueError, TypeError):
            return []
    if isinstance(value, dict):
        for key in ("detections", "labels", "results", "data", "predictions"):
            if isinstance(value.get(key), list):
                return value[key]
        return [value]
    return list(value) if isinstance(value, (list, tuple)) else []


# ─────────────────────────────────────────────────────────────────────────────
# Image IO
# ─────────────────────────────────────────────────────────────────────────────

_GRADIO_FILE_PREFIX = "/gradio_api/file="


def _load_image(value, label="image"):
    """Accept every shape an image port can carry and return a PIL Image."""
    if value is None or value == "":
        raise ValueError(f"No {label} supplied — connect or upload one.")
    if isinstance(value, Image.Image):
        return value

    src = value
    if isinstance(value, dict):
        src = (
            value.get("path")
            or value.get("url")
            or value.get("name")
            or value.get("src")
            or value.get("data")
            or ""
        )
    if isinstance(src, (bytes, bytearray)):
        return Image.open(io.BytesIO(bytes(src)))
    if isinstance(src, (list, tuple)) and src:
        # e.g. an ImageSlider-style [before, after] payload — take the last
        return _load_image(src[-1], label)
    if not isinstance(src, str) or not src.strip():
        raise ValueError(f"Could not read {label} (got {type(value).__name__}).")

    src = src.strip()

    if src.startswith("data:"):
        _, _, payload = src.partition(",")
        payload = payload.strip()
        if not payload:
            raise ValueError(f"Empty data URI for {label}.")
        pad = "=" * (-len(payload) % 4)
        return Image.open(io.BytesIO(base64.b64decode(payload + pad)))

    # "/gradio_api/file=C:\..." — or the same wrapped in an absolute URL
    if _GRADIO_FILE_PREFIX in src:
        src = src.split(_GRADIO_FILE_PREFIX, 1)[1]
        src = urllib.parse.unquote(src)

    if src.startswith(("http://", "https://")):
        req = urllib.request.Request(src, headers={"User-Agent": "workflow1111"})
        with urllib.request.urlopen(req, timeout=90) as resp:  # noqa: S310
            return Image.open(io.BytesIO(resp.read()))

    if src.startswith("file://"):
        src = urllib.request.url2pathname(urllib.parse.urlparse(src).path)

    if os.path.isfile(src):
        return Image.open(src)
    unquoted = urllib.parse.unquote(src)
    if os.path.isfile(unquoted):
        return Image.open(unquoted)

    raise ValueError(f"Could not resolve {label}: {src[:120]!r}")


def _rgb(img):
    """Flatten to RGB over white so JPEG-bound ops never crash on alpha."""
    if img.mode == "RGB":
        return img
    if img.mode in ("RGBA", "LA", "P"):
        img = img.convert("RGBA")
        flat = Image.new("RGB", img.size, (255, 255, 255))
        flat.paste(img, mask=img.split()[-1])
        return flat
    return img.convert("RGB")


def _has_alpha(img):
    return img.mode in ("RGBA", "LA") or (img.mode == "P" and "transparency" in img.info)


_JPEG_ABOVE_PX = 1_600_000  # ~1600×1000; keeps data URIs from ballooning


def _emit(img, info_text=None, quality=93):
    """PIL Image → the value an image output port should carry.

    Returns ``{"path": <temp file>, "url": <data: URI>}``, which is the only
    shape that satisfies all four consumers at once:

    * **REST API** — `_from_output` takes ``path`` first, and a gradio `Image`
      output component needs a real file. Handing it a bare ``data:`` URI makes
      gradio treat the URI as a filename and join it to the CWD
      (``OSError: [Errno 22] ...\\data:image\\png;base64,...``).
    * **Canvas** — the frontend prefers ``url``; a ``data:`` URI renders
      immediately with no round trip and no `allowed_paths` juggling.
    * **Chaining into a `model` node** — `_img_url` also prefers ``url``, and
      `InferenceClient` accepts a ``data:`` URI (a ``/gradio_api/file=`` path
      would be meaningless to a remote provider).
    * **Chaining into another `fn`** — `_load_image` prefers ``path``.

    `space` nodes are the one consumer this does *not* serve (`handle_file`
    cannot read a ``data:`` URI), which is why no `space` node is ever fed from
    an `fn` node — they take uploaded reference images only.
    """
    if img.mode not in ("RGB", "RGBA", "L"):
        img = img.convert("RGBA" if _has_alpha(img) else "RGB")

    buf = io.BytesIO()
    force_png = bool(info_text) or _has_alpha(img)
    if force_png or (img.width * img.height) <= _JPEG_ABOVE_PX:
        params = {}
        if info_text:
            meta = PngImagePlugin.PngInfo()
            # "parameters" is the key Automatic1111 itself writes, so these
            # images round-trip through the PNG Info pipeline (and through
            # real A1111 installs).
            meta.add_text("parameters", str(info_text))
            meta.add_text("Software", "Workflow1111")
            params["pnginfo"] = meta
        img.save(buf, format="PNG", optimize=True, **params)
        mime = "image/png"
    else:
        _rgb(img).save(buf, format="JPEG", quality=int(quality), optimize=True, subsampling=1)
        mime = "image/jpeg"

    raw = buf.getvalue()
    ext = "png" if mime.endswith("png") else "jpg"
    path = os.path.join(tempfile.gettempdir(), f"wf1111_{os.urandom(8).hex()}.{ext}")
    with open(path, "wb") as f:
        f.write(raw)
    return {
        "path": path,
        "url": f"data:{mime};base64," + base64.b64encode(raw).decode("ascii"),
    }


def _emit_uri(img, info_text=None, quality=93):
    """Just the ``data:`` URI — for places that need a bare string, such as the
    image reference handed to a chat-completion request."""
    return _emit(img, info_text, quality)["url"]


def _font(size):
    for name in ("arial.ttf", "segoeui.ttf", "DejaVuSans.ttf", "Helvetica.ttc"):
        try:
            return ImageFont.truetype(name, size)
        except OSError:
            continue
    try:
        return ImageFont.load_default(size=size)
    except TypeError:  # Pillow < 10
        return ImageFont.load_default()


# ─────────────────────────────────────────────────────────────────────────────
# 1 · Prompt engineering — Automatic1111's "Styles"
# ─────────────────────────────────────────────────────────────────────────────

# name → (positive suffix, extra negative terms)
STYLE_PRESETS = {
    "None": ("", ""),
    "Photorealistic": (
        "photorealistic, 35mm photograph, natural lighting, shallow depth of field, "
        "highly detailed, sharp focus",
        "illustration, painting, drawing, cartoon, anime, 3d render, cgi",
    ),
    "Cinematic": (
        "cinematic film still, dramatic rim lighting, anamorphic lens, film grain, "
        "moody color grade, wide shot",
        "flat lighting, snapshot, low contrast",
    ),
    "Anime": (
        "anime key visual, cel shading, vibrant colors, clean linework, "
        "studio-quality illustration",
        "photorealistic, photograph, 3d render, western comic",
    ),
    "Digital Art": (
        "digital painting, concept art, trending on artstation, dramatic lighting, "
        "highly detailed brushwork",
        "photograph, low effort sketch",
    ),
    "Oil Painting": (
        "oil on canvas, visible impasto brush strokes, rich pigment, classical composition, "
        "gallery lighting",
        "digital, vector, photograph, flat shading",
    ),
    "Watercolour": (
        "delicate watercolour painting, soft wet-on-wet washes, visible paper texture, "
        "loose expressive edges",
        "harsh outlines, digital, 3d render, photograph",
    ),
    "3D Render": (
        "octane render, physically based materials, global illumination, soft shadows, "
        "subsurface scattering, 8k",
        "flat illustration, 2d, sketch, painting",
    ),
    "Pixel Art": (
        "16-bit pixel art, limited palette, crisp dithering, isometric sprite",
        "smooth gradients, photorealistic, antialiasing, blur",
    ),
    "Line Art": (
        "clean black and white line art, bold inked contours, minimal hatching, white background",
        "color, shading, gradient, photograph, texture",
    ),
    "Studio Product": (
        "professional product photography, seamless white sweep, three-point softbox lighting, "
        "crisp reflections, commercial catalogue shot",
        "clutter, busy background, harsh shadows, low resolution",
    ),
    "Fantasy Concept": (
        "epic fantasy concept art, volumetric god rays, intricate ornamentation, "
        "matte painting, sweeping scale",
        "modern clothing, urban, mundane, photograph",
    ),
    "Neon Cyberpunk": (
        "cyberpunk megacity, neon signage, wet reflective asphalt, volumetric fog, "
        "teal and magenta lighting, blade runner mood",
        "daylight, rural, pastel, medieval",
    ),
    "Analog Film": (
        "portra 400 analog film photograph, halation, subtle grain, faded highlights, "
        "warm cast, candid framing",
        "digital clarity, oversharpened, hdr, 3d render",
    ),
}

QUALITY_TAGS = "masterpiece, best quality, highly detailed, intricate detail, sharp focus"

BASE_NEGATIVE = (
    "lowres, worst quality, low quality, jpeg artifacts, blurry, out of focus, "
    "watermark, signature, text, username, logo, cropped, bad anatomy, "
    "extra limbs, extra fingers, missing fingers, deformed hands, mutated, "
    "disfigured, poorly drawn face, long neck, duplicate, error"
)

SAFETY_NEGATIVE = "nsfw, nude, gore, blood, violence, disturbing imagery"


def apply_style(prompt, style, extra_tags, quality_boost):
    """Compose the final positive prompt: subject + style preset + your tags."""
    prompt = _text(prompt)
    if not prompt:
        raise ValueError("Prompt is empty — describe what you want to see.")
    style_name = _choice(style, list(STYLE_PRESETS), "None")
    parts = [prompt, STYLE_PRESETS[style_name][0], _text(extra_tags)]
    if _flag(quality_boost, True):
        parts.append(QUALITY_TAGS)

    seen, out = set(), []
    for chunk in parts:
        for tag in (t.strip() for t in chunk.split(",")):
            key = tag.lower()
            if tag and key not in seen:
                seen.add(key)
                out.append(tag)
    return ", ".join(out)


def build_negative(negative, style, use_base, safety_filter):
    """Compose the negative prompt from your text + the style's counter-tags."""
    style_name = _choice(style, list(STYLE_PRESETS), "None")
    parts = [_text(negative)]
    if _flag(use_base, True):
        parts.append(BASE_NEGATIVE)
    parts.append(STYLE_PRESETS[style_name][1])
    if _flag(safety_filter, True):
        parts.append(SAFETY_NEGATIVE)

    seen, out = set(), []
    for chunk in parts:
        for tag in (t.strip() for t in chunk.split(",")):
            key = tag.lower()
            if tag and key not in seen:
                seen.add(key)
                out.append(tag)
    return ", ".join(out)


# Automatic1111's aspect presets, as (width, height)
ASPECTS = {
    "Custom": None,
    "1:1 Square": (1024, 1024),
    "3:2 Landscape": (1216, 832),
    "2:3 Portrait": (832, 1216),
    "16:9 Widescreen": (1344, 768),
    "9:16 Vertical": (768, 1344),
    "4:3 Classic": (1152, 896),
    "3:4 Tall": (896, 1152),
}


def sampler_settings(steps, cfg_scale, seed, aspect, width, height):
    """Validate and normalize the sampler block.

    Returns (steps, cfg_scale, seed, width, height). Guardrails matter here:
    FLUX is served by fal-ai, which hard-rejects ``guidance_scale`` below 1.0
    with a 422, and dimensions must be multiples of 16. A seed of -1 rolls a
    fresh one and reports it, exactly like A1111.
    """
    steps = _num(steps, 4, lo=1, hi=50, integer=True)
    cfg = round(_num(cfg_scale, 1.0, lo=1.0, hi=20.0), 2)

    seed = _num(seed, -1, lo=-1, hi=_MAX_SEED, integer=True)
    if seed < 0:
        seed = _RNG.randint(0, _MAX_SEED)

    preset = ASPECTS.get(_choice(aspect, list(ASPECTS), "Custom"))
    if preset:
        width, height = preset
    width = _num(width, 1024, lo=256, hi=1536, integer=True)
    height = _num(height, 1024, lo=256, hi=1536, integer=True)
    width -= width % 16
    height -= height % 16

    return steps, cfg, seed, width, height


def generation_info(prompt, negative, steps, cfg_scale, seed, width, height, model_id):
    """The Automatic1111 'generation parameters' block, verbatim in its format.

    `postprocess` embeds this into the PNG's ``parameters`` text chunk, so the
    PNG Info pipeline (and a real A1111 install) can read it straight back.
    """
    prompt = _text(prompt, "(none)")
    negative = _text(negative)
    steps = _num(steps, 4, integer=True)
    cfg = round(_num(cfg_scale, 1.0), 2)
    seed = _num(seed, 0, integer=True)
    width = _num(width, 1024, integer=True)
    height = _num(height, 1024, integer=True)
    model = _text(model_id, "black-forest-labs/FLUX.1-schnell")

    lines = [prompt]
    if negative:
        lines.append(f"Negative prompt: {negative}")
    lines.append(
        f"Steps: {steps}, Sampler: Euler, CFG scale: {cfg}, Seed: {seed}, "
        f"Size: {width}x{height}, Model: {model}, Backend: HF Inference Providers, "
        f"Version: Workflow1111 (gr.Workflow)"
    )
    return "\n".join(lines)


def prompt_matrix(base_prompt, variations, shared_tags):
    """Split ``variations`` into up to four prompt variants — A1111's prompt matrix.

    Returns (prompt_1..prompt_4, labels). Each variant is fed to its own
    txt2img node so the four render in parallel, then `contact_sheet` tiles
    them into the familiar X/Y grid.
    """
    base = _text(base_prompt)
    if not base:
        raise ValueError("Prompt matrix needs a base prompt.")
    shared = _text(shared_tags)

    raw = [v.strip() for v in re.split(r"[|\n]+", _text(variations)) if v.strip()]
    if not raw:
        raw = ["", "", "", ""]
    raw = (raw + raw * 4)[:4] if len(raw) < 4 else raw[:4]

    prompts, labels = [], []
    for variant in raw:
        parts = [base, variant, shared]
        prompts.append(", ".join(p for p in parts if p))
        labels.append(variant or "base")
    return prompts[0], prompts[1], prompts[2], prompts[3], " | ".join(labels)


def magic_instruction(idea, target_style, verbosity):
    """Wrap a rough idea into an instruction for the prompt-writing LLM."""
    idea = _text(idea)
    if not idea:
        raise ValueError("Give the prompt builder an idea to work from.")
    style = _choice(target_style, list(STYLE_PRESETS), "Cinematic")
    length = _choice(verbosity, ["Concise", "Detailed", "Elaborate"], "Detailed")
    budget = {"Concise": "18", "Detailed": "35", "Elaborate": "60"}[length]
    return (
        "You are a Stable Diffusion prompt engineer. Rewrite the idea below as a single "
        f"image-generation prompt in the '{style}' style.\n"
        f"Rules: comma-separated visual tags only, at most {budget} tags, no sentences, "
        "no preamble, no explanation, no quotes, no markdown, do not mention the rules. "
        "Cover subject, composition, lighting, colour palette, medium and mood.\n"
        f"Idea: {idea}"
    )


_PREAMBLE = re.compile(
    r"^\s*(sure|certainly|here(?:'s| is)|okay|ok|absolutely|of course|prompt)\b[^\n:]*:?\s*",
    re.I,
)

_TAG_EDGES = re.compile(r"""^[\s"'`*\-–—]+|[\s"'`*.;:]+$""")


def clean_prompt(raw, max_tags):
    """Strip an LLM's chattiness down to a clean comma-separated prompt."""
    text = _text(raw)
    if not text:
        raise ValueError("The language model returned nothing to clean up.")

    text = re.sub(r"```[a-zA-Z]*\n?", "", text).replace("```", "")
    text = re.sub(r"^\s*#+\s*.*$", "", text, flags=re.M)          # md headings
    text = re.sub(r"<think>.*?</think>", "", text, flags=re.S | re.I)
    text = re.sub(r"\*\*(.+?)\*\*", r"\1", text)
    text = _PREAMBLE.sub("", text.strip())
    text = text.strip().strip('"').strip("'")
    # Prefer the densest line — LLMs often add a trailing note after the prompt
    lines = [ln.strip(" -*\t") for ln in text.splitlines() if ln.strip()]
    if lines:
        text = max(lines, key=lambda ln: ln.count(","))

    cap = _num(max_tags, 40, lo=1, hi=120, integer=True)
    seen, tags = set(), []
    # Strip quoting/bullet punctuation per tag — a wrapping quote survives the
    # line-level strip when the closing quote is not the final character.
    for tag in (_TAG_EDGES.sub("", t) for t in text.split(",")):
        key = tag.lower()
        if tag and key not in seen:
            seen.add(key)
            tags.append(tag)
        if len(tags) >= cap:
            break
    if not tags:
        raise ValueError("Could not extract a usable prompt from the model output.")
    return ", ".join(tags)


def txt2img(prompt, negative_prompt, steps, cfg_scale, seed, width, height,
            model_id, oauth_token: Optional[OAuthToken] = None):
    """Text-to-image against HF Inference Providers — the A1111 txt2img box.

    This is deliberately an `fn` node rather than a `model` node. The canvas
    rewrites a `model` node's input ports to match the endpoint's canonical
    schema, and `text_to_image`'s schema is just ``prompt`` — so negative
    prompt, steps, CFG, seed and size were being silently dropped the moment
    the graph was opened in a browser. Calling `InferenceClient` here keeps the
    whole control surface, and `fn` ports are left alone.

    Returns a ``data:`` URI, which chains into both model and fn nodes.
    """
    from huggingface_hub import InferenceClient

    prompt = _text(prompt)
    if not prompt:
        raise ValueError("Prompt is empty — describe what you want to see.")

    token = _hf_token(oauth_token)
    model = _text(model_id, "black-forest-labs/FLUX.1-schnell")
    kwargs = {
        "prompt": prompt,
        "num_inference_steps": _num(steps, 4, lo=1, hi=50, integer=True),
        # fal-ai rejects guidance_scale < 1 with a 422, so never send one.
        "guidance_scale": round(_num(cfg_scale, 1.0, lo=1.0, hi=20.0), 2),
        "width": _num(width, 1024, lo=256, hi=1536, integer=True),
        "height": _num(height, 1024, lo=256, hi=1536, integer=True),
    }
    negative = _text(negative_prompt)
    if negative:
        kwargs["negative_prompt"] = negative
    seed = _num(seed, -1, lo=-1, hi=_MAX_SEED, integer=True)
    if seed >= 0:
        kwargs["seed"] = seed

    try:
        image = InferenceClient(model=model, token=token,
                                provider="auto").text_to_image(**kwargs)
    except Exception as e:
        detail = str(e)
        if "402" in detail or "quota" in detail.lower() or "credits" in detail.lower():
            raise ValueError(
                f"Inference credits exhausted for {model}. Wait for the quota to "
                "reset, or point this node at a different model."
            ) from e
        if "401" in detail or "403" in detail:
            raise ValueError(f"Not authorized for {model} — check your token.") from e
        if "not supported" in detail.lower():
            raise ValueError(
                f"No enabled provider serves {model} for text-to-image. Try "
                "black-forest-labs/FLUX.1-schnell or FLUX.1-dev."
            ) from e
        raise ValueError(f"{model} failed: {detail[:300]}") from e

    return _emit(image)


def _hf_token(oauth_token):
    token = (getattr(oauth_token, "token", None)
             or os.environ.get("HF_TOKEN") or _saved_hf_token())
    if not token:
        raise ValueError(
            "No Hugging Face token. Run `hf auth login`, set HF_TOKEN, or sign "
            "in with the button at the top of the canvas."
        )
    return token


def _chat(model_id, token, text, image=None, max_tokens=512):
    """One chat-completion call, streamed.

    Streamed rather than buffered because the router's gateway drops a
    non-streaming request at ~120s; keeping bytes moving bounds the call by the
    model instead of by an idle proxy.
    """
    from huggingface_hub import InferenceClient

    content = []
    if text:
        content.append({"type": "text", "text": text})
    if image is not None:
        content.append({"type": "image_url", "image_url": {"url": image}})
    if not content:
        raise ValueError("Nothing to send — connect a prompt or an image.")

    client = InferenceClient(model=model_id, token=token, provider="auto")
    parts, finish = [], None
    try:
        for chunk in client.chat_completion(
            [{"role": "user", "content": content}],
            max_tokens=int(max_tokens), stream=True,
        ):
            if not chunk.choices:
                continue
            choice = chunk.choices[0]
            if getattr(choice, "delta", None) and choice.delta.content:
                parts.append(choice.delta.content)
            if choice.finish_reason:
                finish = choice.finish_reason
    except Exception as e:
        detail = str(e)
        if "not supported" in detail.lower():
            raise ValueError(
                f"No enabled provider serves {model_id}. Try "
                "Qwen/Qwen3-4B-Instruct-2507 or google/gemma-3-27b-it."
            ) from e
        if "402" in detail or "quota" in detail.lower():
            raise ValueError(f"Inference credits exhausted for {model_id}.") from e
        raise ValueError(f"{model_id} failed: {detail[:300]}") from e

    out = "".join(parts).strip()
    if not out:
        raise ValueError(f"{model_id} returned no text (finish_reason={finish}).")
    return out


def chat_llm(prompt, model_id, max_tokens, oauth_token: Optional[OAuthToken] = None):
    """Text-only LLM call.

    An `fn` node rather than a `model` node for the same reason as `txt2img`:
    the canvas normalizes a `chat_completion` node's ports to the schema's
    (image, text) pair, and the wired prompt was not reaching the model.
    """
    prompt = _text(prompt)
    if not prompt:
        raise ValueError("Nothing to send to the language model.")
    return _chat(_text(model_id, "Qwen/Qwen3-4B-Instruct-2507"),
                 _hf_token(oauth_token), prompt,
                 max_tokens=_num(max_tokens, 512, lo=32, hi=4096, integer=True))


def interrogate(image, instruction, model_id, max_tokens,
                oauth_token: Optional[OAuthToken] = None):
    """Vision-language call — CLIP-interrogate, essentially.

    The image is **required**: without this guard the model cheerfully invents
    a description of an image it was never given, which looks like a working
    result and is entirely fabricated.
    """
    img = _load_image(image, "image to interrogate")  # raises if absent
    return _chat(_text(model_id, "Qwen/Qwen2.5-VL-72B-Instruct"),
                 _hf_token(oauth_token),
                 _text(instruction, "Describe this image."),
                 image=_emit_uri(img),
                 max_tokens=_num(max_tokens, 512, lo=32, hi=4096, integer=True))


def _image_file(image, label="image"):
    """Materialize any accepted image value as a temp file path.

    A **path**, not bytes: handing `InferenceClient` raw bytes makes the router
    reject the call with "No content type provided and no default one
    configured", whereas from a path huggingface_hub infers the MIME type.
    """
    img = _load_image(image, label)
    path = os.path.join(tempfile.gettempdir(), f"wf1111_in_{os.urandom(8).hex()}.jpg")
    _rgb(img).save(path, format="JPEG", quality=94, optimize=True)
    return path


def detect_objects(image, model_id, min_score, oauth_token: Optional[OAuthToken] = None):
    """Object detection, returning the detections as a **JSON string**.

    An `fn` node calling `InferenceClient` rather than a `model` node, because
    the canvas destroys `json`-typed port values: it stringifies them with
    JavaScript's `String(obj)` instead of `JSON.stringify`, so the downstream
    node receives the literal text ``"[object Object]"`` and sees zero
    detections. Text ports survive intact, so the detections travel as JSON
    text and `_as_list` parses them back.
    """
    from huggingface_hub import InferenceClient

    model = _text(model_id, "facebook/detr-resnet-50")
    floor = _num(min_score, 0.0, lo=0.0, hi=1.0)
    client = InferenceClient(model=model, token=_hf_token(oauth_token), provider="auto")
    try:
        results = client.object_detection(image=_image_file(image, "image to analyse"))
    except Exception as e:
        raise ValueError(f"{model} failed: {str(e)[:300]}") from e

    found = []
    for r in results:
        box = getattr(r, "box", None) or {}
        get = (lambda k: getattr(box, k, None)) if not isinstance(box, dict) else box.get
        try:
            coords = {k: int(get(k)) for k in ("xmin", "ymin", "xmax", "ymax")}
        except (TypeError, ValueError):
            continue
        score = float(getattr(r, "score", 0.0) or 0.0)
        if score < floor:
            continue
        found.append({"label": str(getattr(r, "label", "object")),
                      "score": round(score, 4), "box": coords})
    found.sort(key=lambda d: d["score"], reverse=True)
    return json.dumps(found)


def classify_image(image, model_id, oauth_token: Optional[OAuthToken] = None):
    """Image classification, returning the labels as a **JSON string**
    (same reason as `detect_objects`)."""
    from huggingface_hub import InferenceClient

    model = _text(model_id, "google/vit-base-patch16-224")
    client = InferenceClient(model=model, token=_hf_token(oauth_token), provider="auto")
    try:
        results = client.image_classification(
            image=_image_file(image, "image to classify"))
    except Exception as e:
        raise ValueError(f"{model} failed: {str(e)[:300]}") from e

    return json.dumps([{"label": str(getattr(r, "label", "?")),
                        "score": round(float(getattr(r, "score", 0.0) or 0.0), 5)}
                       for r in results])


def top_labels(labels, top_k, min_score):
    """Format an image-classification payload. Returns (text, json)."""
    items = _as_list(labels)
    k = _num(top_k, 5, lo=1, hi=25, integer=True)
    floor = _num(min_score, 0.0, lo=0.0, hi=1.0)

    rows = []
    for item in items:
        if not isinstance(item, dict):
            continue
        score = _num(item.get("score"), 0.0)
        if score < floor:
            continue
        rows.append({"label": _text(item.get("label"), "?"), "score": round(score, 4)})
    rows.sort(key=lambda r: r["score"], reverse=True)
    rows = rows[:k]

    if not rows:
        return "No labels above the score threshold.", "[]"
    width = max(len(r["label"]) for r in rows)
    lines = [
        f"{r['label']:<{width}}  {r['score'] * 100:5.1f}%  {'█' * max(1, int(r['score'] * 24))}"
        for r in rows
    ]
    # JSON *text*, not a list: a `json` port would reach the canvas as
    # "[object Object]" (see `detect_objects`).
    return "\n".join(lines), json.dumps(rows, indent=2)


# ─────────────────────────────────────────────────────────────────────────────
# 2 · Image operators
# ─────────────────────────────────────────────────────────────────────────────

_RESAMPLE = {
    "Lanczos": Image.LANCZOS,
    "Bicubic": Image.BICUBIC,
    "Bilinear": Image.BILINEAR,
    "Nearest (pixel art)": Image.NEAREST,
}


def _resize(img, factor, method):
    factor = _num(factor, 1.0, lo=1.0, hi=4.0)
    if factor <= 1.001:
        return img
    resample = _RESAMPLE[_choice(method, list(_RESAMPLE), "Lanczos")]
    w = min(int(img.width * factor), 4096)
    h = min(int(img.height * factor), 4096)
    return img.resize((w, h), resample)


def _vignette(img, strength):
    strength = _num(strength, 0.0, lo=0.0, hi=1.0)
    if strength <= 0.001:
        return img
    w, h = img.size
    yy, xx = np.mgrid[0:h, 0:w]
    cx, cy = (w - 1) / 2.0, (h - 1) / 2.0
    dist = np.sqrt(((xx - cx) / cx) ** 2 + ((yy - cy) / cy) ** 2) / np.sqrt(2.0)
    mask = np.clip(1.0 - strength * np.clip(dist, 0, 1) ** 2.2, 0.0, 1.0)

    arr = np.asarray(_rgb(img)).astype(np.float32)
    arr *= mask[..., None]
    return Image.fromarray(np.clip(arr, 0, 255).astype(np.uint8), "RGB")


def _grain(img, amount, seed=None):
    amount = _num(amount, 0.0, lo=0.0, hi=1.0)
    if amount <= 0.001:
        return img
    rng = np.random.default_rng(None if seed is None else int(seed) % (2**32))
    arr = np.asarray(_rgb(img)).astype(np.float32)
    noise = rng.normal(0.0, amount * 26.0, arr.shape[:2])[..., None]
    return Image.fromarray(np.clip(arr + noise, 0, 255).astype(np.uint8), "RGB")


def _watermark(img, text, opacity=0.72):
    text = _text(text)
    if not text:
        return img
    base = img.convert("RGBA")
    layer = Image.new("RGBA", base.size, (0, 0, 0, 0))
    draw = ImageDraw.Draw(layer)
    size = max(13, int(min(base.size) * 0.032))
    font = _font(size)
    pad = max(8, size // 2)

    box = draw.textbbox((0, 0), text, font=font)
    tw, th = box[2] - box[0], box[3] - box[1]
    x, y = base.width - tw - pad, base.height - th - pad - box[1]

    draw.rectangle(
        [x - pad // 2, y + box[1] - pad // 3, x + tw + pad // 2, y + box[1] + th + pad // 3],
        fill=(0, 0, 0, int(90 * opacity)),
    )
    draw.text((x, y), text, font=font, fill=(255, 255, 255, int(255 * opacity)))
    return Image.alpha_composite(base, layer).convert("RGB")


def _border(img, width_pct, color="#0f0f12"):
    width_pct = _num(width_pct, 0.0, lo=0.0, hi=0.2)
    if width_pct <= 0.0005:
        return img
    pad = max(1, int(min(img.size) * width_pct))
    return ImageOps.expand(_rgb(img), border=pad, fill=color)


def postprocess(
    image,
    upscale,
    upscale_method,
    sharpen,
    saturation,
    contrast,
    brightness,
    vignette,
    grain,
    border,
    watermark,
    embed_info,
):
    """The txt2img finishing chain — A1111's post-processing, one node.

    Runs in a deliberate order (resize → tone → sharpen → optical → framing) so
    grain and vignette are not resampled and the watermark stays crisp. When
    ``embed_info`` carries a generation-parameters block it is written into the
    PNG's ``parameters`` chunk.
    """
    img = _load_image(image)

    img = _resize(img, upscale, upscale_method)

    sat = _num(saturation, 1.0, lo=0.0, hi=2.5)
    con = _num(contrast, 1.0, lo=0.2, hi=2.5)
    bri = _num(brightness, 1.0, lo=0.2, hi=2.5)
    if abs(sat - 1.0) > 0.01:
        img = ImageEnhance.Color(img).enhance(sat)
    if abs(con - 1.0) > 0.01:
        img = ImageEnhance.Contrast(img).enhance(con)
    if abs(bri - 1.0) > 0.01:
        img = ImageEnhance.Brightness(img).enhance(bri)

    sharp = _num(sharpen, 0.0, lo=0.0, hi=2.0)
    if sharp > 0.01:
        img = img.filter(
            ImageFilter.UnsharpMask(radius=1.8, percent=int(80 * sharp), threshold=3)
        )

    img = _vignette(img, vignette)
    img = _grain(img, grain)
    img = _border(img, border)
    img = _watermark(img, watermark)

    return _emit(img, info_text=_text(embed_info) or None)


def prep_image(image, max_side, mode, strip_alpha):
    """Normalize an image for the next stage — and bridge model → model.

    A `model` node's image output cannot be wired straight into another
    model's ``image_to_image`` port (gradio hands the provider a
    ``/gradio_api/file=`` path it cannot resolve). Routing it through this node
    re-emits the pixels as a ``data:`` URI, which does chain. It also caps the
    long edge, which keeps img2img latency sane.
    """
    img = _load_image(image)
    mode_name = _choice(mode, ["Fit", "Cover (crop)", "Stretch", "Pad to square"], "Fit")
    side = _num(max_side, 1024, lo=256, hi=2048, integer=True)

    if _flag(strip_alpha, True):
        img = _rgb(img)

    if mode_name == "Pad to square":
        img = ImageOps.contain(img, (side, side), Image.LANCZOS)
        canvas = Image.new(img.mode, (side, side), (255, 255, 255) if img.mode == "RGB" else 0)
        canvas.paste(img, ((side - img.width) // 2, (side - img.height) // 2))
        img = canvas
    elif mode_name == "Cover (crop)":
        img = ImageOps.fit(img, (side, side), Image.LANCZOS, centering=(0.5, 0.5))
    elif mode_name == "Stretch":
        img = img.resize((side, side), Image.LANCZOS)
    else:  # Fit — preserve aspect, only ever downscale
        if max(img.size) > side:
            img = ImageOps.contain(img, (side, side), Image.LANCZOS)

    return _emit(img)


def extras_upscale(image, factor, method, sharpen, denoise, restore_contrast):
    """A1111's 'Extras' upscaler, done locally. Returns (image, report).

    Honest about what it is: high-quality Lanczos resampling with an unsharp
    pass — not a GAN. It is instant, deterministic and never hits a quota,
    which makes it the right default; the AuraSR Space node next to it on the
    canvas is there when you want real hallucinated detail.
    """
    img = _load_image(image)
    before = img.size

    if _flag(denoise, False):
        img = img.filter(ImageFilter.MedianFilter(size=3))

    img = _resize(img, factor, method)

    sharp = _num(sharpen, 0.45, lo=0.0, hi=2.0)
    if sharp > 0.01:
        img = img.filter(
            ImageFilter.UnsharpMask(radius=2.2, percent=int(95 * sharp), threshold=2)
        )
    if _flag(restore_contrast, True):
        img = ImageEnhance.Contrast(img).enhance(1.04)

    mp = (img.width * img.height) / 1e6
    report = (
        f"Upscaled {before[0]}×{before[1]} → {img.width}×{img.height}  ({mp:.2f} MP)\n"
        f"Resampler: {_choice(method, list(_RESAMPLE), 'Lanczos')}   "
        f"Unsharp: {sharp:.2f}   Denoise: {'on' if _flag(denoise, False) else 'off'}"
    )
    return _emit(img), report


# --- ControlNet-style preprocessors -----------------------------------------


def _sobel(gray):
    """Gradient magnitude + direction via Sobel."""
    a = np.asarray(gray, dtype=np.float32)
    kx = np.array([[-1, 0, 1], [-2, 0, 2], [-1, 0, 1]], dtype=np.float32)
    ky = np.array([[-1, -2, -1], [0, 0, 0], [1, 2, 1]], dtype=np.float32)

    p = np.pad(a, 1, mode="edge")
    win = np.lib.stride_tricks.sliding_window_view(p, (3, 3))
    gx = np.einsum("ijkl,kl->ij", win, kx)
    gy = np.einsum("ijkl,kl->ij", win, ky)
    return np.hypot(gx, gy), np.arctan2(gy, gx)


def _canny(gray, low, high):
    """Canny: Sobel → non-maximum suppression → hysteresis."""
    mag, theta = _sobel(gray)
    # Normalize against the 99th percentile, not the maximum: a single specular
    # highlight can be 3× the typical strong edge, and dividing by it crushes
    # every real contour below the threshold (a photo yielded 0.7% edge pixels
    # instead of ~5%). Clipping keeps that one outlier from mattering.
    scale = float(np.percentile(mag, 99)) or float(mag.max())
    if scale > 0:
        mag = np.clip(mag / scale * 255.0, 0, 255)

    # non-maximum suppression, directions quantized to 4 bins
    angle = (np.degrees(theta) % 180.0)
    nms = np.zeros_like(mag)
    m = mag
    pad = np.pad(m, 1, mode="constant")
    neighbours = {
        0: (pad[1:-1, 2:], pad[1:-1, :-2]),      # 0°   → E / W
        1: (pad[:-2, 2:], pad[2:, :-2]),         # 45°  → NE / SW
        2: (pad[:-2, 1:-1], pad[2:, 1:-1]),      # 90°  → N / S
        3: (pad[:-2, :-2], pad[2:, 2:]),         # 135° → NW / SE
    }
    bins = np.digitize(angle, [22.5, 67.5, 112.5, 157.5]) % 4
    for b, (n1, n2) in neighbours.items():
        sel = bins == b
        nms[sel] = np.where((m[sel] >= n1[sel]) & (m[sel] >= n2[sel]), m[sel], 0)

    strong, weak = nms >= high, (nms >= low) & (nms < high)

    # hysteresis: iteratively promote weak pixels touching strong ones
    keep = strong.copy()
    for _ in range(12):
        grown = keep.copy()
        for dy in (-1, 0, 1):
            for dx in (-1, 0, 1):
                if dx or dy:
                    grown |= np.roll(np.roll(keep, dy, 0), dx, 1)
        promoted = grown & weak & ~keep
        if not promoted.any():
            break
        keep |= promoted
    return (keep * 255).astype(np.uint8)


CONTROL_MODES = [
    "Canny edges",
    "Line art",
    "Soft sketch",
    "Luma depth (approx)",
    "Posterize",
    "Threshold",
    "Grayscale",
]


def controlnet_preprocess(image, mode, low_threshold, high_threshold, invert, blur):
    """Annotator previews, computed locally.

    Real depth estimation is unavailable — `InferenceClient` has no
    ``depth_estimation`` method and the legacy ``api-inference`` host no longer
    resolves — so 'Luma depth' is an honest luminance-based approximation, not
    a monocular depth model. The edge modes are the genuine article.
    """
    img = _load_image(image)
    mode_name = _choice(mode, CONTROL_MODES, "Canny edges")

    blur_r = _num(blur, 0.0, lo=0.0, hi=6.0)
    work = img.filter(ImageFilter.GaussianBlur(blur_r)) if blur_r > 0.05 else img
    gray = _rgb(work).convert("L")

    low = _num(low_threshold, 60, lo=1, hi=254, integer=True)
    high = _num(high_threshold, 160, lo=2, hi=255, integer=True)
    if high <= low:
        high = min(255, low + 20)

    if mode_name == "Canny edges":
        out = Image.fromarray(_canny(gray.filter(ImageFilter.GaussianBlur(1.1)), low, high), "L")
    elif mode_name == "Line art":
        edges = gray.filter(ImageFilter.FIND_EDGES)
        edges = ImageOps.autocontrast(edges).filter(ImageFilter.MaxFilter(3))
        out = ImageOps.invert(edges)
    elif mode_name == "Soft sketch":
        inv = ImageOps.invert(gray).filter(ImageFilter.GaussianBlur(max(1.5, blur_r or 3.0)))
        a = np.asarray(gray, np.float32)
        b = np.asarray(inv, np.float32)
        dodge = np.clip(a * 255.0 / np.maximum(255.0 - b, 1.0), 0, 255)
        out = Image.fromarray(dodge.astype(np.uint8), "L")
    elif mode_name == "Luma depth (approx)":
        eq = ImageOps.autocontrast(gray.filter(ImageFilter.GaussianBlur(2.0)))
        out = eq.point(lambda v: int(255 * (v / 255.0) ** 0.72))
    elif mode_name == "Posterize":
        out = ImageOps.posterize(gray, 3)
    elif mode_name == "Threshold":
        out = gray.point(lambda v: 255 if v >= low else 0, mode="L")
    else:  # Grayscale
        out = ImageOps.autocontrast(gray)

    if _flag(invert, False):
        out = ImageOps.invert(out)
    return _emit(out.convert("RGB"))


# --- detection, masking, grids ----------------------------------------------

_PALETTE = [
    (255, 92, 92), (86, 204, 242), (255, 199, 84), (129, 236, 160),
    (200, 143, 255), (255, 145, 200), (120, 180, 255), (255, 170, 110),
]


def _boxes(detections, min_score):
    floor = _num(min_score, 0.5, lo=0.0, hi=1.0)
    out = []
    for det in _as_list(detections):
        if not isinstance(det, dict):
            continue
        box = det.get("box") or det.get("bbox") or {}
        if not isinstance(box, dict):
            continue
        try:
            x0, y0 = float(box["xmin"]), float(box["ymin"])
            x1, y1 = float(box["xmax"]), float(box["ymax"])
        except (KeyError, TypeError, ValueError):
            continue
        score = _num(det.get("score"), 0.0)
        if score < floor:
            continue
        out.append(
            {
                "label": _text(det.get("label"), "object"),
                "score": score,
                "box": (min(x0, x1), min(y0, y1), max(x0, x1), max(y0, y1)),
            }
        )
    out.sort(key=lambda d: d["score"], reverse=True)
    return out


def draw_detections(image, detections, min_score, show_labels):
    """Overlay DETR boxes. Returns (annotated image, summary text)."""
    img = _rgb(_load_image(image)).copy()
    found = _boxes(detections, min_score)

    draw = ImageDraw.Draw(img, "RGBA")
    # Sized generously on purpose: a canvas node renders a 768px image at
    # roughly a third of its size, where a hairline box is invisible.
    stroke = max(3, int(min(img.size) * 0.008))
    font = _font(max(15, int(min(img.size) * 0.034)))

    for i, det in enumerate(found):
        color = _PALETTE[i % len(_PALETTE)]
        x0, y0, x1, y1 = det["box"]
        draw.rectangle([x0, y0, x1, y1], outline=color + (255,), width=stroke)
        if _flag(show_labels, True):
            tag = f"{det['label']} {det['score'] * 100:.0f}%"
            tb = draw.textbbox((0, 0), tag, font=font)
            tw, th = tb[2] - tb[0], tb[3] - tb[1]
            ly = max(0, y0 - th - stroke * 3)
            draw.rectangle([x0, ly, x0 + tw + stroke * 3, ly + th + stroke * 2],
                           fill=color + (230,))
            draw.text((x0 + stroke * 1.5, ly + stroke - tb[1]), tag, font=font,
                      fill=(20, 20, 24, 255))

    if found:
        counts = {}
        for d in found:
            counts[d["label"]] = counts.get(d["label"], 0) + 1
        summary = f"{len(found)} object(s) detected\n" + "\n".join(
            f"  • {n}× {label}" for label, n in
            sorted(counts.items(), key=lambda kv: -kv[1])
        )
    else:
        summary = "No objects above the score threshold."
    return _emit(img), summary


def mask_from_detections(image, detections, label_filter, min_score, feather, invert, preview):
    """Build an inpainting mask from detections — A1111's masked-region workflow.

    White = the region to repaint. ``label_filter`` accepts a comma-separated
    list ('cat, dog'); blank means every detection.
    """
    img = _load_image(image)
    w, h = img.size
    wanted = {t.strip().lower() for t in _text(label_filter).split(",") if t.strip()}

    mask = Image.new("L", (w, h), 0)
    draw = ImageDraw.Draw(mask)
    hits = 0
    for det in _boxes(detections, min_score):
        if wanted and det["label"].lower() not in wanted:
            continue
        x0, y0, x1, y1 = det["box"]
        draw.rectangle(
            [max(0, x0), max(0, y0), min(w - 1, x1), min(h - 1, y1)], fill=255
        )
        hits += 1

    if not hits:
        raise ValueError(
            "No detections matched — lower the score threshold or clear the label filter."
        )

    blur = _num(feather, 6, lo=0, hi=64)
    if blur > 0.5:
        mask = mask.filter(ImageFilter.GaussianBlur(blur))
    if _flag(invert, False):
        mask = ImageOps.invert(mask)

    if _flag(preview, False):
        # red overlay on the source, for eyeballing the region
        overlay = Image.new("RGB", (w, h), (255, 60, 60))
        return _emit(Image.composite(overlay, _rgb(img), mask.point(lambda v: v // 2)))
    return _emit(mask.convert("RGB"))


def contact_sheet(image_1, image_2, image_3, image_4, labels, columns, gap, title):
    """Tile up to four images into A1111's X/Y grid, with captions."""
    sources = [image_1, image_2, image_3, image_4]
    tiles = []
    for i, src in enumerate(sources):
        if src in (None, ""):
            continue
        try:
            tiles.append(_rgb(_load_image(src, f"image_{i + 1}")))
        except ValueError:
            continue
    if not tiles:
        raise ValueError("Connect at least one image to the contact sheet.")

    caption = [c.strip() for c in _text(labels).split("|")]
    cols = _num(columns, 2, lo=1, hi=4, integer=True)
    cols = min(cols, len(tiles))
    rows = (len(tiles) + cols - 1) // cols
    pad = _num(gap, 14, lo=0, hi=80, integer=True)

    cell = min(640, max(t.width for t in tiles))
    tiles = [ImageOps.fit(t, (cell, cell), Image.LANCZOS, centering=(0.5, 0.5)) for t in tiles]

    head = _text(title)
    label_h = max(26, cell // 16)
    head_h = int(label_h * 1.7) if head else 0

    sheet_w = cols * cell + pad * (cols + 1)
    sheet_h = head_h + rows * (cell + label_h) + pad * (rows + 1)
    sheet = Image.new("RGB", (sheet_w, sheet_h), (16, 16, 20))
    draw = ImageDraw.Draw(sheet)

    if head:
        f = _font(int(label_h * 0.95))
        draw.text((pad, pad // 2 + 2), head, font=f, fill=(240, 240, 245))

    f = _font(int(label_h * 0.68))
    for i, tile in enumerate(tiles):
        r, c = divmod(i, cols)
        x = pad + c * (cell + pad)
        y = head_h + pad + r * (cell + label_h + pad)
        sheet.paste(tile, (x, y))
        text = caption[i] if i < len(caption) and caption[i] else f"#{i + 1}"
        if len(text) > 46:
            text = text[:43] + "…"
        draw.text((x + 3, y + cell + 5), text, font=f, fill=(196, 199, 210))

    return _emit(sheet)


def png_info(image):
    """A1111's 'PNG Info' tab: recover generation parameters from a file.

    Returns (report, fields-as-JSON-text). The fields are serialized rather
    than returned as a dict because a `json` port arrives in the canvas as
    "[object Object]" (see `detect_objects`).
    """
    img = _load_image(image)
    img.load()  # force chunk parsing so text metadata is populated

    meta = {k: v for k, v in (img.info or {}).items()
            if isinstance(v, (str, int, float)) and k not in ("icc_profile",)}

    fields = {
        "width": img.width,
        "height": img.height,
        "mode": img.mode,
        "format": img.format or "unknown",
        "megapixels": round(img.width * img.height / 1e6, 2),
    }

    raw = str(meta.get("parameters") or meta.get("Comment") or "").strip()
    if raw:
        lines = raw.splitlines()
        fields["prompt"] = lines[0].strip()
        for line in lines[1:]:
            if line.lower().startswith("negative prompt:"):
                fields["negative_prompt"] = line.split(":", 1)[1].strip()
            elif ":" in line:
                for pair in re.split(r",\s*(?=[A-Z][A-Za-z ]*:)", line):
                    if ":" in pair:
                        k, v = pair.split(":", 1)
                        fields[k.strip().lower().replace(" ", "_")] = v.strip()

    try:
        exif = img.getexif()
        if exif:
            from PIL.ExifTags import TAGS
            for tag, value in exif.items():
                name = TAGS.get(tag, str(tag))
                if isinstance(value, (str, int, float)) and name != "MakerNote":
                    fields.setdefault(f"exif_{name.lower()}", value)
    except Exception:  # EXIF is best-effort; never fail the node over it
        pass

    head = f"{fields['format']}  {img.width}×{img.height}  ({fields['megapixels']} MP, {img.mode})"
    if raw:
        body = "\n\n── Generation parameters ──\n" + raw
    else:
        extra = [f"{k}: {v}" for k, v in meta.items()][:12]
        body = (
            "\n\nNo generation parameters embedded in this file.\n"
            + ("Other metadata:\n" + "\n".join(extra) if extra else
               "This image carries no text metadata at all.")
        )
    return head + body, json.dumps(fields, indent=2, default=str)


# What `app.py` binds onto the canvas. Keys must match the "fn" field of each
# operator node in workflow.json.
BIND = {
    "apply_style": apply_style,
    "build_negative": build_negative,
    "sampler_settings": sampler_settings,
    "generation_info": generation_info,
    "prompt_matrix": prompt_matrix,
    "magic_instruction": magic_instruction,
    "clean_prompt": clean_prompt,
    "txt2img": txt2img,
    "chat_llm": chat_llm,
    "interrogate": interrogate,
    "detect_objects": detect_objects,
    "classify_image": classify_image,
    "top_labels": top_labels,
    "postprocess": postprocess,
    "prep_image": prep_image,
    "extras_upscale": extras_upscale,
    "controlnet_preprocess": controlnet_preprocess,
    "draw_detections": draw_detections,
    "mask_from_detections": mask_from_detections,
    "contact_sheet": contact_sheet,
    "png_info": png_info,
}