mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 02:54:26 +00:00
kevinveenbirkenbachandClaude Opus 4.8
9c029870fe
python -m cli.meta.ci.logs.download fetches a run's job logs and artefacts in parallel, with conclusion filters (-s/-f/-c/-k), a jobs.json/summary.tsv manifest, per-request jitter to dodge API rate limits, and idempotent re-runs. python -m cli.meta.ci.logs.analyze clusters the downloaded logs by their root failing task. Rename the agent CI-triage page debug/pipeline.md to debug/ci.md, point its download/inspect step at the new CLI, and reference the tooling from the contributing CI-debugging page. Add unit tests for both commands. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
|
|
_URL_RE = re.compile(
|
|
r"github\.com/(?P<owner>[^/]+)/(?P<repo>[^/]+)/actions/runs/(?P<run>\d+)"
|
|
)
|
|
|
|
|
|
def gh_proc(args: list[str]) -> subprocess.CompletedProcess[str]:
|
|
"""Run ``gh`` with the given arguments and return the completed process."""
|
|
return subprocess.run(["gh", *args], capture_output=True, text=True, check=False)
|
|
|
|
|
|
def gh(args: list[str]) -> str:
|
|
"""Run ``gh`` and return stdout, raising on a non-zero exit."""
|
|
proc = gh_proc(args)
|
|
if proc.returncode != 0:
|
|
raise subprocess.CalledProcessError(
|
|
proc.returncode, proc.args, proc.stdout, proc.stderr
|
|
)
|
|
return proc.stdout
|
|
|
|
|
|
def resolve_run(ref: str, repo_override: str | None) -> tuple[str, str, str]:
|
|
"""Return (owner, repo, run_id) from a run id or a run/job URL.
|
|
|
|
A bare run id resolves owner/repo from ``repo_override`` or the current gh repo.
|
|
"""
|
|
m = _URL_RE.search(ref)
|
|
if m:
|
|
return m.group("owner"), m.group("repo"), m.group("run")
|
|
if not ref.isdigit():
|
|
raise ValueError(f"not a run id or a run URL: {ref!r}")
|
|
nwo = (
|
|
repo_override
|
|
or gh(
|
|
["repo", "view", "--json", "nameWithOwner", "--jq", ".nameWithOwner"]
|
|
).strip()
|
|
)
|
|
owner, _, repo = nwo.partition("/")
|
|
if not repo:
|
|
raise ValueError(
|
|
f"could not resolve OWNER/REPO (got {nwo!r}); pass --repo OWNER/REPO"
|
|
)
|
|
return owner, repo, ref
|
|
|
|
|
|
def list_jobs(owner: str, repo: str, run: str) -> list[dict]:
|
|
out = gh(
|
|
[
|
|
"api",
|
|
f"repos/{owner}/{repo}/actions/runs/{run}/jobs",
|
|
"--paginate",
|
|
"--jq",
|
|
".jobs[] | {id, name, conclusion, status}",
|
|
]
|
|
)
|
|
return [json.loads(line) for line in out.splitlines() if line.strip()]
|
|
|
|
|
|
def list_artifacts(owner: str, repo: str, run: str) -> list[str]:
|
|
out = gh(
|
|
[
|
|
"api",
|
|
f"repos/{owner}/{repo}/actions/runs/{run}/artifacts",
|
|
"--paginate",
|
|
"--jq",
|
|
".artifacts[].name",
|
|
]
|
|
)
|
|
return [line.strip() for line in out.splitlines() if line.strip()]
|