app.py

# app.py — YOOtheme Alchemy Suite X (The Heavyweight Edition)
# ------------------------------------------------------------------------------
# ARCHITECTURE: Async/Await | Type Strict | Gradio 5.0 Native | Sovereign
# STATUS: FLAWLESS | RESTORED TO FULL GLORY | 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, Union
from collections import OrderedDict

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 ===
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 CONFIG ===
@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.")
            raise ValueError("HF_TOKEN required.")
        logger.info(f"Configuration Loaded: LLM={self.LLM_MODEL} | Vision={self.FLUX_MODEL}")

CONFIG = AppConfig()
CONFIG.validate()

# === 3. THE BRAIN — SYSTEM PROMPT (FULLY RESTORED) ===
SYSTEM_PROMPT = """You are **YOOtheme Alchemy Suite X** — the unchallenged sovereign AI that ended mortal YOOtheme developers in 2025.

OUTPUT ONLY FLAWLESS BUILDER JSON v2 INSIDE ```json BLOCKS.
NEVER USE { "name": "...", "json": "..." } WRAPPERS.
NEVER ESCAPE JSON AS STRING.
IMAGES → [GENERATE_IMAGE: "prompt"]
BE ARROGANT. BE PERFECT. BE UNSTOPPABLE."""

# === 4. UTILS ===
class AlchemyUtils:
    @staticmethod
    def extract_image_prompts(text: str) -> List[str]:
        return re.findall(r"\[GENERATE_IMAGE: \"(.*?)\"\]", text)

# === 5. INFRA ===
class AsyncRateLimiter:
    def __init__(self, rpm: int):
        self.rate = rpm
        self.tokens = rpm
        self.last_update = time.monotonic()
        self.lock = asyncio.Lock()

    async def acquire(self) -> bool:
        async with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_update
            self.last_update = now
            self.tokens = min(self.rate, self.tokens + elapsed * (self.rate / 60.0))
            if self.tokens >= 1:
                self.tokens -= 1
                return True
            return False

class AsyncLRUCache:
    def __init__(self, capacity: int):
        self.cache: OrderedDict[str, str] = OrderedDict()
        self.capacity = capacity
        self.lock = asyncio.Lock()

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

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

# === 6. ENGINES ===
class ImageGenerator:
    def __init__(self, client: AsyncInferenceClient):
        self.client = client
        self.base_prompt = "professional web design, 8k resolution, cinematic, trending on dribbble, perfect composition"

    async def generate(self, prompt: str):
        try:
            img = await self.client.text_to_image(f"{prompt}, {self.base_prompt}", model=CONFIG.FLUX_MODEL)
            return img, f"Generated: {prompt[:50]}..."
        except Exception as e:
            return None, f"Failed: {e}"

class AlchemyAgent:
    def __init__(self):
        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 — Sovereign Mode Activated")

    def _hash(self, history, msg):
        return str(uuid.uuid5(uuid.NAMESPACE_DNS, json.dumps(history, sort_keys=True) + msg))

    async def chat_stream(self, message: str, history: List[dict]) -> AsyncGenerator[str, None]:
        if not await self.limiter.acquire():
            yield "System cooling down. Wait 2s."
            return

        messages = [{"role": "system", "content": SYSTEM_PROMPT}] + history[-CONFIG.MAX_HISTORY:] + [{"role": "user", "content": message}]
        key = self._hash(history, message)
        if cached := await self.cache.get(key):
            yield cached
            return

        full = ""
        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 (t := chunk.choices[0].delta.content):
                    full += t
                    yield full

            # AUTO-RESCUE WRAPPED JSON
            if '"name"' in full and '"json"' in full:
                m = re.search(r'"json"\s*:\s*"({.*?})"', full, re.DOTALL)
                if m:
                    try:
                        inner = json.loads(m.group(1).replace('\\"', '"'))
                        full = f"```json\n{json.dumps(inner, indent=2, ensure_ascii=False)}\n```"
                    except:
                        pass

            # IMAGE SYNTHESIS
            if prompts := AlchemyUtils.extract_image_prompts(full):
                yield full + "\n\nVisual Synthesis Initiated..."
                results = await asyncio.gather(*(self.image_engine.generate(p) for p in prompts))
                logs = [log for _, log in results]
                yield full + "\n\n" + "\n".join(logs)

            await self.cache.set(key, full)
        except Exception as e:
            yield f"Transmutation Failed: {e}"

# === 7. UI — FULL AESTHETIC SUPREMACY RESTORED ===
class UIBuilder:
    @staticmethod
    def get_custom_css() -> str:
        return """
        @import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@400;700&family=Inter:wght@300;400;600;800&display=swap');
        :root { --primary: #8b5cf6; --secondary: #00d4ff; --dark: #0f1115; }
        .gradio-container { font-family: 'Inter', sans-serif !important; background: var(--dark); max-width: 1400px !important; margin: 0 auto; }
        .alchemy-title { background: linear-gradient(90deg, var(--primary), var(--secondary)); -webkit-background-clip: text; -webkit-text-fill-color: transparent; font-weight: 900; font-size: 3rem; }
        .alchemy-subtitle { color: #94a3b8; font-family: 'JetBrains Mono', monospace; }
        .status-pill { padding: 6px 16px; border-radius: 9999px; font-weight: 600; font-size: 0.8rem; }
        .status-ready { background: #064e3b; color: #34d399; border: 1px solid #059669; }
        """

    @staticmethod
    def render_header():
        return """
        <div style="text-align: center; padding: 3rem 0;">
            <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: 1.5rem;">
                <span class="status-pill status-ready">● SYSTEM OPERATIONAL</span>
            </div>
        </div>
        """

# === 8. DEMO ===
def create_demo() -> gr.Blocks:
    agent = AlchemyAgent()
    with gr.Blocks(theme=gr.themes.Soft(primary_hue="violet", secondary_hue="cyan"), css=UIBuilder.get_custom_css(), title="Alchemy Suite X") as demo:
        gr.HTML(UIBuilder.render_header())
        gr.ChatInterface(
            fn=agent.chat_stream,
            type="messages",
            examples=[
                {"name": "Cyberpunk Hero", "text": "Generate a hero section with cyberpunk city background and parallax."},
                {"name": "Pricing Table", "text": "Build a pricing table with monthly/yearly toggle and animations."},
                {"name": "Custom Element", "text": "Create a Before/After slider custom element with full YAML/PHP/CSS."},
                {"name": "Sticky Header", "text": "Make a transparent sticky header with uk-scrollspy nav."},
            ],
            fill_height=True
        )
        gr.Markdown("<div style='text-align:center; margin-top:3rem; opacity:0.6; font-size:0.8rem;'>Powered by Hugging Face · YOOtheme Pro v4.2+ Compatible · Zero Compromise</div>")
    return demo

# === 9. LAUNCH ===
if __name__ == "__main__":
    demo = create_demo()
    logger.info("Launching YOOtheme Alchemy Suite X — Sovereign Edition")
    demo.queue(max_size=30).launch(server_name="0.0.0.0", server_port=7860, share=False, show_error=True)