Files
ComfyAssets-ComfyUI_PromptM…/py/config.py
T
Vito f68e92b3b6 feat(system): maintenance backend (#313)
* feat(maintenance): reviewed, read-only maintenance plans

POST /prompt_manager/maintenance/preview reviews duplicate_prompts,
optimize, unused_components or missing_links on its own mode=ro snapshot
and answers a Plan (scope, protected reasons, backup policy, media, at most
50 detail rows). The fingerprint, targets and judged sources stay in a
server-side PlanStore under a single-use token (15 min, 8 plans, 16 MiB,
never truncated; a restart forgets them), bound to the database it read.

The previews reuse the legacy queries: duplicate groups and the merge are
shared with cleanup_duplicates, ORPHANED_PROMPTS_SQL is built from the
#239 protection predicates the plan reports, and missing links are judged
by the health check's link_states. Legacy /maintenance keeps its shape;
vacuum now reports success:false when VACUUM fails, and
cleanup_orphaned_images skips links under unavailable sources.

* feat(maintenance): apply reviewed plans as Activity jobs

POST /prompt_manager/maintenance/apply {token} starts a maintenance job
(records admission, not cancellable, groups changed/protected/errors).
The token is consumed on the loop turn the job starts: a second apply is
409 plan-used naming the job, a busy refusal keeps the token, an unknown
or expired one is 410 plan-expired, and a restore since the review is
409 storage-changed.

With the safety backup policy on (maintenance.safety_backup, default true,
saved through the settings route with the revision), the job first writes
a backup into a temp file beside the database, makes it a standalone
rollback-journal file, verifies it, fsyncs it and renames it onto a unique
<db>.backup_<YYYYmmdd_HHMMSS_ffffff>[_n] name; any failure leaves no file
and nothing changed. The apply then rechecks the reviewed fingerprint
inside BEGIN IMMEDIATE (plan-changed writes nothing), deletes in batches of
400 ids, counts changed and protected records, logs the backup by name
only and prunes old safety backups. With the policy off the log and the
status line say "No safety backup".

* feat(maintenance): find and remove duplicate files through Activity

Duplicate image files move onto the job registry:

- POST /prompt_manager/jobs/duplicates finds content-identical media
  files over every effective source as a cancellable read job
  ("Find duplicate image files", group groups_found); the server holds
  the groups per job and GET /prompt_manager/maintenance/duplicates/{id}
  pages them ten at a time.
- The duplicate_files preview binds a removal to the reviewed keepers.
  Every keeper and removal is checked again: containment as text before
  any filesystem access, then not a link, still inside its source, same
  size and hash. The plan lists the database links that move.
- Apply runs duplicate_cleanup (records, not cancellable): the promised
  safety backup, then per file a recheck, the unlink with every handle
  closed and its thumbnail. A changed file is kept, the first failed
  unlink stops the job with a coded message (never the OS text), and only
  removed files have their image links and favorites moved to the kept
  copy.
- The legacy scan, stream, status and delete routes keep their shapes on
  the same registry kinds; the hashing, grouping and thumbnail removal
  are shared with them.

Coordinator rulings on Tasks 15-16:

- Safety backups share the restore pool of 10 `<db>.backup_*` files; the
  pool is now also pruned when an apply fails after its backup.
- Plans report the safety backup's size and the free disk space, and
  apply refuses up front (507 insufficient-space) when the backup or
  VACUUM does not fit, so no job fails halfway for space.

* fix(maintenance): never remove the kept copy; move records before unlink

Final review of #313:

- C1: the duplicate find tells files apart by (st_dev, st_ino), so a
  source listed twice, a symlinked source inside another and a hardlink
  are one file. The preview and the job refuse a removal that
  os.path.samestat matches with its kept copy.
- I1: each file's records move to its verified kept copy in one short
  transaction before that file is unlinked; a failed move deletes
  nothing and a failed unlink leaves the records on the identical kept
  copy. The status line counts files removed and not removed.
- I2: links are looked up under every spelling of the source (the real
  path, the configured and monitored folders, ComfyUI's output).
- I5: image tags join the kept copy as their union and the removed
  copy's pending, undecided Needs review items are settled; the plan
  says so.
- I3: the VACUUM disk check measures SQLite's real temporary folder.
- M1: the token is looked at again with no await before the job starts.
- M2: the backup decision is made once and handed to the job.
- M3: backups are linked into place; startup sweeps interrupted temps.
- M4: Optimize runs alone (new "exclusive" admission) and its plan says
  the database is locked while VACUUM runs.
- M5: the reviewed duplicate merge is strict: collisions and Library
  titles are protected, and errors roll back. The legacy cleanup is
  unchanged.
- M7: 2,000 keep entries within a 1 MiB body. M8: event waits in tests.
- M9: the preview binds files by size, mtime and inode without hashing;
  the job hashes right before each unlink.

* fix(maintenance): protect one file reached through two mounts

Re-review of #313, round 2:

- R1: a removal on another device than its kept copy, with the same
  size and mtimes within 2 s, is protected when its inode is equal or its
  path within its source is the same ("May be the same file reached
  through another mount or share"), in the preview and again right before
  each unlink. The job holds the kept copy open during the unlink, so on
  Windows deleting the kept copy itself through another path fails.
- R2: a file's spellings are found by containment: every configured,
  monitored, ComfyUI-output or listed folder whose stat identity is a
  folder above the file. Parent spellings of nested sources and case
  variants match; favorites move under every listed root holding the
  file. Lookups stay index searches.
- m1: a failed unlink's row says the links now name the kept copy.
- m2: a volume with no file ids reads "Can't confirm this is a separate
  file on this drive".

* fix(maintenance): stat duplicate candidates with os.stat

Python 3.10's Path.stat uses the os.stat bound when pathlib was imported,
so the find and the per-file checks could see different stat calls. The
find now calls os.stat directly, like every other check; the mount tests
failed on py3.10 only.

* fix(maintenance): recheck the kept copy right before each unlink

Re-review 2 of #313:

- n1: the cross-mount guard also matches when one path within its source
  ends with the other on a whole folder name (sources listed at different
  depths, e.g. 2024/x.png and x.png on a FUSE or noserverino mount).
- n2: after a file's records move and right before its unlink, the held
  kept copy must still be the file at its path (fstat of the hold against
  os.stat) and the cross-mount guard must still hold; otherwise the file
  is kept, its records move back to it, and it counts as not removed.

* test(maintenance): the kept-copy race tests hold on Windows

On Windows the removal job holds the kept copy open without
FILE_SHARE_DELETE, so the tests' attempt to delete or replace it during
the record move is refused. The tests now assert that native outcome:
the kept copy stays and the removal goes ahead.

* fix(maintenance): restore a backup of any size the disk has room for

The legacy restore refused the maintainer's own 1.6 GB safety backup with
"File too large. Maximum size is 100MB", a fixed cap since #153. Free
disk space now decides: the upload, its staged copy and the swap must fit,
under a 64 GiB ceiling. A request with a Content-Length that does not fit
is refused before anything is written, one without as soon as its running
size does not, with 507 insufficient-space and a message without paths.
The upload still streams and its temp file is removed on every path.
2026-10-24 05:28:44 -07:00

879 lines
36 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""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 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"
# The one answer for every path outside the allowed folders. The filesystem
# root and the home directory answer it too, so no caller learns the server's
# layout (home path, user name) from a refusal (#249 S4).
OUTSIDE_ALLOWED_REASON = (
"Gallery root must be inside a ComfyUI directory (output, input, temp, user) "
f"or one listed in {EXTRA_GALLERY_ROOTS_ENV}"
)
# 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 _lexical_path(path: str) -> str:
"""``path`` normalised for comparison without touching the filesystem.
normpath collapses ``.`` and ``..`` as text and normcase folds case on
Windows; neither opens, stats nor resolves anything (no SMB request for a
UNC path). A symlink is only seen by :func:`_canonical_path` later.
"""
return os.path.normcase(os.path.normpath(path))
def _lexically_inside(path: str, parents: List[str]) -> bool:
"""Whether the normalised ``path`` is one of ``parents`` or below one.
Pure text on normalised spellings (:func:`_lexical_path` or
:func:`_canonical_path`): no filesystem access.
"""
sep = os.path.sep
return any(
path == parent or path.startswith(parent.rstrip(sep) + sep)
for parent in parents
)
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:
if isinstance(monitoring["enabled"], bool):
cls.MONITORING_ENABLED = monitoring["enabled"]
else: # a hand-edited "false" would otherwise switch it on
config_logger.warning(
"Ignoring gallery monitoring enabled: expected true or false"
)
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.
"""
return [canonical for canonical, _spelling in cls._allowed_parent_pairs()]
@classmethod
def _allowed_parent_pairs(cls) -> List[Tuple[str, str]]:
"""``(canonical, lexical spelling)`` of each allowed parent.
Only trusted directories (ComfyUI's own and the environment's) are
resolved here; the spelling is how ComfyUI or the environment wrote
the directory, so a submitted path can be compared without resolving it.
"""
pairs = []
seen = set()
for candidate in _comfyui_directories() + _extra_gallery_roots():
try:
canonical = _canonical_path(candidate)
spelling = _lexical_path(os.path.abspath(candidate))
except (OSError, ValueError):
continue
if _is_filesystem_root(canonical) or not os.path.isdir(canonical):
continue
if canonical not in seen:
seen.add(canonical)
pairs.append((canonical, spelling))
return pairs
@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
ComfyUI directory outside it (``--output-directory D:\\Renders``) and of
each directory listed in PROMPT_MANAGER_EXTRA_GALLERY_ROOTS, so such a
directory is addressed by its own name (``Renders/sub``) rather than by
an absolute path, and two folders of one name stay apart. The same
anchors drive the public (relative) form of paths in API responses.
"""
anchors = []
base = _comfyui_base_directory()
candidates = [base] if base else []
try:
base_key = _canonical_path(base) if base else None
except (OSError, ValueError):
base_key = None
for directory in _comfyui_directories():
try:
canonical = _canonical_path(directory)
except (OSError, ValueError):
continue
if base_key is None or not _lexically_inside(canonical, [base_key]):
candidates.append(os.path.dirname(canonical))
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).
"""
code, reason = cls.gallery_root_problem(path)
return code is None, reason
@classmethod
def gallery_root_problem(cls, path: Any) -> Tuple[Optional[str], str]:
"""``(code, reason)`` why ``path`` may not be a gallery root, else (None, "").
Codes: ``empty``, ``unresolvable``, ``outside-allowed``, ``not-found``
and ``not-directory``. No reason names the path.
Containment is decided twice. First lexically, with no filesystem
access at all: a path (UNC, ``\\\\?\\``, a device, anything) whose
normalised spelling lies outside every allowed parent answers
``outside-allowed`` before it is resolved, so the server never opens
or stats it (no SMB connection, no existence or timing oracle). Only
then is it resolved and checked again, which catches a symlink out.
The filesystem root and the home directory answer ``outside-allowed``
too: one answer for every caller (validate-source, the settings save,
config.json), and the detail goes to the log only.
"""
if not isinstance(path, str) or not path.strip():
return "empty", "Gallery root must be a non-empty path"
pairs = cls._allowed_parent_pairs()
spellings = [spelling for pair in pairs for spelling in pair]
try:
lexical = cls._lexical_candidates(path)
except (TypeError, ValueError):
return "unresolvable", "Gallery root could not be resolved"
if not any(_lexically_inside(candidate, spellings) for candidate in lexical):
return "outside-allowed", OUTSIDE_ALLOWED_REASON
try:
canonical = cls.resolve_gallery_root(path)
except (OSError, ValueError):
return "unresolvable", "Gallery root could not be resolved"
if _is_filesystem_root(canonical):
config_logger.info("Gallery root refused: a filesystem root")
return "outside-allowed", OUTSIDE_ALLOWED_REASON
if canonical == _canonical_path(os.path.expanduser("~")):
config_logger.info("Gallery root refused: the home directory")
return "outside-allowed", OUTSIDE_ALLOWED_REASON
if not _lexically_inside(canonical, [canonical for canonical, _ in pairs]):
return "outside-allowed", OUTSIDE_ALLOWED_REASON
if not os.path.exists(canonical):
return "not-found", "Gallery root does not exist"
if not os.path.isdir(canonical):
return "not-directory", "Gallery root is not a directory"
return None, ""
@classmethod
def _lexical_candidates(cls, path: str) -> List[str]:
"""Every spelling ``path`` may resolve to, normalised as text (no IO).
An absolute path is itself; a relative one is joined to each
:meth:`path_anchors` entry, as :meth:`resolve_gallery_root` tries them.
"""
path = path.strip()
if os.path.isabs(path):
return [_lexical_path(path)]
anchors = cls.path_anchors()
return [_lexical_path(os.path.join(anchor, path)) for anchor in anchors]
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
# Maintenance: a reviewed apply takes a verified safety backup of the
# database first (System › Backup & export › Recovery policy, #249 S6).
SAFETY_BACKUP = True
@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,
},
"maintenance": {"safety_backup": cls.SAFETY_BACKUP},
"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
)
maintenance = new_config.get("maintenance", {})
if isinstance(maintenance, dict) and isinstance(
maintenance.get("safety_backup"), bool
):
cls.SAFETY_BACKUP = maintenance["safety_backup"]
# 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}")