theavinash02/llm-code-deployer

1import os
2import requests
3import time
4from fastapi import FastAPI, BackgroundTasks, HTTPException, Request
5from pydantic import BaseModel, HttpUrl
6from typing import List, Optional
7from dotenv import load_dotenv
8
9from utils import decode_attachment
10from llm_handler import generate_app_code, generate_readme
11from github_handler import deploy_project, create_or_get_repo, get_file_content
12
13load_dotenv()
14app = FastAPI()
15STUDENT_SECRET = os.getenv("STUDENT_SECRET")
16
17# --- Pydantic Models for automatic request validation ---
18class Attachment(BaseModel):
19    name: str
20    url: str
21
22class TaskRequest(BaseModel):
23    email: str
24    secret: str
25    task: str
26    round: int
27    nonce: str
28    brief: str
29    checks: List[str]
30    evaluation_url: HttpUrl
31    attachments: Optional[List[Attachment]] = None
32
33# --- Background Task Processing Logic ---
34def process_task(data: TaskRequest):
35    """The main function to handle the build and deploy process."""
36    print(f"Processing task: {data.task} (Round {data.round})")
37    brief = data.brief
38    
39    if data.attachments:
40        for att in data.attachments:
41            decoded = decode_attachment(att.dict())
42            if decoded:
43                brief += f"\n\n--- Attachment: {decoded['name']} ---\n{decoded['content']}"
44
45    code_to_update = None
46    if data.round > 1:
47        repo = create_or_get_repo(data.task)
48        code_to_update = get_file_content(repo, 'index.html')
49        if not code_to_update:
50            print("Warning: Could not fetch existing code for round 2.")
51
52    app_code = generate_app_code(brief, existing_code=code_to_update)
53    if not app_code:
54        print("Error: LLM failed to generate code."); return
55
56    readme_content = generate_readme(brief, app_code)
57    
58    files_to_commit = {'index.html': app_code, 'README.md': readme_content}
59    repo_details = deploy_project(data.task, files_to_commit, data.round)
60
61    payload = {
62        "email": data.email, "task": data.task, "round": data.round,
63        "nonce": data.nonce, "repo_url": repo_details['repo_url'],
64        "commit_sha": repo_details['commit_sha'], "pages_url": repo_details['pages_url'],
65    }
66    
67    notify_evaluation(str(data.evaluation_url), payload)
68
69def notify_evaluation(url: str, payload: dict):
70    """Sends a POST request to the evaluation URL with exponential backoff retries."""
71    max_retries, delay = 5, 1
72    for i in range(max_retries):
73        try:
74            print(f"Notifying evaluation server at {url}...")
75            response = requests.post(url, json=payload, headers={'Content-Type': 'application/json'})
76            if response.status_code == 200:
77                print("Notification successful!"); return
78            else:
79                print(f"Attempt {i+1} failed with status {response.status_code}: {response.text}")
80        except requests.exceptions.RequestException as e:
81            print(f"Attempt {i+1} failed with exception: {e}")
82        
83        if i < max_retries - 1:
84            print(f"Retrying in {delay} seconds..."); time.sleep(delay); delay *= 2
85    
86    print("Failed to notify evaluation server after all retries.")
87
88# --- API Endpoint ---
89@app.post("/api-endpoint")
90async def handle_request(task_request: TaskRequest, background_tasks: BackgroundTasks):
91    """API endpoint to receive, validate, and process task requests."""
92    if task_request.secret != STUDENT_SECRET:
93        raise HTTPException(status_code=401, detail="Unauthorized: Invalid secret")
94    
95    background_tasks.add_task(process_task, task_request)
96    
97    return {"message": "Request received and is being processed."}
98
99@app.get("/")
100def read_root():
101    return {"status": "ok", "message": "API is running. Send POST requests to /api-endpoint"}