path: root/bot/exts/events/hacktoberfest/_utils.py

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 discord.ext import commands

from bot.bot import Bot
from bot.constants import Colours, NEGATIVE_REPLIES, Tokens

log = logging.getLogger(__name__)

# Constants for `.hacktoberfest issue`
ISSUES_REQUEST_HEADERS = {
    "User-Agent": "Python Discord Hacktoberbot",
    "Accept": "application / vnd.github.v3 + json"
}
if GITHUB_TOKEN := Tokens.github:
    ISSUES_REQUEST_HEADERS["Authorization"] = f"token {GITHUB_TOKEN}"

# Constants for `.hacktoberfest stats`
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

STATS_REQUEST_HEADERS = {"User-Agent": "Python Discord Hacktoberbot"}
GITHUB_TOPICS_ACCEPT_HEADER = {"Accept": "application/vnd.github.mercy-preview+json"}

# using repo topics API during preview period requires an accept header
if GITHUB_TOKEN := Tokens.github:
    STATS_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."
)

URL = "https://api.github.com/search/issues?per_page=100&q=is:issue+label:hacktoberfest+language:python+state:open"


# Util functions for `.hacktoberfest timeleft`
def in_hacktober() -> bool:
    """Return True if the current time is within Hacktoberfest."""
    _, end, start = load_date()

    now = datetime.utcnow()

    return start <= now <= end


def load_date() -> tuple[datetime, datetime, datetime]:
    """Return of a tuple of the current time and the end and start times of the next October."""
    now = datetime.utcnow()
    year = now.year
    if now.month > 10:
        year += 1
    end = datetime(year, 11, 1, 12)  # November 1st 12:00 (UTC-12:00)
    start = datetime(year, 9, 30, 10)  # September 30th 10:00 (UTC+14:00)
    return now, end, start


# Util functions for `.hacktoberfest stats`
async def get_stats(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 get_october_prs(ctx.bot, 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 build_stats_embed(ctx.bot, 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_stats_embed(bot: Bot, 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 categorize_prs(bot, 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"{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 = build_prs_string(in_review, github_username) or "None"
    accepted_str = 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(bot: Bot, 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 fetch_url(bot, base_url, STATS_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 = 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 has_label(item, ["invalid", "spam"]):
            if not await is_accepted(bot, 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 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 fetch_url(bot, topics_query_url, GITHUB_TOPICS_ACCEPT_HEADER, {"q": query_params})
        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(bot: Bot, url: str, headers: dict, params: dict) -> dict:
    """Retrieve API response from URL."""
    async with bot.http_session.get(url, headers=headers, params=params) as resp:
        return await resp.json()


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(bot: Bot, 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 fetch_url(bot, query_url, STATS_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 has_label(jsonresp, "hacktoberfest-accepted"):
        return True

    # checking approval
    query_url += "/reviews"
    jsonresp2 = await fetch_url(bot, query_url, STATS_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


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(bot: Bot, prs: list[dict]) -> tuple[list[dict], list[dict]]:
    """
    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 is_accepted(bot, pr):
            accepted.append(pr)

    return in_review, accepted


def build_prs_string(prs: list[dict], 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)


def contributionator(n: int) -> str:
    """Return "contribution" or "contributions" based on the value of n."""
    if n == 1:
        return "contribution"
    else:
        return "contributions"


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


# Util functions for `.hacktoberfest issue`
async def get_issues(ctx: commands.Context, option: str) -> Optional[dict]:
    """Get a list of the python issues with the label 'hacktoberfest' from the Github api."""
    if option == "beginner":
        if (ctx.message.created_at.replace(tzinfo=None) - ctx.cog.cache_timer_beginner).seconds <= 60:
            log.debug("using cache")
            return ctx.cog.cache_beginner
    elif (ctx.message.created_at.replace(tzinfo=None) - ctx.cog.cache_timer_normal).seconds <= 60:
        log.debug("using cache")
        return ctx.cog.cache_normal

    if option == "beginner":
        url = URL + '+label:"good first issue"'
        if ctx.cog.cache_beginner is not None:
            page = random.randint(1, min(1000, ctx.cog.cache_beginner["total_count"]) // 100)
            url += f"&page={page}"
    else:
        url = URL
        if ctx.cog.cache_normal is not None:
            page = random.randint(1, min(1000, ctx.cog.cache_normal["total_count"]) // 100)
            url += f"&page={page}"

    log.debug(f"making api request to url: {url}")
    async with ctx.bot.http_session.get(url, headers=ISSUES_REQUEST_HEADERS) as response:
        if response.status != 200:
            log.error(f"expected 200 status (got {response.status}) by the GitHub api.")
            await ctx.send(
                f"ERROR: expected 200 status (got {response.status}) by the GitHub api.\n"
                f"{await response.text()}"
            )
            return None
        data = await response.json()

        if len(data["items"]) == 0:
            log.error(f"no issues returned by GitHub API, with url: {response.url}")
            await ctx.send(f"ERROR: no issues returned by GitHub API, with url: {response.url}")
            return None

        if option == "beginner":
            ctx.cog.cache_beginner = data
            ctx.cog.cache_timer_beginner = ctx.message.created_at.replace(tzinfo=None)
        else:
            ctx.cog.cache_normal = data
            ctx.cog.cache_timer_normal = ctx.message.created_at.replace(tzinfo=None)

        return data


def format_issues_embed(issue: dict) -> discord.Embed:
    """Format the issue data into a embed."""
    title = issue["title"]
    issue_url = issue["url"].replace("api.", "").replace("/repos/", "/")
    # issues can have empty bodies, which in that case GitHub doesn't include the key in the API response
    body = issue.get("body", "")
    labels = [label["name"] for label in issue["labels"]]

    embed = discord.Embed(title=title)
    embed.description = body[:500] + "..." if len(body) > 500 else body
    embed.add_field(name="labels", value="\n".join(labels))
    embed.url = issue_url
    embed.set_footer(text=issue_url)

    return embed