* 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.
This commit is contained in:
Vito
@@ -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
|
||||
|
||||
+19
-8
@@ -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()
|
||||
|
||||
+13
-21
@@ -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)})
|
||||
)
|
||||
|
||||
@@ -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()
|
||||
|
||||
+158
-1
@@ -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 "<lora:...>".
|
||||
g = flux_graph()
|
||||
g["50"] = {
|
||||
"class_type": "Lora Loader (LoraManager)",
|
||||
"inputs": {
|
||||
"model": ["1", 0],
|
||||
"clip": ["2", 0],
|
||||
"text": "<lora:style:1.00>",
|
||||
},
|
||||
}
|
||||
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": "<lora:style:1.00>"},
|
||||
}
|
||||
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": "<lora:style:1.00>"},
|
||||
}
|
||||
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 "<lora:...>" syntax.
|
||||
g = flux_graph(text=["50", 2])
|
||||
g["50"] = {
|
||||
"class_type": "Lora Loader (LoraManager)",
|
||||
"inputs": {"model": ["1", 0], "text": "<lora:style:1.00>"},
|
||||
}
|
||||
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": "<lora:style:1.00>"},
|
||||
}
|
||||
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"])
|
||||
|
||||
@@ -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": "<lora:x:1>"}},
|
||||
"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()
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user