feat: PromptSearchList batch workflow support & image linking v3.2.2

database/operations.py

@@ -190,6 +190,7 @@ def search_prompts(
         folder: Optional[str] = None,
         limit: int = 100,
         offset: int = 0,
+        tag_partial: bool = False,
     ) -> List[Dict[str, Any]]:
         """
         Search prompts with various filters.
@@ -205,6 +206,7 @@ def search_prompts(
             folder: Filter by subfolder name in generated image paths
             limit: Maximum number of results
             offset: Number of results to skip
+            tag_partial: Use LIKE matching for tags instead of exact match
 
         Returns:
             List of dictionaries containing prompt data
@@ -222,12 +224,20 @@ def search_prompts(
 
         if tags:
             for tag in tags:
-                query_parts.append(
-                    "AND prompts.id IN ("
-                    "  SELECT pt.prompt_id FROM prompt_tags pt"
-                    "  JOIN tags t ON pt.tag_id = t.id WHERE t.name = ?)"
-                )
-                params.append(tag)
+                if tag_partial:
+                    query_parts.append(
+                        "AND prompts.id IN ("
+                        "  SELECT pt.prompt_id FROM prompt_tags pt"
+                        "  JOIN tags t ON pt.tag_id = t.id WHERE t.name LIKE ?)"
+                    )
+                    params.append(f"%{tag}%")
+                else:
+                    query_parts.append(
+                        "AND prompts.id IN ("
+                        "  SELECT pt.prompt_id FROM prompt_tags pt"
+                        "  JOIN tags t ON pt.tag_id = t.id WHERE t.name = ?)"
+                    )
+                    params.append(tag)
 
         if folder:
             # Escape LIKE wildcards in the folder name

prompt_search_list.py

@@ -5,6 +5,7 @@
 OUTPUT_IS_LIST=True, allowing direct connection to nodes that accept list inputs.
 """
 
+import re
 from typing import Any, Dict, List, Tuple
 
 try:
@@ -79,7 +80,7 @@ def INPUT_TYPES(cls) -> InputTypeDict:
                     IO.STRING,
                     {
                         "default": "",
-                        "tooltip": "Comma-separated list of tags to filter by",
+                        "tooltip": "Comma-separated list of tags to filter by (partial match)",
                     },
                 ),
                 "min_rating": (
@@ -100,20 +101,33 @@ def INPUT_TYPES(cls) -> InputTypeDict:
                         "tooltip": "Maximum number of results to return",
                     },
                 ),
+                "skip_multipart": (
+                    "BOOLEAN",
+                    {
+                        "default": True,
+                        "tooltip": "Skip prompts containing Clip_1/Clip_2/etc. multi-part markers",
+                    },
+                ),
             },
         }
 
-    RETURN_TYPES = (IO.STRING,)
-    RETURN_NAMES = ("prompts",)
-    OUTPUT_IS_LIST = (True,)
-    OUTPUT_TOOLTIPS = ("List of prompt texts matching the search criteria.",)
+    RETURN_TYPES = (IO.STRING, IO.STRING, IO.INT)
+    RETURN_NAMES = ("prompts", "preview", "count")
+    OUTPUT_IS_LIST = (True, False, False)
+    OUTPUT_TOOLTIPS = (
+        "List of prompt texts matching the search criteria.",
+        "Preview of all found prompts (for display).",
+        "Number of prompts found.",
+    )
     FUNCTION = "search"
     OUTPUT_NODE = True
     CATEGORY = "🫶 ComfyAssets/🧠 Prompts"
     DESCRIPTION = (
         "Searches the prompt database and outputs matching prompts as a list. "
         "Use this node to retrieve stored prompts for batch processing workflows. "
-        "Connect to nodes that accept list inputs like String OutputList or batch processors."
+        "'prompts' output sends each prompt individually to downstream nodes (batch). "
+        "'preview' output shows all found prompts as one text block. "
+        "'count' output shows how many prompts were found."
     )
 
     def search(
@@ -123,7 +137,8 @@ def search(
         tags: str = "",
         min_rating: int = 0,
         limit: int = 50,
-    ) -> Tuple[List[str]]:
+        skip_multipart: bool = True,
+    ):
         """
         Search for prompts matching the given criteria.
 
@@ -135,46 +150,77 @@ def search(
             limit: Maximum number of results to return
 
         Returns:
-            Tuple containing a list of prompt text strings
+            Dict with 'ui' display info and 'result' tuple
         """
         try:
             # Parse tags from comma-separated string
             tags_list = None
             if tags and tags.strip():
                 tags_list = [tag.strip() for tag in tags.split(",") if tag.strip()]
 
-            # Perform search
+            # Perform search with partial tag matching for discovery
             results = self.db.search_prompts(
                 text=text.strip() if text and text.strip() else None,
                 category=category.strip() if category and category.strip() else None,
                 tags=tags_list,
+                tag_partial=True,
                 rating_min=min_rating if min_rating > 0 else None,
                 limit=limit,
             )
 
-            # Extract just the prompt text from results
-            prompt_texts = [r["text"] for r in results if r.get("text")]
-
-            self.logger.debug(
-                f"Search returned {len(prompt_texts)} prompts "
-                f"(text='{text[:20]}...' if text else '', category='{category}', "
-                f"tags={tags_list}, min_rating={min_rating}, limit={limit})"
-            )
-
-            # Return empty list if no results (not an error)
+            # Extract prompt text, collapsing newlines to spaces so downstream
+            # nodes that split on \n (e.g. StringOutputList) treat each DB
+            # entry as a single prompt.
+            prompt_texts = [
+                " ".join(r["text"].split()) for r in results if r.get("text")
+            ]
+
+            # Filter out multi-part prompts (Clip_1/Clip_2 markers)
+            if skip_multipart:
+                prompt_texts = [
+                    p for p in prompt_texts if not re.search(r"Clip_\d+", p)
+                ]
+
+            # Filter out prompts that are only LoRA tags with no actual content.
+            # Uses subtraction instead of repeated groups to avoid backtracking.
+            prompt_texts = [
+                p for p in prompt_texts if re.sub(r"<lora:[^>]+>", "", p).strip()
+            ]
+
+            count = len(prompt_texts)
+
+            # Build a preview: numbered list of truncated prompts
+            preview_lines = []
+            for i, p in enumerate(prompt_texts, 1):
+                truncated = p[:120] + "..." if len(p) > 120 else p
+                preview_lines.append(f"[{i}] {truncated}")
+            preview = "\n".join(preview_lines) if preview_lines else "No results found"
+
+            # Must return at least one element — ComfyUI's slice_dict
+            # indexes into OUTPUT_IS_LIST outputs and crashes on empty lists.
             if not prompt_texts:
                 self.logger.info("Search returned no results")
-                return ([],)
+                return {
+                    "ui": {"text": ["No results found"]},
+                    "result": ([""], preview, count),
+                }
 
-            return (prompt_texts,)
+            return {
+                "ui": {"text": [f"Found {count} prompts"]},
+                "result": (prompt_texts, preview, count),
+            }
 
         except Exception as e:
             self.logger.error(f"Search error: {e}", exc_info=True)
-            # Return empty list on error to avoid breaking workflows
-            return ([],)
+            return {
+                "ui": {"text": [f"Search error: {e}"]},
+                "result": ([""], f"Error: {e}", 0),
+            }
 
     @classmethod
-    def IS_CHANGED(cls, text="", category="", tags="", min_rating=0, limit=50):
+    def IS_CHANGED(
+        cls, text="", category="", tags="", min_rating=0, limit=50, skip_multipart=True
+    ):
         """
         Always re-execute to get fresh results from the database.
 

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "promptmanager"
 description = "A powerful ComfyUI custom node that extends the standard text encoder with persistent prompt storage, advanced search capabilities, and an automatic image gallery system using SQLite."
-version = "3.2.1"
+version = "3.2.2"
 license = {file = "LICENSE"}
 dependencies = ["# Core dependencies for PromptManager", "# Note: Most dependencies are already included with ComfyUI", "# Already included with Python standard library:", "# - sqlite3", "# - hashlib", "# - json", "# - datetime", "# - os", "# - typing", "# - threading", "# - uuid", "# Required for gallery functionality:", "watchdog>=2.1.0              # For file system monitoring", "Pillow>=8.0.0                # For image metadata extraction (usually included with ComfyUI)", "# Optional dependencies for enhanced search functionality:", "# fuzzywuzzy[speedup]>=0.18.0  # For fuzzy string matching (optional)", "# sqlalchemy>=1.4.0            # For advanced ORM features (optional)", "# Development dependencies (optional):", "# pytest>=6.0.0                # For running tests", "# black>=22.0.0                # For code formatting", "# flake8>=4.0.0                # For linting", "# mypy>=0.910                   # For type checking"]
 

tests/test_prompt_tracker.py

@@ -34,6 +34,8 @@ def _make_tracker(prompt_timeout=600, cleanup_interval=300):
     tracker._local = threading.local()
     tracker.active_prompts = {}
     tracker.lock = threading.Lock()
+    tracker._prompt_queue = []
+    tracker._queue_lock = threading.Lock()
     tracker.cleanup_interval = cleanup_interval
     tracker.prompt_timeout = prompt_timeout
     # Don't start cleanup thread in tests
@@ -256,5 +258,59 @@ def test_expired_prompts_cleaned(self):
         self.assertEqual(len(expired_ids), 1)
 
 
+class TestPromptQueue(unittest.TestCase):
+    """Test the FIFO prompt queue for batch workflow image linking."""
+
+    def setUp(self):
+        self.tracker = _make_tracker()
+
+    def test_queue_empty_initially(self):
+        result = self.tracker.pop_next_prompt()
+        self.assertIsNone(result)
+
+    def test_set_prompt_pushes_to_queue(self):
+        self.tracker.set_current_prompt("p1", {"prompt_id": 10})
+        self.assertEqual(len(self.tracker._prompt_queue), 1)
+
+    def test_pop_returns_fifo_order(self):
+        self.tracker.set_current_prompt("first", {"prompt_id": 1})
+        self.tracker.set_current_prompt("second", {"prompt_id": 2})
+        self.tracker.set_current_prompt("third", {"prompt_id": 3})
+
+        p1 = self.tracker.pop_next_prompt()
+        p2 = self.tracker.pop_next_prompt()
+        p3 = self.tracker.pop_next_prompt()
+
+        self.assertEqual(p1["id"], 1)
+        self.assertEqual(p2["id"], 2)
+        self.assertEqual(p3["id"], 3)
+
+    def test_pop_returns_none_when_exhausted(self):
+        self.tracker.set_current_prompt("only", {"prompt_id": 1})
+        self.tracker.pop_next_prompt()
+        result = self.tracker.pop_next_prompt()
+        self.assertIsNone(result)
+
+    def test_queue_preserves_prompt_text(self):
+        self.tracker.set_current_prompt("hello world", {"prompt_id": 5})
+        ctx = self.tracker.pop_next_prompt()
+        self.assertEqual(ctx["text"], "hello world")
+
+    def test_batch_simulation(self):
+        """Simulate a batch workflow: encode 10 prompts, then pop 10 in order."""
+        ids = list(range(100, 110))
+        for pid in ids:
+            self.tracker.set_current_prompt(f"prompt_{pid}", {"prompt_id": pid})
+
+        popped_ids = []
+        while True:
+            ctx = self.tracker.pop_next_prompt()
+            if ctx is None:
+                break
+            popped_ids.append(ctx["id"])
+
+        self.assertEqual(popped_ids, ids)
+
+
 if __name__ == "__main__":
     unittest.main()