Ashoklumarsimhadri's picture
Upload 23 files
b7cdd54 verified
Raw
History Blame
7.27 kB
from __future__ import annotations
import os
from pathlib import Path
from typing import Any
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel, Field
from .services.agent_controller import agent_controller
from .services.file_service import ensure_within_directory
from .services.zip_service import create_project_zip
load_dotenv()
ROOT_DIR = Path(__file__).resolve().parent.parent
APP_DIR = ROOT_DIR / "app"
GENERATED_DIR = ROOT_DIR / "generated"
DEFAULT_PORT = 7860
app = FastAPI(title="Project Agent")
app.mount("/static", StaticFiles(directory=APP_DIR / "static"), name="static")
templates = Jinja2Templates(directory=str(APP_DIR / "templates"))
class StackSelectionPayload(BaseModel):
language: str = "Auto"
frontend: str = "Auto"
backend: str = "Auto"
database: str = "Auto"
aiTools: str = "Auto"
deployment: str = "Auto"
class EnvVariablePayload(BaseModel):
name: str
value: str = ""
description: str = ""
class RequiredInputPayload(BaseModel):
name: str
required: bool = True
example: str = ""
whereToAdd: str = ".env"
purpose: str = ""
class SuggestRequest(BaseModel):
idea: str = Field(..., min_length=1)
selectedStack: StackSelectionPayload | None = None
generationMode: str = "fast"
finalRequirements: str = ""
class AgentAnalyzeRequest(BaseModel):
idea: str = Field(..., min_length=1)
class AgentQuestionPayload(BaseModel):
id: str
question: str
type: str
options: list[str] = Field(default_factory=list)
default: str = ""
reason: str = ""
class AgentAnalyzeResponse(BaseModel):
understanding: str
assumptions: list[str] = Field(default_factory=list)
suggestedStack: StackSelectionPayload = Field(default_factory=StackSelectionPayload)
stackReasons: list[str] = Field(default_factory=list)
questions: list[AgentQuestionPayload] = Field(default_factory=list)
detectedProjectType: str = "full-stack"
confidence: int = 0
class AgentFinalizeRequest(BaseModel):
idea: str = Field(..., min_length=1)
answers: dict[str, Any] = Field(default_factory=dict)
suggestedStack: StackSelectionPayload = Field(default_factory=StackSelectionPayload)
class AgentFinalizeResponse(BaseModel):
finalRequirements: str = ""
selectedStack: StackSelectionPayload = Field(default_factory=StackSelectionPayload)
assumptions: list[str] = Field(default_factory=list)
class ModulePayload(BaseModel):
name: str
purpose: str = ""
keyFiles: list[str] = Field(default_factory=list)
class FilePayload(BaseModel):
path: str
content: str = ""
class PreviewPayload(BaseModel):
projectName: str
detectedUserChoices: list[str] = Field(default_factory=list)
selectedStack: StackSelectionPayload = Field(default_factory=StackSelectionPayload)
chosenStack: list[str] = Field(default_factory=list)
assumptions: list[str] = Field(default_factory=list)
summary: str = ""
problemStatement: str = ""
architecture: list[str] = Field(default_factory=list)
modules: list[ModulePayload] = Field(default_factory=list)
packageRequirements: list[str] = Field(default_factory=list)
installCommands: list[str] = Field(default_factory=list)
runCommands: list[str] = Field(default_factory=list)
requiredInputs: list[RequiredInputPayload] = Field(default_factory=list)
envVariables: list[EnvVariablePayload] = Field(default_factory=list)
fileTree: str = ""
files: list[FilePayload] = Field(default_factory=list)
class ZipRequest(BaseModel):
preview: PreviewPayload
@app.get("/", response_class=HTMLResponse)
async def index(request: Request) -> HTMLResponse:
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/api/suggest")
async def suggest_project(payload: SuggestRequest) -> JSONResponse:
idea = payload.idea.strip()
if not idea:
raise HTTPException(status_code=400, detail="Please enter a project idea.")
try:
preview = await agent_controller.generate_files(
idea,
payload.selectedStack.model_dump() if payload.selectedStack else None,
payload.generationMode,
payload.finalRequirements,
)
except RuntimeError as exc:
raise HTTPException(status_code=502, detail=str(exc)) from exc
return JSONResponse(preview)
@app.post("/api/agent/analyze")
async def analyze_agent(payload: AgentAnalyzeRequest) -> JSONResponse:
idea = payload.idea.strip()
if not idea:
raise HTTPException(status_code=400, detail="Please enter a project idea.")
return JSONResponse(agent_controller.analyze_idea(idea))
@app.post("/api/agent/finalize")
async def finalize_agent(payload: AgentFinalizeRequest) -> JSONResponse:
idea = payload.idea.strip()
if not idea:
raise HTTPException(status_code=400, detail="Please enter a project idea.")
finalized = agent_controller.finalize_requirements(
idea,
payload.answers,
payload.suggestedStack.model_dump(),
)
return JSONResponse(finalized)
@app.post("/api/zip")
async def build_zip(payload: ZipRequest) -> JSONResponse:
try:
normalized_preview = agent_controller.validate_project(payload.preview.model_dump())
result = create_project_zip(normalized_preview, GENERATED_DIR)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except OSError as exc:
raise HTTPException(status_code=500, detail=f"Could not create ZIP: {exc}") from exc
return JSONResponse(result)
@app.get("/downloads/{filename}")
async def download_zip(filename: str) -> FileResponse:
download_path = GENERATED_DIR / filename
if download_path.name != filename:
raise HTTPException(status_code=404, detail="File not found.")
if download_path.suffix != ".zip":
raise HTTPException(status_code=404, detail="File not found.")
if not download_path.exists():
raise HTTPException(status_code=404, detail="File not found.")
if not ensure_within_directory(GENERATED_DIR, download_path):
raise HTTPException(status_code=404, detail="File not found.")
return FileResponse(
download_path,
media_type="application/zip",
filename=download_path.name,
)
@app.exception_handler(HTTPException)
async def http_exception_handler(_: Request, exc: HTTPException) -> JSONResponse:
if exc.status_code == 404:
return JSONResponse(status_code=404, content={"detail": exc.detail})
return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail})
@app.exception_handler(Exception)
async def unhandled_exception_handler(_: Request, exc: Exception) -> JSONResponse:
return JSONResponse(status_code=500, content={"detail": f"Unexpected server error: {exc}"})
if __name__ == "__main__":
import uvicorn
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=int(os.getenv("PORT", str(DEFAULT_PORT))),
)