Add runner-logs mise task for Forgejo Actions
Authenticated runner-logs script that lists workflow runs and fetches job logs. Uses Forgejo API token from 1Password and auto-detects the repo from git remote. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
140b3f3e9f
commit
dfbcb72e94
1 changed files with 254 additions and 0 deletions
254
mise-tasks/runner-logs
Executable file
254
mise-tasks/runner-logs
Executable file
|
|
@ -0,0 +1,254 @@
|
|||
#!/usr/bin/env -S uv run --script
|
||||
# /// script
|
||||
# requires-python = ">=3.12"
|
||||
# dependencies = ["httpx>=0.28.1", "rich>=14.0.0", "typer>=0.24.0"]
|
||||
# ///
|
||||
#MISE description="List recent Forgejo Actions runs or fetch logs for a specific job"
|
||||
#USAGE arg "[run_number]" help="Run number to show jobs for (omit to list recent runs)"
|
||||
#USAGE flag "--job -j <job>" help="Job index (0-based) to fetch logs for"
|
||||
#USAGE flag "--repo <repo>" help="Forge repo (owner/name), default: detected from git remote"
|
||||
#USAGE flag "--limit -n <limit>" help="Max runs to display (0 for all)"
|
||||
#USAGE flag "--token <token>" help="Forgejo API token (default: read from 1Password)"
|
||||
"""List recent Forgejo Actions runs and fetch job logs.
|
||||
|
||||
Usage:
|
||||
mise run runner-logs # list recent runs (default 15)
|
||||
mise run runner-logs -n 0 # list ALL runs
|
||||
mise run runner-logs --repo eblume/hermes # list runs for a different repo
|
||||
mise run runner-logs 474 # show jobs in run 474
|
||||
mise run runner-logs 474 -j 1 # fetch logs for job 1 of run 474
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from typing import Annotated
|
||||
|
||||
import httpx
|
||||
import typer
|
||||
from rich.console import Console
|
||||
from rich.table import Table
|
||||
|
||||
FORGE_URL = "https://forge.ops.eblu.me"
|
||||
FORGE_API = f"{FORGE_URL}/api/v1"
|
||||
OP_TOKEN_REF = "op://vg6xf6vvfmoh5hqjjhlhbeoaie/w3663ffnvkewbftncqxtcpeavy/api-token"
|
||||
|
||||
app = typer.Typer(add_completion=False)
|
||||
|
||||
|
||||
def resolve_token(explicit_token: str | None, console: Console) -> str:
|
||||
"""Resolve Forgejo API token: explicit flag > FORGEJO_TOKEN env > 1Password."""
|
||||
if explicit_token:
|
||||
return explicit_token
|
||||
env_token = os.environ.get("FORGEJO_TOKEN", "").strip()
|
||||
if env_token:
|
||||
return env_token
|
||||
console.print("[dim]Reading Forgejo API token from 1Password...[/dim]")
|
||||
result = subprocess.run(
|
||||
["op", "read", OP_TOKEN_REF],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def detect_repo_from_git() -> str | None:
|
||||
"""Sniff owner/repo from the git remote 'origin' URL."""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "remote", "get-url", "origin"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True,
|
||||
)
|
||||
except (subprocess.CalledProcessError, FileNotFoundError):
|
||||
return None
|
||||
url = result.stdout.strip()
|
||||
# Match SSH (git@host:owner/repo.git) or HTTPS (https://host/owner/repo.git)
|
||||
m = re.search(r"[/:]([^/]+/[^/]+?)(?:\.git)?$", url)
|
||||
if not m:
|
||||
return None
|
||||
candidate = m.group(1)
|
||||
# Only use it if the remote points at our forge
|
||||
if "forge.ops.eblu.me" in url or "forge.eblu.me" in url:
|
||||
return candidate
|
||||
return None
|
||||
|
||||
|
||||
def auth_headers(token: str) -> dict[str, str]:
|
||||
return {"Authorization": f"token {token}"}
|
||||
|
||||
|
||||
def fetch_tasks(repo: str, token: str) -> list[dict]:
|
||||
"""Fetch all tasks from the Forgejo API, paginating if needed."""
|
||||
tasks: list[dict] = []
|
||||
page = 1
|
||||
while True:
|
||||
resp = httpx.get(
|
||||
f"{FORGE_API}/repos/{repo}/actions/tasks",
|
||||
params={"page": page, "limit": 50},
|
||||
headers=auth_headers(token),
|
||||
timeout=15,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
batch = resp.json().get("workflow_runs", [])
|
||||
if not batch:
|
||||
break
|
||||
tasks.extend(batch)
|
||||
page += 1
|
||||
return tasks
|
||||
|
||||
|
||||
def list_runs(repo: str, limit: int, token: str, console: Console) -> None:
|
||||
"""List recent workflow runs, grouped by run number."""
|
||||
tasks = fetch_tasks(repo, token)
|
||||
|
||||
# Group tasks by run_number
|
||||
runs: dict[int, list[dict]] = {}
|
||||
for t in tasks:
|
||||
rn = t["run_number"]
|
||||
runs.setdefault(rn, []).append(t)
|
||||
|
||||
table = Table(title=f"Recent runs — {repo}")
|
||||
table.add_column("Run #", style="cyan", no_wrap=True)
|
||||
table.add_column("Status")
|
||||
table.add_column("Jobs")
|
||||
table.add_column("Title")
|
||||
table.add_column("Event")
|
||||
|
||||
shown = 0
|
||||
for rn in sorted(runs, reverse=True):
|
||||
if limit > 0 and shown >= limit:
|
||||
break
|
||||
|
||||
jobs = sorted(runs[rn], key=lambda x: x["id"])
|
||||
|
||||
# Aggregate status: worst status wins
|
||||
statuses = [j.get("status", "") for j in jobs]
|
||||
if "failure" in statuses:
|
||||
status, style = "failure", "red"
|
||||
elif "running" in statuses or "waiting" in statuses:
|
||||
status, style = "running", "yellow"
|
||||
elif all(s == "success" for s in statuses):
|
||||
status, style = "success", "green"
|
||||
else:
|
||||
status, style = statuses[0], "yellow"
|
||||
|
||||
job_names = ", ".join(j.get("name", "?")[:30] for j in jobs)
|
||||
title = (jobs[0].get("display_title") or "")[:40]
|
||||
event = jobs[0].get("event", "")
|
||||
|
||||
table.add_row(
|
||||
str(rn),
|
||||
f"[{style}]{status}[/{style}]",
|
||||
job_names,
|
||||
title,
|
||||
event,
|
||||
)
|
||||
shown += 1
|
||||
|
||||
console.print(table)
|
||||
console.print("\n[dim]Use: mise run runner-logs <run#> to see jobs in a run[/dim]")
|
||||
console.print("[dim] mise run runner-logs <run#> -j N to fetch logs for job N[/dim]")
|
||||
|
||||
|
||||
def show_jobs(run_number: int, repo: str, token: str, console: Console) -> None:
|
||||
"""Show the jobs within a specific run."""
|
||||
tasks = fetch_tasks(repo, token)
|
||||
|
||||
jobs = sorted(
|
||||
[t for t in tasks if t["run_number"] == run_number],
|
||||
key=lambda x: x["id"],
|
||||
)
|
||||
if not jobs:
|
||||
typer.echo(f"Error: No jobs found for run #{run_number}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
table = Table(title=f"Jobs in run #{run_number} — {repo}")
|
||||
table.add_column("Job #", style="cyan", no_wrap=True)
|
||||
table.add_column("Status")
|
||||
table.add_column("Name")
|
||||
table.add_column("Created")
|
||||
|
||||
for i, job in enumerate(jobs):
|
||||
status = job.get("status", "")
|
||||
style = "green" if status == "success" else "red" if status == "failure" else "yellow"
|
||||
table.add_row(
|
||||
str(i),
|
||||
f"[{style}]{status}[/{style}]",
|
||||
job.get("name", ""),
|
||||
job.get("created_at", ""),
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
console.print(f"\n[dim]Use: mise run runner-logs {run_number} -j N to fetch logs for job N[/dim]")
|
||||
|
||||
|
||||
def fetch_log(run_number: int, job_index: int, repo: str, token: str) -> None:
|
||||
"""Fetch logs for a specific job via the Forgejo web endpoint."""
|
||||
url = f"{FORGE_URL}/{repo}/actions/runs/{run_number}/jobs/{job_index}/attempt/1/logs"
|
||||
resp = httpx.get(url, headers=auth_headers(token), timeout=30, follow_redirects=True)
|
||||
if resp.status_code == 404:
|
||||
typer.echo(
|
||||
f"Error: No logs found for run #{run_number} job {job_index}",
|
||||
err=True,
|
||||
)
|
||||
typer.echo(f"URL: {url}", err=True)
|
||||
raise typer.Exit(1)
|
||||
resp.raise_for_status()
|
||||
sys.stdout.write(resp.text)
|
||||
|
||||
|
||||
@app.command()
|
||||
def main(
|
||||
run_number: Annotated[
|
||||
int | None,
|
||||
typer.Argument(help="Run number to show jobs for (omit to list recent runs)"),
|
||||
] = None,
|
||||
job: Annotated[
|
||||
int | None,
|
||||
typer.Option("--job", "-j", help="Job index (0-based) to fetch logs for"),
|
||||
] = None,
|
||||
repo: Annotated[
|
||||
str | None,
|
||||
typer.Option("--repo", help="Forge repo (owner/name), default: detected from git remote"),
|
||||
] = None,
|
||||
limit: Annotated[
|
||||
int,
|
||||
typer.Option("--limit", "-n", help="Max runs to display (0 for all)"),
|
||||
] = 15,
|
||||
token: Annotated[
|
||||
str | None,
|
||||
typer.Option("--token", help="Forgejo API token (default: read from 1Password)"),
|
||||
] = None,
|
||||
) -> None:
|
||||
"""List recent Forgejo Actions runs or fetch logs for a specific job."""
|
||||
console = Console()
|
||||
|
||||
if repo is None:
|
||||
repo = detect_repo_from_git()
|
||||
if repo is None:
|
||||
typer.echo(
|
||||
"Error: could not detect repo from git remote; use --repo owner/name",
|
||||
err=True,
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
console.print(f"[dim]Detected repo: {repo}[/dim]")
|
||||
|
||||
resolved_token = resolve_token(token, console)
|
||||
|
||||
if run_number is None:
|
||||
if job is not None:
|
||||
typer.echo("Error: --job requires a run number", err=True)
|
||||
raise typer.Exit(1)
|
||||
list_runs(repo, limit, resolved_token, console)
|
||||
elif job is None:
|
||||
show_jobs(run_number, repo, resolved_token, console)
|
||||
else:
|
||||
fetch_log(run_number, job, repo, resolved_token)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
app()
|
||||
Loading…
Add table
Add a link
Reference in a new issue