From 2e60fe3cbe6effc11440e0b89d3d9f73b8fedba1 Mon Sep 17 00:00:00 2001 From: CyberShadowX Date: Thu, 28 May 2026 14:38:32 +0800 Subject: [PATCH] Resolve nested Smart Metadata Reader text outputs --- smart_metadata_reader/adapters/base.py | 44 +++- smart_metadata_reader/adapters/clip_text.py | 4 + smart_metadata_reader/adapters/pysssss.py | 4 + .../adapters/smart_reader.py | 244 ++++++++++++++++++ .../adapters/text_sources.py | 4 + smart_metadata_reader/graph.py | 21 ++ smart_metadata_reader/metadata_reader.py | 24 ++ smart_metadata_reader/models.py | 3 + smart_metadata_reader/resolver.py | 22 ++ tests/test_metadata_reader.py | 236 +++++++++++++++++ tests/test_node_contract.py | 13 + 11 files changed, 618 insertions(+), 1 deletion(-) create mode 100644 smart_metadata_reader/adapters/smart_reader.py diff --git a/smart_metadata_reader/adapters/base.py b/smart_metadata_reader/adapters/base.py index 766f6c2..205bbeb 100644 --- a/smart_metadata_reader/adapters/base.py +++ b/smart_metadata_reader/adapters/base.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Protocol +from typing import Any, Protocol from ..graph import GraphIndex from ..models import NodeRecord, PromptSegment @@ -18,6 +18,7 @@ class NodeAdapter(Protocol): role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: ... @@ -44,12 +45,24 @@ class TextResolutionContext: prefer_cached_text: bool = True, trace: TraceCollector | None = None, max_depth: int = 40, + parse_max_depth: int | None = None, + parameter_index: int = 0, + include_raw_json: bool = True, + base_dir: str | None = None, + recursion_stack: tuple[str, ...] = (), + unresolved: list[dict[str, Any]] | None = None, ) -> None: self.graph = graph self.registry = registry or AdapterRegistry() self.prefer_cached_text = prefer_cached_text self.trace = trace or TraceCollector() self.max_depth = max_depth + self.parse_max_depth = parse_max_depth if parse_max_depth is not None else max_depth + self.parameter_index = parameter_index + self.include_raw_json = include_raw_json + self.base_dir = base_dir + self.recursion_stack = recursion_stack + self.unresolved = unresolved def resolve_node( self, @@ -57,6 +70,7 @@ class TextResolutionContext: role: str, path: list[str], depth: int = 0, + output_index: int | None = None, ) -> list[PromptSegment]: if depth > self.max_depth: self.trace.add("TEXT_TRACE", f"max depth reached at node {node_id}") @@ -87,6 +101,7 @@ class TextResolutionContext: role=role, path=path + [f"{node.class_type} {node.node_id}"], depth=depth + 1, + output_index=output_index, ) def resolve_input( @@ -140,6 +155,7 @@ class TextResolutionContext: role=role, path=path + [f"link[{target_id}, {output_index}]"], depth=depth + 1, + output_index=output_index, ) if isinstance(value, list): @@ -173,3 +189,29 @@ class TextResolutionContext: lowered = node.class_type.lower() markers = ("gemini", "chatgpt", "claude", "llm", "language_model") return any(marker in lowered for marker in markers) + + def record_unresolved( + self, + node: NodeRecord, + field: str, + role: str, + reason: str, + **extra: Any, + ) -> None: + entry: dict[str, Any] = { + "node_id": node.node_id, + "class_type": node.class_type, + "field": field, + "role": role, + "reason": reason, + } + entry.update(extra) + if self.unresolved is not None: + self.unresolved.append(entry) + self.trace.add( + f"{role.upper()}_TRACE", + ( + f"UNRESOLVED {node.class_type} {node.node_id} " + f"field={field} role={role} reason={reason}" + ), + ) diff --git a/smart_metadata_reader/adapters/clip_text.py b/smart_metadata_reader/adapters/clip_text.py index 90172cf..aa80a7d 100644 --- a/smart_metadata_reader/adapters/clip_text.py +++ b/smart_metadata_reader/adapters/clip_text.py @@ -15,7 +15,9 @@ class CLIPTextEncodeAdapter: role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: + del output_index return context.resolve_input(node, "text", role, path, depth) @@ -30,7 +32,9 @@ class CLIPTextEncodeSDXLAdapter: role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: + del output_index segments: list[PromptSegment] = [] for field in ("text", "text_g", "text_l"): if field in node.inputs: diff --git a/smart_metadata_reader/adapters/pysssss.py b/smart_metadata_reader/adapters/pysssss.py index f54257a..86f286f 100644 --- a/smart_metadata_reader/adapters/pysssss.py +++ b/smart_metadata_reader/adapters/pysssss.py @@ -15,7 +15,9 @@ class StringFunctionAdapter: role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: + del output_index action = str(node.inputs.get("action", "append")).lower() fields = ("text_a", "text_b", "text_c") if action != "append": @@ -41,7 +43,9 @@ class ShowTextAdapter: role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: + del output_index cached_text = self._cached_text(node, context) if cached_text: text, source = cached_text diff --git a/smart_metadata_reader/adapters/smart_reader.py b/smart_metadata_reader/adapters/smart_reader.py new file mode 100644 index 0000000..a4f05e2 --- /dev/null +++ b/smart_metadata_reader/adapters/smart_reader.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import os +import re +from pathlib import Path +from typing import Any + +from .base import TextResolutionContext +from ..models import NodeRecord, PromptSegment + + +SMART_READER_OUTPUT_FALLBACK = { + 0: "image", + 1: "mask", + 2: "positive", + 3: "negative", +} +PROMPT_OUTPUT_ROLES = {"positive", "negative"} +ALLOWED_IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp"} + + +class SmartMetadataReaderAdapter: + def matches(self, node: NodeRecord) -> bool: + return node.class_type == "SmartMetadataReader" + + def resolve( + self, + node: NodeRecord, + context: TextResolutionContext, + role: str, + path: list[str], + depth: int, + output_index: int | None = None, + ) -> list[PromptSegment]: + if output_index is None: + return self._unresolved( + node, + context, + role, + output_index=-1, + output_role="unknown", + referenced_image="", + reason="SmartMetadataReader output index missing", + ) + + output_role = self._output_role(node, context, output_index) + image_name = self._image_name(node) + if output_role not in PROMPT_OUTPUT_ROLES: + return self._unresolved( + node, + context, + role, + output_index=output_index, + output_role=output_role or "unknown", + referenced_image=image_name or "", + reason="SmartMetadataReader output is not a prompt text output", + ) + + if role != output_role: + context.trace.add( + f"{role.upper()}_TRACE", + f"SmartMetadataReader output {output_role} used in {role} chain", + ) + + image_path, failure_reason = self._resolve_referenced_image(context, image_name) + if image_path is None: + return self._unresolved( + node, + context, + role, + output_index=output_index, + output_role=output_role, + referenced_image=image_name or "", + reason=failure_reason, + ) + + stack_key = os.path.normcase(str(image_path)) + if stack_key in context.recursion_stack: + return self._unresolved( + node, + context, + role, + output_index=output_index, + output_role=output_role, + referenced_image=image_name or "", + reason="recursive SmartMetadataReader image reference", + ) + + try: + from ..metadata_reader import parse_metadata_bundle, read_metadata + + nested_bundle = read_metadata(image_path) + nested_result = parse_metadata_bundle( + nested_bundle, + parameter_index=context.parameter_index, + prefer_cached_text=context.prefer_cached_text, + include_raw_json=context.include_raw_json, + max_depth=context.parse_max_depth, + _recursion_stack=context.recursion_stack + (stack_key,), + ) + except Exception as exc: + return self._unresolved( + node, + context, + role, + output_index=output_index, + output_role=output_role, + referenced_image=image_name or "", + reason=f"referenced image not resolvable: {exc}", + ) + + text = nested_result.positive if output_role == "positive" else nested_result.negative + if not text and nested_result.status_message in {"FAILED", "NO_METADATA", "PARTIAL"}: + return self._unresolved( + node, + context, + role, + output_index=output_index, + output_role=output_role, + referenced_image=image_name or "", + reason=( + "SmartMetadataReader output is runtime value; " + "referenced image did not resolve prompt text" + ), + ) + + context.trace.add( + f"{role.upper()}_TRACE", + ( + f"Resolved SmartMetadataReader {output_role} output " + f"from referenced image {image_name}" + ), + ) + if not text.strip(): + return [] + return [ + PromptSegment( + text=text, + node_id=node.node_id, + class_type=node.class_type, + field=output_role, + path=path + + [ + f"{node.class_type} {node.node_id}.output[{output_index}] {output_role}", + f"referenced image {image_name}", + ], + confidence=min(max(nested_result.confidence, 0.75), 0.95), + ) + ] + + def _output_role( + self, + node: NodeRecord, + context: TextResolutionContext, + output_index: int, + ) -> str | None: + output_name = context.graph.workflow_output_name(node.node_id, output_index) + role = _role_from_output_name(output_name) + if role is not None: + return role + return SMART_READER_OUTPUT_FALLBACK.get(output_index) + + def _image_name(self, node: NodeRecord) -> str | None: + value = node.inputs.get("image") + if isinstance(value, str) and value: + return value + widgets = node.widgets_values + if isinstance(widgets, dict): + value = widgets.get("image") + if isinstance(value, str) and value: + return value + elif isinstance(widgets, list): + for value in widgets: + if isinstance(value, str) and _allowed_image_extension(value): + return value + return None + + def _resolve_referenced_image( + self, + context: TextResolutionContext, + image_name: str | None, + ) -> tuple[Path | None, str]: + if not image_name: + return None, "SmartMetadataReader referenced image filename missing" + if not context.base_dir: + return None, "SmartMetadataReader referenced image base_dir unavailable" + if not _safe_relative_image_path(image_name): + return None, "unsafe referenced image path" + + base_dir = Path(context.base_dir).resolve() + candidate = (base_dir / image_name.replace("\\", "/")).resolve() + try: + candidate.relative_to(base_dir) + except ValueError: + return None, "unsafe referenced image path" + if not candidate.exists(): + return None, "referenced image not found" + return candidate, "" + + def _unresolved( + self, + node: NodeRecord, + context: TextResolutionContext, + role: str, + output_index: int, + output_role: str, + referenced_image: str, + reason: str, + ) -> list[PromptSegment]: + context.record_unresolved( + node=node, + field=f"output[{output_index}]", + role=role, + reason=reason, + output_index=output_index, + resolved_output_role=output_role, + referenced_image_filename=referenced_image, + ) + return [] + + +def _role_from_output_name(output_name: str | None) -> str | None: + if not output_name: + return None + normalized = output_name.strip().lower().replace(" ", "_") + if normalized in PROMPT_OUTPUT_ROLES: + return normalized + return None + + +def _safe_relative_image_path(image_name: str) -> bool: + normalized = image_name.replace("\\", "/") + if not _allowed_image_extension(normalized): + return False + if Path(normalized).is_absolute() or re.match(r"^[A-Za-z]:", normalized): + return False + parts = [part for part in normalized.split("/") if part] + if not parts or any(part == ".." for part in parts): + return False + return True + + +def _allowed_image_extension(image_name: str) -> bool: + return Path(image_name).suffix.lower() in ALLOWED_IMAGE_EXTENSIONS diff --git a/smart_metadata_reader/adapters/text_sources.py b/smart_metadata_reader/adapters/text_sources.py index 4dd894b..8d8bf9a 100644 --- a/smart_metadata_reader/adapters/text_sources.py +++ b/smart_metadata_reader/adapters/text_sources.py @@ -28,7 +28,9 @@ class DeepTranslatorTextAdapter: role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: + del output_index return context.resolve_input(node, "text", role, path, depth) @@ -51,7 +53,9 @@ class GenericTextAdapter: role: str, path: list[str], depth: int, + output_index: int | None = None, ) -> list[PromptSegment]: + del output_index segments: list[PromptSegment] = [] for field in COMMON_TEXT_FIELDS: if field in node.inputs: diff --git a/smart_metadata_reader/graph.py b/smart_metadata_reader/graph.py index 90905b1..db67e3c 100644 --- a/smart_metadata_reader/graph.py +++ b/smart_metadata_reader/graph.py @@ -95,6 +95,23 @@ class GraphIndex: return value, "workflow widget cache fallback" return None + def workflow_output_name(self, node_id: str | int, output_index: int) -> str | None: + node = self._workflow_nodes.get(str(node_id)) + if node is None or not isinstance(node.outputs, list): + return None + if output_index < 0 or output_index >= len(node.outputs): + return None + output = node.outputs[output_index] + if isinstance(output, str): + return output + if not isinstance(output, dict): + return None + for field in ("name", "label", "display_name", "displayName"): + value = output.get(field) + if isinstance(value, str) and value: + return value + return None + def _normalize_prompt_nodes( self, prompt: dict[str, Any] | None, @@ -114,6 +131,7 @@ class GraphIndex: class_type=class_type, inputs=dict(inputs) if isinstance(inputs, dict) else {}, widgets_values=None, + outputs=raw_node.get("outputs"), ) return nodes @@ -143,6 +161,7 @@ class GraphIndex: class_type=class_type, inputs=dict(inputs) if isinstance(inputs, dict) else {}, widgets_values=raw_node.get("widgets_values"), + outputs=raw_node.get("outputs"), ) return nodes @@ -153,3 +172,5 @@ class GraphIndex: continue if prompt_node.widgets_values is None: prompt_node.widgets_values = workflow_node.widgets_values + if prompt_node.outputs is None: + prompt_node.outputs = workflow_node.outputs diff --git a/smart_metadata_reader/metadata_reader.py b/smart_metadata_reader/metadata_reader.py index 09540b2..b3fdaec 100644 --- a/smart_metadata_reader/metadata_reader.py +++ b/smart_metadata_reader/metadata_reader.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import os from pathlib import Path from typing import Any @@ -45,6 +46,17 @@ def _source_format( return "none" +def _initial_recursion_stack( + bundle: MetadataBundle, + recursion_stack: tuple[str, ...] | None, +) -> tuple[str, ...]: + if recursion_stack is not None: + return recursion_stack + if not bundle.image_path: + return () + return (os.path.normcase(str(Path(bundle.image_path).resolve())),) + + def read_metadata(image_path: str | Path) -> MetadataBundle: path = Path(image_path) with Image.open(path) as image: @@ -73,6 +85,8 @@ def read_metadata(image_path: str | Path) -> MetadataBundle: user_comment_raw=user_comment_raw, image_description_raw=image_description_raw, software_raw=software_raw, + image_path=str(path), + base_dir=str(path.parent), ) @@ -82,14 +96,18 @@ def parse_metadata_bundle( prefer_cached_text: bool = True, include_raw_json: bool = True, max_depth: int = 40, + _recursion_stack: tuple[str, ...] | None = None, ) -> ParseResult: + recursion_stack = _initial_recursion_stack(bundle, _recursion_stack) if bundle.prompt_raw: try: return _parse_comfyui_bundle( bundle=bundle, + parameter_index=parameter_index, prefer_cached_text=prefer_cached_text, include_raw_json=include_raw_json, max_depth=max_depth, + recursion_stack=recursion_stack, ) except _ComfyGraphStartError as exc: if bundle.parameters_raw: @@ -125,9 +143,11 @@ class _ComfyGraphStartError(Exception): def _parse_comfyui_bundle( bundle: MetadataBundle, + parameter_index: int, prefer_cached_text: bool, include_raw_json: bool, max_depth: int, + recursion_stack: tuple[str, ...], ) -> ParseResult: prompt = _load_required_json(bundle.prompt_raw, "prompt") workflow, workflow_error = _load_optional_json(bundle.workflow_raw, "workflow") @@ -143,6 +163,10 @@ def _parse_comfyui_bundle( graph=graph, max_depth=max_depth, prefer_cached_text=prefer_cached_text, + parameter_index=parameter_index, + include_raw_json=include_raw_json, + base_dir=bundle.base_dir, + recursion_stack=recursion_stack, ) positive_segments = resolver.resolve_positive(sampler) negative_segments = resolver.resolve_negative(sampler) diff --git a/smart_metadata_reader/models.py b/smart_metadata_reader/models.py index 2d0b21a..1639d85 100644 --- a/smart_metadata_reader/models.py +++ b/smart_metadata_reader/models.py @@ -16,6 +16,8 @@ class MetadataBundle: user_comment_raw: str | None = None image_description_raw: str | None = None software_raw: str | None = None + image_path: str | None = None + base_dir: str | None = None @dataclass @@ -24,6 +26,7 @@ class NodeRecord: class_type: str inputs: dict[str, Any] widgets_values: list[Any] | dict[str, Any] | None = None + outputs: list[Any] | None = None @dataclass diff --git a/smart_metadata_reader/resolver.py b/smart_metadata_reader/resolver.py index 675ebd3..6566015 100644 --- a/smart_metadata_reader/resolver.py +++ b/smart_metadata_reader/resolver.py @@ -6,6 +6,7 @@ from .adapters.base import AdapterRegistry, TextResolutionContext from .adapters.clip_text import CLIPTextEncodeAdapter, CLIPTextEncodeSDXLAdapter from .adapters.conditioning import ConditioningAdapter from .adapters.pysssss import ShowTextAdapter, StringFunctionAdapter +from .adapters.smart_reader import SmartMetadataReaderAdapter from .adapters.text_sources import DeepTranslatorTextAdapter, GenericTextAdapter from .graph import GraphIndex from .models import NodeRecord, PromptSegment @@ -19,10 +20,18 @@ class ConditioningResolver: adapters: AdapterRegistry | None = None, max_depth: int = 40, prefer_cached_text: bool = True, + parameter_index: int = 0, + include_raw_json: bool = True, + base_dir: str | None = None, + recursion_stack: tuple[str, ...] = (), ) -> None: self.graph = graph self.max_depth = max_depth self.prefer_cached_text = prefer_cached_text + self.parameter_index = parameter_index + self.include_raw_json = include_raw_json + self.base_dir = base_dir + self.recursion_stack = recursion_stack self.trace = TraceCollector() self.unresolved: list[dict[str, Any]] = [] self._active: set[tuple[str, str, str]] = set() @@ -80,6 +89,7 @@ class ConditioningResolver: path=[f"{sampler.class_type} {sampler.node_id}.{field}"], depth=0, field=field, + output_index=None, ) def _resolve_value( @@ -89,6 +99,7 @@ class ConditioningResolver: path: list[str], depth: int, field: str | None, + output_index: int | None = None, ) -> list[PromptSegment]: if depth > self.max_depth: self._record_unresolved_by_values( @@ -122,6 +133,7 @@ class ConditioningResolver: path=path + [f"link[{node_id}, {output_index}]"], depth=depth + 1, field=field or "", + output_index=output_index, ) return [] @@ -133,6 +145,7 @@ class ConditioningResolver: path: list[str], depth: int, field: str, + output_index: int | None = None, ) -> list[PromptSegment]: node = self.graph.get_node(node_id) if node is None: @@ -177,12 +190,19 @@ class ConditioningResolver: prefer_cached_text=self.prefer_cached_text, trace=self.trace, max_depth=self.max_depth - depth, + parse_max_depth=self.max_depth, + parameter_index=self.parameter_index, + include_raw_json=self.include_raw_json, + base_dir=self.base_dir, + recursion_stack=self.recursion_stack, + unresolved=self.unresolved, ) return text_context.resolve_node( node_id=node.node_id, role=role, path=path, depth=0, + output_index=output_index, ) if self._conditioning_adapter.matches(node): @@ -230,6 +250,7 @@ class ConditioningResolver: path=path + [f"{node.class_type} {node.node_id}.{field}"], depth=depth + 1, field=field, + output_index=None, ) ) return segments @@ -287,6 +308,7 @@ class ConditioningResolver: registry.register(CLIPTextEncodeSDXLAdapter()) registry.register(StringFunctionAdapter()) registry.register(ShowTextAdapter()) + registry.register(SmartMetadataReaderAdapter()) registry.register(DeepTranslatorTextAdapter()) registry.register(GenericTextAdapter()) return registry diff --git a/tests/test_metadata_reader.py b/tests/test_metadata_reader.py index 9fb3b9f..077f31d 100644 --- a/tests/test_metadata_reader.py +++ b/tests/test_metadata_reader.py @@ -14,6 +14,107 @@ def _save_jpeg_with_user_comment(path, comment: str, prefix: bytes = b"ASCII\x00 Image.new("RGB", (32, 24), (20, 40, 60)).save(path, exif=exif) +def _save_png_with_prompt(path, prompt, workflow=None): + pnginfo = PngInfo() + pnginfo.add_text("prompt", json.dumps(prompt)) + if workflow is not None: + pnginfo.add_text("workflow", json.dumps(workflow)) + Image.new("RGB", (32, 24), (20, 40, 60)).save(path, pnginfo=pnginfo) + + +def _simple_prompt_graph(positive="original positive prompt", negative="original negative prompt"): + return { + "1": {"class_type": "SaveImage", "inputs": {"images": ["2", 0]}}, + "2": {"class_type": "VAEDecode", "inputs": {"samples": ["3", 0]}}, + "3": { + "class_type": "KSampler", + "inputs": { + "positive": ["4", 0], + "negative": ["5", 0], + "seed": 111, + "steps": 12, + "cfg": 5.5, + "sampler_name": "euler", + "scheduler": "normal", + }, + }, + "4": {"class_type": "CLIPTextEncode", "inputs": {"text": positive}}, + "5": {"class_type": "CLIPTextEncode", "inputs": {"text": negative}}, + } + + +def _upscale_prompt_graph(reader_image="003.png"): + return { + "22": { + "class_type": "FaceDetailer", + "inputs": { + "image": ["24", 0], + "positive": ["26", 0], + "negative": ["27", 0], + }, + }, + "24": { + "class_type": "UltimateSDUpscale", + "inputs": { + "image": ["40", 0], + "positive": ["26", 0], + "negative": ["27", 0], + "seed": 222, + "steps": 9, + "cfg": 1.5, + "sampler_name": "Euler", + "scheduler": "normal", + "tile_width": 768, + "seam_fix_mode": "half tile", + }, + }, + "26": {"class_type": "CLIPTextEncode", "inputs": {"text": ["37", 0]}}, + "27": {"class_type": "CLIPTextEncode", "inputs": {"text": ["38", 0]}}, + "36": {"class_type": "SmartMetadataReader", "inputs": {"image": reader_image}}, + "37": {"class_type": "ShowText|pysssss", "inputs": {"text": ["36", 2]}}, + "38": {"class_type": "ShowText|pysssss", "inputs": {"text": ["36", 3]}}, + "40": {"class_type": "LoadImage", "inputs": {"image": "source.png"}}, + "90": { + "class_type": "GeminiChatNode", + "inputs": {"prompt": "unused Gemini prompt template"}, + }, + "91": {"class_type": "ShowText|pysssss", "inputs": {"text": ["90", 0]}}, + "99": {"class_type": "PreviewImage", "inputs": {"images": ["22", 0]}}, + } + + +def _upscale_workflow(reader_image="003.png"): + return { + "nodes": [ + { + "id": 36, + "type": "SmartMetadataReader", + "inputs": {"image": reader_image}, + "outputs": [ + {"name": "image"}, + {"name": "mask"}, + {"name": "positive"}, + {"name": "negative"}, + {"name": "seed"}, + {"name": "steps"}, + {"name": "cfg"}, + {"name": "width"}, + {"name": "height"}, + {"name": "model_name"}, + {"name": "filename"}, + {"name": "setting"}, + ], + }, + { + "id": 91, + "type": "ShowText|pysssss", + "inputs": {"text": ["90", 0]}, + "widgets_values": ["wrong unused prompt"], + }, + ] + } + + def test_read_metadata_extracts_png_text_chunks_and_dimensions(tmp_path): from smart_metadata_reader.metadata_reader import read_metadata @@ -373,3 +474,138 @@ def test_exif_user_comment_civitai_metadata_fallback(tmp_path): assert result.partial_result["source_format"] == "EXIF UserComment / Civitai metadata" assert result.status_message == "EXIF UserComment / Civitai metadata fallback" assert "NO_METADATA" not in result.setting + + +def test_nested_smart_metadata_reader_outputs_resolve_referenced_image_prompts(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + original_path = tmp_path / "003.png" + _save_png_with_prompt(original_path, _simple_prompt_graph()) + + upscale_path = tmp_path / "upscaled.png" + _save_png_with_prompt( + upscale_path, + _upscale_prompt_graph(), + workflow=_upscale_workflow(), + ) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "original positive prompt" + assert result.negative == "original negative prompt" + assert result.status_message != "FAILED" + assert not (result.status_message == "OK" and result.positive == "" and result.negative == "") + assert "Resolved SmartMetadataReader positive output from referenced image 003.png" in result.debug_trace + assert "Resolved SmartMetadataReader negative output from referenced image 003.png" in result.debug_trace + + +def test_unconnected_showtext_workflow_cache_does_not_pollute_nested_reader_prompt(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + original_path = tmp_path / "003.png" + _save_png_with_prompt(original_path, _simple_prompt_graph()) + upscale_path = tmp_path / "upscaled.png" + _save_png_with_prompt( + upscale_path, + _upscale_prompt_graph(), + workflow=_upscale_workflow(), + ) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "original positive prompt" + assert "wrong unused prompt" not in result.positive + assert "wrong unused prompt" not in result.negative + + +def test_nested_smart_metadata_reader_falls_back_to_current_output_contract_without_workflow_outputs(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + original_path = tmp_path / "003.png" + _save_png_with_prompt(original_path, _simple_prompt_graph()) + upscale_path = tmp_path / "upscaled.png" + _save_png_with_prompt(upscale_path, _upscale_prompt_graph()) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "original positive prompt" + assert result.negative == "original negative prompt" + + +def test_nested_smart_metadata_reader_prefers_workflow_output_names_over_indices(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + original_path = tmp_path / "003.png" + _save_png_with_prompt(original_path, _simple_prompt_graph()) + prompt = _upscale_prompt_graph() + prompt["37"]["inputs"]["text"] = ["36", 3] + prompt["38"]["inputs"]["text"] = ["36", 2] + workflow = _upscale_workflow() + workflow["nodes"][0]["outputs"][2] = {"name": "negative"} + workflow["nodes"][0]["outputs"][3] = {"name": "positive"} + upscale_path = tmp_path / "upscaled.png" + _save_png_with_prompt(upscale_path, prompt, workflow=workflow) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "original positive prompt" + assert result.negative == "original negative prompt" + + +def test_nested_smart_metadata_reader_missing_image_returns_partial_unresolved(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + upscale_path = tmp_path / "upscaled.png" + _save_png_with_prompt( + upscale_path, + _upscale_prompt_graph(reader_image="missing.png"), + workflow=_upscale_workflow(reader_image="missing.png"), + ) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "" + assert result.negative == "" + assert result.status_message == "PARTIAL" + assert result.confidence <= 0.6 + assert result.partial_result["unresolved"][0]["class_type"] == "SmartMetadataReader" + assert result.partial_result["unresolved"][0]["output_index"] == 2 + assert result.partial_result["unresolved"][0]["resolved_output_role"] == "positive" + assert result.partial_result["unresolved"][0]["referenced_image_filename"] == "missing.png" + assert "referenced image not found" in result.partial_result["unresolved"][0]["reason"] + assert "Unresolved:" in result.setting + + +def test_nested_smart_metadata_reader_rejects_unsafe_image_paths(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + for unsafe_path in ("../secret.png", "C:\\Users\\xxx\\secret.png"): + upscale_path = tmp_path / f"upscaled-{unsafe_path[0].encode().hex()}.png" + _save_png_with_prompt( + upscale_path, + _upscale_prompt_graph(reader_image=unsafe_path), + workflow=_upscale_workflow(reader_image=unsafe_path), + ) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "" + assert result.status_message == "PARTIAL" + assert result.partial_result["unresolved"][0]["class_type"] == "SmartMetadataReader" + assert "unsafe referenced image path" in result.partial_result["unresolved"][0]["reason"] + + +def test_nested_smart_metadata_reader_respects_actual_connected_output_role(tmp_path): + from smart_metadata_reader.metadata_reader import parse_metadata_bundle, read_metadata + + original_path = tmp_path / "003.png" + _save_png_with_prompt(original_path, _simple_prompt_graph()) + prompt = _upscale_prompt_graph() + prompt["37"]["inputs"]["text"] = ["36", 3] + upscale_path = tmp_path / "upscaled.png" + _save_png_with_prompt(upscale_path, prompt, workflow=_upscale_workflow()) + + result = parse_metadata_bundle(read_metadata(upscale_path)) + + assert result.positive == "original negative prompt" + assert "SmartMetadataReader output negative used in positive chain" in result.debug_trace diff --git a/tests/test_node_contract.py b/tests/test_node_contract.py index 02f3e64..80b1b55 100644 --- a/tests/test_node_contract.py +++ b/tests/test_node_contract.py @@ -366,6 +366,19 @@ def test_smart_metadata_reader_return_contract(): assert node_cls.CATEGORY +def test_smart_metadata_reader_text_output_indices_match_parser_mapping(): + import nodes + + node_cls = nodes.SmartMetadataReader + + assert node_cls.RETURN_NAMES[0] == "image" + assert node_cls.RETURN_NAMES[1] == "mask" + assert node_cls.RETURN_NAMES[2] == "positive" + assert node_cls.RETURN_NAMES[3] == "negative" + assert node_cls.RETURN_TYPES[2] == "STRING" + assert node_cls.RETURN_TYPES[3] == "STRING" + + def test_smart_metadata_reader_method_returns_expected_tuple_order(tmp_path, monkeypatch): import nodes