""" Prompt Registry - Centralized prompt management with override hierarchy Supports: task-level overrides → DB prompts → GlobalAIPrompt (REQUIRED) """ import logging from typing import Dict, Any, Optional, Tuple from django.db import models logger = logging.getLogger(__name__) class PromptRegistry: """ Centralized prompt registry with hierarchical resolution: 1. Task-level prompt_override (if exists) 2. DB prompt for (account, function) 3. GlobalAIPrompt (REQUIRED - no hardcoded fallbacks) """ # Removed ALL hardcoded prompts - GlobalAIPrompt is now the ONLY source of default prompts # To add/modify prompts, use Django admin: /admin/system/globalaiprompt/ # Mapping from function names to prompt types FUNCTION_TO_PROMPT_TYPE = { 'auto_cluster': 'clustering', 'generate_ideas': 'ideas', 'generate_content': 'content_generation', 'generate_images': 'image_prompt_extraction', 'extract_image_prompts': 'image_prompt_extraction', 'generate_image_prompts': 'image_prompt_extraction', 'generate_site_structure': 'site_structure_generation', 'optimize_content': 'optimize_content', # Phase 8: Universal Content Types 'generate_product_content': 'product_generation', 'generate_service_page': 'service_generation', 'generate_taxonomy': 'taxonomy_generation', } # Mapping of prompt types to their prefix numbers and display names # Format: {prompt_type: (number, display_name)} # GP = Global Prompt, CP = Custom Prompt PROMPT_PREFIX_MAP = { 'clustering': ('01', 'Clustering'), 'ideas': ('02', 'Ideas'), 'content_generation': ('03', 'ContentGen'), 'image_prompt_extraction': ('04', 'ImagePrompts'), 'site_structure_generation': ('05', 'SiteStructure'), 'optimize_content': ('06', 'OptimizeContent'), 'product_generation': ('07', 'ProductGen'), 'service_generation': ('08', 'ServiceGen'), 'taxonomy_generation': ('09', 'TaxonomyGen'), 'image_prompt_template': ('10', 'ImageTemplate'), 'negative_prompt': ('11', 'NegativePrompt'), } @classmethod def get_prompt_prefix(cls, prompt_type: str, is_custom: bool) -> str: """ Generate prompt prefix for tracking. Args: prompt_type: The prompt type (e.g., 'clustering', 'ideas') is_custom: True if using custom/account-specific prompt, False if global Returns: Prefix string like "##GP01-Clustering" or "##CP01-Clustering" """ prefix_info = cls.PROMPT_PREFIX_MAP.get(prompt_type, ('00', prompt_type.title())) number, display_name = prefix_info prefix_type = 'CP' if is_custom else 'GP' return f"##{prefix_type}{number}-{display_name}" @classmethod def get_prompt_with_metadata( cls, function_name: str, account: Optional[Any] = None, task: Optional[Any] = None, context: Optional[Dict[str, Any]] = None ) -> Tuple[str, bool, str]: """ Get prompt for a function with metadata about source. Priority: 1. task.prompt_override (if task provided and has override) 2. DB prompt for (account, function) - marked as custom if is_customized=True 3. GlobalAIPrompt (REQUIRED - no hardcoded fallbacks) Args: function_name: AI function name (e.g., 'auto_cluster', 'generate_ideas') account: Account object (optional) task: Task object with optional prompt_override (optional) context: Additional context for prompt rendering (optional) Returns: Tuple of (prompt_string, is_custom, prompt_type) - prompt_string: The rendered prompt - is_custom: True if using custom/account prompt, False if global - prompt_type: The prompt type identifier """ # Step 1: Get prompt type prompt_type = cls.FUNCTION_TO_PROMPT_TYPE.get(function_name, function_name) # Step 2: Check task-level override (always considered custom) if task and hasattr(task, 'prompt_override') and task.prompt_override: logger.info(f"Using task-level prompt override for {function_name}") prompt = task.prompt_override return cls._render_prompt(prompt, context or {}), True, prompt_type # Step 3: Try DB prompt (account-specific) if account: try: from igny8_core.modules.system.models import AIPrompt db_prompt = AIPrompt.objects.get( account=account, prompt_type=prompt_type, is_active=True ) # Check if prompt is customized is_custom = db_prompt.is_customized logger.info(f"Using {'customized' if is_custom else 'default'} account prompt for {function_name} (account {account.id})") prompt = db_prompt.prompt_value return cls._render_prompt(prompt, context or {}), is_custom, prompt_type except Exception as e: logger.debug(f"No account-specific prompt found for {function_name}: {e}") # Step 4: Try GlobalAIPrompt (platform-wide default) - REQUIRED try: from igny8_core.modules.system.global_settings_models import GlobalAIPrompt global_prompt = GlobalAIPrompt.objects.get( prompt_type=prompt_type, is_active=True ) logger.info(f"Using global default prompt for {function_name} from GlobalAIPrompt") prompt = global_prompt.prompt_value return cls._render_prompt(prompt, context or {}), False, prompt_type except Exception as e: error_msg = ( f"ERROR: Global prompt '{prompt_type}' not found for function '{function_name}'. " f"Please configure it in Django admin at: /admin/system/globalaiprompt/. " f"Error: {e}" ) logger.error(error_msg) raise ValueError(error_msg) @classmethod def get_prompt( cls, function_name: str, account: Optional[Any] = None, task: Optional[Any] = None, context: Optional[Dict[str, Any]] = None ) -> str: """ Get prompt for a function with hierarchical resolution. Priority: 1. task.prompt_override (if task provided and has override) 2. DB prompt for (account, function) 3. GlobalAIPrompt (REQUIRED - no hardcoded fallbacks) Args: function_name: AI function name (e.g., 'auto_cluster', 'generate_ideas') account: Account object (optional) task: Task object with optional prompt_override (optional) context: Additional context for prompt rendering (optional) Returns: Prompt string ready for formatting """ prompt, _, _ = cls.get_prompt_with_metadata(function_name, account, task, context) return prompt @classmethod def _render_prompt(cls, prompt_template: str, context: Dict[str, Any]) -> str: """ Render prompt template with context variables. Supports both .format() style ({variable}) and placeholder replacement ([IGNY8_*]). Args: prompt_template: Prompt template string context: Context variables for rendering Returns: Rendered prompt string """ if not context: return prompt_template rendered = prompt_template # Step 1: Replace [IGNY8_*] placeholders first (always do this) for key, value in context.items(): placeholder = f'[IGNY8_{key.upper()}]' if placeholder in rendered: rendered = rendered.replace(placeholder, str(value)) logger.debug(f"Replaced placeholder {placeholder} with {len(str(value))} characters") # Step 2: Try .format() style for {variable} placeholders (if any remain) # Normalize context keys - convert UPPER to lowercase for .format() normalized_context = {} for key, value in context.items(): # Try both original key and lowercase version normalized_context[key] = value normalized_context[key.lower()] = value # Only try .format() if there are {variable} placeholders if '{' in rendered and '}' in rendered: try: rendered = rendered.format(**normalized_context) except (KeyError, ValueError, IndexError) as e: # If .format() fails, log warning but keep the [IGNY8_*] replacements logger.warning(f"Failed to format prompt with .format(): {e}. Using [IGNY8_*] replacements only.") return rendered @classmethod def get_image_prompt_template(cls, account: Optional[Any] = None) -> str: """ Get image prompt template. Returns template string (not rendered) - caller should format with .format() """ prompt_type = 'image_prompt_template' # Try DB prompt if account: try: from igny8_core.modules.system.models import AIPrompt db_prompt = AIPrompt.objects.get( account=account, prompt_type=prompt_type, is_active=True ) return db_prompt.prompt_value except Exception: pass # Try GlobalAIPrompt try: from igny8_core.modules.system.global_settings_models import GlobalAIPrompt global_prompt = GlobalAIPrompt.objects.get( prompt_type=prompt_type, is_active=True ) return global_prompt.prompt_value except Exception: # Fallback for image_prompt_template return '{image_type} image for blog post titled "{post_title}": {image_prompt}' @classmethod def get_negative_prompt(cls, account: Optional[Any] = None) -> str: """ Get negative prompt. Returns template string (not rendered). """ prompt_type = 'negative_prompt' # Try DB prompt if account: try: from igny8_core.modules.system.models import AIPrompt db_prompt = AIPrompt.objects.get( account=account, prompt_type=prompt_type, is_active=True ) return db_prompt.prompt_value except Exception: pass # Try GlobalAIPrompt try: from igny8_core.modules.system.global_settings_models import GlobalAIPrompt global_prompt = GlobalAIPrompt.objects.get( prompt_type=prompt_type, is_active=True ) return global_prompt.prompt_value except Exception: # Fallback for negative_prompt return 'text, watermark, logo, overlay, title, caption, writing on walls, writing on objects, UI, infographic elements, post title' # Convenience function for backward compatibility def get_prompt(function_name: str, account=None, task=None, context=None) -> str: """Get prompt using registry""" return PromptRegistry.get_prompt(function_name, account=account, task=task, context=context) def get_prompt_with_prefix(function_name: str, account=None, task=None, context=None) -> Tuple[str, str]: """ Get prompt with its tracking prefix. Args: function_name: AI function name account: Account object (optional) task: Task object with optional prompt_override (optional) context: Additional context for prompt rendering (optional) Returns: Tuple of (prompt_string, prefix_string) - prompt_string: The rendered prompt - prefix_string: The tracking prefix (e.g., '##GP01-Clustering' or '##CP01-Clustering') """ prompt, is_custom, prompt_type = PromptRegistry.get_prompt_with_metadata( function_name, account=account, task=task, context=context ) prefix = PromptRegistry.get_prompt_prefix(prompt_type, is_custom) return prompt, prefix def get_prompt_prefix_for_function(function_name: str, account=None, task=None) -> str: """ Get just the prefix for a function without fetching the full prompt. Useful when the prompt was already fetched elsewhere. Args: function_name: AI function name account: Account object (optional) task: Task object with optional prompt_override (optional) Returns: The tracking prefix (e.g., '##GP01-Clustering' or '##CP01-Clustering') """ prompt_type = PromptRegistry.FUNCTION_TO_PROMPT_TYPE.get(function_name, function_name) # Check for task-level override (always custom) if task and hasattr(task, 'prompt_override') and task.prompt_override: return PromptRegistry.get_prompt_prefix(prompt_type, is_custom=True) # Check for account-specific prompt if account: try: from igny8_core.modules.system.models import AIPrompt db_prompt = AIPrompt.objects.get( account=account, prompt_type=prompt_type, is_active=True ) return PromptRegistry.get_prompt_prefix(prompt_type, is_custom=db_prompt.is_customized) except Exception: pass # Fallback to global (not custom) return PromptRegistry.get_prompt_prefix(prompt_type, is_custom=False)