#!/usr/bin/env python3
"""Parse public GitHub Actions logs into a compact CI-failure report.

The parser intentionally uses GitHub's unauthenticated public REST API by
default.  A ``GITHUB_TOKEN`` may be supplied for a private or rate-limited
repository, but the tool never prints or stores that token.
"""

from __future__ import annotations

import argparse
import io
import json
import os
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
import zipfile
from pathlib import Path
from typing import Any, Mapping, Sequence


class GitHubActionsError(RuntimeError):
    """Raised when the run URL or GitHub log response is unusable."""


def parse_run_url(run_url: str) -> tuple[str, str, str]:
    """Extract ``owner``, ``repository`` and ``run_id`` from a run URL."""

    parsed = urllib.parse.urlparse(run_url.strip())
    if parsed.scheme not in {"http", "https"} or parsed.netloc.lower() != "github.com":
        raise GitHubActionsError("run_url must be an https://github.com/.../actions/runs/<id> URL")
    parts = [part for part in parsed.path.split("/") if part]
    if len(parts) < 5 or parts[2] != "actions" or parts[3] != "runs" or not parts[4].isdigit():
        raise GitHubActionsError("run_url must contain /<owner>/<repo>/actions/runs/<numeric-id>")
    return parts[0], parts[1], parts[4]


def github_request(url: str, token: str | None = None) -> bytes:
    """Fetch bytes from GitHub while keeping authentication details private."""

    headers = {
        "Accept": "application/vnd.github+json, application/zip",
        "User-Agent": "Proofwork-GitHub-Actions-Parser/1.0",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    if token:
        headers["Authorization"] = f"Bearer {token}"
    request = urllib.request.Request(url, headers=headers, method="GET")
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return response.read()
    except urllib.error.HTTPError as exc:
        if exc.code == 404:
            raise GitHubActionsError("GitHub could not find the run or its logs; it may be private") from exc
        if exc.code == 403:
            raise GitHubActionsError("GitHub denied the request or its rate limit was reached") from exc
        raise GitHubActionsError(f"GitHub returned HTTP {exc.code} while fetching logs") from exc
    except urllib.error.URLError as exc:
        raise GitHubActionsError("GitHub log request failed") from exc


def fetch_run_logs(run_url: str, token: str | None = None) -> bytes:
    """Download the ZIP log bundle for a GitHub Actions run URL."""

    owner, repository, run_id = parse_run_url(run_url)
    api_url = (
        f"https://api.github.com/repos/{urllib.parse.quote(owner)}/"
        f"{urllib.parse.quote(repository)}/actions/runs/{run_id}/logs"
    )
    return github_request(api_url, token)


def extract_log_files(payload: bytes) -> dict[str, str]:
    """Decode a GitHub Actions log ZIP into a filename-to-text mapping."""

    try:
        archive = zipfile.ZipFile(io.BytesIO(payload))
    except (zipfile.BadZipFile, OSError) as exc:
        raise GitHubActionsError("GitHub returned logs that were not a valid ZIP archive") from exc
    files: dict[str, str] = {}
    with archive:
        for member in archive.infolist():
            if member.is_dir():
                continue
            try:
                files[member.filename] = archive.read(member).decode("utf-8", errors="replace")
            except (KeyError, OSError, RuntimeError) as exc:
                raise GitHubActionsError("could not read a file from the GitHub log archive") from exc
    if not files:
        raise GitHubActionsError("GitHub returned an empty log archive")
    return files


def _clean_log_line(line: str) -> str:
    """Remove GitHub workflow annotations and terminal whitespace."""

    cleaned = re.sub(r"^\s*##\[(?:error|warning|debug)\]\s*", "", line)
    return cleaned.strip()


def _failure_type(text: str) -> tuple[str, str]:
    """Return a stable failure type and suggested remediation category."""

    lowered = text.lower()
    if re.search(r"\b(?:eslint|pylint|flake8|ruff|stylelint|lint)\b", lowered) or re.search(
        r"\b(?:no-unused-vars|undefined-variable|invalid-name|e\d{3,4}|w\d{3,4})\b", lowered
    ):
        return "lint", "lint"
    if re.search(
        r"(?:error\s*ts\d+|typescript|\btsc\b|compilation failed|compile failed|module not found|webpack|build failed)",
        lowered,
    ):
        return "build", "build"
    if re.search(
        r"(?:pytest|jest|test suites?:|tests?:|assertionerror|failed tests?|\bfailed\b.*\btest)",
        lowered,
    ):
        return "test", "tests"
    return "workflow", "workflow"


def _candidate(line: str) -> bool:
    """Identify a line likely to describe a failure rather than normal output."""

    lowered = line.lower()
    return (
        "##[error]" in lowered
        or bool(re.search(r"\b(?:failed|failure|error|exception|assertionerror)\b", lowered))
        or bool(re.search(r"\b(?:test suites?:|tests?:)\s+.*\bfailed\b", lowered))
    )


def _first_error(lines: Sequence[str]) -> tuple[int, str] | None:
    """Return the first useful failure line, skipping generic shell noise."""

    for index, raw in enumerate(lines):
        cleaned = _clean_log_line(raw)
        if not cleaned or not _candidate(raw):
            continue
        if cleaned.lower() in {"error", "failure", "failed"} and index + 1 < len(lines):
            next_line = _clean_log_line(lines[index + 1])
            if next_line:
                return index + 1, next_line
        return index, cleaned
    return None


def _step_for(lines: Sequence[str], index: int, filename: str) -> str:
    """Find the nearest GitHub Actions group or run command for an error."""

    for candidate in reversed(lines[: index + 1]):
        match = re.search(r"##\[group\](?:Run\s+)?(.+)$", candidate)
        if match:
            return _clean_log_line(match.group(1))[:200]
    for candidate in reversed(lines[: index + 1]):
        if candidate.strip().startswith("Run "):
            return candidate.strip()[:200]
    return Path(filename).stem[:200] or "unknown step"


def _stack_excerpt(lines: Sequence[str], index: int) -> str:
    """Collect a bounded stack/error excerpt around the first failure."""

    start = max(0, index - 5)
    end = min(len(lines), index + 12)
    excerpt = [_clean_log_line(line) for line in lines[start:end]]
    excerpt = [line for line in excerpt if line]
    return "\n".join(excerpt)[:4000]


def parse_logs(log_files: Mapping[str, str]) -> dict[str, Any]:
    """Parse log files into a JSON-serializable failure summary."""

    best: tuple[int, str, str, int, str] | None = None
    all_text_parts: list[str] = []
    for filename in sorted(log_files):
        text = log_files[filename]
        all_text_parts.append(text)
        lines = text.splitlines()
        found = _first_error(lines)
        if found is None:
            continue
        index, message = found
        failure_type, _ = _failure_type(message + "\n" + text)
        priority = {"lint": 3, "build": 2, "test": 2, "workflow": 1}[failure_type]
        score = priority * 100000 - index
        if best is None or score > best[0]:
            best = (score, filename, message, index, failure_type)

    combined = "\n".join(all_text_parts)
    if best is None:
        failure_type, suggested = _failure_type(combined)
        return {
            "failing_step": None,
            "error_message": None,
            "stack_trace": None,
            "suggested_fix_category": suggested,
            "failure_type": failure_type,
            "log_files": sorted(log_files),
            "parse_status": "no_failure_marker_found",
        }

    _, filename, message, index, failure_type = best
    lines = log_files[filename].splitlines()
    _, suggested = _failure_type(message + "\n" + combined)
    return {
        "failing_step": _step_for(lines, index, filename),
        "error_message": message[:1000],
        "stack_trace": _stack_excerpt(lines, index),
        "suggested_fix_category": suggested,
        "failure_type": failure_type,
        "log_files": sorted(log_files),
        "parse_status": "failure_found",
    }


def build_report(run_url: str, token: str | None = None) -> dict[str, Any]:
    """Fetch and parse a run's logs, returning the final report object."""

    payload = fetch_run_logs(run_url, token)
    report = parse_logs(extract_log_files(payload))
    report["run_url"] = run_url
    return report


def main(argv: Sequence[str] | None = None) -> int:
    """Run the command-line interface and print one JSON report."""

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("run_url", help="public GitHub Actions run URL")
    parser.add_argument(
        "--logs-zip",
        type=Path,
        help="parse a previously downloaded logs ZIP instead of making an API request",
    )
    args = parser.parse_args(argv)
    try:
        if args.logs_zip is not None:
            report = parse_logs(extract_log_files(args.logs_zip.read_bytes()))
            report["run_url"] = args.run_url
        else:
            report = build_report(args.run_url, os.environ.get("GITHUB_TOKEN"))
    except (GitHubActionsError, OSError) as exc:
        print(json.dumps({"error": str(exc)}), file=sys.stderr)
        return 2
    print(json.dumps(report, indent=2, sort_keys=True))
    return 0


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