700 lines
24 KiB
Python
700 lines
24 KiB
Python
#!/usr/bin/env python3
|
|
"""Collect Gitea Actions CI evidence: version, runs, jobs, and job logs.
|
|
|
|
This is an evidence-collection helper for the gitea-fix-ci skill. It automates
|
|
the programmatic parts of the skill's Procedure (probe version, list failing
|
|
runs, list jobs, download a failing job's log) so the agent does not hand-build
|
|
API paths or blindly pick job index 0. It stops at evidence: classification,
|
|
fix plans, and code edits stay with the agent.
|
|
|
|
Scope decisions (see SKILL.md):
|
|
- Evidence only. No classification, no fix, no edits.
|
|
- Gitea Actions REST API only. No legacy web-scraping fallback; when the API is
|
|
absent the tool points the agent back to SKILL.md's manual/web path.
|
|
- Auth via GITEA_TOKEN (preferred) or an explicitly requested `git credential`
|
|
lookup; base URL/owner/repo are derived from `git remote -v`. Secrets are
|
|
never placed on argv, logs, or exception text.
|
|
|
|
Zero third-party dependencies (stdlib urllib only).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
TOKEN_ENV = "GITEA_TOKEN"
|
|
DEFAULT_REMOTE = "origin"
|
|
DEFAULT_LOG_TAIL = 40
|
|
REQUEST_TIMEOUT = 30 # seconds; a hung Gitea/proxy must not block the session
|
|
CREDENTIAL_TIMEOUT = 10 # seconds; helpers must not block on an interactive UI
|
|
USER_AGENT = "gitea-fix-ci/fetch_ci_logs"
|
|
|
|
# https://host/owner/repo(.git) or git@host:owner/repo(.git) or
|
|
# ssh://git@host[:port]/owner/repo(.git)
|
|
_HTTP_REMOTE_RE = re.compile(
|
|
r"^(?P<scheme>https?)://(?:[^@/]+@)?(?P<host>[^/]+)/"
|
|
r"(?P<owner>[^/]+)/(?P<repo>[^/]+?)(?:\.git)?/?$"
|
|
)
|
|
_SCP_REMOTE_RE = re.compile(
|
|
r"^(?:ssh://)?(?:[^@]+@)?(?P<host>[^:/]+)(?::\d+)?[:/]"
|
|
r"(?P<owner>[^/]+)/(?P<repo>[^/]+?)(?:\.git)?/?$"
|
|
)
|
|
|
|
|
|
class ConfigError(ValueError):
|
|
"""Missing or malformed configuration (remote, token, arguments)."""
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
"""The Gitea API returned an error or unexpected payload."""
|
|
|
|
|
|
def _eprint(*args: object) -> None:
|
|
print(*args, file=sys.stderr)
|
|
|
|
|
|
def _origin(url: str) -> tuple[str, str, int]:
|
|
"""Return a normalized (scheme, hostname, effective port) tuple."""
|
|
try:
|
|
parsed = urllib.parse.urlsplit(url)
|
|
hostname = parsed.hostname
|
|
port = parsed.port
|
|
except ValueError:
|
|
raise ConfigError("invalid Gitea URL") from None
|
|
|
|
if parsed.scheme.lower() not in {"http", "https"} or not parsed.netloc:
|
|
raise ConfigError("Gitea URL must use http:// or https://")
|
|
if parsed.username is not None or parsed.password is not None:
|
|
raise ConfigError("Gitea URL must not contain embedded credentials")
|
|
if not hostname:
|
|
raise ConfigError("Gitea URL must include a host")
|
|
|
|
scheme = parsed.scheme.lower()
|
|
effective_port = port if port is not None else (443 if scheme == "https" else 80)
|
|
return scheme, hostname.rstrip(".").lower(), effective_port
|
|
|
|
|
|
def _normalize_base_url(value: str) -> str:
|
|
"""Validate and normalize the configured API base URL."""
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ConfigError("Gitea base URL is required")
|
|
value = value.strip().rstrip("/")
|
|
try:
|
|
parsed = urllib.parse.urlsplit(value)
|
|
except ValueError:
|
|
raise ConfigError("invalid Gitea base URL") from None
|
|
if parsed.query or parsed.fragment:
|
|
raise ConfigError("Gitea base URL must not contain a query or fragment")
|
|
_origin(value)
|
|
return value
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RepoTarget:
|
|
base_url: str # e.g. https://git.example.com
|
|
owner: str
|
|
repo: str
|
|
|
|
def __post_init__(self) -> None:
|
|
object.__setattr__(self, "base_url", _normalize_base_url(self.base_url))
|
|
if not self.owner or not self.repo:
|
|
raise ConfigError("Gitea owner and repository are required")
|
|
|
|
@property
|
|
def api_root(self) -> str:
|
|
return f"{self.base_url}/api/v1"
|
|
|
|
def repo_path(self, suffix: str) -> str:
|
|
owner = urllib.parse.quote(self.owner, safe="")
|
|
repo = urllib.parse.quote(self.repo, safe="")
|
|
return f"{self.api_root}/repos/{owner}/{repo}{suffix}"
|
|
|
|
@property
|
|
def origin(self) -> tuple[str, str, int]:
|
|
return _origin(self.base_url)
|
|
|
|
@property
|
|
def credential_host(self) -> str:
|
|
parsed = urllib.parse.urlsplit(self.base_url)
|
|
hostname = parsed.hostname or ""
|
|
if ":" in hostname and not hostname.startswith("["):
|
|
hostname = f"[{hostname}]"
|
|
if parsed.port is not None:
|
|
hostname = f"{hostname}:{parsed.port}"
|
|
return hostname
|
|
|
|
@property
|
|
def credential_path(self) -> str:
|
|
base_path = urllib.parse.urlsplit(self.base_url).path.strip("/")
|
|
owner = urllib.parse.quote(self.owner, safe="")
|
|
repo = urllib.parse.quote(self.repo, safe="")
|
|
repository_path = f"{owner}/{repo}.git"
|
|
return f"{base_path}/{repository_path}" if base_path else repository_path
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ApiAuth:
|
|
"""An authorization header whose secret is deliberately absent from repr."""
|
|
|
|
source: str
|
|
authorization: str = field(repr=False)
|
|
|
|
def __post_init__(self) -> None:
|
|
scheme, separator, credential = self.authorization.partition(" ")
|
|
has_control = any(
|
|
ord(character) < 32 or ord(character) == 127
|
|
for character in self.authorization
|
|
)
|
|
try:
|
|
self.authorization.encode("ascii")
|
|
except UnicodeEncodeError:
|
|
raise ConfigError("invalid authorization value") from None
|
|
if (
|
|
not self.source
|
|
or not separator
|
|
or not scheme
|
|
or not credential
|
|
or has_control
|
|
):
|
|
raise ConfigError("invalid authorization value")
|
|
|
|
|
|
def _require_https(target: RepoTarget) -> None:
|
|
if target.origin[0] != "https":
|
|
raise ConfigError("credentials require an HTTPS Gitea base URL")
|
|
|
|
|
|
def _run_git(args: list[str]) -> str:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", *args],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
except OSError as exc:
|
|
raise ConfigError(f"cannot run git {' '.join(args)}: {exc}") from exc
|
|
if result.returncode != 0:
|
|
detail = result.stderr.strip()
|
|
suffix = f": {detail}" if detail else ""
|
|
raise ConfigError(f"git {' '.join(args)} failed{suffix}")
|
|
return result.stdout.strip()
|
|
|
|
|
|
def _remote_url(remote: str) -> str:
|
|
url = _run_git(["remote", "get-url", remote])
|
|
if not url:
|
|
raise ConfigError(f"remote '{remote}' has no URL")
|
|
return url
|
|
|
|
|
|
def parse_remote_url(url: str) -> tuple[str, str, str]:
|
|
"""Return (base_url, owner, repo) parsed from a git remote URL.
|
|
|
|
Only http(s) remotes yield a usable API base URL. SSH remotes give host and
|
|
path but no scheme, so we assume https for the API base.
|
|
"""
|
|
http_match = _HTTP_REMOTE_RE.match(url)
|
|
if http_match:
|
|
base = f"{http_match.group('scheme')}://{http_match.group('host')}"
|
|
return base, http_match.group("owner"), http_match.group("repo")
|
|
|
|
scp_match = _SCP_REMOTE_RE.match(url)
|
|
if scp_match:
|
|
# No scheme in an SSH remote; the API is reached over https by default.
|
|
base = f"https://{scp_match.group('host')}"
|
|
return base, scp_match.group("owner"), scp_match.group("repo")
|
|
|
|
raise ConfigError(
|
|
"cannot parse owner/repo from the configured git remote. Gitea remotes "
|
|
"are expected as <host>/<owner>/<repo>; for nested or "
|
|
"non-standard paths, pass --base-url/--owner/--repo explicitly."
|
|
)
|
|
|
|
|
|
def resolve_target(args: argparse.Namespace) -> RepoTarget:
|
|
base_url = args.base_url
|
|
owner = args.owner
|
|
repo = args.repo
|
|
if not (base_url and owner and repo):
|
|
url = _remote_url(args.remote)
|
|
parsed_base, parsed_owner, parsed_repo = parse_remote_url(url)
|
|
base_url = base_url or parsed_base
|
|
owner = owner or parsed_owner
|
|
repo = repo or parsed_repo
|
|
return RepoTarget(base_url=base_url, owner=owner, repo=repo)
|
|
|
|
|
|
def _token() -> str | None:
|
|
token = os.getenv(TOKEN_ENV)
|
|
if token and token.strip():
|
|
return token.strip()
|
|
return None
|
|
|
|
|
|
def _git_credential_auth(target: RepoTarget) -> ApiAuth:
|
|
"""Resolve repository-scoped HTTP Basic credentials without prompting."""
|
|
_require_https(target)
|
|
credential_query = (
|
|
"protocol=https\n"
|
|
f"host={target.credential_host}\n"
|
|
f"path={target.credential_path}\n\n"
|
|
)
|
|
helper_env = os.environ.copy()
|
|
helper_env["GIT_TERMINAL_PROMPT"] = "0"
|
|
helper_env["GCM_INTERACTIVE"] = "Never"
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "credential", "fill"],
|
|
input=credential_query,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
timeout=CREDENTIAL_TIMEOUT,
|
|
env=helper_env,
|
|
)
|
|
except subprocess.TimeoutExpired:
|
|
raise ConfigError("git credential lookup timed out") from None
|
|
except OSError:
|
|
raise ConfigError("cannot run git credential lookup") from None
|
|
|
|
if result.returncode != 0:
|
|
# stdout/stderr may contain credentials or helper-specific secret data.
|
|
raise ConfigError("git credential lookup failed or requires interaction")
|
|
|
|
fields: dict[str, str] = {}
|
|
for line in result.stdout.splitlines():
|
|
key, separator, value = line.partition("=")
|
|
if separator:
|
|
fields[key] = value
|
|
|
|
username = fields.get("username", "")
|
|
password = fields.get("password", "")
|
|
if not username or not password or ":" in username:
|
|
raise ConfigError("git credential lookup returned unusable credentials")
|
|
|
|
encoded = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii")
|
|
return ApiAuth(source="git credential", authorization=f"Basic {encoded}")
|
|
|
|
|
|
def resolve_auth(args: argparse.Namespace, target: RepoTarget) -> ApiAuth | None:
|
|
"""Resolve authentication once, with environment tokens taking precedence."""
|
|
token = _token()
|
|
if token:
|
|
_require_https(target)
|
|
return ApiAuth(source=TOKEN_ENV, authorization=f"token {token}")
|
|
if getattr(args, "use_git_credential", False):
|
|
return _git_credential_auth(target)
|
|
return None
|
|
|
|
|
|
class SameOriginRedirectHandler(urllib.request.HTTPRedirectHandler):
|
|
"""Follow redirects only while they stay on the configured Gitea origin."""
|
|
|
|
def __init__(self, target: RepoTarget) -> None:
|
|
super().__init__()
|
|
self._origin = target.origin
|
|
|
|
def redirect_request(
|
|
self,
|
|
req: urllib.request.Request,
|
|
fp: Any,
|
|
code: int,
|
|
msg: str,
|
|
headers: Any,
|
|
newurl: str,
|
|
) -> urllib.request.Request | None:
|
|
absolute_url = urllib.parse.urljoin(req.full_url, newurl)
|
|
try:
|
|
redirect_origin = _origin(absolute_url)
|
|
except ConfigError:
|
|
raise ConfigError("refusing an invalid Gitea redirect") from None
|
|
if redirect_origin != self._origin:
|
|
raise ConfigError("refusing a redirect outside the configured origin")
|
|
return super().redirect_request(req, fp, code, msg, headers, absolute_url)
|
|
|
|
|
|
class ApiClient:
|
|
"""Small, origin-bound client for read-only Gitea API requests."""
|
|
|
|
def __init__(
|
|
self,
|
|
target: RepoTarget,
|
|
*,
|
|
auth: ApiAuth | None = None,
|
|
opener: Any | None = None,
|
|
) -> None:
|
|
if auth:
|
|
_require_https(target)
|
|
self.target = target
|
|
self.auth = auth
|
|
self._opener = opener or urllib.request.build_opener(
|
|
SameOriginRedirectHandler(target)
|
|
)
|
|
|
|
def request(self, url: str, *, accept_json: bool = True) -> tuple[int, bytes, str]:
|
|
"""Perform an origin-bound GET and return status, body, content type."""
|
|
try:
|
|
request_origin = _origin(url)
|
|
except ConfigError:
|
|
raise ConfigError("refusing an invalid Gitea API URL") from None
|
|
if request_origin != self.target.origin:
|
|
raise ConfigError("refusing a request outside the configured origin")
|
|
|
|
headers = {"User-Agent": USER_AGENT}
|
|
if accept_json:
|
|
headers["Accept"] = "application/json"
|
|
if self.auth:
|
|
headers["Authorization"] = self.auth.authorization
|
|
|
|
request = urllib.request.Request(url, headers=headers, method="GET")
|
|
try:
|
|
with self._opener.open(request, timeout=REQUEST_TIMEOUT) as response:
|
|
body = response.read()
|
|
content_type = response.headers.get("Content-Type", "")
|
|
return response.status, body, content_type
|
|
except urllib.error.HTTPError as exc:
|
|
body = exc.read() if hasattr(exc, "read") else b""
|
|
content_type = exc.headers.get("Content-Type", "") if exc.headers else ""
|
|
return exc.code, body, content_type
|
|
except urllib.error.URLError as exc:
|
|
detail = str(exc.reason)
|
|
if self.auth:
|
|
authorization = self.auth.authorization
|
|
secret = authorization.partition(" ")[2]
|
|
detail = detail.replace(authorization, "<redacted>")
|
|
if secret:
|
|
detail = detail.replace(secret, "<redacted>")
|
|
raise ApiError(f"request to Gitea failed: {detail}") from None
|
|
|
|
def get_json(self, url: str) -> Any:
|
|
status, body, _ = self.request(url, accept_json=True)
|
|
if status != 200:
|
|
raise ApiError(_explain_status(status, self.target, self.auth))
|
|
try:
|
|
return json.loads(body.decode("utf-8"))
|
|
except (UnicodeError, json.JSONDecodeError) as exc:
|
|
raise ApiError(f"cannot parse JSON from {url}: {exc}") from None
|
|
|
|
|
|
def _explain_status(
|
|
status: int, target: RepoTarget, auth: ApiAuth | None = None
|
|
) -> str:
|
|
if status == 401:
|
|
if auth is None:
|
|
return (
|
|
f"authentication required (HTTP 401). Set {TOKEN_ENV} or rerun "
|
|
"with --use-git-credential; do not paste credentials into chat."
|
|
)
|
|
return (
|
|
f"authentication rejected (HTTP 401) for {auth.source}. Check that "
|
|
"the credential is valid for this Gitea instance and repository."
|
|
)
|
|
if status == 403:
|
|
if auth is None:
|
|
return (
|
|
f"permission denied (HTTP 403) without authentication. Set "
|
|
f"{TOKEN_ENV} or rerun with --use-git-credential."
|
|
)
|
|
return (
|
|
f"permission denied (HTTP 403) using {auth.source}. Grant repository "
|
|
"and Actions read permission to the selected credential."
|
|
)
|
|
if status == 404:
|
|
return (
|
|
"endpoint not found (HTTP 404). Check the repository coordinates and "
|
|
"Gitea version; older Gitea (1.21 and earlier) lacks the Actions "
|
|
"run/job/log API. Fall back to the web UI or pasted logs per SKILL.md."
|
|
)
|
|
return f"unexpected HTTP {status} from {target.base_url}"
|
|
|
|
|
|
def cmd_version(
|
|
args: argparse.Namespace,
|
|
target: RepoTarget,
|
|
client: ApiClient,
|
|
) -> int:
|
|
payload = client.get_json(f"{target.api_root}/version")
|
|
version = payload.get("version") if isinstance(payload, dict) else None
|
|
if not version:
|
|
raise ApiError("version endpoint returned no 'version' field")
|
|
if args.json:
|
|
print(json.dumps({"version": version}, ensure_ascii=False))
|
|
else:
|
|
print(f"gitea version: {version}")
|
|
print(f"instance: {target.base_url}")
|
|
return 0
|
|
|
|
|
|
def _as_list(payload: Any, key: str) -> list[dict[str, Any]]:
|
|
"""Gitea may wrap collections in an object or return a bare list."""
|
|
if isinstance(payload, list):
|
|
return [item for item in payload if isinstance(item, dict)]
|
|
if isinstance(payload, dict):
|
|
inner = payload.get(key)
|
|
if isinstance(inner, list):
|
|
return [item for item in inner if isinstance(item, dict)]
|
|
return []
|
|
|
|
|
|
def _outcome(item: dict[str, Any]) -> str:
|
|
"""Resolve the effective result of a run/job.
|
|
|
|
Gitea's Actions objects follow the GitHub-compatible two-field model: a
|
|
lifecycle `status` (queued/in_progress/completed) plus a terminal
|
|
`conclusion` (success/failure/cancelled/...). A completed job reports
|
|
`status=completed, conclusion=failure`, so a naive `status or conclusion`
|
|
would stop at "completed" and miss the failure. Prefer `conclusion`; fall
|
|
back to `status` only when there is no conclusion yet.
|
|
"""
|
|
conclusion = (item.get("conclusion") or "").strip()
|
|
status = (item.get("status") or "").strip()
|
|
return (conclusion or status).lower()
|
|
|
|
|
|
def _run_row(run: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": run.get("id"),
|
|
"name": run.get("name") or run.get("workflow_id") or "",
|
|
"outcome": _outcome(run),
|
|
"event": run.get("event") or "",
|
|
"branch": run.get("head_branch") or run.get("branch") or "",
|
|
"sha": (run.get("head_sha") or run.get("commit_sha") or "")[:12],
|
|
"url": run.get("html_url") or run.get("url") or "",
|
|
}
|
|
|
|
|
|
def cmd_runs(
|
|
args: argparse.Namespace,
|
|
target: RepoTarget,
|
|
client: ApiClient,
|
|
) -> int:
|
|
query: dict[str, str] = {}
|
|
if args.branch:
|
|
query["branch"] = args.branch
|
|
if args.event:
|
|
query["event"] = args.event
|
|
if args.status:
|
|
query["status"] = args.status
|
|
if args.sha:
|
|
query["head_sha"] = args.sha
|
|
if args.limit:
|
|
query["limit"] = str(args.limit)
|
|
suffix = "/actions/runs"
|
|
if query:
|
|
suffix += "?" + urllib.parse.urlencode(query)
|
|
payload = client.get_json(target.repo_path(suffix))
|
|
runs = [_run_row(run) for run in _as_list(payload, "workflow_runs")]
|
|
if args.sha:
|
|
runs = [row for row in runs if row["sha"].startswith(args.sha[:12])]
|
|
runs = runs[: args.limit] if args.limit else runs
|
|
|
|
if args.json:
|
|
print(json.dumps(runs, ensure_ascii=False, indent=2))
|
|
return 0
|
|
if not runs:
|
|
print("no matching workflow runs")
|
|
return 1
|
|
print(f"{len(runs)} run(s):")
|
|
for row in runs:
|
|
print(
|
|
f" run {row['id']} [{row['outcome']}] {row['name']} "
|
|
f"{row['event']} {row['branch']} {row['sha']}"
|
|
)
|
|
if row["url"]:
|
|
print(f" {row['url']}")
|
|
return 0
|
|
|
|
|
|
def _job_row(job: dict[str, Any], index: int) -> dict[str, Any]:
|
|
return {
|
|
"index": index,
|
|
"id": job.get("id"),
|
|
"name": job.get("name") or "",
|
|
"outcome": _outcome(job),
|
|
}
|
|
|
|
|
|
def _is_failed(outcome: str) -> bool:
|
|
return outcome.lower() in {"failure", "failed", "error"}
|
|
|
|
|
|
def cmd_jobs(
|
|
args: argparse.Namespace,
|
|
target: RepoTarget,
|
|
client: ApiClient,
|
|
) -> int:
|
|
payload = client.get_json(target.repo_path(f"/actions/runs/{args.run_id}/jobs"))
|
|
jobs = [_job_row(job, index) for index, job in enumerate(_as_list(payload, "jobs"))]
|
|
failed = [job for job in jobs if _is_failed(job["outcome"])]
|
|
|
|
if args.json:
|
|
print(
|
|
json.dumps(
|
|
{"jobs": jobs, "failed_job_ids": [j["id"] for j in failed]},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
if not jobs:
|
|
print(f"run {args.run_id} has no jobs (or endpoint unavailable)")
|
|
return 1
|
|
print(f"run {args.run_id} jobs:")
|
|
for job in jobs:
|
|
marker = " <== FAILED" if _is_failed(job["outcome"]) else ""
|
|
print(
|
|
f" job {job['id']} index={job['index']} "
|
|
f"[{job['outcome']}] {job['name']}{marker}"
|
|
)
|
|
if failed:
|
|
ids = ", ".join(str(job["id"]) for job in failed)
|
|
print(f"failed job id(s): {ids}")
|
|
print("fetch a failed job's log with: logs <job-id>")
|
|
else:
|
|
print("no failed jobs on this run")
|
|
return 0
|
|
|
|
|
|
def cmd_logs(
|
|
args: argparse.Namespace,
|
|
target: RepoTarget,
|
|
client: ApiClient,
|
|
) -> int:
|
|
status, body, _ = client.request(
|
|
target.repo_path(f"/actions/jobs/{args.job_id}/logs"),
|
|
accept_json=False,
|
|
)
|
|
if status != 200:
|
|
raise ApiError(_explain_status(status, target, client.auth))
|
|
|
|
text = body.decode("utf-8", errors="replace")
|
|
lines = text.splitlines()
|
|
|
|
if args.out:
|
|
out_path = Path(args.out).expanduser()
|
|
out_path.write_text(text, encoding="utf-8")
|
|
else:
|
|
handle = tempfile.NamedTemporaryFile(
|
|
prefix=f"gitea-job-{args.job_id}-",
|
|
suffix=".log",
|
|
delete=False,
|
|
mode="w",
|
|
encoding="utf-8",
|
|
)
|
|
handle.write(text)
|
|
handle.close()
|
|
out_path = Path(handle.name)
|
|
|
|
tail = args.tail if args.tail is not None else DEFAULT_LOG_TAIL
|
|
tail_lines = lines[-tail:] if tail > 0 else lines
|
|
|
|
if args.json:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"job_id": args.job_id,
|
|
"log_path": str(out_path),
|
|
"total_lines": len(lines),
|
|
"tail": tail_lines,
|
|
},
|
|
ensure_ascii=False,
|
|
indent=2,
|
|
)
|
|
)
|
|
return 0
|
|
|
|
print(f"job {args.job_id} log saved to: {out_path}")
|
|
print(f"total lines: {len(lines)} (showing last {len(tail_lines)})")
|
|
print("-" * 60)
|
|
for line in tail_lines:
|
|
print(line)
|
|
return 0
|
|
|
|
|
|
def _build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(
|
|
description="Collect Gitea Actions CI evidence (version/runs/jobs/logs).",
|
|
)
|
|
parser.add_argument("--remote", default=DEFAULT_REMOTE, help="git remote name")
|
|
parser.add_argument("--base-url", help="Gitea base URL override")
|
|
parser.add_argument("--owner", help="repository owner override")
|
|
parser.add_argument("--repo", help="repository name override")
|
|
parser.add_argument(
|
|
"--use-git-credential",
|
|
action="store_true",
|
|
help=(
|
|
"explicitly use repository-scoped git credentials over HTTPS "
|
|
f"when {TOKEN_ENV} is unset"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--json", action="store_true", help="emit machine-readable JSON"
|
|
)
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("version", help="probe /api/v1/version")
|
|
|
|
runs = sub.add_parser("runs", help="list workflow runs")
|
|
runs.add_argument("--branch")
|
|
runs.add_argument("--event")
|
|
runs.add_argument("--status", default="failure")
|
|
runs.add_argument("--sha")
|
|
runs.add_argument("--limit", type=int, default=20)
|
|
|
|
jobs = sub.add_parser("jobs", help="list jobs of a run, flag failed ones")
|
|
jobs.add_argument("run_id")
|
|
|
|
logs = sub.add_parser("logs", help="download a job log, print its tail")
|
|
logs.add_argument("job_id")
|
|
logs.add_argument("--out", help="write full log here instead of a temp file")
|
|
logs.add_argument(
|
|
"--tail", type=int, help=f"tail lines (default {DEFAULT_LOG_TAIL})"
|
|
)
|
|
|
|
return parser
|
|
|
|
|
|
_COMMANDS = {
|
|
"version": cmd_version,
|
|
"runs": cmd_runs,
|
|
"jobs": cmd_jobs,
|
|
"logs": cmd_logs,
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
if hasattr(sys.stdout, "reconfigure"):
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
args = _build_parser().parse_args(argv)
|
|
try:
|
|
target = resolve_target(args)
|
|
auth = resolve_auth(args, target)
|
|
client = ApiClient(target, auth=auth)
|
|
return _COMMANDS[args.command](args, target, client)
|
|
except ConfigError as exc:
|
|
_eprint(f"ERROR: {exc}")
|
|
return 2
|
|
except ApiError as exc:
|
|
_eprint(f"ERROR: {exc}")
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|