From 44177974cb2ea12d4ea6473783f023e34ab474e9 Mon Sep 17 00:00:00 2001 From: Vito Date: Wed, 28 Oct 2026 00:32:30 +0000 Subject: [PATCH] fix(scan): read sampler-less graphs from the one wired text, retire the heuristic (#252) (#397) * fix(scan): read sampler-less graphs from the one wired text, retire the heuristic (#252) A graph where no node takes a positive or negative input fell back to the pre-redesign _extract_readable_prompt heuristic: the last node with any string text input won, and with none it read stale UI workflow widgets. It stored LoRA syntax, display-node reports, install notes, model file names and numbers. The scan now asks sampler_less_prompt: only the executed API graph counts; a candidate is a node with a text input that another node consumes; blank and numeric texts are never prompts; this repository's own positive nodes are authoritative; one distinct text links, several (or an unreadable wired text that no candidate builds) stay unlinked and try the recorded-run attribution. The heuristic and its helpers are removed; A1111 files read the parsed positive prompt directly. * fix(scan): refuse loader side outputs, seeded templates and keep A1111 text (#252) Review findings: an unreadable text link counts as built from a candidate only through that node's first output, so a LoRA loader's trigger-word output is unknown, not its text. A node with a seed input holds a template, never the prompt that ran. A sampler-less graph the rule cannot decide still links the A1111 positive a metadata saver wrote beside it, as the old scan did. Only plain numbers are numeric; words such as nan or inf are prompts. * fix(scan): linear plain-number check for sampler-less texts (#252) The numeric pattern had overlapping quantifiers that backtrack quadratically on a long digit run ending in a non-digit; it now has none. --- py/api/__init__.py | 193 ------------------------------ py/api/admin.py | 27 +++-- tests/test_api_admin.py | 34 ++---- tests/test_api_core.py | 92 -------------- tests/test_pair_scan.py | 159 +++++++++++++++++++++++- tests/test_sampler_less_prompt.py | 112 +++++++++++++++++ utils/prompt_pair_resolver.py | 112 +++++++++++++++++ 7 files changed, 414 insertions(+), 315 deletions(-) create mode 100644 tests/test_sampler_less_prompt.py diff --git a/py/api/__init__.py b/py/api/__init__.py index 485578c..def2186 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -1107,196 +1107,3 @@ class PromptManagerAPI( result["parameters"][field] = metadata[field] return result - - def _extract_readable_prompt(self, parsed_data): - """Extract human-readable prompt text from ComfyUI/A1111 data.""" - - def safe_to_string(value): - if isinstance(value, str): - return value - elif isinstance(value, list): - return " ".join(str(item) for item in value if item) - elif value is not None: - return str(value) - return None - - # First check if we already extracted a positive prompt (A1111 format) - if parsed_data.get("positive_prompt"): - return parsed_data["positive_prompt"] - - # Check if prompt is already a string - if isinstance(parsed_data.get("prompt"), str): - return parsed_data["prompt"] - - # Check if prompt is a simple value that can be converted - if parsed_data.get("prompt") and not isinstance( - parsed_data.get("prompt"), dict - ): - return safe_to_string(parsed_data["prompt"]) - - prompt_data = parsed_data.get("prompt") - if isinstance(prompt_data, dict): - # Use the enhanced logic from parse-metadata.py - positive_prompt = self._extract_positive_prompt_from_comfyui_data( - prompt_data - ) - if positive_prompt: - return positive_prompt - - # Check workflow data if available - workflow_data = parsed_data.get("workflow") - if isinstance(workflow_data, dict): - positive_prompt = self._extract_positive_prompt_from_comfyui_data( - workflow_data - ) - if positive_prompt: - return positive_prompt - - # Check parameters for positive prompt - if parsed_data.get("parameters", {}).get("positive"): - return safe_to_string(parsed_data["parameters"]["positive"]) - - return None - - def _get_node_inputs(self, node): - """Safely get inputs from a node, handling both dict and list formats. - - Old format: inputs is a dict with direct key-value pairs - inputs = {"text": "my prompt", "seed": 123} - - New format: inputs is a list of connection objects - inputs = [ - {"name": "text", "type": "STRING", "link": null, - "widget": {"name": "text"}}, - {"name": "clip", "type": "CLIP", "link": 11} - ] - """ - if not isinstance(node, dict): - return {} - - inputs = node.get("inputs", {}) - - # If inputs is already a dict, return it - if isinstance(inputs, dict): - return inputs - - # If inputs is a list, convert to dict format - if isinstance(inputs, list): - inputs_dict = {} - for input_item in inputs: - if isinstance(input_item, dict) and "name" in input_item: - name = input_item["name"] - inputs_dict[name] = input_item - return inputs_dict - - return {} - - def _find_text_in_node(self, node): - """Try to find text content in a node using various strategies. - - Handles both old and new workflow formats. - """ - if not isinstance(node, dict): - return None - - # Strategy 1: Check normalized inputs for 'text' field - inputs = self._get_node_inputs(node) - if "text" in inputs and isinstance(inputs["text"], str): - return inputs["text"] - - # Strategy 2: For text encoder nodes, check widgets_values - class_type = node.get("class_type", node.get("type", "")) - text_encoder_types = [ - "CLIPTextEncode", - "CLIPTextEncodeSDXL", - "CLIPTextEncodeSDXLRefiner", - "CLIPTextEncodeFlux", - "PromptManager", - "PromptManagerText", - "BNK_CLIPTextEncoder", - "Text Encoder", - "CLIP Text Encode", - ] - - if any( - encoder_type.lower() in class_type.lower() - for encoder_type in text_encoder_types - ): - widgets_values = node.get("widgets_values", []) - if widgets_values and len(widgets_values) > 0: - if isinstance(widgets_values[0], str) and widgets_values[0].strip(): - return widgets_values[0] - - return None - - def _extract_positive_prompt_from_comfyui_data(self, data): - """Extract positive prompt from ComfyUI data. - - Handles both old and new formats. - """ - if not isinstance(data, dict): - return None - - # Build nodes dictionary - nodes_by_id = {} - if "nodes" in data: - # Handle nodes array format - for node in data["nodes"]: - if isinstance(node, dict): - nid = node.get("id") - if nid is not None: - nodes_by_id[nid] = node - else: - # Handle flat dictionary format (node_id -> node_data) - for nid_str, node in data.items(): - try: - nid = int(nid_str) - except (ValueError, TypeError): - nid = nid_str - if isinstance(node, dict): - if "id" in node: - nid = node["id"] - nodes_by_id[nid] = node - - if not nodes_by_id: - return None - - # First, try to find positive/negative connection pattern - pos_id = None - for node in nodes_by_id.values(): - if isinstance(node, dict): - inputs = self._get_node_inputs(node) - if "positive" in inputs and "negative" in inputs: - try: - pos_input = inputs["positive"] - if isinstance(pos_input, list) and len(pos_input) > 0: - pos_id = int(pos_input[0]) - break - except (ValueError, TypeError, IndexError): - continue - - # Get text from the positive node - if pos_id is not None and pos_id in nodes_by_id: - text_val = self._find_text_in_node(nodes_by_id[pos_id]) - if text_val: - return text_val - - # Fallback: find any text encoder node with text content - text_nodes = [] - for node in nodes_by_id.values(): - if isinstance(node, dict): - text_val = self._find_text_in_node(node) - if text_val: - # Try to determine if this is positive or negative - node_title = node.get("title", "").lower() - if "neg" not in node_title and "negative" not in node_title: - # Prioritize non-negative prompts - text_nodes.insert(0, text_val) - else: - text_nodes.append(text_val) - - # Return the first positive-looking prompt - if text_nodes: - return text_nodes[0] - - return None diff --git a/py/api/admin.py b/py/api/admin.py index 4ebf8ee..ec3f6db 100644 --- a/py/api/admin.py +++ b/py/api/admin.py @@ -47,7 +47,7 @@ try: extract_generation_params, parse_page_json, ) - from ...utils.prompt_pair_resolver import resolve_prompt_pair + from ...utils.prompt_pair_resolver import resolve_prompt_pair, sampler_less_prompt from ...utils.review_capture import reviewable except ImportError: from database.pair_ingestion import blank_positive, ingest_resolved_pair @@ -62,7 +62,7 @@ except ImportError: extract_generation_params, parse_page_json, ) - from utils.prompt_pair_resolver import resolve_prompt_pair + from utils.prompt_pair_resolver import resolve_prompt_pair, sampler_less_prompt from utils.review_capture import reviewable @@ -1754,13 +1754,24 @@ class AdminRoutesMixin: canonical.get("parameters"), str ): return resolution["state"] + elif canonical.get("prompt"): + # No node takes a positive or negative input (a guider or wrapper + # graph): the one wired text, never a guess (#252). + found = sampler_less_prompt(canonical["prompt"]) + prompt_text = found["text"] + if prompt_text is None and isinstance(canonical.get("parameters"), str): + # A metadata saver also wrote A1111 text: its positive prompt. + parsed = self._parse_comfyui_prompt(metadata)["positive_prompt"] + prompt_text = parsed.strip() if parsed and parsed.strip() else None + if prompt_text is None: + linked = self._link_recorded_run( + media_file, canonical, detailed, history + ) + if linked is not None: + return linked + return "unresolved" if found["state"] == "ambiguous" else "found" else: - parsed = self._parse_comfyui_prompt(metadata) - if not (parsed.get("prompt") or parsed.get("parameters")): - return None - prompt_text = self._extract_readable_prompt(parsed) - if prompt_text is not None and not isinstance(prompt_text, str): - prompt_text = str(prompt_text) + prompt_text = self._parse_comfyui_prompt(metadata)["positive_prompt"] if not (prompt_text and prompt_text.strip()): return "found" prompt_text = prompt_text.strip() diff --git a/tests/test_api_admin.py b/tests/test_api_admin.py index a5b85ba..120c1e9 100644 --- a/tests/test_api_admin.py +++ b/tests/test_api_admin.py @@ -846,6 +846,14 @@ class TestSettingsReadMonitorThroughModule(AdminAPITestCase): self.assertEqual(data["settings"]["gallery_root_paths"], ["output"]) +def _wired_prompt(text): + """A sampler-less graph whose one encoder feeds a guider (#252).""" + return { + "1": {"class_type": "CLIPTextEncode", "inputs": {"text": text}}, + "2": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}}, + } + + def _raise(*args, **kwargs): raise RuntimeError("boom") @@ -2186,11 +2194,7 @@ class TestOutputScanInternals(AdminAPITestCase): def test_ingest_outcomes(self): f = Path("gen.png") - meta = { - "prompt": json.dumps( - {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}}} - ) - } + meta = {"prompt": json.dumps(_wired_prompt("a cat"))} self.assertIsNone(self.api._ingest_scanned_file(f, {})) self.assertIsNone(self.api._ingest_scanned_file(f, {"prompt": "{}"})) self.assertEqual( @@ -2202,31 +2206,19 @@ class TestOutputScanInternals(AdminAPITestCase): def test_ingest_link_failures_are_tolerated(self): f = Path("gen.png") - meta = { - "prompt": json.dumps( - {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a dog"}}} - ) - } + meta = {"prompt": json.dumps(_wired_prompt("a dog"))} self.api.db.link_image_to_prompt = _raise self.assertEqual(self.api._ingest_scanned_file(f, meta), "added") self.assertEqual(self.api._ingest_scanned_file(f, meta), "found") def test_ingest_save_failure_counts_as_found(self): f = Path("gen.png") - meta = { - "prompt": json.dumps( - {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a bird"}}} - ) - } + meta = {"prompt": json.dumps(_wired_prompt("a bird"))} self.api.db.save_prompt = lambda *a, **k: None self.assertEqual(self.api._ingest_scanned_file(f, meta), "found") def test_batch_counts_and_per_file_errors(self): - meta = { - "prompt": json.dumps( - {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a fish"}}} - ) - } + meta = {"prompt": json.dumps(_wired_prompt("a fish"))} self.api.db.get_prompt_by_hash = _raise counts = self.api._ingest_scan_batch_sync( [(Path("a.png"), meta), (Path("b.png"), {})] @@ -2915,7 +2907,7 @@ class TestScanReadsVideos(AdminAPITestCase): shutil.which("ffmpeg") and shutil.which("ffprobe"), "ffmpeg not installed" ) async def test_scan_adds_prompts_found_in_videos(self): - graph = {"1": {"class_type": "CLIPTextEncode", "inputs": {"text": "a cat"}}} + graph = _wired_prompt("a cat") _write_mp4( self.output_dir / "clip.mp4", json.dumps({"prompt": json.dumps(graph)}) ) diff --git a/tests/test_api_core.py b/tests/test_api_core.py index 07e6ffa..bf46c32 100644 --- a/tests/test_api_core.py +++ b/tests/test_api_core.py @@ -835,7 +835,6 @@ class TestMetadataParsing(unittest.TestCase): self.assertEqual(parsed["positive_prompt"], "a cat") self.assertEqual(parsed["negative_prompt"], "dog") self.assertEqual(parsed["parameters"]["parameters"], meta["parameters"]) - self.assertEqual(self.api._extract_readable_prompt(parsed), "a cat") def test_parse_comfyui_fields(self): meta = { @@ -856,97 +855,6 @@ class TestMetadataParsing(unittest.TestCase): self.assertEqual(parsed["prompt"], "plain text") self.assertEqual(parsed["workflow"], {"nodes": []}) - def test_extract_readable_prompt_variants(self): - extract = self.api._extract_readable_prompt - self.assertEqual(extract({"prompt": "plain"}), "plain") - self.assertEqual(extract({"prompt": ["a", "", "b"]}), "a b") - self.assertEqual(extract({"prompt": 42}), "42") - self.assertEqual( - extract({"prompt": None, "parameters": {"positive": ["x", "y"]}}), "x y" - ) - self.assertIsNone(extract({"prompt": {}, "parameters": {}})) - - graph = { - "1": {"class_type": "CLIPTextEncode", "inputs": {"text": "pos"}}, - "2": {"class_type": "CLIPTextEncode", "inputs": {"text": "neg"}}, - "3": { - "class_type": "KSampler", - "inputs": {"positive": ["1", 0], "negative": ["2", 0]}, - }, - } - self.assertEqual(extract({"prompt": graph}), "pos") - self.assertEqual(extract({"prompt": {"x": 1}, "workflow": graph}), "pos") - - def test_node_inputs_and_text_lookup(self): - self.assertEqual(self.api._get_node_inputs("nope"), {}) - self.assertEqual(self.api._get_node_inputs({"inputs": 5}), {}) - self.assertEqual(self.api._get_node_inputs({"inputs": {"a": 1}}), {"a": 1}) - listed = self.api._get_node_inputs( - {"inputs": [{"name": "text", "link": None}, {"nope": 1}, "junk"]} - ) - self.assertEqual(list(listed), ["text"]) - - self.assertIsNone(self.api._find_text_in_node(None)) - self.assertEqual(self.api._find_text_in_node({"inputs": {"text": "hi"}}), "hi") - self.assertEqual( - self.api._find_text_in_node( - {"type": "CLIPTextEncode", "widgets_values": ["w"]} - ), - "w", - ) - self.assertIsNone( - self.api._find_text_in_node( - {"type": "CLIPTextEncode", "widgets_values": [" "]} - ) - ) - self.assertIsNone(self.api._find_text_in_node({"type": "KSampler"})) - - def test_positive_prompt_from_nodes_array_and_fallbacks(self): - extract = self.api._extract_positive_prompt_from_comfyui_data - self.assertIsNone(extract("x")) - self.assertIsNone(extract({})) - self.assertIsNone(extract({"nodes": []})) - - workflow = { - "nodes": [ - {"id": 1, "type": "CLIPTextEncode", "widgets_values": ["pos"]}, - { - "id": 2, - "type": "CLIPTextEncode", - "widgets_values": ["neg"], - "title": "Negative", - }, - {"id": 3, "type": "KSampler", "inputs": [{"name": "seed", "link": 4}]}, - "junk", - ] - } - self.assertEqual(extract(workflow), "pos") - - only_negative = { - "nodes": [ - { - "id": 2, - "type": "CLIPTextEncode", - "widgets_values": ["neg"], - "title": "neg", - }, - ] - } - self.assertEqual(extract(only_negative), "neg") - - bad_link = { - "1": {"class_type": "CLIPTextEncode", "inputs": {"text": "pos"}}, - "3": { - "class_type": "KSampler", - "inputs": {"positive": "oops", "negative": []}, - }, - "4": { - "class_type": "KSampler", - "inputs": {"positive": ["zz", 0], "negative": []}, - }, - } - self.assertEqual(extract(bad_link), "pos") - if __name__ == "__main__": unittest.main() diff --git a/tests/test_pair_scan.py b/tests/test_pair_scan.py index 1597b1f..2ee7015 100644 --- a/tests/test_pair_scan.py +++ b/tests/test_pair_scan.py @@ -115,7 +115,12 @@ class TestPairScan(AdminAPITestCase): self.assertEqual(second["pairs_added"], 0) def test_legacy_single_prompt_second_pass_not_reported_as_new_link(self): - meta = {"prompt": json.dumps({"1": graph()["1"]})} + # A sampler-less graph: one encoder wired into a guider (#252). + single = { + "1": graph()["1"], + "2": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}}, + } + meta = {"prompt": json.dumps(single)} path = self.output_dir / "old.png" self.assertEqual(self.api._ingest_scanned_file(path, meta), "added") self.assertEqual(self.api._ingest_scanned_file(path, meta), "unchanged") @@ -292,3 +297,155 @@ class TestPairScan(AdminAPITestCase): self.api.db.get_image_prompt_info(str(path))["workflow_data"], workflow ) self.assertEqual(self.api._ingest_scanned_file(path, metadata), "unchanged") + + +def flux_graph(text="a red fox", encoder="CLIPTextEncode"): + """A sampler-less Flux graph: the prompt reaches a guider's conditioning (#252).""" + return { + "1": {"class_type": "UNETLoader", "inputs": {"unet_name": "flux.safetensors"}}, + "2": {"class_type": "DualCLIPLoader", "inputs": {"clip_name1": "t5"}}, + "6": {"class_type": encoder, "inputs": {"text": text, "clip": ["2", 0]}}, + "22": { + "class_type": "BasicGuider", + "inputs": {"model": ["1", 0], "conditioning": ["6", 0]}, + }, + "13": { + "class_type": "SamplerCustomAdvanced", + "inputs": {"guider": ["22", 0], "sigmas": ["17", 0]}, + }, + "17": {"class_type": "BasicScheduler", "inputs": {"steps": 20}}, + "9": {"class_type": "SaveImage", "inputs": {"images": ["13", 0]}}, + } + + +class TestSamplerLessScan(AdminAPITestCase): + """Graphs where no node takes a positive or negative input (#252).""" + + def _scanned(self, name, prompt_graph, workflow=None): + meta = {"prompt": json.dumps(prompt_graph)} + if workflow is not None: + meta["workflow"] = json.dumps(workflow) + return self.api._ingest_scanned_file(self.output_dir / name, meta) + + def _texts(self): + return [p["text"] for p in self.api.db.get_recent_prompts()["prompts"]] + + def test_the_one_wired_prompt_is_linked(self): + self.assertEqual(self._scanned("flux.png", flux_graph()), "added") + self.assertEqual(self._texts(), ["a red fox"]) + + def test_a_text_node_feeding_the_encoder_is_one_prompt(self): + g = flux_graph(text=["30", 0]) + g["30"] = {"class_type": "TextInput", "inputs": {"text": "a red fox"}} + self.assertEqual(self._scanned("textinput.png", g), "added") + self.assertEqual(self._texts(), ["a red fox"]) + + def test_a_display_node_that_feeds_nothing_is_never_the_prompt(self): + # A real scan stored a "=== GIT CLONE REPORT ===" display text this way. + g = flux_graph() + g["40"] = { + "class_type": "easy showAnything", + "inputs": {"text": "=== GIT CLONE REPORT ===\nTotal repositories: 1"}, + } + self.assertEqual(self._scanned("report.png", g), "added") + self.assertEqual(self._texts(), ["a red fox"]) + + def test_only_unconsumed_texts_store_nothing(self): + g = { + "40": {"class_type": "easy showAnything", "inputs": {"text": "report"}}, + "41": {"class_type": "Textbox", "inputs": {"text": "install note"}}, + } + self.assertEqual(self._scanned("notes.png", g), "found") + self.assertEqual(self._texts(), []) + + def test_two_wired_texts_are_ambiguous_and_stay_unlinked(self): + # A LoRA loader's text input is wired too: the scan stored "". + g = flux_graph() + g["50"] = { + "class_type": "Lora Loader (LoraManager)", + "inputs": { + "model": ["1", 0], + "clip": ["2", 0], + "text": "", + }, + } + g["22"]["inputs"]["model"] = ["50", 0] + g["6"]["inputs"]["clip"] = ["50", 1] + self.assertEqual(self._scanned("lora.png", g), "unresolved") + self.assertEqual(self._texts(), []) + + def test_an_untraceable_wired_text_makes_the_graph_ambiguous(self): + # The encoder's text comes from a node no rule reads: never guess the other. + g = flux_graph(text=["60", 0]) + g["60"] = {"class_type": "SomeLLMNode", "inputs": {"seed": 1}} + g["50"] = { + "class_type": "Lora Loader (LoraManager)", + "inputs": {"model": ["1", 0], "text": ""}, + } + g["22"]["inputs"]["model"] = ["50", 0] + self.assertEqual(self._scanned("llm.png", g), "unresolved") + self.assertEqual(self._texts(), []) + + def test_this_repositorys_prompt_node_is_authoritative(self): + g = flux_graph(text="a red fox", encoder="PromptManager") + g["50"] = { + "class_type": "Lora Loader (LoraManager)", + "inputs": {"model": ["1", 0], "text": ""}, + } + g["22"]["inputs"]["model"] = ["50", 0] + self.assertEqual(self._scanned("own.png", g), "added") + self.assertEqual(self._texts(), ["a red fox"]) + + def test_a_numeric_text_is_never_stored_as_a_prompt(self): + self.assertEqual(self._scanned("number.png", flux_graph(text="2048")), "found") + self.assertEqual(self._texts(), []) + + def test_workflow_widget_values_are_never_read(self): + # The API text is linked from a node no rule reads; the UI workflow + # still holds a stale widget value, which the old scan stored. + g = flux_graph(text=["70", 0]) + g["70"] = {"class_type": "PromptBuilder", "inputs": {"prompt": "x", "n": 2}} + workflow = { + "nodes": [ + {"id": 6, "type": "CLIPTextEncode", "widgets_values": ["stale text"]} + ] + } + self.assertEqual(self._scanned("stale.png", g, workflow), "unresolved") + self.assertEqual(self._texts(), []) + + def test_a_lora_loaders_trigger_word_output_is_not_its_text(self): + # The encoder reads the loader's trigger-word output (slot 2), not its + # text input: the prompt is unknown, never the "" syntax. + g = flux_graph(text=["50", 2]) + g["50"] = { + "class_type": "Lora Loader (LoraManager)", + "inputs": {"model": ["1", 0], "text": ""}, + } + g["22"]["inputs"]["model"] = ["50", 0] + self.assertEqual(self._scanned("trigger.png", g), "unresolved") + self.assertEqual(self._texts(), []) + + def test_a_seeded_text_node_holds_a_template_not_the_prompt(self): + g = flux_graph(text=["80", 0]) + g["80"] = { + "class_type": "RandomPrompt", + "inputs": {"text": "a {red|blue} fox", "seed": 5}, + } + self.assertEqual(self._scanned("random.png", g), "unresolved") + self.assertEqual(self._texts(), []) + + def test_a1111_parameters_beside_an_undecided_graph_are_read(self): + # Metadata savers write both; the old scan read the A1111 positive. + g = flux_graph() + g["50"] = { + "class_type": "Lora Loader (LoraManager)", + "inputs": {"model": ["1", 0], "text": ""}, + } + g["22"]["inputs"]["model"] = ["50", 0] + meta = { + "prompt": json.dumps(g), + "parameters": "a red fox\nNegative prompt: blur\nSteps: 20", + } + path = self.output_dir / "saver.png" + self.assertEqual(self.api._ingest_scanned_file(path, meta), "added") + self.assertEqual(self._texts(), ["a red fox"]) diff --git a/tests/test_sampler_less_prompt.py b/tests/test_sampler_less_prompt.py new file mode 100644 index 0000000..62ee6fb --- /dev/null +++ b/tests/test_sampler_less_prompt.py @@ -0,0 +1,112 @@ +"""``sampler_less_prompt``: the one wired text of a graph with no sampler (#252).""" + +import time +import unittest + +from utils.prompt_pair_resolver import sampler_less_prompt + + +def _guided(**nodes): + """``nodes`` plus a guider consuming node ``"1"``.""" + graph = {"9": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}}} + graph.update(nodes) + return graph + + +class TestSamplerLessPrompt(unittest.TestCase): + def test_not_a_graph_or_too_large_is_absent(self): + for graph in (None, [], {}, "text"): + self.assertEqual(sampler_less_prompt(graph)["state"], "absent") + big = {str(i): {"class_type": "X", "inputs": {}} for i in range(3)} + self.assertEqual(sampler_less_prompt(big, max_visits=2)["state"], "absent") + + def test_one_literal_text(self): + graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": " a fox "}}}) + self.assertEqual( + sampler_less_prompt(graph), + {"state": "resolved", "text": "a fox", "texts": ["a fox"]}, + ) + + def test_a_number_literal_is_never_a_prompt(self): + for value in (2048, "2048", "1.5", " 7 "): + graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": value}}}) + self.assertEqual(sampler_less_prompt(graph)["state"], "absent", value) + + def test_a_long_digit_run_with_a_letter_is_text_and_fast(self): + text = "9" * 50_000 + "x" + graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": text}}}) + started = time.monotonic() + self.assertEqual(sampler_less_prompt(graph)["text"], text) + self.assertLess(time.monotonic() - started, 1.0) + + def test_number_like_words_are_prompts(self): + for value in ("nan", "inf", "Infinity", "1_000"): + graph = _guided(**{"1": {"class_type": "Enc", "inputs": {"text": value}}}) + self.assertEqual(sampler_less_prompt(graph)["text"], value) + + def test_an_own_node_with_unreadable_text_beside_another_text_is_ambiguous(self): + graph = _guided( + **{ + "1": {"class_type": "PromptManager", "inputs": {"text": ["5", 0]}}, + "5": {"class_type": "LLM", "inputs": {"seed": 3}}, + "2": {"class_type": "Enc", "inputs": {"text": ""}}, + "3": {"class_type": "Sink", "inputs": {"c": ["2", 0]}}, + } + ) + self.assertEqual(sampler_less_prompt(graph)["state"], "ambiguous") + + def test_unhashable_class_type_is_an_unknown_node_not_a_crash(self): + graph = _guided(**{"1": {"class_type": ["x"], "inputs": {"text": "a fox"}}}) + self.assertEqual(sampler_less_prompt(graph)["text"], "a fox") + + def test_own_negative_node_is_not_authoritative(self): + graph = _guided( + **{ + "1": {"class_type": "Enc", "inputs": {"text": "a fox"}}, + "2": { + "class_type": "PromptManagerNegative", + "inputs": {"text": "blur"}, + }, + "3": {"class_type": "Sink", "inputs": {"c": ["2", 0]}}, + } + ) + found = sampler_less_prompt(graph) + self.assertEqual(found["state"], "ambiguous") + self.assertEqual(found["texts"], ["a fox", "blur"]) + + def test_own_node_text_wins_over_an_untraceable_text(self): + graph = _guided( + **{ + "1": {"class_type": "PromptManagerText", "inputs": {"text": "a fox"}}, + "2": {"class_type": "Enc", "inputs": {"text": ["5", 0]}}, + "5": {"class_type": "LLM", "inputs": {"seed": 3}}, + "6": {"class_type": "Sink", "inputs": {"c": ["2", 0]}}, + } + ) + self.assertEqual(sampler_less_prompt(graph)["text"], "a fox") + + def test_two_own_texts_are_ambiguous(self): + graph = _guided( + **{ + "1": {"class_type": "PromptManager", "inputs": {"text": "a fox"}}, + "2": {"class_type": "PromptManager", "inputs": {"text": "a cat"}}, + "3": {"class_type": "Sink", "inputs": {"c": ["2", 0]}}, + } + ) + self.assertEqual(sampler_less_prompt(graph)["state"], "ambiguous") + + def test_only_an_untraceable_text_is_ambiguous_with_nothing_to_show(self): + graph = _guided( + **{ + "1": {"class_type": "Enc", "inputs": {"text": ["5", 0]}}, + "5": {"class_type": "LLM", "inputs": {"seed": 3}}, + } + ) + self.assertEqual( + sampler_less_prompt(graph), + {"state": "ambiguous", "text": None, "texts": []}, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index b45c948..1520048 100644 --- a/utils/prompt_pair_resolver.py +++ b/utils/prompt_pair_resolver.py @@ -2578,3 +2578,115 @@ def resolve_text_input( return resolver.text(inputs.get(input_name), "string")[0] except _Unresolved: return None + + +_NUMBER = re.compile(r"[+-]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][+-]?\d+)?") + + +def _numeric(text: str) -> bool: + """Whether ``text`` is a plain number (a widget value such as ``2048``).""" + return _NUMBER.fullmatch(text) is not None + + +def sampler_less_prompt(graph: Any, *, max_visits: int = 500) -> Dict[str, Any]: + """The positive prompt of an API graph with no sampler pair, if one (#252). + + For graphs ``_no_sampler`` finds ``absent`` (no node takes a ``positive`` + or ``negative`` input, such as a Flux guider or a video wrapper). Only the + executed API graph is read, never the UI workflow. A candidate is a node + with a ``text`` input whose output another node consumes: a text that + feeds nothing did not make the image. Its text is traced as a STRING + (``resolve_text_input``); a linked text no rule reads is unknown unless + it is built from a candidate's text (``_covered``). Blank and + numeric texts are never prompts, and a node with a seed input holds a + template, not the prompt (unknown). This repository's own positive-text + nodes are authoritative: when they give a text, only they count. + + Returns ``{"state", "text", "texts"}``: ``resolved`` with the one distinct + text; ``ambiguous`` when several texts or an unknown one could be the + prompt (``texts`` lists the readable ones); ``absent`` when none. + """ + absent = {"state": "absent", "text": None, "texts": []} + if not isinstance(graph, dict) or not graph or len(graph) > max_visits: + return absent + nodes = {str(key): value for key, value in graph.items()} + consumed = { + str(value[0]) + for node in nodes.values() + if isinstance(node, dict) and isinstance(node.get("inputs"), dict) + for value in node["inputs"].values() + if _link(value) + } + own: list = [] + other: list = [] + unknown = False + for node_id, node in nodes.items(): + inputs = node.get("inputs") if isinstance(node, dict) else None + if node_id not in consumed or not _has_text_input(node): + continue + if _seeded(inputs): + # Generated at run time (a wildcard or a language model): its + # text input is a template, not the prompt that ran. + unknown = True + continue + value = inputs["text"] + text = resolve_text_input(nodes, node_id, "text", max_visits=max_visits) + if text is None: + # A literal that is not a string (a number) is never a prompt. + if _link(value) and not _covered(nodes, value): + unknown = True + continue + text = text.strip() + if not text or _numeric(text): + continue + kind = node.get("class_type") + if isinstance(kind, str) and kind in _OWN_CLASSES and "Negative" not in kind: + own.append(text) + else: + other.append(text) + texts = list(dict.fromkeys(own or other)) + if len(texts) == 1 and (own or not unknown): + return {"state": "resolved", "text": texts[0], "texts": texts} + if texts or unknown: + return {"state": "ambiguous", "text": None, "texts": texts} + return absent + + +def _has_text_input(node: Any) -> bool: + inputs = node.get("inputs") if isinstance(node, dict) else None + return isinstance(inputs, dict) and "text" in inputs + + +def _seeded(inputs: dict) -> bool: + """A seed input: the node's string output is a run-time value (the + resolver's ``runtime-value`` rule).""" + return any(isinstance(name, str) and name.endswith("seed") for name in inputs) + + +def _covered(nodes: dict, link: list) -> bool: + """Whether an unreadable text link is built from a candidate's text. + + True when a node upstream of ``link`` is read through its first output + (slot 0, the legacy encoder rule's text) and has an unseeded ``text`` + input: that candidate is read on its own, and whatever else is joined to + it at run time (a caption, a language model) is no stored prompt to + choose. Another output of such a node (a loader's trigger words) is not + its text. False when the text comes only from inputs no candidate holds + (a builder's ``prompt`` field): the prompt is unknown then. + """ + seen, stack = set(), [(str(link[0]), link[1])] + while stack: + node_id, slot = stack.pop() + if (node_id, slot) in seen: + continue + seen.add((node_id, slot)) + node = nodes.get(node_id) + inputs = node.get("inputs") if isinstance(node, dict) else None + if not isinstance(inputs, dict): + continue + if "text" in inputs and slot == 0 and not _seeded(inputs): + return True + stack.extend( + (str(value[0]), value[1]) for value in inputs.values() if _link(value) + ) + return False