Ampel-Waechter: Telegram bei ROTEN CI-Laeufen (no-agent-Cron, 30 min)
Meldet nur NEUE Fehlschlaege (Merkliste), running zaehlt nicht als gesehen. Teil 3 der Wasserdicht-Runde. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,67 @@
|
|||||||
|
#!/usr/bin/env bash
|
||||||
|
# Ampel-Wächter (22.07.2026, Wasserdicht-Runde): meldet ROTE CI-Läufe aller
|
||||||
|
# Gitea-Repos per Telegram. Läuft als no-agent-Cron (stdout leer = still,
|
||||||
|
# klassisches Watchdog-Muster). Nur NEUE Fehlschläge werden gemeldet —
|
||||||
|
# Status-Merkliste unter ~/.hermes/state/ampel-waechter-gesehen.txt.
|
||||||
|
set -uo pipefail
|
||||||
|
|
||||||
|
STATE="$HOME/.hermes/state/ampel-waechter-gesehen.txt"
|
||||||
|
mkdir -p "$(dirname "$STATE")"; touch "$STATE"
|
||||||
|
|
||||||
|
TOKEN=$(grep -m1 'git\.tobisniceshomelab' "$HOME/.git-credentials" 2>/dev/null | sed -E 's#https://[^:]+:([^@]+)@.*#\1#')
|
||||||
|
[ -n "$TOKEN" ] || exit 0
|
||||||
|
|
||||||
|
python3 - "$TOKEN" "$STATE" <<'PY'
|
||||||
|
import json, sys, urllib.request
|
||||||
|
|
||||||
|
token, state_pfad = sys.argv[1], sys.argv[2]
|
||||||
|
BASE = "https://git.tobisniceshomelab.ddnsfree.com/api/v1"
|
||||||
|
WEB = "https://git.tobisniceshomelab.ddnsfree.com"
|
||||||
|
|
||||||
|
|
||||||
|
def api(pfad):
|
||||||
|
req = urllib.request.Request(BASE + pfad, headers={"Authorization": "token " + token})
|
||||||
|
with urllib.request.urlopen(req, timeout=20) as r:
|
||||||
|
return json.load(r)
|
||||||
|
|
||||||
|
|
||||||
|
try:
|
||||||
|
gesehen = set(open(state_pfad).read().split())
|
||||||
|
except OSError:
|
||||||
|
gesehen = set()
|
||||||
|
|
||||||
|
neu = []
|
||||||
|
try:
|
||||||
|
repos = api("/user/repos?limit=50")
|
||||||
|
except Exception:
|
||||||
|
sys.exit(0)
|
||||||
|
|
||||||
|
for repo in repos:
|
||||||
|
full = repo.get("full_name", "")
|
||||||
|
if not full or repo.get("archived"):
|
||||||
|
continue
|
||||||
|
try:
|
||||||
|
d = api(f"/repos/{full}/actions/tasks?limit=1")
|
||||||
|
except Exception:
|
||||||
|
continue
|
||||||
|
rows = d.get("workflow_runs", d if isinstance(d, list) else d.get("entries", []))
|
||||||
|
for t in rows[:1]:
|
||||||
|
kennung = f"{full}#{t.get('id')}"
|
||||||
|
status = t.get("status")
|
||||||
|
if status == "failure" and kennung not in gesehen:
|
||||||
|
neu.append((full, t.get("head_sha", "")[:7], t.get("name", "")))
|
||||||
|
# Nur ABGESCHLOSSENE Läufe merken — ein 'running', das später rot wird,
|
||||||
|
# darf nicht schon als gesehen gelten.
|
||||||
|
if status in ("failure", "success", "cancelled", "skipped"):
|
||||||
|
gesehen.add(kennung)
|
||||||
|
|
||||||
|
if neu:
|
||||||
|
print("🔴 CI-AMPEL ROT:")
|
||||||
|
for full, sha, name in neu:
|
||||||
|
print(f"• {full} @ {sha} ({name})")
|
||||||
|
print(f" {WEB}/{full}/actions")
|
||||||
|
print("Nichts aus diesen Ständen ist ‚fertig', solange die Ampel rot ist.")
|
||||||
|
|
||||||
|
with open(state_pfad, "w") as f:
|
||||||
|
f.write("\n".join(sorted(gesehen)))
|
||||||
|
PY
|
||||||
Reference in New Issue
Block a user