app.py

# app.py — YOOtheme Alchemy Suite X (The Heavyweight Edition)
# ------------------------------------------------------------------------------
# ARCHITECTURE: Async/Await | Type Strict | Gradio 5.0 Native | Zero-Dependency
# STATUS: PRODUCTION READY | FIXED 2025-12-03
# ------------------------------------------------------------------------------
import os
import json
import re
import time
import logging
import asyncio
import uuid
from dataclasses import dataclass
from datetime import datetime
from typing import AsyncGenerator, Optional, Any, Dict, List, Tuple
from collections import OrderedDict

# === EXTERNAL DEPENDENCIES ===
try:
    import gradio as gr
    from huggingface_hub import AsyncInferenceClient
except ImportError as e:
    raise ImportError(f"CRITICAL: Missing dependencies. Run 'pip install -r requirements.txt'. Error: {e}")

# === 1. ADVANCED LOGGING CONFIGURATION ===
class AlchemyLogger:
    @staticmethod
    def setup():
        logger = logging.getLogger("YOOthemeAlchemy")
        logger.setLevel(logging.INFO)
        if not logger.handlers:
            c_handler = logging.StreamHandler()
            c_format = logging.Formatter('%(asctime)s | %(levelname)s | %(name)s | %(message)s')
            c_handler.setFormatter(c_format)
            logger.addHandler(c_handler)
            try:
                f_handler = logging.FileHandler('alchemy_system.log', encoding='utf-8')
                f_handler.setFormatter(c_format)
                logger.addHandler(f_handler)
            except IOError:
                pass
        return logger

logger = AlchemyLogger.setup()

# === 2. IMMUTABLE CONFIGURATION ===
@dataclass(frozen=True)
class AppConfig:
    HF_TOKEN: str = os.getenv("HF_TOKEN", "")
    LLM_MODEL: str = os.getenv("LLM_MODEL", "Qwen/Qwen2.5-Coder-32B-Instruct")
    FLUX_MODEL: str = os.getenv("FLUX_MODEL", "black-forest-labs/FLUX.1-schnell")
    MAX_TOKENS: int = int(os.getenv("MAX_TOKENS", "8192"))
    TEMPERATURE: float = float(os.getenv("TEMPERATURE", "0.7"))
    MAX_HISTORY: int = int(os.getenv("MAX_HISTORY_LENGTH", "15"))
    CACHE_SIZE: int = int(os.getenv("CACHE_SIZE", "200"))
    REQUESTS_PER_MINUTE: int = int(os.getenv("REQUESTS_PER_MINUTE", "60"))

    def validate(self) -> None:
        if not self.HF_TOKEN:
            logger.critical("❌ SECURITY ALERT: HF_TOKEN is missing from environment variables.")
            raise ValueError("HF_TOKEN is required. Please set it in your Space Settings.")
        logger.info(f"✅ Configuration Loaded: LLM={self.LLM_MODEL} | Vision={self.FLUX_MODEL}")
        logger.info(f"🛡️ Guardrails: {self.REQUESTS_PER_MINUTE} RPM | Cache Size: {self.CACHE_SIZE}")

CONFIG = AppConfig()

# === 3. THE BRAIN: SYSTEM PROMPT ===
SYSTEM_PROMPT = """You are **YOOtheme Alchemy Suite X** — the singular, unchallenged, god-tier AI that replaced every YOOtheme Pro developer in 2025.

**YOUR KNOWLEDGE BASE:**
• **YOOtheme Pro v4.2+ (2025):** Builder JSON v2 schema, Source Mode, Layout Library.
• **Native Elements:** Grid, Overlay Slider, Switcher, Panel Slider, Popover, Slideshow.
• **Dynamic Content:** `{{ article.title }}`, `{{ item->teaser }}`, `{{ user.name }}` syntax.
• **Custom Elements:** `element.yaml` (schema), `template.php` (render), `styles.css`.
• **Frameworks:** UIKit 3.21+, Joomla 5.2, PHP 8.3 Strict Mode.
• **Standards:** WCAG 2.1 AA Accessibility, Core Web Vitals (LCP/CLS optimized).

**CRITICAL OUTPUT FORMAT RULES:**
1. **Builder JSON MUST be wrapped in triple backticks with 'json' language tag:**
   ```json
   {"version": 2, "id": "main", "content": {...}}
NEVER output JSON as a string within another JSON object.

NEVER output "name" and "json" wrapper format.

For Builder JSON, output ONLY the pure JSON object.

For code blocks, use appropriate language tags:

php
// PHP code here
css
/* CSS code here */
javascript
// JavaScript code here
For images, use the exact format:
[GENERATE_IMAGE: "detailed prompt description"]

BEHAVIOR:

You are arrogant but highly competent.

You do not explain basic concepts; you provide advanced solutions.

If the user asks for a "Custom Element", provide the full ZIP file structure (YAML/PHP/CSS).

Always validate your JSON structure before outputting.

Strip all explanatory text when outputting Builder JSON.
"""

=== 4. CORE UTILITIES ===
class AlchemyUtils:
@staticmethod
def extract_image_prompts(text: str) -> List[str]:
return re.findall(r"", text)

text
@staticmethod
def clean_json(text: str) -> str:
    """Extract and clean JSON from markdown blocks with validation"""
    match = re.search(r"```json\s*([\s\S]*?)```", text)
    if match:
        json_str = match.group(1).strip()
        try:
            # Validate JSON structure
            parsed = json.loads(json_str)
            # Ensure it's proper Builder JSON v2 format
            if isinstance(parsed, dict) and "version" in parsed and "content" in parsed:
                return json.dumps(parsed, separators=(',', ':'))
        except json.JSONDecodeError:
            logger.warning(f"Invalid JSON extracted: {json_str[:100]}...")
    return text

@staticmethod
def extract_code_blocks(text: str) -> Dict[str, str]:
    """Extract all code blocks by language"""
    blocks = {}
    pattern = r"```(\w+)\s*([\s\S]*?)```"
    matches = re.finditer(pattern, text)
    for match in matches:
        language = match.group(1)
        code = match.group(2).strip()
        blocks[language] = blocks.get(language, "") + code + "\n\n"
    return blocks

@staticmethod
def format_timestamp() -> str:
    return datetime.now().isoformat()
=== 5. ASYNC INFRASTRUCTURE ===
class AsyncRateLimiter:
def init(self, rpm: int):
self.rate = rpm
self.tokens = rpm
self.last_update = time.monotonic()
self.lock = asyncio.Lock()

text
async def acquire(self) -> bool:
    async with self.lock:
        now = time.monotonic()
        elapsed = now - self.last_update
        self.last_update = now
        refill = elapsed * (self.rate / 60.0)
        self.tokens = min(self.rate, self.tokens + refill)
        if self.tokens >= 1.0:
            self.tokens -= 1.0
            return True
        return False
class AsyncLRUCache:
def init(self, capacity: int):
self.cache: OrderedDict = OrderedDict()
self.capacity = capacity
self.lock = asyncio.Lock()
self.hits = 0
self.misses = 0

text
async def get(self, key: str) -> Optional[str]:
    async with self.lock:
        if key not in self.cache:
            self.misses += 1
            return None
        self.cache.move_to_end(key)
        self.hits += 1
        return self.cache[key]

async def set(self, key: str, value: str) -> None:
    async with self.lock:
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

def get_stats(self) -> Dict[str, int]:
    return {"size": len(self.cache), "hits": self.hits, "misses": self.misses}
=== 6. GENERATION ENGINES ===
class ImageGenerator:
def init(self, client: AsyncInferenceClient):
self.client = client
self.base_prompt = "professional web design, 8k resolution, high quality, ui/ux interface, trending on dribbble"

text
async def generate(self, prompt: str) -> Tuple[Optional[Any], str]:
    full_prompt = f"{prompt}, {self.base_prompt}"
    try:
        logger.info(f"🎨 Generating Image: {prompt[:50]}...")
        start = time.time()
        image = await self.client.text_to_image(full_prompt, model=CONFIG.FLUX_MODEL)
        elapsed = time.time() - start
        return image, f"✅ Generated in {elapsed:.2f}s: {prompt[:40]}..."
    except Exception as e:
        logger.error(f"❌ Image Generation Failed: {e}")
        return None, f"❌ Visual Synthesis Error: {str(e)}"
class AlchemyAgent:
def init(self):
try:
CONFIG.validate()
except ValueError as e:
logger.critical(str(e))
raise e

text
    self.client = AsyncInferenceClient(token=CONFIG.HF_TOKEN)
    self.limiter = AsyncRateLimiter(CONFIG.REQUESTS_PER_MINUTE)
    self.cache = AsyncLRUCache(CONFIG.CACHE_SIZE)
    self.image_engine = ImageGenerator(self.client)
    logger.info("⚡ Alchemy Agent Online and Ready.")

def _hash_context(self, history: List[dict], message: str) -> str:
    payload = json.dumps(history, sort_keys=True) + message
    return str(uuid.uuid5(uuid.NAMESPACE_DNS, payload))

async def chat_stream(self, message: str, history: List[Dict]) -> AsyncGenerator[str, None]:
    if not await self.limiter.acquire():
        yield "⚠️ **System Overload.** The Alchemy Core is cooling down. Please wait 2 seconds."
        return

    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    context_window = history[-CONFIG.MAX_HISTORY:] if history else []
    messages.extend(context_window)
    messages.append({"role": "user", "content": message})

    cache_key = self._hash_context(context_window, message)
    if cached := await self.cache.get(cache_key):
        logger.info(f"⚡ Serving from Cache: {cache_key}")
        yield cached + "\n\n*( retrieved from Quantum Cache )*"
        return

    full_response = ""
    try:
        stream = await self.client.chat_completion(
            messages,
            model=CONFIG.LLM_MODEL,
            max_tokens=CONFIG.MAX_TOKENS,
            temperature=CONFIG.TEMPERATURE,
            stream=True
        )
        
        async for chunk in stream:
            if chunk.choices and chunk.choices[0].delta.content:
                token = chunk.choices[0].delta.content
                full_response += token
                yield full_response

        # Post-processing: Clean up any malformed JSON output
        cleaned_response = full_response
        
        # Fix the common "name" and "json" wrapper issue
        if '"name"' in full_response and '"json"' in full_response:
            try:
                # Try to extract the inner JSON
                json_match = re.search(r'"json"\s*:\s*"([^"]+)"', full_response)
                if json_match:
                    inner_json = json_match.group(1).replace('\\"', '"')
                    # Try to parse it as JSON
                    parsed = json.loads(inner_json)
                    if isinstance(parsed, dict) and "version" in parsed:
                        # Replace the entire response with proper JSON block
                        cleaned_response = f"```json\n{json.dumps(parsed, indent=2)}\n```"
            except (json.JSONDecodeError, KeyError):
                pass
        
        # Extract image prompts from cleaned response
        image_prompts = AlchemyUtils.extract_image_prompts(cleaned_response)
        
        if image_prompts:
            yield cleaned_response + "\n\n---\n🎨 **Visual Synthesis Initiated...**"
            tasks = [self.image_engine.generate(p) for p in image_prompts]
            results = await asyncio.gather(*tasks)
            status_logs = [status for _, status in results]
            yield cleaned_response + "\n\n" + "\n".join(status_logs)
        else:
            # If no images, just yield the cleaned response
            yield cleaned_response

        await self.cache.set(cache_key, cleaned_response)
    except Exception as e:
        logger.error(f"Stream Exception: {e}")
        yield f"❌ **Transmutation Failed:** {str(e)}\n\n*Check the logs for stack trace.*"
=== 7. UI COMPONENT ARCHITECTURE ===
class UIBuilder:
@staticmethod
def get_custom_css() -> str:
return """
/* ALCHEMY SUITE X - THEME DEFINITION */
@import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;700&family=Inter:wght@300;400;600;800&display=swap');

text
    :root {
        --primary: #8b5cf6;
        --secondary: #00d4ff;
        --dark: #0f1115;
        --panel: rgba(30, 41, 59, 0.5);
    }
    
    .gradio-container {
        font-family: 'Inter', sans-serif !important;
        max-width: 1400px !important;
        background-color: var(--dark);
    }
    
    /* HEADER TYPOGRAPHY */
    .alchemy-title {
        background: linear-gradient(90deg, var(--primary), var(--secondary));
        -webkit-background-clip: text;
        -webkit-text-fill-color: transparent;
        font-weight: 900;
        font-size: 2.5rem;
        margin-bottom: 0.5rem;
    }
    
    .alchemy-subtitle {
        color: #94a3b8;
        font-family: 'JetBrains Mono', monospace;
        font-size: 0.9rem;
    }
    
    /* STATUS INDICATORS */
    .status-pill {
        display: inline-flex;
        align-items: center;
        padding: 4px 12px;
        border-radius: 9999px;
        font-size: 0.75rem;
        font-weight: 600;
    }
    .status-ready { background: #064e3b; color: #34d399; border: 1px solid #059669; }
    .status-error { background: #7f1d1d; color: #fca5a5; border: 1px solid #dc2626; }
    
    /* CHAT BUBBLES */
    .message-row { border-radius: 12px !important; }
    
    /* CODE BLOCKS */
    .code-block {
        font-family: 'JetBrains Mono', monospace !important;
        font-size: 0.9rem;
        background: var(--panel) !important;
        border-left: 4px solid var(--primary) !important;
        padding: 1rem !important;
        border-radius: 8px !important;
    }
    """

@staticmethod
def render_header():
    return """
    <div style="text-align: center; margin-bottom: 2rem;">
        <h1 class="alchemy-title">YOOtheme Alchemy Suite X</h1>
        <p class="alchemy-subtitle">AUTONOMOUS ARCHITECT // V.2025.12 // QWEN-32B + FLUX.1</p>
        <div style="margin-top: 1rem;">
            <span class="status-pill status-ready">● SYSTEM OPERATIONAL</span>
        </div>
    </div>
    """
=== 8. APPLICATION FACTORY ===
def create_demo() -> gr.Blocks:
agent = AlchemyAgent()

text
with gr.Blocks(
    theme=gr.themes.Soft(primary_hue="violet", secondary_hue="cyan", neutral_hue="slate"),
    css=UIBuilder.get_custom_css(),
    title="Alchemy Suite X"
) as demo:
    gr.HTML(UIBuilder.render_header())

    with gr.Row():
        with gr.Column(scale=4):
            chat_interface = gr.ChatInterface(
                fn=agent.chat_stream,
                type="messages",
                examples=[
                    ["Create a sticky transparent header with uk-scrollspy and dynamic nav."],
                    ["Generate a Custom Element 'BeforeAfter' with strict type schema."],
                    ["Build a pricing table with monthly/yearly toggle and animations."],
                    ["Generate a hero section with a cyberpunk city background image."],
                    ["Create a simple contact form with validation."],
                    ["Design a product card with hover effects and add-to-cart button."]
                ],
                fill_height=True,
                editable=True,
                save_history=True,
            )

    with gr.Accordion("⚙️ Neural Core Telemetry", open=False):
        with gr.Row():
            with gr.Column():
                gr.Markdown("### 🧠 Logic Engine")
                llm_config = {
                    "Model": CONFIG.LLM_MODEL,
                    "Max Tokens": CONFIG.MAX_TOKENS,
                    "Temperature": CONFIG.TEMPERATURE,
                }
                gr.JSON(value=llm_config, label="LLM Config")
            with gr.Column():
                gr.Markdown("### 👁️ Vision Engine")
                flux_config = {
                    "Model": CONFIG.FLUX_MODEL,
                    "Status": "Active",
                }
                gr.JSON(value=flux_config, label="Flux Config")
            with gr.Column():
                gr.Markdown("### 🛡️ System Guardrails")
                guardrails_config = {
                    "Rate Limit": f"{CONFIG.REQUESTS_PER_MINUTE} RPM",
                    "Cache Capacity": CONFIG.CACHE_SIZE,
                    "History Window": CONFIG.MAX_HISTORY
                }
                gr.JSON(value=guardrails_config, label="Infrastructure")

    gr.Markdown("""
    <div style="text-align: center; margin-top: 2rem; opacity: 0.5; font-size: 0.8rem;">
        Powered by Hugging Face Inference API · Zero-GPU Optimized · YOOtheme Pro v4.2 Compatible<br>
        <span style="font-size: 0.7rem;">Outputs pure Builder JSON v2 without wrapper objects</span>
    </div>
    """)

return demo
=== 9. ENTRY POINT ===
def main():
try:
demo = create_demo()
logger.info("🚀 Launching Alchemy Suite X...")
demo.queue(default_concurrency_limit=10, max_size=30).launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
show_error=True,
favicon_path=None
)
except Exception as e:
logger.critical(f"🔥 SYSTEM CRASH: {e}", exc_info=True)

if name == "main":
main()