507 lines
16 KiB
Python
507 lines
16 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
from typing import Any
|
||
|
||
import requests
|
||
import yaml
|
||
from fastapi import Depends, FastAPI, HTTPException
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy import delete, 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 AutonomousPipeline, Base, User, UserChatMessage # noqa: E402
|
||
from backend.pipeline_migrate import migrate_definition # 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]
|
||
|
||
|
||
class JenkinsPhaseConfigIn(BaseModel):
|
||
job: str = ""
|
||
server_url: str = ""
|
||
parameters: dict[str, str] = Field(default_factory=dict)
|
||
|
||
|
||
class LowCodePhaseIn(BaseModel):
|
||
id: str
|
||
name: str = ""
|
||
description: str = ""
|
||
jenkins: JenkinsPhaseConfigIn = Field(default_factory=JenkinsPhaseConfigIn)
|
||
|
||
|
||
class GateCheckIn(BaseModel):
|
||
"""Условие для перехода конвейера на следующий этап (имя переменной → ожидаемое значение)."""
|
||
|
||
variable: str = ""
|
||
expected: str = ""
|
||
|
||
|
||
class StageIn(BaseModel):
|
||
"""Этап (например стенд): внутри него выполняются фазы; checks — гейты перед выходом на следующий этап."""
|
||
|
||
id: str
|
||
name: str = ""
|
||
description: str = ""
|
||
checks: list[GateCheckIn] = Field(default_factory=list)
|
||
phases: list[LowCodePhaseIn] = Field(default_factory=list)
|
||
|
||
|
||
class AutonomousPipelineSave(BaseModel):
|
||
title: str = ""
|
||
stages: list[StageIn] = Field(default_factory=list)
|
||
|
||
|
||
class ValidateGateChecksIn(BaseModel):
|
||
"""Проверка условий перехода: для каждого элемента checks сравнивается values[variable] с expected."""
|
||
|
||
checks: list[GateCheckIn] = Field(default_factory=list)
|
||
values: dict[str, Any] = Field(default_factory=dict)
|
||
|
||
|
||
class GateCheckFailure(BaseModel):
|
||
variable: str
|
||
reason: str # "missing" | "mismatch"
|
||
expected: str | None = None
|
||
actual: str | None = None
|
||
|
||
|
||
class ValidateGateChecksOut(BaseModel):
|
||
ok: bool
|
||
failures: list[GateCheckFailure] = Field(default_factory=list)
|
||
|
||
|
||
def evaluate_gate_checks(
|
||
checks: list[GateCheckIn],
|
||
values: dict[str, Any],
|
||
) -> tuple[bool, list[GateCheckFailure]]:
|
||
"""Сравнивает ожидания с фактическими значениями (строки после strip). Строки с пустым variable игнорируются."""
|
||
normalized: dict[str, str] = {}
|
||
for k, v in values.items():
|
||
key = str(k).strip()
|
||
if not key:
|
||
continue
|
||
if v is None:
|
||
normalized[key] = ""
|
||
else:
|
||
normalized[key] = str(v).strip()
|
||
|
||
failures: list[GateCheckFailure] = []
|
||
for c in checks:
|
||
var = (c.variable or "").strip()
|
||
if not var:
|
||
continue
|
||
exp = (c.expected or "").strip()
|
||
if var not in normalized:
|
||
failures.append(
|
||
GateCheckFailure(
|
||
variable=var,
|
||
reason="missing",
|
||
expected=exp,
|
||
actual=None,
|
||
)
|
||
)
|
||
continue
|
||
act = normalized[var]
|
||
if act != exp:
|
||
failures.append(
|
||
GateCheckFailure(
|
||
variable=var,
|
||
reason="mismatch",
|
||
expected=exp,
|
||
actual=act,
|
||
)
|
||
)
|
||
ok = len(failures) == 0
|
||
return ok, failures
|
||
|
||
|
||
class JenkinsTriggerRequest(BaseModel):
|
||
server_url: str = ""
|
||
job: str
|
||
parameters: dict[str, str] = Field(default_factory=dict)
|
||
|
||
|
||
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.get("/pipelines/autonomous")
|
||
def get_autonomous_pipeline(
|
||
db: Session = Depends(get_db),
|
||
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
||
):
|
||
if user is None:
|
||
return {"title": "", "stages": [], "updated_at": None}
|
||
row = db.get(AutonomousPipeline, user.id)
|
||
if row is None:
|
||
return {"title": "", "stages": [], "updated_at": None}
|
||
defn = migrate_definition(row.definition if isinstance(row.definition, dict) else {})
|
||
stages = defn.get("stages", [])
|
||
return {
|
||
"title": row.title,
|
||
"stages": stages,
|
||
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
|
||
}
|
||
|
||
|
||
@app.put("/pipelines/autonomous")
|
||
def put_autonomous_pipeline(
|
||
body: AutonomousPipelineSave,
|
||
db: Session = Depends(get_db),
|
||
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
||
):
|
||
if user is None:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
try:
|
||
row = db.get(AutonomousPipeline, user.id)
|
||
dumped = [s.model_dump() for s in body.stages]
|
||
new_def = {"schema_version": 2, "stages": dumped}
|
||
if row is None:
|
||
row = AutonomousPipeline(
|
||
user_id=user.id,
|
||
title=body.title,
|
||
definition=new_def,
|
||
)
|
||
db.add(row)
|
||
else:
|
||
row.title = body.title
|
||
row.definition = new_def
|
||
db.commit()
|
||
db.refresh(row)
|
||
return {
|
||
"title": row.title,
|
||
"stages": row.definition.get("stages", []),
|
||
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
|
||
}
|
||
except SQLAlchemyError as e:
|
||
db.rollback()
|
||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||
|
||
|
||
@app.post("/pipelines/autonomous/validate-gates", response_model=ValidateGateChecksOut)
|
||
def validate_autonomous_gates(
|
||
body: ValidateGateChecksIn,
|
||
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
||
):
|
||
"""Проверяет, что для каждого условия в `checks` в `values` есть совпадающее значение."""
|
||
if user is None:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
ok, failures = evaluate_gate_checks(body.checks, body.values)
|
||
return ValidateGateChecksOut(ok=ok, failures=failures)
|
||
|
||
|
||
@app.post("/jenkins/trigger")
|
||
def jenkins_trigger(
|
||
req: JenkinsTriggerRequest,
|
||
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
||
):
|
||
if user is None:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
from backend.jenkins_client import default_jenkins_credentials, trigger_jenkins_job
|
||
|
||
default_url, ju, jt, verify_ssl = default_jenkins_credentials()
|
||
base = (req.server_url or "").strip() or default_url
|
||
if not base:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail="Укажите Jenkins URL в фазе или задайте JENKINS_DEFAULT_URL в окружении backend.",
|
||
)
|
||
if not ju or not jt:
|
||
raise HTTPException(
|
||
status_code=503,
|
||
detail="Задайте JENKINS_USER и JENKINS_TOKEN (API token) в окружении backend.",
|
||
)
|
||
job = (req.job or "").strip()
|
||
if not job:
|
||
raise HTTPException(status_code=400, detail="Поле job обязательно.")
|
||
|
||
ok, _code, msg, detail = trigger_jenkins_job(
|
||
server_url=base,
|
||
job_path=job,
|
||
parameters=req.parameters,
|
||
username=ju,
|
||
token=jt,
|
||
verify_ssl=verify_ssl,
|
||
)
|
||
if not ok:
|
||
raise HTTPException(status_code=502, detail={"message": msg, **detail})
|
||
return {"ok": True, "message": msg, **detail}
|
||
|
||
|
||
@app.get("/chat/history")
|
||
def chat_history(
|
||
limit: int = 500,
|
||
db: Session = Depends(get_db),
|
||
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
||
):
|
||
if user is None:
|
||
return {"messages": []}
|
||
lim = max(1, min(limit, 2000))
|
||
rows = (
|
||
db.execute(
|
||
select(UserChatMessage)
|
||
.where(UserChatMessage.user_id == user.id)
|
||
.order_by(UserChatMessage.created_at.asc())
|
||
.limit(lim)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
return {
|
||
"messages": [{"role": r.role, "content": r.content} for r in rows],
|
||
}
|
||
|
||
|
||
@app.delete("/chat/history")
|
||
def chat_history_clear(
|
||
db: Session = Depends(get_db),
|
||
user: User | None = Depends(get_current_user) if REQUIRE_AUTH_FOR_GENERATE else None,
|
||
):
|
||
if user is None:
|
||
return {"ok": True}
|
||
db.execute(delete(UserChatMessage).where(UserChatMessage.user_id == user.id))
|
||
db.commit()
|
||
return {"ok": True}
|
||
|
||
|
||
@app.post("/chat")
|
||
def chat_with_ollama(
|
||
req: ChatRequest,
|
||
db: Session = Depends(get_db),
|
||
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"]
|
||
|
||
if user is not None:
|
||
last = req.messages[-1]
|
||
if last.role == "user":
|
||
try:
|
||
db.add(
|
||
UserChatMessage(
|
||
user_id=user.id,
|
||
role="user",
|
||
content=last.content,
|
||
)
|
||
)
|
||
db.add(
|
||
UserChatMessage(
|
||
user_id=user.id,
|
||
role="assistant",
|
||
content=content,
|
||
)
|
||
)
|
||
db.commit()
|
||
except SQLAlchemyError:
|
||
db.rollback()
|
||
|
||
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")))
|