Vito Sansevero
1d730d8e97
The output scan skipped every file the pair resolver called ambiguous or unresolved, so workflows where several samplers share one prompt (a base pass, an upscale pass and a detailer) never reached the Library from a scan. It now links the positive prompt the shared parser traces for those files, the prompt the inspector and the legacy pages show. Files the resolver decides are stored as before, and graphs without a recognized sampler keep the legacy extractor (#252). Read-only scans of a 38,602-file output folder into temporary databases: - links: 1,500 before, 7,954 after; none of the 1,500 dropped or changed; - of the added links that differ from the pre-redesign scan, 1,275 replace a number, a LoRA tag, a fragment or a truncated prompt that scan stored.
172 lines
7.3 KiB
Python
172 lines
7.3 KiB
Python
"""Metadata scanning enriches known pairs and conservatively leaves conflicts."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from tests.test_api_admin import AdminAPITestCase
|
|
|
|
|
|
def graph(positive="cat", negative="blur"):
|
|
return {
|
|
"1": {"class_type": "CLIPTextEncode", "inputs": {"text": positive}},
|
|
"2": {"class_type": "CLIPTextEncode", "inputs": {"text": negative}},
|
|
"3": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["1", 0], "negative": ["2", 0]},
|
|
},
|
|
}
|
|
|
|
|
|
def sampled_graph(positive="cat", negative="blur"):
|
|
"""A KSampler with a model link, as real workflows have."""
|
|
g = graph(positive, negative)
|
|
g["4"] = {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}}
|
|
g["3"]["inputs"]["model"] = ["4", 0]
|
|
return g
|
|
|
|
|
|
class TestPairScan(AdminAPITestCase):
|
|
def test_pair_scan_repeated_is_unchanged_and_empty_negative_is_supported(self):
|
|
path = self.output_dir / "pair.png"
|
|
meta = {"prompt": json.dumps(graph(negative=""))}
|
|
self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
|
|
self.assertEqual(self.api._ingest_scanned_file(path, meta), "unchanged")
|
|
pairs = self.api.db.list_prompt_pairs()["pairs"]
|
|
self.assertEqual(len(pairs), 1)
|
|
self.assertEqual(pairs[0]["negative_text"], "")
|
|
self.assertEqual(pairs[0]["run_count"], 0)
|
|
|
|
def _scanned(self, name, prompt_graph):
|
|
meta = {"prompt": json.dumps(prompt_graph)}
|
|
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_samplers_sharing_one_prompt_link_that_prompt(self):
|
|
g = sampled_graph()
|
|
g["5"] = {**g["3"]}
|
|
self.assertEqual(self._scanned("multi.png", g), "added")
|
|
self.assertEqual(self._texts(), ["cat"])
|
|
|
|
def test_samplers_with_different_prompts_link_the_first_sampler_prompt(self):
|
|
# The prompt the inspector and the legacy panels show for this file.
|
|
g = sampled_graph()
|
|
g["6"] = {"class_type": "CLIPTextEncode", "inputs": {"text": "dog"}}
|
|
g["12"] = {**g["3"], "inputs": {**g["3"]["inputs"], "positive": ["6", 0]}}
|
|
self.assertEqual(self._scanned("different.png", g), "added")
|
|
self.assertEqual(self._texts(), ["cat"])
|
|
|
|
def test_an_unsupported_pass_through_still_links_the_traced_prompt(self):
|
|
g = sampled_graph()
|
|
g["7"] = {
|
|
"class_type": "MyConditioningTweak",
|
|
"inputs": {"conditioning": ["1", 0]},
|
|
}
|
|
g["3"]["inputs"]["positive"] = ["7", 0]
|
|
g["5"] = {**g["3"]}
|
|
self.assertEqual(self._scanned("tweak.png", g), "added")
|
|
self.assertEqual(self._texts(), ["cat"])
|
|
|
|
def test_a_graph_without_a_traceable_prompt_saves_nothing(self):
|
|
g = sampled_graph()
|
|
g["5"] = {**g["3"]}
|
|
for sampler in ("3", "5"):
|
|
g[sampler]["inputs"]["positive"] = ["9", 0]
|
|
g["9"] = {
|
|
"class_type": "BasicGuider",
|
|
"inputs": {"width": 2048, "height": 2048},
|
|
}
|
|
self.assertEqual(self._scanned("untraceable.png", g), "ambiguous")
|
|
self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0)
|
|
self.assertEqual(self._texts(), [])
|
|
|
|
def test_numeric_widgets_are_never_stored_as_a_prompt(self):
|
|
# The pre-redesign scan stored "2048" from a resolution widget here.
|
|
g = {
|
|
"1": {"class_type": "CLIPTextEncode", "inputs": {"text": ["8", 0]}},
|
|
"8": {"class_type": "DisplayText", "inputs": {"text_size": 2048}},
|
|
"3": {
|
|
"class_type": "SamplerCustomAdvanced",
|
|
"inputs": {"guider": ["4", 0], "sigmas": ["5", 0]},
|
|
},
|
|
"4": {"class_type": "BasicGuider", "inputs": {"conditioning": ["1", 0]}},
|
|
}
|
|
self.assertIn(
|
|
self._scanned("numbers.png", g), ("ambiguous", "unresolved", "found")
|
|
)
|
|
self.assertEqual(self._texts(), [])
|
|
|
|
def test_batch_exposes_pair_counts_and_actual_unchanged_status(self):
|
|
meta = {"parameters": "cat\nNegative prompt: blur\nSteps: 20, Sampler: Euler"}
|
|
batch = [(Path(self.output_dir / "a.png"), meta)]
|
|
first = self.api._ingest_scan_batch_sync(batch)
|
|
self.assertEqual(first["pairs_added"], 1)
|
|
self.assertEqual(first["pair_links_added"], 1)
|
|
second = self.api._ingest_scan_batch_sync(batch)
|
|
self.assertEqual(second["unchanged"], 1)
|
|
self.assertEqual(second["linked"], 0)
|
|
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"]})}
|
|
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")
|
|
|
|
async def test_cancel_finishes_at_safe_boundary_and_restart_is_idempotent(self):
|
|
from py.api.scan_job import ScanJob
|
|
|
|
job = ScanJob()
|
|
self.api._scan_job = job
|
|
response = await self.client.post("/prompt_manager/scan/cancel")
|
|
self.assertEqual(response.status, 200)
|
|
self.assertTrue((await response.json())["cancellation_requested"])
|
|
counts = self.api._ingest_scan_batch_sync(
|
|
[(self.output_dir / "cancel.png", {"prompt": json.dumps(graph())})]
|
|
)
|
|
self.assertEqual(counts["processed"], 0)
|
|
self.assertEqual(self.api.db.list_prompt_pairs()["total"], 0)
|
|
job.finish()
|
|
self.api._scan_job = ScanJob()
|
|
first = self.api._ingest_scan_batch_sync(
|
|
[(self.output_dir / "cancel.png", {"prompt": json.dumps(graph())})]
|
|
)
|
|
self.assertEqual(first["pairs_added"], 1)
|
|
self.assertEqual(
|
|
self.api._ingest_scan_batch_sync(
|
|
[(self.output_dir / "cancel.png", {"prompt": json.dumps(graph())})]
|
|
)["unchanged"],
|
|
1,
|
|
)
|
|
|
|
def test_legacy_found_outcome_is_not_counted_twice(self):
|
|
result = self.api._ingest_scan_batch_sync(
|
|
[(self.output_dir / "blank.png", {"parameters": "Negative prompt: blur"})]
|
|
)
|
|
self.assertEqual(result["processed"], 1)
|
|
self.assertEqual(result["found"], 1)
|
|
|
|
def test_ambiguous_a1111_does_not_fall_back_to_legacy_guess(self):
|
|
meta = {
|
|
"parameters": (
|
|
"cat\nNegative prompt: blur\nNegative prompt: noise"
|
|
"\nSteps: 20, Sampler: Euler"
|
|
)
|
|
}
|
|
self.assertEqual(
|
|
self.api._ingest_scanned_file(self.output_dir / "ambiguous.png", meta),
|
|
"ambiguous",
|
|
)
|
|
self.assertEqual(self.api.db.get_recent_prompts()["prompts"], [])
|
|
|
|
def test_raw_png_workflow_is_stored_as_object_and_repeat_is_unchanged(self):
|
|
path = self.output_dir / "workflow.png"
|
|
workflow = {"nodes": [{"id": 1, "type": "PromptManagerPair"}]}
|
|
metadata = {"prompt": json.dumps(graph()), "workflow": json.dumps(workflow)}
|
|
self.assertEqual(self.api._ingest_scanned_file(path, metadata), "added")
|
|
self.assertEqual(
|
|
self.api.db.get_image_prompt_info(str(path))["workflow_data"], workflow
|
|
)
|
|
self.assertEqual(self.api._ingest_scanned_file(path, metadata), "unchanged")
|