Spaces:
Sleeping
Sleeping
File size: 4,440 Bytes
42f12bd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 | """
Main FastAPI application for LLM Code Deployment Project
Receives task requests, generates code with LLM, deploys to GitHub, notifies evaluation API
"""
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel
from typing import List, Dict, Optional
import os
from dotenv import load_dotenv
import logging
# Import our custom modules
from llm_generator import generate_app_code, generate_readme
from github_deployer import deploy_to_github
from notifier import notify_evaluation_api
# Load environment variables
load_dotenv()
# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Initialize FastAPI app
app = FastAPI(title="LLM Code Deployment API")
# Pydantic models for request validation
class Attachment(BaseModel):
name: str
url: str # data URI format
class TaskRequest(BaseModel):
email: str
secret: str
task: str
round: int
nonce: str
brief: str
checks: List[str]
evaluation_url: str
attachments: Optional[List[Attachment]] = []
class TaskResponse(BaseModel):
status: str
message: str
# Background task processor
async def process_task(request: TaskRequest):
"""
Background task to process the entire workflow:
1. Generate code with LLM
2. Deploy to GitHub
3. Notify evaluation API
"""
try:
logger.info(f"Processing task: {request.task}, Round: {request.round}")
# Step 1: Generate HTML code and README using LLM
logger.info("Generating code with LLM...")
html_content = generate_app_code(
brief=request.brief,
checks=request.checks,
attachments=request.attachments
)
readme_content = generate_readme(
task_name=request.task,
brief=request.brief,
checks=request.checks
)
logger.info("✓ Code generation complete")
# Step 2: Deploy to GitHub
logger.info("Deploying to GitHub...")
repo_url, commit_sha, pages_url = deploy_to_github(
task_name=request.task,
html_content=html_content,
readme_content=readme_content,
round_num=request.round
)
logger.info(f"✓ Deployed to GitHub: {repo_url}")
logger.info(f"✓ Pages URL: {pages_url}")
# Step 3: Notify evaluation API
logger.info("Notifying evaluation API...")
success = notify_evaluation_api(
evaluation_url=request.evaluation_url,
email=request.email,
task=request.task,
round=request.round,
nonce=request.nonce,
repo_url=repo_url,
commit_sha=commit_sha,
pages_url=pages_url
)
if success:
logger.info("✓ Evaluation API notified successfully")
else:
logger.error("✗ Failed to notify evaluation API")
except Exception as e:
logger.error(f"✗ Error processing task: {str(e)}")
raise
@app.get("/")
async def root():
"""Health check endpoint"""
return {
"status": "online",
"message": "LLM Code Deployment API is running",
"endpoints": {
"build": "/build (POST)"
}
}
@app.post("/build", response_model=TaskResponse)
async def build_app(request: TaskRequest, background_tasks: BackgroundTasks):
"""
Main endpoint that receives task requests and processes them
Flow:
1. Verify secret
2. Return HTTP 200 immediately
3. Process task in background
"""
# CRITICAL: Verify secret
expected_secret = os.getenv("MY_SECRET")
if request.secret != expected_secret:
logger.warning(f"Invalid secret attempt for task: {request.task}")
raise HTTPException(status_code=401, detail="Invalid secret")
logger.info(f"✓ Secret verified for task: {request.task}")
# Add background task to process asynchronously
background_tasks.add_task(process_task, request)
# Return HTTP 200 immediately (as required)
return TaskResponse(
status="accepted",
message=f"Task {request.task} accepted and processing in background"
)
# Run with: uvicorn main:app --reload --port 8000
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|