devx/src/devx/tools/pr_logs.py

#!/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 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.tokens import get_developer_token
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."""
    try:
        token = get_developer_token()
    except click.ClickException:
        raise click.ClickException(_("CI_GITEA_TOKEN is not set.")) from None

    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