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 from sqlalchemy.exc import SQLAlchemyError 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: http://host:11434/api/generate — для чата берётся тот же хост + /api/chat OLLAMA_API_URL = _env("OLLAMA_API_URL", "http://host.docker.internal:11434/api/generate") def _ollama_chat_url() -> str: base = OLLAMA_API_URL.replace("/api/generate", "").rstrip("/") return f"{base}/api/chat" OLLAMA_CHAT_URL = _env("OLLAMA_CHAT_URL", _ollama_chat_url()) REQUIRE_AUTH_FOR_GENERATE = _env("REQUIRE_AUTH_FOR_GENERATE", "true").lower() in ("1", "true", "yes", "on") from backend.auth import ( # noqa: E402 AUTH_VERSION, create_access_token, get_current_user, hash_password, password_needs_update, verify_password, ) 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" class ChatMessageIn(BaseModel): role: str content: str class ChatRequest(BaseModel): model: str messages: list[ChatMessageIn] 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(): # lightweight DB probe to surface connection issues early db_ok = True db_error = None try: with engine.connect() as conn: conn.exec_driver_sql("SELECT 1") except Exception as e: db_ok = False db_error = str(e) return {"status": "ok", "db_ok": db_ok, "db_error": db_error, "auth_version": AUTH_VERSION} @app.on_event("startup") def _startup(): # Ensure tables exist; if DB is unreachable, endpoints will surface the error. try: Base.metadata.create_all(bind=engine) except Exception: # Avoid crashing the service hard; /health will show db_error. pass 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") pw_bytes_len = len(req.password.encode("utf-8")) try: 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} except HTTPException: raise except SQLAlchemyError as e: db.rollback() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") except Exception as e: db.rollback() raise HTTPException( status_code=500, detail=f"Internal error: {str(e)} (password_bytes_len={pw_bytes_len})", ) @app.post("/auth/login") def login(req: LoginRequest, db: Session = Depends(get_db)): email = req.email.strip().lower() if not req.password: raise HTTPException(status_code=400, detail="Password is required") try: 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") if password_needs_update(user.password_hash): user.password_hash = hash_password(req.password) db.add(user) db.commit() token = create_access_token(user_id=str(user.id)) return {"access_token": token, "token_type": "bearer"} except HTTPException: raise except SQLAlchemyError as e: raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") except Exception as e: raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}") @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("/chat") def chat_with_ollama( req: ChatRequest, user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None, ): if not req.messages: raise HTTPException(status_code=400, detail="messages must not be empty") for m in req.messages: if m.role not in ("user", "assistant", "system"): raise HTTPException(status_code=400, detail=f"invalid role: {m.role}") payload = { "model": req.model, "messages": [{"role": m.role, "content": m.content} for m in req.messages], "stream": False, } try: response = requests.post(OLLAMA_CHAT_URL, json=payload, timeout=120) response.raise_for_status() data = response.json() msg = data.get("message") or {} content = msg.get("content", "") if not content and isinstance(data.get("response"), str): content = data["response"] return {"role": "assistant", "content": content} except requests.HTTPError as e: detail = e.response.text if e.response is not None else str(e) raise HTTPException(status_code=502, detail=f"Ollama HTTP error: {detail}") except Exception as e: raise HTTPException(status_code=500, detail=f"Ollama error: {str(e)}") @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")))