Upload 5 files

core/__init__.py

@@ -0,0 +1 @@
+# NAIA-WEB Core Package

core/api_service.py

@@ -0,0 +1,388 @@
+"""
+NAIA-WEB API Service
+NAI Image Generation API communication layer
+
+Reference: NAIA2.0/core/api_service.py (260-460)
+"""
+
+import aiohttp
+import asyncio
+import zipfile
+import io
+import json
+import random
+import base64
+from dataclasses import dataclass
+from typing import Optional, Tuple, Dict, Any, List
+from PIL import Image
+
+from utils.constants import NAI_API_URL, MODEL_ID_MAP
+
+
+def process_reference_image(file_path: str) -> str:
+    """
+    Process reference image for character reference API.
+    Normalizes aspect ratio and encodes to base64.
+
+    Reference: NAIA2.0/modules/character_reference_module.py _file_to_base64
+    """
+    try:
+        original_image = Image.open(file_path)
+        width, height = original_image.size
+        aspect_ratio = width / height
+
+        # Standard aspect ratios (ratio, canvas_width, canvas_height)
+        ratios = {
+            '2:3': (2/3, 1024, 1536),
+            '3:2': (3/2, 1536, 1024),
+            '1:1': (1/1, 1472, 1472)
+        }
+
+        # Find closest standard ratio
+        closest_ratio = min(ratios.keys(), key=lambda k: abs(aspect_ratio - ratios[k][0]))
+        target_ratio, canvas_width, canvas_height = ratios[closest_ratio]
+
+        print(f"NAIA-WEB: Reference image {width}x{height} ({aspect_ratio:.2f}) → {closest_ratio} ({canvas_width}x{canvas_height})")
+
+        # Create black canvas
+        canvas = Image.new('RGB', (canvas_width, canvas_height), (0, 0, 0))
+
+        # Resize to fit canvas (preserve aspect ratio)
+        if width / canvas_width > height / canvas_height:
+            new_width = canvas_width
+            new_height = int(height * (canvas_width / width))
+        else:
+            new_height = canvas_height
+            new_width = int(width * (canvas_height / height))
+
+        resized_image = original_image.resize((new_width, new_height), Image.Resampling.LANCZOS)
+
+        # Center on canvas
+        x_offset = (canvas_width - new_width) // 2
+        y_offset = (canvas_height - new_height) // 2
+
+        # Handle RGBA transparency
+        if resized_image.mode == 'RGBA':
+            canvas = canvas.convert('RGBA')
+            canvas.paste(resized_image, (x_offset, y_offset), resized_image)
+            rgb_canvas = Image.new('RGB', (canvas_width, canvas_height), (0, 0, 0))
+            rgb_canvas.paste(canvas, (0, 0), canvas)
+            canvas = rgb_canvas
+        else:
+            canvas.paste(resized_image, (x_offset, y_offset))
+
+        # Encode to base64
+        buffer = io.BytesIO()
+        canvas.save(buffer, format="PNG", optimize=False)
+        return base64.b64encode(buffer.getvalue()).decode("utf-8")
+
+    except Exception as e:
+        print(f"NAIA-WEB: Failed to process reference image: {e}")
+        # Fallback: use original file bytes
+        with open(file_path, "rb") as f:
+            return base64.b64encode(f.read()).decode("utf-8")
+
+
+class NAIAPIError(Exception):
+    """Custom exception for NAI API errors"""
+    def __init__(self, status_code: int, message: str, debug_info: Optional[Dict] = None):
+        self.status_code = status_code
+        self.message = message
+        self.debug_info = debug_info or {}
+        super().__init__(f"NAI API Error ({status_code}): {message}")
+
+
+@dataclass
+class CharacterReferenceData:
+    """Character reference data for NAID4.5"""
+    image_base64: str  # Base64 encoded image
+    style_aware: bool = True  # Include style from reference
+    fidelity: float = 0.75  # How closely to follow the reference (0.0-1.0)
+
+
+@dataclass
+class GenerationParameters:
+    """Parameters for image generation request"""
+    prompt: str
+    negative_prompt: str
+    width: int
+    height: int
+    steps: int = 28
+    scale: float = 5.0
+    cfg_rescale: float = 0.4  # NAIA2.0 default
+    sampler: str = "k_euler"
+    seed: Optional[int] = None
+    model: str = "NAID4.5F"
+    noise_schedule: str = "native"
+    # Character prompts: List of (prompt, negative) tuples
+    character_prompts: List[Tuple[str, str]] = None
+    # Character reference (NAID4.5 feature)
+    character_reference: Optional[CharacterReferenceData] = None
+
+
+class NAIAPIService:
+    """
+    Service for communicating with NAI image generation API.
+
+    Handles V4.5 model API calls with proper payload structure.
+    """
+
+    def __init__(self):
+        self._session: Optional[aiohttp.ClientSession] = None
+        # Debug info storage
+        self._last_payload: Optional[Dict] = None
+        self._last_response_status: Optional[int] = None
+        self._last_response_text: Optional[str] = None
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        """Get or create aiohttp session"""
+        if self._session is None or self._session.closed:
+            self._session = aiohttp.ClientSession()
+        return self._session
+
+    async def generate_image(
+        self,
+        token: str,
+        params: GenerationParameters
+    ) -> Tuple[Image.Image, Dict[str, Any]]:
+        """
+        Call NAI API to generate an image.
+
+        Args:
+            token: NAI API authentication token
+            params: Generation parameters
+
+        Returns:
+            Tuple of (PIL Image, metadata dict)
+
+        Raises:
+            NAIAPIError: If API call fails
+        """
+        session = await self._get_session()
+
+        # Get model name from mapping
+        model_name = MODEL_ID_MAP.get(params.model, "nai-diffusion-4-5-full")
+
+        # Determine seed
+        seed = params.seed if params.seed and params.seed > 0 else random.randint(0, 2**32 - 1)
+
+        # Build V4 prompt structure
+        v4_prompt = {
+            "caption": {
+                "base_caption": params.prompt,
+                "char_captions": []
+            },
+            "use_coords": False,
+            "use_order": True
+        }
+
+        v4_negative_prompt = {
+            "caption": {
+                "base_caption": params.negative_prompt,
+                "char_captions": []
+            },
+            "legacy_uc": False
+        }
+
+        # Add character prompts if provided (NAID4.5 feature)
+        if params.character_prompts:
+            for char_prompt, char_negative in params.character_prompts:
+                if char_prompt.strip():
+                    # Default center position (no 5x5 grid feature)
+                    centers = [{"x": 0.5, "y": 0.5}]
+                    v4_prompt["caption"]["char_captions"].append({
+                        "char_caption": char_prompt.strip(),
+                        "centers": centers
+                    })
+                    v4_negative_prompt["caption"]["char_captions"].append({
+                        "char_caption": char_negative.strip() if char_negative else "",
+                        "centers": centers
+                    })
+            if v4_prompt["caption"]["char_captions"]:
+                print(f"NAIA-WEB: Added {len(v4_prompt['caption']['char_captions'])} character prompt(s)")
+
+        # Build API parameters (matching NAI V4 structure)
+        api_parameters = {
+            "width": params.width,
+            "height": params.height,
+            "n_samples": 1,
+            "seed": seed,
+            "extra_noise_seed": seed,
+            "sampler": params.sampler,
+            "steps": params.steps,
+            "scale": params.scale,
+            "cfg_rescale": params.cfg_rescale,
+            "noise_schedule": params.noise_schedule,
+            "negative_prompt": params.negative_prompt,
+            # V4 specific parameters
+            "params_version": 3,
+            "add_original_image": True,
+            "legacy": False,
+            "legacy_uc": False,
+            "autoSmea": True,
+            "prefer_brownian": True,
+            "ucPreset": 0,
+            "use_coords": False,
+            "v4_prompt": v4_prompt,
+            "v4_negative_prompt": v4_negative_prompt,
+            "skip_cfg_above_sigma": None,
+        }
+
+        # Add character reference if provided (NAID4.5 feature)
+        if params.character_reference:
+            ref = params.character_reference
+            # Build description based on style_aware setting
+            if ref.style_aware:
+                description = {
+                    "caption": {"base_caption": "character&style", "char_captions": []},
+                    "legacy_uc": False
+                }
+            else:
+                description = {
+                    "caption": {"base_caption": "character", "char_captions": []},
+                    "legacy_uc": False
+                }
+
+            api_parameters["director_reference_descriptions"] = [description]
+            api_parameters["director_reference_images"] = [ref.image_base64]
+            api_parameters["director_reference_information_extracted"] = [1]
+            api_parameters["director_reference_secondary_strength_values"] = [ref.fidelity]
+            api_parameters["director_reference_strength_values"] = [1]
+            api_parameters["controlnet_strength"] = 1
+            api_parameters["inpaintImg2ImgStrength"] = 1
+            api_parameters["normalize_reference_strength_multiple"] = True
+
+            print(f"NAIA-WEB: Character reference enabled (style_aware={ref.style_aware}, fidelity={ref.fidelity})")
+
+        # Build request payload
+        payload = {
+            "input": params.prompt,
+            "model": model_name,
+            "action": "generate",
+            "parameters": api_parameters
+        }
+
+        # Headers - matching NAIA2.0 (no Accept header)
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json"
+        }
+
+        # Store for debugging
+        self._last_payload = payload
+        self._last_response_status = None
+        self._last_response_text = None
+
+        max_retries = 2
+        last_error = None
+
+        for attempt in range(max_retries):
+            try:
+                async with session.post(
+                    NAI_API_URL,
+                    json=payload,
+                    headers=headers,
+                    timeout=aiohttp.ClientTimeout(total=180)  # NAIA2.0 uses 180s
+                ) as response:
+                    self._last_response_status = response.status
+
+                    if response.status == 200:
+                        zip_data = await response.read()
+                        image = self._extract_image_from_zip(zip_data)
+
+                        metadata = {
+                            "seed": seed,
+                            "model": params.model,
+                            "steps": params.steps,
+                            "scale": params.scale,
+                            "sampler": params.sampler,
+                            "width": params.width,
+                            "height": params.height,
+                        }
+
+                        return image, metadata
+                    else:
+                        error_text = await response.text()
+                        self._last_response_text = error_text
+
+                        debug_info = {
+                            "model": model_name,
+                            "status": response.status,
+                            "response": error_text[:500],  # Truncate long responses
+                            "token_length": len(token) if token else 0,
+                            "token_prefix": token[:10] + "..." if token and len(token) > 10 else token
+                        }
+                        last_error = NAIAPIError(response.status, error_text, debug_info)
+
+                        # Don't retry on client errors (4xx)
+                        if 400 <= response.status < 500:
+                            raise last_error
+
+            except aiohttp.ClientError as e:
+                self._last_response_text = str(e)
+                last_error = NAIAPIError(0, f"Network error: {str(e)}")
+
+            # Wait before retry
+            if attempt < max_retries - 1:
+                await asyncio.sleep(1)
+
+        raise last_error or NAIAPIError(0, "Unknown error")
+
+    def _extract_image_from_zip(self, zip_data: bytes) -> Image.Image:
+        """Extract image from NAI response zip"""
+        with zipfile.ZipFile(io.BytesIO(zip_data)) as zf:
+            # Find PNG file in zip
+            image_files = [f for f in zf.namelist() if f.endswith('.png')]
+            if not image_files:
+                raise NAIAPIError(0, "No image found in response")
+
+            image_bytes = zf.read(image_files[0])
+            return Image.open(io.BytesIO(image_bytes))
+
+    async def close(self):
+        """Close the aiohttp session"""
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    def get_debug_info(self) -> Dict[str, Any]:
+        """Return debug info from last request"""
+        return {
+            "last_status": self._last_response_status,
+            "last_response": self._last_response_text,
+            "last_payload_keys": list(self._last_payload.keys()) if self._last_payload else None,
+            "last_model": self._last_payload.get("model") if self._last_payload else None,
+        }
+
+
+def format_api_error(error: NAIAPIError) -> str:
+    """Format API error for user display with debug info"""
+    base_msg = ""
+    if error.status_code == 401:
+        base_msg = "Authentication failed. Please check your API token."
+    elif error.status_code == 402:
+        base_msg = "Insufficient Anlas. Please check your account balance."
+    elif error.status_code == 429:
+        base_msg = "Rate limited. Please wait before trying again."
+    elif error.status_code >= 500:
+        base_msg = "NAI server error. Please try again later."
+    elif error.status_code == 0:
+        base_msg = f"Connection error: {error.message}"
+    else:
+        base_msg = f"API Error ({error.status_code}): {error.message}"
+
+    # Add debug info if available
+    if error.debug_info:
+        debug_parts = []
+        if "token_length" in error.debug_info:
+            debug_parts.append(f"Token length: {error.debug_info['token_length']}")
+        if "token_prefix" in error.debug_info:
+            debug_parts.append(f"Token prefix: {error.debug_info['token_prefix']}")
+        if "model" in error.debug_info:
+            debug_parts.append(f"Model: {error.debug_info['model']}")
+        if "response" in error.debug_info:
+            debug_parts.append(f"Response: {error.debug_info['response']}")
+
+        if debug_parts:
+            base_msg += "\n\n[Debug Info]\n" + "\n".join(debug_parts)
+
+    return base_msg

core/autocomplete_service.py

@@ -0,0 +1,336 @@
+"""
+NAIA-WEB Autocomplete Service
+Tag autocomplete functionality for prompt input fields
+
+Reference: NAIA2.0/core/autocomplete_manager.py, NAIA2.0/core/tag_data_manager.py
+"""
+
+from typing import List, Dict, Tuple, Optional
+from dataclasses import dataclass
+import time
+
+
+@dataclass
+class TagResult:
+    """Single tag search result"""
+    tag: str
+    count: int
+    category: str = "general"  # general, artist, character
+
+
+class AutocompleteService:
+    """
+    Autocomplete service for tag suggestions.
+
+    Provides fast tag search with:
+    - Prefix matching (highest priority)
+    - Contains matching
+    - Category-aware search (general, artist, character)
+    - Frequency-based sorting
+
+    Usage:
+        service = AutocompleteService()
+        results = service.search("blue")  # Returns list of TagResult
+    """
+
+    _instance: Optional['AutocompleteService'] = None
+
+    def __new__(cls):
+        """Singleton pattern for shared data across requests"""
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self):
+        if self._initialized:
+            return
+
+        self._generals: Dict[str, int] = {}
+        self._artists: Dict[str, int] = {}
+        self._characters: Dict[str, int] = {}
+        self._combined: Dict[str, Tuple[int, str]] = {}  # tag -> (count, category)
+
+        self._load_data()
+        self._initialized = True
+
+    def _load_data(self):
+        """Load tag data from source files"""
+        print("AutocompleteService: Loading tag data...")
+        start_time = time.time()
+
+        # Load generals (general tags)
+        try:
+            from data.autocomplete.result_dupl import generals
+            self._generals = dict(generals)
+            print(f"  - Loaded {len(self._generals):,} general tags")
+        except ImportError as e:
+            print(f"  - Failed to load generals: {e}")
+            self._generals = {}
+
+        # Load artists
+        try:
+            from data.autocomplete.artist_dictionary import artist_dict
+            # artist_dict has artist names as keys with counts
+            self._artists = {}
+            for key, value in artist_dict.items():
+                if isinstance(value, int):
+                    self._artists[key] = value
+                elif isinstance(value, (list, tuple)) and len(value) > 0:
+                    # Some entries might be [count, ...] format
+                    self._artists[key] = value[0] if isinstance(value[0], int) else 0
+            print(f"  - Loaded {len(self._artists)} artists")
+        except ImportError as e:
+            print(f"  - Failed to load artists: {e}")
+            self._artists = {}
+
+        # Load characters
+        try:
+            from data.autocomplete.danbooru_character import character_dict_count
+            self._characters = dict(character_dict_count)
+            print(f"  - Loaded {len(self._characters):,} characters")
+        except ImportError as e:
+            print(f"  - Failed to load characters: {e}")
+            self._characters = {}
+
+        # Build combined index
+        self._build_combined_index()
+
+        elapsed = time.time() - start_time
+        print(f"AutocompleteService: Loaded {len(self._combined):,} total tags in {elapsed:.2f}s")
+
+    def _build_combined_index(self):
+        """Build combined index with category information"""
+        self._combined = {}
+
+        # Add generals
+        for tag, count in self._generals.items():
+            self._combined[tag] = (count, "general")
+
+        # Add artists (may override generals with same name)
+        for tag, count in self._artists.items():
+            self._combined[tag] = (count, "artist")
+
+        # Add characters
+        for tag, count in self._characters.items():
+            if tag not in self._combined or count > self._combined[tag][0]:
+                self._combined[tag] = (count, "character")
+
+    def search(
+        self,
+        query: str,
+        limit: int = 20,
+        category: Optional[str] = None
+    ) -> List[TagResult]:
+        """
+        Search for tags matching query.
+
+        Args:
+            query: Search query (minimum 1 character)
+            limit: Maximum results to return
+            category: Optional filter by category ('general', 'artist', 'character')
+
+        Returns:
+            List of TagResult sorted by relevance and frequency
+        """
+        if not query or len(query) < 1:
+            return []
+
+        query_lower = query.lower().strip()
+
+        # Select data source based on category
+        if category == "artist":
+            source = {k: (v, "artist") for k, v in self._artists.items()}
+        elif category == "character":
+            source = {k: (v, "character") for k, v in self._characters.items()}
+        elif category == "general":
+            source = {k: (v, "general") for k, v in self._generals.items()}
+        else:
+            source = self._combined
+
+        # Separate matches by type
+        exact_matches = []
+        prefix_matches = []
+        contains_matches = []
+
+        for tag, (count, cat) in source.items():
+            tag_lower = tag.lower()
+
+            if tag_lower == query_lower:
+                exact_matches.append(TagResult(tag=tag, count=count, category=cat))
+            elif tag_lower.startswith(query_lower):
+                prefix_matches.append(TagResult(tag=tag, count=count, category=cat))
+            elif query_lower in tag_lower:
+                contains_matches.append(TagResult(tag=tag, count=count, category=cat))
+
+        # Sort each group by count (descending)
+        exact_matches.sort(key=lambda x: x.count, reverse=True)
+        prefix_matches.sort(key=lambda x: x.count, reverse=True)
+        contains_matches.sort(key=lambda x: x.count, reverse=True)
+
+        # Combine: exact > prefix > contains
+        results = exact_matches + prefix_matches + contains_matches
+
+        return results[:limit]
+
+    def search_artists(self, query: str, limit: int = 20) -> List[TagResult]:
+        """Search only artists"""
+        return self.search(query, limit=limit, category="artist")
+
+    def search_characters(self, query: str, limit: int = 20) -> List[TagResult]:
+        """Search only characters"""
+        return self.search(query, limit=limit, category="character")
+
+    def search_generals(self, query: str, limit: int = 20) -> List[TagResult]:
+        """Search only general tags"""
+        return self.search(query, limit=limit, category="general")
+
+    def get_popular_tags(self, limit: int = 100, category: Optional[str] = None) -> List[TagResult]:
+        """Get most popular tags"""
+        if category == "artist":
+            source = [(k, v, "artist") for k, v in self._artists.items()]
+        elif category == "character":
+            source = [(k, v, "character") for k, v in self._characters.items()]
+        elif category == "general":
+            source = [(k, v, "general") for k, v in self._generals.items()]
+        else:
+            source = [(k, v, c) for k, (v, c) in self._combined.items()]
+
+        # Sort by count
+        source.sort(key=lambda x: x[1], reverse=True)
+
+        return [TagResult(tag=t, count=c, category=cat) for t, c, cat in source[:limit]]
+
+    def get_stats(self) -> Dict[str, int]:
+        """Get statistics about loaded data"""
+        return {
+            "generals": len(self._generals),
+            "artists": len(self._artists),
+            "characters": len(self._characters),
+            "total": len(self._combined)
+        }
+
+
+# Convenience function for simple usage
+def search_tags(query: str, limit: int = 20) -> List[Dict]:
+    """
+    Simple function to search tags.
+
+    Returns list of dicts: [{"tag": str, "count": int, "category": str}, ...]
+    """
+    service = AutocompleteService()
+    results = service.search(query, limit=limit)
+    return [{"tag": r.tag, "count": r.count, "category": r.category} for r in results]
+
+
+def get_autocomplete_service() -> AutocompleteService:
+    """Get the singleton AutocompleteService instance"""
+    return AutocompleteService()
+
+
+# =============================================================================
+# Gradio API Functions
+# =============================================================================
+
+def gradio_search_tags(query: str, limit: int = 20) -> List[List]:
+    """
+    Search tags for Gradio Dataframe component.
+
+    Args:
+        query: Search query
+        limit: Maximum results
+
+    Returns:
+        List of [tag, count, category] for Dataframe display
+    """
+    if not query or len(query.strip()) < 1:
+        return []
+
+    service = get_autocomplete_service()
+    results = service.search(query.strip(), limit=limit)
+
+    return [[r.tag, r.count, r.category] for r in results]
+
+
+def gradio_search_tags_json(query: str, limit: int = 20) -> List[Dict]:
+    """
+    Search tags and return as JSON-serializable list.
+
+    For JavaScript consumption via Gradio's js parameter.
+
+    Returns:
+        [{"tag": str, "count": int, "category": str}, ...]
+    """
+    if not query or len(query.strip()) < 1:
+        return []
+
+    service = get_autocomplete_service()
+    results = service.search(query.strip(), limit=limit)
+
+    return [
+        {"tag": r.tag, "count": r.count, "category": r.category}
+        for r in results
+    ]
+
+
+def gradio_get_completion(current_text: str, cursor_position: int, limit: int = 10) -> List[Dict]:
+    """
+    Get autocomplete suggestions based on current cursor position.
+
+    Extracts the current token (word being typed) and returns suggestions.
+
+    Args:
+        current_text: Full text content
+        cursor_position: Cursor position in text
+        limit: Maximum suggestions
+
+    Returns:
+        List of suggestions with metadata
+    """
+    if not current_text:
+        return []
+
+    # Extract current token (word at cursor)
+    # Tokens are separated by commas
+    text_before_cursor = current_text[:cursor_position]
+
+    # Find the start of current token (after last comma)
+    last_comma = text_before_cursor.rfind(',')
+    token_start = last_comma + 1 if last_comma >= 0 else 0
+
+    # Extract and clean the current token
+    current_token = text_before_cursor[token_start:].strip()
+
+    if len(current_token) < 1:
+        return []
+
+    # Search for matches
+    service = get_autocomplete_service()
+    results = service.search(current_token, limit=limit)
+
+    return [
+        {
+            "tag": r.tag,
+            "count": r.count,
+            "category": r.category,
+            "token_start": token_start,
+            "token_end": cursor_position
+        }
+        for r in results
+    ]
+
+
+def preload_autocomplete_data() -> Dict:
+    """
+    Preload autocomplete data and return statistics.
+    Call this at app startup to warm up the cache.
+
+    Returns:
+        Statistics about loaded data
+    """
+    service = get_autocomplete_service()
+    stats = service.get_stats()
+    return {
+        "status": "loaded",
+        "stats": stats
+    }

core/generation_service.py

@@ -0,0 +1,161 @@
+"""
+NAIA-WEB Generation Service
+High-level orchestration of prompt processing and API calls
+
+Reference: NAIA2.0/core/generation_controller.py
+"""
+
+from dataclasses import dataclass, field
+from typing import Optional, Set, List, Tuple
+from PIL import Image
+
+from .api_service import (
+    NAIAPIService,
+    NAIAPIError,
+    GenerationParameters,
+    CharacterReferenceData,
+    format_api_error
+)
+from .prompt_processor import PromptProcessor, PromptContext
+
+
+@dataclass
+class GenerationRequest:
+    """User-facing generation request"""
+    positive_prompt: str
+    negative_prompt: str
+    resolution: str              # e.g., "832 x 1216"
+    model: str                   # e.g., "NAID4.5F"
+    steps: int = 28
+    scale: float = 5.0
+    cfg_rescale: float = 0.4     # NAIA2.0 default
+    sampler: str = "k_euler"
+    seed: Optional[int] = None
+
+    # Prompt processing options
+    use_quality_tags: bool = True
+    pre_prompt: str = ""
+    post_prompt: str = ""
+    auto_hide_tags: Set[str] = field(default_factory=set)
+
+    # Character prompt data (NAID4.5 feature)
+    # List of (prompt, negative) tuples for each active character
+    character_prompts: List[Tuple[str, str]] = field(default_factory=list)
+
+    # Character reference (NAID4.5 feature)
+    character_reference: Optional[CharacterReferenceData] = None
+
+
+@dataclass
+class GenerationResult:
+    """Result of generation attempt"""
+    success: bool
+    image: Optional[Image.Image] = None
+    seed_used: int = 0
+    error_message: str = ""
+    processed_prompt: str = ""
+    processed_negative: str = ""
+    metadata: dict = field(default_factory=dict)
+
+
+class GenerationService:
+    """
+    High-level service coordinating prompt processing and image generation.
+
+    This service:
+    1. Parses user input into structured parameters
+    2. Runs prompt through processing pipeline
+    3. Calls NAI API
+    4. Returns processed result
+    """
+
+    def __init__(self):
+        self.api_service = NAIAPIService()
+        self.prompt_processor = PromptProcessor()
+
+    def _parse_resolution(self, resolution: str) -> tuple[int, int]:
+        """Parse resolution string like '832 x 1216' into (width, height)"""
+        try:
+            parts = resolution.lower().replace(' ', '').split('x')
+            if len(parts) != 2:
+                raise ValueError()
+            return int(parts[0]), int(parts[1])
+        except (ValueError, IndexError):
+            # Default resolution
+            return 832, 1216
+
+    async def generate(
+        self,
+        token: str,
+        request: GenerationRequest,
+    ) -> GenerationResult:
+        """
+        Execute a generation request.
+
+        Args:
+            token: NAI API token
+            request: Generation parameters
+
+        Returns:
+            GenerationResult with image or error information
+        """
+        try:
+            # Parse resolution
+            width, height = self._parse_resolution(request.resolution)
+
+            # Process prompts through pipeline
+            context = PromptContext(
+                positive_prompt=request.positive_prompt,
+                negative_prompt=request.negative_prompt,
+                use_quality_tags=request.use_quality_tags,
+                pre_prompt=request.pre_prompt,
+                post_prompt=request.post_prompt,
+                auto_hide_tags=request.auto_hide_tags,
+            )
+            processed_context = self.prompt_processor.process(context)
+
+            # Build API parameters
+            params = GenerationParameters(
+                prompt=processed_context.positive_prompt,
+                negative_prompt=processed_context.negative_prompt,
+                width=width,
+                height=height,
+                steps=request.steps,
+                scale=request.scale,
+                cfg_rescale=request.cfg_rescale,
+                sampler=request.sampler,
+                seed=request.seed,
+                model=request.model,
+                character_prompts=request.character_prompts if request.character_prompts else None,
+                character_reference=request.character_reference,
+            )
+
+            # Call API
+            image, metadata = await self.api_service.generate_image(
+                token=token,
+                params=params
+            )
+
+            return GenerationResult(
+                success=True,
+                image=image,
+                seed_used=metadata.get("seed", 0),
+                processed_prompt=processed_context.positive_prompt,
+                processed_negative=processed_context.negative_prompt,
+                metadata=metadata
+            )
+
+        except NAIAPIError as e:
+            return GenerationResult(
+                success=False,
+                error_message=format_api_error(e)
+            )
+        except Exception as e:
+            return GenerationResult(
+                success=False,
+                error_message=f"Unexpected error: {str(e)}"
+            )
+
+    async def close(self):
+        """Cleanup resources"""
+        await self.api_service.close()

core/prompt_processor.py

@@ -0,0 +1,323 @@
+"""
+NAIA-WEB Prompt Processor
+Pipeline-based prompt processing with hooks
+
+Reference: NAIA2.0/core/prompt_processor.py, NAIA2.0/modules/prompt_engineering_module.py
+"""
+
+import re
+from dataclasses import dataclass, field
+from typing import List, Set, Tuple
+
+from utils.constants import QUALITY_TAGS_POSITIVE, QUALITY_TAGS_NEGATIVE
+
+
+@dataclass
+class PromptContext:
+    """
+    Context passed through the prompt processing pipeline.
+
+    Carries all prompt-related data and settings through each stage.
+    """
+    positive_prompt: str
+    negative_prompt: str
+
+    # Processing flags
+    use_quality_tags: bool = True
+
+    # Pre/Post prompt additions
+    pre_prompt: str = ""
+    post_prompt: str = ""
+
+    # Auto hide tags (tags to remove) - supports patterns
+    auto_hide_tags: Set[str] = field(default_factory=set)
+
+    # Removed tags tracking
+    removed_tags: List[str] = field(default_factory=list)
+
+    # Processing log for debugging
+    processing_log: List[str] = field(default_factory=list)
+
+
+class PromptProcessor:
+    """
+    Pipeline-based prompt processor.
+
+    Processing order:
+    1. Add pre-prompt
+    2. Main prompt
+    3. Add post-prompt
+    4. Inject quality tags (if enabled)
+    5. Remove auto-hide tags
+    6. Clean up formatting
+    """
+
+    def process(self, context: PromptContext) -> PromptContext:
+        """
+        Run the full processing pipeline on a prompt context.
+
+        Args:
+            context: Initial prompt context
+
+        Returns:
+            Processed prompt context
+        """
+        # Step 1: Build positive prompt with pre/post
+        context = self._build_positive_prompt(context)
+
+        # Step 2: Inject quality tags
+        if context.use_quality_tags:
+            context = self._inject_quality_tags(context)
+
+        # Step 3: Remove auto-hide tags
+        if context.auto_hide_tags:
+            context = self._remove_auto_hide_tags(context)
+
+        # Step 4: Clean up formatting
+        context = self._cleanup_prompt(context)
+
+        return context
+
+    # Person tag sets for reordering (from NAIA2.0)
+    PERSON_TAGS = {
+        "boys": {"1boy", "2boys", "3boys", "4boys", "5boys", "6+boys"},
+        "girls": {"1girl", "2girls", "3girls", "4girls", "5girls", "6+girls"},
+        "others": {"1other", "2others", "3others", "4others", "5others", "6+others"}
+    }
+    ALL_PERSON_TAGS = PERSON_TAGS["boys"] | PERSON_TAGS["girls"] | PERSON_TAGS["others"]
+
+    def _build_positive_prompt(self, context: PromptContext) -> PromptContext:
+        """
+        Combine pre-prompt, main prompt, and post-prompt.
+
+        Person tags (1girl, 2boys, etc.) are extracted from main prompt
+        and moved to the front in order: boys -> girls -> others.
+
+        Final order: [person tags], [pre-prompt], [main prompt], [post-prompt]
+        """
+        # Parse main prompt into tags
+        main_tags = [t.strip() for t in context.positive_prompt.split(',') if t.strip()]
+
+        # Extract person tags from main prompt
+        person_tags_found = []
+        other_main_tags = []
+
+        for tag in main_tags:
+            if tag.lower() in {pt.lower() for pt in self.ALL_PERSON_TAGS}:
+                person_tags_found.append(tag)
+            else:
+                other_main_tags.append(tag)
+
+        # Sort person tags: boys -> girls -> others
+        sorted_person_tags = sorted(
+            person_tags_found,
+            key=lambda tag: (
+                0 if tag.lower() in {pt.lower() for pt in self.PERSON_TAGS["boys"]} else
+                1 if tag.lower() in {pt.lower() for pt in self.PERSON_TAGS["girls"]} else 2
+            )
+        )
+
+        if sorted_person_tags:
+            context.processing_log.append(f"Person tags moved to front: {', '.join(sorted_person_tags)}")
+
+        # Build final prompt: [person tags], [pre-prompt], [main prompt], [post-prompt]
+        parts = []
+
+        # 1. Person tags (extracted from main prompt)
+        if sorted_person_tags:
+            parts.append(", ".join(sorted_person_tags))
+
+        # 2. Pre-prompt
+        if context.pre_prompt.strip():
+            parts.append(context.pre_prompt.strip())
+            context.processing_log.append("Added pre-prompt")
+
+        # 3. Main prompt (without person tags)
+        if other_main_tags:
+            parts.append(", ".join(other_main_tags))
+
+        # 4. Post-prompt
+        if context.post_prompt.strip():
+            parts.append(context.post_prompt.strip())
+            context.processing_log.append("Added post-prompt")
+
+        context.positive_prompt = ", ".join(parts)
+        return context
+
+    def _inject_quality_tags(self, context: PromptContext) -> PromptContext:
+        """
+        Inject quality tags if enabled.
+
+        Positive quality tags are only appended to the END of the prompt
+        if the user's post_prompt does NOT contain "quality".
+        This allows users to customize quality tags via post_prompt.
+
+        Negative quality tags are always appended.
+        """
+        # Check if post_prompt contains "quality" (case-insensitive)
+        has_quality_in_post = "quality" in context.post_prompt.lower()
+
+        # Append positive quality tags only if post_prompt doesn't have "quality"
+        if not has_quality_in_post:
+            if context.positive_prompt:
+                context.positive_prompt = f"{context.positive_prompt}, {QUALITY_TAGS_POSITIVE}"
+            else:
+                context.positive_prompt = QUALITY_TAGS_POSITIVE
+            context.processing_log.append("Appended positive quality tags (post_prompt has no 'quality')")
+        else:
+            context.processing_log.append("Skipped positive quality tags (post_prompt has 'quality')")
+
+        # Append quality tags to negative prompt (always)
+        if context.negative_prompt:
+            context.negative_prompt = f"{context.negative_prompt}, {QUALITY_TAGS_NEGATIVE}"
+        else:
+            context.negative_prompt = QUALITY_TAGS_NEGATIVE
+
+        context.processing_log.append("Injected negative quality tags")
+        return context
+
+    def _remove_auto_hide_tags(self, context: PromptContext) -> PromptContext:
+        """
+        Remove auto-hide tags from the prompt with pattern support.
+
+        Pattern syntax (from NAIA2.0):
+        - `tag`: Exact match removal
+        - `_pattern_`: Remove tags containing 'pattern' (e.g., _hair_ → blonde hair)
+        - `_pattern`: Remove tags ending with 'pattern'
+        - `pattern_`: Remove tags starting with 'pattern'
+        - `~keyword`: Protect keyword from removal
+        """
+        if not context.auto_hide_tags:
+            return context
+
+        # Parse tags from positive prompt
+        tags = [t.strip() for t in context.positive_prompt.split(',') if t.strip()]
+
+        # Separate protected keywords (starting with ~) and patterns
+        protected_keywords = set()
+        auto_hide_patterns = []
+
+        for item in context.auto_hide_tags:
+            item = item.strip()
+            if not item:
+                continue
+            if item.startswith('~'):
+                # Protected keyword
+                protected_keywords.add(item[1:].strip().lower())
+            else:
+                auto_hide_patterns.append(item)
+
+        # Build removal list
+        to_remove = set()
+
+        for pattern in auto_hide_patterns:
+            pattern_lower = pattern.lower()
+
+            # Pattern matching logic from NAIA2.0
+            if pattern.startswith('__') and pattern.endswith('__') and len(pattern) > 4:
+                # __pattern__: contains match (double underscore)
+                # Remove all underscores for search
+                search_term = pattern[2:-2].replace('_', '')
+                for tag in tags:
+                    if search_term.lower() in tag.lower().replace(' ', ''):
+                        to_remove.add(tag)
+
+            elif pattern.startswith('_') and pattern.endswith('_') and len(pattern) > 2:
+                # _pattern_: contains match (single underscore, space-based)
+                search_term = pattern[1:-1].replace('_', ' ')
+                for tag in tags:
+                    if search_term.lower() in tag.lower():
+                        to_remove.add(tag)
+
+            elif pattern.startswith('_') and not pattern.endswith('_'):
+                # _pattern: ends with match
+                search_term = pattern[1:].replace('_', ' ')
+                for tag in tags:
+                    if tag.lower().endswith(search_term.lower()):
+                        to_remove.add(tag)
+
+            elif pattern.endswith('_') and not pattern.startswith('_'):
+                # pattern_: starts with match
+                search_term = pattern[:-1].replace('_', ' ')
+                for tag in tags:
+                    if tag.lower().startswith(search_term.lower()):
+                        to_remove.add(tag)
+
+            else:
+                # Exact match
+                for tag in tags:
+                    if tag.lower() == pattern_lower:
+                        to_remove.add(tag)
+
+        # Remove protected keywords from removal list
+        if protected_keywords:
+            protected_to_keep = set()
+            for tag in to_remove:
+                tag_lower = tag.lower()
+                for protected in protected_keywords:
+                    if protected in tag_lower or tag_lower == protected:
+                        protected_to_keep.add(tag)
+                        break
+            to_remove -= protected_to_keep
+
+            if protected_to_keep:
+                context.processing_log.append(f"Protected tags: {', '.join(protected_to_keep)}")
+
+        # Apply removal
+        filtered = [t for t in tags if t not in to_remove]
+        context.removed_tags = list(to_remove)
+
+        context.positive_prompt = ", ".join(filtered)
+
+        if to_remove:
+            context.processing_log.append(f"Auto-hide removed {len(to_remove)} tags: {', '.join(sorted(to_remove))}")
+        else:
+            context.processing_log.append("Auto-hide: no tags matched")
+
+        return context
+
+    def _cleanup_prompt(self, context: PromptContext) -> PromptContext:
+        """Clean up prompt formatting"""
+        # Process positive prompt
+        context.positive_prompt = self._clean_text(context.positive_prompt)
+
+        # Process negative prompt
+        context.negative_prompt = self._clean_text(context.negative_prompt)
+
+        context.processing_log.append("Cleaned up formatting")
+        return context
+
+    def _clean_text(self, text: str) -> str:
+        """Clean a single text string"""
+        if not text:
+            return ""
+
+        # Remove extra whitespace
+        text = ' '.join(text.split())
+
+        # Remove duplicate commas
+        text = re.sub(r',\s*,+', ',', text)
+
+        # Remove spaces around commas
+        text = re.sub(r'\s*,\s*', ', ', text)
+
+        # Strip leading/trailing commas and whitespace
+        text = text.strip(' ,')
+
+        return text
+
+
+def parse_tags_from_text(text: str) -> List[str]:
+    """
+    Parse comma-separated tags from text.
+
+    Args:
+        text: Comma-separated tag string
+
+    Returns:
+        List of individual tags (stripped)
+    """
+    if not text:
+        return []
+
+    return [t.strip() for t in text.split(',') if t.strip()]