"""Assign and notify **new** contributors on `good first issue` threads. Here *new* means the commenter has **no merged PRs and no open PRs** in this repo where they are the PR author (GitHub Search API). **One or more merged PRs** means they are not treated as a first-time contributor for this automation (GitHub's ``FIRST_TIME_CONTRIBUTOR`` / ``FIRST_TIMER`` flags are not used). Also skips repo insiders (OWNER / MEMBER / COLLABORATOR), bots, closed issues, comments on **pull request** threads (``issue_comment`` fires for PRs too; those use ``issue.pull_request``), and commenters already listed as assignees. **One open assignment per eligible new contributor** in this repo: if they already have another **open** issue assigned (Search API), they cannot be auto-assigned here. At most **one** auto-assignment per issue: if anyone else is already an assignee, further eligible commenters are skipped (manual pre-assignments count). """ from __future__ import annotations import json import os import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any GOOD_FIRST_LABEL = "good first issue" # Do not auto-assign maintainers/collaborators; # eligibility is 0 merged + 0 open PRs as author + not insider. EXCLUDED_COMMENTER_ASSOCIATIONS = frozenset({"OWNER", "MEMBER", "COLLABORATOR"}) GITHUB_API = "https://api.github.com" def _github_api_url(path: str, query: dict[str, str]) -> str: encoded = urllib.parse.urlencode(query) return f"{GITHUB_API}{path}?{encoded}" def screen_event_without_api(event: dict[str, Any]) -> str | None: """Return a skip reason before calling the GitHub API, or None if checks should continue.""" issue = event.get("issue") or {} comment = event.get("comment") or {} if issue.get("pull_request") is not None: return "comment_on_pull_request" if issue.get("state") != "open": return "issue_not_open" labels = issue.get("labels") or [] if not isinstance(labels, list): return "invalid_labels" names = {item.get("name") for item in labels if isinstance(item, dict)} if GOOD_FIRST_LABEL not in names: return "not_good_first_issue" c_user = comment.get("user") or {} if c_user.get("type") == "Bot": return "bot_commenter" c_login = c_user.get("login") or "" if not c_login: return "missing_commenter_login" c_assoc = comment.get("author_association") or "" if c_assoc in EXCLUDED_COMMENTER_ASSOCIATIONS: return "commenter_repo_insider" assignees = issue.get("assignees") or [] if isinstance(assignees, list): assigned_logins = { a.get("login") for a in assignees if isinstance(a, dict) and a.get("login") } if c_login in assigned_logins: return "already_assignee" if assigned_logins: return "issue_already_claimed" return None def assign_decision( *, skip_reason_pre_api: str | None, merged_pr_count_for_commenter: int, open_pr_count_for_commenter: int, open_assigned_issue_count_for_commenter: int, ) -> tuple[bool, str]: """Return (should_assign_and_comment, skip_reason_or_empty). Eligible "new contributor" means ``merged_pr_count_for_commenter == 0`` and ``open_pr_count_for_commenter == 0`` (PR author in this repo, via Search API). They must also have no other open issues assigned in this repo (``open_assigned_issue_count_for_commenter == 0``). """ if skip_reason_pre_api is not None: return False, skip_reason_pre_api if open_pr_count_for_commenter > 0: return False, "has_open_prs" if merged_pr_count_for_commenter > 0: return False, "has_merged_prs" if open_assigned_issue_count_for_commenter > 0: return False, "already_has_open_assigned_issue" return True, "" def _request_json(url: str, token: str) -> Any: parsed = urllib.parse.urlparse(url) if parsed.scheme != "https" or parsed.netloc != "api.github.com": raise ValueError("GitHub API URL must target https://api.github.com") req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {token}", "Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28", }, method="GET", ) with ( urllib.request.urlopen(req, timeout=60) as resp ): # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected return json.loads(resp.read().decode("utf-8")) def _search_issue_total_count(query: str, token: str) -> int: url = _github_api_url("/search/issues", {"q": query}) try: data = _request_json(url, token) except urllib.error.URLError as exc: print(f"GitHub search failed: {exc}", file=sys.stderr) raise total = data.get("total_count") if not isinstance(total, int): return 0 return total def fetch_merged_pr_count(owner: str, repo: str, login: str, token: str) -> int: q = f"repo:{owner}/{repo} is:pr is:merged author:{login}" return _search_issue_total_count(q, token) def fetch_open_pr_count(owner: str, repo: str, login: str, token: str) -> int: q = f"repo:{owner}/{repo} is:pr is:open author:{login}" return _search_issue_total_count(q, token) def fetch_open_assigned_issue_count(owner: str, repo: str, login: str, token: str) -> int: """Count open issues in this repo where ``login`` is an assignee.""" q = f"repo:{owner}/{repo} is:issue is:open assignee:{login}" return _search_issue_total_count(q, token) def build_assign_notice_body(*, assignee_login: str) -> str: return f"@{assignee_login} You've been **assigned** to this issue. Thanks for picking it up." def set_github_output(name: str, value: str) -> None: path = os.environ.get("GITHUB_OUTPUT") if not path: return with open(path, "a", encoding="utf-8") as fh: fh.write(f"{name}={value}\n") def main() -> int: event_path = os.environ.get("GITHUB_EVENT_PATH") repository = os.environ.get("GITHUB_REPOSITORY") token = os.environ.get("GITHUB_TOKEN") if not event_path or not repository or not token: print("Missing GITHUB_EVENT_PATH, GITHUB_REPOSITORY, or GITHUB_TOKEN.", file=sys.stderr) return 1 raw = Path(event_path).read_text(encoding="utf-8") event = json.loads(raw) pre = screen_event_without_api(event) merged_count = 0 open_count = 0 open_assigned_issue_count = 0 if pre is None: owner, _, repo = repository.partition("/") if not owner or not repo: print("Invalid GITHUB_REPOSITORY.", file=sys.stderr) return 1 comment = event.get("comment") or {} c_login = (comment.get("user") or {}).get("login") or "" try: merged_count = fetch_merged_pr_count(owner, repo, c_login, token) open_count = fetch_open_pr_count(owner, repo, c_login, token) open_assigned_issue_count = fetch_open_assigned_issue_count(owner, repo, c_login, token) except urllib.error.URLError as exc: print(f"GitHub API request failed: {exc}", file=sys.stderr) return 1 should, reason = assign_decision( skip_reason_pre_api=pre, merged_pr_count_for_commenter=merged_count, open_pr_count_for_commenter=open_count, open_assigned_issue_count_for_commenter=open_assigned_issue_count, ) if not should: print(f"Skip: {reason}") set_github_output("should_assign", "false") return 0 comment_user = (event.get("comment") or {}).get("user") or {} login = comment_user.get("login") if isinstance(comment_user, dict) else "" if not isinstance(login, str) or not login: print("Missing commenter login.", file=sys.stderr) return 1 body = build_assign_notice_body(assignee_login=login) Path("assign_comment.md").write_text(body, encoding="utf-8") set_github_output("should_assign", "true") print("Wrote assign_comment.md; assignment will be applied in workflow.") return 0 if __name__ == "__main__": raise SystemExit(main())