diff --git a/prompt_manager.py b/prompt_manager.py index e316f9f..6c3000c 100644 --- a/prompt_manager.py +++ b/prompt_manager.py @@ -3,19 +3,20 @@ PromptManager: Main custom node implementation that extends CLIPTextEncode with persistent prompt storage and search capabilities. """ -import hashlib import datetime +import hashlib import json -import webbrowser import os import time -from typing import Optional, Dict, Any, Tuple, List +import webbrowser +from typing import Any, Dict, List, Optional, Tuple # Import logging system try: from .utils.logging_config import get_logger except ImportError: import sys + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from utils.logging_config import get_logger @@ -25,28 +26,29 @@ except ImportError: # Fallback for older ComfyUI versions class ComfyNodeABC: pass - + class IO: STRING = "STRING" - CLIP = "CLIP" + CLIP = "CLIP" CONDITIONING = "CONDITIONING" - + InputTypeDict = dict try: from .database.operations import PromptDatabase - from .utils.prompt_tracker import PromptTracker, PromptExecutionContext - from .utils.image_monitor import ImageMonitor from .utils.comfyui_integration import get_comfyui_integration + from .utils.image_monitor import ImageMonitor + from .utils.prompt_tracker import PromptExecutionContext, PromptTracker except ImportError: # For direct imports when not in a package - import sys import os + import sys + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from database.operations import PromptDatabase - from utils.prompt_tracker import PromptTracker, PromptExecutionContext - from utils.image_monitor import ImageMonitor from utils.comfyui_integration import get_comfyui_integration + from utils.image_monitor import ImageMonitor + from utils.prompt_tracker import PromptExecutionContext, PromptTracker class PromptManager(ComfyNodeABC): @@ -57,81 +59,100 @@ class PromptManager(ComfyNodeABC): - Metadata management (categories, tags, ratings, notes) - Duplicate detection via SHA256 hashing """ - + def __init__(self): - self.logger = get_logger('prompt_manager.node') + self.logger = get_logger("prompt_manager.node") self.logger.debug("Initializing PromptManager node") - + self.db = PromptDatabase() self.prompt_tracker = PromptTracker(self.db) self.image_monitor = ImageMonitor(self.db, self.prompt_tracker) self.comfyui_integration = get_comfyui_integration() - + # Start image monitoring automatically self._start_gallery_system() self.logger.debug("PromptManager node initialization completed") - + @classmethod def INPUT_TYPES(cls) -> InputTypeDict: return { "required": { - "text": (IO.STRING, { - "multiline": True, - "dynamicPrompts": True, - "tooltip": "The text prompt to be encoded and saved to database." - }), - "clip": (IO.CLIP, { - "tooltip": "The CLIP model used for encoding the text." - }) + "text": ( + IO.STRING, + { + "multiline": True, + "dynamicPrompts": True, + "tooltip": "The text prompt to be encoded and saved to database.", + }, + ), + "clip": ( + IO.CLIP, + {"tooltip": "The CLIP model used for encoding the text."}, + ), }, "optional": { - "category": (IO.STRING, { - "default": "", - "tooltip": "Optional category for organizing prompts (e.g., 'landscapes', 'portraits')" - }), - "tags": (IO.STRING, { - "default": "", - "tooltip": "Comma-separated tags for the prompt (e.g., 'anime, detailed, sunset')" - }), - "search_text": (IO.STRING, { - "default": "", - "tooltip": "Search for past prompts containing this text" - }), - "prepend_text": (IO.STRING, { - "tooltip": "Text to prepend to the main prompt (connected STRING nodes will be added before the main text)" - }), - "append_text": (IO.STRING, { - "tooltip": "Text to append to the main prompt (connected STRING nodes will be added after the main text)" - }) - } + "category": ( + IO.STRING, + { + "default": "", + "tooltip": "Optional category for organizing prompts (e.g., 'landscapes', 'portraits')", + }, + ), + "tags": ( + IO.STRING, + { + "default": "", + "tooltip": "Comma-separated tags for the prompt (e.g., 'anime, detailed, sunset')", + }, + ), + "search_text": ( + IO.STRING, + { + "default": "", + "tooltip": "Search for past prompts containing this text", + }, + ), + "prepend_text": ( + IO.STRING, + { + "tooltip": "Text to prepend to the main prompt (connected STRING nodes will be added before the main text)" + }, + ), + "append_text": ( + IO.STRING, + { + "tooltip": "Text to append to the main prompt (connected STRING nodes will be added after the main text)" + }, + ), + }, } - + RETURN_TYPES = (IO.CONDITIONING, IO.STRING) OUTPUT_TOOLTIPS = ( "A conditioning containing the embedded text used to guide the diffusion model.", - "The final combined text string (with prepend/append applied) that was encoded." + "The final combined text string (with prepend/append applied) that was encoded.", ) FUNCTION = "encode" - CATEGORY = "ComfyAssets/Text" + CATEGORY = "🫶 ComfyAssets/🧠 Prompts" DESCRIPTION = ( "Encodes a text prompt using a CLIP model into an embedding that can be used to guide " "the diffusion model towards generating specific images. Additionally saves all prompts " "to a local SQLite database with optional metadata for search and retrieval." ) - + def encode( - self, - clip, + self, + clip, text: str, category: str = "", tags: str = "", search_text: str = "", prepend_text: str = "", - append_text: str = "" + append_text: str = "", ) -> Tuple[Any]: """ Encode the text prompt and save it to the database. - + Args: clip: The CLIP model for encoding text: The text prompt to encode @@ -140,10 +161,10 @@ class PromptManager(ComfyNodeABC): search_text: Text to search for in past prompts prepend_text: Text to prepend to the main prompt append_text: Text to append to the main prompt - + Returns: Tuple containing the conditioning for the diffusion model and the final text string - + Raises: RuntimeError: If clip input is invalid """ @@ -154,16 +175,16 @@ class PromptManager(ComfyNodeABC): final_text += text if append_text and append_text.strip(): final_text += " " + append_text.strip() - + # Use the combined text for encoding encoding_text = final_text - + # For database storage, save the original main text with metadata about prepend/append storage_text = text - + # Search functionality is now handled by the JavaScript UI # The search parameters are still available for backend processing if needed - + # Validate CLIP model if clip is None: error_msg = ( @@ -173,78 +194,85 @@ class PromptManager(ComfyNodeABC): ) self.logger.error("CLIP validation failed: clip input is None") raise RuntimeError(error_msg) - + # Save prompt to database and set execution context for gallery tracking prompt_id = None if storage_text and storage_text.strip(): self.logger.debug(f"Processing prompt text: {storage_text[:100]}...") - + # Add prepend/append info to tags if they exist extended_tags = self._parse_tags(tags) or [] if prepend_text and prepend_text.strip(): extended_tags.append(f"prepend:{prepend_text.strip()[:50]}") if append_text and append_text.strip(): extended_tags.append(f"append:{append_text.strip()[:50]}") - + try: prompt_id = self._save_prompt_to_database( text=storage_text.strip(), # Always strip whitespace category=category.strip() if category else None, - tags=extended_tags if extended_tags else None + tags=extended_tags if extended_tags else None, ) - + # Set current prompt for image tracking if prompt_id: execution_id = self.prompt_tracker.set_current_prompt( prompt_text=encoding_text.strip(), # Use final combined text for tracking additional_data={ - 'category': category.strip() if category else None, - 'tags': extended_tags, - 'prompt_id': prompt_id, - 'prepend_text': prepend_text.strip() if prepend_text else None, - 'append_text': append_text.strip() if append_text else None, - 'final_text': encoding_text.strip() # Store final combined text - } + "category": category.strip() if category else None, + "tags": extended_tags, + "prompt_id": prompt_id, + "prepend_text": ( + prepend_text.strip() if prepend_text else None + ), + "append_text": append_text.strip() if append_text else None, + "final_text": encoding_text.strip(), # Store final combined text + }, ) - self.logger.debug(f"Set execution context: {execution_id} for prompt ID: {prompt_id}") - + self.logger.debug( + f"Set execution context: {execution_id} for prompt ID: {prompt_id}" + ) + except Exception as e: # Log error but don't fail the encoding self.logger.warning(f"Failed to save prompt to database: {e}") # Already logged above, no need for additional print - + # Perform standard CLIP text encoding using the combined text - self.logger.debug(f"Performing CLIP text encoding on combined text: {encoding_text[:100]}...") + self.logger.debug( + f"Performing CLIP text encoding on combined text: {encoding_text[:100]}..." + ) tokens = clip.tokenize(encoding_text) conditioning = clip.encode_from_tokens_scheduled(tokens) - + # Register with ComfyUI integration for standard metadata compatibility node_id = f"promptmanager_{int(time.time() * 1000)}" # Unique node ID - self.comfyui_integration.register_prompt(node_id, encoding_text.strip(), { - 'category': category.strip() if category else None, - 'tags': extended_tags, - 'prompt_id': prompt_id, - 'prepend_text': prepend_text.strip() if prepend_text else None, - 'append_text': append_text.strip() if append_text else None - }) - + self.comfyui_integration.register_prompt( + node_id, + encoding_text.strip(), + { + "category": category.strip() if category else None, + "tags": extended_tags, + "prompt_id": prompt_id, + "prepend_text": prepend_text.strip() if prepend_text else None, + "append_text": append_text.strip() if append_text else None, + }, + ) + self.logger.debug("CLIP encoding completed successfully") return (conditioning, encoding_text) - + def _save_prompt_to_database( - self, - text: str, - category: Optional[str] = None, - tags: Optional[list] = None + self, text: str, category: Optional[str] = None, tags: Optional[list] = None ) -> Optional[int]: """ Save the prompt to the SQLite database. - + Args: text: The prompt text category: Optional category tags: List of tags - + Returns: The prompt ID if saved successfully, None otherwise """ @@ -252,83 +280,82 @@ class PromptManager(ComfyNodeABC): # Generate hash for duplicate detection prompt_hash = self._generate_hash(text) self.logger.debug(f"Generated hash for prompt: {prompt_hash[:16]}...") - + # Check if prompt already exists existing = self.db.get_prompt_by_hash(prompt_hash) if existing: - self.logger.info(f"Found existing prompt with ID {existing['id']}, updating metadata") + self.logger.info( + f"Found existing prompt with ID {existing['id']}, updating metadata" + ) # Update metadata if this is a duplicate with new info if any([category, tags]): self.db.update_prompt_metadata( - prompt_id=existing['id'], - category=category, - tags=tags + prompt_id=existing["id"], category=category, tags=tags ) self.logger.debug("Updated metadata for existing prompt") - return existing['id'] - + return existing["id"] + # Save new prompt - self.logger.debug(f"Saving new prompt with category: {category}, tags: {tags}") - prompt_id = self.db.save_prompt( - text=text, - category=category, - tags=tags, - prompt_hash=prompt_hash + self.logger.debug( + f"Saving new prompt with category: {category}, tags: {tags}" ) - + prompt_id = self.db.save_prompt( + text=text, category=category, tags=tags, prompt_hash=prompt_hash + ) + if prompt_id: self.logger.debug(f"Successfully saved new prompt with ID: {prompt_id}") else: self.logger.warning("Failed to save prompt - no ID returned") - + return prompt_id - + except Exception as e: self.logger.error(f"Error saving prompt to database: {e}") # Already logged above, no need for additional print return None - + def _generate_hash(self, text: str) -> str: """Generate SHA256 hash for the prompt text.""" # Normalize text for consistent hashing (strip whitespace, normalize case) normalized_text = text.strip().lower() - return hashlib.sha256(normalized_text.encode('utf-8')).hexdigest() - + return hashlib.sha256(normalized_text.encode("utf-8")).hexdigest() + def _parse_tags(self, tags_string: str) -> Optional[list]: """Parse comma-separated tags string into a list.""" if not tags_string or not tags_string.strip(): return None - - tags = [tag.strip() for tag in tags_string.split(',') if tag.strip()] + + tags = [tag.strip() for tag in tags_string.split(",") if tag.strip()] return tags if tags else None - + def _search_prompts(self, search_text: str = "") -> List[Dict[str, Any]]: """Search for past prompts by text content.""" try: if not search_text or not search_text.strip(): return [] - + results = self.db.search_prompts( text=search_text.strip(), category=None, tags=None, rating_min=None, - limit=50 + limit=50, ) - + return results - + except Exception as e: self.logger.error(f"Error searching prompts: {e}") return [] - + def _open_web_interface(self): """Open the web interface in the default browser.""" try: # Look for a web interface directory current_dir = os.path.dirname(os.path.abspath(__file__)) web_dir = os.path.join(current_dir, "web_interface") - + if os.path.exists(web_dir): # If web interface exists, try to start it index_path = os.path.join(web_dir, "index.html") @@ -336,17 +363,21 @@ class PromptManager(ComfyNodeABC): webbrowser.open(f"file://{index_path}") self.logger.info("Web interface opened in browser") else: - self.logger.warning(f"Web interface directory found but no index.html. Please check {web_dir} for setup instructions") + self.logger.warning( + f"Web interface directory found but no index.html. Please check {web_dir} for setup instructions" + ) else: - self.logger.info("Web interface not yet implemented. This feature will open a web-based prompt management interface when the web_interface directory is created.") - + self.logger.info( + "Web interface not yet implemented. This feature will open a web-based prompt management interface when the web_interface directory is created." + ) + except Exception as e: self.logger.error(f"Error opening web interface: {e}") - + def search_prompts_api(self, search_text: str = "") -> List[Dict[str, Any]]: """API method for JavaScript UI to search prompts.""" return self._search_prompts(search_text=search_text) - + def get_recent_prompts_api(self, limit: int = 20) -> List[Dict[str, Any]]: """API method for JavaScript UI to get recent prompts.""" try: @@ -354,42 +385,42 @@ class PromptManager(ComfyNodeABC): except Exception as e: self.logger.error(f"Error getting recent prompts: {e}") return [] - + def _start_gallery_system(self): """Initialize and start the gallery monitoring system.""" try: self.logger.debug("Starting gallery system...") - + # Start image monitoring self.image_monitor.start_monitoring() - + self.logger.debug("Gallery system started successfully") - + except Exception as e: self.logger.error(f"Failed to start gallery system: {e}") self.logger.warning("Gallery features will be disabled") - + def get_gallery_status(self) -> Dict[str, Any]: """Get status of the gallery system.""" return { - 'image_monitor': self.image_monitor.get_status(), - 'prompt_tracker': self.prompt_tracker.get_status() + "image_monitor": self.image_monitor.get_status(), + "prompt_tracker": self.prompt_tracker.get_status(), } - + def cleanup_gallery_system(self): """Clean up gallery system resources.""" try: - if hasattr(self, 'image_monitor'): + if hasattr(self, "image_monitor"): self.image_monitor.stop_monitoring() self.logger.debug("Gallery system cleaned up") except Exception as e: self.logger.error(f"Error cleaning up gallery system: {e}") - + def __del__(self): """Cleanup when object is destroyed.""" self.cleanup_gallery_system() - + @classmethod def IS_CHANGED(cls, **kwargs): """Always process to ensure database saving.""" - return float("NaN") # Always execute \ No newline at end of file + return float("NaN") # Always execute diff --git a/prompt_manager_text.py b/prompt_manager_text.py index 920e1af..d75125e 100644 --- a/prompt_manager_text.py +++ b/prompt_manager_text.py @@ -1,21 +1,22 @@ """ -PromptManagerText: A text-only version of PromptManager that outputs STRING +PromptManagerText: A text-only version of PromptManager that outputs STRING without CLIP encoding, while maintaining all database and search features. """ -import hashlib import datetime +import hashlib import json -import webbrowser import os import time -from typing import Optional, Dict, Any, Tuple, List +import webbrowser +from typing import Any, Dict, List, Optional, Tuple # Import logging system try: from .utils.logging_config import get_logger except ImportError: import sys + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from utils.logging_config import get_logger @@ -25,28 +26,29 @@ except ImportError: # Fallback for older ComfyUI versions class ComfyNodeABC: pass - + class IO: STRING = "STRING" - CLIP = "CLIP" + CLIP = "CLIP" CONDITIONING = "CONDITIONING" - + InputTypeDict = dict try: from .database.operations import PromptDatabase - from .utils.prompt_tracker import PromptTracker, PromptExecutionContext - from .utils.image_monitor import ImageMonitor from .utils.comfyui_integration import get_comfyui_integration + from .utils.image_monitor import ImageMonitor + from .utils.prompt_tracker import PromptExecutionContext, PromptTracker except ImportError: # For direct imports when not in a package - import sys import os + import sys + sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from database.operations import PromptDatabase - from utils.prompt_tracker import PromptTracker, PromptExecutionContext - from utils.image_monitor import ImageMonitor from utils.comfyui_integration import get_comfyui_integration + from utils.image_monitor import ImageMonitor + from utils.prompt_tracker import PromptExecutionContext, PromptTracker class PromptManagerText(ComfyNodeABC): @@ -59,78 +61,96 @@ class PromptManagerText(ComfyNodeABC): - Duplicate detection via SHA256 hashing - Text concatenation with prepend/append functionality """ - + def __init__(self): - self.logger = get_logger('prompt_manager_text.node') + self.logger = get_logger("prompt_manager_text.node") self.logger.debug("Initializing PromptManagerText node") - + self.db = PromptDatabase() self.prompt_tracker = PromptTracker(self.db) self.image_monitor = ImageMonitor(self.db, self.prompt_tracker) self.comfyui_integration = get_comfyui_integration() - + # Start image monitoring automatically self._start_gallery_system() self.logger.debug("PromptManagerText node initialization completed") - + @classmethod def INPUT_TYPES(cls) -> InputTypeDict: return { "required": { - "text": (IO.STRING, { - "multiline": True, - "dynamicPrompts": True, - "tooltip": "The text prompt to be processed and saved to database." - }) + "text": ( + IO.STRING, + { + "multiline": True, + "dynamicPrompts": True, + "tooltip": "The text prompt to be processed and saved to database.", + }, + ) }, "optional": { - "category": (IO.STRING, { - "default": "", - "tooltip": "Optional category for organizing prompts (e.g., 'landscapes', 'portraits')" - }), - "tags": (IO.STRING, { - "default": "", - "tooltip": "Comma-separated tags for the prompt (e.g., 'anime, detailed, sunset')" - }), - "search_text": (IO.STRING, { - "default": "", - "tooltip": "Search for past prompts containing this text" - }), - "prepend_text": (IO.STRING, { - "default": "", - "tooltip": "Text to prepend to the main prompt (connected STRING nodes will be added before the main text)" - }), - "append_text": (IO.STRING, { - "default": "", - "tooltip": "Text to append to the main prompt (connected STRING nodes will be added after the main text)" - }) - } + "category": ( + IO.STRING, + { + "default": "", + "tooltip": "Optional category for organizing prompts (e.g., 'landscapes', 'portraits')", + }, + ), + "tags": ( + IO.STRING, + { + "default": "", + "tooltip": "Comma-separated tags for the prompt (e.g., 'anime, detailed, sunset')", + }, + ), + "search_text": ( + IO.STRING, + { + "default": "", + "tooltip": "Search for past prompts containing this text", + }, + ), + "prepend_text": ( + IO.STRING, + { + "default": "", + "tooltip": "Text to prepend to the main prompt (connected STRING nodes will be added before the main text)", + }, + ), + "append_text": ( + IO.STRING, + { + "default": "", + "tooltip": "Text to append to the main prompt (connected STRING nodes will be added after the main text)", + }, + ), + }, } - + RETURN_TYPES = (IO.STRING,) OUTPUT_TOOLTIPS = ( "The final combined text string (with prepend/append applied) ready for use in other nodes.", ) FUNCTION = "process_text" - CATEGORY = "ComfyAssets/Text" + CATEGORY = "🫶 ComfyAssets/🧠 Prompts" DESCRIPTION = ( "Processes and manages text prompts with database storage and search capabilities. " "Outputs a plain STRING that can be used with any node that accepts text input. " "Includes all PromptManager features: categorization, tagging, search, and prepend/append functionality." ) - + def process_text( - self, + self, text: str, category: str = "", tags: str = "", search_text: str = "", prepend_text: str = "", - append_text: str = "" + append_text: str = "", ) -> Tuple[str]: """ Process the text prompt and save it to the database. - + Args: text: The text prompt to process category: Optional category for organization @@ -138,7 +158,7 @@ class PromptManagerText(ComfyNodeABC): search_text: Text to search for in past prompts prepend_text: Text to prepend to the main prompt append_text: Text to append to the main prompt - + Returns: Tuple containing the final processed text string """ @@ -149,77 +169,82 @@ class PromptManagerText(ComfyNodeABC): final_text += text if append_text and append_text.strip(): final_text += " " + append_text.strip() - + # For database storage, save the original main text with metadata about prepend/append storage_text = text - + # Search functionality is now handled by the JavaScript UI # The search parameters are still available for backend processing if needed - + # Save prompt to database and set execution context for gallery tracking prompt_id = None if storage_text and storage_text.strip(): self.logger.debug(f"Processing prompt text: {storage_text[:100]}...") - + # Add prepend/append info to tags if they exist extended_tags = self._parse_tags(tags) or [] if prepend_text and prepend_text.strip(): extended_tags.append(f"prepend:{prepend_text.strip()[:50]}") if append_text and append_text.strip(): extended_tags.append(f"append:{append_text.strip()[:50]}") - + try: prompt_id = self._save_prompt_to_database( text=storage_text.strip(), # Always strip whitespace category=category.strip() if category else None, - tags=extended_tags if extended_tags else None + tags=extended_tags if extended_tags else None, ) - + # Set current prompt for image tracking if prompt_id: execution_id = self.prompt_tracker.set_current_prompt( prompt_text=final_text.strip(), # Use final combined text for tracking additional_data={ - 'category': category.strip() if category else None, - 'tags': extended_tags, - 'prompt_id': prompt_id, - 'prepend_text': prepend_text.strip() if prepend_text else None, - 'append_text': append_text.strip() if append_text else None - } + "category": category.strip() if category else None, + "tags": extended_tags, + "prompt_id": prompt_id, + "prepend_text": ( + prepend_text.strip() if prepend_text else None + ), + "append_text": append_text.strip() if append_text else None, + }, ) - self.logger.debug(f"Set execution context: {execution_id} for prompt ID: {prompt_id}") - + self.logger.debug( + f"Set execution context: {execution_id} for prompt ID: {prompt_id}" + ) + except Exception as e: # Log error but don't fail the processing self.logger.warning(f"Failed to save prompt to database: {e}") - + # Register with ComfyUI integration for standard metadata compatibility node_id = f"promptmanagertext_{int(time.time() * 1000)}" # Unique node ID - self.comfyui_integration.register_prompt(node_id, final_text.strip(), { - 'category': category.strip() if category else None, - 'tags': extended_tags, - 'prompt_id': prompt_id, - 'prepend_text': prepend_text.strip() if prepend_text else None, - 'append_text': append_text.strip() if append_text else None - }) - + self.comfyui_integration.register_prompt( + node_id, + final_text.strip(), + { + "category": category.strip() if category else None, + "tags": extended_tags, + "prompt_id": prompt_id, + "prepend_text": prepend_text.strip() if prepend_text else None, + "append_text": append_text.strip() if append_text else None, + }, + ) + self.logger.debug(f"Text processing completed: {final_text[:100]}...") return (final_text,) - + def _save_prompt_to_database( - self, - text: str, - category: Optional[str] = None, - tags: Optional[list] = None + self, text: str, category: Optional[str] = None, tags: Optional[list] = None ) -> Optional[int]: """ Save the prompt to the SQLite database. - + Args: text: The prompt text category: Optional category tags: List of tags - + Returns: The prompt ID if saved successfully, None otherwise """ @@ -227,82 +252,81 @@ class PromptManagerText(ComfyNodeABC): # Generate hash for duplicate detection prompt_hash = self._generate_hash(text) self.logger.debug(f"Generated hash for prompt: {prompt_hash[:16]}...") - + # Check if prompt already exists existing = self.db.get_prompt_by_hash(prompt_hash) if existing: - self.logger.info(f"Found existing prompt with ID {existing['id']}, updating metadata") + self.logger.info( + f"Found existing prompt with ID {existing['id']}, updating metadata" + ) # Update metadata if this is a duplicate with new info if any([category, tags]): self.db.update_prompt_metadata( - prompt_id=existing['id'], - category=category, - tags=tags + prompt_id=existing["id"], category=category, tags=tags ) self.logger.debug("Updated metadata for existing prompt") - return existing['id'] - + return existing["id"] + # Save new prompt - self.logger.debug(f"Saving new prompt with category: {category}, tags: {tags}") - prompt_id = self.db.save_prompt( - text=text, - category=category, - tags=tags, - prompt_hash=prompt_hash + self.logger.debug( + f"Saving new prompt with category: {category}, tags: {tags}" ) - + prompt_id = self.db.save_prompt( + text=text, category=category, tags=tags, prompt_hash=prompt_hash + ) + if prompt_id: self.logger.debug(f"Successfully saved new prompt with ID: {prompt_id}") else: self.logger.warning("Failed to save prompt - no ID returned") - + return prompt_id - + except Exception as e: self.logger.error(f"Error saving prompt to database: {e}") return None - + def _generate_hash(self, text: str) -> str: """Generate SHA256 hash for the prompt text.""" # Normalize text for consistent hashing (strip whitespace, normalize case) normalized_text = text.strip().lower() - return hashlib.sha256(normalized_text.encode('utf-8')).hexdigest() - + return hashlib.sha256(normalized_text.encode("utf-8")).hexdigest() + def _parse_tags(self, tags_string: str) -> Optional[list]: """Parse comma-separated tags string into a list.""" if not tags_string or not tags_string.strip(): return None - - tags = [tag.strip() for tag in tags_string.split(',') if tag.strip()] + + tags = [tag.strip() for tag in tags_string.split(",") if tag.strip()] return tags if tags else None - + def _search_prompts(self, search_text: str = "") -> List[Dict[str, Any]]: """Search for past prompts by text content.""" try: if not search_text or not search_text.strip(): return [] - + results = self.db.search_prompts( text=search_text.strip(), category=None, tags=None, rating_min=None, - limit=50 + limit=50, ) - + return results - + except Exception as e: self.logger.error(f"Error searching prompts: {e}") return [] - + def _open_web_interface(self): """Open the web interface in the default browser.""" try: # Look for a web interface directory current_dir = os.path.dirname(os.path.abspath(__file__)) web_dir = os.path.join(current_dir, "web_interface") - + if os.path.exists(web_dir): # If web interface exists, try to start it index_path = os.path.join(web_dir, "index.html") @@ -310,17 +334,21 @@ class PromptManagerText(ComfyNodeABC): webbrowser.open(f"file://{index_path}") self.logger.info("Web interface opened in browser") else: - self.logger.warning(f"Web interface directory found but no index.html. Please check {web_dir} for setup instructions") + self.logger.warning( + f"Web interface directory found but no index.html. Please check {web_dir} for setup instructions" + ) else: - self.logger.info("Web interface not yet implemented. This feature will open a web-based prompt management interface when the web_interface directory is created.") - + self.logger.info( + "Web interface not yet implemented. This feature will open a web-based prompt management interface when the web_interface directory is created." + ) + except Exception as e: self.logger.error(f"Error opening web interface: {e}") - + def search_prompts_api(self, search_text: str = "") -> List[Dict[str, Any]]: """API method for JavaScript UI to search prompts.""" return self._search_prompts(search_text=search_text) - + def get_recent_prompts_api(self, limit: int = 20) -> List[Dict[str, Any]]: """API method for JavaScript UI to get recent prompts.""" try: @@ -328,42 +356,42 @@ class PromptManagerText(ComfyNodeABC): except Exception as e: self.logger.error(f"Error getting recent prompts: {e}") return [] - + def _start_gallery_system(self): """Initialize and start the gallery monitoring system.""" try: self.logger.debug("Starting gallery system...") - + # Start image monitoring self.image_monitor.start_monitoring() - + self.logger.debug("Gallery system started successfully") - + except Exception as e: self.logger.error(f"Failed to start gallery system: {e}") self.logger.warning("Gallery features will be disabled") - + def get_gallery_status(self) -> Dict[str, Any]: """Get status of the gallery system.""" return { - 'image_monitor': self.image_monitor.get_status(), - 'prompt_tracker': self.prompt_tracker.get_status() + "image_monitor": self.image_monitor.get_status(), + "prompt_tracker": self.prompt_tracker.get_status(), } - + def cleanup_gallery_system(self): """Clean up gallery system resources.""" try: - if hasattr(self, 'image_monitor'): + if hasattr(self, "image_monitor"): self.image_monitor.stop_monitoring() self.logger.debug("Gallery system cleaned up") except Exception as e: self.logger.error(f"Error cleaning up gallery system: {e}") - + def __del__(self): """Cleanup when object is destroyed.""" self.cleanup_gallery_system() - + @classmethod def IS_CHANGED(cls, **kwargs): """Always process to ensure database saving.""" - return float("NaN") # Always execute \ No newline at end of file + return float("NaN") # Always execute