scripts/update_status.py aktualisiert
This commit is contained in:
+635
-128
@@ -1,5 +1,11 @@
|
||||
|
||||
#!/usr/bin/env python3
|
||||
"""Gitea Actions status dashboard. Standard library only; reads config from same repo."""
|
||||
"""
|
||||
Gitea Actions Status Dashboard
|
||||
Kompatibel mit Gitea 1.27.x
|
||||
Nur Python-Standardbibliothek erforderlich.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import datetime as dt
|
||||
import json
|
||||
@@ -8,180 +14,681 @@ import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
BASE = os.environ['GITEA_URL'].rstrip('/')
|
||||
API = BASE + '/api/v1'
|
||||
TOKEN = os.environ['GITEA_API_TOKEN']
|
||||
OWNER = os.environ.get('STATUS_OWNER', 'VULN')
|
||||
STATUS_REPO = os.environ.get('STATUS_REPOSITORY', 'status')
|
||||
BRANCH = os.environ.get('STATUS_BRANCH', 'main')
|
||||
|
||||
# ==========================================
|
||||
# Konfiguration
|
||||
# ==========================================
|
||||
|
||||
BASE = os.environ["GITEA_URL"].rstrip("/")
|
||||
API = BASE + "/api/v1"
|
||||
|
||||
TOKEN = os.environ["GITEA_API_TOKEN"]
|
||||
|
||||
OWNER = os.environ.get("STATUS_OWNER", "VULN")
|
||||
STATUS_REPO = os.environ.get("STATUS_REPOSITORY", "status")
|
||||
BRANCH = os.environ.get("STATUS_BRANCH", "main")
|
||||
|
||||
NOW = dt.datetime.now(dt.timezone.utc)
|
||||
TIMEZONE = ZoneInfo("Europe/Berlin")
|
||||
|
||||
PAGE_SIZE = 50
|
||||
MAX_PAGES = 5
|
||||
|
||||
# Markdown-Emoji-Codes
|
||||
ICONS = {
|
||||
"success": ":green_circle:",
|
||||
"failure": ":red_circle:",
|
||||
"running": ":large_blue_circle:",
|
||||
"waiting": ":yellow_circle:",
|
||||
"unknown": ":white_circle:",
|
||||
"warning": ":warning:",
|
||||
"error": ":x:",
|
||||
}
|
||||
|
||||
|
||||
def req(path, method='GET', data=None):
|
||||
body = json.dumps(data).encode() if data is not None else None
|
||||
# ==========================================
|
||||
# Hilfsfunktionen
|
||||
# ==========================================
|
||||
|
||||
def log(message):
|
||||
print(message, flush=True)
|
||||
|
||||
|
||||
def quote(value):
|
||||
return urllib.parse.quote(str(value), safe="")
|
||||
|
||||
|
||||
def req(path, method="GET", data=None):
|
||||
body = (
|
||||
json.dumps(data).encode("utf-8")
|
||||
if data is not None
|
||||
else None
|
||||
)
|
||||
|
||||
request = urllib.request.Request(
|
||||
API + path, data=body, method=method,
|
||||
headers={'Authorization': 'token ' + TOKEN, 'Accept': 'application/json',
|
||||
'Content-Type': 'application/json', 'User-Agent': 'vuln-status/1.0'})
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=35) as r:
|
||||
return json.load(r)
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code == 404:
|
||||
raise FileNotFoundError(f'HTTP 404: {path}') from e
|
||||
err = e.read(500).decode(errors='replace')
|
||||
raise RuntimeError(f'HTTP {e.code} {path}: {err}') from e
|
||||
API + path,
|
||||
data=body,
|
||||
method=method,
|
||||
headers={
|
||||
"Authorization": "token " + TOKEN,
|
||||
"Accept": "application/json",
|
||||
"Content-Type": "application/json",
|
||||
"User-Agent": "vuln-status/2.0",
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
with urllib.request.urlopen(
|
||||
request,
|
||||
timeout=35
|
||||
) as response:
|
||||
|
||||
payload = response.read()
|
||||
|
||||
if not payload:
|
||||
return {}
|
||||
|
||||
return json.loads(payload)
|
||||
|
||||
except urllib.error.HTTPError as exc:
|
||||
|
||||
message = exc.read(1000).decode(
|
||||
errors="replace"
|
||||
)
|
||||
|
||||
raise RuntimeError(
|
||||
f"HTTP {exc.code}: {path}: {message}"
|
||||
) from exc
|
||||
|
||||
except urllib.error.URLError as exc:
|
||||
|
||||
raise RuntimeError(
|
||||
f"Verbindungsfehler: {path}: {exc}"
|
||||
) from exc
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Repository-Dateien
|
||||
# ==========================================
|
||||
|
||||
def content_path(filename):
|
||||
return '/repos/{}/{}/contents/{}?ref={}'.format(
|
||||
urllib.parse.quote(OWNER, safe=''),
|
||||
urllib.parse.quote(STATUS_REPO, safe=''),
|
||||
urllib.parse.quote(filename, safe='/'),
|
||||
urllib.parse.quote(BRANCH, safe=''))
|
||||
|
||||
return (
|
||||
f"/repos/{quote(OWNER)}/"
|
||||
f"{quote(STATUS_REPO)}/contents/"
|
||||
f"{urllib.parse.quote(filename, safe='/')}"
|
||||
f"?ref={quote(BRANCH)}"
|
||||
)
|
||||
|
||||
|
||||
def get_content(filename):
|
||||
obj = req(content_path(filename))
|
||||
return base64.b64decode(obj['content']).decode('utf-8'), obj['sha']
|
||||
|
||||
obj = req(content_path(filename))
|
||||
|
||||
content = base64.b64decode(
|
||||
obj["content"]
|
||||
).decode("utf-8")
|
||||
|
||||
return content, obj["sha"]
|
||||
|
||||
|
||||
def update_readme(content, sha):
|
||||
|
||||
path = (
|
||||
f"/repos/{quote(OWNER)}/"
|
||||
f"{quote(STATUS_REPO)}/contents/README.md"
|
||||
)
|
||||
|
||||
data = {
|
||||
"branch": BRANCH,
|
||||
"sha": sha,
|
||||
"content": base64.b64encode(
|
||||
content.encode("utf-8")
|
||||
).decode("ascii"),
|
||||
"message": (
|
||||
"docs: update vulnerability "
|
||||
"sync status [skip ci]"
|
||||
),
|
||||
}
|
||||
|
||||
return req(path, "PUT", data)
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Datum und Laufzeit
|
||||
# ==========================================
|
||||
|
||||
def parse_time(raw):
|
||||
if not raw or raw.startswith('0001-'):
|
||||
|
||||
if not raw or str(raw).startswith("0001-"):
|
||||
return None
|
||||
|
||||
try:
|
||||
return dt.datetime.fromisoformat(raw.replace('Z', '+00:00')).astimezone(dt.timezone.utc)
|
||||
except (ValueError, AttributeError):
|
||||
value = dt.datetime.fromisoformat(
|
||||
str(raw).replace("Z", "+00:00")
|
||||
)
|
||||
|
||||
if value.tzinfo is None:
|
||||
value = value.replace(
|
||||
tzinfo=dt.timezone.utc
|
||||
)
|
||||
|
||||
return value.astimezone(dt.timezone.utc)
|
||||
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def fmt(value):
|
||||
|
||||
if value is None:
|
||||
return '–'
|
||||
local = value.astimezone(ZoneInfo('Europe/Berlin'))
|
||||
return local.strftime('%d.%m.%Y %H:%M %Z')
|
||||
return "–"
|
||||
|
||||
return value.astimezone(TIMEZONE).strftime(
|
||||
"%d.%m.%Y %H:%M %Z"
|
||||
)
|
||||
|
||||
|
||||
def elapsed(start, end):
|
||||
|
||||
if not start or not end or end < start:
|
||||
return '–'
|
||||
sec = int((end - start).total_seconds())
|
||||
return f'{sec // 3600}h {sec % 3600 // 60}m' if sec >= 3600 else f'{sec // 60}m {sec % 60:02d}s'
|
||||
return "–"
|
||||
|
||||
seconds = int(
|
||||
(end - start).total_seconds()
|
||||
)
|
||||
|
||||
hours, remaining = divmod(seconds, 3600)
|
||||
minutes, seconds = divmod(remaining, 60)
|
||||
|
||||
if hours:
|
||||
return f"{hours}h {minutes:02d}m"
|
||||
|
||||
return f"{minutes}m {seconds:02d}s"
|
||||
|
||||
|
||||
def safe_cell(text):
|
||||
return str(text).replace('|', '\\|').replace('\n', ' ')
|
||||
def safe_cell(value):
|
||||
|
||||
return (
|
||||
str(value)
|
||||
.replace("|", "\\|")
|
||||
.replace("\n", " ")
|
||||
.replace("\r", " ")
|
||||
)
|
||||
|
||||
|
||||
def api_runs(owner, repo, workflow, page, limit=50):
|
||||
# Gitea 1.27: Repository-level endpoint; avoids 404 caused by
|
||||
# guessed/nonexistent workflow filenames.
|
||||
path = '/repos/{}/{}/actions/runs?{}' .format(
|
||||
urllib.parse.quote(owner, safe=''),
|
||||
urllib.parse.quote(repo, safe=''),
|
||||
urllib.parse.urlencode({'page': page, 'limit': limit}))
|
||||
# ==========================================
|
||||
# Gitea Actions API
|
||||
# ==========================================
|
||||
|
||||
def api_runs(owner, repo, page=1):
|
||||
|
||||
params = urllib.parse.urlencode({
|
||||
"page": page,
|
||||
"limit": PAGE_SIZE,
|
||||
})
|
||||
|
||||
path = (
|
||||
f"/repos/{quote(owner)}/"
|
||||
f"{quote(repo)}/actions/runs?{params}"
|
||||
)
|
||||
|
||||
result = req(path)
|
||||
runs = result.get('workflow_runs', [])
|
||||
if workflow:
|
||||
# Gitea run.path is typically .gitea/workflows/<filename>@<branch>.
|
||||
# Restrict to configured workflow, never show a different workflow as success.
|
||||
def matches(run):
|
||||
name = str(run.get('path') or '').split('@', 1)[0].rsplit('/', 1)[-1]
|
||||
return name == workflow
|
||||
runs = [run for run in runs if matches(run)]
|
||||
|
||||
return result.get("workflow_runs", [])
|
||||
|
||||
|
||||
def workflow_matches(run, workflow):
|
||||
|
||||
if not workflow:
|
||||
return True
|
||||
|
||||
path = str(run.get("path") or "")
|
||||
|
||||
# Beispiel:
|
||||
# .gitea/workflows/sync.yml@refs/heads/main
|
||||
|
||||
filename = (
|
||||
path.split("@", 1)[0]
|
||||
.rsplit("/", 1)[-1]
|
||||
)
|
||||
|
||||
return filename == workflow
|
||||
|
||||
|
||||
def get_runs(owner, repo, workflow=None):
|
||||
|
||||
runs = []
|
||||
|
||||
for page in range(1, MAX_PAGES + 1):
|
||||
|
||||
# Erst API-Seite abrufen
|
||||
chunk = api_runs(owner, repo, page)
|
||||
|
||||
if not chunk:
|
||||
break
|
||||
|
||||
# Danach nach Workflow filtern
|
||||
matched = [
|
||||
run for run in chunk
|
||||
if workflow_matches(run, workflow)
|
||||
]
|
||||
|
||||
runs.extend(matched)
|
||||
|
||||
# Pagination anhand der ungefilterten
|
||||
# API-Ergebnisse prüfen.
|
||||
if len(chunk) < PAGE_SIZE:
|
||||
break
|
||||
|
||||
runs.sort(
|
||||
key=lambda run: (
|
||||
parse_time(run.get("created_at"))
|
||||
or dt.datetime.min.replace(
|
||||
tzinfo=dt.timezone.utc
|
||||
),
|
||||
run.get("id") or 0,
|
||||
),
|
||||
reverse=True,
|
||||
)
|
||||
|
||||
return runs
|
||||
|
||||
|
||||
def source_status(cfg, source):
|
||||
owner = source.get('owner', cfg['owner'])
|
||||
repo = source['repo']
|
||||
workflow = source.get('workflow')
|
||||
limit_age = dt.timedelta(hours=source.get('max_age_hours', cfg.get('max_age_hours', 26)))
|
||||
all_runs = []
|
||||
# Up to 250 runs; sufficient for typical daily sources, can be adjusted.
|
||||
for page in range(1, 6):
|
||||
chunk = api_runs(owner, repo, workflow, page)
|
||||
all_runs.extend(chunk)
|
||||
if len(chunk) < 50:
|
||||
break
|
||||
all_runs.sort(key=lambda r: (parse_time(r.get('created_at')) or dt.datetime.min.replace(tzinfo=dt.timezone.utc), r.get('id') or 0), reverse=True)
|
||||
latest = all_runs[0] if all_runs else None
|
||||
success = next((r for r in all_runs if (r.get('conclusion') or '').lower() == 'success'), None)
|
||||
latest_success = parse_time(success.get('completed_at')) or parse_time(success.get('created_at')) if success else None
|
||||
if not latest:
|
||||
state = '⚪ Kein Lauf'
|
||||
elif (latest.get('status') or '').lower() in ('in_progress', 'running'):
|
||||
state = '🔵 Läuft'
|
||||
elif (latest.get('status') or '').lower() in ('queued', 'pending', 'requested', 'waiting'):
|
||||
state = '🟡 Wartet'
|
||||
elif (latest.get('conclusion') or '').lower() == 'success':
|
||||
state = '🟢 Erfolgreich'
|
||||
elif (latest.get('conclusion') or '').lower() in ('failure', 'timed_out', 'startup_failure'):
|
||||
state = '🔴 Fehlgeschlagen'
|
||||
elif (latest.get('conclusion') or '').lower() in ('cancelled', 'skipped'):
|
||||
state = '⚪ ' + latest['conclusion']
|
||||
else:
|
||||
state = '🟡 ' + str(latest.get('status') or 'Unbekannt')
|
||||
if latest_success is None or NOW - latest_success > limit_age:
|
||||
state += ' / ⚠️ Überfällig'
|
||||
started = parse_time(latest.get('started_at')) if latest else None
|
||||
ended = parse_time(latest.get('completed_at')) if latest else None
|
||||
latest_url = latest.get('html_url') if latest else None
|
||||
link = f'[#{latest["id"]}]({latest_url})' if latest_url and latest else '–'
|
||||
return state, fmt(latest_success), elapsed(started, ended), link
|
||||
# ==========================================
|
||||
# Jobstatus
|
||||
# ==========================================
|
||||
|
||||
def get_state(run):
|
||||
|
||||
if not run:
|
||||
return (
|
||||
ICONS["unknown"] + " Kein Lauf"
|
||||
)
|
||||
|
||||
status = str(
|
||||
run.get("status") or ""
|
||||
).lower()
|
||||
|
||||
conclusion = str(
|
||||
run.get("conclusion") or ""
|
||||
).lower()
|
||||
|
||||
if status in (
|
||||
"in_progress",
|
||||
"running",
|
||||
):
|
||||
return ICONS["running"] + " Läuft"
|
||||
|
||||
if status in (
|
||||
"queued",
|
||||
"pending",
|
||||
"requested",
|
||||
"waiting",
|
||||
):
|
||||
return ICONS["waiting"] + " Wartet"
|
||||
|
||||
if conclusion == "success":
|
||||
return ICONS["success"] + " Erfolgreich"
|
||||
|
||||
if conclusion in (
|
||||
"failure",
|
||||
"timed_out",
|
||||
"startup_failure",
|
||||
):
|
||||
return ICONS["failure"] + " Fehlgeschlagen"
|
||||
|
||||
if conclusion in (
|
||||
"cancelled",
|
||||
"skipped",
|
||||
):
|
||||
return (
|
||||
ICONS["unknown"]
|
||||
+ " "
|
||||
+ conclusion.capitalize()
|
||||
)
|
||||
|
||||
return (
|
||||
ICONS["unknown"]
|
||||
+ " "
|
||||
+ (status or "Unbekannt")
|
||||
)
|
||||
|
||||
|
||||
def source_status(cfg, source):
|
||||
|
||||
owner = source.get(
|
||||
"owner",
|
||||
cfg.get("owner", OWNER)
|
||||
)
|
||||
|
||||
repo = source["repo"]
|
||||
|
||||
workflow = source.get("workflow")
|
||||
|
||||
max_age_hours = source.get(
|
||||
"max_age_hours",
|
||||
cfg.get("max_age_hours", 26)
|
||||
)
|
||||
|
||||
limit_age = dt.timedelta(
|
||||
hours=max_age_hours
|
||||
)
|
||||
|
||||
runs = get_runs(
|
||||
owner,
|
||||
repo,
|
||||
workflow
|
||||
)
|
||||
|
||||
latest = runs[0] if runs else None
|
||||
|
||||
# Letzten erfolgreichen Lauf bestimmen.
|
||||
success = next(
|
||||
(
|
||||
run for run in runs
|
||||
if str(
|
||||
run.get("conclusion") or ""
|
||||
).lower() == "success"
|
||||
),
|
||||
None
|
||||
)
|
||||
|
||||
latest_success = None
|
||||
|
||||
if success:
|
||||
latest_success = (
|
||||
parse_time(success.get("completed_at"))
|
||||
or parse_time(success.get("updated_at"))
|
||||
or parse_time(success.get("created_at"))
|
||||
)
|
||||
|
||||
# Status ermitteln
|
||||
state = get_state(latest)
|
||||
|
||||
# Überfälligkeit überprüfen
|
||||
overdue = (
|
||||
latest_success is None
|
||||
or NOW - latest_success > limit_age
|
||||
)
|
||||
|
||||
if overdue:
|
||||
state += (
|
||||
" / "
|
||||
+ ICONS["warning"]
|
||||
+ " Überfällig"
|
||||
)
|
||||
|
||||
# Laufzeit ermitteln
|
||||
started = None
|
||||
ended = None
|
||||
|
||||
if latest:
|
||||
|
||||
started = (
|
||||
parse_time(latest.get("started_at"))
|
||||
or parse_time(latest.get("run_started_at"))
|
||||
)
|
||||
|
||||
status = str(
|
||||
latest.get("status") or ""
|
||||
).lower()
|
||||
|
||||
if status == "completed":
|
||||
|
||||
ended = (
|
||||
parse_time(latest.get("completed_at"))
|
||||
or parse_time(latest.get("updated_at"))
|
||||
)
|
||||
|
||||
if not started:
|
||||
started = parse_time(
|
||||
latest.get("created_at")
|
||||
)
|
||||
|
||||
duration = elapsed(started, ended)
|
||||
|
||||
# Workflow-Link
|
||||
link = "–"
|
||||
|
||||
if latest:
|
||||
|
||||
run_id = latest.get("id")
|
||||
|
||||
url = latest.get("html_url")
|
||||
|
||||
if not url and run_id:
|
||||
url = (
|
||||
f"{BASE}/{quote(owner)}/"
|
||||
f"{quote(repo)}/actions/runs/"
|
||||
f"{run_id}"
|
||||
)
|
||||
|
||||
if url and run_id:
|
||||
link = f"[#{run_id}]({url})"
|
||||
|
||||
return {
|
||||
"state": state,
|
||||
"last_success": fmt(latest_success),
|
||||
"duration": duration,
|
||||
"link": link,
|
||||
"overdue": overdue,
|
||||
}
|
||||
|
||||
|
||||
# ==========================================
|
||||
# README erzeugen
|
||||
# ==========================================
|
||||
|
||||
def build_readme(cfg, rows, stats):
|
||||
|
||||
timestamp = fmt(NOW)
|
||||
|
||||
content = [
|
||||
"# Vulnerability Database – Sync Status",
|
||||
"",
|
||||
"Automatisch generiertes Dashboard "
|
||||
"für Gitea Actions.",
|
||||
"",
|
||||
f"**Letzte Prüfung:** {timestamp}",
|
||||
"",
|
||||
"## Übersicht",
|
||||
"",
|
||||
f"- **Datenquellen:** {stats['total']}",
|
||||
f"- **Erfolgreich:** {stats['success']}",
|
||||
f"- **Laufend:** {stats['running']}",
|
||||
f"- **Wartend:** {stats['waiting']}",
|
||||
f"- **Fehlgeschlagen:** {stats['failed']}",
|
||||
f"- **Überfällig:** {stats['overdue']}",
|
||||
f"- **API-Fehler:** {stats['errors']}",
|
||||
"",
|
||||
"## Synchronisierungsstatus",
|
||||
"",
|
||||
"| Datenquelle | Aktueller Status | "
|
||||
"Letzter Erfolg | Dauer letzter Lauf | "
|
||||
"Letzter Job |",
|
||||
"|---|---|---|---|---|",
|
||||
]
|
||||
|
||||
content.extend(rows)
|
||||
|
||||
content.extend([
|
||||
"",
|
||||
"## Statusbeschreibung",
|
||||
"",
|
||||
f"- {ICONS['success']} **Erfolgreich:** "
|
||||
"Letzter Job erfolgreich abgeschlossen.",
|
||||
f"- {ICONS['running']} **Läuft:** "
|
||||
"Job wird aktuell ausgeführt.",
|
||||
f"- {ICONS['waiting']} **Wartet:** "
|
||||
"Job wartet auf einen Runner.",
|
||||
f"- {ICONS['failure']} **Fehlgeschlagen:** "
|
||||
"Letzter Job wurde mit Fehler beendet.",
|
||||
f"- {ICONS['warning']} **Überfällig:** "
|
||||
"Kein ausreichend aktueller erfolgreicher Lauf.",
|
||||
f"- {ICONS['error']} **API-Fehler:** "
|
||||
"Repository, Tokenrechte oder API prüfen.",
|
||||
"",
|
||||
"## Konfiguration",
|
||||
"",
|
||||
f"- **Standard-Maximalalter:** "
|
||||
f"{cfg.get('max_age_hours', 26)} Stunden",
|
||||
f"- **Zeitzone:** Europe/Berlin",
|
||||
f"- **Workflow-Abfrage:** Gitea REST API",
|
||||
f"- **Aktualisierung:** Alle 15 Minuten "
|
||||
"oder manuell",
|
||||
"",
|
||||
"## Links",
|
||||
"",
|
||||
"- [Vulnerability Database]"
|
||||
"(https://vulnerability.apex-project.de)",
|
||||
"- [Projektwebsite]"
|
||||
"(https://www.apex-project.de)",
|
||||
"",
|
||||
])
|
||||
|
||||
return "\n".join(content)
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Hauptprogramm
|
||||
# ==========================================
|
||||
|
||||
def main():
|
||||
config_text, _ = get_content('sources.json')
|
||||
|
||||
log("======================================")
|
||||
log("Vulnerability Status Dashboard")
|
||||
log("======================================")
|
||||
|
||||
config_text, _ = get_content(
|
||||
"sources.json"
|
||||
)
|
||||
|
||||
cfg = json.loads(config_text)
|
||||
|
||||
rows = []
|
||||
errors = []
|
||||
for src in cfg['sources']:
|
||||
owner = src.get('owner', cfg['owner'])
|
||||
repo = src['repo']
|
||||
workflow = src.get('workflow')
|
||||
name = src['name']
|
||||
url = BASE + '/' + urllib.parse.quote(owner) + '/' + urllib.parse.quote(repo) + '/actions'
|
||||
|
||||
stats = {
|
||||
"total": len(cfg["sources"]),
|
||||
"success": 0,
|
||||
"running": 0,
|
||||
"waiting": 0,
|
||||
"failed": 0,
|
||||
"overdue": 0,
|
||||
"errors": 0,
|
||||
}
|
||||
|
||||
for src in cfg["sources"]:
|
||||
|
||||
owner = src.get(
|
||||
"owner",
|
||||
cfg.get("owner", OWNER)
|
||||
)
|
||||
|
||||
repo = src["repo"]
|
||||
name = src.get("name", repo)
|
||||
|
||||
url = (
|
||||
f"{BASE}/{quote(owner)}/"
|
||||
f"{quote(repo)}/actions"
|
||||
)
|
||||
|
||||
try:
|
||||
state, last_success, duration, link = source_status(cfg, src)
|
||||
|
||||
result = source_status(cfg, src)
|
||||
|
||||
state = result["state"]
|
||||
last_success = result["last_success"]
|
||||
duration = result["duration"]
|
||||
link = result["link"]
|
||||
|
||||
if result["overdue"]:
|
||||
stats["overdue"] += 1
|
||||
|
||||
if state.startswith(ICONS["success"]):
|
||||
stats["success"] += 1
|
||||
|
||||
elif state.startswith(ICONS["running"]):
|
||||
stats["running"] += 1
|
||||
|
||||
elif state.startswith(ICONS["waiting"]):
|
||||
stats["waiting"] += 1
|
||||
|
||||
elif state.startswith(ICONS["failure"]):
|
||||
stats["failed"] += 1
|
||||
|
||||
except Exception as exc:
|
||||
state, last_success, duration, link = '❌ API-Fehler', '–', '–', '–'
|
||||
errors.append(f'{repo}: {exc}')
|
||||
rows.append(f'| [{safe_cell(name)}]({url}) | {state} | {last_success} | {duration} | {link} |')
|
||||
print(f'{repo}: {state}', flush=True)
|
||||
content = ('# Vulnerability Database – Sync Status\n\n'
|
||||
'Automatisch generiertes Dashboard für Gitea Actions.\n\n'
|
||||
f'**Letzte Prüfung:** {NOW.strftime("%d.%m.%Y %H:%M UTC")} \n'
|
||||
f'**Überfällig:** Kein erfolgreicher Lauf innerhalb von {cfg.get("max_age_hours", 26)} Stunden '
|
||||
'(pro Quelle anpassbar).\n\n'
|
||||
'| Datenquelle | Aktueller Status | Letzter Erfolg | Dauer letzter Lauf | Letzter Job |\n'
|
||||
'|---|---|---|---|---|\n' + '\n'.join(rows) + '\n\n'
|
||||
'## Hinweise\n\n'
|
||||
'- 🔴 Fehlgeschlagen: Der neueste Workflow ist gescheitert.\n'
|
||||
'- ⚠️ Überfällig: Es gibt keinen ausreichend aktuellen erfolgreichen Lauf, auch wenn der letzte alte Lauf grün war.\n'
|
||||
'- 🔵 Läuft / 🟡 Wartet: Der neueste Lauf ist noch nicht abgeschlossen.\n'
|
||||
'- ❌ API-Fehler: Repository, Tokenrechte oder API prüfen.\n'
|
||||
'- Die Dauer bezieht sich auf den zuletzt gestarteten Lauf, soweit Start- und Endzeit vorliegen.\n'
|
||||
'- Status und Zeitangaben werden alle 15 Minuten oder manuell aktualisiert.\n\n'
|
||||
'## Links\n\n'
|
||||
'- [Vulnerability Database](https://vulnerability.apex-project.de)\n'
|
||||
'- [Projektwebsite](https://www.apex-project.de)\n')
|
||||
existing, sha = get_content('README.md')
|
||||
|
||||
state = (
|
||||
ICONS["error"] + " API-Fehler"
|
||||
)
|
||||
|
||||
last_success = "–"
|
||||
duration = "–"
|
||||
link = "–"
|
||||
|
||||
stats["errors"] += 1
|
||||
|
||||
errors.append(
|
||||
f"{repo}: {exc}"
|
||||
)
|
||||
|
||||
rows.append(
|
||||
f"| [{safe_cell(name)}]({url}) "
|
||||
f"| {state} "
|
||||
f"| {last_success} "
|
||||
f"| {duration} "
|
||||
f"| {link} |"
|
||||
)
|
||||
|
||||
log(f"{repo}: {state}")
|
||||
|
||||
# ======================================
|
||||
# README aktualisieren
|
||||
# ======================================
|
||||
|
||||
content = build_readme(
|
||||
cfg,
|
||||
rows,
|
||||
stats
|
||||
)
|
||||
|
||||
existing, sha = get_content(
|
||||
"README.md"
|
||||
)
|
||||
|
||||
if existing == content:
|
||||
print('README unverändert.')
|
||||
|
||||
log("README unverändert.")
|
||||
|
||||
else:
|
||||
path = '/repos/{}/{}/contents/README.md'.format(urllib.parse.quote(OWNER), urllib.parse.quote(STATUS_REPO))
|
||||
req(path, 'PUT', {'branch': BRANCH, 'sha': sha,
|
||||
'content': base64.b64encode(content.encode()).decode(),
|
||||
'message': 'docs: update vulnerability sync status [skip ci]'})
|
||||
print('README aktualisiert.')
|
||||
|
||||
update_readme(content, sha)
|
||||
|
||||
log("README aktualisiert.")
|
||||
|
||||
# ======================================
|
||||
# Fehlerbehandlung
|
||||
# ======================================
|
||||
|
||||
if errors:
|
||||
|
||||
for error in errors:
|
||||
print('ERROR: ' + error, file=sys.stderr)
|
||||
print(
|
||||
"ERROR: " + error,
|
||||
file=sys.stderr
|
||||
)
|
||||
|
||||
sys.exit(1)
|
||||
|
||||
log("======================================")
|
||||
log("Dashboard erfolgreich aktualisiert")
|
||||
log("======================================")
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user