From 9bf8362d2226bfb2aba8171e50720072b9c25063 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 28 Oct 2026 06:09:38 -0700 Subject: [PATCH] feat(ingestion): one per-file prompt decision for the scan and the monitor utils/file_prompt.py holds decide_file_prompt, the ladder the output scan runs today plus the inspector's rule (inspection profile, then the recorded saver) before the legacy parser, with a SCAN and a MONITOR policy. Junk text (blank, numbers, LoRA tags only, model file names) is never stored. graph_prompt_text and parse_comfyui_prompt move here; the parser accepts parsed metadata (#247). --- tests/test_file_prompt.py | 379 ++++++++++++++++++++++++++++++++++++ utils/file_prompt.py | 398 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 777 insertions(+) create mode 100644 tests/test_file_prompt.py create mode 100644 utils/file_prompt.py diff --git a/tests/test_file_prompt.py b/tests/test_file_prompt.py new file mode 100644 index 00000000..9ae150b1 --- /dev/null +++ b/tests/test_file_prompt.py @@ -0,0 +1,379 @@ +"""The per-file prompt decision shared by the scan and the monitor (#247). + +One test per case, named by the ladder step of the design +(``docs/egm/247-foreign-workflows``). The ladder does no file or database +I/O: history comes in through ``lookup``. +""" + +import json +import unittest +from unittest import mock + +from database.run_attribution import RecordedOutput +from tests.test_pair_scan import flux_graph, sampled_graph +from tests.test_prompt_attribution import issue_graph, pair_evidence +from utils import file_prompt +from utils.file_prompt import Policy, decide_file_prompt, junk_text +from utils.prompt_inspector import InspectionResult + +try: + from tests import inspection_parity as parity + from tests import saver_anchor_graphs as shapes +except ImportError: # pragma: no cover - discover -s tests puts tests/ first + import inspection_parity as parity + import saver_anchor_graphs as shapes + +SHARED_NAME = "IMG_GEN_00371_.png" +LORA = "" + + +def _decide( + metadata, + *, + policy=Policy.SCAN, + lookup=None, + in_output=True, + file_name=SHARED_NAME, + file_size=None, +): + return decide_file_prompt( + metadata, + file_name=file_name, + file_size=file_size, + in_output=in_output, + lookup=lookup, + policy=policy, + ) + + +def _graph(graph): + return {"prompt": json.dumps(graph)} + + +class _Lookup: + """A history lookup that counts its calls.""" + + def __init__(self, saver=None): + self.calls = 0 + self.found = RecordedOutput(saver, {}) if saver is not None else None + + def __call__(self): + self.calls += 1 + return self.found + + +def _inspected(state, malformed=False): + resolution = {"state": state, "positive": None, "negative": None} + return InspectionResult(resolution, malformed=malformed) + + +def _untraced_by_both(): + """A sampled graph whose positive no profile traces; the legacy parser does.""" + return mock.patch.object( + file_prompt, "inspection_resolution", return_value=_inspected("unresolved") + ) + + +def _gated(graph): + """``graph`` past the default profile's node gate, so it is unresolved.""" + graph = dict(graph) + for index in range(510): + graph[f"pad{index}"] = {"class_type": "Note", "inputs": {}} + return graph + + +class TestStepsBeforeResolving(unittest.TestCase): + def test_step_1_no_metadata_tries_the_recorded_run(self): + for metadata in ({}, None): + decision = _decide(metadata) + self.assertEqual(decision.kind, "never") + self.assertTrue(decision.try_recorded_run) + self.assertIsNone(decision.unlinked_status) + self.assertEqual(decision.canonical, {}) + + def test_step_3_neither_prompt_nor_parameters_tries_the_recorded_run(self): + decision = _decide({"workflow": json.dumps({"nodes": []})}) + self.assertEqual(decision.kind, "never") + self.assertTrue(decision.try_recorded_run) + self.assertIsNone(decision.unlinked_status) + self.assertEqual(decision.canonical, {"workflow": {"nodes": []}}) + + def test_strings_are_parsed_and_malformed_ones_kept(self): + decision = _decide({"prompt": "{not json", "workflow": "{}"}) + self.assertEqual(decision.canonical["prompt"], "{not json") + self.assertEqual(decision.canonical["workflow"], {}) + + def test_the_file_size_is_recorded_unless_the_metadata_has_one(self): + meta = _graph(sampled_graph()) + self.assertEqual( + _decide(meta, file_size=42).canonical["file_info"], {"size": 42} + ) + meta["file_info"] = {"size": 7, "width": 3} + self.assertEqual( + _decide(meta, file_size=42).canonical["file_info"], {"size": 7, "width": 3} + ) + self.assertNotIn("file_info", _decide(_graph(sampled_graph())).canonical) + + +class TestResolvedSteps(unittest.TestCase): + def test_step_6_a_blank_positive_counts_as_unresolved(self): + meta = { + "prompt": json.dumps(issue_graph()), + "comfyassets_prompt_pairs": json.dumps(pair_evidence()), + } + decision = _decide(meta, policy=Policy.MONITOR, in_output=False) + self.assertEqual(decision.kind, "never") + self.assertEqual(decision.resolution["state"], "unresolved") + + def test_step_7_a_resolved_pair(self): + lookup = _Lookup("9") + decision = _decide(_graph(sampled_graph("cat", "blur")), lookup=lookup) + self.assertEqual(decision.kind, "pair") + self.assertEqual(decision.source, "resolver") + self.assertEqual(decision.text, "cat") + self.assertEqual(decision.resolution["negative"]["text"], "blur") + self.assertIsNone(decision.review) + self.assertEqual(lookup.calls, 0, "decided before step 8: history not asked") + + def test_step_7_a_blank_negative_stays_a_pair(self): + decision = _decide(_graph(sampled_graph("cat", ""))) + self.assertEqual(decision.kind, "pair") + self.assertEqual(decision.resolution["negative"]["text"], "") + + +class TestUndecidedGraph(unittest.TestCase): + def test_step_8_1_the_inspector_alone_decides(self): + decision = _decide({"prompt": parity.build("f03")["prompt"]}) + self.assertEqual(decision.kind, "positive") + self.assertEqual(decision.source, "inspector") + self.assertEqual(decision.text, parity.POS) + self.assertEqual(decision.resolution["state"], "unresolved") + + def test_step_8_1_the_recorded_saver_names_its_branch(self): + lookup = _Lookup("23") + decision = _decide(_graph(shapes.shared_prefix()), lookup=lookup) + self.assertEqual(decision.kind, "positive") + self.assertEqual(decision.source, "recorded-run") + self.assertEqual(decision.text, shapes.BRANCH_TEXTS[1]) + self.assertEqual(decision.review, "ambiguous") + self.assertEqual(lookup.calls, 1) + + def test_step_8_1_stores_the_storage_text_without_prepend_or_append(self): + graph = shapes.shared_prefix() + graph["20"]["inputs"]["text"] = ["25", 1] + graph["25"] = { + "class_type": "PromptManagerSplit", + "inputs": { + "text": shapes.BRANCH_TEXTS[1], + "prepend_text": "trigger", + "append_text": "detail", + }, + } + decision = _decide(_graph(graph), lookup=_Lookup("23")) + self.assertEqual(decision.source, "recorded-run") + self.assertEqual(decision.text, shapes.BRANCH_TEXTS[1]) + + def test_step_8_1_a_file_name_naming_another_saver_anchors_nothing(self): + graph = shapes.branch_bench() + for index in (1, 3, 4): + graph[str(index * 10 + 3)]["inputs"]["filename_prefix"] = "bench/shared" + decision = _decide( + _graph(graph), lookup=_Lookup("43"), file_name="s45_branch_2_00001_.png" + ) + # The file name alone names branch 2; history's branch 4 is not used. + self.assertEqual(decision.source, "inspector") + self.assertEqual(decision.text, shapes.BRANCH_TEXTS[1]) + + def test_step_8_2_the_monitor_waits_for_history_inside_the_output_folder(self): + decision = _decide(_graph(shapes.shared_prefix()), policy=Policy.MONITOR) + self.assertEqual(decision.kind, "history") + + def test_step_8_2_outside_the_output_folder_steps_8_3_and_8_4_apply(self): + decision = _decide( + _graph(shapes.shared_prefix()), policy=Policy.MONITOR, in_output=False + ) + self.assertEqual(decision.kind, "never") + with _untraced_by_both(): + decision = _decide( + _graph(_gated(sampled_graph())), policy=Policy.MONITOR, in_output=False + ) + self.assertEqual( + (decision.kind, decision.source), ("positive", "legacy-parser") + ) + + def test_step_8_3_an_ambiguous_file_is_left_unlinked_by_the_monitor(self): + lookup = _Lookup("missing") + decision = _decide( + _graph(shapes.shared_prefix()), policy=Policy.MONITOR, lookup=lookup + ) + self.assertEqual(decision.kind, "never") + self.assertEqual(decision.unlinked_status, "ambiguous") + self.assertEqual(lookup.calls, 1) + + def test_step_8_4_an_ambiguous_file_gets_the_legacy_answer_in_the_scan(self): + decision = _decide(_graph(shapes.shared_prefix()), lookup=_Lookup()) + self.assertEqual(decision.kind, "positive") + self.assertEqual(decision.source, "legacy-parser") + self.assertEqual(decision.text, shapes.BRANCH_TEXTS[0]) + self.assertEqual(decision.review, "ambiguous") + + def test_step_8_4_an_untraceable_file_gets_the_legacy_answer_in_both(self): + for policy in Policy: + for inspected in (_inspected("unresolved"), _inspected("x", True)): + with ( + self.subTest(policy=policy, malformed=inspected.malformed), + mock.patch.object( + file_prompt, "inspection_resolution", return_value=inspected + ), + ): + decision = _decide( + _graph(_gated(sampled_graph("cat"))), + policy=policy, + lookup=_Lookup(), + ) + self.assertEqual(decision.kind, "positive") + self.assertEqual(decision.source, "legacy-parser") + self.assertEqual(decision.text, "cat") + + def test_step_8_4_nothing_traced_tries_the_recorded_run(self): + graph = sampled_graph() + graph["5"] = {**graph["3"]} + for sampler in ("3", "5"): + graph[sampler]["inputs"]["positive"] = ["9", 0] + graph["9"] = {"class_type": "BasicGuider", "inputs": {"width": 2048}} + decision = _decide(_graph(graph), lookup=_Lookup()) + self.assertEqual(decision.kind, "never") + self.assertTrue(decision.try_recorded_run) + # Ambiguous without a choice is reported unresolved. + self.assertEqual(decision.unlinked_status, "unresolved") + self.assertEqual(decision.review, "untraced") + + +class TestStepsWithoutAnUndecidedGraph(unittest.TestCase): + def test_step_9_ambiguous_parameters_only(self): + meta = { + "parameters": ( + "cat\nNegative prompt: blur\nNegative prompt: noise" + "\nSteps: 20, Sampler: Euler" + ) + } + decision = _decide(meta) + self.assertEqual(decision.kind, "never") + self.assertFalse(decision.try_recorded_run) + self.assertEqual(decision.unlinked_status, "unresolved") + + def test_step_10_unresolved_without_a1111_text(self): + decision = _decide({"prompt": {}, "parameters": 5}) + self.assertEqual(decision.kind, "never") + self.assertFalse(decision.try_recorded_run) + self.assertEqual(decision.unlinked_status, "unresolved") + + def test_step_11_sampler_less_with_one_text(self): + decision = _decide(_graph(flux_graph())) + self.assertEqual((decision.kind, decision.source), ("positive", "sampler-less")) + self.assertEqual(decision.text, "a red fox") + + def test_step_11_sampler_less_with_several_texts(self): + graph = flux_graph() + graph["50"] = { + "class_type": "Lora Loader (LoraManager)", + "inputs": {"model": ["1", 0], "clip": ["2", 0], "text": "a lora note"}, + } + graph["22"]["inputs"]["model"] = ["50", 0] + graph["6"]["inputs"]["clip"] = ["50", 1] + decision = _decide(_graph(graph)) + self.assertEqual(decision.kind, "never") + self.assertTrue(decision.try_recorded_run) + self.assertEqual(decision.unlinked_status, "unresolved") + meta = {**_graph(graph), "parameters": "a red fox\nNegative prompt: x"} + decision = _decide(meta) + self.assertEqual((decision.kind, decision.source), ("positive", "a1111")) + self.assertEqual(decision.text, "a red fox") + + def test_step_11_sampler_less_without_a_text(self): + graph = {"40": {"class_type": "easy showAnything", "inputs": {"text": "x"}}} + decision = _decide(_graph(graph)) + self.assertEqual(decision.kind, "never") + self.assertTrue(decision.try_recorded_run) + self.assertEqual(decision.unlinked_status, "found") + + def test_step_12_a1111_only(self): + decision = _decide({"parameters": "a cat"}) + self.assertEqual((decision.kind, decision.source), ("positive", "a1111")) + self.assertEqual(decision.text, "a cat") + decision = _decide({"parameters": "Negative prompt: blur"}) + self.assertEqual((decision.kind, decision.unlinked_status), ("never", "found")) + self.assertFalse(decision.try_recorded_run) + + def test_step_12_parsed_metadata_never_raises(self): + # The monitor hands parsed values: a dict prompt, non-string parameters. + decision = _decide({"prompt": {}, "parameters": "a cat"}) + self.assertEqual(decision.text, "a cat") + decision = _decide({"parameters": 5}) + self.assertEqual((decision.kind, decision.unlinked_status), ("never", "found")) + + +class TestJunkGate(unittest.TestCase): + JUNK = { + "blank": " ", + "number": "2048", + "lora-only": f"{LORA}, |", + "model-file": "models/checkpoints/Flux-Dev.SafeTensors", + } + + def test_each_junk_class_is_named(self): + for name, text in self.JUNK.items(): + with self.subTest(name=name): + self.assertEqual(junk_text(text), name) + for ending in ("ckpt", "pt", "pth", "bin", "gguf", "sft"): + self.assertEqual(junk_text(f"model.{ending}"), "model-file") + + def test_near_misses_are_not_junk(self): + for text in ( + f"a red fox {LORA}", + "a.safetensors b.safetensors", + "nan", + "a portrait saved as final.safetensors", + "4k", + ): + with self.subTest(text=text): + self.assertIsNone(junk_text(text)) + + def test_step_13_junk_from_any_step_is_never_stored(self): + cases = { + "resolver": _graph(sampled_graph("2048", "blur")), + "sampler-less": _graph(flux_graph(text=LORA)), + "a1111": {"parameters": "model.safetensors"}, + } + for step, meta in cases.items(): + with self.subTest(step=step): + decision = _decide(meta) + self.assertEqual(decision.kind, "never") + self.assertTrue(decision.source.startswith("junk:"), decision.source) + self.assertTrue(decision.try_recorded_run) + self.assertEqual(decision.unlinked_status, "found") + + def test_step_13_a_junk_legacy_answer_is_never_stored(self): + with _untraced_by_both(): + decision = _decide(_graph(_gated(sampled_graph(LORA))), lookup=_Lookup()) + self.assertEqual(decision.source, "junk:lora-only") + self.assertEqual(decision.kind, "never") + + +class TestParseComfyuiPrompt(unittest.TestCase): + def test_a_parsed_prompt_and_non_string_values_are_kept(self): + parsed = file_prompt.parse_comfyui_prompt( + {"prompt": {"1": {}}, "workflow": {"nodes": []}, "steps": 20} + ) + self.assertEqual(parsed["prompt"], {"1": {}}) + self.assertEqual(parsed["workflow"], {"nodes": []}) + self.assertEqual(parsed["parameters"]["steps"], 20) + self.assertIsNone(parsed["positive_prompt"]) + + def test_non_string_parameters_are_ignored(self): + parsed = file_prompt.parse_comfyui_prompt({"parameters": {"a": 1}}) + self.assertIsNone(parsed["positive_prompt"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/utils/file_prompt.py b/utils/file_prompt.py new file mode 100644 index 00000000..d9fabd67 --- /dev/null +++ b/utils/file_prompt.py @@ -0,0 +1,398 @@ +"""The per-file prompt decision shared by the output scan and the monitor (#247). + +``decide_file_prompt`` is the ladder: which prompt one output file names, +from its metadata alone plus, when the graph cannot decide, the save node +recorded history says wrote it. It does no file or database I/O; the caller +reads the metadata, the file size and history (through ``lookup``) and +stores the ``Decision``. + +The default resolver profile decides first, as ingestion always has. Only +for a file it leaves ambiguous or unresolved does the ladder ask the +inspector's resolve (``inspection_resolution``: the inspection profile, then +the recorded saver anchor), then the #170 browser parser (the legacy floor). +The policies differ in one place: for a file the inspector calls ambiguous, +the scan keeps the legacy floor and the monitor stores nothing. Junk text +(``junk_text``) is never stored, whichever step found it. +""" + +import json +import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Callable, Mapping, Optional + +from .generation_params import ( + PLACEHOLDERS, + GenerationParamsError, + extract_generation_params, + parse_page_json, +) +from .prompt_inspector import inspection_resolution +from .prompt_pair_resolver import _numeric, resolve_prompt_pair, sampler_less_prompt +from .review_capture import reviewable + +try: + from ..database.pair_ingestion import blank_positive +except ImportError: + from database.pair_ingestion import blank_positive + + +class Policy(Enum): + SCAN = "scan" + MONITOR = "monitor" + + +PAIR = "pair" +POSITIVE = "positive" +HISTORY = "history" +NEVER = "never" +KINDS = (PAIR, POSITIVE, HISTORY, NEVER) + +_UNDECIDED = ("ambiguous", "unresolved") + + +@dataclass(frozen=True) +class Decision: + """What to store for one file (see the design's "The ladder"). + + ``kind``: one of ``KINDS``. ``text``: the storage text of the positive. + ``canonical``: the parsed metadata stored with the link. ``resolution``: + the resolution that decided a ``pair``, else the default profile's + result. ``source``: which step named the text (log lines and the scan's + outcome detail). ``review``: ``ambiguous`` or ``untraced`` when the + default profile was ambiguous (the scan's Needs review lists). + ``try_recorded_run``: the scan tries the recorded run's stored prompt + when nothing else links. ``unlinked_status``: the scan status when + nothing links (None: no prompt metadata). + """ + + kind: str + text: Optional[str] = None + canonical: dict = field(default_factory=dict) + resolution: Optional[dict] = None + source: str = "none" + review: Optional[str] = None + try_recorded_run: bool = False + unlinked_status: Optional[str] = None + + +@dataclass(frozen=True) +class _File: + """The ladder's inputs beside the metadata.""" + + metadata: Mapping + canonical: dict + resolution: dict + review: Optional[str] + file_name: Optional[str] + in_output: bool + lookup: Optional[Callable[[], Any]] + policy: Policy + + +def decide_file_prompt( + metadata: Any, + *, + file_name: Optional[str], + file_size: Optional[int], + in_output: bool, + lookup: Optional[Callable[[], Any]], + policy: Policy, +) -> Decision: + """The ladder for one file's raw ``metadata``. + + ``prompt``/``workflow`` may be JSON strings (the scan) or parsed values + (the monitor). ``lookup`` returns the file's ``RecordedOutput`` or None; + it is None when history must not be asked yet. It is called at most + once, and only for a graph the default profile leaves undecided. + """ + if not metadata: + return Decision(NEVER, try_recorded_run=True) + canonical = _parsed(metadata) + if not canonical.get("prompt") and not canonical.get("parameters"): + return Decision(NEVER, canonical=canonical, try_recorded_run=True) + _with_size(canonical, file_size) + resolution = resolve_prompt_pair(canonical) + if blank_positive(resolution): + # Only prepend/append text: no stored prompt can own it (#342) + resolution = {**resolution, "state": "unresolved"} + found = _File( + metadata, + canonical, + resolution, + _review(resolution), + file_name, + in_output, + lookup, + policy, + ) + return _gated(_decide(found)) + + +def _decide(found: _File) -> Decision: + state = found.resolution["state"] + graph = found.canonical.get("prompt") + if state == "resolved": + positive = found.resolution["positive"] + text = positive.get("storage_text", positive["text"]) + return _decision(found, PAIR, text, "resolver") + if graph and state in _UNDECIDED: + return _undecided_graph(found) + if state == "ambiguous": + return _never(found, _unlinked(found)) + if state == "unresolved" and not isinstance(found.canonical.get("parameters"), str): + return _never(found, "unresolved") + if graph: + return _sampler_less(found, graph) + text = _a1111_positive(found.metadata) + if text is None: + return _never(found, "found") + return _decision(found, POSITIVE, text, "a1111") + + +def _undecided_graph(found: _File) -> Decision: + recorded = found.lookup() if found.lookup is not None else None + saver = recorded.saver_id if recorded is not None else None + inspected = inspection_resolution(found.metadata, found.file_name, saver) + text = _inspected_text(inspected.resolution) + if text is not None: + source = "recorded-run" if inspected.by_record else "inspector" + return _decision(found, POSITIVE, text, source) + if found.policy is Policy.MONITOR: + if found.lookup is None and found.in_output: + return _decision(found, HISTORY, None, "none") + if not inspected.malformed and inspected.resolution["state"] == "ambiguous": + return _never(found, _unlinked(found), try_recorded_run=True) + text = graph_prompt_text(found.metadata) + if text is not None: + return _decision(found, POSITIVE, text, "legacy-parser") + return _never(found, _unlinked(found), try_recorded_run=True) + + +def _sampler_less(found: _File, graph: Any) -> Decision: + # No node takes a positive or negative input (a guider or wrapper graph): + # the one wired text, never a guess (#252). + wired = sampler_less_prompt(graph) + if wired["text"] is not None: + return _decision(found, POSITIVE, wired["text"], "sampler-less") + if isinstance(found.canonical.get("parameters"), str): + # A metadata saver also wrote A1111 text: its positive prompt. + text = _a1111_positive(found.metadata) + if text is not None: + return _decision(found, POSITIVE, text, "a1111") + status = "unresolved" if wired["state"] == "ambiguous" else "found" + return _never(found, status, try_recorded_run=True) + + +def _gated(decision: Decision) -> Decision: + """Junk text is never stored; the file reports ``found`` (step 13).""" + if decision.kind not in (PAIR, POSITIVE): + return decision + junk = junk_text(decision.text) + if junk is None: + return decision + return Decision( + NEVER, + canonical=decision.canonical, + resolution=decision.resolution, + source=f"junk:{junk}", + review=decision.review, + try_recorded_run=True, + unlinked_status="found", + ) + + +def _decision(found: _File, kind: str, text: Optional[str], source: str) -> Decision: + return Decision( + kind, + text=text, + canonical=found.canonical, + resolution=found.resolution, + source=source, + review=found.review, + ) + + +def _never(found: _File, status: str, *, try_recorded_run: bool = False) -> Decision: + return Decision( + NEVER, + canonical=found.canonical, + resolution=found.resolution, + review=found.review, + try_recorded_run=try_recorded_run, + unlinked_status=status, + ) + + +def _review(resolution: dict) -> Optional[str]: + if resolution["state"] != "ambiguous": + return None + return "ambiguous" if reviewable(resolution) else "untraced" + + +def _unlinked(found: _File) -> str: + """Ambiguous without a choice is reported unresolved.""" + if found.review == "untraced": + return "unresolved" + return found.resolution["state"] + + +def _inspected_text(resolution: dict) -> Optional[str]: + """The storage text of a positive the inspector shows, else None.""" + if resolution.get("state") in _UNDECIDED: + return None + positive = resolution.get("positive") + if not isinstance(positive, dict): + return None + text = positive.get("storage_text", positive.get("text")) + return text if isinstance(text, str) and text.strip() else None + + +def _parsed(metadata: Mapping) -> dict: + canonical = dict(metadata) + for key in ("prompt", "workflow"): + if isinstance(canonical.get(key), str): + try: + canonical[key] = json.loads(canonical[key]) + except (ValueError, TypeError): + # Preserve malformed metadata for conservative resolver rejection. + pass + return canonical + + +def _with_size(canonical: dict, file_size: Optional[int]) -> None: + """Record the file's size in ``file_info`` unless the metadata carries one. + + Stored as ``generated_images.file_size``; the "new" scan scope skips a + known file only while its size is unchanged. + """ + info = canonical.get("file_info") + info = dict(info) if isinstance(info, dict) else {} + if info.get("size") is None and file_size is not None: + info["size"] = file_size + if info: + canonical["file_info"] = info + + +def _a1111_positive(metadata: Mapping) -> Optional[str]: + text = parse_comfyui_prompt(metadata)["positive_prompt"] + return text.strip() if text and text.strip() else None + + +_LORA_TAG = re.compile(r"]*>") +_LORA_SEPARATORS = " \t\r\n\f\v,;|" +_MODEL_FILE = re.compile(r"\S*\.(?:safetensors|ckpt|pt|pth|bin|gguf|sft)", re.I) + + +def junk_text(text: Any) -> Optional[str]: + """The #252 junk class of ``text``, or None when it may be a prompt.""" + stripped = text.strip() if isinstance(text, str) else "" + if not stripped: + return "blank" + if _numeric(stripped): + return "number" + if not _LORA_TAG.sub("", stripped).strip(_LORA_SEPARATORS): + return "lora-only" + if _MODEL_FILE.fullmatch(stripped): + return "model-file" + return None + + +def graph_prompt_text(metadata: Mapping) -> Optional[str]: + """The positive prompt the browser parser traces from the graphs, or None. + + Records parse as the legacy pages parsed them (NaN reads as null). + """ + records = {} + for key in ("prompt", "workflow"): + value = metadata.get(key) + records[key] = parse_page_json(value) if isinstance(value, str) else value + try: + positive = extract_generation_params(records)["positivePrompt"] + except GenerationParamsError: + return None + if not isinstance(positive, str) or positive == PLACEHOLDERS["positivePrompt"]: + return None + return positive.strip() or None + + +_COMMON_FIELDS = ( + "positive", + "negative", + "steps", + "cfg", + "sampler", + "scheduler", + "seed", +) + + +def _json_or_raw(value: Any) -> Any: + if not isinstance(value, str): + return value + try: + return json.loads(value) + except json.JSONDecodeError: + return value + + +def parse_comfyui_prompt(metadata: Mapping) -> dict: + """Parse ComfyUI and A1111 prompt data from metadata. + + Accepts the file's strings or parsed values: a parsed ``prompt`` or + ``workflow`` is kept as it is and a non-string ``parameters`` is ignored. + """ + result = { + "prompt": None, + "workflow": None, + "parameters": {}, + "positive_prompt": None, + "negative_prompt": None, + } + params = metadata.get("parameters") + if isinstance(params, str): + _parse_a1111(params, result) + if result["positive_prompt"] is None: + for key in ("prompt", "workflow"): + if key in metadata: + result[key] = _json_or_raw(metadata[key]) + for name in _COMMON_FIELDS: + if name in metadata: + result["parameters"][name] = _json_or_raw(metadata[name]) + return result + + +_SETTINGS_FIELDS = ("Steps", "Sampler", "Schedule type", "CFG scale", "Seed") + + +def _parse_a1111(params: str, result: dict) -> None: + """A1111/Forge ``parameters``: a possibly multi-line positive, then a + "Negative prompt:" section, then a trailing "Steps: ..." settings line. + Taking the first line as the positive would truncate a multi-line prompt, + so the text is split on the settings line instead. + """ + body = params + settings = "" + found = re.search(r"\nSteps:\s", body) + if found: + settings = body[found.start() + 1 :].strip() + body = body[: found.start()] + negative = None + found = re.search(r"(?im)^\s*negative prompt:\s*", body) + if found: + negative = body[found.end() :].strip() + body = body[: found.start()] + elif body.lower().lstrip().startswith("negative prompt:"): + negative = body.lower().lstrip()[len("negative prompt:") :].strip() + body = "" + result["positive_prompt"] = body.strip() or None + if negative: + result["negative_prompt"] = negative + result["parameters"]["parameters"] = params + for name in _SETTINGS_FIELDS if settings else (): + found = re.search( + r"(?i)(?:^|,\s*)" + re.escape(name) + r":\s*([^,]+)", settings + ) + if found: + key = name.lower().replace(" ", "_") + result["parameters"][key] = found.group(1).strip()