diff options
Diffstat (limited to 'bot/exts/events/hacktoberfest/hacktoberstats.py')
| -rw-r--r-- | bot/exts/events/hacktoberfest/hacktoberstats.py | 437 |
1 files changed, 0 insertions, 437 deletions
diff --git a/bot/exts/events/hacktoberfest/hacktoberstats.py b/bot/exts/events/hacktoberfest/hacktoberstats.py deleted file mode 100644 index 72067dbe..00000000 --- a/bot/exts/events/hacktoberfest/hacktoberstats.py +++ /dev/null @@ -1,437 +0,0 @@ -import logging -import random -import re -from collections import Counter -from datetime import datetime, timedelta -from typing import Optional, Union -from urllib.parse import quote_plus - -import discord -from async_rediscache import RedisCache -from discord.ext import commands - -from bot.bot import Bot -from bot.constants import Colours, Month, NEGATIVE_REPLIES, Tokens -from bot.utils.decorators import in_month - -log = logging.getLogger(__name__) - -CURRENT_YEAR = datetime.now().year # Used to construct GH API query -PRS_FOR_SHIRT = 4 # Minimum number of PRs before a shirt is awarded -REVIEW_DAYS = 14 # number of days needed after PR can be mature - -REQUEST_HEADERS = {"User-Agent": "Python Discord Hacktoberbot"} -# using repo topics API during preview period requires an accept header -GITHUB_TOPICS_ACCEPT_HEADER = {"Accept": "application/vnd.github.mercy-preview+json"} -if GITHUB_TOKEN := Tokens.github: - REQUEST_HEADERS["Authorization"] = f"token {GITHUB_TOKEN}" - GITHUB_TOPICS_ACCEPT_HEADER["Authorization"] = f"token {GITHUB_TOKEN}" - -GITHUB_NONEXISTENT_USER_MESSAGE = ( - "The listed users cannot be searched either because the users do not exist " - "or you do not have permission to view the users." -) - - -class HacktoberStats(commands.Cog): - """Hacktoberfest statistics Cog.""" - - # Stores mapping of user IDs and GitHub usernames - linked_accounts = RedisCache() - - def __init__(self, bot: Bot): - self.bot = bot - - @in_month(Month.SEPTEMBER, Month.OCTOBER, Month.NOVEMBER) - @commands.group(name="hacktoberstats", aliases=("hackstats",), invoke_without_command=True) - async def hacktoberstats_group(self, ctx: commands.Context, github_username: str = None) -> None: - """ - Display an embed for a user's Hacktoberfest contributions. - - If invoked without a subcommand or github_username, get the invoking user's stats if they've - linked their Discord name to GitHub using .stats link. If invoked with a github_username, - get that user's contributions - """ - if not github_username: - author_id, author_mention = self._author_mention_from_context(ctx) - - if await self.linked_accounts.contains(author_id): - github_username = await self.linked_accounts.get(author_id) - logging.info(f"Getting stats for {author_id} linked GitHub account '{github_username}'") - else: - msg = ( - f"{author_mention}, you have not linked a GitHub account\n\n" - f"You can link your GitHub account using:\n```\n{ctx.prefix}hackstats link github_username\n```\n" - f"Or query GitHub stats directly using:\n```\n{ctx.prefix}hackstats github_username\n```" - ) - await ctx.send(msg) - return - - await self.get_stats(ctx, github_username) - - @in_month(Month.SEPTEMBER, Month.OCTOBER, Month.NOVEMBER) - @hacktoberstats_group.command(name="link") - async def link_user(self, ctx: commands.Context, github_username: str = None) -> None: - """ - Link the invoking user's Github github_username to their Discord ID. - - Linked users are stored in Redis: User ID => GitHub Username. - """ - author_id, author_mention = self._author_mention_from_context(ctx) - if github_username: - if await self.linked_accounts.contains(author_id): - old_username = await self.linked_accounts.get(author_id) - log.info(f"{author_id} has changed their github link from '{old_username}' to '{github_username}'") - await ctx.send(f"{author_mention}, your GitHub username has been updated to: '{github_username}'") - else: - log.info(f"{author_id} has added a github link to '{github_username}'") - await ctx.send(f"{author_mention}, your GitHub username has been added") - - await self.linked_accounts.set(author_id, github_username) - else: - log.info(f"{author_id} tried to link a GitHub account but didn't provide a username") - await ctx.send(f"{author_mention}, a GitHub username is required to link your account") - - @in_month(Month.SEPTEMBER, Month.OCTOBER, Month.NOVEMBER) - @hacktoberstats_group.command(name="unlink") - async def unlink_user(self, ctx: commands.Context) -> None: - """Remove the invoking user's account link from the log.""" - author_id, author_mention = self._author_mention_from_context(ctx) - - stored_user = await self.linked_accounts.pop(author_id, None) - if stored_user: - await ctx.send(f"{author_mention}, your GitHub profile has been unlinked") - logging.info(f"{author_id} has unlinked their GitHub account") - else: - await ctx.send(f"{author_mention}, you do not currently have a linked GitHub account") - logging.info(f"{author_id} tried to unlink their GitHub account but no account was linked") - - async def get_stats(self, ctx: commands.Context, github_username: str) -> None: - """ - Query GitHub's API for PRs created by a GitHub user during the month of October. - - PRs with an 'invalid' or 'spam' label are ignored - - For PRs created after October 3rd, they have to be in a repository that has a - 'hacktoberfest' topic, unless the PR is labelled 'hacktoberfest-accepted' for it - to count. - - If a valid github_username is provided, an embed is generated and posted to the channel - - Otherwise, post a helpful error message - """ - async with ctx.typing(): - prs = await self.get_october_prs(github_username) - - if prs is None: # Will be None if the user was not found - await ctx.send( - embed=discord.Embed( - title=random.choice(NEGATIVE_REPLIES), - description=f"GitHub user `{github_username}` was not found.", - colour=discord.Colour.red() - ) - ) - return - - if prs: - stats_embed = await self.build_embed(github_username, prs) - await ctx.send("Here are some stats!", embed=stats_embed) - else: - await ctx.send(f"No valid Hacktoberfest PRs found for '{github_username}'") - - async def build_embed(self, github_username: str, prs: list[dict]) -> discord.Embed: - """Return a stats embed built from github_username's PRs.""" - logging.info(f"Building Hacktoberfest embed for GitHub user: '{github_username}'") - in_review, accepted = await self._categorize_prs(prs) - - n = len(accepted) + len(in_review) # Total number of PRs - if n >= PRS_FOR_SHIRT: - shirtstr = f"**{github_username} is eligible for a T-shirt or a tree!**" - elif n == PRS_FOR_SHIRT - 1: - shirtstr = f"**{github_username} is 1 PR away from a T-shirt or a tree!**" - else: - shirtstr = f"**{github_username} is {PRS_FOR_SHIRT - n} PRs away from a T-shirt or a tree!**" - - stats_embed = discord.Embed( - title=f"{github_username}'s Hacktoberfest", - color=Colours.purple, - description=( - f"{github_username} has made {n} valid " - f"{self._contributionator(n)} in " - f"October\n\n" - f"{shirtstr}\n\n" - ) - ) - - stats_embed.set_thumbnail(url=f"https://www.github.com/{github_username}.png") - stats_embed.set_author( - name="Hacktoberfest", - url="https://hacktoberfest.digitalocean.com", - icon_url="https://avatars1.githubusercontent.com/u/35706162?s=200&v=4" - ) - - # This will handle when no PRs in_review or accepted - review_str = self._build_prs_string(in_review, github_username) or "None" - accepted_str = self._build_prs_string(accepted, github_username) or "None" - stats_embed.add_field( - name=":clock1: In Review", - value=review_str - ) - stats_embed.add_field( - name=":tada: Accepted", - value=accepted_str - ) - - logging.info(f"Hacktoberfest PR built for GitHub user '{github_username}'") - return stats_embed - - async def get_october_prs(self, github_username: str) -> Optional[list[dict]]: - """ - Query GitHub's API for PRs created during the month of October by github_username. - - PRs with an 'invalid' or 'spam' label are ignored unless it is merged or approved - - For PRs created after October 3rd, they have to be in a repository that has a - 'hacktoberfest' topic, unless the PR is labelled 'hacktoberfest-accepted' for it - to count. - - If PRs are found, return a list of dicts with basic PR information - - For each PR: - { - "repo_url": str - "repo_shortname": str (e.g. "python-discord/sir-lancebot") - "created_at": datetime.datetime - "number": int - } - - Otherwise, return empty list. - None will be returned when the GitHub user was not found. - """ - log.info(f"Fetching Hacktoberfest Stats for GitHub user: '{github_username}'") - base_url = "https://api.github.com/search/issues" - action_type = "pr" - is_query = "public" - not_query = "draft" - date_range = f"{CURRENT_YEAR}-09-30T10:00Z..{CURRENT_YEAR}-11-01T12:00Z" - per_page = "300" - query_params = ( - f"+type:{action_type}" - f"+is:{is_query}" - f"+author:{quote_plus(github_username)}" - f"+-is:{not_query}" - f"+created:{date_range}" - f"&per_page={per_page}" - ) - - log.debug(f"GitHub query parameters generated: {query_params}") - - jsonresp = await self._fetch_url(base_url, REQUEST_HEADERS, {"q": query_params}) - if "message" in jsonresp: - # One of the parameters is invalid, short circuit for now - api_message = jsonresp["errors"][0]["message"] - - # Ignore logging non-existent users or users we do not have permission to see - if api_message == GITHUB_NONEXISTENT_USER_MESSAGE: - log.debug(f"No GitHub user found named '{github_username}'") - return - else: - log.error(f"GitHub API request for '{github_username}' failed with message: {api_message}") - return [] # No October PRs were found due to error - - if jsonresp["total_count"] == 0: - # Short circuit if there aren't any PRs - log.info(f"No October PRs found for GitHub user: '{github_username}'") - return [] - - logging.info(f"Found {len(jsonresp['items'])} Hacktoberfest PRs for GitHub user: '{github_username}'") - outlist = [] # list of pr information dicts that will get returned - oct3 = datetime(int(CURRENT_YEAR), 10, 3, 23, 59, 59, tzinfo=None) - hackto_topics = {} # cache whether each repo has the appropriate topic (bool values) - for item in jsonresp["items"]: - shortname = self._get_shortname(item["repository_url"]) - itemdict = { - "repo_url": f"https://www.github.com/{shortname}", - "repo_shortname": shortname, - "created_at": datetime.strptime( - item["created_at"], "%Y-%m-%dT%H:%M:%SZ" - ), - "number": item["number"] - } - - # If the PR has 'invalid' or 'spam' labels, the PR must be - # either merged or approved for it to be included - if self._has_label(item, ["invalid", "spam"]): - if not await self._is_accepted(itemdict): - continue - - # PRs before oct 3 no need to check for topics - # continue the loop if 'hacktoberfest-accepted' is labelled then - # there is no need to check for its topics - if itemdict["created_at"] < oct3: - outlist.append(itemdict) - continue - - # Checking PR's labels for "hacktoberfest-accepted" - if self._has_label(item, "hacktoberfest-accepted"): - outlist.append(itemdict) - continue - - # No need to query GitHub if repo topics are fetched before already - if hackto_topics.get(shortname): - outlist.append(itemdict) - continue - # Fetch topics for the PR's repo - topics_query_url = f"https://api.github.com/repos/{shortname}/topics" - log.debug(f"Fetching repo topics for {shortname} with url: {topics_query_url}") - jsonresp2 = await self._fetch_url(topics_query_url, GITHUB_TOPICS_ACCEPT_HEADER) - if jsonresp2.get("names") is None: - log.error(f"Error fetching topics for {shortname}: {jsonresp2['message']}") - continue # Assume the repo doesn't have the `hacktoberfest` topic if API request errored - - # PRs after oct 3 that doesn't have 'hacktoberfest-accepted' label - # must be in repo with 'hacktoberfest' topic - if "hacktoberfest" in jsonresp2["names"]: - hackto_topics[shortname] = True # Cache result in the dict for later use if needed - outlist.append(itemdict) - return outlist - - async def _fetch_url(self, url: str, headers: dict, params: dict) -> dict: - """Retrieve API response from URL.""" - async with self.bot.http_session.get(url, headers=headers, params=params) as resp: - return await resp.json() - - @staticmethod - def _has_label(pr: dict, labels: Union[list[str], str]) -> bool: - """ - Check if a PR has label 'labels'. - - 'labels' can be a string or a list of strings, if it's a list of strings - it will return true if any of the labels match. - """ - if not pr.get("labels"): # if PR has no labels - return False - if isinstance(labels, str) and any(label["name"].casefold() == labels for label in pr["labels"]): - return True - for item in labels: - if any(label["name"].casefold() == item for label in pr["labels"]): - return True - return False - - async def _is_accepted(self, pr: dict) -> bool: - """Check if a PR is merged, approved, or labelled hacktoberfest-accepted.""" - # checking for merge status - query_url = f"https://api.github.com/repos/{pr['repo_shortname']}/pulls/{pr['number']}" - jsonresp = await self._fetch_url(query_url, REQUEST_HEADERS) - - if message := jsonresp.get("message"): - log.error(f"Error fetching PR stats for #{pr['number']} in repo {pr['repo_shortname']}:\n{message}") - return False - - if jsonresp.get("merged"): - return True - - # checking for the label, using `jsonresp` which has the label information - if self._has_label(jsonresp, "hacktoberfest-accepted"): - return True - - # checking approval - query_url += "/reviews" - jsonresp2 = await self._fetch_url(query_url, REQUEST_HEADERS) - if isinstance(jsonresp2, dict): - # if API request is unsuccessful it will be a dict with the error in 'message' - log.error( - f"Error fetching PR reviews for #{pr['number']} in repo {pr['repo_shortname']}:\n" - f"{jsonresp2['message']}" - ) - return False - # if it is successful it will be a list instead of a dict - if len(jsonresp2) == 0: # if PR has no reviews - return False - - # loop through reviews and check for approval - for item in jsonresp2: - if item.get("status") == "APPROVED": - return True - return False - - @staticmethod - def _get_shortname(in_url: str) -> str: - """ - Extract shortname from https://api.github.com/repos/* URL. - - e.g. "https://api.github.com/repos/python-discord/sir-lancebot" - | - V - "python-discord/sir-lancebot" - """ - exp = r"https?:\/\/api.github.com\/repos\/([/\-\_\.\w]+)" - return re.findall(exp, in_url)[0] - - async def _categorize_prs(self, prs: list[dict]) -> tuple: - """ - Categorize PRs into 'in_review' and 'accepted' and returns as a tuple. - - PRs created less than 14 days ago are 'in_review', PRs that are not - are 'accepted' (after 14 days review period). - - PRs that are accepted must either be merged, approved, or labelled - 'hacktoberfest-accepted. - """ - now = datetime.now() - oct3 = datetime(CURRENT_YEAR, 10, 3, 23, 59, 59, tzinfo=None) - in_review = [] - accepted = [] - for pr in prs: - if (pr["created_at"] + timedelta(REVIEW_DAYS)) > now: - in_review.append(pr) - elif (pr["created_at"] <= oct3) or await self._is_accepted(pr): - accepted.append(pr) - - return in_review, accepted - - @staticmethod - def _build_prs_string(prs: list[tuple], user: str) -> str: - """ - Builds a discord embed compatible string for a list of PRs. - - Repository name with the link to pull requests authored by 'user' for - each PR. - """ - base_url = "https://www.github.com/" - str_list = [] - repo_list = [pr["repo_shortname"] for pr in prs] - prs_list = Counter(repo_list).most_common(5) # get first 5 counted PRs - more = len(prs) - sum(i[1] for i in prs_list) - - for pr in prs_list: - # for example: https://www.github.com/python-discord/bot/pulls/octocat - # will display pull requests authored by octocat. - # pr[1] is the number of PRs to the repo - string = f"{pr[1]} to [{pr[0]}]({base_url}{pr[0]}/pulls/{user})" - str_list.append(string) - if more: - str_list.append(f"...and {more} more") - - return "\n".join(str_list) - - @staticmethod - def _contributionator(n: int) -> str: - """Return "contribution" or "contributions" based on the value of n.""" - if n == 1: - return "contribution" - else: - return "contributions" - - @staticmethod - def _author_mention_from_context(ctx: commands.Context) -> tuple[str, str]: - """Return stringified Message author ID and mentionable string from commands.Context.""" - author_id = str(ctx.author.id) - author_mention = ctx.author.mention - - return author_id, author_mention - - -def setup(bot: Bot) -> None: - """Load the Hacktober Stats Cog.""" - bot.add_cog(HacktoberStats(bot)) |