import io import json import os import logging import uuid import asyncio import zipfile from datetime import datetime from contextlib import asynccontextmanager from typing import Any, Optional, List import asyncpg import httpx from fastapi import FastAPI, File, HTTPException, UploadFile, BackgroundTasks from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from starlette.concurrency import run_in_threadpool logging.basicConfig(level=logging.INFO) logger = logging.getLogger("dolor3v-unified-backend") # --------------------------------------------------------------------------- # Config: LLM Providers # --------------------------------------------------------------------------- GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") CEREBRAS_API_KEY = os.environ.get("CEREBRAS_API_KEY", "") OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "") GROQ_URL = "https://api.groq.com/openai/v1/chat/completions" CEREBRAS_URL = "https://api.cerebras.ai/v1/chat/completions" OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" VIRTUAL_MODELS = { "auto": None, "groq-llama-3.3-70b": ("groq", "llama-3.3-70b-versatile"), "cerebras-glm-4.7": ("cerebras", "zai-glm-4.7"), "openrouter-gpt-oss-120b-free": ("openrouter", "openai/gpt-oss-120b:free"), } AUTO_CHAIN = [ ("groq", "llama-3.3-70b-versatile"), ("cerebras", "zai-glm-4.7"), ("openrouter", "openai/gpt-oss-120b:free"), ] # --------------------------------------------------------------------------- # Config: Postgres with Dual-Mode In-Memory Fallbacks # --------------------------------------------------------------------------- POSTGRES_SERVER = os.environ.get("POSTGRES_SERVER", "") POSTGRES_PORT = os.environ.get("POSTGRES_PORT", "5432") POSTGRES_USER = os.environ.get("POSTGRES_USER", "") POSTGRES_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "") POSTGRES_DB = os.environ.get("POSTGRES_DB", "") POSTGRES_SSL = os.environ.get("POSTGRES_SSL", "false").strip().lower() in ("1", "true", "yes") db_pool: Optional[asyncpg.Pool] = None IN_MEMORY_PROJECTS = {} IN_MEMORY_VERSIONS = {} IN_MEMORY_BUILDS = {} IN_MEMORY_ASSETS = [] async def init_db_pool() -> asyncpg.Pool: database_url = os.environ.get("DATABASE_URL", "").strip() if database_url: pool = await asyncpg.create_pool(dsn=database_url, min_size=1, max_size=5, ssl="require" if os.environ.get("POSTGRES_SSL","").lower() in ("1","true","yes") else None) return pool pool = await asyncpg.create_pool( host=POSTGRES_SERVER, port=int(POSTGRES_PORT), user=POSTGRES_USER, password=POSTGRES_PASSWORD, database=POSTGRES_DB, min_size=1, max_size=5, timeout=10, ssl=True if POSTGRES_SSL else None, ) async with pool.acquire() as conn: await conn.execute("""CREATE TABLE IF NOT EXISTS dolor3v_projects (page_id TEXT PRIMARY KEY, data JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())""") await conn.execute("""CREATE TABLE IF NOT EXISTS dolor3v_assets (id SERIAL PRIMARY KEY, name TEXT NOT NULL, url TEXT NOT NULL, type TEXT NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())""") await conn.execute("""CREATE TABLE IF NOT EXISTS dolor3v_versions (id SERIAL PRIMARY KEY, project_id TEXT NOT NULL, name TEXT NOT NULL, data JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT now())""") await conn.execute("""CREATE TABLE IF NOT EXISTS dolor3v_builds (id TEXT PRIMARY KEY, project_id TEXT NOT NULL, platform TEXT NOT NULL, status TEXT NOT NULL, logs TEXT NOT NULL, download_url TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT now())""") return pool @asynccontextmanager async def lifespan(app: FastAPI): global db_pool try: db_pool = await init_db_pool() logger.info("Database pool initialized — postgres_connected=True") except Exception as exc: logger.error( "Failed to initialize Postgres pool. Falling back to memory: %s", exc, ) db_pool = None yield if db_pool is not None: await db_pool.close() app = FastAPI(title="Dolor3v AI Workspace Core Engine", lifespan=lifespan) app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) # --------------------------------------------------------------------------- # Config: Google Drive # --------------------------------------------------------------------------- GDRIVE_SERVICE_ACCOUNT_JSON = os.environ.get("GDRIVE_SERVICE_ACCOUNT_JSON", "") GDRIVE_FOLDER_ID = os.environ.get("GDRIVE_FOLDER_ID", "") _drive_service_cache: Optional[Any] = None def _get_drive_service_sync() -> Any: global _drive_service_cache if _drive_service_cache is not None: return _drive_service_cache if not GDRIVE_SERVICE_ACCOUNT_JSON: raise RuntimeError("GDRIVE_SERVICE_ACCOUNT_JSON is not configured") from google.oauth2 import service_account from googleapiclient.discovery import build info = json.loads(GDRIVE_SERVICE_ACCOUNT_JSON) credentials = service_account.Credentials.from_service_account_info(info, scopes=["https://www.googleapis.com/auth/drive"]) _drive_service_cache = build("drive", "v3", credentials=credentials, cache_discovery=False) return _drive_service_cache def _upload_one_sync(service: Any, filename: str, mimetype: str, content: bytes) -> dict: from googleapiclient.http import MediaIoBaseUpload media = MediaIoBaseUpload(io.BytesIO(content), mimetype=mimetype or "application/octet-stream", resumable=False) file_metadata = {"name": filename, "parents": [GDRIVE_FOLDER_ID]} created = service.files().create(body=file_metadata, media_body=media, fields="id, name, webContentLink, webViewLink").execute() service.permissions().create(fileId=created["id"], body={"role": "reader", "type": "anyone"}).execute() src = created.get("webContentLink") or f"https://drive.google.com/uc?export=view&id={created['id']}" return {"src": src, "name": created.get("name", filename), "type": "image"} # --------------------------------------------------------------------------- # Models # --------------------------------------------------------------------------- class ChatMessage(BaseModel): role: str; content: str class ChatRequest(BaseModel): model: str = "auto"; messages: list[ChatMessage]; temperature: Optional[float] = None; max_tokens: Optional[int] = None class ProjectSaveRequest(BaseModel): pageId: str; data: dict[str, Any] class PublishRequest(BaseModel): project_id: str; platform: str = "web" class NativeGenerateRequest(BaseModel): prompt: str; platform: str = "flutter"; project_id: str class NativeBuildRequest(BaseModel): project_id: str; platform: str; config: Optional[dict[str, Any]] = None class CodeGenerateRequest(BaseModel): prompt: str; language: str; context: Optional[str] = None class CodeRefactorRequest(BaseModel): code: str; instructions: str; language: str class CodeFixRequest(BaseModel): code: str; error: str; language: str class VersionSaveRequest(BaseModel): project_id: str; name: str; data: dict[str, Any] class VersionRestoreRequest(BaseModel): project_id: str; version_id: int # --------------------------------------------------------------------------- # AI Engine Fallback & Completions # --------------------------------------------------------------------------- async def call_groq(model: str, messages: list[dict], **kwargs) -> dict[str, Any]: if not GROQ_API_KEY: raise RuntimeError("GROQ_API_KEY not configured") async with httpx.AsyncClient(timeout=60) as client: resp = await client.post(GROQ_URL, headers={"Authorization": f"Bearer {GROQ_API_KEY}"}, json={"model": model, "messages": messages, **kwargs}) if resp.status_code!= 200: raise RuntimeError(f"Groq {resp.status_code}: {resp.text[:500]}") return resp.json() async def call_cerebras(model: str, messages: list[dict], **kwargs) -> dict[str, Any]: if not CEREBRAS_API_KEY: raise RuntimeError("CEREBRAS_API_KEY not configured") async with httpx.AsyncClient(timeout=60) as client: resp = await client.post(CEREBRAS_URL, headers={"Authorization": f"Bearer {CEREBRAS_API_KEY}"}, json={"model": model, "messages": messages, **kwargs}) if resp.status_code!= 200: raise RuntimeError(f"Cerebras {resp.status_code}: {resp.text[:500]}") return resp.json() async def call_openrouter(model: str, messages: list[dict], **kwargs) -> dict[str, Any]: if not OPENROUTER_API_KEY: raise RuntimeError("OPENROUTER_API_KEY not configured") async with httpx.AsyncClient(timeout=60) as client: resp = await client.post(OPENROUTER_URL, headers={"Authorization": f"Bearer {OPENROUTER_API_KEY}"}, json={"model": model, "messages": messages, **kwargs}) if resp.status_code!= 200: raise RuntimeError(f"OpenRouter {resp.status_code}: {resp.text[:500]}") return resp.json() PROVIDER_FUNCS = {"groq": call_groq, "cerebras": call_cerebras, "openrouter": call_openrouter} async def get_ai_completion(system_prompt: str, user_prompt: str, temperature: float = 0.3) -> str: messages = [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}] errors = [] for provider, real_model in AUTO_CHAIN: try: result = await PROVIDER_FUNCS[provider](real_model, messages, temperature=temperature) return result.get("choices", [])[0].get("message", {}).get("content", "") except Exception as exc: logger.warning("Provider %s failed: %s", provider, exc); errors.append(str(exc)) raise HTTPException(status_code=502, detail={"error": "All AI providers failed", "details": errors}) BUILD_JOBS = {} from mcp_routes import router as mcp_router app.include_router(mcp_router) from builder_routes import router as builder_router app.include_router(builder_router)