gh_log_parser.py

"""Parse GitHub Actions failures into compact, machine-readable triage data.

The module intentionally uses only the Python standard library.  The HTTP
client is small and injectable, which keeps the parser deterministic in tests
and makes it possible to run offline against a saved log file.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen


JSONDict = Dict[str, Any]
_RUN_URL = re.compile(
    r"^https?://github\.com/(?P<owner>[^/]+)/(?P<repo>[^/]+)"
    r"/actions/runs/(?P<run_id>\d+)(?:/.*)?/?$"
)
_TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T[^ ]+\s+")
_EXCEPTION = re.compile(
    r"(?:^|\s)(?:[A-Za-z_][\w.]*Error|[A-Za-z_][\w.]*Exception):?\s*(.*)$"
)


class GitHubAPIError(RuntimeError):
    """Raised when the GitHub REST API cannot provide a requested resource."""


@dataclass(frozen=True)
class RepositoryRun:
    """The repository and run identifier extracted from a public Actions URL."""

    owner: str
    repo: str
    run_id: str

    @property
    def repository(self) -> str:
        """Return the canonical ``owner/repository`` string."""
        return f"{self.owner}/{self.repo}"


class GitHubClient:
    """Minimal GitHub API client with an injectable URL opener."""

    def __init__(
        self,
        token: Optional[str] = None,
        opener: Optional[Callable[..., Any]] = None,
        api_base: str = "https://api.github.com",
    ) -> None:
        self.token = token
        self.opener = opener or urlopen
        self.api_base = api_base.rstrip("/")

    def _request(self, path: str) -> Tuple[bytes, str]:
        url = path if path.startswith("http") else f"{self.api_base}{path}"
        headers = {
            "Accept": "application/vnd.github+json",
            "User-Agent": "workprotocol-gh-log-parser/1.0",
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        request = Request(url, headers=headers)
        try:
            with self.opener(request, timeout=20) as response:
                return response.read(), response.headers.get_content_type()
        except HTTPError as exc:
            raise GitHubAPIError(f"GitHub API returned HTTP {exc.code} for {path}") from exc
        except URLError as exc:
            raise GitHubAPIError(f"Could not reach GitHub API: {exc.reason}") from exc

    def get_json(self, path: str) -> JSONDict:
        """Fetch a JSON object from GitHub."""
        payload, _ = self._request(path)
        try:
            value = json.loads(payload.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise GitHubAPIError(f"GitHub returned invalid JSON for {path}") from exc
        if not isinstance(value, dict):
            raise GitHubAPIError(f"GitHub returned a non-object response for {path}")
        return value

    def get_text(self, path: str) -> str:
        """Fetch a text response, preserving replacement characters safely."""
        payload, _ = self._request(path)
        return payload.decode("utf-8", errors="replace")


def parse_run_url(run_url: str) -> RepositoryRun:
    """Extract ``owner``, ``repo`` and ``run_id`` from a GitHub Actions URL."""
    match = _RUN_URL.match(run_url.strip())
    if not match:
        raise ValueError(
            "run URL must look like https://github.com/OWNER/REPO/actions/runs/123"
        )
    return RepositoryRun(**match.groupdict())


def _clean_line(line: str) -> str:
    """Remove GitHub's timestamp and command annotations from a log line."""
    cleaned = _TIMESTAMP.sub("", line.rstrip())
    return re.sub(r"^##\[[^]]+\]\s*", "", cleaned).strip()


def _step_name(line: str) -> Optional[str]:
    """Return a step title from a group or shell ``Run`` line."""
    group = re.search(r"##\[group\](.+)$", line)
    if group:
        return group.group(1).strip()
    run = re.search(r"\bRun\s+(.+)$", _clean_line(line))
    return run.group(1).strip() if run else None


def classify_failure(text: str) -> Tuple[str, str]:
    """Classify a log as test, build, lint, or runtime failure."""
    lowered = text.lower()
    if re.search(r"pytest|jest|test(?:s)? failed|assertionerror|assertion failed", lowered):
        return "test", "test-failure"
    if re.search(
        r"typescript|\btsc\b|compilation failed|compile error|webpack|vite build", lowered
    ):
        return "build", "build-configuration"
    if re.search(r"eslint|pylint|flake8|ruff|lint(?:ing)? error|lint failed", lowered):
        return "lint", "linting-rules"
    return "runtime", "runtime-error"


def _first_error(lines: Sequence[str]) -> str:
    """Select the most useful concise error message from log lines."""
    for line in lines:
        clean = _clean_line(line)
        if "##[error]" in line:
            message = clean.removeprefix("Error: ").strip()
            if message:
                return message[:500]
    for line in lines:
        clean = _clean_line(line)
        if _EXCEPTION.search(clean) or re.search(r"\b(?:FAIL|failed|fatal):", clean, re.I):
            return clean[:500]
    return "No explicit error line found; inspect the captured stack trace."


def _stack_trace(lines: Sequence[str]) -> str:
    """Extract a bounded Python/JavaScript-style stack trace."""
    start = next(
        (index for index, line in enumerate(lines) if "Traceback (most recent call last):" in line),
        None,
    )
    if start is not None:
        selected: List[str] = []
        for line in lines[start : start + 18]:
            clean = _clean_line(line)
            if clean or not selected:
                selected.append(clean)
            if len(selected) >= 12:
                break
        return "\n".join(selected).strip()

    js_lines = [_clean_line(line) for line in lines if re.search(r"\bat\s+.+\(.+\)", line)]
    return "\n".join(js_lines[:12]).strip()


def analyze_log(log_text: str, job_name: str = "unknown") -> JSONDict:
    """Analyze one Actions job log without network access."""
    lines = log_text.splitlines()
    current_step = job_name
    observed_steps: List[str] = []
    for line in lines:
        step = _step_name(line)
        if step:
            current_step = step
            if step not in observed_steps:
                observed_steps.append(step)
    error_message = _first_error(lines)
    stack_trace = _stack_trace(lines)
    combined = "\n".join(lines)
    failure_type, suggested = classify_failure(combined)
    return {
        "job_name": job_name,
        "failing_step": current_step,
        "error_message": error_message,
        "stack_trace": stack_trace,
        "failure_type": failure_type,
        "suggested_fix_category": suggested,
        "steps_seen": observed_steps,
    }


class ActionsLogParser:  # pylint: disable=too-few-public-methods
    """Fetch and analyze the first failed job in a GitHub Actions run."""

    def __init__(self, client: GitHubClient) -> None:
        self.client = client

    def analyze(self, run_url: str) -> JSONDict:
        """Return a structured report for ``run_url``."""
        run = parse_run_url(run_url)
        jobs_path = f"/repos/{run.repository}/actions/runs/{run.run_id}/jobs?per_page=100"
        jobs_payload = self.client.get_json(jobs_path)
        jobs = jobs_payload.get("jobs", [])
        failed = [job for job in jobs if job.get("conclusion") == "failure"]
        if not failed:
            return {
                "run_url": run_url,
                "repository": run.repository,
                "run_id": run.run_id,
                "job_name": None,
                "failing_step": None,
                "error_message": "No failed job was reported for this run.",
                "stack_trace": "",
                "failure_type": "none",
                "suggested_fix_category": "none",
                "jobs_examined": len(jobs),
            }

        job = failed[0]
        job_id = job.get("id")
        if not job_id:
            raise GitHubAPIError("GitHub returned a failed job without an id")
        log_text = self.client.get_text(f"/repos/{run.repository}/actions/jobs/{job_id}/logs")
        report = analyze_log(log_text, str(job.get("name", "failed job")))
        report.update(
            {
                "run_url": run_url,
                "repository": run.repository,
                "run_id": run.run_id,
                "jobs_examined": len(jobs),
            }
        )
        return report


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("run_url", nargs="?", help="GitHub Actions run URL")
    parser.add_argument("--log-file", help="Analyze a saved job log instead of calling GitHub")
    parser.add_argument("--job-name", default="offline log", help="Job name for --log-file")
    parser.add_argument("--token", default=os.environ.get("GITHUB_TOKEN"), help=argparse.SUPPRESS)
    return parser


def main(argv: Optional[Sequence[str]] = None) -> int:
    """Run the command-line interface and return a process exit code."""
    args = _build_parser().parse_args(argv)
    try:
        if args.log_file:
            with open(args.log_file, "r", encoding="utf-8") as handle:
                report = analyze_log(handle.read(), args.job_name)
        elif args.run_url:
            report = ActionsLogParser(GitHubClient(token=args.token)).analyze(args.run_url)
        else:
            raise ValueError("provide a run URL or --log-file")
    except (OSError, ValueError, GitHubAPIError) as exc:
        print(json.dumps({"error": str(exc)}, ensure_ascii=False), file=sys.stderr)
        return 2
    print(json.dumps(report, indent=2, ensure_ascii=False, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())