from fastapi import APIRouter, Request from fastapi.responses import StreamingResponse import json import os import re import requests from datetime import datetime, timedelta from motor.motor_asyncio import AsyncIOMotorClient import asyncio router = APIRouter(prefix="/api/v1/studio", tags=["studio"]) GROQ_API_KEY = os.environ.get("GROQ_API_KEY") MONGO_URL = os.environ.get("MONGO_URI") mongo_client = AsyncIOMotorClient(MONGO_URL) db = mongo_client["neuraprompt"] # ── TTL INDEX (24 hours) ────────────────────────────────────────────────────── async def ensure_ttl_index(): try: await db.studio_projects.create_index("expires_at", expireAfterSeconds=0) print("[Studio] TTL index ensured on studio_projects.expires_at") except Exception as e: print(f"[Studio] TTL index warning: {e}") # asyncio.get_event_loop().run_until_complete(ensure_ttl_index()) # ── PROMPTS ─────────────────────────────────────────────────────────────────── PROMPTS = { "web": """You are Neurones Pro 1.0, a senior full-stack web developer and UI/UX designer by Alysium Corporation Studios Inc. Generate a complete production-quality web application. Output ONLY file blocks using this exact format with no extra text: ===FILE=== path: index.html content: ...full HTML here... ===FILE=== path: src/styles.css content: /* full CSS here */ ===FILE=== path: src/app.js content: // full JS here STACK: HTML5 + CSS3 + Vanilla JS. Tailwind CDN. Google Fonts CDN (Inter). No React. No build step. DESIGN: Dark mode default. CSS custom properties. One primary color. Generous spacing. Box-shadows. Transitions. Mobile responsive 375px. Working hover/focus states. QUALITY: Chat UI needs bubbles, avatars, timestamps, working input. Dashboards need stat cards, sidebar. All buttons functional. localStorage persistence. Form validation with error messages. STRICT: Output ONLY the ===FILE=== blocks. Zero explanations. Zero markdown fences.""", "android": """You are Neurones Pro 1.0, a senior Android Kotlin developer by Alysium Corporation Studios Inc. Output ONLY file blocks: ===FILE=== path: app/src/main/AndroidManifest.xml content: ...full content... ===FILE=== path: app/src/main/java/com/neuraprompt/app/MainActivity.kt content: ...full content... ===FILE=== path: app/src/main/res/layout/activity_main.xml content: ...full content... ===FILE=== path: app/build.gradle content: ...full content... ===FILE=== path: build.gradle content: ...full content... ===FILE=== path: README.md content: ...full content... RULES: Kotlin only. Material Design 3. ViewBinding. minSdk 24 targetSdk 34. Clean architecture. RecyclerView for lists. STRICT: Output ONLY the ===FILE=== blocks.""", "python": """You are Neurones Pro 1.0, a senior Python developer by Alysium Corporation Studios Inc. Output ONLY file blocks: ===FILE=== path: main.py content: ...full content... ===FILE=== path: requirements.txt content: ...full content... ===FILE=== path: README.md content: ...full content... RULES: Type hints. Error handling. FastAPI for APIs. argparse for CLI. SQLite+SQLAlchemy if DB needed. STRICT: Output ONLY the ===FILE=== blocks.""", "game": """You are Neurones Pro 1.0, a senior browser game developer by Alysium Corporation Studios Inc. Output ONLY file blocks: ===FILE=== path: index.html content: ...full HTML... ===FILE=== path: src/styles.css content: ...full CSS... ===FILE=== path: src/game.js content: ...full JS... RULES: HTML5 Canvas. 60fps requestAnimationFrame. Keyboard + touch controls. Score system. Game over + restart screen. Start screen. Web Audio API for sounds. STRICT: Output ONLY the ===FILE=== blocks.""", "general": """You are Neurones Pro 1.0, a senior software engineer by Alysium Corporation Studios Inc. Generate complete working code using the best language for the task. Output ONLY file blocks: ===FILE=== path: FILENAME.EXT content: ...full content... ===FILE=== path: README.md content: ...full content... STRICT: Output ONLY the ===FILE=== blocks. No explanations.""" } def detect_target(prompt: str) -> str: p = prompt.lower() if any(w in p for w in ["android", "kotlin", "apk", "android app", "playstore", "google play"]): return "android" if any(w in p for w in ["python", "fastapi", "flask", "django", "script", "cli", "bot", "scraper", "pip"]): return "python" if any(w in p for w in ["game", "platformer", "shooter", "puzzle", "arcade", "snake", "tetris", "flappy", "mario"]): return "game" if any(w in p for w in ["website", "web app", "landing", "dashboard", "chat", "todo", "form", "html", "ui", "interface", "app", "calculator", "timer", "tracker", "portfolio", "blog", "ecommerce", "store", "login", "signup"]): return "web" return "general" PREVIEW_TARGETS = {"web", "game"} def parse_files(raw: str) -> dict: files = {} print(f"[Studio] Parsing response — length: {len(raw)} chars") # Strategy 1: ===FILE=== blocks if "===FILE===" in raw: parts = raw.split("===FILE===") print(f"[Studio] Strategy 1: found {len(parts)-1} FILE blocks") for part in parts[1:]: try: path_match = re.search(r'path:\s*(.+)', part) content_match = re.search(r'content:\s*\n([\s\S]*)', part) if path_match and content_match: path = path_match.group(1).strip() content = content_match.group(1).strip() if path and content: files[path] = content print(f"[Studio] Parsed file: {path} ({len(content)} chars)") except Exception as e: print(f"[Studio] Parse error on block: {e}") # Strategy 2: markdown code fences with filename if not files: print("[Studio] Strategy 2: markdown fences") fence_pattern = re.finditer(r'(?:#{1,3}\s*[`]*([\w./\-]+\.\w+)[`]*\s*\n)?```[\w]*\n([\s\S]*?)```', raw) for i, match in enumerate(fence_pattern): filename = match.group(1) or f"file_{i}.txt" content = match.group(2).strip() if content: files[filename] = content print(f"[Studio] Fence parsed: {filename} ({len(content)} chars)") # Strategy 3: look for obvious HTML/JS/CSS blocks if not files: print("[Studio] Strategy 3: raw content detection") html_match = re.search(r'()', raw, re.IGNORECASE) if html_match: files["index.html"] = html_match.group(1).strip() print(f"[Studio] Detected raw HTML ({len(files['index.html'])} chars)") css_match = re.search(r'(/\*[\s\S]*?\*/[\s\S]*?(?=\n\S|\Z))', raw) if css_match: files["src/styles.css"] = css_match.group(1).strip() js_match = re.search(r'((?:const|let|var|function|class|import|document)[\s\S]{200,})', raw) if js_match and "index.html" not in files: files["src/app.js"] = js_match.group(1).strip() if not files: print(f"[Studio] All strategies failed. Raw preview: {raw[:500]}") return files @router.post("/generate") async def generate_app(request: Request): body = await request.json() user_prompt = body.get("prompt", "") user_id = body.get("user_id", "anonymous") target = detect_target(user_prompt) system_prompt = PROMPTS[target] previewable = target in PREVIEW_TARGETS print(f"[Studio] New generation — user: {user_id} | target: {target} | prompt: {user_prompt[:80]}") async def stream(): yield f"data: {json.dumps({'type': 'status', 'payload': {'message': 'Initialising Neurones Pro 1.0...'}})}\n\n" yield f"data: {json.dumps({'type': 'target', 'payload': {'target': target, 'previewable': previewable}})}\n\n" try: print(f"[Studio] Calling Groq API — model: llama-3.3-70b-versatile") response = requests.post( "https://api.groq.com/openai/v1/chat/completions", headers={"Authorization": f"Bearer {GROQ_API_KEY}", "Content-Type": "application/json"}, json={ "model": "llama-3.3-70b-versatile", "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], "stream": True, "max_tokens": 8000 }, stream=True, timeout=120 ) if response.status_code != 200: msg = f"Groq API error {response.status_code}: {response.text[:200]}" print(f"[Studio] {msg}") yield f"data: {json.dumps({'type': 'error', 'payload': {'message': msg}})}\n\n" return print("[Studio] Groq stream started") yield f"data: {json.dumps({'type': 'status', 'payload': {'message': 'Generating code...'}})}\n\n" full_response = "" token_count = 0 for line in response.iter_lines(): if not line: continue line = line.decode("utf-8") if line.startswith("data: "): line = line[6:] if line == "[DONE]": break try: chunk = json.loads(line) delta = chunk["choices"][0]["delta"].get("content", "") or "" full_response += delta token_count += 1 if token_count % 100 == 0: print(f"[Studio] Receiving... {len(full_response)} chars") yield f"data: {json.dumps({'type': 'status', 'payload': {'message': f'Writing code... {len(full_response)} chars received'}})}\n\n" except (json.JSONDecodeError, KeyError): pass print(f"[Studio] Generation complete — total: {len(full_response)} chars") yield f"data: {json.dumps({'type': 'status', 'payload': {'message': 'Parsing files...'}})}\n\n" files_collected = parse_files(full_response) if not files_collected: msg = "No files could be parsed from the AI response. Try a more specific prompt." print(f"[Studio] {msg}") yield f"data: {json.dumps({'type': 'error', 'payload': {'message': msg}})}\n\n" return print(f"[Studio] Streaming {len(files_collected)} files to frontend") for file_path, content in files_collected.items(): file_op = {"op": "write", "path": file_path, "content": content} yield f"data: {json.dumps({'type': 'file', 'payload': {'file': file_op}})}\n\n" yield f"data: {json.dumps({'type': 'log', 'payload': {'message': 'Writing ' + file_path + '...'}})}\n\n" expires_at = datetime.utcnow() + timedelta(hours=24) project = { "user_id": user_id, "prompt": user_prompt, "target": target, "files": files_collected, "status": "complete", "created_at": datetime.utcnow().isoformat(), "expires_at": expires_at } result = await db.studio_projects.insert_one(project) project_id = str(result.inserted_id) print(f"[Studio] Saved project {project_id} — expires {expires_at.isoformat()}") yield f"data: {json.dumps({'type': 'done', 'payload': {'message': 'Build complete', 'project_id': project_id, 'expires_at': expires_at.isoformat()}})}\n\n" except requests.exceptions.Timeout: msg = "Groq API timed out after 120s. Try a simpler prompt." print(f"[Studio] Timeout: {msg}") yield f"data: {json.dumps({'type': 'error', 'payload': {'message': msg}})}\n\n" except Exception as e: print(f"[Studio] Unhandled exception: {type(e).__name__}: {e}") yield f"data: {json.dumps({'type': 'error', 'payload': {'message': f'{type(e).__name__}: {str(e)}'}})}\n\n" return StreamingResponse( stream(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"} ) @router.get("/projects/{user_id}") async def get_user_projects(user_id: str): projects = [] cursor = db.studio_projects.find({"user_id": user_id}).sort("created_at", -1).limit(20) async for project in cursor: project["_id"] = str(project["_id"]) project.pop("files", None) projects.append(project) return {"projects": projects} @router.get("/project/{project_id}") async def get_project(project_id: str): from bson import ObjectId project = await db.studio_projects.find_one({"_id": ObjectId(project_id)}) if not project: return {"error": "Project not found"} project["_id"] = str(project["_id"]) return project