Files
ComfyAssets-ComfyUI_PromptM…/py/config.py
T
Vito 1b8fbdbb75 feat(annotations): image tag routes, model gate and scored suggestions (#278)
* 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).
2026-10-22 10:36:58 -07:00

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}")