139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import requests
|
|
import yaml
|
|
from fastapi import Depends, FastAPI, HTTPException
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
app = FastAPI(title="AI CI/CD Assistant")
|
|
|
|
|
|
def _env(name: str, default: str) -> str:
|
|
value = os.getenv(name)
|
|
return value if value else default
|
|
|
|
|
|
# Example for local Ollama: http://localhost:11434/api/generate
|
|
OLLAMA_API_URL = _env("OLLAMA_API_URL", "http://host.docker.internal:11434/api/generate")
|
|
REQUIRE_AUTH_FOR_GENERATE = _env("REQUIRE_AUTH_FOR_GENERATE", "true").lower() in ("1", "true", "yes", "on")
|
|
|
|
from backend.auth import create_access_token, get_current_user, hash_password, verify_password # noqa: E402
|
|
from backend.db import engine, get_db # noqa: E402
|
|
from backend.models import Base, User # noqa: E402
|
|
|
|
|
|
class GenerationRequest(BaseModel):
|
|
description: str
|
|
platform: str = "GitHub Actions" # or GitLab CI
|
|
model: str = "codellama"
|
|
|
|
|
|
def validate_config(config_text: str):
|
|
try:
|
|
yaml.safe_load(config_text)
|
|
return True, "Success"
|
|
except Exception as e:
|
|
return False, str(e)
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.on_event("startup")
|
|
def _startup():
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
|
|
class RegisterRequest(BaseModel):
|
|
email: str
|
|
password: str
|
|
full_name: str | None = None
|
|
|
|
|
|
class LoginRequest(BaseModel):
|
|
email: str
|
|
password: str
|
|
|
|
|
|
@app.post("/auth/register")
|
|
def register(req: RegisterRequest, db: Session = Depends(get_db)):
|
|
email = req.email.strip().lower()
|
|
if not email or not req.password:
|
|
raise HTTPException(status_code=400, detail="Email and password are required")
|
|
|
|
exists = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
|
|
if exists:
|
|
raise HTTPException(status_code=409, detail="Email already registered")
|
|
|
|
user = User(
|
|
email=email,
|
|
full_name=req.full_name.strip() if req.full_name else None,
|
|
password_hash=hash_password(req.password),
|
|
is_active=True,
|
|
is_admin=False,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
|
|
return {"id": str(user.id), "email": user.email, "full_name": user.full_name, "is_admin": user.is_admin}
|
|
|
|
|
|
@app.post("/auth/login")
|
|
def login(req: LoginRequest, db: Session = Depends(get_db)):
|
|
email = req.email.strip().lower()
|
|
user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
|
|
if not user or not verify_password(req.password, user.password_hash):
|
|
raise HTTPException(status_code=401, detail="Invalid credentials")
|
|
if not user.is_active:
|
|
raise HTTPException(status_code=403, detail="User is inactive")
|
|
|
|
token = create_access_token(user_id=str(user.id))
|
|
return {"access_token": token, "token_type": "bearer"}
|
|
|
|
|
|
@app.get("/auth/me")
|
|
def me(user: User = Depends(get_current_user)):
|
|
return {"id": str(user.id), "email": user.email, "full_name": user.full_name, "is_admin": user.is_admin}
|
|
|
|
|
|
@app.post("/generate")
|
|
async def generate_pipeline(
|
|
req: GenerationRequest,
|
|
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
|
):
|
|
prompt = (
|
|
f"Write only a valid {req.platform} YAML configuration for: {req.description}. "
|
|
"No explanations, just code."
|
|
)
|
|
|
|
payload = {"model": req.model, "prompt": prompt, "stream": False}
|
|
|
|
try:
|
|
response = requests.post(OLLAMA_API_URL, json=payload, timeout=60)
|
|
response.raise_for_status()
|
|
raw_text = response.json().get("response", "")
|
|
|
|
clean_code = raw_text.replace("```yaml", "").replace("```", "").strip()
|
|
is_valid, error_msg = validate_config(clean_code)
|
|
|
|
return {
|
|
"status": "ok" if is_valid else "error",
|
|
"config": clean_code,
|
|
"validation_error": error_msg if not is_valid else None,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Ollama error: {str(e)}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
|
|
uvicorn.run(app, host="0.0.0.0", port=int(_env("PORT", "8000")))
|