blkot-ComfyUI-tgBridge/nodes.py

from __future__ import annotations

import json
from pathlib import Path
import os
import traceback

try:
    from .tgbridge.config import load_config, ConfigError
    from .tgbridge.formatter import build_caption, build_detail_messages, build_json_metadata
    from .tgbridge.metadata import build_metadata, resolve_send_type, select_files
    from .tgbridge.reference_image import save_reference_image
    from .tgbridge.telegram_bot import TelegramBotClient
except ImportError:
    from tgbridge.config import load_config, ConfigError
    from tgbridge.formatter import build_caption, build_detail_messages, build_json_metadata
    from tgbridge.metadata import build_metadata, resolve_send_type, select_files
    from tgbridge.reference_image import save_reference_image
    from tgbridge.telegram_bot import TelegramBotClient

class TelegramWorkflowReport:
    @classmethod
    def INPUT_TYPES(cls):
        string = ("STRING", {"default": ""})
        multiline = ("STRING", {"default": "", "multiline": True})
        return {
            "required": {
                "enabled": ("BOOLEAN", {"default": True}),
                "upload_mode": (["final_only", "all_files"], {"default": "final_only"}),
                "send_as": (["auto", "video", "photo", "document"], {"default": "auto"}),
                "message_mode": (
                    ["caption_only", "caption_and_reply", "caption_reply_and_json"],
                    {"default": "caption_and_reply"},
                ),
                "failure_mode": (["warn", "raise"], {"default": "warn"}),
            },
            "optional": {
                "filenames": ("VHS_FILENAMES",),
                "title": string,
                "positive_prompt": multiline,
                "negative_prompt": multiline,
                "extra_notes": multiline,
                "base_model": string,
                "base_model_obj": ("WANVIDEOMODEL",),
                "high_model": string,
                "high_model_obj": ("WANVIDEOMODEL",),
                "low_model": string,
                "low_model_obj": ("WANVIDEOMODEL",),
                "vae": string,
                "lora_1_name": string,
                "lora_1_strength": ("FLOAT", {"default": 0.0, "step": 0.01}),
                "lora_2_name": string,
                "lora_2_strength": ("FLOAT", {"default": 0.0, "step": 0.01}),
                "lora_3_name": string,
                "lora_3_strength": ("FLOAT", {"default": 0.0, "step": 0.01}),
                "lora_4_name": string,
                "lora_4_strength": ("FLOAT", {"default": 0.0, "step": 0.01}),
                "lora_5_name": string,
                "lora_5_strength": ("FLOAT", {"default": 0.0, "step": 0.01}),
                "high_loras_obj": ("WANVIDLORA",),
                "high_loras": multiline,
                "low_loras_obj": ("WANVIDLORA",),
                "low_loras": multiline,
                "extra_loras": multiline,
                "seed": ("INT", {"default": -1, "min": -1}),
                "steps": ("INT", {"default": 0, "min": 0}),
                "cfg": ("FLOAT", {"default": 0.0, "min": 0.0, "step": 0.1}),
                "sampler": string,
                "scheduler": string,
                "width": ("INT", {"default": 0, "min": 0}),
                "height": ("INT", {"default": 0, "min": 0}),
                "frames": ("INT", {"default": 0, "min": 0}),
                "fps": ("FLOAT", {"default": 0.0, "min": 0.0, "step": 0.01}),
                "reference_label": string,
                "reference_image": ("IMAGE",),
                "send_reference_image": ("BOOLEAN", {"default": False}),
            },
        }

    RETURN_TYPES = ("VHS_FILENAMES", "STRING",)
    RETURN_NAMES = ("filenames", "message_preview",)
    FUNCTION = "upload"
    CATEGORY = "Telegram"
    OUTPUT_NODE = True

    def _extract_model_name(self, model_obj):
        """Extract model name from WANVIDEOMODEL object."""
        try:
            model_name = model_obj.model["model_name"]
        except (KeyError, AttributeError, TypeError):
            return None
        if isinstance(model_name, str) and model_name:
            return os.path.basename(model_name)
        return None

    def _format_lora_list(self, lora_list):
        """Format WANVIDLORA list as string (one per line)."""
        if lora_list is None:
            return None
        lora_entries = []
        if isinstance(lora_list, list):
            for lora in lora_list:
                if isinstance(lora, dict):
                    name = lora.get("name", "unknown")
                    strength = lora.get("strength", 0.0)
                    if isinstance(strength, list):
                        strength_str = str(strength)
                    else:
                        strength_str = f"{strength:.2f}"
                    lora_entries.append(f"{name}({strength_str})")
        return "\n".join(lora_entries) if lora_entries else None

    def _process_model_inputs(self, kwargs):
        """Process WANVIDEOMODEL and WANVIDLORA inputs to extract strings."""
        result = dict(kwargs)

        # Extract model names from WANVIDEOMODEL objects (obj inputs take priority)
        if "base_model_obj" in kwargs and kwargs["base_model_obj"] is not None:
            name = self._extract_model_name(kwargs["base_model_obj"])
            if name and not result.get("base_model"):
                result["base_model"] = name

        if "high_model_obj" in kwargs and kwargs["high_model_obj"] is not None:
            name = self._extract_model_name(kwargs["high_model_obj"])
            if name and not result.get("high_model"):
                result["high_model"] = name

        if "low_model_obj" in kwargs and kwargs["low_model_obj"] is not None:
            name = self._extract_model_name(kwargs["low_model_obj"])
            if name and not result.get("low_model"):
                result["low_model"] = name

        # Extract lora lists from WANVIDLORA objects (obj inputs take priority)
        if "high_loras_obj" in kwargs and kwargs["high_loras_obj"] is not None:
            formatted = self._format_lora_list(kwargs["high_loras_obj"])
            if formatted and not result.get("high_loras"):
                result["high_loras"] = formatted

        if "low_loras_obj" in kwargs and kwargs["low_loras_obj"] is not None:
            formatted = self._format_lora_list(kwargs["low_loras_obj"])
            if formatted and not result.get("low_loras"):
                result["low_loras"] = formatted

        return result

    def upload(
        self,
        filenames=None,
        enabled=True,
        upload_mode="final_only",
        send_as="auto",
        message_mode="caption_and_reply",
        failure_mode="warn",
        **kwargs,
    ):
        # Process WANVIDEOMODEL and WANVIDLORA inputs to extract strings
        processed_kwargs = self._process_model_inputs(kwargs)

        # Try to load config, but use defaults for testing if not configured
        try:
            config = load_config()
            parse_mode = config.parse_mode
        except ConfigError:
            # Config not set up - use defaults for testing/preview
            parse_mode = "HTML"
            print("[ComfyUI-tgBridge] Config not found - running in test mode (preview only)")

        # Build preview message regardless of enabled state or filenames
        if filenames is not None:
            _, output_files = filenames
            files, _ = select_files(output_files, upload_mode)
        else:
            files = []

        metadata = build_metadata(processed_kwargs, files)
        caption = build_caption(metadata, parse_mode)
        detail_messages = list(build_detail_messages(metadata, parse_mode))

        # Build preview text
        preview_lines = ["=== Telegram Message Preview ==="]
        preview_lines.append(f"\n📎 Caption:\n{caption}")
        if detail_messages:
            preview_lines.append("\n📝 Follow-up Messages:")
            for i, msg in enumerate(detail_messages, 1):
                preview_lines.append(f"  {i}. {msg}")
        preview_lines.append(f"\n📁 Files to upload: {[str(f) for f in files]}")
        preview_lines.append(f"📤 Send as: {send_as}")
        preview_lines.append(f"💬 Message mode: {message_mode}")
        message_preview = "\n".join(preview_lines)

        if not enabled or filenames is None:
            return (filenames, message_preview)

        try:
            self._upload(
                filenames=filenames,
                upload_mode=upload_mode,
                send_as=send_as,
                message_mode=message_mode,
                **processed_kwargs,
            )
        except ConfigError as exc:
            # Config not set up - informative message for user
            if failure_mode == "raise":
                raise
            print(f"[ComfyUI-tgBridge] Telegram not configured: {exc}")
            print("[ComfyUI-tgBridge] Set COMFY_TG_BOT_TOKEN and COMFY_TG_CHAT_ID environment variables, or create config/telegram.json")
        except Exception as exc:
            if failure_mode == "raise":
                raise
            print(f"[ComfyUI-tgBridge] Telegram upload skipped: {exc}")
            traceback.print_exc()

        return (filenames, message_preview)

    def _upload(self, filenames, upload_mode, send_as, message_mode, **kwargs):
        _save_output, output_files = filenames
        files, missing = select_files(output_files, upload_mode)
        for missing_path in missing:
            print(f"[ComfyUI-tgBridge] Missing output file skipped: {missing_path}")
        if not files:
            print("[ComfyUI-tgBridge] No files to send - skipping Telegram upload (test mode)")
            return

        config = load_config()
        client = TelegramBotClient(config)
        metadata = build_metadata(kwargs, files)
        caption = build_caption(metadata, config.parse_mode)

        first_message_id = None
        for index, path in enumerate(files):
            media_caption = caption if index == 0 else ""
            send_type = resolve_send_type(path, send_as)
            sent = client.send_media(path, send_type, media_caption)
            if first_message_id is None:
                first_message_id = sent.message_id

        if message_mode in {"caption_and_reply", "caption_reply_and_json"}:
            for message in build_detail_messages(metadata, config.parse_mode):
                client.send_message(message, reply_to_message_id=first_message_id)

        reference_path = None
        try:
            if kwargs.get("send_reference_image") and kwargs.get("reference_image") is not None:
                reference_path = save_reference_image(kwargs["reference_image"])
                caption_text = kwargs.get("reference_label") or "Reference image"
                client.send_photo(reference_path, caption=caption_text, reply_to_message_id=first_message_id)
        finally:
            if reference_path:
                Path(reference_path).unlink(missing_ok=True)

        if message_mode == "caption_reply_and_json":
            content = build_json_metadata(metadata)
            client.send_document_bytes(
                content,
                filename="comfyui_generation_metadata.json",
                caption="Generation metadata JSON",
                reply_to_message_id=first_message_id,
            )


class WanVideoModelToString:
    """Helper node to extract model name from WANVIDEOMODEL for use in string inputs."""

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {
                "model": ("WANVIDEOMODEL",),
            },
        }

    RETURN_TYPES = ("STRING",)
    RETURN_NAMES = ("model_name",)
    FUNCTION = "extract"
    CATEGORY = "Telegram"
    DESCRIPTION = "Extracts the model name from a WANVIDEOMODEL object as a string."

    def extract(self, model):
        # The model name is stored in model.model["model_name"] via __getitem__
        try:
            model_name = model.model["model_name"]
        except (KeyError, AttributeError, TypeError):
            model_name = "unknown"
        # If it's a full path, get just the basename
        if isinstance(model_name, str) and model_name != "unknown":
            model_name = os.path.basename(model_name)
        return (model_name,)


class WanVideoLoraListToString:
    """Helper node to extract lora names and strengths from WANVIDLORA for use in string inputs."""

    @classmethod
    def INPUT_TYPES(cls):
        return {
            "required": {},
            "optional": {
                "high_loras": ("WANVIDLORA",),
                "low_loras": ("WANVIDLORA",),
                "format": (["list", "comma_separated", "numbered"], {"default": "list"}),
            },
        }

    RETURN_TYPES = ("STRING", "STRING",)
    RETURN_NAMES = ("high_loras", "low_loras",)
    FUNCTION = "extract"
    CATEGORY = "Telegram"
    DESCRIPTION = "Extracts lora names and strengths from WANVIDLORA lists as formatted strings."

    def extract(self, high_loras=None, low_loras=None, format="comma_separated"):
        def format_lora_list(lora_list, fmt):
            if lora_list is None:
                return ""
            lora_entries = []
            if isinstance(lora_list, list):
                for lora in lora_list:
                    if isinstance(lora, dict):
                        name = lora.get("name", "unknown")
                        strength = lora.get("strength", 0.0)
                        if isinstance(strength, list):
                            strength_str = str(strength)
                        else:
                            strength_str = f"{strength:.2f}"
                        lora_entries.append(f"{name}({strength_str})")

            if fmt == "list":
                return "\n".join(lora_entries) if lora_entries else ""
            elif fmt == "numbered":
                return "\n".join(f"{i+1}. {entry}" for i, entry in enumerate(lora_entries)) if lora_entries else ""
            else:  # comma_separated
                return ", ".join(lora_entries) if lora_entries else ""

        return (
            format_lora_list(high_loras, format),
            format_lora_list(low_loras, format),
        )


class TelegramAutoReport:
    """Simplified Telegram report that reads workflow metadata from saved files."""

    @classmethod
    def INPUT_TYPES(cls):
        string = ("STRING", {"default": ""})
        multiline = ("STRING", {"default": "", "multiline": True})
        return {
            "required": {
                "filenames": ("VHS_FILENAMES",),
                "enabled": ("BOOLEAN", {"default": True}),
            },
            "optional": {
                "title": string,
                "extra_notes": multiline,
                "send_as": (["auto", "video", "photo", "document"], {"default": "auto"}),
                "message_mode": (
                    ["caption_only", "caption_and_reply", "caption_reply_and_json"],
                    {"default": "caption_and_reply"},
                ),
                "failure_mode": (["warn", "raise"], {"default": "warn"}),
            },
        }

    RETURN_TYPES = ("VHS_FILENAMES", "STRING",)
    RETURN_NAMES = ("filenames", "message_preview",)
    FUNCTION = "upload"
    CATEGORY = "Telegram"
    OUTPUT_NODE = True

    def _metadata_candidates(self, output_files):
        candidates = []
        seen = set()

        def add(path):
            candidate = Path(path)
            key = str(candidate)
            if key not in seen and candidate.exists():
                seen.add(key)
                candidates.append(candidate)

        for raw_path in output_files:
            path = Path(raw_path)
            add(path)
            add(path.with_suffix(".json"))
            add(path.with_suffix(".png"))
            add(path.with_suffix(".webp"))

        return candidates

    def _read_metadata_payload(self, metadata_path):
        path = Path(metadata_path)
        try:
            if path.suffix.lower() == ".json":
                return json.loads(path.read_text(encoding="utf-8"))
            if path.suffix.lower() in {".png", ".webp"}:
                return self._read_image_metadata(path)
        except Exception as exc:
            print(f"[ComfyUI-tgBridge] Failed to read metadata from {path}: {exc}")
        return None

    def _read_image_metadata(self, image_path):
        try:
            from PIL import Image
        except ImportError:
            print("[ComfyUI-tgBridge] Pillow is required to read embedded image metadata")
            return None

        with Image.open(image_path) as image:
            info = dict(image.info)

        payload = {}
        for key in ("prompt", "workflow"):
            value = info.get(key)
            if value is not None:
                payload[key] = self._parse_json_value(value)
        return payload or None

    def _extract_from_metadata_files(self, output_files):
        for path in self._metadata_candidates(output_files):
            payload = self._read_metadata_payload(path)
            if not isinstance(payload, dict):
                continue

            extracted = self._extract_from_payload(payload)
            if extracted:
                print(f"[ComfyUI-tgBridge] Extracted metadata from: {path.name}")
                return extracted

        return {}

    def _extract_from_payload(self, data):
        extracted = {}
        workflow = self._parse_json_value(data.get("workflow"))
        prompt = self._parse_json_value(data.get("prompt"))

        if isinstance(workflow, dict):
            title = workflow.get("title") or workflow.get("name")
            if title:
                extracted["title"] = title

        if isinstance(prompt, dict):
            prompt_values = self._extract_from_prompt_graph(prompt)
            extracted.update({key: value for key, value in prompt_values.items() if value not in ("", None, ())})

        if isinstance(workflow, dict) and workflow.get("nodes"):
            workflow_values = self._extract_from_workflow_nodes(workflow["nodes"])
            for key, value in workflow_values.items():
                if key not in extracted and value not in ("", None, ()):
                    extracted[key] = value

        return extracted

    def _extract_from_prompt_graph(self, prompt):
        nodes = {
            str(node_id): node
            for node_id, node in prompt.items()
            if isinstance(node, dict)
        }
        extracted = {}

        sampler = self._find_sampler_node(nodes.values())
        if sampler:
            inputs = sampler.get("inputs", {})
            self._copy_first(extracted, inputs, "seed", ("seed", "noise_seed"))
            self._copy_first(extracted, inputs, "steps", ("steps",))
            self._copy_first(extracted, inputs, "cfg", ("cfg", "cfg_scale"))
            self._copy_first(extracted, inputs, "sampler", ("sampler_name", "sampler"))
            self._copy_first(extracted, inputs, "scheduler", ("scheduler",))

            positive = self._resolve_prompt_text(nodes, inputs.get("positive"))
            negative = self._resolve_prompt_text(nodes, inputs.get("negative"))
            if positive:
                extracted["positive_prompt"] = positive
            if negative:
                extracted["negative_prompt"] = negative

        if not extracted.get("positive_prompt"):
            positive, negative = self._fallback_prompt_text(nodes.values())
            if positive:
                extracted["positive_prompt"] = positive
            if negative:
                extracted["negative_prompt"] = negative

        self._extract_models_and_loras_from_prompt(nodes.values(), extracted)
        self._extract_video_settings(nodes.values(), extracted)
        return extracted

    def _extract_from_workflow_nodes(self, nodes):
        extracted = {}
        positive = []
        negative = []

        for node in nodes:
            if not isinstance(node, dict):
                continue
            node_type = str(node.get("type") or node.get("class_type") or "")
            title = str(node.get("title") or "")
            values = node.get("widgets_values")

            text_value = self._workflow_text_value(values)
            label = f"{node_type} {title}".lower()
            if text_value and "negative" in label:
                negative.append(text_value)
            elif text_value and ("positive" in label or "prompt" in label or "text" in label):
                positive.append(text_value)

            self._extract_model_from_inputs(node_type, self._workflow_inputs(node), extracted)

        if positive:
            extracted["positive_prompt"] = "\n".join(positive)
        if negative:
            extracted["negative_prompt"] = "\n".join(negative)
        return extracted

    def _find_sampler_node(self, nodes):
        for node in nodes:
            class_type = str(node.get("class_type", ""))
            if "sampler" in class_type.lower():
                return node
        return None

    def _resolve_prompt_text(self, nodes, link):
        node_id = self._linked_node_id(link)
        if node_id is None or node_id not in nodes:
            return ""
        return self._collect_text_from_node(nodes, node_id, set())

    def _collect_text_from_node(self, nodes, node_id, visited):
        if node_id in visited:
            return ""
        visited.add(node_id)

        node = nodes.get(node_id, {})
        inputs = node.get("inputs", {})
        text = inputs.get("text")
        if isinstance(text, str):
            return text

        texts = []
        for value in inputs.values():
            linked_id = self._linked_node_id(value)
            if linked_id is not None:
                linked_text = self._collect_text_from_node(nodes, linked_id, visited)
                if linked_text:
                    texts.append(linked_text)
        return "\n".join(texts)

    def _fallback_prompt_text(self, nodes):
        positive = []
        negative = []
        unlabeled = []
        for node in nodes:
            class_type = str(node.get("class_type", ""))
            inputs = node.get("inputs", {})
            text = inputs.get("text")
            if not isinstance(text, str) or not text.strip():
                continue
            label = class_type.lower()
            if "negative" in label:
                negative.append(text)
            elif "positive" in label:
                positive.append(text)
            else:
                unlabeled.append(text)

        if not positive and unlabeled:
            positive.append(unlabeled[0])
            negative.extend(unlabeled[1:2])
        return "\n".join(positive), "\n".join(negative)

    def _extract_models_and_loras_from_prompt(self, nodes, extracted):
        for node in nodes:
            class_type = str(node.get("class_type", ""))
            inputs = node.get("inputs", {})
            self._extract_model_from_inputs(class_type, inputs, extracted)
            self._extract_loras_from_inputs(class_type, inputs, extracted)

    def _extract_model_from_inputs(self, class_type, inputs, extracted):
        lowered = class_type.lower()
        if "model" not in lowered and "loader" not in lowered and "checkpoint" not in lowered:
            return

        model_name = self._first_string(
            inputs,
            (
                "model_name",
                "ckpt_name",
                "unet_name",
                "checkpoint",
                "checkpoint_name",
                "vae_name",
            ),
        )
        if not model_name:
            return

        model_name = os.path.basename(model_name)
        if "vae" in lowered or "vae_name" in inputs:
            extracted.setdefault("vae", model_name)
        elif "high" in lowered:
            extracted.setdefault("high_model", model_name)
        elif "low" in lowered:
            extracted.setdefault("low_model", model_name)
        else:
            extracted.setdefault("base_model", model_name)

    def _extract_loras_from_inputs(self, class_type, inputs, extracted):
        if "lora" not in class_type.lower() and not any("lora" in str(key).lower() for key in inputs):
            return

        entries = []
        for key, value in inputs.items():
            key_text = str(key).lower()
            if "lora" in key_text and isinstance(value, str) and value.lower() != "none":
                strength = self._lora_strength(inputs)
                entries.append(self._format_lora(value, strength))
            elif "lora" in key_text and isinstance(value, list):
                entries.extend(self._format_lora_list(value))

        if not entries:
            return

        target = "high_loras" if "high" in class_type.lower() else "low_loras" if "low" in class_type.lower() else "extra_loras"
        existing = extracted.get(target, "")
        extracted[target] = "\n".join(line for line in (existing, "\n".join(entries)) if line)

    def _extract_video_settings(self, nodes, extracted):
        for node in nodes:
            inputs = node.get("inputs", {})
            self._copy_first(extracted, inputs, "width", ("width", "empty_latent_width"))
            self._copy_first(extracted, inputs, "height", ("height", "empty_latent_height"))
            self._copy_first(extracted, inputs, "frames", ("frames", "num_frames", "frame_count", "video_length", "length"))
            self._copy_first(extracted, inputs, "fps", ("fps", "frame_rate"))

    def _workflow_inputs(self, node):
        inputs = {}
        raw_inputs = node.get("inputs")
        if isinstance(raw_inputs, dict):
            return raw_inputs
        if isinstance(raw_inputs, list):
            for item in raw_inputs:
                if isinstance(item, dict) and "name" in item and "value" in item:
                    inputs[item["name"]] = item["value"]
        return inputs

    def _workflow_text_value(self, values):
        if isinstance(values, str):
            return values
        if isinstance(values, list):
            for value in values:
                if isinstance(value, str) and "\n" in value:
                    return value
            for value in values:
                if isinstance(value, str) and value.strip():
                    return value
        if isinstance(values, dict):
            value = values.get("text")
            if isinstance(value, str):
                return value
        return ""

    def _format_lora_list(self, values):
        entries = []
        for value in values:
            if isinstance(value, dict):
                name = value.get("name") or value.get("lora_name")
                if name and str(name).lower() != "none":
                    entries.append(self._format_lora(str(name), value.get("strength")))
            elif isinstance(value, str) and value.lower() != "none":
                entries.append(value)
        return entries

    def _format_lora(self, name, strength):
        if strength in (None, ""):
            return os.path.basename(name)
        try:
            return f"{os.path.basename(name)}: {float(strength):g}"
        except (TypeError, ValueError):
            return f"{os.path.basename(name)}: {strength}"

    def _lora_strength(self, inputs):
        for key in ("strength_model", "strength_clip", "strength", "lora_strength"):
            value = inputs.get(key)
            if value not in (None, ""):
                return value
        return None

    def _first_string(self, inputs, keys):
        for key in keys:
            value = inputs.get(key)
            if isinstance(value, str) and value and value.lower() != "none":
                return value
        return ""

    def _copy_first(self, extracted, inputs, target, keys):
        if target in extracted:
            return
        for key in keys:
            value = inputs.get(key)
            if value not in (None, "") and not isinstance(value, (list, dict)):
                extracted[target] = value
                return

    def _linked_node_id(self, value):
        if isinstance(value, (list, tuple)) and value:
            return str(value[0])
        return None

    def _parse_json_value(self, value):
        if isinstance(value, str):
            try:
                return json.loads(value)
            except json.JSONDecodeError:
                return value
        return value

    def upload(
        self,
        filenames,
        enabled=True,
        title="",
        extra_notes="",
        send_as="auto",
        message_mode="caption_and_reply",
        failure_mode="warn",
    ):
        if filenames is None:
            return (filenames, "")

        # Try to load config
        config = None
        config_error = None
        try:
            config = load_config()
            parse_mode = config.parse_mode
        except ConfigError as exc:
            parse_mode = "HTML"
            config_error = exc
            print("[ComfyUI-tgBridge] Config not found - running in test mode (preview only)")

        # Get output files
        _, output_files = filenames
        files, _ = select_files(output_files, "final_only")

        if not files:
            return (filenames, "[Telegram Auto] No files to process")

        extracted = self._extract_from_metadata_files(output_files)
        metadata_values = {
            "title": title or extracted.get("title", ""),
            "positive_prompt": extracted.get("positive_prompt", ""),
            "negative_prompt": extracted.get("negative_prompt", ""),
            "extra_notes": extra_notes,
            "base_model": extracted.get("base_model", ""),
            "high_model": extracted.get("high_model", ""),
            "low_model": extracted.get("low_model", ""),
            "vae": extracted.get("vae", ""),
            "high_loras": extracted.get("high_loras", ""),
            "low_loras": extracted.get("low_loras", ""),
            "extra_loras": extracted.get("extra_loras", ""),
            "seed": extracted.get("seed", -1),
            "steps": extracted.get("steps", 0),
            "cfg": extracted.get("cfg", 0),
            "sampler": extracted.get("sampler", ""),
            "scheduler": extracted.get("scheduler", ""),
            "width": extracted.get("width", 0),
            "height": extracted.get("height", 0),
            "frames": extracted.get("frames", 0),
            "fps": extracted.get("fps", 0),
        }
        metadata = build_metadata(metadata_values, files)

        # Build caption and messages
        caption = build_caption(metadata, parse_mode)
        detail_messages = list(build_detail_messages(metadata, parse_mode))

        # Build preview
        preview_lines = ["=== Telegram Auto Report Preview ==="]
        preview_lines.append(f"\n📎 Caption:\n{caption}")
        if detail_messages:
            preview_lines.append("\n📝 Follow-up Messages:")
            for i, msg in enumerate(detail_messages, 1):
                preview_lines.append(f"  {i}. {msg}")
        preview_lines.append(f"\n📁 Files to upload: {[str(f) for f in files]}")
        message_preview = "\n".join(preview_lines)

        if not enabled:
            return (filenames, message_preview)

        if config_error is not None:
            if failure_mode == "raise":
                raise config_error
            print(f"[ComfyUI-tgBridge] Telegram not configured: {config_error}")
            return (filenames, message_preview)

        try:
            client = TelegramBotClient(config)

            # Send files
            first_message_id = None
            for index, path in enumerate(files):
                media_caption = caption if index == 0 else ""
                send_type = resolve_send_type(path, send_as)
                sent = client.send_media(path, send_type, media_caption)
                if first_message_id is None:
                    first_message_id = sent.message_id

            # Send follow-up messages
            if message_mode in {"caption_and_reply", "caption_reply_and_json"}:
                for message in detail_messages:
                    client.send_message(message, reply_to_message_id=first_message_id)

            # Send JSON metadata if requested
            if message_mode == "caption_reply_and_json":
                content = build_json_metadata(metadata)
                client.send_document_bytes(
                    content,
                    filename="comfyui_generation_metadata.json",
                    caption="Generation metadata JSON",
                    reply_to_message_id=first_message_id,
                )

        except ConfigError as exc:
            if failure_mode == "raise":
                raise
            print(f"[ComfyUI-tgBridge] Telegram not configured: {exc}")
        except Exception as exc:
            if failure_mode == "raise":
                raise
            print(f"[ComfyUI-tgBridge] Upload skipped: {exc}")
            traceback.print_exc()

        return (filenames, message_preview)


NODE_CLASS_MAPPINGS = {
    "TelegramWorkflowReport": TelegramWorkflowReport,
    "TelegramAutoReport": TelegramAutoReport,
    "WanVideoModelToString": WanVideoModelToString,
    "WanVideoLoraListToString": WanVideoLoraListToString,
}

NODE_DISPLAY_NAME_MAPPINGS = {
    "TelegramWorkflowReport": "Telegram Workflow Report",
    "TelegramAutoReport": "Telegram Auto Workflow Report",
    "WanVideoModelToString": "WanVideo Model → String",
    "WanVideoLoraListToString": "WanVideo Lora List → String",
}