nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(library): filter prompts by output folder and browse tags with Any (#444)

Browse Source 
Refine this view on the Prompt library gains the concept's Folder (folders
below an output root, read when the dialog opens) and a Tag match select
(All / Any). The listing takes folder and tag_mode, matches folders on the
normalized file_path under every spelling of each output root, echoes both
so a stale server is an error, and saved Library views keep them. Recorded
as D150.

Refs #444
This commit is contained in:
Vito Sansevero
2026-10-28 17:15:25 -07:00
parent 4040007264
commit 6372bc7c63
17 changed files with 768 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/library.py
+70 -16
View File
@@ -5,6 +5,7 @@ import sqlite3
from collections.abc import Iterator
from contextlib import contextmanager
from .library_folders import folder_condition, list_library_folders
from .library_tokens import library_edit_token
_ELIGIBLE = """(EXISTS(SELECT 1 FROM prompt_pairs ep WHERE ep.positive_prompt_id=p.id)
@@ -121,7 +122,44 @@ def _archive(value: str) -> str:
return ARCHIVE_SCOPES[value]
def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list]:
# How the Tags filter matches (#444): every tag (AND) or any of them (OR).
TAG_MODES = ("all", "any")
def _tag_mode(tag_mode: object) -> str:
if not isinstance(tag_mode, str) or tag_mode not in TAG_MODES:
raise ValueError("tag_mode must be one of " + ", ".join(TAG_MODES))
return tag_mode
def _tag_conditions(tags: list[str], any_tag: bool) -> tuple[list, list, list]:
"""Parent-arm and pair-arm tag conditions and the tags' parameters."""
parent_exists = (
"EXISTS(SELECT 1 FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
"WHERE pt.prompt_id=p.id AND t.name{test})"
)
pair_exists = (
"EXISTS(SELECT 1 FROM prompt_pair_tags pt JOIN tags t ON t.id=pt.tag_id "
"WHERE pt.pair_id=pp.id AND t.name{test})"
)
if any_tag and tags:
# One condition per arm: the card, or one active pair, carries any tag.
test = " IN (" + ",".join("?" for _ in tags) + ")"
return [parent_exists.format(test=test)], [pair_exists.format(test=test)], tags
return (
[parent_exists.format(test="=?") for _ in tags],
[pair_exists.format(test="=?") for _ in tags],
tags,
)
def _filter(
text: str,
category: str | None,
tags: list[str],
tag_mode: str = "all",
folder_prefixes: list[str] | None = None,
) -> tuple[str, list]:
if not isinstance(text, str) or len(text) > 500:
raise ValueError("Search text must be a string of at most 500 characters")
if category is not None and (not isinstance(category, str) or len(category) > 200):
@@ -146,17 +184,13 @@ def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list
pair.append("pp.category=?")
parent_params.append(category)
pair_params.append(category)
for tag in tags:
parent.append(
"EXISTS(SELECT 1 FROM prompt_tags pt JOIN tags t ON t.id=pt.tag_id "
"WHERE pt.prompt_id=p.id AND t.name=?)"
)
pair.append(
"EXISTS(SELECT 1 FROM prompt_pair_tags pt JOIN tags t ON t.id=pt.tag_id "
"WHERE pt.pair_id=pp.id AND t.name=?)"
)
parent_params.append(tag)
pair_params.append(tag)
parent_tags, pair_tags, tag_params = _tag_conditions(
tags, _tag_mode(tag_mode) == "any"
)
parent.extend(parent_tags)
pair.extend(pair_tags)
parent_params.extend(tag_params)
pair_params.extend(tag_params)
condition = (
"("
+ (" AND ".join(parent) or "1=1")
@@ -168,7 +202,13 @@ def _filter(text: str, category: str | None, tags: list[str]) -> tuple[str, list
+ " AND ".join(pair)
+ ")"
)
return f"({_ELIGIBLE}) AND ({condition})", [*parent_params, *pair_params]
params = [*parent_params, *pair_params]
if folder_prefixes:
# A whole-card condition around both arms: images link to the parent.
folder, folder_params = folder_condition(folder_prefixes)
condition = f"({condition}) AND {folder}"
params.extend(folder_params)
return f"({_ELIGIBLE}) AND ({condition})", params
class LibraryOperations:
@@ -251,6 +291,8 @@ class LibraryOperations:
text: str = "",
category: str | None = None,
tags: list[str] | None = None,
tag_mode: str = "all",
folder_prefixes: list[str] | None = None,
sort: str = "newest",
limit: int = 40,
after: dict | None = None,
@@ -258,7 +300,9 @@ class LibraryOperations:
_limit(limit)
if sort not in _ORDERS:
raise ValueError("Unknown library sort order")
where, params = _filter(text, category, [] if tags is None else tags)
where, params = _filter(
text, category, [] if tags is None else tags, tag_mode, folder_prefixes
)
if after is not None:
if (
not isinstance(after, dict)
@@ -311,6 +355,8 @@ class LibraryOperations:
"items": rows,
"total": total,
"variant_total": variant_total,
# Echoed so a client can tell this read applied its tag match.
"tag_mode": tag_mode,
"has_more": has_more,
"next_key": next_key,
}
@@ -321,6 +367,8 @@ class LibraryOperations:
text: str = "",
category: str | None = None,
tags: list[str] | None = None,
tag_mode: str = "all",
folder_prefixes: list[str] | None = None,
sort: str = "newest",
limit: int = 40,
after: dict | None = None,
@@ -333,6 +381,8 @@ class LibraryOperations:
text=text,
category=category,
tags=tags,
tag_mode=tag_mode,
folder_prefixes=folder_prefixes,
sort=sort,
limit=limit,
after=after,
@@ -463,9 +513,13 @@ class LibraryOperations:
"categories": [row["category"] for row in rows],
}
def get_library_options(self) -> dict:
def get_library_options(self, folder_roots=None) -> dict:
"""Categories; with *folder_roots* (root spellings) also ``folders``."""
with self.library_snapshot() as (conn, epoch, _ceiling):
return self._get_library_options(conn, epoch)
result = self._get_library_options(conn, epoch)
if folder_roots is not None:
result["folders"] = list_library_folders(conn, folder_roots, _ELIGIBLE)
return result
def get_library_previews(self, positive_ids: list[int]) -> dict[int, list[dict]]:
"""Read at most two links per parent in one bounded page query."""
database/library_folders.py
+111
View File
@@ -0,0 +1,111 @@
"""Library output-folder filter (#444): folder names, root prefixes and listing.
A folder is a ``/``-joined path relative to an output root, as Images' folder
filter and the legacy ``/prompt_manager/subfolders`` name them. A parent
matches a folder when one of its linked images (``generated_images.prompt_id``,
the link its previews and image count read) lies in that folder or below it,
under any spelling of any output root.
Matching compares ``generated_images.file_path``, which is always
``normcase(normpath(abspath(path)))`` (``models.normalize_image_path``), with
prefixes built by the same normalization here, so case folding is the
platform's own on both sides (Windows folds non-ASCII too, unlike SQLite's
``lower()``) and the comparison is an exact ``substr`` with no wildcards.
"""
import os
import sqlite3
_MAX_FOLDER = 1024
_MAX_PREFIXES = 64
def normalize_folder(value: object) -> str:
"""The folder as ``a/b``: ``\\`` read as ``/``, outer slashes dropped.
Empty means no folder filter. ``.``/``..``/empty segments, a drive or
control characters are refused: the value only ever names a folder below a
root, never a path of its own.
"""
if not isinstance(value, str) or len(value) > _MAX_FOLDER:
raise ValueError(f"folder must be text of at most {_MAX_FOLDER} characters")
folder = value.replace("\\", "/").strip("/")
if not folder:
return ""
parts = folder.split("/")
if (
any(part in ("", ".", "..") for part in parts)
or ":" in parts[0]
or any(ord(char) < 32 or ord(char) == 127 for char in folder)
):
raise ValueError("folder must be a folder below an output root")
return folder
def _base(root: object, pathmod) -> str:
return pathmod.normcase(pathmod.normpath(str(root)))
def folder_prefixes(roots, folder: str, pathmod=os.path) -> list[str]:
"""One ``file_path`` prefix per root spelling for a normalized folder."""
prefixes = []
for root in roots:
joined = pathmod.join(_base(root, pathmod), *folder.split("/"))
prefix = pathmod.normcase(pathmod.normpath(joined)) + pathmod.sep
if prefix not in prefixes:
prefixes.append(prefix)
if not prefixes or len(prefixes) > _MAX_PREFIXES:
raise ValueError("No output folder to filter by")
return prefixes
def folder_condition(prefixes: list[str]) -> tuple[str, list]:
"""A condition on alias ``p``: one linked image under any prefix."""
tests = " OR ".join("substr(fg.file_path,1,?)=?" for _ in prefixes)
params = [value for prefix in prefixes for value in (len(prefix), prefix)]
return (
"EXISTS(SELECT 1 FROM generated_images fg WHERE fg.prompt_id=p.id "
f"AND ({tests}))",
params,
)
def relative_folder(image_path: str, roots, pathmod=os.path) -> str | None:
"""The folder of *image_path* below the first root spelling holding it.
The spelling of the stored path is kept (its case on Windows); ``None``
for an image directly in a root or outside every root.
"""
directory = pathmod.normpath(pathmod.dirname(image_path))
folded = pathmod.normcase(directory)
for root in roots:
base = _base(root, pathmod)
start = base if base.endswith(pathmod.sep) else base + pathmod.sep
if not folded.startswith(start):
continue
# normcase keeps the length for ordinary names; else show the folded form.
rest = (directory if len(directory) == len(folded) else folded)[len(start) :]
parts = rest.split(pathmod.sep)
if any(part in ("", ".", "..") for part in parts):
return None
return "/".join(parts)
return None
def list_library_folders(
conn: sqlite3.Connection, roots, eligible: str, pathmod=os.path
) -> list[str]:
"""Every folder (and ancestor) holding an image of an eligible parent."""
folders = set()
for (image_path,) in conn.execute(
"SELECT DISTINCT g.image_path FROM generated_images g "
"JOIN prompts p ON p.id=g.prompt_id WHERE " + eligible
):
if not isinstance(image_path, str) or not image_path:
continue
folder = relative_folder(image_path, roots, pathmod)
if folder is None:
continue
parts = folder.split("/")
folders.update("/".join(parts[:index]) for index in range(1, len(parts) + 1))
return sorted(folders, key=lambda value: (value.casefold(), value))
database/saved_views.py
+5
View File
@@ -33,6 +33,8 @@ MAX_TAG_LENGTH = 50
_IMAGE_SORTS = ("newest", "oldest", "name", "name-desc", "rating")
_LIBRARY_SORTS = ("newest", "oldest", "name", "rating")
_MEDIA = ("all", "image", "video")
# How a Library view's Tags match (#444): every tag or any of them.
_TAG_MODES = ("all", "any")
_CONTROL = re.compile(r"[\x00-\x1f\x7f]")
_SOURCE = re.compile(r"all|0|[1-9][0-9]{0,8}")
# A collection id as the listing's ``collection`` query takes it (#331): none
@@ -67,6 +69,9 @@ _STATE_FIELDS = {
"category": ("text", 100, ""),
"sort": ("choice", _LIBRARY_SORTS, "newest"),
"tags": ("tags", None, ()),
# #444: the Tags match and the output folder (relative, "/"-joined).
"tag_mode": ("choice", _TAG_MODES, "all"),
"folder": ("text", 1024, ""),
},
}