* feat(annotations): add scoped observe, capture preview and tag save routes
POST /prompt_manager/image-annotations/observe captures the file named by
the listing identity {root_index, root_scope, path} (scope checked against
the current roots before any file access) and answers the saved tag set of
exactly those bytes, refreshing a matching record's verified fingerprint.
GET/DELETE .../capture/{token} serve the private copy (no-store, nosniff,
pinned with acquire/release_use while it streams) and release it.
PUT .../tags re-hashes the current file and writes only when it still
holds the captured bytes, under the storage epoch and tag revision
(file-changed, storage-changed, tag-conflict, tag-limit, replayed).
Bodies are capped at 64 KiB with unknown fields refused; envelopes never
carry server paths or locators. Restores invalidate every capture and the
host's cleanup removes the capture directory.
* feat(autotag): scored suggestions with a configurable count behind a shared model gate
- py/model_gate.py: one process-wide, non-blocking gate for loading,
swapping, running, downloading and unloading a tagging model. Worker
calls run shielded; a release requested while one still runs takes
effect only when it exits (client disconnect keeps the gate).
- Legacy /autotag/single, /start, /download/{type} and /unload hold the
gate and answer 409 {code: "model-busy"} while another operation holds
it; their bodies are otherwise unchanged. /autotag/models gains
suggestion_limit.
- AutoTagService.generate_tag_suggestions: WD14 (name, probability) best
first, GGUF/HF (name, None) in output order, first `limit` distinct
case-insensitively; generate_tags keeps its return; the image file is
closed before inference returns.
- POST /prompt_manager/image-annotations/suggest runs one engine on the
pinned capture, with strict per-engine options, never downloads
(409 model-unavailable) and never writes.
- config.json autotag.defaults.suggestion_limit, integer 1..100, default
10, validated on load.
- Capture probe decides Generate from the image header alone, so prompt
metadata over the inspection budgets no longer blocks it.
* feat(images): match image tags in the Tags filter, search and inspector
The workspace Images listing now admits a file under a Tags filter when
every requested tag is carried by its linked positive or its current image
tag set, and q also matches image tag names (case-folded substring). Image
tags count only while a byte record is verified at the file's present
(size, mtime_ns) (spec amendment S1): the tagged records are streamed
once, only listed files carrying a filter tag or q hit are stat'ed, and
their linked positives are read through the file_path index so tags,
category, rating and the positive count combine.
The rooted inspector envelope gains a validated image_tags facet
(current / none / unavailable) read at the inspection's own stable stat.
A failed image-tag read fails a Tags filter like a failed link read and
degrades a q search to no image tags; plain listings read none. Legacy
gallery and search responses are unchanged.
* fix(annotations): answer a fixed coded message when the filtered image list fails
A failed prompt-link or image-tag read under a Tags, category or rating
filter returned the raw exception text (SQL or database messages) in the
500 body. The workspace listing now answers code listing-failed with a
fixed sentence and logs the traceback. The legacy output listing and the
inspector image-tag read now log with exc_info as well.
* fix(annotations): drop suggestion names that Save would reject
A JoyCaption part over 50 characters or holding a newline came back as a
preselected chip, so the first Save of the dialog failed with a 400 and
nothing was stored. generate_tag_suggestions now applies the vocabulary
tag-name rules before the case-insensitive dedupe and the limit count, so
the requested number of valid chips still comes back.
* fix(annotations): stop changing the process warnings filters during capture
header_dimensions wrapped Image.open in warnings.catch_warnings with an
error filter for DecompressionBombWarning. catch_warnings is not
thread-safe, so two overlapping captures could leave that error filter
installed for every later Pillow open in the ComfyUI process. Only
Pillow's DecompressionBombError is now mapped to HeaderTooLarge; a size
that only warns is refused by the existing inference pixel cap.
* fix(annotations): free pins on cancellation, serve preview ranges, code every storage error
- A preview cancelled while its file opens closes the handle and only
then ends the pin, so a doomed capture is never unlinked while open.
- An observe cancelled while capturing or reading releases the capture
instead of leaving it until the idle expiry.
- The model gate takes an on_skipped callback that runs once when the
work never starts; Generate passes the capture's release_use, so an
executor refusal no longer leaks a pin against the capture budget.
- The preview answers a single byte range with 206 and Content-Range
(416 for an unsatisfiable or multi-range header), still streamed inside
the pin, so Safari and WebKit can play video captures.
- _coded maps tag-limit (400), metadata-conflict and vocabulary-conflict,
and falls back to 409 with a fixed sentence for any other code instead
of raising KeyError.
* fix(annotations): deny case and Unicode spellings of a listed file
On macOS (case- and normalization-insensitive APFS) a crafted target such
as A.PNG or an NFD name opened the listed a.png or NFC file while
realpath kept the requested spelling, so the tags were stored under a
second locator the Images filter never shows. Capture and verify now
require each path part to be spelled as its directory lists it and deny
any other case or Unicode spelling as an alias (409 source-changed, like
a symlink alias). A missing file with no other spelling keeps its usual
outcome.
* test(annotations): one shared folder_paths stub for the API test modules
test_api_images and test_api_prompts installed a MagicMock folder_paths,
so whenever one of them was imported before test_api_autotag the later
patch("folder_paths.get_output_directory") in test_api_history_outputs
left a mock attribute behind and two test_diagnostics_config tests
failed. tests/host_stubs.py now installs one real-module stub, used by
test_api_images, test_api_prompts and test_api_autotag (and through
test_api_images by the annotation modules, whose copy is gone), so the
modules pass in either import order.
* fix(annotations): derive the suggestion cap from the image tag set cap
py/autotag.py now takes MAX_SUGGESTION_LIMIT from database.annotations
MAX_IMAGE_TAGS. The two copies that cannot import it (py/config.py loads
before the database, utils/inspection_response.py stays storage-free) are
pinned equal by a test, so the 100 limits cannot drift apart.
* docs(testing): record capture aliases, preview ranges and the PR B follow-ups
TESTING.md now states that case and Unicode spellings are aliases, that
the preview serves byte ranges inside the capture pin, where the
cancellation tests live, and two tracked follow-ups: the legacy batch's
ungated cleanup unload under its own last inference, and the three files
over the 800-line ceiling (split py/autotag.py first).
772 lines
30 KiB
Python
772 lines
30 KiB
Python
"""Configuration module for ComfyUI PromptManager.
|
|
|
|
This module provides centralized configuration management for the PromptManager
|
|
extension, including gallery monitoring settings, database configuration,
|
|
web interface options, and performance tuning parameters.
|
|
|
|
The configuration is organized into two main classes:
|
|
- GalleryConfig: Settings for image monitoring and gallery functionality
|
|
- PromptManagerConfig: General settings for the PromptManager core features
|
|
|
|
Configuration can be loaded from and saved to JSON files for persistence.
|
|
|
|
Example:
|
|
from config import PromptManagerConfig
|
|
config = PromptManagerConfig.get_config()
|
|
PromptManagerConfig.load_from_file('custom_config.json')
|
|
"""
|
|
|
|
# PromptManager/py/config.py
|
|
|
|
# Extension configuration
|
|
extension_name = "PromptManager"
|
|
|
|
# Get server instance and routes (same pattern as ComfyUI_Assets)
|
|
from server import PromptServer # noqa: E402
|
|
|
|
server_instance = PromptServer.instance
|
|
routes = server_instance.routes
|
|
|
|
# Extension info
|
|
extension_uri = None # Will be set in __init__.py
|
|
|
|
import os # noqa: E402
|
|
from pathlib import Path # noqa: E402
|
|
from typing import Any, Dict, List, Optional, Tuple # noqa: E402
|
|
|
|
# Import logging system
|
|
try:
|
|
from ..utils.logging_config import get_logger
|
|
except ImportError:
|
|
import sys
|
|
|
|
current_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
sys.path.insert(0, current_dir)
|
|
from utils.logging_config import get_logger
|
|
|
|
# Initialize logger for config operations
|
|
config_logger = get_logger("prompt_manager.config")
|
|
|
|
# Environment variable listing extra directories (os.pathsep-separated) that
|
|
# may be used as gallery roots in addition to ComfyUI's own directories.
|
|
EXTRA_GALLERY_ROOTS_ENV = "PROMPT_MANAGER_EXTRA_GALLERY_ROOTS"
|
|
|
|
# Environment variable overriding where config.json is read from and saved to.
|
|
CONFIG_PATH_ENV = "PROMPT_MANAGER_CONFIG_PATH"
|
|
|
|
# ComfyUI folder_paths getters whose directories are valid gallery parents.
|
|
_COMFYUI_DIRECTORY_GETTERS = (
|
|
"get_output_directory",
|
|
"get_input_directory",
|
|
"get_temp_directory",
|
|
"get_user_directory",
|
|
)
|
|
|
|
|
|
def _canonical_path(path: str) -> str:
|
|
"""Return a path normalised for comparison (realpath + normcase)."""
|
|
return os.path.normcase(os.path.realpath(path))
|
|
|
|
|
|
def write_private_json(path: str, data: Any) -> None:
|
|
"""Write ``data`` as JSON to ``path`` readable by the current user only.
|
|
|
|
config.json can hold an API key (integrations.lora_manager), so it is
|
|
created with mode 0600 and, on POSIX, an existing more open file is
|
|
tightened too. Windows has no POSIX modes; there the chmod is a best
|
|
effort that only affects the read-only bit. Parent directories are
|
|
created as needed. Raises OSError when the file cannot be written.
|
|
"""
|
|
import json
|
|
|
|
parent_dir = os.path.dirname(path)
|
|
if parent_dir:
|
|
os.makedirs(parent_dir, exist_ok=True)
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as f:
|
|
json.dump(data, f, indent=2)
|
|
if os.name == "posix":
|
|
os.chmod(path, 0o600)
|
|
|
|
|
|
def _is_filesystem_root(path: str) -> bool:
|
|
"""True for '/' on POSIX and drive roots such as 'C:\\' on Windows."""
|
|
return os.path.dirname(path) == path
|
|
|
|
|
|
def _import_folder_paths():
|
|
"""ComfyUI's folder_paths module, or None outside ComfyUI."""
|
|
try:
|
|
import folder_paths
|
|
except ImportError:
|
|
return None
|
|
return folder_paths
|
|
|
|
|
|
def _comfyui_base_directory() -> Optional[str]:
|
|
"""ComfyUI's base directory: folder_paths.base_path, else the output dir's
|
|
parent."""
|
|
folder_paths = _import_folder_paths()
|
|
if folder_paths is None:
|
|
return None
|
|
base = getattr(folder_paths, "base_path", None)
|
|
if isinstance(base, str) and base:
|
|
return base
|
|
getter = getattr(folder_paths, "get_output_directory", None)
|
|
if getter is None:
|
|
return None
|
|
try:
|
|
output_dir = getter()
|
|
except Exception:
|
|
return None
|
|
if isinstance(output_dir, str) and output_dir:
|
|
return os.path.dirname(os.path.realpath(output_dir))
|
|
return None
|
|
|
|
|
|
def _comfyui_directories() -> List[str]:
|
|
"""Directories reported by ComfyUI's folder_paths module, if importable."""
|
|
folder_paths = _import_folder_paths()
|
|
if folder_paths is None:
|
|
return []
|
|
|
|
found = []
|
|
for getter_name in _COMFYUI_DIRECTORY_GETTERS:
|
|
getter = getattr(folder_paths, getter_name, None)
|
|
if getter is None:
|
|
continue
|
|
try:
|
|
directory = getter()
|
|
except Exception:
|
|
continue
|
|
if isinstance(directory, str) and directory:
|
|
found.append(directory)
|
|
return found
|
|
|
|
|
|
def _extra_gallery_roots() -> List[str]:
|
|
"""Directories the user opted in through PROMPT_MANAGER_EXTRA_GALLERY_ROOTS."""
|
|
raw = os.environ.get(EXTRA_GALLERY_ROOTS_ENV, "")
|
|
return [entry.strip() for entry in raw.split(os.pathsep) if entry.strip()]
|
|
|
|
|
|
class GalleryConfig:
|
|
"""Configuration class for the gallery monitoring and image processing system.
|
|
|
|
This class manages all settings related to automatic image monitoring,
|
|
prompt tracking, database cleanup, web interface display, and performance
|
|
optimization for the gallery functionality.
|
|
|
|
All configuration values are class attributes that can be modified at runtime
|
|
or loaded from external configuration files.
|
|
|
|
Attributes:
|
|
MONITORING_ENABLED (bool): Enable/disable automatic image monitoring
|
|
MONITORING_DIRECTORIES (List[str]): Directories to monitor for new images
|
|
SUPPORTED_EXTENSIONS (List[str]): Image file extensions to process
|
|
PROCESSING_DELAY (float): Delay in seconds before processing new files
|
|
PROMPT_TIMEOUT (int): Seconds to keep prompt context active
|
|
CLEANUP_INTERVAL (int): Seconds between cleanup of expired prompts
|
|
AUTO_CLEANUP_MISSING_FILES (bool): Automatically remove missing file records
|
|
MAX_IMAGE_AGE_DAYS (int): Maximum age in days before cleaning up images
|
|
IMAGES_PER_PAGE (int): Number of images to display per page in web UI
|
|
THUMBNAIL_SIZE (int): Size in pixels for generated thumbnails
|
|
ENABLE_SEARCH (bool): Enable search functionality in web interface
|
|
ENABLE_METADATA_VIEW (bool): Enable metadata viewing for images
|
|
MAX_CONCURRENT_PROCESSING (int): Maximum concurrent image processing tasks
|
|
METADATA_EXTRACTION_TIMEOUT (int): Timeout for metadata extraction operations
|
|
"""
|
|
|
|
# Image monitoring settings
|
|
MONITORING_ENABLED = True
|
|
MONITORING_DIRECTORIES = [] # Auto-detect if empty
|
|
SUPPORTED_EXTENSIONS = [".png", ".jpg", ".jpeg", ".webp", ".gif"]
|
|
PROCESSING_DELAY = 2.0 # Seconds to wait before processing new files
|
|
|
|
# Prompt tracking settings
|
|
PROMPT_TIMEOUT = (
|
|
600 # Seconds to keep prompt context active (10 min for long generations)
|
|
)
|
|
CLEANUP_INTERVAL = 300 # Seconds between cleanup of expired prompts
|
|
|
|
# Database settings
|
|
AUTO_CLEANUP_MISSING_FILES = True
|
|
MAX_IMAGE_AGE_DAYS = 365 # Clean up images older than this
|
|
|
|
# Web interface settings
|
|
IMAGES_PER_PAGE = 20
|
|
THUMBNAIL_SIZE = 256
|
|
ENABLE_SEARCH = True
|
|
ENABLE_METADATA_VIEW = True
|
|
|
|
# Performance settings
|
|
MAX_CONCURRENT_PROCESSING = 3
|
|
METADATA_EXTRACTION_TIMEOUT = 10 # Seconds
|
|
|
|
@classmethod
|
|
def get_config(cls) -> Dict[str, Any]:
|
|
"""Get the complete gallery configuration as a structured dictionary.
|
|
|
|
Returns:
|
|
Dict[str, Any]: A nested dictionary containing all gallery configuration
|
|
sections: monitoring, tracking, database, web_interface, and
|
|
performance. Each section contains the relevant configuration
|
|
parameters as key-value pairs.
|
|
|
|
Example:
|
|
config = GalleryConfig.get_config()
|
|
monitoring_enabled = config['monitoring']['enabled']
|
|
images_per_page = config['web_interface']['images_per_page']
|
|
"""
|
|
return {
|
|
"monitoring": {
|
|
"enabled": cls.MONITORING_ENABLED,
|
|
"directories": cls.MONITORING_DIRECTORIES,
|
|
"extensions": cls.SUPPORTED_EXTENSIONS,
|
|
"processing_delay": cls.PROCESSING_DELAY,
|
|
},
|
|
"tracking": {
|
|
"prompt_timeout": cls.PROMPT_TIMEOUT,
|
|
"cleanup_interval": cls.CLEANUP_INTERVAL,
|
|
},
|
|
"database": {
|
|
"auto_cleanup": cls.AUTO_CLEANUP_MISSING_FILES,
|
|
"max_image_age_days": cls.MAX_IMAGE_AGE_DAYS,
|
|
},
|
|
"web_interface": {
|
|
"images_per_page": cls.IMAGES_PER_PAGE,
|
|
"thumbnail_size": cls.THUMBNAIL_SIZE,
|
|
"enable_search": cls.ENABLE_SEARCH,
|
|
"enable_metadata_view": cls.ENABLE_METADATA_VIEW,
|
|
},
|
|
"performance": {
|
|
"max_concurrent_processing": cls.MAX_CONCURRENT_PROCESSING,
|
|
"metadata_extraction_timeout": cls.METADATA_EXTRACTION_TIMEOUT,
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def update_config(cls, new_config: Dict[str, Any]):
|
|
"""Update gallery configuration attributes from a dictionary.
|
|
|
|
Takes a nested dictionary with gallery configuration sections and updates
|
|
the corresponding class attributes. Only updates attributes that are
|
|
present in the input dictionary, leaving others unchanged.
|
|
|
|
Args:
|
|
new_config (Dict[str, Any]): Nested dictionary containing gallery
|
|
configuration updates. Should follow the same structure as returned
|
|
by get_config(). Valid top-level keys are: 'monitoring', 'tracking',
|
|
'database', 'web_interface', 'performance'.
|
|
|
|
Example:
|
|
gallery_settings = {
|
|
'monitoring': {'enabled': False},
|
|
'web_interface': {'images_per_page': 50}
|
|
}
|
|
GalleryConfig.update_config(gallery_settings)
|
|
"""
|
|
monitoring = new_config.get("monitoring", {})
|
|
if "enabled" in monitoring:
|
|
cls.MONITORING_ENABLED = monitoring["enabled"]
|
|
if "directories" in monitoring:
|
|
directories = cls._validated_directories(monitoring["directories"])
|
|
if directories is not None:
|
|
cls.MONITORING_DIRECTORIES = directories
|
|
if "extensions" in monitoring:
|
|
cls.SUPPORTED_EXTENSIONS = monitoring["extensions"]
|
|
if "processing_delay" in monitoring:
|
|
cls.PROCESSING_DELAY = monitoring["processing_delay"]
|
|
|
|
tracking = new_config.get("tracking", {})
|
|
if "prompt_timeout" in tracking:
|
|
cls.PROMPT_TIMEOUT = tracking["prompt_timeout"]
|
|
if "cleanup_interval" in tracking:
|
|
cls.CLEANUP_INTERVAL = tracking["cleanup_interval"]
|
|
|
|
database = new_config.get("database", {})
|
|
if "auto_cleanup" in database:
|
|
cls.AUTO_CLEANUP_MISSING_FILES = database["auto_cleanup"]
|
|
if "max_image_age_days" in database:
|
|
cls.MAX_IMAGE_AGE_DAYS = database["max_image_age_days"]
|
|
|
|
web_interface = new_config.get("web_interface", {})
|
|
if "images_per_page" in web_interface:
|
|
cls.IMAGES_PER_PAGE = web_interface["images_per_page"]
|
|
if "thumbnail_size" in web_interface:
|
|
cls.THUMBNAIL_SIZE = web_interface["thumbnail_size"]
|
|
if "enable_search" in web_interface:
|
|
cls.ENABLE_SEARCH = web_interface["enable_search"]
|
|
if "enable_metadata_view" in web_interface:
|
|
cls.ENABLE_METADATA_VIEW = web_interface["enable_metadata_view"]
|
|
|
|
performance = new_config.get("performance", {})
|
|
if "max_concurrent_processing" in performance:
|
|
cls.MAX_CONCURRENT_PROCESSING = performance["max_concurrent_processing"]
|
|
if "metadata_extraction_timeout" in performance:
|
|
cls.METADATA_EXTRACTION_TIMEOUT = performance["metadata_extraction_timeout"]
|
|
|
|
@classmethod
|
|
def _validated_directories(cls, directories: Any) -> Optional[List[str]]:
|
|
"""Gallery roots from a config payload that pass validate_gallery_root.
|
|
|
|
A hand-edited config.json is as untrusted as a settings request:
|
|
entries that fail validation are dropped with a warning (naming only
|
|
the basename), and a value that is not a list is ignored entirely
|
|
(``None`` is returned so the current roots stay as they are).
|
|
"""
|
|
if not isinstance(directories, list):
|
|
config_logger.warning(
|
|
"Ignoring gallery monitoring directories: expected a list"
|
|
)
|
|
return None
|
|
kept = []
|
|
for entry in directories:
|
|
ok, reason = cls.validate_gallery_root(entry)
|
|
if ok:
|
|
kept.append(entry)
|
|
continue
|
|
label = os.path.basename(entry) if isinstance(entry, str) else entry
|
|
config_logger.warning(f"Ignoring gallery root {label!r}: {reason}")
|
|
return kept
|
|
|
|
@classmethod
|
|
def allowed_gallery_parents(cls) -> List[str]:
|
|
"""Canonical directories under which gallery roots may live.
|
|
|
|
Filesystem roots are never allowed as parents, even when listed in
|
|
the environment, because that would re-open the whole disk.
|
|
"""
|
|
parents = []
|
|
for candidate in _comfyui_directories() + _extra_gallery_roots():
|
|
try:
|
|
canonical = _canonical_path(candidate)
|
|
except (OSError, ValueError):
|
|
continue
|
|
if _is_filesystem_root(canonical) or not os.path.isdir(canonical):
|
|
continue
|
|
if canonical not in parents:
|
|
parents.append(canonical)
|
|
return parents
|
|
|
|
@classmethod
|
|
def path_anchors(cls) -> List[str]:
|
|
"""Canonical directories that relative gallery paths are resolved against.
|
|
|
|
The ComfyUI base directory comes first, followed by the parent of each
|
|
directory listed in PROMPT_MANAGER_EXTRA_GALLERY_ROOTS, so an extra
|
|
root is addressed by its own name (``gallery/sub``) rather than by an
|
|
absolute path. The same anchors drive the public (relative) form of
|
|
paths in API responses.
|
|
"""
|
|
anchors = []
|
|
base = _comfyui_base_directory()
|
|
candidates = [base] if base else []
|
|
candidates.extend(os.path.dirname(root) for root in _extra_gallery_roots())
|
|
for candidate in candidates:
|
|
try:
|
|
canonical = _canonical_path(candidate)
|
|
except (OSError, ValueError):
|
|
continue
|
|
if canonical not in anchors:
|
|
anchors.append(canonical)
|
|
return anchors
|
|
|
|
@classmethod
|
|
def resolve_gallery_root(cls, path: str) -> str:
|
|
"""Canonical absolute path for a gallery root given in any accepted form.
|
|
|
|
Absolute paths are canonicalised as-is. Relative paths are tried
|
|
against each :meth:`path_anchors` entry and the first existing match
|
|
wins; otherwise the first anchor (or the current directory when there
|
|
is none) is used, so the caller's existence check reports it.
|
|
"""
|
|
path = path.strip()
|
|
if os.path.isabs(path):
|
|
return _canonical_path(path)
|
|
anchors = cls.path_anchors()
|
|
for anchor in anchors:
|
|
candidate = os.path.join(anchor, path)
|
|
if os.path.exists(candidate):
|
|
return _canonical_path(candidate)
|
|
return _canonical_path(os.path.join(anchors[0], path) if anchors else path)
|
|
|
|
@classmethod
|
|
def validate_gallery_root(cls, path: Any) -> Tuple[bool, str]:
|
|
"""Check whether ``path`` may be used as a gallery root.
|
|
|
|
Returns:
|
|
(True, "") when the directory lies inside one of
|
|
:meth:`allowed_gallery_parents`; otherwise (False, reason).
|
|
"""
|
|
if not isinstance(path, str) or not path.strip():
|
|
return False, "Gallery root must be a non-empty path"
|
|
|
|
try:
|
|
canonical = cls.resolve_gallery_root(path)
|
|
except (OSError, ValueError):
|
|
return False, "Gallery root could not be resolved"
|
|
|
|
if _is_filesystem_root(canonical):
|
|
return False, "Gallery root cannot be a filesystem root"
|
|
if canonical == _canonical_path(os.path.expanduser("~")):
|
|
return False, "Gallery root cannot be the home directory"
|
|
if not os.path.exists(canonical):
|
|
return False, "Gallery root does not exist"
|
|
if not os.path.isdir(canonical):
|
|
return False, "Gallery root is not a directory"
|
|
|
|
candidate = Path(canonical)
|
|
for parent in cls.allowed_gallery_parents():
|
|
if candidate.is_relative_to(Path(parent)):
|
|
return True, ""
|
|
|
|
return (
|
|
False,
|
|
"Gallery root must be inside a ComfyUI directory "
|
|
f"(output, input, temp, user) or one listed in {EXTRA_GALLERY_ROOTS_ENV}",
|
|
)
|
|
|
|
|
|
class IntegrationConfig:
|
|
"""Configuration for third-party extension integrations.
|
|
|
|
Manages opt-in integration settings for extensions like LoraManager.
|
|
All integrations are disabled by default so PromptManager works standalone.
|
|
"""
|
|
|
|
# LoraManager integration
|
|
LORA_MANAGER_ENABLED = False
|
|
LORA_MANAGER_PATH = "" # Auto-detected if empty
|
|
LORA_TRIGGER_WORDS_ENABLED = False # Auto-inject trigger words into prompts
|
|
CIVITAI_API_KEY = "" # Required to download NSFW example images
|
|
|
|
@classmethod
|
|
def get_config(cls) -> Dict[str, Any]:
|
|
return {
|
|
"lora_manager": {
|
|
"enabled": cls.LORA_MANAGER_ENABLED,
|
|
"path": cls.LORA_MANAGER_PATH,
|
|
"trigger_words_enabled": cls.LORA_TRIGGER_WORDS_ENABLED,
|
|
"civitai_api_key": cls.CIVITAI_API_KEY,
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def update_config(cls, new_config: Dict[str, Any]):
|
|
lora = new_config.get("lora_manager", {})
|
|
if "enabled" in lora:
|
|
cls.LORA_MANAGER_ENABLED = lora["enabled"]
|
|
if "path" in lora:
|
|
cls.LORA_MANAGER_PATH = lora["path"]
|
|
if "trigger_words_enabled" in lora:
|
|
cls.LORA_TRIGGER_WORDS_ENABLED = lora["trigger_words_enabled"]
|
|
if "civitai_api_key" in lora:
|
|
cls.CIVITAI_API_KEY = lora["civitai_api_key"]
|
|
|
|
|
|
# Number of tag suggestions the Tag dialogs request by default (D68). The cap
|
|
# is the image tag set limit: more suggestions could never all be kept.
|
|
DEFAULT_SUGGESTION_LIMIT = 10
|
|
MAX_SUGGESTION_LIMIT = 100 # pinned to MAX_IMAGE_TAGS by tests/test_config.py
|
|
|
|
|
|
class AutotagConfig:
|
|
"""Persisted tagging defaults under ``autotag.defaults`` in config.json.
|
|
|
|
``suggestion_limit`` is how many suggestions Generate returns (the top N
|
|
by score), an integer in ``1..MAX_SUGGESTION_LIMIT``. A hand-edited value
|
|
outside that range, or of another type, falls back to the default with a
|
|
warning rather than failing the load.
|
|
"""
|
|
|
|
SUGGESTION_LIMIT = DEFAULT_SUGGESTION_LIMIT
|
|
|
|
@classmethod
|
|
def suggestion_limit(cls) -> int:
|
|
return cls.SUGGESTION_LIMIT
|
|
|
|
@staticmethod
|
|
def valid_suggestion_limit(value: Any) -> bool:
|
|
return (
|
|
type(value) is int and 1 <= value <= MAX_SUGGESTION_LIMIT
|
|
) # bool, float and str are rejected
|
|
|
|
@classmethod
|
|
def get_config(cls) -> Dict[str, Any]:
|
|
return {"defaults": {"suggestion_limit": cls.SUGGESTION_LIMIT}}
|
|
|
|
@classmethod
|
|
def update_config(cls, new_config: Any) -> None:
|
|
if not isinstance(new_config, dict):
|
|
config_logger.warning("Ignoring autotag settings that are not an object")
|
|
return
|
|
defaults = new_config.get("defaults", {})
|
|
if not isinstance(defaults, dict):
|
|
config_logger.warning("Ignoring autotag.defaults that is not an object")
|
|
return
|
|
if "suggestion_limit" not in defaults:
|
|
return
|
|
value = defaults["suggestion_limit"]
|
|
if cls.valid_suggestion_limit(value):
|
|
cls.SUGGESTION_LIMIT = value
|
|
return
|
|
config_logger.warning(
|
|
"autotag.defaults.suggestion_limit must be an integer from 1 to "
|
|
f"{MAX_SUGGESTION_LIMIT}; using {DEFAULT_SUGGESTION_LIMIT}"
|
|
)
|
|
cls.SUGGESTION_LIMIT = DEFAULT_SUGGESTION_LIMIT
|
|
|
|
|
|
class PromptManagerConfig:
|
|
"""Main configuration class for PromptManager core functionality.
|
|
|
|
This class manages configuration for database operations, web UI behavior,
|
|
performance settings, and integrates gallery configuration. It provides
|
|
methods for loading and saving configuration from/to JSON files.
|
|
|
|
The configuration is organized into logical sections:
|
|
- Database: Settings for SQLite operations and data management
|
|
- Web UI: User interface behavior and display options
|
|
- Performance: Optimization and resource management settings
|
|
- Gallery: Embedded gallery configuration (via GalleryConfig)
|
|
|
|
Attributes:
|
|
DEFAULT_DB_PATH (str): Default path for the SQLite database file
|
|
ENABLE_DUPLICATE_DETECTION (bool): Enable automatic duplicate detection
|
|
ENABLE_AUTO_SAVE (bool): Enable automatic saving of prompts
|
|
RESULT_TIMEOUT (int): Auto-hide timeout for results in ComfyUI node
|
|
SHOW_TEST_BUTTON (bool): Show API test button in node interface
|
|
WEBUI_DISPLAY_MODE (str): Display mode for web UI ('popup' or 'newtab')
|
|
MAX_SEARCH_RESULTS (int): Maximum number of search results to return
|
|
ENABLE_FUZZY_SEARCH (bool): Enable fuzzy search capabilities
|
|
AUTO_BACKUP_INTERVAL (int): Hours between automatic database backups
|
|
"""
|
|
|
|
# Database settings
|
|
DEFAULT_DB_PATH = "prompts.db"
|
|
ENABLE_DUPLICATE_DETECTION = True
|
|
ENABLE_AUTO_SAVE = True
|
|
|
|
# Web UI settings
|
|
RESULT_TIMEOUT = 5 # Seconds to auto-hide results in ComfyUI node
|
|
SHOW_TEST_BUTTON = False # Show API test button in node UI
|
|
WEBUI_DISPLAY_MODE = "newtab" # 'popup' or 'newtab'
|
|
INFINITE_SCROLL = False # Dashboard list: load the next page on scroll by default
|
|
|
|
# Performance settings
|
|
MAX_SEARCH_RESULTS = 100
|
|
ENABLE_FUZZY_SEARCH = False # Requires fuzzywuzzy
|
|
AUTO_BACKUP_INTERVAL = 24 # Hours
|
|
# Threads for the output scan and thumbnail generation; half the cores by
|
|
# default so ComfyUI's own generation keeps CPU headroom. 1..max_worker_threads().
|
|
DEFAULT_WORKER_THREADS = max(1, (os.cpu_count() or 1) // 2)
|
|
WORKER_THREADS = DEFAULT_WORKER_THREADS
|
|
|
|
@classmethod
|
|
def max_worker_threads(cls) -> int:
|
|
"""Largest accepted worker-thread count: the detected core count."""
|
|
return os.cpu_count() or 1
|
|
|
|
@classmethod
|
|
def clamp_worker_threads(cls, value, fallback: int) -> int:
|
|
"""Coerce a worker count from config.json into ``[1, max_worker_threads()]``.
|
|
|
|
Non-numeric values (and booleans) return ``fallback`` so a hand-edited
|
|
file cannot disable the setting.
|
|
"""
|
|
if isinstance(value, bool):
|
|
return fallback
|
|
if isinstance(value, str) and value.strip().isdigit():
|
|
value = int(value.strip())
|
|
if not isinstance(value, int):
|
|
return fallback
|
|
return max(1, min(value, cls.max_worker_threads()))
|
|
|
|
@classmethod
|
|
def get_config(cls) -> Dict[str, Any]:
|
|
"""Get the complete PromptManager configuration as a structured dictionary.
|
|
|
|
Returns:
|
|
Dict[str, Any]: A nested dictionary containing all configuration sections:
|
|
- database: Database-related settings
|
|
- web_ui: Web interface configuration
|
|
- performance: Performance and optimization settings
|
|
- gallery: Complete gallery configuration (from GalleryConfig)
|
|
|
|
Example:
|
|
config = PromptManagerConfig.get_config()
|
|
db_path = config['database']['default_path']
|
|
max_results = config['performance']['max_search_results']
|
|
"""
|
|
return {
|
|
"database": {
|
|
"default_path": cls.DEFAULT_DB_PATH,
|
|
"enable_duplicate_detection": cls.ENABLE_DUPLICATE_DETECTION,
|
|
"enable_auto_save": cls.ENABLE_AUTO_SAVE,
|
|
},
|
|
"web_ui": {
|
|
"result_timeout": cls.RESULT_TIMEOUT,
|
|
"show_test_button": cls.SHOW_TEST_BUTTON,
|
|
"webui_display_mode": cls.WEBUI_DISPLAY_MODE,
|
|
"infinite_scroll": cls.INFINITE_SCROLL,
|
|
},
|
|
"performance": {
|
|
"max_search_results": cls.MAX_SEARCH_RESULTS,
|
|
"enable_fuzzy_search": cls.ENABLE_FUZZY_SEARCH,
|
|
"auto_backup_interval": cls.AUTO_BACKUP_INTERVAL,
|
|
"worker_threads": cls.WORKER_THREADS,
|
|
},
|
|
"gallery": GalleryConfig.get_config(),
|
|
"integrations": IntegrationConfig.get_config(),
|
|
"autotag": AutotagConfig.get_config(),
|
|
}
|
|
|
|
@classmethod
|
|
def get_config_path(cls) -> str:
|
|
"""Path of the persisted config.json.
|
|
|
|
Honours ``PROMPT_MANAGER_CONFIG_PATH`` when set; otherwise the file
|
|
lives at the repository root next to ``pyproject.toml``.
|
|
"""
|
|
override = os.environ.get(CONFIG_PATH_ENV, "").strip()
|
|
if override:
|
|
return override
|
|
repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
return os.path.join(repo_root, "config.json")
|
|
|
|
@classmethod
|
|
def load_from_file(cls, config_path: Optional[str] = None):
|
|
"""Load configuration settings from a JSON file.
|
|
|
|
Reads configuration from the specified JSON file and updates the current
|
|
configuration attributes. If the file doesn't exist or contains invalid
|
|
JSON, logs an appropriate message and continues with default values.
|
|
|
|
Args:
|
|
config_path (str): Path to the JSON configuration file to load.
|
|
Can be relative or absolute path. Defaults to
|
|
:meth:`get_config_path`.
|
|
|
|
Raises:
|
|
The method handles all exceptions internally and logs errors rather
|
|
than propagating them, ensuring the system continues with defaults.
|
|
|
|
Example:
|
|
PromptManagerConfig.load_from_file('custom_config.json')
|
|
PromptManagerConfig.load_from_file('/path/to/config.json')
|
|
"""
|
|
import json
|
|
|
|
if config_path is None:
|
|
config_path = cls.get_config_path()
|
|
|
|
if os.path.exists(config_path):
|
|
try:
|
|
with open(config_path, "r") as f:
|
|
config = json.load(f)
|
|
cls.update_config(config)
|
|
config_logger.info(f"Loaded configuration from {config_path}")
|
|
except Exception as e:
|
|
config_logger.error(f"Error loading config from {config_path}: {e}")
|
|
else:
|
|
config_logger.info(f"Config file not found: {config_path}, using defaults")
|
|
|
|
@classmethod
|
|
def save_to_file(cls, config_path: Optional[str] = None):
|
|
"""Save the current configuration to a JSON file.
|
|
|
|
Serializes the complete configuration (including gallery settings) to
|
|
a JSON file. Creates the directory structure if it doesn't exist.
|
|
|
|
Args:
|
|
config_path (str): Path where the JSON configuration file should be saved.
|
|
Parent directories will be created if they don't exist.
|
|
Defaults to :meth:`get_config_path`.
|
|
|
|
Raises:
|
|
The method handles all exceptions internally and logs errors rather
|
|
than propagating them.
|
|
|
|
Example:
|
|
PromptManagerConfig.save_to_file('backup_config.json')
|
|
PromptManagerConfig.save_to_file('/etc/comfyui/prompt_manager.json')
|
|
"""
|
|
if config_path is None:
|
|
config_path = cls.get_config_path()
|
|
|
|
try:
|
|
write_private_json(config_path, cls.get_config())
|
|
config_logger.info(f"Saved configuration to {config_path}")
|
|
except Exception as e:
|
|
config_logger.error(f"Error saving config to {config_path}: {e}")
|
|
|
|
@classmethod
|
|
def update_config(cls, new_config: Dict[str, Any]):
|
|
"""Update configuration attributes from a dictionary.
|
|
|
|
Takes a nested dictionary with configuration sections and updates
|
|
the corresponding class attributes. Only updates attributes that
|
|
are present in the input dictionary, leaving others unchanged.
|
|
|
|
Args:
|
|
new_config (Dict[str, Any]): Nested dictionary containing configuration
|
|
updates. Should follow the same structure as returned by get_config().
|
|
Valid top-level keys are: 'database', 'web_ui', 'performance',
|
|
'gallery'.
|
|
|
|
Example:
|
|
new_settings = {
|
|
'database': {'default_path': 'custom.db'},
|
|
'performance': {'max_search_results': 50}
|
|
}
|
|
PromptManagerConfig.update_config(new_settings)
|
|
"""
|
|
database = new_config.get("database", {})
|
|
if "default_path" in database:
|
|
cls.DEFAULT_DB_PATH = database["default_path"]
|
|
if "enable_duplicate_detection" in database:
|
|
cls.ENABLE_DUPLICATE_DETECTION = database["enable_duplicate_detection"]
|
|
if "enable_auto_save" in database:
|
|
cls.ENABLE_AUTO_SAVE = database["enable_auto_save"]
|
|
|
|
web_ui = new_config.get("web_ui", {})
|
|
if "result_timeout" in web_ui:
|
|
cls.RESULT_TIMEOUT = web_ui["result_timeout"]
|
|
if "show_test_button" in web_ui:
|
|
cls.SHOW_TEST_BUTTON = web_ui["show_test_button"]
|
|
if "webui_display_mode" in web_ui:
|
|
cls.WEBUI_DISPLAY_MODE = web_ui["webui_display_mode"]
|
|
if isinstance(web_ui.get("infinite_scroll"), bool):
|
|
cls.INFINITE_SCROLL = web_ui["infinite_scroll"]
|
|
|
|
performance = new_config.get("performance", {})
|
|
if "max_search_results" in performance:
|
|
cls.MAX_SEARCH_RESULTS = performance["max_search_results"]
|
|
if "enable_fuzzy_search" in performance:
|
|
cls.ENABLE_FUZZY_SEARCH = performance["enable_fuzzy_search"]
|
|
if "auto_backup_interval" in performance:
|
|
cls.AUTO_BACKUP_INTERVAL = performance["auto_backup_interval"]
|
|
if "worker_threads" in performance:
|
|
cls.WORKER_THREADS = cls.clamp_worker_threads(
|
|
performance["worker_threads"], cls.WORKER_THREADS
|
|
)
|
|
|
|
# Update gallery config
|
|
if "gallery" in new_config:
|
|
GalleryConfig.update_config(new_config["gallery"])
|
|
|
|
# Update integration config
|
|
if "integrations" in new_config:
|
|
IntegrationConfig.update_config(new_config["integrations"])
|
|
|
|
if "autotag" in new_config:
|
|
AutotagConfig.update_config(new_config["autotag"])
|
|
|
|
|
|
# Load configuration on import
|
|
try:
|
|
PromptManagerConfig.load_from_file()
|
|
except Exception as e:
|
|
config_logger.error(f"Error during config initialization: {e}")
|