import os import json import asyncio from pathlib import Path from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel, Field from typing import List, Optional from bs4 import BeautifulSoup import google.generativeai as genai from netlify_deployer import deploy_to_netlify from supabase_client import get_supabase_client, provision_tenant, SUPABASE_URL, SUPABASE_ANON_KEY from telnyx_client import spawn_telnyx_assistant from image_generate_client import generate_3d_icon from visual_polish_agent import visual_polish_loop # Setup directories BASE_DIR = Path(__file__).parent.parent TEMPLATES_DIR = BASE_DIR / "Sunaiva" / "website_factory" / "templates" OUTPUT_DIR = BASE_DIR / "Sunaiva" / "website_factory" / "generated_sites" OUTPUT_DIR.mkdir(parents=True, exist_ok=True) # Fetch API Key GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "") if not GEMINI_API_KEY: try: with open(BASE_DIR / "config" / "secrets.env", "r") as f: for line in f: if line.startswith("GEMINI_API_KEY="): GEMINI_API_KEY = line.split("=")[1].strip().strip('"\'') except Exception: pass if GEMINI_API_KEY: genai.configure(api_key=GEMINI_API_KEY) app = FastAPI(title="AgileAdapt Website Generation Factory", version="2.0.0") # --------------------------------------------------------- # Pydantic Schemas for Strict Structured Outputs # --------------------------------------------------------- class ProspectIntakeRequest(BaseModel): business_name: str transcript_summary: str target_audience: str class ResearchOutput(BaseModel): business_name: str core_services: List[str] brand_archetype: str target_audience: str customer_objections: List[str] class CopywriterOutput(BaseModel): hero_headline: str hero_subheadline: str service_titles: List[str] service_descriptions: List[str] faq_questions: List[str] faq_answers: List[str] call_to_action: str class DesignerOutput(BaseModel): primary_color_hex: str background_gradient_class: str lucide_icons: List[str] class AssemblyOutput(BaseModel): custom_css: str content: str # --------------------------------------------------------- # Core Agentic Pipeline Functions # --------------------------------------------------------- async def run_researcher(transcript: str, biz_name: str) -> str: """Agent 1: Distills raw conversation into structured JSON.""" model = genai.GenerativeModel("gemini-2.5-flash") prompt = f"Analyze this transcript/summary for {biz_name}: {transcript}. Extract the core services, the brand vibe/archetype, audience, and any customer objections mentioned into strict JSON." response = await model.generate_content_async( prompt, generation_config=genai.types.GenerationConfig(response_mime_type="application/json") ) return response.text async def run_copywriter(research_json: str) -> str: """Agent 2: Writes high-converting copy handling objections.""" model = genai.GenerativeModel("gemini-3.1-pro-preview") # Elite Copy prompt = f"You are a direct response copywriter. Given this business brief: {research_json}. Write high-converting English copy for a single page website. Write an FAQ that explicitly resolves the customer objections. Return strict JSON." response = await model.generate_content_async( prompt, generation_config=genai.types.GenerationConfig(response_mime_type="application/json") ) return response.text async def run_designer(research_json: str) -> str: """Agent 3: Selects aesthetic tokens.""" model = genai.GenerativeModel("gemini-2.5-flash") prompt = f"Given this business profile: {research_json}. Select a modern Tailwind background gradient, a primary hex color for glowing elements, and 3 specific Lucide icon names relevant to the services. Return strict JSON." response = await model.generate_content_async( prompt, generation_config=genai.types.GenerationConfig(response_mime_type="application/json") ) return response.text async def run_architect(copy_json: str, design_json: str, biz_name: str) -> str: """Agent 4: The Build Engine.""" model = genai.GenerativeModel("gemini-3.1-pro-preview") # Elite Assembly prompt = f"""You are the Master Web Architect. Assemble the final HTML content for {biz_name}. Use Glassmorphism UI (backdrop-filter: blur(16px), subtle borders). Use these exact aesthetic tokens: {design_json}. Use this exact copy: {copy_json}. MUST USE Lucide icons: ``. CRITICAL: You MUST embed the following brand logo image tag prominently in the hero section or top-left nav: `Brand Icon` (Customize classes as needed but use that exact src string). Return strictly JSON with strictly two keys: 'custom_css' and 'content' (which contains all the HTML).""" response = await model.generate_content_async( prompt, generation_config=genai.types.GenerationConfig(response_mime_type="application/json") ) return response.text # --------------------------------------------------------- # Factory Orchestration Logic # --------------------------------------------------------- async def execute_p1_pipeline(request: ProspectIntakeRequest): print(f"[{request.business_name}] Initiating Zero-Touch Pipeline...") # Run Agent 1 research_json = await run_researcher(request.transcript_summary, request.business_name) print(f"[{request.business_name}] Research Extracted.") # Fan-Out Phase: Run Agents 2 and 3 + Telnyx Backend API in parallel rs_data = {} try: rs_data = json.loads(research_json) except Exception: pass core_offer = rs_data.get("core_offer", "general services") target_arch = rs_data.get("brand_archetype", "professional") copywriter_task = asyncio.create_task(run_copywriter(research_json)) designer_task = asyncio.create_task(run_designer(research_json)) telnyx_task = asyncio.create_task(spawn_telnyx_assistant(request.business_name, core_offer, target_arch)) logo_task = asyncio.create_task(generate_3d_icon(request.business_name, core_offer)) copy_json, design_json, assistant_id, logo_url = await asyncio.gather(copywriter_task, designer_task, telnyx_task, logo_task) print(f"[{request.business_name}] Copy, Aesthetics, Voice Agent, and 3D Logo generated.") # Gather Phase: Run Agent 4 final_json = await run_architect(copy_json, design_json, request.business_name) print(f"[{request.business_name}] Final DOM Assembled.") try: final_data = json.loads(final_json) custom_css = final_data.get("custom_css", "") body_content = final_data.get("content", "") except json.JSONDecodeError: print("Architect JSON parsing failed. Executing fallback.") body_content = final_json custom_css = "" # Inject into Template with open(TEMPLATES_DIR / "base_glassmorphism.html", "r", encoding="utf-8") as f: template = f.read() # Wrap the LLM's CSS in style tags if it didn't include them if custom_css and "" final_html = template.replace("{{business_name}}", request.business_name) final_html = final_html.replace("{{custom_css}}", custom_css) final_html = final_html.replace("{{content}}", body_content) # Generate temporary tenant UUID to inject before provision import uuid tenant_id = str(uuid.uuid4()) final_html = final_html.replace("{{tenant_id}}", tenant_id) final_html = final_html.replace("{{supabase_url}}", SUPABASE_URL or "SUPERBASE_URL_MISSING") final_html = final_html.replace("{{supabase_anon_key}}", SUPABASE_ANON_KEY or "SUPABASE_ANON_MISSING") final_html = final_html.replace("{{telnyx_assistant_id}}", assistant_id) final_html = final_html.replace("{{logo_url}}", logo_url) # AST Self-Healing Pass try: soup = BeautifulSoup(final_html, "html.parser") final_html = str(soup) print(f"[{request.business_name}] AST Self-Healing complete. DOM structure validated.") except Exception as e: print(f"[{request.business_name}] AST parsing failed: {e}") # --- Phase 6: The Autonomous Visual Polish Loop (AIVPL) --- print(f"[{request.business_name}] Initiating Design Critic (AIVPL)...") final_html = await visual_polish_loop(final_html, max_iterations=2, target_score=90) print(f"[{request.business_name}] AIVPL Phase Complete. Final Aesthetics locked.") # Save Output safe_name = "".join([c for c in request.business_name if c.isalpha() or c.isdigit()]).lower() output_path = OUTPUT_DIR / f"{safe_name}.html" with open(output_path, "w", encoding="utf-8") as f: f.write(final_html) print(f"[{request.business_name}] [SUCCESS] Production Build Complete! Saved to {output_path}") # Live Netlify API Deployment project_slug = f"agile-adapt-{safe_name}" deploy_url = await deploy_to_netlify(html_content=final_html, project_name=project_slug) # Phase 2: Supabase Provisioning sb_client = get_supabase_client() if sb_client: # We manually pass the pre-generated ID we injected in HTML # Ensure our provision_tenant function handles it, or we insert it directly. # UPDATE: we need to modify provision to accept id, but since we need it in HTML now, # let's just insert it here directly or update provision_tenant. try: sb_client.table("tenants").insert({ "id": tenant_id, "business_name": request.business_name, "subdomain": project_slug, "telnyx_assistant_id": assistant_id, "html_payload": final_html }).execute() print(f"[{request.business_name}] \U0001f512 Database tenant row provisioned securely.") except Exception as e: print(f"[{request.business_name}] Supabase provision failed: {e}") return deploy_url @app.post("/api/generate") async def trigger_factory(request: ProspectIntakeRequest, background_tasks: BackgroundTasks): """ Webhook target for Telnyx AI Assistant. Accepts conversational summary and triggers the async pipeline. """ if not GEMINI_API_KEY: raise HTTPException(status_code=500, detail="Gemini API Key missing.") # Schedule the build process in the background to return 200 OK instantly to Telnyx background_tasks.add_task(execute_p1_pipeline, request) return { "status": "Engine Initialized", "message": f"Pipeline started for {request.business_name}. ETA to live URL: 60s.", "tenant_id": "pending_generation" } if __name__ == "__main__": import uvicorn print("\nšŸš€ AgileAdapt Factory API Starting on Port 8000...\n") uvicorn.run("website_factory_orchestrator:app", host="0.0.0.0", port=8000, reload=True)