DEVX-91: feat: add pr_status, pr_logs, pr_label tools

src/devx/tools/pr_logs.py

@@ -0,0 +1,187 @@
+#!/usr/bin/env python3
+"""Fetch logs for failed CI jobs on a pull request.
+
+Lists CI jobs for the latest workflow run of a PR's branch, then fetches
+and prints the logs of any failed jobs. Useful for diagnosing CI failures
+without navigating the web UI.
+
+Usage::
+
+    # Show failed job logs for PR #42
+    python -m devx.tools.pr_logs --pr 42
+
+    # Show failed job logs for current branch's PR
+    python -m devx.tools.pr_logs
+
+    # Show logs for a specific job (by name)
+    python -m devx.tools.pr_logs --pr 42 --job quality
+
+    # Show last N lines of each failed job's logs
+    python -m devx.tools.pr_logs --pr 42 --tail 50
+
+The repository is auto-detected from ``DEVX_REPO_OWNER`` /
+``DEVX_REPO_NAME`` or ``GITHUB_REPOSITORY`` environment variables.
+"""
+
+from __future__ import annotations
+
+import os
+
+import click
+from dotenv import load_dotenv
+
+from devx.api_clients import APIError, GiteaClient
+from devx.config import GITEA_API_URL, REPO_OWNER
+from devx.i18n import _
+from devx.tools.create_pr import get_repo_name
+from devx.tools.pr_status import _get_current_branch_pr
+
+load_dotenv()
+
+
+def _get_pr_sha(client: GiteaClient, pr_number: int) -> str:
+    """Fetch the head SHA of a PR."""
+    pr = client.get_pr(pr_number)
+    return pr.get("head", {}).get("sha", "")
+
+
+def _find_latest_run_by_sha(client: GiteaClient, sha: str) -> dict | None:
+    """Find the latest workflow run for a commit SHA.
+
+    Gitea Actions API doesn't set head_branch for pull_request events,
+    so we filter by head_sha instead.
+    """
+    data = client.list_action_runs(limit=50)
+    for run in data.get("workflow_runs", []):
+        if run.get("head_sha", "").startswith(sha):
+            return run
+    return None
+
+
+def _find_failed_jobs(jobs: list[dict]) -> list[dict]:
+    """Return jobs with conclusion 'failure'."""
+    return [j for j in jobs if j.get("conclusion") == "failure"]
+
+
+def _find_job_by_name(jobs: list[dict], name: str) -> dict | None:
+    """Find a job by name (case-insensitive partial match)."""
+    name_lower = name.lower()
+    for j in jobs:
+        if name_lower in j.get("name", "").lower():
+            return j
+    return None
+
+
+def _print_job_summary(jobs: list[dict]) -> None:
+    """Print a summary table of all jobs and their status."""
+    for j in jobs:
+        name = j.get("name", "?")
+        conclusion = j.get("conclusion", "pending")
+        status = j.get("status", "?")
+        symbol = "[FAIL]" if conclusion == "failure" else "[OK]" if conclusion == "success" else f"[{conclusion}]"
+        click.echo(f"  {symbol} {name} (status: {status}, conclusion: {conclusion})")
+
+
+def _print_failed_steps(job: dict) -> list[int]:
+    """Print failed steps for a job. Returns list of failed step numbers."""
+    failed_steps = []
+    for step in job.get("steps", []):
+        if step.get("conclusion") == "failure":
+            name = step.get("name", "?")
+            num = step.get("number", "?")
+            click.echo(f"    FAILED step #{num}: {name}")
+            failed_steps.append(num)
+    return failed_steps
+
+
+def _print_logs(client: GiteaClient, job_id: int, tail: int = 0) -> None:
+    """Fetch and print logs for a job. If tail > 0, print only last N lines."""
+    try:
+        logs = client.get_action_job_logs(job_id)
+    except APIError as e:
+        click.echo(_("    Could not fetch logs: {error}", error=str(e)))
+        return
+
+    if tail > 0:
+        lines = logs.strip().split("\n")
+        if len(lines) > tail:
+            click.echo(f"    ... (showing last {tail} of {len(lines)} lines)")
+            logs = "\n".join(lines[-tail:])
+
+    for line in logs.split("\n"):
+        click.echo(f"    {line}")
+
+
+@click.command()
+@click.option("--pr", "pr_number", type=int, default=None, help="PR number (default: auto-detect from current branch).")
+@click.option("--job", default=None, help="Job name to show logs for (partial match, case-insensitive).")
+@click.option("--tail", type=int, default=80, show_default=True, help="Show last N lines of logs (0 = all).")
+@click.option("--owner", default=None, help="Repository owner (default: DEVX_REPO_OWNER).")
+@click.option("--repo", default=None, help="Repository name (default: DEVX_REPO_NAME or GITHUB_REPOSITORY).")
+def cli(
+    pr_number: int | None,
+    job: str | None,
+    tail: int,
+    owner: str | None,
+    repo: str | None,
+) -> None:
+    """Fetch logs for failed CI jobs on a pull request."""
+    token = os.environ.get("CI_GITEA_TOKEN", "")
+    if not token:
+        raise click.ClickException(_("CI_GITEA_TOKEN is not set."))
+
+    repo_owner = owner or REPO_OWNER
+    if not repo_owner:
+        raise click.ClickException(_("Repository owner not set. Use --owner or DEVX_REPO_OWNER env var."))
+    repo_name = repo or get_repo_name()
+
+    client = GiteaClient(GITEA_API_URL, token, repo_owner, repo_name)
+
+    if pr_number is None:
+        pr_number = _get_current_branch_pr(client)
+        click.echo(_("Fetching logs for PR #{pr_number}...", pr_number=pr_number))
+
+    sha = _get_pr_sha(client, pr_number)
+    if not sha:
+        raise click.ClickException(_("Could not determine head SHA for PR #{pr_number}.", pr_number=pr_number))
+
+    run = _find_latest_run_by_sha(client, sha)
+    if not run:
+        raise click.ClickException(_("No workflow runs found for SHA {sha}.", sha=sha[:8]))
+
+    run_id = run.get("id", 0)
+    run_status = run.get("status", "?")
+    click.echo(_("Latest run: #{run_id} (status: {status})", run_id=run_id, status=run_status))
+    click.echo("")
+
+    jobs = client.get_action_run_jobs(run_id)
+    if not jobs:
+        click.echo(_("No jobs found for run #{run_id}.", run_id=run_id))
+        return
+
+    _print_job_summary(jobs)
+    click.echo("")
+
+    if job:
+        target = _find_job_by_name(jobs, job)
+        if not target:
+            raise click.ClickException(_("No job matching '{job}' found.", job=job))
+        click.echo(f"Logs for job '{target.get('name', '?')}' (id={target.get('id')}):")
+        _print_failed_steps(target)
+        click.echo("")
+        _print_logs(client, target["id"], tail)
+    else:
+        failed = _find_failed_jobs(jobs)
+        if not failed:
+            click.echo(_("No failed jobs."))
+            return
+        for fj in failed:
+            click.echo(f"Logs for failed job '{fj.get('name', '?')}' (id={fj.get('id')}):")
+            _print_failed_steps(fj)
+            click.echo("")
+            _print_logs(client, fj["id"], tail)
+            click.echo("")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()  # pragma: no cover