ci-ai: KI-Schicht für die Gitea-CI (PR-Review, Security-Erklärung, CI-Failure-Analyse)
Provider-agnostische LLM-Anbindung (Anthropic jetzt, Jetson/Ollama später per Env-Var umschaltbar). Stdlib-only Skripte + Beispiel-Workflows. GitLab-Duo-Parität für self-hosted Gitea + act_runner. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
71
scripts/ai_explain.py
Normal file
71
scripts/ai_explain.py
Normal file
@@ -0,0 +1,71 @@
|
||||
"""Features B & C — Klartext-Erklärung von Scan-Funden / CI-Fehlern.
|
||||
|
||||
Aufruf:
|
||||
python ai_explain.py <kind> <textdatei>
|
||||
|
||||
kind = "security" -> erklärt Trivy/Gitleaks-Funde (GitLab-Duo-Parität)
|
||||
kind = "ci-failure" -> erklärt ein fehlgeschlagenes Job-Log
|
||||
|
||||
Postet das Ergebnis als idempotenten PR-Kommentar (eigener Marker je kind).
|
||||
Ohne PR-Kontext (z.B. push ohne PR) wird nur ins Job-Log geschrieben.
|
||||
"""
|
||||
import sys
|
||||
|
||||
import context
|
||||
import gitea_api
|
||||
import llm
|
||||
|
||||
KINDS = {
|
||||
"security": {
|
||||
"marker": "<!-- ci-ai:security -->",
|
||||
"heading": "## 🛡️ Security-Scan — Erklärung der Funde",
|
||||
"system": (
|
||||
"Du bist ein Security-Engineer. Erkläre die folgenden Scanner-Funde "
|
||||
"(Trivy/Gitleaks) auf Deutsch in Klartext: was wurde gefunden, warum "
|
||||
"ist es ein Risiko, und wie behebt man es konkret. Priorisiere nach "
|
||||
"Schweregrad. Keine Funde erfinden."
|
||||
),
|
||||
"intro": "Der Security-Scan ist fehlgeschlagen. Hier die Funde im Klartext:",
|
||||
},
|
||||
"ci-failure": {
|
||||
"marker": "<!-- ci-ai:ci-failure -->",
|
||||
"heading": "## 🔧 CI-Fehler — Analyse",
|
||||
"system": (
|
||||
"Du bist ein erfahrener CI/CD-Engineer. Analysiere das folgende Log "
|
||||
"eines fehlgeschlagenen CI-Jobs auf Deutsch: nenne die wahrscheinliche "
|
||||
"Ursache und einen konkreten Fix. Knapp und konkret."
|
||||
),
|
||||
"intro": "Ein CI-Job ist fehlgeschlagen. Wahrscheinliche Ursache:",
|
||||
},
|
||||
}
|
||||
|
||||
LOG_CAP = 16000 # Zeichen aus dem Log (Ende ist meist am aussagekräftigsten)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
if len(sys.argv) != 3 or sys.argv[1] not in KINDS:
|
||||
print("Usage: ai_explain.py <security|ci-failure> <textdatei>", file=sys.stderr)
|
||||
return 2
|
||||
|
||||
kind = KINDS[sys.argv[1]]
|
||||
with open(sys.argv[2], encoding="utf-8", errors="replace") as fh:
|
||||
text = fh.read()
|
||||
if len(text) > LOG_CAP:
|
||||
text = "[... Anfang gekürzt ...]\n" + text[-LOG_CAP:]
|
||||
|
||||
answer = llm.complete(kind["system"], f"{kind['intro']}\n\n```\n{text}\n```", max_tokens=1200)
|
||||
body = f"{kind['heading']}\n\n{answer}"
|
||||
|
||||
owner, repo = context.repo_slug()
|
||||
pr = context.pull_request()
|
||||
if pr and pr["index"]:
|
||||
cid = gitea_api.upsert_comment(owner, repo, pr["index"], kind["marker"], body)
|
||||
print(f"[ai_explain] Kommentar gepostet/aktualisiert (id={cid}).")
|
||||
else:
|
||||
print("[ai_explain] Kein PR-Kontext — Ausgabe nur im Log:\n")
|
||||
print(body)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
101
scripts/ai_review.py
Normal file
101
scripts/ai_review.py
Normal file
@@ -0,0 +1,101 @@
|
||||
"""Feature A — KI-PR-Review + Summary.
|
||||
|
||||
Holt den PR-Diff über die Gitea-API, lässt ihn vom konfigurierten LLM
|
||||
zusammenfassen + reviewen und postet das Ergebnis als EINEN idempotenten
|
||||
Kommentar (Marker -> editiert sich bei jedem Push, statt zu spammen).
|
||||
|
||||
Erwartete Env-Vars: siehe llm.py + gitea_api.py, plus die Actions-Kontext-
|
||||
Variablen (GITHUB_REPOSITORY, GITHUB_EVENT_PATH).
|
||||
"""
|
||||
import sys
|
||||
|
||||
import context
|
||||
import gitea_api
|
||||
import llm
|
||||
|
||||
MARKER = "<!-- ci-ai:review -->"
|
||||
DIFF_CAP = 40000 # Zeichen — darüber wird der Diff gekürzt (Token-/Kostenschutz)
|
||||
|
||||
# Dateien, die für ein Review wenig Sinn ergeben (Lockfiles, Binär, generiert).
|
||||
SKIP_TOKENS = (
|
||||
".lock", "package-lock.json", "yarn.lock", "pnpm-lock.yaml",
|
||||
".min.js", ".min.css", ".map",
|
||||
".png", ".jpg", ".jpeg", ".gif", ".ico", ".pdf", ".woff", ".woff2", ".ttf",
|
||||
"/vendor/", "/node_modules/", "/dist/", "/build/",
|
||||
)
|
||||
|
||||
SYSTEM = (
|
||||
"Du bist ein erfahrener, pragmatischer Code-Reviewer. Antworte auf Deutsch, "
|
||||
"knapp und konkret. Erfinde nichts, was nicht im Diff steht."
|
||||
)
|
||||
|
||||
PROMPT = """\
|
||||
Reviewe den folgenden Pull-Request.
|
||||
|
||||
PR-Titel: {title}
|
||||
PR-Beschreibung: {body}
|
||||
|
||||
Gib genau diese zwei Abschnitte aus (Markdown):
|
||||
|
||||
### Zusammenfassung
|
||||
2–4 Sätze: Was ändert dieser PR und warum.
|
||||
|
||||
### Review
|
||||
Stichpunkte zu: möglichen Bugs, Sicherheits-/Edge-Cases, fehlenden Tests,
|
||||
Stil/Lesbarkeit. Nur echte, im Diff belegbare Punkte — wenn alles gut aussieht,
|
||||
sag das ehrlich. Markiere ernste Probleme mit ⚠️.
|
||||
|
||||
--- DIFF ---
|
||||
{diff}
|
||||
"""
|
||||
|
||||
|
||||
def filter_diff(diff: str) -> str:
|
||||
"""Entfernt uninteressante Dateien (per `diff --git`-Sektion)."""
|
||||
if not diff:
|
||||
return ""
|
||||
sections = diff.split("diff --git ")
|
||||
kept = []
|
||||
for sec in sections[1:]: # sections[0] ist der Vorspann (leer)
|
||||
header = sec.split("\n", 1)[0].lower()
|
||||
if any(tok in header for tok in SKIP_TOKENS):
|
||||
continue
|
||||
kept.append("diff --git " + sec)
|
||||
return "".join(kept).strip()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
owner, repo = context.repo_slug()
|
||||
pr = context.pull_request()
|
||||
if not pr or not pr["index"]:
|
||||
print("[ai_review] Kein Pull-Request im Event — übersprungen.")
|
||||
return 0
|
||||
if pr["author"].lower().endswith("-bot") or pr["author"].lower() == "ci-ai":
|
||||
print("[ai_review] PR vom Bot selbst — übersprungen.")
|
||||
return 0
|
||||
|
||||
diff = filter_diff(gitea_api.get_pr_diff(owner, repo, pr["index"]))
|
||||
if not diff:
|
||||
print("[ai_review] Leerer/gefilterter Diff — nichts zu reviewen.")
|
||||
return 0
|
||||
|
||||
truncated = False
|
||||
if len(diff) > DIFF_CAP:
|
||||
diff = diff[:DIFF_CAP]
|
||||
truncated = True
|
||||
|
||||
answer = llm.complete(
|
||||
SYSTEM,
|
||||
PROMPT.format(title=pr["title"], body=pr["body"][:1000], diff=diff),
|
||||
max_tokens=1800,
|
||||
)
|
||||
|
||||
note = "\n\n_⚠️ Diff war groß und wurde gekürzt — Review nur über den Anfang._" if truncated else ""
|
||||
body = f"## 🤖 KI-Review\n\n{answer}{note}"
|
||||
cid = gitea_api.upsert_comment(owner, repo, pr["index"], MARKER, body)
|
||||
print(f"[ai_review] Kommentar gepostet/aktualisiert (id={cid}).")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
43
scripts/context.py
Normal file
43
scripts/context.py
Normal file
@@ -0,0 +1,43 @@
|
||||
"""Liest den Gitea-Actions-Event-Kontext (GitHub-kompatible Env-Vars).
|
||||
|
||||
Gitea spiegelt die GITHUB_*-Variablen: GITHUB_REPOSITORY (owner/repo),
|
||||
GITHUB_EVENT_PATH (JSON des auslösenden Events), GITHUB_EVENT_NAME.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
|
||||
|
||||
def repo_slug():
|
||||
"""(owner, repo) aus GITHUB_REPOSITORY."""
|
||||
full = os.environ.get("GITHUB_REPOSITORY", "")
|
||||
if "/" not in full:
|
||||
raise SystemExit("[context] GITHUB_REPOSITORY fehlt oder ist ungültig.")
|
||||
owner, repo = full.split("/", 1)
|
||||
return owner, repo
|
||||
|
||||
|
||||
def event():
|
||||
"""Geparstes Event-JSON (oder {} wenn nicht vorhanden)."""
|
||||
path = os.environ.get("GITHUB_EVENT_PATH", "")
|
||||
if path and os.path.exists(path):
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
return {}
|
||||
|
||||
|
||||
def pull_request():
|
||||
"""PR-Infos aus dem Event oder None, wenn das Event kein PR ist.
|
||||
|
||||
Liefert dict: index, title, body, author, base_ref, head_sha.
|
||||
"""
|
||||
pr = event().get("pull_request")
|
||||
if not pr:
|
||||
return None
|
||||
return {
|
||||
"index": pr.get("number"),
|
||||
"title": pr.get("title", ""),
|
||||
"body": pr.get("body", "") or "",
|
||||
"author": (pr.get("user") or {}).get("login", ""),
|
||||
"base_ref": (pr.get("base") or {}).get("ref", ""),
|
||||
"head_sha": (pr.get("head") or {}).get("sha", ""),
|
||||
}
|
||||
70
scripts/gitea_api.py
Normal file
70
scripts/gitea_api.py
Normal file
@@ -0,0 +1,70 @@
|
||||
"""Schlanke Gitea-API-Helfer (stdlib only).
|
||||
|
||||
- PR-Diff holen
|
||||
- Kommentare IDEMPOTENT posten: ein versteckter HTML-Marker identifiziert den
|
||||
Bot-Kommentar; existiert er schon, wird er editiert statt neu angelegt
|
||||
(kein Zuspammen bei jedem Push).
|
||||
|
||||
Env-Vars:
|
||||
GITEA_API Basis-URL der API, Default https://git.eulernest.eu/api/v1
|
||||
GITEA_TOKEN Token fürs Kommentieren (Actions-Auto-Token ODER PAT).
|
||||
Fallback: GITEA_BOT_TOKEN.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
_BASE = os.environ.get("GITEA_API", "https://git.eulernest.eu/api/v1").rstrip("/")
|
||||
_TOKEN = os.environ.get("GITEA_TOKEN") or os.environ.get("GITEA_BOT_TOKEN", "")
|
||||
|
||||
|
||||
def _request(method: str, path: str, *, data=None, raw: bool = False):
|
||||
url = f"{_BASE}{path}"
|
||||
headers = {}
|
||||
if _TOKEN:
|
||||
headers["Authorization"] = f"token {_TOKEN}"
|
||||
body = None
|
||||
if data is not None:
|
||||
body = json.dumps(data).encode("utf-8")
|
||||
headers["Content-Type"] = "application/json"
|
||||
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||||
content = resp.read().decode("utf-8", "replace")
|
||||
if raw:
|
||||
return content
|
||||
return json.loads(content) if content.strip() else None
|
||||
except urllib.error.HTTPError as exc:
|
||||
detail = exc.read().decode("utf-8", "replace")
|
||||
raise SystemExit(f"[gitea] {method} {path} -> HTTP {exc.code}: {detail[:400]}")
|
||||
except urllib.error.URLError as exc:
|
||||
raise SystemExit(f"[gitea] {method} {path} fehlgeschlagen: {exc.reason}")
|
||||
|
||||
|
||||
def get_pr_diff(owner: str, repo: str, index: int) -> str:
|
||||
"""Unified Diff eines PRs (Gitea unterstützt den .diff-Suffix)."""
|
||||
return _request("GET", f"/repos/{owner}/{repo}/pulls/{index}.diff", raw=True)
|
||||
|
||||
|
||||
def upsert_comment(owner: str, repo: str, index: int, marker: str, body: str) -> int:
|
||||
"""Legt einen Kommentar an ODER editiert den vorhandenen mit gleichem Marker.
|
||||
|
||||
PR-Kommentare sind in Gitea Issue-Kommentare (gleicher Endpunkt).
|
||||
"""
|
||||
full = f"{body}\n\n{marker}"
|
||||
existing = _request("GET", f"/repos/{owner}/{repo}/issues/{index}/comments") or []
|
||||
for comment in existing:
|
||||
if marker in (comment.get("body") or ""):
|
||||
_request(
|
||||
"PATCH",
|
||||
f"/repos/{owner}/{repo}/issues/comments/{comment['id']}",
|
||||
data={"body": full},
|
||||
)
|
||||
return comment["id"]
|
||||
created = _request(
|
||||
"POST",
|
||||
f"/repos/{owner}/{repo}/issues/{index}/comments",
|
||||
data={"body": full},
|
||||
)
|
||||
return created["id"]
|
||||
92
scripts/llm.py
Normal file
92
scripts/llm.py
Normal file
@@ -0,0 +1,92 @@
|
||||
"""Provider-agnostischer LLM-Client für die KI-CI.
|
||||
|
||||
Bewusst NUR stdlib (urllib) — so braucht der CI-Job kein `pip install` und
|
||||
läuft in jedem schlanken Python-Image sofort.
|
||||
|
||||
Umschalten zwischen Backends über Env-Vars:
|
||||
AI_PROVIDER=anthropic -> api.anthropic.com/v1/messages (Claude, jetzt)
|
||||
AI_PROVIDER=ollama -> <AI_BASE_URL>/chat/completions (Jetson, später)
|
||||
|
||||
Weitere Env-Vars:
|
||||
AI_MODEL Modellname (Default je nach Provider)
|
||||
ANTHROPIC_API_KEY (für anthropic)
|
||||
AI_BASE_URL OpenAI-kompatible Basis-URL (für ollama),
|
||||
Default http://192.168.178.20:11434/v1
|
||||
JETSON_API_KEY Bearer-Token (für ollama, optional)
|
||||
|
||||
Der Wechsel auf den Jetson ist damit ein reiner Env-Change in der Workflow-
|
||||
Datei — KEINE Code-Änderung.
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
|
||||
|
||||
def complete(system: str, user: str, *, max_tokens: int = 1500) -> str:
|
||||
"""Schickt (system, user) ans konfigurierte Backend, gibt reinen Text zurück."""
|
||||
provider = os.environ.get("AI_PROVIDER", "anthropic").lower()
|
||||
if provider == "anthropic":
|
||||
model = os.environ.get("AI_MODEL", "claude-3-5-haiku-latest")
|
||||
return _anthropic(system, user, model, max_tokens)
|
||||
if provider == "ollama":
|
||||
model = os.environ.get("AI_MODEL", "qwen-light")
|
||||
return _ollama_openai(system, user, model, max_tokens)
|
||||
raise SystemExit(f"[llm] Unbekannter AI_PROVIDER: {provider!r}")
|
||||
|
||||
|
||||
def _post(url: str, headers: dict, payload: dict) -> dict:
|
||||
data = json.dumps(payload).encode("utf-8")
|
||||
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=180) as resp:
|
||||
return json.loads(resp.read().decode("utf-8"))
|
||||
except urllib.error.HTTPError as exc:
|
||||
body = exc.read().decode("utf-8", "replace")
|
||||
raise SystemExit(f"[llm] HTTP {exc.code} von {url}: {body[:600]}")
|
||||
except urllib.error.URLError as exc:
|
||||
raise SystemExit(f"[llm] Verbindung zu {url} fehlgeschlagen: {exc.reason}")
|
||||
|
||||
|
||||
def _anthropic(system: str, user: str, model: str, max_tokens: int) -> str:
|
||||
try:
|
||||
key = os.environ["ANTHROPIC_API_KEY"]
|
||||
except KeyError:
|
||||
raise SystemExit("[llm] ANTHROPIC_API_KEY ist nicht gesetzt.")
|
||||
resp = _post(
|
||||
"https://api.anthropic.com/v1/messages",
|
||||
{
|
||||
"x-api-key": key,
|
||||
"anthropic-version": "2023-06-01",
|
||||
"content-type": "application/json",
|
||||
},
|
||||
{
|
||||
"model": model,
|
||||
"max_tokens": max_tokens,
|
||||
"system": system,
|
||||
"messages": [{"role": "user", "content": user}],
|
||||
},
|
||||
)
|
||||
parts = [b.get("text", "") for b in resp.get("content", []) if b.get("type") == "text"]
|
||||
return "".join(parts).strip()
|
||||
|
||||
|
||||
def _ollama_openai(system: str, user: str, model: str, max_tokens: int) -> str:
|
||||
base = os.environ.get("AI_BASE_URL", "http://192.168.178.20:11434/v1").rstrip("/")
|
||||
headers = {"content-type": "application/json"}
|
||||
key = os.environ.get("JETSON_API_KEY", "")
|
||||
if key:
|
||||
headers["Authorization"] = f"Bearer {key}"
|
||||
resp = _post(
|
||||
f"{base}/chat/completions",
|
||||
headers,
|
||||
{
|
||||
"model": model,
|
||||
"max_tokens": max_tokens,
|
||||
"messages": [
|
||||
{"role": "system", "content": system},
|
||||
{"role": "user", "content": user},
|
||||
],
|
||||
},
|
||||
)
|
||||
return resp["choices"][0]["message"]["content"].strip()
|
||||
Reference in New Issue
Block a user