# 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 """

YOOtheme Alchemy Suite X

AUTONOMOUS ARCHITECT // V.2025.12 // QWEN-32B + FLUX.1

● SYSTEM OPERATIONAL
""" # === 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("
Powered by Hugging Face · YOOtheme Pro v4.2+ Compatible · Zero Compromise
") 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)