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).
This commit is contained in:
Vito Sansevero
@@ -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 = "<lora:style:1.00>"
|
||||
|
||||
|
||||
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}, <lora:detail:0.5> |",
|
||||
"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()
|
||||
@@ -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:[^>]*>")
|
||||
_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()
|
||||
Reference in New Issue
Block a user