penpot/scripts/gh.py

#!/usr/bin/env python3
"""
gh.py — Multi-purpose CLI helper for penpot/penpot GitHub operations.

Uses GitHub GraphQL and REST APIs via the authenticated ``gh`` CLI.

Subcommands:
  issues      List issues in a milestone (or unassigned with milestone=none)
  prs         Fetch details for one or more PRs (by number or milestone)
  advisories  List or inspect GitHub security advisories
  link-issue  Explicitly link a GitHub issue to a pull request

Usage:
  python3 scripts/gh.py issues <milestone-title>            (default: state=closed)
  python3 scripts/gh.py issues "2.16.0" --state all
  python3 scripts/gh.py issues "2.16.0" --exclude "release blocker,no changelog"
  python3 scripts/gh.py issues "2.16.0" --label "bug"       (include only issues with label)
  python3 scripts/gh.py issues "2.16.0" --label "bug,regression" --exclude "no changelog"
  python3 scripts/gh.py issues "2.16.0" --compare CHANGES.md
  python3 scripts/gh.py issues none                         (issues with no milestone)
  python3 scripts/gh.py issues none --label "enhancement"
  python3 scripts/gh.py issues none --state open
  python3 scripts/gh.py prs 9179 9204 9311
  python3 scripts/gh.py prs --file prs.txt
  cat prs.txt | python3 scripts/gh.py prs --stdin
  python3 scripts/gh.py prs --milestone "2.16.0"            (default: state=merged)
  python3 scripts/gh.py prs --milestone "2.16.0" --state all
  python3 scripts/gh.py advisories                          (list all advisories)
  python3 scripts/gh.py advisories --severity critical      (filter by severity)
  python3 scripts/gh.py advisories GHSA-xvj6-fh9w-gjw7     (single advisory detail)
  python3 scripts/gh.py link-issue 11235 11243

Prerequisites:
  - gh CLI authenticated (gh auth status)
  - Python 3.8+
"""

import argparse
import json
import re
import subprocess
import sys
from typing import Any


REPO = "penpot/penpot"
OWNER = "penpot"
REPO_NAME = "penpot"


# ─────────────────────────────────────────────
#  Shared helpers
# ─────────────────────────────────────────────


def run_gh_graphql(query: str, variables: dict) -> Any:
    """Run a GraphQL query via ``gh api graphql --input -``."""
    payload = json.dumps({"query": query, "variables": variables})
    cmd = ["gh", "api", "graphql", "--input", "-"]
    result = subprocess.run(cmd, input=payload, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"gh error: {result.stderr}", file=sys.stderr)
        sys.exit(1)
    body = json.loads(result.stdout)
    if "errors" in body:
        for err in body["errors"]:
            print(f"GraphQL error: {err.get('message')}", file=sys.stderr)
        sys.exit(1)
    return body["data"]


def run_gh_rest(path: str) -> Any:
    """Run a REST API call via ``gh api``."""
    cmd = ["gh", "api", path]
    result = subprocess.run(cmd, capture_output=True, text=True)
    if result.returncode != 0:
        print(f"gh error: {result.stderr}", file=sys.stderr)
        sys.exit(1)
    return json.loads(result.stdout)


# ─────────────────────────────────────────────
#  Subcommand: link-issue
# ─────────────────────────────────────────────

GQL_LINK_TARGETS_QUERY = """\
query($owner: String!, $repo: String!, $issueNumber: Int!, $prNumber: Int!) {
  repository(owner: $owner, name: $repo) {
    issue(number: $issueNumber) { id number }
    pullRequest(number: $prNumber) { id number }
  }
}
"""

GQL_ADD_CLOSE_ISSUE_REFERENCES = """\
mutation($issueId: ID!, $pullRequestIds: [ID!]!) {
  addCloseIssueReferences(input: {issueId: $issueId, pullRequestIds: $pullRequestIds}) {
    issue { id number }
  }
}
"""

GQL_VERIFY_ISSUE_LINK_QUERY = """\
query($owner: String!, $repo: String!, $issueNumber: Int!, $prNumber: Int!) {
  repository(owner: $owner, name: $repo) {
    issue(number: $issueNumber) {
      number
      state
      closedByPullRequestsReferences(
        includeClosedPrs: true
        userLinkedOnly: true
        first: 100
      ) {
        nodes { number state url }
      }
    }
    pullRequest(number: $prNumber) {
      number
      state
      closingIssuesReferences(first: 100) {
        nodes { number state url }
      }
    }
  }
}
"""


def link_issue_to_pr(issue_number: int, pr_number: int) -> dict:
    """Add and verify an explicit GitHub issue-to-PR link."""
    if issue_number <= 0 or pr_number <= 0:
        raise ValueError("issue and pull request numbers must be positive")

    variables = {
        "owner": OWNER,
        "repo": REPO_NAME,
        "issueNumber": issue_number,
        "prNumber": pr_number,
    }
    target_data = run_gh_graphql(GQL_LINK_TARGETS_QUERY, variables)
    repository = target_data.get("repository") or {}
    issue = repository.get("issue") or {}
    pull_request = repository.get("pullRequest") or {}
    if not issue.get("id"):
        raise RuntimeError(f"issue #{issue_number} was not found in {REPO}")
    if not pull_request.get("id"):
        raise RuntimeError(f"pull request #{pr_number} was not found in {REPO}")

    mutation_data = run_gh_graphql(
        GQL_ADD_CLOSE_ISSUE_REFERENCES,
        {
            "issueId": issue["id"],
            "pullRequestIds": [pull_request["id"]],
        },
    )
    mutation_result = mutation_data.get("addCloseIssueReferences") or {}
    linked_issue = mutation_result.get("issue") or {}
    if linked_issue.get("number") != issue_number:
        raise RuntimeError(f"GitHub did not link issue #{issue_number}")

    verification_data = run_gh_graphql(GQL_VERIFY_ISSUE_LINK_QUERY, variables)
    repository = verification_data.get("repository") or {}
    issue = repository.get("issue") or {}
    pull_request = repository.get("pullRequest") or {}
    if not issue or not pull_request:
        raise RuntimeError("GitHub did not return both link targets during verification")

    issue_links = [
        node
        for node in issue["closedByPullRequestsReferences"]["nodes"]
        if node.get("number") == pr_number
    ]
    pr_links = [
        node
        for node in pull_request["closingIssuesReferences"]["nodes"]
        if node.get("number") == issue_number
    ]
    if not issue_links or not pr_links:
        raise RuntimeError(
            f"issue #{issue_number} and pull request #{pr_number} are not linked"
        )

    return {
        "linked": True,
        "issue": {
            "number": issue["number"],
            "state": issue["state"],
            "linked_pull_requests": issue_links,
        },
        "pull_request": {
            "number": pull_request["number"],
            "state": pull_request["state"],
            "linked_issues": pr_links,
        },
    }


def cmd_link_issue(args: argparse.Namespace) -> None:
    """Handle the ``link-issue`` subcommand."""
    print(
        f"Linking issue #{args.issue_number} to pull request #{args.pr_number}...",
        file=sys.stderr,
    )
    try:
        result = link_issue_to_pr(args.issue_number, args.pr_number)
    except (ValueError, RuntimeError) as error:
        print(f"ERROR: {error}", file=sys.stderr)
        sys.exit(1)

    print(
        f"Verified issue #{args.issue_number} -> pull request #{args.pr_number}",
        file=sys.stderr,
    )
    print(json.dumps(result, indent=2))


# ─────────────────────────────────────────────
#  Shared: milestone lookup
# ─────────────────────────────────────────────

GQL_FIND_MILESTONE_QUERY = """\
query($owner: String!, $repo: String!, $title: String!) {
  repository(owner: $owner, name: $repo) {
    milestones(query: $title, first: 20, states: [OPEN CLOSED]) {
      nodes {
        number
        title
        state
        issues(states: [OPEN]) { totalCount }
        closed_issues: issues(states: [CLOSED]) { totalCount }
      }
    }
  }
}
"""


def find_milestone(title: str) -> dict:
    """Look up milestone by title via GraphQL, return {number, title, open_issues, closed_issues}."""
    variables = {"owner": OWNER, "repo": REPO_NAME, "title": title}
    data = run_gh_graphql(GQL_FIND_MILESTONE_QUERY, variables)
    nodes = data["repository"]["milestones"]["nodes"]
    for ms in nodes:
        if ms["title"] == title:
            return {
                "number": ms["number"],
                "title": ms["title"],
                "open_issues": ms["issues"]["totalCount"],
                "closed_issues": ms["closed_issues"]["totalCount"],
            }
    print(f"ERROR: Milestone \"{title}\" not found in {REPO}", file=sys.stderr)
    sys.exit(1)


# ─────────────────────────────────────────────
#  Subcommand: issues
# ─────────────────────────────────────────────

GQL_ISSUES_QUERY = """\
query($owner: String!, $repo: String!, $milestone: Int!, $cursor: String) {
  repository(owner: $owner, name: $repo) {
    milestone(number: $milestone) {
      issues(first: 100, after: $cursor, states: __STATES__) {
        totalCount
        pageInfo { hasNextPage endCursor }
          nodes {
            ... on Issue {
              number
              title
              state
              issueType { name }
              labels(first: 20) { nodes { name } }
              closedByPullRequestsReferences(first: 5) { nodes { number } }
              projectItems(first: 10) {
                nodes {
                  project { title }
                  fieldValueByName(name: "Status") {
                    ... on ProjectV2ItemFieldSingleSelectValue {
                      name
                    }
                  }
                }
              }
            }
        }
      }
    }
  }
}
"""


GQL_NO_MILESTONE_QUERY = """\
query($query: String!, $cursor: String) {
  search(
    query: $query
    type: ISSUE
    first: 100
    after: $cursor
  ) {
    issueCount
    pageInfo { hasNextPage endCursor }
    nodes {
      ... on Issue {
        number
        title
        state
        milestone { title }
        issueType { name }
        labels(first: 20) { nodes { name } }
        closedByPullRequestsReferences(first: 5) { nodes { number } }
        projectItems(first: 10) {
          nodes {
            project { title }
            fieldValueByName(name: "Status") {
              ... on ProjectV2ItemFieldSingleSelectValue {
                name
              }
            }
          }
        }
      }
    }
  }
}
"""


def fetch_no_milestone_issues(states: str, labels: str | None = None) -> list[dict]:
    """
    Fetch all issues that belong to NO milestone via paginated GraphQL search.

    Args:
        states: GraphQL states enum array literal, e.g. ``"[CLOSED]"`` or ``"[OPEN CLOSED]"``
        labels: optional comma-separated labels to include (built into the search query)

    Returns:
        List of {number, title, state, milestone, issue_type, labels, closing_prs, project_status}
    """
    all_nodes: list[dict] = []
    cursor: str | None = None

    # Map states enum literal to search qualifiers
    state_qualifiers = {
        "[OPEN]": "is:open",
        "[CLOSED]": "is:closed",
        "[OPEN CLOSED]": "",
    }
    state_q = state_qualifiers.get(states, "")
    label_q = ""
    if labels:
        for lbl in labels.split(","):
            label_q += f" label:\"{lbl.strip()}\""
    search_query = f"repo:{OWNER}/{REPO_NAME} is:issue no:milestone{state_q}{label_q}".strip()
    while True:
        variables: dict[str, Any] = {
            "query": search_query,
            "cursor": cursor,
        }
        data = run_gh_graphql(GQL_NO_MILESTONE_QUERY, variables)
        search = data["search"]
        page_info = search["pageInfo"]

        for node in search["nodes"]:
            if node is None:
                continue
            issue_type = node.get("issueType")
            ms = node.get("milestone")
            project_status = None
            for pi in (node.get("projectItems") or {}).get("nodes") or []:
                project = pi.get("project") or {}
                if project.get("title") == "Main":
                    status_field = pi.get("fieldValueByName") or {}
                    project_status = status_field.get("name")
                    break
            all_nodes.append({
                "number": node["number"],
                "title": node["title"],
                "state": node["state"],
                "milestone": ms["title"] if ms else None,
                "issue_type": issue_type["name"] if issue_type else None,
                "labels": [lbl["name"] for lbl in node["labels"]["nodes"]],
                "closing_prs": [pr["number"] for pr in node["closedByPullRequestsReferences"]["nodes"]],
                "project_status": project_status,
            })

        total = len(all_nodes)
        print(f"  ... fetched {total} issues so far", file=sys.stderr)

        if not page_info["hasNextPage"]:
            break
        cursor = page_info["endCursor"]

    return all_nodes


def fetch_milestone_issues(milestone_num: int, states: str) -> list[dict]:
    """
    Fetch all issues in a milestone via paginated GraphQL.

    Args:
        milestone_num: milestone number
        states: GraphQL states enum array literal, e.g. ``"[CLOSED]"`` or ``"[OPEN CLOSED]"``

    Returns:
        List of {number, title, state, issue_type: str|None, labels: [str], closing_prs: [int]}
    """
    query = GQL_ISSUES_QUERY.replace("__STATES__", states)
    all_nodes: list[dict] = []
    cursor: str | None = None

    while True:
        variables: dict[str, Any] = {
            "owner": OWNER,
            "repo": REPO_NAME,
            "milestone": milestone_num,
            "cursor": cursor,
        }
        data = run_gh_graphql(query, variables)
        issues = data["repository"]["milestone"]["issues"]
        page_info = issues["pageInfo"]

        for node in issues["nodes"]:
            if node is None:
                continue
            issue_type = node.get("issueType")
            # Extract project status from the "Main" project board (if present)
            project_status = None
            for pi in (node.get("projectItems") or {}).get("nodes") or []:
                project = pi.get("project") or {}
                if project.get("title") == "Main":
                    status_field = pi.get("fieldValueByName") or {}
                    project_status = status_field.get("name")
                    break
            all_nodes.append({
                "number": node["number"],
                "title": node["title"],
                "state": node["state"],
                "issue_type": issue_type["name"] if issue_type else None,
                "labels": [lbl["name"] for lbl in node["labels"]["nodes"]],
                "closing_prs": [pr["number"] for pr in node["closedByPullRequestsReferences"]["nodes"]],
                "project_status": project_status,
            })

        total = len(all_nodes)
        print(f"  ... fetched {total} issues so far", file=sys.stderr)

        if not page_info["hasNextPage"]:
            break
        cursor = page_info["endCursor"]

    return all_nodes


def load_existing_issue_numbers(filepath: str) -> set[int]:
    """Parse all ``#NNNN`` references from a file (e.g. CHANGES.md)."""
    pattern = re.compile(r"#(\d{3,5})\b")
    nums: set[int] = set()
    with open(filepath) as f:
        for line in f:
            for m in pattern.finditer(line):
                nums.add(int(m.group(1)))
    return nums


def cmd_issues(args: argparse.Namespace) -> None:
    """Handle the ``issues`` subcommand."""

    # Map state to GraphQL enum array literal
    state_map = {"open": "[OPEN]", "closed": "[CLOSED]", "all": "[OPEN CLOSED]"}
    gql_states = state_map[args.state]

    # ── No-milestone path ──────────────────────────────────────────
    if args.milestone and args.milestone.lower() == "none":
        print("Fetching issues with NO milestone...", file=sys.stderr)
        issues = fetch_no_milestone_issues(gql_states, labels=args.label)
        print(f"Fetched {len(issues)} issues total", file=sys.stderr)

    # ── Milestone path ─────────────────────────────────────────────
    else:
        print(f"Looking up milestone \"{args.milestone}\"...", file=sys.stderr)
        ms = find_milestone(args.milestone)
        print(f"Milestone #{ms['number']}: {ms['open_issues']} open, {ms['closed_issues']} closed",
              file=sys.stderr)
        print(f"Fetching {args.state} issues via GraphQL...", file=sys.stderr)
        issues = fetch_milestone_issues(ms["number"], gql_states)
        print(f"Fetched {len(issues)} issues total", file=sys.stderr)

    # Filter by excluded labels
    if args.exclude:
        exclusions = set(label.strip() for label in args.exclude.split(","))
        filtered = [issue for issue in issues
                    if not any(lbl in exclusions for lbl in issue["labels"])]
        print(f"After excluding labels: {len(filtered)} issues", file=sys.stderr)
        issues = filtered

    # Exclude issues with type "Task" (internal chores) — opt out with --include-tasks
    if not args.include_tasks:
        tasks = [iss for iss in issues if iss.get("issue_type") == "Task"]
        if tasks:
            issues = [iss for iss in issues if iss.get("issue_type") != "Task"]
            print(f"After excluding Task issues: {len(issues)} issues (removed {len(tasks)}: {[t['number'] for t in tasks]})", file=sys.stderr)

    # Filter by included labels (--label) — issue must have ALL specified labels
    if args.label:
        inclusions = set(label.strip() for label in args.label.split(","))
        filtered = [issue for issue in issues
                    if all(lbl in issue["labels"] for lbl in inclusions)]
        print(f"After filtering by labels: {len(filtered)} issues", file=sys.stderr)
        issues = filtered

    # Filter out issues with "Rejected" project status (unless --include-rejected)
    if not args.include_rejected:
        rejected = [iss for iss in issues if iss.get("project_status") == "Rejected"]
        if rejected:
            issues = [iss for iss in issues if iss.get("project_status") != "Rejected"]
            print(f"After excluding rejected: {len(issues)} issues (removed {len(rejected)}: {[r['number'] for r in rejected]})", file=sys.stderr)

    # Filter to issues NOT yet in the comparison file (if --compare given)
    if args.compare:
        existing_nums = load_existing_issue_numbers(args.compare)
        missing = [iss for iss in issues if iss["number"] not in existing_nums]
        missing.sort(key=lambda x: x["number"])
        print(f"Issues not yet in changelog: {len(missing)}", file=sys.stderr)
        issues = missing

    print(json.dumps(issues, indent=2))


# ─────────────────────────────────────────────
#  Subcommand: prs
# ─────────────────────────────────────────────

PRS_BATCH_SIZE = 50

GQL_PRS_QUERY_ITEM = """\
    pr_{num}: pullRequest(number: {num}) {{
      number
      title
      body
      state
      mergedAt
      createdAt
      author {{ login }}
      labels(first: 20) {{ nodes {{ name }} }}
      closingIssuesReferences(first: 5) {{ nodes {{ number }} }}
    }}
"""

GQL_PRS_QUERY_WRAPPER = """\
query($owner: String!, $repo: String!) {{
  repository(owner: $owner, name: $repo) {{
{items}
  }}
}}
"""


def fetch_prs_batch(pr_numbers: list[int]) -> list[dict]:
    """
    Fetch details for a list of PR numbers in a single GraphQL query.

    Uses numbered aliases (pr_1234, pr_5678, …) so each PR is looked up by
    number in one round-trip. Returns entries in the same order as the input.
    """
    items = "\n".join(
        GQL_PRS_QUERY_ITEM.format(num=n) for n in pr_numbers
    )
    query = GQL_PRS_QUERY_WRAPPER.format(items=items)
    variables = {"owner": OWNER, "repo": REPO_NAME}

    data = run_gh_graphql(query, variables)
    repo = data["repository"]

    results: list[dict] = []
    for num in pr_numbers:
        pr = repo.get(f"pr_{num}")
        if pr is None:
            results.append({
                "number": num,
                "error": "not_found",
            })
            continue
        results.append({
            "number": pr["number"],
            "title": pr["title"],
            "body": pr.get("body"),
            "state": pr["state"],
            "merged_at": pr.get("mergedAt"),
            "created_at": pr.get("createdAt"),
            "author": pr["author"]["login"] if pr["author"] else None,
            "labels": [lbl["name"] for lbl in pr["labels"]["nodes"]],
            "closing_issues": [iss["number"] for iss in pr["closingIssuesReferences"]["nodes"]],
        })
    return results


GQL_MILESTONE_PRS_QUERY = """\
query($owner: String!, $repo: String!, $milestone: Int!, $cursor: String) {
  repository(owner: $owner, name: $repo) {
    milestone(number: $milestone) {
      pullRequests(first: 100, after: $cursor, states: __STATES__) {
        totalCount
        pageInfo { hasNextPage endCursor }
        nodes {
          ... on PullRequest {
            number
            title
            body
            state
            mergedAt
            createdAt
            headRefName
            author { login }
            labels(first: 20) { nodes { name } }
            files(first: 100) { nodes { path } }
            closingIssuesReferences(first: 5) { nodes { number } }
          }
        }
      }
    }
  }
}
"""


def fetch_milestone_prs(milestone_num: int, states: str) -> list[dict]:
    """
    Fetch all pull requests in a milestone via paginated GraphQL.

    Args:
        milestone_num: milestone number
        states: GraphQL states enum array literal, e.g. ``"[MERGED]"`` or ``"[OPEN CLOSED MERGED]"``

    Returns:
        List of {number, title, body, state, merged_at, created_at,
                head_ref_name, author, labels: [str], files: [str],
                closing_issues: [int]}
    """
    query = GQL_MILESTONE_PRS_QUERY.replace("__STATES__", states)
    all_nodes: list[dict] = []
    cursor: str | None = None

    while True:
        variables: dict[str, Any] = {
            "owner": OWNER,
            "repo": REPO_NAME,
            "milestone": milestone_num,
            "cursor": cursor,
        }
        data = run_gh_graphql(query, variables)
        prs = data["repository"]["milestone"]["pullRequests"]
        page_info = prs["pageInfo"]

        for node in prs["nodes"]:
            if node is None:
                continue
            all_nodes.append({
                "number": node["number"],
                "title": node["title"],
                "body": node.get("body"),
                "state": node["state"],
                "merged_at": node.get("mergedAt"),
                "created_at": node.get("createdAt"),
                "head_ref_name": node.get("headRefName"),
                "author": node["author"]["login"] if node["author"] else None,
                "labels": [lbl["name"] for lbl in node["labels"]["nodes"]],
                "files": [file["path"] for file in node["files"]["nodes"]],
                "closing_issues": [iss["number"] for iss in node["closingIssuesReferences"]["nodes"]],
            })

        total = len(all_nodes)
        print(f"  ... fetched {total} PRs so far", file=sys.stderr)

        if not page_info["hasNextPage"]:
            break
        cursor = page_info["endCursor"]

    return all_nodes


def cmd_prs(args: argparse.Namespace) -> None:
    """Handle the ``prs`` subcommand."""

    # ── Milestone path ──────────────────────────────────────────────
    if args.milestone:
        print(f"Looking up milestone \"{args.milestone}\"...", file=sys.stderr)
        ms = find_milestone(args.milestone)

        state_map = {"open": "[OPEN]", "closed": "[CLOSED]", "merged": "[MERGED]", "all": "[OPEN CLOSED MERGED]"}
        gql_states = state_map[args.state]

        print(f"Fetching {args.state} PRs via GraphQL...", file=sys.stderr)
        prs = fetch_milestone_prs(ms["number"], gql_states)
        print(f"Fetched {len(prs)} PRs total", file=sys.stderr)
        print(json.dumps(prs, indent=2))
        return

    # ── Number-based path ───────────────────────────────────────────
    pr_numbers: list[int] = []

    if args.numbers:
        pr_numbers.extend(args.numbers)

    if args.file:
        with open(args.file) as f:
            for line in f:
                line = line.strip()
                if line:
                    pr_numbers.append(int(line))

    if args.stdin:
        for line in sys.stdin:
            line = line.strip()
            if line:
                pr_numbers.append(int(line))

    if not pr_numbers:
        print("ERROR: no PR numbers provided (pass numbers, --file, --stdin, or --milestone)",
              file=sys.stderr)
        sys.exit(1)

    # Deduplicate while preserving order
    seen: set[int] = set()
    pr_numbers = [n for n in pr_numbers if not (n in seen or seen.add(n))]

    print(f"Fetching {len(pr_numbers)} PRs in batches of {PRS_BATCH_SIZE}...",
          file=sys.stderr)

    all_results: list[dict] = []
    for i in range(0, len(pr_numbers), PRS_BATCH_SIZE):
        batch = pr_numbers[i : i + PRS_BATCH_SIZE]
        print(f"  batch {i // PRS_BATCH_SIZE + 1}: PRs {batch[0]}..{batch[-1]}",
              file=sys.stderr)
        all_results.extend(fetch_prs_batch(batch))

    print(json.dumps(all_results, indent=2))


# ─────────────────────────────────────────────
#  Subcommand: advisories
# ─────────────────────────────────────────────


def fetch_advisories() -> list[dict]:
    """Fetch all security advisories for the repository via REST API."""
    all_advisories: list[dict] = []
    page = 1

    while True:
        advisories = run_gh_rest(
            f"repos/{REPO}/security-advisories?per_page=100&page={page}"
        )
        all_advisories.extend(advisories)

        if len(advisories) < 100:
            break
        page += 1

    return all_advisories


def fetch_advisory(ghsa_id: str) -> dict:
    """Fetch a single security advisory by GHSA ID."""
    return run_gh_rest(f"repos/{REPO}/security-advisories/{ghsa_id}")


def format_advisory_summary(adv: dict) -> dict:
    """Extract a summary view of an advisory."""
    return {
        "ghsa_id": adv["ghsa_id"],
        "cve_id": adv.get("cve_id"),
        "severity": adv.get("severity"),
        "cvss_score": (adv.get("cvss") or {}).get("score"),
        "state": adv.get("state"),
        "summary": adv.get("summary"),
        "cwes": [c["cwe_id"] for c in adv.get("cwes", [])],
        "published_at": adv.get("published_at"),
        "closed_at": adv.get("closed_at"),
        "url": adv.get("html_url"),
    }


def format_advisory_detail(adv: dict) -> dict:
    """Extract full detail view of an advisory."""
    summary = format_advisory_summary(adv)
    summary["description"] = adv.get("description")
    summary["vulnerabilities"] = [
        {
            "package": v.get("package", {}).get("name"),
            "vulnerable_version_range": v.get("vulnerable_version_range"),
            "patched_versions": v.get("patched_versions"),
        }
        for v in adv.get("vulnerabilities", [])
    ]
    summary["credits"] = [
        {"login": c.get("user", {}).get("login"), "type": c.get("type")}
        for c in adv.get("credits_detailed", [])
    ]
    summary["created_at"] = adv.get("created_at")
    summary["updated_at"] = adv.get("updated_at")
    summary["withdrawn_at"] = adv.get("withdrawn_at")
    return summary


def cmd_advisories(args: argparse.Namespace) -> None:
    """Handle the ``advisories`` subcommand."""

    # ── Single advisory detail ──────────────────────────────
    if args.ghsa_id:
        ghsa_id = args.ghsa_id.upper()
        if not ghsa_id.startswith("GHSA-"):
            ghsa_id = f"GHSA-{ghsa_id}"
        print(f"Fetching advisory {ghsa_id}...", file=sys.stderr)
        adv = fetch_advisory(ghsa_id)
        print(json.dumps(format_advisory_detail(adv), indent=2))
        return

    # ── List all advisories ─────────────────────────────────
    print("Fetching security advisories...", file=sys.stderr)
    advisories = fetch_advisories()
    print(f"Fetched {len(advisories)} advisories", file=sys.stderr)

    results = [format_advisory_summary(adv) for adv in advisories]

    # Apply filters
    if args.severity:
        sev = args.severity.lower()
        results = [r for r in results if (r.get("severity") or "").lower() == sev]
        print(f"After severity filter ({sev}): {len(results)} advisories", file=sys.stderr)

    if args.state:
        st = args.state.lower()
        results = [r for r in results if (r.get("state") or "").lower() == st]
        print(f"After state filter ({st}): {len(results)} advisories", file=sys.stderr)

    print(json.dumps(results, indent=2))


# ─────────────────────────────────────────────
#  CLI entrypoint
# ─────────────────────────────────────────────


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Multi-purpose CLI helper for penpot/penpot GitHub operations"
    )
    sub = parser.add_subparsers(dest="command", required=True, title="subcommands")

    # --- issues ---
    p_issues = sub.add_parser("issues", help="List issues in a milestone (or use 'none' for unassigned)")
    p_issues.add_argument("milestone", help="Milestone title (e.g. '2.16.0') or 'none' for issues with no milestone")
    p_issues.add_argument(
        "--state", choices=["open", "closed", "all"], default="closed",
        help="Issue state filter (default: closed)"
    )
    p_issues.add_argument(
        "--exclude", "--exclude-labels",
        help="Comma-separated labels to exclude, e.g. 'release blocker,no changelog'"
    )
    p_issues.add_argument(
        "--label", "--labels",
        help="Comma-separated labels to include (issue must have ALL specified), e.g. 'bug' or 'bug,regression'"
    )
    p_issues.add_argument(
        "--compare",
        help="Path to CHANGES.md; only show issues NOT yet referenced in that file"
    )
    p_issues.add_argument(
        "--include-rejected", action="store_true",
        help="Include issues with 'Rejected' project status (excluded by default)"
    )
    p_issues.add_argument(
        "--include-tasks", action="store_true",
        help="Include issues with type 'Task' (excluded by default, they are internal chores)"
    )
    p_issues.set_defaults(func=cmd_issues)

    # --- prs ---
    p_prs = sub.add_parser("prs", help="Fetch details for one or more PRs (by number or milestone)")
    p_prs.add_argument(
        "numbers", type=int, nargs="*",
        help="PR numbers to fetch (space-separated)"
    )
    p_prs.add_argument(
        "--file", type=str,
        help="File with one PR number per line"
    )
    p_prs.add_argument(
        "--stdin", action="store_true",
        help="Read PR numbers from stdin (one per line)"
    )
    p_prs.add_argument(
        "--milestone", type=str,
        help="Milestone title, e.g. '2.16.0' (fetches all PRs in the milestone)"
    )
    p_prs.add_argument(
        "--state", choices=["open", "closed", "merged", "all"], default="merged",
        help="PR state filter when using --milestone (default: merged)"
    )
    p_prs.set_defaults(func=cmd_prs)

    # --- link-issue ---
    p_link = sub.add_parser(
        "link-issue",
        aliases=["link"],
        help="Explicitly link an issue to a pull request and verify both sides",
    )
    p_link.add_argument("issue_number", type=int, help="Issue number")
    p_link.add_argument("pr_number", type=int, help="Pull request number")
    p_link.set_defaults(func=cmd_link_issue)

    # --- advisories ---
    p_adv = sub.add_parser("advisories", help="List or inspect GitHub security advisories")
    p_adv.add_argument(
        "ghsa_id", nargs="?",
        help="GHSA ID to fetch (e.g. 'GHSA-xvj6-fh9w-gjw7'); omit to list all"
    )
    p_adv.add_argument(
        "--severity", choices=["critical", "high", "medium", "low"],
        help="Filter by severity level"
    )
    p_adv.add_argument(
        "--state", choices=["triage", "draft", "published", "closed", "withdrawn"],
        help="Filter by advisory state"
    )
    p_adv.set_defaults(func=cmd_advisories)

    args = parser.parse_args()
    args.func(args)


if __name__ == "__main__":
    main()