Spaces:
Running
Running
| 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: | |
| <!DOCTYPE html> | |
| ...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'(<!DOCTYPE html[\s\S]*?</html>)', 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 | |
| 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"} | |
| ) | |
| 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} | |
| 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 | |