scripts/update_status.py aktualisiert
This commit is contained in:
+205
-623
@@ -1,11 +1,5 @@
|
||||
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Gitea Actions Status Dashboard
|
||||
Kompatibel mit Gitea 1.27.x
|
||||
Nur Python-Standardbibliothek erforderlich.
|
||||
"""
|
||||
|
||||
"""Gitea Actions dashboard with self-hosted SVG status icons (stdlib only)."""
|
||||
import base64
|
||||
import datetime as dt
|
||||
import json
|
||||
@@ -14,681 +8,269 @@ import sys
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 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")
|
||||
|
||||
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")
|
||||
TZ = ZoneInfo('Europe/Berlin')
|
||||
PAGE_SIZE, MAX_PAGES = 50, 5
|
||||
|
||||
PAGE_SIZE = 50
|
||||
MAX_PAGES = 5
|
||||
|
||||
# Markdown-Emoji-Codes
|
||||
# The URLs use Gitea's raw file route; no externally hosted graphics.
|
||||
ICONS = {
|
||||
"success": ":green_circle:",
|
||||
"failure": ":red_circle:",
|
||||
"running": ":large_blue_circle:",
|
||||
"waiting": ":yellow_circle:",
|
||||
"unknown": ":white_circle:",
|
||||
"warning": ":warning:",
|
||||
"error": ":x:",
|
||||
'success': ('#208637', 'check'),
|
||||
'failure': ('#d1242f', 'cross'),
|
||||
'running': ('#0969da', 'play'),
|
||||
'waiting': ('#b78103', 'clock'),
|
||||
'unknown': ('#6e7781', 'question'),
|
||||
'warning': ('#b78103', 'alert'),
|
||||
'error': ('#d1242f', 'cross'),
|
||||
}
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Hilfsfunktionen
|
||||
# ==========================================
|
||||
|
||||
def log(message):
|
||||
print(message, flush=True)
|
||||
def q(value):
|
||||
return urllib.parse.quote(str(value), safe='')
|
||||
|
||||
|
||||
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/2.0",
|
||||
},
|
||||
)
|
||||
|
||||
def req(path, method='GET', data=None):
|
||||
encoded = json.dumps(data).encode('utf-8') if data is not None else None
|
||||
request = urllib.request.Request(API + path, data=encoded, method=method, headers={
|
||||
'Authorization': 'token ' + TOKEN,
|
||||
'Accept': 'application/json',
|
||||
'Content-Type': 'application/json',
|
||||
'User-Agent': 'vuln-status/3.0',
|
||||
})
|
||||
try:
|
||||
with urllib.request.urlopen(
|
||||
request,
|
||||
timeout=35
|
||||
) as response:
|
||||
|
||||
with urllib.request.urlopen(request, timeout=40) 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
|
||||
return json.loads(payload) if payload else {}
|
||||
except urllib.error.HTTPError as e:
|
||||
detail = e.read(500).decode(errors='replace')
|
||||
raise RuntimeError(f'HTTP {e.code} {path}: {detail}') from e
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Repository-Dateien
|
||||
# ==========================================
|
||||
|
||||
def content_path(filename):
|
||||
|
||||
return (
|
||||
f"/repos/{quote(OWNER)}/"
|
||||
f"{quote(STATUS_REPO)}/contents/"
|
||||
f"{urllib.parse.quote(filename, safe='/')}"
|
||||
f"?ref={quote(BRANCH)}"
|
||||
)
|
||||
def file_path(filename):
|
||||
return f'/repos/{q(OWNER)}/{q(STATUS_REPO)}/contents/{urllib.parse.quote(filename, safe="/")}'
|
||||
|
||||
|
||||
def get_content(filename):
|
||||
|
||||
obj = req(content_path(filename))
|
||||
|
||||
content = base64.b64decode(
|
||||
obj["content"]
|
||||
).decode("utf-8")
|
||||
|
||||
return content, obj["sha"]
|
||||
def get_file(filename):
|
||||
try:
|
||||
result = req(file_path(filename) + '?ref=' + q(BRANCH))
|
||||
except RuntimeError as e:
|
||||
if str(e).startswith('HTTP 404 '):
|
||||
return None, None
|
||||
raise
|
||||
return base64.b64decode(result['content']).decode('utf-8'), result['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]"
|
||||
),
|
||||
def put_file(filename, content, message, old=None):
|
||||
if old is None:
|
||||
existing, sha = get_file(filename)
|
||||
else:
|
||||
existing, sha = old
|
||||
if existing == content:
|
||||
return False
|
||||
body = {
|
||||
'branch': BRANCH,
|
||||
'content': base64.b64encode(content.encode('utf-8')).decode('ascii'),
|
||||
'message': message + ' [skip ci]',
|
||||
}
|
||||
if sha:
|
||||
body['sha'] = sha
|
||||
req(file_path(filename), 'PUT' if sha else 'POST', body)
|
||||
return True
|
||||
|
||||
return req(path, "PUT", data)
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Datum und Laufzeit
|
||||
# ==========================================
|
||||
|
||||
def parse_time(raw):
|
||||
|
||||
if not raw or str(raw).startswith("0001-"):
|
||||
if not raw or str(raw).startswith('0001-'):
|
||||
return None
|
||||
|
||||
try:
|
||||
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)
|
||||
|
||||
result = dt.datetime.fromisoformat(str(raw).replace('Z', '+00:00'))
|
||||
if result.tzinfo is None:
|
||||
result = result.replace(tzinfo=dt.timezone.utc)
|
||||
return result.astimezone(dt.timezone.utc)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def fmt(value):
|
||||
|
||||
if value is None:
|
||||
return "–"
|
||||
|
||||
return value.astimezone(TIMEZONE).strftime(
|
||||
"%d.%m.%Y %H:%M %Z"
|
||||
)
|
||||
def display_time(value):
|
||||
return value.astimezone(TZ).strftime('%d.%m.%Y %H:%M %Z') if value else '–'
|
||||
|
||||
|
||||
def elapsed(start, end):
|
||||
|
||||
if not start or not end or end < start:
|
||||
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"
|
||||
return '–'
|
||||
seconds = int((end - start).total_seconds())
|
||||
h, rem = divmod(seconds, 3600)
|
||||
m, s = divmod(rem, 60)
|
||||
return f'{h}h {m:02d}m' if h else f'{m}m {s:02d}s'
|
||||
|
||||
|
||||
def safe_cell(value):
|
||||
|
||||
return (
|
||||
str(value)
|
||||
.replace("|", "\\|")
|
||||
.replace("\n", " ")
|
||||
.replace("\r", " ")
|
||||
)
|
||||
def cell(value):
|
||||
return str(value).replace('|', '\\|').replace('\n', ' ').replace('\r', ' ')
|
||||
|
||||
|
||||
# ==========================================
|
||||
# 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)
|
||||
|
||||
return result.get("workflow_runs", [])
|
||||
def svg(name):
|
||||
color, shape = ICONS[name]
|
||||
graphic = {
|
||||
'check': '<circle cx="12" cy="12" r="10"/><path d="m7.3 12.3 3.2 3.1 6.3-6.6"/>',
|
||||
'cross': '<circle cx="12" cy="12" r="10"/><path d="m8 8 8 8M16 8l-8 8"/>',
|
||||
'clock': '<circle cx="12" cy="12" r="10"/><path d="M12 6v6l4 2"/>',
|
||||
'play': '<circle cx="12" cy="12" r="10"/><path d="m10 7 6 5-6 5z"/>',
|
||||
'question': '<circle cx="12" cy="12" r="10"/><path d="M9 9a3 3 0 1 1 4 3c-1 .6-1 1-1 2M12 17h.01"/>',
|
||||
'alert': '<path d="M12 2 23 21H1z"/><path d="M12 9v5M12 18h.01"/>',
|
||||
}[shape]
|
||||
return (f'<svg xmlns="http://www.w3.org/2000/svg" width="24" height="24" '
|
||||
f'viewBox="0 0 24 24" fill="none" stroke="{color}" '
|
||||
f'stroke-width="2" stroke-linecap="round" stroke-linejoin="round">'
|
||||
f'{graphic}</svg>\n')
|
||||
|
||||
|
||||
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 icon_md(name):
|
||||
# Absolute image URL works in the Gitea README from all link contexts.
|
||||
# The repository must be readable by the viewer; private repos require login.
|
||||
url = (f'{BASE}/{q(OWNER)}/{q(STATUS_REPO)}/raw/branch/'
|
||||
f'{q(BRANCH)}/assets/status-{name}.svg')
|
||||
return f''
|
||||
|
||||
|
||||
def get_runs(owner, repo, workflow=None):
|
||||
|
||||
runs = []
|
||||
|
||||
def get_runs(owner, repo, workflow):
|
||||
found = []
|
||||
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.
|
||||
params = urllib.parse.urlencode({'page': page, 'limit': PAGE_SIZE})
|
||||
path = f'/repos/{q(owner)}/{q(repo)}/actions/runs?{params}'
|
||||
chunk = req(path).get('workflow_runs', [])
|
||||
for run in chunk:
|
||||
if workflow:
|
||||
name = str(run.get('path') or '').split('@', 1)[0].rsplit('/', 1)[-1]
|
||||
if name != workflow:
|
||||
continue
|
||||
found.append(run)
|
||||
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
|
||||
found.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 found
|
||||
|
||||
|
||||
# ==========================================
|
||||
# Jobstatus
|
||||
# ==========================================
|
||||
|
||||
def get_state(run):
|
||||
|
||||
def current_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")
|
||||
)
|
||||
return 'unknown', 'Kein Lauf'
|
||||
status = str(run.get('status') or '').lower()
|
||||
conclusion = str(run.get('conclusion') or '').lower()
|
||||
if status in ('running', 'in_progress'):
|
||||
return 'running', 'Läuft'
|
||||
if status in ('queued', 'pending', 'requested', 'waiting'):
|
||||
return 'waiting', 'Wartet'
|
||||
if conclusion == 'success':
|
||||
return 'success', 'Erfolgreich'
|
||||
if conclusion in ('failure', 'timed_out', 'startup_failure'):
|
||||
return 'failure', 'Fehlgeschlagen'
|
||||
if conclusion in ('cancelled', 'skipped'):
|
||||
return 'unknown', conclusion.capitalize()
|
||||
return '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
|
||||
)
|
||||
|
||||
def inspect_source(cfg, source):
|
||||
owner = source.get('owner', cfg.get('owner', OWNER))
|
||||
repo = source['repo']
|
||||
runs = get_runs(owner, repo, source.get('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
|
||||
)
|
||||
|
||||
successful = next((r for r in runs if str(r.get('conclusion') or '').lower() == 'success'), None)
|
||||
success_at = (parse_time(successful.get('completed_at')) or
|
||||
parse_time(successful.get('updated_at')) or
|
||||
parse_time(successful.get('created_at'))) if successful else None
|
||||
max_age = float(source.get('max_age_hours', cfg.get('max_age_hours', 26)))
|
||||
overdue = success_at is None or NOW - success_at > dt.timedelta(hours=max_age)
|
||||
kind, state = current_state(latest)
|
||||
state_text = f'{icon_md(kind)} {state}'
|
||||
if overdue:
|
||||
state += (
|
||||
" / "
|
||||
+ ICONS["warning"]
|
||||
+ " Überfällig"
|
||||
)
|
||||
state_text += f' / {icon_md("warning")} Überfällig'
|
||||
started = (parse_time(latest.get('started_at')) or
|
||||
parse_time(latest.get('run_started_at')) or
|
||||
parse_time(latest.get('created_at'))) if latest else None
|
||||
completed = (parse_time(latest.get('completed_at')) or
|
||||
parse_time(latest.get('updated_at'))) if latest and str(latest.get('status') or '').lower() == 'completed' else None
|
||||
duration = elapsed(started, completed)
|
||||
link = '–'
|
||||
if latest and latest.get('id'):
|
||||
url = latest.get('html_url') or f'{BASE}/{q(owner)}/{q(repo)}/actions/runs/{latest["id"]}'
|
||||
link = f'[{latest["id"]}]({url})'
|
||||
return kind, overdue, state_text, display_time(success_at), duration, link
|
||||
|
||||
# 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():
|
||||
cfg_raw, _ = get_file('sources.json')
|
||||
if cfg_raw is None:
|
||||
raise RuntimeError('sources.json nicht gefunden')
|
||||
cfg = json.loads(cfg_raw)
|
||||
# Ensure all SVG graphics exist before referencing them in README.
|
||||
for name in ICONS:
|
||||
filename = f'assets/status-{name}.svg'
|
||||
changed = put_file(filename, svg(name), f'assets: update {name} status icon')
|
||||
if changed:
|
||||
print(f'SVG erstellt: {filename}', flush=True)
|
||||
|
||||
log("======================================")
|
||||
log("Vulnerability Status Dashboard")
|
||||
log("======================================")
|
||||
|
||||
config_text, _ = get_content(
|
||||
"sources.json"
|
||||
)
|
||||
|
||||
cfg = json.loads(config_text)
|
||||
|
||||
rows = []
|
||||
errors = []
|
||||
|
||||
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"
|
||||
)
|
||||
|
||||
counts = dict(total=len(cfg['sources']), success=0, failure=0, running=0,
|
||||
waiting=0, overdue=0, error=0, unknown=0)
|
||||
errors, rows = [], []
|
||||
for src in cfg['sources']:
|
||||
owner = src.get('owner', cfg.get('owner', OWNER))
|
||||
repo = src['repo']
|
||||
title = src.get('name', repo)
|
||||
url = f'{BASE}/{q(owner)}/{q(repo)}/actions'
|
||||
try:
|
||||
|
||||
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
|
||||
|
||||
kind, overdue, state, success, duration, link = inspect_source(cfg, src)
|
||||
counts[kind] += 1
|
||||
if overdue:
|
||||
counts['overdue'] += 1
|
||||
except Exception as exc:
|
||||
state, success, duration, link = f'{icon_md("error")} API-Fehler', '–', '–', '–'
|
||||
counts['error'] += 1
|
||||
errors.append(f'{repo}: {exc}')
|
||||
rows.append(f'| [{cell(title)}]({url}) | {state} | {success} | {duration} | {link} |')
|
||||
print(f'{repo}: {state.split(" ", 1)[-1]}', flush=True)
|
||||
|
||||
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:
|
||||
|
||||
log("README unverändert.")
|
||||
|
||||
else:
|
||||
|
||||
update_readme(content, sha)
|
||||
|
||||
log("README aktualisiert.")
|
||||
|
||||
# ======================================
|
||||
# Fehlerbehandlung
|
||||
# ======================================
|
||||
|
||||
lines = [
|
||||
'# Vulnerability Database – Sync Status', '',
|
||||
'Automatisch generiertes Dashboard für Gitea Actions.', '',
|
||||
f'**Letzte Prüfung:** {display_time(NOW)} ',
|
||||
f'**Überfällig:** Kein erfolgreicher Lauf innerhalb von {cfg.get("max_age_hours", 26)} Stunden (pro Quelle anpassbar).',
|
||||
'', '## Übersicht', '',
|
||||
f'- Erfolgreich: **{counts["success"]}**',
|
||||
f'- Laufend: **{counts["running"]}**',
|
||||
f'- Wartend: **{counts["waiting"]}**',
|
||||
f'- Fehlgeschlagen: **{counts["failure"]}**',
|
||||
f'- Überfällig: **{counts["overdue"]}**',
|
||||
f'- API-Fehler: **{counts["error"]}**',
|
||||
'', '## Synchronisierungsstatus', '',
|
||||
'| Datenquelle | Status | Letzter Erfolg | Dauer letzter Lauf | Letzter Job |',
|
||||
'|---|---|---|---|---|',
|
||||
*rows, '', '## Legende', '',
|
||||
f'- {icon_md("success")} Erfolgreich',
|
||||
f'- {icon_md("waiting")} Wartet',
|
||||
f'- {icon_md("running")} Läuft',
|
||||
f'- {icon_md("failure")} Fehlgeschlagen',
|
||||
f'- {icon_md("warning")} Überfällig',
|
||||
f'- {icon_md("error")} API-Fehler',
|
||||
'', '## Links', '',
|
||||
'- [Vulnerability Database](https://vulnerability.apex-project.de)',
|
||||
'- [Projektwebsite](https://www.apex-project.de)', '',
|
||||
]
|
||||
content = '\n'.join(lines)
|
||||
changed = put_file('README.md', content, 'docs: update vulnerability sync status')
|
||||
print('README aktualisiert.' if changed else 'README unverändert.', flush=True)
|
||||
if errors:
|
||||
|
||||
for error in errors:
|
||||
print(
|
||||
"ERROR: " + error,
|
||||
file=sys.stderr
|
||||
)
|
||||
|
||||
for err in errors:
|
||||
print('ERROR:', err, 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