code:
import asyncio
from datetime import datetime, timedelta
from typing import List, Tuple

import discord
from motor.motor_asyncio import AsyncIOMotorCollection

from .handler import command
from ..config import config
from ..utils import format_name, character_permissions_check

N_DAYS = 7
CONFIRM_EMOJI = "✅"
CANCEL_EMOJI = "❎"

CONFIRM_PROMPT = """
**[Confirmation Requested]**
Characters with credited authors and artists that are still active have protected play stats.
This information should not be shared publically!

Exceptions are made for characters that have not been updated in 12+ months.
Are you sure you have permission to display statistics for **{character}** in this channel?
(This prompt will auto-cancel in 30 seconds by default.)
"""

LOCKOUT_ROLE_ID: int = 701460498708955166

def get_date_range() -> List[datetime]:
    """Get a list of date boundaries stretching back N_DAYS, in ascending order.

    The first list element is the inclusive lower end of the date range.
    The last list element is the exclusive upper end of the date range.
    """
    upper_bound = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
    ret = [upper_bound - timedelta(days=N_DAYS - i) for i in range(N_DAYS)]
    ret.append(upper_bound)
    return ret

def construct_match_filter(
    dates: List[datetime], report_types: List[str], character: str
) -> dict:
    """Construct the query for the first stage of the overall aggregation pipeline.
    """
    return {
        "date": {"$gte": dates[0], "$lt": dates[-1]},
        "type": {"$in": report_types},
        "$or": [
            {"table.1": character},
            {"table.2": character},
            {"table.3": character},
            {"table.4": character},
        ],
    }

def construct_facet_subpipeline(dates: List[datetime], report_type: str) -> List[dict]:
    """Construct the subpipeline for each part of the $facet second stage in the
    aggregation pipeline.
    """
    return [
        {"$match": {"type": report_type}},
        {
            "$bucket": {
                "groupBy": "$date",
                "boundaries": list(dates),
                "output": {"count": {"$sum": 1},},
            }
        },
    ]

def construct_pipeline(character: str) -> List[dict]:
    """Construct the overall aggregation pipeline used for the playstats command.
    """
    dates = get_date_range()

    return [
        {
            "$match": construct_match_filter(
                dates, ["start_game", "end_game"], character
            )
        },
        {
            "$facet": {
                "start_game": construct_facet_subpipeline(dates, "start_game"),
                "end_game": construct_facet_subpipeline(dates, "end_game"),
            }
        },
    ]

async def count_totals(coll: AsyncIOMotorCollection) -> Tuple[int, int]:
    """Count total number of games started and ended over the recent stats window."""
    dates = get_date_range()

    pipeline = [
        {
            "$match": {
                "date": {"$gte": dates[0], "$lt": dates[-1]},
                "type": {"$in": ["start_game", "end_game"]},
            }
        },
        {"$group": {"_id": "$type", "count": {"$sum": 1},},},
    ]

    results = {}
    async for doc in coll.aggregate(pipeline):
        results[doc["_id"]] = doc["count"]

    return (results.get("start_game", 0), results.get("end_game", 0))

def get_bucket_key(d: datetime) -> Tuple[int, int, int]:
    return (d.year, d.month, d.day)

async def do_aggregation(
    character: str, coll: AsyncIOMotorCollection
) -> List[Tuple[Tuple[int, int, int], int, int]]:
    cursor = coll.aggregate(construct_pipeline(character))
    doc = (await cursor.to_list(1))[0]

    # Need to be a bit careful here-- the each of the second-stage $bucket
    # operations only returns info for buckets that contain at least one element.
    #
    # We need to make sure that we actually match together each bucket document
    # based on its "_id" attribute, instead of just zipping together the two arrays
    # of buckets.

    buckets = {}

    for bucket in doc["start_game"]:
        buckets[get_bucket_key(bucket["_id"])] = [bucket["count"], 0]

    for bucket in doc["end_game"]:
        key = get_bucket_key(bucket["_id"])
        if key in buckets:
            buckets[key][1] = bucket["count"]
        else:
            buckets[key] = [0, bucket["count"]]

    # Convert the buckets dict into a list of (key, start_game, end_game)
    buckets = [(k, v[0], v[1]) for (k, v) in buckets.items()]

    # Sort on keys:
    return sorted(buckets, key=lambda t: t[0])

def format_bucket_rate(n_finished: int, days=1, desc="game") -> str:
    if n_finished == 0:
        return "no rate information"

    interval = timedelta(days=days) / n_finished
    total_seconds = interval.total_seconds()
    timestr = ""

    if total_seconds >= 3570:  # (3600 - 30) seconds = 59.5 minutes
        hours, r = divmod(interval, timedelta(hours=1))
        minutes = round(r / timedelta(minutes=1))

        if minutes == 60:
            hours += 1
            minutes = 0

        timestr = str(hours) + " hour"
        if hours != 1:
            timestr += "s"

        if minutes > 0:
            timestr += ", {} minute".format(minutes)
            if minutes != 1:
                timestr += "s"
    elif total_seconds > 59.5:
        minutes = round(interval / timedelta(minutes=1))
        timestr = str(minutes) + " minute"
        if minutes != 1:
            timestr += "s"
    else:
        seconds = round(interval.total_seconds())
        timestr = str(seconds) + " second"
        if seconds != 1:
            timestr += "s"

    return "1 " + desc + " every " + timestr

def format_bucket(bucket: Tuple[Tuple[int, int, int], int, int]) -> str:
    return "`{:04d}-{:02d}-{:02d}:` {:s} (and {:s})".format(
        *(bucket[0]),
        format_bucket_rate(bucket[2], desc="completion"),
        format_bucket_rate(bucket[1], desc="started"),
    )

def format_lead_bucket(bucket: Tuple[Tuple[int, int, int], int, int]) -> str:
    return "`{:04d}-{:02d}-{:02d}:` {:s} (and {:s})".format(
        *(bucket[0]),
        format_bucket_rate(bucket[2], desc="game completed"),
        format_bucket_rate(bucket[1], desc="game started"),
    )

async def do_confirmation(
    client: discord.Client,
    channel: discord.TextChannel,
    confirm_user: discord.Member,
    confirmation_msg: str,
) -> bool:
    confirm_msg: discord.Message = await channel.send(confirmation_msg)
    await confirm_msg.add_reaction(CONFIRM_EMOJI)
    await confirm_msg.add_reaction(CANCEL_EMOJI)

    def check(reaction, user):
        return (
            user == confirm_user
            and reaction.message.id == confirm_msg.id
            and (
                str(reaction.emoji) == CONFIRM_EMOJI
                or str(reaction.emoji) == CANCEL_EMOJI
            )
        )

    try:
        reaction, _ = await client.wait_for("reaction_add", timeout=30.0, check=check)
        return str(reaction.emoji) == CONFIRM_EMOJI
    except asyncio.TimeoutError:
        return False
    finally:
        await confirm_msg.delete()

@command(
    "playstats", summary="Get play statistics for a character.",
)
async def cmd_playstats(client: discord.Client, msg: discord.Message, args: List[str]):
    character = args[0].casefold()
    character_name = format_name(character)

    if not isinstance(msg.channel, discord.DMChannel):
        return await client.reply(msg, "This command can only be used via DMs.",)

    authorized = await character_permissions_check(client, character, msg.author.id)

    # Nothing to see here, folks.
    dev_server = client.get_guild(config.primary_server_id)
    user: discord.Member = await dev_server.fetch_member(msg.author.id)
    if LOCKOUT_ROLE_ID is not None and user is not None:
        authorized = authorized and not any(
            role.id == LOCKOUT_ROLE_ID for role in user.roles
        )

    if not authorized:
        await client.error_notify(
            "Unauthorized user {}#{} attempted to get play stats for character `{}`".format(
                msg.author.name, msg.author.discriminator, character
            )
        )

        return await client.reply(
            msg, "You don't have authorization for that command.",
        )

    await client.log_notify(
        "User {}#{} retrieved play stats for character `{}`".format(
            msg.author.name, msg.author.discriminator, character
        )
    )

    buckets = await do_aggregation(character, client.db.usage_reports)
    if all((t[1] == 0) and (t[2] == 0) for t in buckets):
        return await client.reply(
            msg,
            "I don't have any recent usage reports for `{}`. Did you spell their ID correctly?".format(
                character
            ),
        )

    total_game_counts = await count_totals(client.db.usage_reports)
    character_completed_sum = 0
    character_started_sum = 0

    for _, started, ended in buckets:
        character_completed_sum += ended
        character_started_sum += started

    reply_msg = "**[Recent Play Statistics for {}]**\n".format(character_name)
    reply_msg += format_lead_bucket(buckets[0])
    if len(buckets) > 1:
        reply_msg += "\n" + "\n".join(map(format_bucket, buckets[1:]))
    reply_msg += "\n`     Total:` {:s} / {:s}".format(
        format_bucket_rate(character_completed_sum, days=7, desc="completion"),
        format_bucket_rate(character_started_sum, days=7, desc="started"),
    )
    reply_msg += "\nData collected from {:,d} complete games (out of {:,d} total games started) over the past {:d} days.".format(
        total_game_counts[1], total_game_counts[0], N_DAYS
    )

    return await msg.channel.send(reply_msg)


code:
import asyncio
from pathlib import Path

from .handler import command
from ..config import config
from ..git import git_ops
from ..git.info import get_incomplete_jobs
from ..utils import get_http_session

cached_extensions = [
    "bmp",
    "ejs",
    "jpeg",
    "pdf",
    "ps",
    "ttf",
    "class",
    "eot",
    "jpg",
    "pict",
    "svg",
    "webp",
    "css",
    "eps",
    "js",
    "pls",
    "svgz",
    "woff",
    "csv",
    "gif",
    "mid",
    "png",
    "swf",
    "woff2",
    "doc",
    "ico",
    "midi",
    "ppt",
    "tif",
    "xls",
    "docx",
    "jar",
    "otf",
    "pptx",
    "tiff",
    "xlsx",
]

async def cloudflare_purge_files(client, files):
    headers = {
        "X-Auth-Email": config.cf_auth_email,
        "X-Auth-Key": config.cf_apikey,
        "Content-Type": "application/json",
    }

    endpoint = (
        "https://api.cloudflare.com/client/v4/zones/"
        + config.cf_zone_id
        + "/purge_cache"
    )
    payloads = []

    current_file_list = []
    for file in files:
        if file[0] == "/":
            current_file_list.append("https://spnati.net/" + file[1:])
        else:
            current_file_list.append("https://spnati.net/" + file)

        if len(current_file_list) == 30:
            payloads.append({"files": current_file_list.copy()})
            current_file_list = []

    payloads.append({"files": current_file_list.copy()})

    responses = []
    session = get_http_session()

    for payload in payloads:
        async with session.post(endpoint, headers=headers, json=payload) as resp:
            responses.append(await resp.json())

    return responses

@command(
    "purge_cache",
    summary="Purge files from the Cloudflare cache.",
    authorized_only=True,
)
async def cmd_purge_files(client, msg, args):
    purge_target = args[0]
    force_purge = False

    if len(args) > 1:
        force_purge = args[1] == "force"

    if not force_purge:
        incomplete_jobs = await get_incomplete_jobs(client, config.upstream_project_id)
        for job in incomplete_jobs:
            if job["stage"] == "deploy" and job["ref"] == "master":
                return await client.reply(
                    msg,
                    "Error: a deployment job is currently pending on `master`. Try again once that completes (usually in 5-10 minutes).",
                )

    if purge_target == "core":
        purge_list = [
            "css/spni.css",
            "css/spniDialogueUtilities.css",
            "js/save.js",
            "js/player.js",
            "js/table.js",
            "js/spniAI.js",
            "js/spniBehaviour.js",
            "js/spniClothing.js",
            "js/spniCore.js",
            "js/spniDevMode.js",
            "js/spniDisplay.js",
            "js/spniEpilogue.js",
            "js/spniForfeit.js",
            "js/spniGallery.js",
            "js/spniGame.js",
            "js/spniOption.js",
            "js/spniPoker.js",
            "js/spniSelect.js",
            "js/spniTitle.js",
        ]
    else:
        purge_list = []

        await msg.channel.send("    - Listing opponent files...")

        if git_ops.GIT_LOCK.locked():
            await msg.channel.send(
                "    - Waiting for other Git operations to complete first (this won't take long)..."
            )

        async with git_ops.GIT_LOCK:
            await git_ops.reset_ops_repo()
            opp_dir = Path(config.ops_repo_dir).joinpath("opponents", purge_target)

            for p in opp_dir.rglob("*"):
                if p.is_file() and p.suffix[1:] in cached_extensions:
                    purge_list.append(str(p.relative_to(config.ops_repo_dir)))

    await msg.channel.send(
        "    - Purging **" + str(len(purge_list)) + "** files from CF cache..."
    )

    responses = await cloudflare_purge_files(client, purge_list)
    n_total = 0
    n_success = 0
    errors = []
    for response in responses:
        n_total += 1

        if response["success"]:
            n_success += 1
        else:
            errors.extend(response["errors"])

    await client.reply(
        msg, "**{}** out of **{}** requests succeeded.".format(n_success, n_total)
    )

    if len(errors) > 0:
        await msg.channel.send("**Errors received:**")
        for error in errors:
            await msg.channel.send(
                "Code **{}**: `{}`".format(error["code"], error["message"])
            )

repo_name: spnati/serve

license: mit


code:
import io
import csv
from datetime import datetime, timedelta, date, time
import os.path as osp
import zipfile

from bson import ObjectId
import discord

from .. import linecount
from .. import utils
from .handler import command
from ..config import config
from ..reporting import formatting
from ..reporting.routing import is_feedback_enabled

async def get_usage_reports(db_client, start_dt, end_dt, report_type=None):
    query = {"date": {"$gte": start_dt, "$lte": end_dt}}

    if report_type is not None:
        query["type"] = report_type

    cursor = db_client.usage_reports.find(query)
    cursor.sort("date")

    buf = io.StringIO(newline="")

    writer = csv.DictWriter(
        buf,
        [
            "date",
            "type",
            "commit",
            "session",
            "game",
            "userAgent",
            "origin",
            "ip_hash",
            "table-1",
            "table-2",
            "table-3",
            "table-4",
            "winner",
            "chosen-id",
            "chosen-title",
        ],
        extrasaction="ignore",
    )

    writer.writeheader()

    n_docs = 0
    async for doc in cursor:
        row = {
            "date": doc["date"].isoformat(),
            "type": doc["type"],
            "commit": doc.get("commit", ""),
            "session": doc["session"],
            "ip_hash": doc.get("ip", ""),
            "game": doc.get("game", ""),
            "userAgent": doc["userAgent"],
            "origin": doc["origin"],
            "winner": doc.get("winner", ""),
        }

        if doc["table"] is not None:
            row["table-1"] = doc["table"].get("1", "")
            row["table-2"] = doc["table"].get("2", "")
            row["table-3"] = doc["table"].get("3", "")
            row["table-4"] = doc["table"].get("4", "")
        else:
            row["table-1"] = ""
            row["table-2"] = ""
            row["table-3"] = ""
            row["table-4"] = ""

        if "chosen" in doc and doc["chosen"] is not None:
            row["chosen-id"] = doc["chosen"].get("id", "")
            row["chosen-title"] = doc["chosen"].get("title", "")

        writer.writerow(row)
        n_docs += 1

    buf.seek(0, io.SEEK_SET)

    return buf

async def get_bug_reports(db_client, start_dt, end_dt, report_type, character):
    query = {
        "date": {"$gte": start_dt, "$lte": end_dt},
        "type": report_type,
        "status": {"$ne": "spam"},
    }
    if character is not None:
        query["character"] = character

    cursor = db_client.bug_reports.find(query)

    cursor.sort("date")

    buf = io.StringIO(newline="")

    writer = csv.DictWriter(
        buf,
        [
            "id",
            "date",
            "status",
            "session",
            "commit",
            "userAgent",
            "origin",
            "ip_hash",
            "type",
            "character",
            "description",
            "js_errors",
            "table-1",
            "table-2",
            "table-3",
            "table-4",
        ],
    )

    writer.writeheader()

    async for doc in cursor:
        row = {
            "id": str(doc["_id"]),
            "date": doc["date"].isoformat(),
            "status": doc.get("status", "open"),
            "commit": doc.get("commit", ""),
            "session": doc["session"],
            "ip_hash": doc.get("ip", ""),
            "userAgent": doc["circumstances"]["userAgent"],
            "origin": doc["circumstances"]["origin"],
            "type": doc["type"],
            "character": doc.get("character", ""),
            "description": doc["description"],
            "js_errors": len(doc["jsErrors"]),
            "table-1": "",
            "table-2": "",
            "table-3": "",
            "table-4": "",
        }

        if doc["table"] is not None:
            for pl in doc["table"]:
                if pl is not None:
                    row["table-" + str(pl["slot"])] = pl["id"]

        writer.writerow(row)

    return buf

def get_date_range(n_days=7):
    end_dt = datetime.combine(
        date.today() + timedelta(days=1), time()
    )  # midnight tomorrow
    start_dt = datetime.combine(
        date.today() - timedelta(days=n_days), time()
    )  # midnight n_days ago

    return start_dt, end_dt

def deflate_textfile(filename, strbuf):
    bio = io.BytesIO()
    with zipfile.ZipFile(
        bio, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=9
    ) as zf:
        with zf.open(filename + ".csv", "w") as outfile:
            outfile.write(strbuf.getvalue().encode("utf-8"))
            strbuf.close()

    sz = bio.tell()
    bio.seek(0, io.SEEK_SET)

    return bio, sz

@command("usage", summary="Get raw usage report data.", authorized_only=True)
async def cmd_list_usage_reports(client, msg, args):
    if msg.author.id not in config.authorized_users:
        return await client.reply(msg, "You are not authorized to access this command.")

    if (msg.channel.id not in config.listen_channels) and not isinstance(
        msg.channel, discord.DMChannel
    ):
        return

    await client.log_notify(
        "Delivering usage report data to user {}#{} (`{}`)".format(
            msg.author.name, msg.author.discriminator, msg.author.id
        )
    )

    n_days = 7
    report_type = None
    if len(args) >= 1:
        n_days = int(args[0])

    if len(args) >= 2:
        report_type = args[1]

    start_dt, end_dt = get_date_range(n_days)
    strbuf = await get_usage_reports(client.db, start_dt, end_dt, report_type)

    if report_type is None:
        filename = "Usage_Data_{}_to_{}".format(
            start_dt.strftime("%Y-%m-%d"), end_dt.strftime("%Y-%m-%d")
        )
    else:
        filename = "{}_Event_Data_{}_to_{}".format(
            report_type, start_dt.strftime("%Y-%m-%d"), end_dt.strftime("%Y-%m-%d")
        )

    bio, sz = deflate_textfile(filename, strbuf)

    if sz > (7.5 * (10 ** 6)):  # 7.5 MB
        with open(osp.join(config.downloads_dir, filename + ".zip"), "wb") as f:
            f.write(bio.read())

        await client.reply(
            msg,
            "Here's your usage report summary!\n\nhttps://spnati.faraway-vision.io/dl/"
            + filename
            + ".zip",
        )
    else:
        await client.reply(
            msg,
            "Here's your usage report summary!",
            file=discord.File(bio, filename + ".zip"),
        )

    bio.close()
    strbuf.close()

@command("feedback-summary", summary="Get raw feedback report data.")
async def cmd_list_feedback_reports(client, msg, args):
    n_days = 7
    char = None

    if len(args) >= 1:
        n_days = int(args[0])

    if len(args) >= 2:
        char = args[1]

    await client.log_notify(
        "Delivering feedback report data to user {}#{} (`{}`)".format(
            msg.author.name, msg.author.discriminator, msg.author.id
        )
    )

    start_dt, end_dt = get_date_range(n_days)
    strbuf = await get_bug_reports(client.db, start_dt, end_dt, "feedback", char)

    filename = "Feedback_Reports_{}_to_{}.csv".format(
        start_dt.strftime("%Y-%m-%d"), end_dt.strftime("%Y-%m-%d")
    )

    bio, sz = deflate_textfile(filename, strbuf)

    if sz > (7.5 * (10 ** 6)):  # 7.5 MB
        with open(osp.join(config.downloads_dir, filename + ".zip"), "wb") as f:
            f.write(bio.read())

        await client.reply(
            msg,
            "Here's your feedback report summary!\n\nhttps://spnati.faraway-vision.io/dl/"
            + filename
            + ".zip",
        )
    else:
        await client.reply(
            msg,
            "Here's your feedback report summary!",
            file=discord.File(bio, filename + ".zip"),
        )

@command("bug-summary", summary="Get raw bug report data.")
async def cmd_list_bug_reports(client, msg, args):
    n_days = 7
    char = None
    report_type = {"$ne": "feedback"}

    if len(args) >= 1:
        n_days = int(args[0])

    if len(args) >= 2:
        report_type = args[1]

    if len(args) >= 3:
        char = args[2]

    await client.log_notify(
        "Delivering bug report data to user {}#{} (`{}`)".format(
            msg.author.name, msg.author.discriminator, msg.author.id
        )
    )

    start_dt, end_dt = get_date_range(n_days)
    strbuf = await get_bug_reports(client.db, start_dt, end_dt, report_type, char)

    filename = "Bug Reports_{}_to_{}.csv".format(
        start_dt.strftime("%Y-%m-%d"), end_dt.strftime("%Y-%m-%d")
    )

    bio, sz = deflate_textfile(filename, strbuf)

    if sz > (7.5 * (10 ** 6)):  # 7.5 MB
        with open(osp.join(config.downloads_dir, filename + ".zip"), "wb") as f:
            f.write(bio.read())

        await client.reply(
            msg,
            "Here's your bug report summary!\n\nhttps://spnati.faraway-vision.io/dl/"
            + filename
            + ".zip",
        )
    else:
        await client.reply(
            msg,
            "Here's your bug report summary!",
            file=discord.File(bio, filename + ".zip"),
        )

@command("bug", summary="Get full details for a bug report.")
async def cmd_get_detailed_bug_report(client, msg, args):
    if len(args) < 1:
        return await client.reply(msg, "USAGE: `b!bug [report id]`")

    await client.log_notify(
        "Delivering bug report details for bug {} to user {}#{} (`{}`)".format(
            args[0], msg.author.name, msg.author.discriminator, msg.author.id
        )
    )

    doc = await client.db.bug_reports.find_one({"_id": ObjectId(args[0])})

    report = await formatting.format_detailed_bug_report(client, doc)
    if report is None:
        await client.reply(msg, "That bug report does not exist!")
    else:
        if len(report) < 1950:
            await client.reply(msg, "```" + report + "```")
        else:
            with io.BytesIO(report.encode("utf-8")) as bio:
                await client.reply(
                    msg,
                    "The bug report is too long to fit into Discord as a message:",
                    file=discord.File(bio, "bug-report-{}.txt".format(args[0])),
                )

@command("feedback", summary="Configure feedback for a character.")
async def cmd_configure_feedback(client, msg, args):
    if (
        msg.author.bot
        or msg.channel.guild is None
        or msg.channel.guild.id != config.primary_server_id
    ):
        return

    if len(args) < 1:
        return await client.reply(
            msg,
            "**USAGE:** `b!feedback [enable|disable|message] [optional message...]`",
        )

    args = msg.content.split(" ", 2)
    args = list(args[1:])

    subcmd = args[0].lower()
    feedback_msg = " ".join(args[1:])

    dev_server = client.get_guild(config.primary_server_id)
    user = await dev_server.fetch_member(msg.author.id)
    if user is None:
        return

    roster = await linecount.get_roster("master")
    norm_channel_name = utils.normalize_channel_name(msg.channel.name)
    character = None

    for c in roster:
        if utils.normalize_folder_name(c) == norm_channel_name:
            character = c
            break
    else:
        return await client.reply(
            msg, "Could not identify character for channel " + msg.channel.name,
        )

    # Check permissions if necessary:
    if (subcmd != "status") and not (
        await utils.character_permissions_check(client, character, msg.author.id)
    ):
        await client.error_notify(
            "Unauthorized user {}#{} attempted to update feedback settings for character {} in <#{}>".format(
                user.name, user.discriminator, character, msg.channel.id
            )
        )

        return await client.reply(
            msg,
            "You don't have authorization for that command. Do you have a role for that character?",
        )

    feedback_enabled = await is_feedback_enabled(client, character)

    if subcmd == "enable":
        await client.redis.hset("feedback_data:" + character, "enabled", 1)

        if not feedback_enabled:
            if len(feedback_msg.strip()) > 0:
                await client.redis.hset(
                    "feedback_data:" + character, "message", feedback_msg
                )

                return await client.reply(
                    msg, "In-game feedback enabled and message changed."
                )

            return await client.reply(msg, "In-game feedback enabled.")
        else:
            if len(feedback_msg.strip()) > 0:
                await client.redis.hset(
                    "feedback_data:" + character, "message", feedback_msg
                )

                return await client.reply(
                    msg,
                    "In-game feedback message changed. (Feedback was already enabled.)",
                )

            return await client.reply(msg, "In-game feedback is already enabled!")
    elif subcmd == "disable":
        await client.redis.hset("feedback_data:" + character, "enabled", 0)

        if not feedback_enabled:
            return await client.reply(msg, "In-game feedback is already disabled!")
        else:
            return await client.reply(msg, "In-game feedback disabled.")
    elif subcmd == "message":
        if len(feedback_msg.strip()) > 0:
            await client.redis.hset(
                "feedback_data:" + character, "message", feedback_msg
            )

            if not feedback_enabled:
                return await client.reply(
                    msg,
                    "In-game feedback message changed. (Warning: feedback is currently disabled.)",
                )
            else:
                return await client.reply(msg, "In-game feedback message changed.")
        else:
            return await client.reply(msg, "You must supply a feedback message!")
    elif subcmd == "status":
        if feedback_enabled:
            reply = "Feedback status: Enabled\n\n"
        else:
            reply = "Feedback status: Disabled\n\n"

        feedback_message = await client.redis.hget(
            "feedback_data:" + character, "message", encoding="utf-8",
        )

        if feedback_message is None or feedback_message == "":
            if feedback_enabled:
                reply += "Feedback message:\n`(no message set)`"
        else:
            reply += "Feedback message:\n`{}`".format(feedback_message)

        return await client.reply(msg, reply)
    else:
        return await client.reply(msg, "Unrecognized subcommand: `{}`".format(subcmd))


code:
from datetime import datetime, timedelta

from .handler import command
from ..reporting import silence, reactions
from ..config import config

@command(
    "silence",
    summary="Silence a user, preventing them from sending bug reports. Additionally marks all of their previous reports from the past 24 hours as spam.",
    authorized_only=True,
)
async def cmd_silence(client, msg, args):
    if msg.author.id not in config.authorized_users:
        return await client.reply(msg, "You are not authorized to access this command.")

    if len(args) < 1:
        return await client.reply(
            msg, "Usage: `silence <ip_hash> [n_hours] [clear_hours]`"
        )

    ip_hash = args[0]
    n_hours = 2
    clear_hours = 24

    if len(args) >= 2:
        try:
            n_hours = int(args[1])
        except ValueError:
            return await client.reply(msg, "{} is not a valid number!".format(args[1]))

    if len(args) >= 3:
        try:
            clear_hours = int(args[2])
        except ValueError:
            return await client.reply(msg, "{} is not a valid number!".format(args[2]))

    await silence.silence_reports_from_ip(
        client, ip_hash, n_hours, msg.author
    )  # noqa: F841

    await client.reply(
        msg,
        "Reports from IP-Hash `{}` have been silenced for {} hours.".format(
            ip_hash, n_hours
        ),
    )

    # find all previous bug reports from this IP:
    end_dt = datetime.utcnow()
    start_dt = end_dt - timedelta(hours=clear_hours)
    cursor = client.db.bug_reports.find(
        {"ip": ip_hash, "date": {"$gte": start_dt, "$lte": end_dt}}
    )

    n = 0
    async for doc in cursor:
        if doc.get("status", "open") == "open":
            await reactions.set_bug_report_status(
                client, doc["_id"], "spam", msg.author
            )
            n += 1

    await msg.channel.send(
        "Closed {} previous reports from the past {} hours as spam.".format(
            n, clear_hours
        )
    )

@command(
    "unsilence",
    summary="Unsilence a user, allowing them to send bug reports again.",
    authorized_only=True,
)
async def cmd_unsilence(client, msg, args):
    if msg.author.id not in config.authorized_users:
        return await client.reply(msg, "You are not authorized to access this command.")

    if len(args) < 1:
        return await client.reply(msg, "Usage: `unsilence <ip_hash>`")

    await client.log_notify(
        "User {}#{} (`{}`) unsilenced IP-hash {}".format(
            msg.author.name, msg.author.discriminator, msg.author.id, args[0]
        )
    )

    ip_hash = args[0]
    await silence.unsilence_ip(client.redis, ip_hash)
    return await client.reply(
        msg, "Reports from IP-Hash `{}` are now unsilenced.".format(ip_hash)
    )


code:
from __future__ import annotations

from aiohttp import web
import discord
from typing import Optional, List, Union, Dict, Any

from ..interactions import InteractionHandler, InteractionContext
from ..config import config
from .. import utils, main

slash_commands: List[SlashCommandNode] = []

class SlashCommandOption(object):
    TYPE_ID: int = None

    def __init__(self, name: str, description: str, required: Optional[bool] = None):
        self.name: str = name
        self.description: str = description
        self.required: Optional[bool] = required

    def convert_value(self, _ctx: InteractionContext, _value: Any) -> Any:
        raise NotImplementedError()

    def serialize(self) -> dict:
        ret = {"name": self.name, "description": self.description, "type": self.TYPE_ID}
        if self.required is not None:
            ret["required"] = self.required
        return ret

class SlashCommandStringOption(SlashCommandOption):
    TYPE_ID = 3

    def convert_value(self, _ctx: InteractionContext, value: Any) -> str:
        return str(value)

class SlashCommandIntegerOption(SlashCommandOption):
    TYPE_ID = 4

    def convert_value(self, _ctx: InteractionContext, value: Any) -> int:
        return int(value)

class SlashCommandBooleanOption(SlashCommandOption):
    TYPE_ID = 5

    def convert_value(self, _ctx: InteractionContext, value: Any) -> bool:
        return bool(value)

class SlashCommandUserOption(SlashCommandOption):
    TYPE_ID = 6

    def convert_value(self, ctx: InteractionContext, value: Any) -> discord.Member:
        dev_server: discord.Guild = ctx.client.get_guild(config.primary_server_id)
        member = dev_server.get_member(int(value))

        if member is not None:
            return member
        else:
            return ctx.guild.get_member(int(value))

class SlashCommandChannelOption(SlashCommandOption):
    TYPE_ID = 7

    def convert_value(self, ctx: InteractionContext, value: Any) -> discord.TextChannel:
        return ctx.client.get_channel(int(value))

class SlashCommandRoleOption(SlashCommandOption):
    TYPE_ID = 8

    def convert_value(self, ctx: InteractionContext, value: Any) -> discord.Role:
        return ctx.guild.get_role(int(value))

class SlashCommandNode(object):
    def __init__(self, name: str, description: str):
        self.name: str = name
        self.description: str = description

    async def handler(self, ctx: InteractionContext, data: dict) -> web.Resource:
        raise NotImplementedError()

    def serialize(self) -> dict:
        raise NotImplementedError()

    async def register_guild(self, client: main.NotifierClient, guild: discord.Guild):
        endpoint = "https://discord.com/api/v8/applications/{}/guilds/{}/commands".format(
            client.user.id, guild.id
        )
        headers = {"Authorization": "Bot " + config.token}

        http_sess = utils.get_http_session()

        async with http_sess.post(
            endpoint, json=self.serialize(), raise_for_status=True, headers=headers
        ) as resp:
            return await resp.json()

    async def register_global(self, client: main.NotifierClient):
        endpoint = "https://discord.com/api/v8/applications/{}/commands".format(
            client.user.id
        )
        headers = {"Authorization": "Bot " + config.token}

        http_sess = utils.get_http_session()

        async with http_sess.post(
            endpoint, json=self.serialize(), raise_for_status=True, headers=headers
        ) as resp:
            return await resp.json()

class SlashCommandLeaf(SlashCommandNode):
    def __init__(
        self,
        name: str,
        description: str,
        func,
        *args,
        authorized_only=False,
        is_top_level=False
    ):
        super().__init__(name, description)
        self.options: Dict[str, SlashCommandOption] = {}
        self.func = func
        self.authorized_only: bool = authorized_only
        self.is_top_level: bool = is_top_level

        for arg in args:
            if not isinstance(arg, SlashCommandOption):
                raise TypeError(
                    "argument must be subclass of SlashCommandOption, not "
                    + str(type(arg).__name__)
                )
            self.options[arg.name] = arg

    def serialize(self) -> dict:
        ret = {
            "name": self.name,
            "description": self.description,
            "options": [opt.serialize() for opt in self.options.values()],
        }

        if not self.is_top_level:
            ret["type"] = 1
        return ret

    async def handler(self, ctx: InteractionContext, data: dict) -> web.Resource:
        if self.authorized_only and ctx.user.id not in config.authorized_users:
            return ctx.reply("You are not authorized to access that command.")

        args = {}
        for opt in data.get("options", []):
            name = opt["name"]
            val = self.options[name].convert_value(ctx, opt["value"])
            args[name] = val
        return await self.func(ctx, **args)

    def __call__(self, *args, **kwargs):
        return self.func(*args, **kwargs)

class SlashCommandGroup(SlashCommandNode):
    def __init__(
        self, name: str, description: str, authorized_only=False, is_subcmd_group=False
    ):
        super().__init__(name, description)
        self.children: Dict[str, SlashCommandNode] = {}
        self.authorized_only: bool = authorized_only
        self.is_subcmd_group: bool = is_subcmd_group

    async def handler(self, ctx: InteractionContext, data: dict) -> web.Resource:
        if self.authorized_only and ctx.user.id not in config.authorized_users:
            return ctx.reply("You are not authorized to access that command.")

        assert len(data["options"]) == 1
        opt = data["options"][0]
        return await self.children[opt["name"]].handler(ctx, opt)

    def subcommand_group(
        self, name: str, description: str, authorized_only=False
    ) -> SlashCommandGroup:
        if self.is_subcmd_group:
            raise RuntimeError("Subcommand groups cannot be nested.")

        group = SlashCommandGroup(
            name, description, authorized_only=authorized_only, is_subcmd_group=True
        )
        self.children[name] = group
        return group

    def subcommand(self, name: str, description: str, *args, authorized_only=False):
        def wrapper(func):
            leaf = SlashCommandLeaf(
                name, description, func, *args, authorized_only=authorized_only
            )
            self.children[name] = leaf
            return leaf

        return wrapper

    def serialize(self) -> dict:
        ret = {
            "name": self.name,
            "description": self.description,
            "options": [node.serialize() for node in self.children.values()],
        }

        if self.is_subcmd_group:
            ret["type"] = 2
        return ret

def slash_command(
    name: str, description: str, authorized_only=False
) -> SlashCommandGroup:
    global slash_commands
    group = SlashCommandGroup(name, description, authorized_only=authorized_only)
    slash_commands.append(group)
    return group

def single_slash_cmd(name: str, description: str, *args, authorized_only=False):
    global slash_commands

    def wrapper(func):
        ret = SlashCommandLeaf(
            name,
            description,
            func,
            *args,
            authorized_only=authorized_only,
            is_top_level=True
        )
        slash_commands.append(ret)

        return ret

    return wrapper


code:
import asyncio
from datetime import datetime, timedelta

import dateutil.parser

from .handler import command
from .. import utils
from ..config import config
from ..site import get_deployed_commit
from ..git.info import get_commit_data

help_text = """
status
- Get bot status.

usage [n_days] [type]
- Retrieve game usage reports from up to n_days ago, optionally filtering on report type. Mods only.

bug-summary [n_days] [type] [character]
- Get a summary of bug reports from up to n_days ago, for the given report type (optional) and character.
  DM only.

feedback-summary [n_days] [character]
- Get a summary of feedback reports from up to n_days ago for the given character. DM only.

bug [id]
- Get detailed information for the given bug report.

linecount [character]
- Get detailed linecount information for a given character.

target_info [character]
- Get character targeting info for a given character.

filter_info [character]
- Get tag targeting info for a given character.

inbound [character]
- List opponents that target a given character.

tags [character]
- List all tags for a given character.

pairs [character]
- Get pairing information for a given character.

playstats [character]
- Get play statistics for a given character. DM only.
    Restricted to users with the character's role.

update [character folder name] [update title]
- Open or update a merge request for updating a character.
    Restricted to users with the character's role.
    Be sure to send a zip file with the updated files, too!

feedback [enable|disable|message|status] [feedback message]
- Configure feedback for the character whose channel this is posted in.
    Restricted to users with the character's role.

sync [key|enable|disable|update] [GitGud username]
 - Configure automatic fork synchronization for the given GitGud username.

deploy-test [gitgud user] [branch name]
- Create a testing environment and deploy a given user's branch to it.

help                                    - Show this help message.

Commands can be prefixed with either 'b!' (i.e. b!status) or with an @-mention.
Commands sent over DM do not require a prefix.
"""

@command("help", summary="Get help with this bot.")
async def cmd_help(client, msg, args):
    return await client.reply_report(msg, "Commands List:", help_text)

@command("reload", summary="Reload the bot configuration.", authorized_only=True)
async def cmd_reload(client, msg, args):
    if msg.author.id not in config.authorized_users:
        return await client.reply(msg, "You are not authorized to access this command.")

    config.load()

CHECK_SITES = ["spnati.net"]

@command("site-status", "Get the current status of spnati.net.")
async def cmd_site_status(client, msg, args):
    deployed_commits = await asyncio.gather(
        *[get_deployed_commit(site) for site in CHECK_SITES]
    )

    commit_info = []
    for commit in deployed_commits:
        if commit is None:
            commit_info.append(None)
        else:
            commit_info.append(await get_commit_data(commit))

    report = ""
    for site, commit in zip(CHECK_SITES, commit_info):
        if commit is not None:
            dt = dateutil.parser.parse(commit["created_at"])
            report += '{:17s}: {} ("{}" - {} at {})\n'.format(
                site,
                commit["short_id"],
                commit["title"],
                commit["author_name"],
                dt.strftime("%c"),
            )
        else:
            report += "{:17s}: ERROR - Could not reach site!\n".format(site)

    return await client.reply_report(msg, "Deployed SPNATI versions:", report)

async def cmd_ping(client, msg, args):
    version = await utils.get_version()
    return await client.reply(msg, "Status: Active (Version {})".format(version))

report_tables = """
[ Report Counts ]
Report Type |  5min  |  1hr   |  1day  |  1wk
 Usage Data | {:^6d} | {:^6d} | {:^6d} | {:^6d}
   Bug Data | {:^6d} | {:^6d} | {:^6d} | {:^6d}

[ Report Rates per Minute ]
Report Type |  5min  |  1hr   |  1day  |  1wk
 Usage Data | {:^6.3f} | {:^6.3f} | {:^6.3f} | {:^6.3f}
  Bug Data  | {:^6.3f} | {:^6.3f} | {:^6.3f} | {:^6.3f}
"""

async def count_in_prev_timespan(collection, days=0, hours=0, minutes=0):
    now = datetime.utcnow()
    lim = now - timedelta(days=days, hours=hours, minutes=minutes)

    return await collection.count_documents(({"date": {"$gte": lim, "$lte": now}}))

@command("status", "Get the current status of this bot.")
async def cmd_status(client, msg, args):
    version = await utils.get_version()
    header = "Status: Active (Version {})".format(version)

    use_report_stats = await asyncio.gather(
        count_in_prev_timespan(client.db.usage_reports, minutes=5),
        count_in_prev_timespan(client.db.usage_reports, hours=1),
        count_in_prev_timespan(client.db.usage_reports, days=1),
        count_in_prev_timespan(client.db.usage_reports, days=7),
    )

    bug_report_stats = await asyncio.gather(
        count_in_prev_timespan(client.db.bug_reports, minutes=5),
        count_in_prev_timespan(client.db.bug_reports, hours=1),
        count_in_prev_timespan(client.db.bug_reports, days=1),
        count_in_prev_timespan(client.db.bug_reports, days=7),
    )

    use_report_rates = [
        use_report_stats[0] / 5,
        use_report_stats[1] / 60,
        use_report_stats[2] / (60 * 24),
        use_report_stats[3] / (60 * 24 * 7),
    ]

    bug_report_rates = [
        bug_report_stats[0] / 5,
        bug_report_stats[1] / 60,
        bug_report_stats[2] / (60 * 24),
        bug_report_stats[3] / (60 * 24 * 7),
    ]

    report = report_tables.format(
        *use_report_stats, *bug_report_stats, *use_report_rates, *bug_report_rates
    )

    if msg.author.id in config.authorized_users:
        uptime = await utils.get_cmd_output("uptime")
        free_out = await utils.get_cmd_output("free -m", False)

        svc_status = await asyncio.gather(
            utils.get_service_info("serve-repo"),
            utils.get_service_info("mongod"),
            utils.get_service_info("nginx"),
            utils.get_service_info("kisekae-server-v2"),
            utils.get_service_info("linecountd"),
            utils.get_service_info("tcnati"),
        )

        svc_report = ""

        for svc_data in svc_status:
            svc_report += "{:26s}: {:>20s} since {} - Main PID: {}\n".format(
                svc_data["Id"],
                "{} ({})".format(svc_data["ActiveState"], svc_data["SubState"]),
                svc_data.get("StateChangeTimestamp", "<unknown>"),
                svc_data.get("MainPID", "<unknown>"),
            )

        report += "\n[ System Uptime ]\n{}\n\n".format(uptime)
        report += "[ Memory Usage (MiB) ]\n{}\n".format(free_out)
        report += "[ Service Status ]\n{}".format(svc_report)

    return await client.reply_report(msg, header, report)


code:
import asyncio
import json
import os
import os.path as osp
import sys
import traceback
from typing import List

from .handler import command
from ..config import config
from ..git import git_ops

MIRROR_CFG_LOCK = asyncio.Lock()
SYNC_NICENESS = 10

# Called as preexec_fn
def set_sync_niceness():
    os.nice(SYNC_NICENESS)

def get_deploy_key_path(name):
    p = osp.realpath(osp.join(config.deploy_keys_dir, name))
    if osp.commonprefix((p, config.deploy_keys_dir)) != config.deploy_keys_dir:
        raise ValueError("Invalid username")

    return p

async def create_ssh_deploy_key(name):
    proc = await asyncio.create_subprocess_exec(
        "ssh-keygen",
        "-t",
        "rsa",
        "-N",
        "",
        "-C",
        "deploy-" + name + "@spnati.faraway-vision.io",
        "-f",
        get_deploy_key_path(name),
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        preexec_fn=set_sync_niceness,
    )

    stdout, stderr = await proc.communicate()
    stderr = stderr.decode("utf-8")
    stdout = stdout.decode("utf-8")

    if proc.returncode != 0:
        raise git_ops.CommandError(
            "ssh-keygen command exited abnormally with return code {}".format(
                proc.returncode
            ),
            stdout,
            stderr,
            "ssh-keygen",
            proc.returncode,
        )

    return proc.returncode, stdout

def deploy_key_exists(name):
    return osp.isfile(get_deploy_key_path(name))

def get_deploy_pubkey(name):
    with open(get_deploy_key_path(name + ".pub"), "r", encoding="utf-8") as f:
        return f.read()

def read_mirror_config():
    with open(config.mirror_config_path, "r", encoding="utf-8") as f:
        return json.load(f)

def write_mirror_config(cfg):
    with open(config.mirror_config_path, "w", encoding="utf-8") as f:
        return json.dump(cfg, f)

SYNC_INSTRUCTIONS = """
To allow the bot to synchronize your fork with the official repository,
add a new Deploy Key to your fork (at https://gitgud.io/{}/spnati/-/settings/repository) with the following settings:

    - **Title:** `Repository Mirroring`
    - **Write Access Allowed** (Very important!)
    - **Key:** ```{}```
"""

@command("sync", summary="Configure automatic git fork synchronization.")
async def cmd_sync(client, msg, args):
    if len(args) < 2:
        return await client.reply(
            msg, "USAGE: `b!sync [enable|disable|key|update] [GitGud Username]`"
        )

    op = args[0].lower()
    username = args[1]

    if username == "origin":
        return

    if op == "enable" or op == "key":
        if not deploy_key_exists(username):
            await msg.channel.send(
                "Creating a new SSH deploy key for `{}`...".format(username)
            )
            await create_ssh_deploy_key(username)

        if op == "enable":
            async with MIRROR_CFG_LOCK:
                cfg = read_mirror_config()
                if username not in cfg:
                    cfg.append(username)

                write_mirror_config(cfg)

            await msg.channel.send("Mirroring configuration updated successfully!")

        pubkey = get_deploy_pubkey(username)
        await msg.channel.send(SYNC_INSTRUCTIONS.format(username, pubkey))
    elif op == "disable":
        async with MIRROR_CFG_LOCK:
            cfg = read_mirror_config()
            if username in cfg:
                cfg.remove(username)
            write_mirror_config(cfg)

        await msg.channel.send("Mirroring configuration updated successfully!")
    elif op == "update":
        await msg.channel.send("Synchronizing Git fork for `{}`...".format(username))

        async with msg.channel.typing():
            proc = await asyncio.create_subprocess_exec(
                sys.executable,
                osp.join(os.getcwd(), "do_mirror_push.py"),
                username,
                cwd=config.mirror_repo_dir,
                stderr=asyncio.subprocess.PIPE,
                preexec_fn=set_sync_niceness,
            )

            _, stderr = await proc.communicate()

        if proc.returncode != 0:
            await msg.channel.send(
                "Synchronization failed. Did you set up the deploy key properly?"
            )
            await msg.channel.send("Error output: ```" + stderr.decode("utf-8") + "```")
        else:
            await msg.channel.send(
                "Successfully synchronized `{}/spnati.git` with official repository.".format(
                    username
                )
            )
    else:
        return await client.reply(
            msg, "USAGE: `b!sync [enable|disable|key|update] [GitGud Username]`"
        )

async def ipc_sync_handler(client, channel):
    while True:
        try:
            while await channel.wait_message():
                payload: List[str] = await channel.get_json()
                proc = await asyncio.create_subprocess_exec(
                    sys.executable,
                    osp.join(os.getcwd(), "do_mirror_push.py"),
                    *payload,
                    cwd=config.mirror_repo_dir,
                    stderr=asyncio.subprocess.PIPE,
                    preexec_fn=set_sync_niceness
                )

                _, stderr = await proc.communicate()
                if len(payload) == 0:
                    formatted_payload = "all users"
                else:
                    formatted_payload = ", ".join(formatted_payload)

                if proc.returncode != 0:
                    stderr = stderr.decode("utf-8")
                    sys.stderr.write(stderr)
                    sys.stderr.flush()

                    await client.error_notify(
                        "sync IPC request for {} failed".format(formatted_payload),
                        ext_data=stderr,
                        ext_data_name="stderr.log",
                    )
                else:
                    await client.log_notify(
                        "handled sync IPC request for " + formatted_payload
                    )
        except:
            traceback.print_exc()

async def init_ipc(client):
    res = await client.redis_sub.subscribe("utilities:sync")
    channel = res[0]

    tsk = asyncio.create_task(ipc_sync_handler(client, channel))
    return tsk


code:
import asyncio

from .handler import command
from .. import update_alerts

@command(
    "staleness-warnings", "Check current staleness warnings for Testing characters."
)
async def cmd_check_warnings(client, msg, args):
    lines = ["**Testing Staleness Warnings:**"]
    async for update_status in update_alerts.get_testing_update_statuses(client):
        if update_status.reached_threshold:
            lines.append(str(update_status))
    return await client.reply(msg, "\n".join(lines))


code:
from pathlib import Path
import re
import time
import urllib.parse
import xml.etree.ElementTree as ET

import aiohttp

from .handler import command
from ..config import config
from ..git import git_ops
from ..utils import get_http_session
from ..autoversion import prepare_version

semver_re = r"[vV]?(?P<major>\d+)\.(?P<minor>\d+)(?:\.(?P<patch>\d+))?(?:\-(?P<prerelease>[a-zA-Z0-9\-]+(?:\.[a-zA-Z0-9\-])*))?(?:\+(?P<build>[a-zA-Z0-9\-]+(?:\.[a-zA-Z0-9\-]+)*))?"

WINDOWS_LINE_ENDING = b"\r\n"
UNIX_LINE_ENDING = b"\n"

@command(
    "test-autoversion",
    summary="Test automatic version message generation",
    authorized_only=True,
)
async def cmd_test_autoversion(client, msg, args):
    if len(args) >= 2:
        start_ref = args[0]
        end_ref = args[1]
    else:
        end_ref = "master"
        start_ref = await git_ops.get_latest_tag("master")

    minor_bump, desc = await prepare_version(start_ref, end_ref)
    if len(desc) == 0:
        return await client.reply(msg, "No changes of note between those refs.")

    if minor_bump:
        minor_bump = "Yes"
    else:
        minor_bump = "No"

    await client.reply(
        msg,
        "**Minor Version Bump?:** {}, **Description:**```{}```".format(
            minor_bump, desc
        ),
    )

@command("version", summary="Increment the game version.", authorized_only=True)
async def cmd_increment_version(client, msg, args):
    if (
        msg.author.id not in config.authorized_users
        and msg.author.id not in config.version_authorized_users
    ):
        return await client.reply(msg, "You are not authorized to access this command.")

    args = msg.content.split(" ", 2)
    args = list(args[1:])

    if len(args) < 2:
        return await client.reply(
            msg, "USAGE: `b!version [minor|patch] [changelog text]`"
        )

    args[0] = args[0].strip().lower()
    if args[0] not in ["minor", "patch"]:
        return await client.reply(
            msg, "USAGE: `b!version [minor|patch] [changelog text]`"
        )

    session = get_http_session()
    async with session.get(
        "https://gitgud.io/api/v4/projects/{}/repository/files/{}/raw?ref=master".format(
            config.upstream_project_id, urllib.parse.quote("version-info.xml", safe="")
        ),
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        text = await resp.text()
        tree = ET.fromstring(text)

        cur_verstring = tree.find("current").get("version").strip()
        m = re.match(semver_re, cur_verstring)

        if m is None:
            return await client.reply(
                msg,
                "ERROR: could not parse current version string (`"
                + cur_verstring
                + "`)",
            )

        version = [int(m.group("major")), int(m.group("minor")), 0]
        try:
            version[2] = int(m.group("patch"))
        except IndexError:
            pass

        if args[0] == "minor":
            version[1] += 1
            version[2] = 0
        elif args[0] == "patch":
            version[2] += 1

        next_verstring = "{:d}.{:d}.{:d}".format(*version)
        timestamp = int(time.time() * 1000)

        tree.find("current").set("version", next_verstring)

        changelog_elem = ET.Element(
            "version", {"number": next_verstring, "timestamp": str(timestamp)}
        )
        changelog_elem.text = args[1]
        changelog_elem.tail = "\n        "

        tree.find("changelog").insert(0, changelog_elem)
        new_xml_data = ET.tostring(tree, encoding="utf-8")

        if git_ops.GIT_LOCK.locked():
            await client.reply(
                msg,
                "Waiting for other Git operations to complete first (this won't take long)...",
            )

        async with git_ops.GIT_LOCK:
            await git_ops.reset_ops_repo()

            base_path = Path(config.ops_repo_dir).resolve()
            new_xml_data = new_xml_data.replace(WINDOWS_LINE_ENDING, UNIX_LINE_ENDING)

            with base_path.joinpath("version-info.xml").open("wb") as f:
                f.write(new_xml_data)

            await git_ops.run_git_command(["add", "./version-info.xml"])
            await git_ops.commit_with_msg(
                "Update version number to v{}\n\nCommand executed by: {}#{}\n".format(
                    next_verstring, msg.author.name, msg.author.discriminator
                )
            )

            try:
                await git_ops.tag_with_msg("v" + next_verstring, args[1])
            except OSError:
                pass

            await git_ops.run_git_command(
                [
                    "push",
                    "--follow-tags",
                    "-u",
                    config.upstream_remote_name,
                    "master:master",
                ]
            )

            return await client.reply(
                msg, "Version number updated to v" + next_verstring
            )


code:
import json
from pathlib import Path
import os
import os.path as osp
import sys

class ConfigStore(object):
    def __init__(self):
        self.config = {
            "token": "",
            "git_apikey": "",
            "cf_apikey": "",
            "cf_auth_email": "",
            "cf_zone_id": "",
            "primary_redis_url": "redis://localhost:6379/0",
            "redis_pubsub_url": "redis://localhost:6379",
            "upstream_project_id": 0,
            "staging_project_id": 0,
            "bot_gitlab_id": 0,
            "upstream_remote_name": "origin",
            "staging_remote_name": "staging",
            "downloads_dir": "",
            "ops_repo_dir": "",
            "deploy_keys_dir": "",
            "mirror_repo_dir": "",
            "mirror_config_path": "",
            "primary_server_id": 0,
            "authorized_users": [],
            "version_authorized_users": [],
            "notification_channels": [],
            "logging_channels": [],
            "manual_bug_report_channels": [],
            "auto_bug_report_channels": [],
            "listen_channels": [],
            "report_channels": [],
            "monika_channels": [],
            "silenced_report_channels": [],
            "feedback_always_channels": [],
            "feedback_redirect_channels": [],
            "general_feedback_channels": [],
            "exclude_character_report_channels": [],
            "character_bug_reporting_enabled": True,
            "summon_prefixes": ["b!", "B!"],
            "masquerade": {},
            "masquerade_enabled": False,
            "spam_uncertain_threshold": 0.8,  # Threshold below which reports are flagged as possible spam
            "spam_likely_threshold": 0.5,  # Threshold below which reports are flagged as likely spam
            "spam_hard_filter": False,
            "interactions_public_key": "",
            "interactions_port": 8050,
        }

        self.load()

    def load(self):
        self.config_file = Path(os.environ["UTILITIES_BOT_CONFIG"]).resolve()

        with self.config_file.open("r", encoding="utf-8") as f:
            data = json.load(f)
            data["git_apikey"] = data["git_apikey"].strip()
            data["cf_apikey"] = data["cf_apikey"].strip()
            data["token"] = data["token"].strip()

            data["deploy_keys_dir"] = osp.realpath(data["deploy_keys_dir"])
            data["ops_repo_dir"] = osp.realpath(data["ops_repo_dir"])
            data["mirror_repo_dir"] = osp.realpath(data["mirror_repo_dir"])
            data["mirror_config_path"] = osp.realpath(data["mirror_config_path"])

            self.config.update(data)

    def save(self):
        with self.config_file.open("w", encoding="utf-8") as f:
            json.dump(self.config, f, indent=4)

    def __getattr__(self, name):
        return self.config[name]

    def __setattr__(self, name, value):
        if name == "config" or name == "config_file":
            object.__setattr__(self, name, value)
        else:
            self.config[name] = value
            self.save()

    def __delattr__(self, name):
        del self.config[name]
        self.save()

    def __dir__(self):
        return list(self.config.keys())

    def get(self, name, default):
        return self.config.get(name, default)

config = ConfigStore()


code:
from __future__ import annotations

from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Iterable, AsyncGenerator, Iterator, List, Optional, Tuple, Union

from . import git_ops
from ..config import config

class LogParseError(Exception):
    pass

class CommitNotFound(Exception):
    pass

class Commit:
    def __init__(
        self,
        obj_id: str,
        author: str,
        author_email: str,
        author_date: datetime,
        committer: str,
        committer_email: str,
        commit_date: datetime,
        subject: str,
        body: str,
    ):
        self.id: str = obj_id

        self.author: str = author
        self.author_email: str = author_email
        self.author_date: datetime = author_date

        self.committer: str = committer
        self.committer_email: str = committer_email
        self.commit_date: datetime = commit_date

        self.subject: str = subject
        self.body: str = body

class CommitReader:
    LOG_LINE_TYPES = {"id", "an", "ae", "ad", "cn", "ce", "cd"}
    LOG_FORMAT = "id:%H%nan:%aN%nae:%aE%nad:%ad%ncn:%cN%nce:%cE%ncd:%cd%n%s%n%b%ne:%H"

    def __init__(self, data: Union[str, bytes]):
        if isinstance(data, bytes):
            data = data.decode("utf-8")
        self._lines: List[str] = data.splitlines()
        self._iter: Iterator[str] = iter(self._lines)

    @staticmethod
    def _parse_raw_date(date_str: str) -> datetime:
        try:
            timestamp, offset = date_str.split(" ", 1)
            timestamp = int(timestamp)

            # offset[0] is always the sign (+/-)
            offset_hours = int(offset[1:3])
            offset_mins = int(offset[3:])
            delta = timedelta(hours=offset_hours, minutes=offset_mins)
            if offset[0] == "-":
                delta = -delta
            return datetime.fromtimestamp(timestamp, timezone(delta))
        except (ValueError, TypeError):
            raise LogParseError("Could not parse raw date string '" + date_str + "'")

    def _get_log_line(self, *, start: bool = False) -> str:
        for line in self._iter:
            line = line.strip()
            if len(line) == 0 and start:
                continue
            return line
        if start:
            raise StopIteration
        raise LogParseError("Expected commit log line, got unexpected EOF")

    def _expect_log_line(self, line_type: str, **kwargs) -> str:
        line = self._get_log_line(**kwargs)
        if not line[0:2] == line_type:
            if line[0:3] in CommitReader.LOG_LINE_TYPES:
                raise LogParseError(
                    "Expected line type '{}', got '{}'".format(line_type, line[0:2]),
                    line,
                )
            else:
                raise LogParseError(
                    "Expected line type '{}', got unknown line type (message line?)".format(
                        line_type
                    ),
                    line,
                )
        return line[3:]

    def __next__(self) -> Commit:
        # _expect_log_line with start=True raises StopIteration
        obj_id = self._expect_log_line("id", start=True)
        author_name = self._expect_log_line("an")
        author_email = self._expect_log_line("ae")
        author_time = CommitReader._parse_raw_date(self._expect_log_line("ad"))
        committer_name = self._expect_log_line("cn")
        committer_email = self._expect_log_line("ce")
        committer_time = CommitReader._parse_raw_date(self._expect_log_line("cd"))

        # NOTE: no empty line between subject and body
        subject = self._get_log_line()
        body = ""

        # look for this specific string to delimit end of message body
        end_line = "e:" + obj_id

        while True:
            line = self._get_log_line()
            if line == end_line:
                break
            body += line + "\n"
        body = body[:-1]  # remove trailing newline

        return Commit(
            obj_id,
            author_name,
            author_email,
            author_time,
            committer_name,
            committer_email,
            committer_time,
            subject,
            body,
        )

    def __iter__(self) -> CommitReader:
        return self

    @classmethod
    async def get_commits(
        cls, *refs: str, cwd: Union[str, Path] = None
    ) -> Tuple[Commit, ...]:
        """Get individual commits from a repository.

        Returns a tuple of retrieved commits.
        """
        try:
            _, data = await git_ops.run_git_command(
                [
                    "show",
                    "--format=" + cls.LOG_FORMAT,
                    "--date=raw",
                    "-s",
                    *refs,
                ],
                cwd=cwd,
            )
        except git_ops.CommandError as e:
            if e.stderr.endswith("unknown revision or path not in the working tree."):
                rev_start = e.stderr.rindex("'")
                rev_end = e.stderr.index("'")
                raise CommitNotFound(e.stderr[rev_start + 1 : rev_end])
            raise

        return tuple(cls(data))

    @classmethod
    async def get_latest_update(
        cls,
        *revs: str,
        filter_paths: Union[str, Path, Iterable[Union[str, Path]]] = "",
        cwd: Union[str, Path] = None
    ) -> Optional[Commit]:
        """Get the last commit touching a path or set of paths.

        Returns either the Commit in question, or None if no commits touch those paths.
        Note that merge commits are skipped.
        """
        if cwd is None:
            cwd = config.ops_repo_dir

        try:
            cmd_args = [
                "log",
                "--format=" + cls.LOG_FORMAT,
                "--date=raw",
                "-s",
                "--max-count=1",
                "--no-merges",
                *revs,
                "--",
            ]

            if isinstance(filter_paths, Path) or isinstance(filter_paths, str):
                cmd_args.append(str(filter_paths))
            else:
                cmd_args.extend(map(str, filter_paths))
            _, data = await git_ops.run_git_command(cmd_args, cwd=cwd)
        except git_ops.CommandError as e:
            if e.stderr.endswith("unknown revision or path not in the working tree."):
                rev_start = e.stderr.rindex("'")
                rev_end = e.stderr.index("'")
                raise CommitNotFound(e.stderr[rev_start + 1 : rev_end])
            raise

        data = data.strip()
        if len(data) == 0:
            return None

        try:
            return next(cls(data))
        except StopIteration:
            return None

    @classmethod
    async def get_log(
        cls,
        *revs: str,
        filter_paths: Union[str, Path, Iterable[Union[str, Path]]] = None,
        cwd: Union[str, Path] = None,
        max_per_call: int = 20,
        skip_merges: bool = False
    ) -> AsyncGenerator[Commit, None]:
        """Get a range of commits from a repository, using 'git log'.

        Can be iterated over (using 'async for') to retrieve commits.
        """
        if cwd is None:
            cwd = config.ops_repo_dir

        cur_commit_num = 0
        while True:
            try:
                cmd_args = [
                    "log",
                    "--format=" + cls.LOG_FORMAT,
                    "--date=raw",
                    "-s",
                    "--max-count=" + str(max_per_call),
                ]

                if cur_commit_num > 0:
                    cmd_args.append("--skip=" + str(cur_commit_num))

                if skip_merges:
                    cmd_args.append("--no-merges")

                cmd_args.extend(revs)

                if filter_paths is not None:
                    cmd_args.append("--")
                    if isinstance(filter_paths, Path) or isinstance(filter_paths, str):
                        cmd_args.append(str(filter_paths))
                    else:
                        cmd_args.extend(map(str, filter_paths))
                _, data = await git_ops.run_git_command(cmd_args, cwd=cwd)
            except git_ops.CommandError as e:
                if e.stderr.endswith(
                    "unknown revision or path not in the working tree."
                ):
                    rev_start = e.stderr.rindex("'")
                    rev_end = e.stderr.index("'")
                    raise CommitNotFound(e.stderr[rev_start + 1 : rev_end])
                raise

            for sub_idx, commit in zip(range(max_per_call), cls(data)):
                yield commit

            # If we read less than a full page from this call, stop iteration.
            if sub_idx < (max_per_call - 1):
                raise StopIteration

            cur_commit_num += max_per_call

class ChangedFile:
    EMPTY_MODE = 6 * b"0"
    EMPTY_HASH = 40 * b"0"

    def __init__(
        self,
        cwd: Path,
        main_line: Tuple[bytes, ...],
        src_path: bytes,
        dest_path: Optional[bytes],
    ):
        cwd = Path(cwd)

        self.src_mode: Optional[str] = (
            main_line[0].decode("utf-8")
            if main_line[0] != ChangedFile.EMPTY_MODE
            else None
        )
        self.dest_mode: Optional[str] = (
            main_line[1].decode("utf-8")
            if main_line[1] != ChangedFile.EMPTY_MODE
            else None
        )

        self.src_hash: Optional[str] = (
            main_line[2].decode("utf-8")
            if main_line[2] != ChangedFile.EMPTY_HASH
            else None
        )
        self.dest_hash: Optional[str] = (
            main_line[3].decode("utf-8")
            if main_line[3] != ChangedFile.EMPTY_HASH
            else None
        )

        self.status: str = main_line[4].decode("utf-8")
        self.score: Optional[int] = None
        if len(self.status) > 1:
            self.score = int(self.status[1:])
            self.status = self.status[0]

        self.src_path: Path = cwd.joinpath(src_path.decode("utf-8"))
        self.dest_path: Optional[Path] = (
            cwd.joinpath(dest_path.decode("utf-8")) if dest_path is not None else None
        )

    async def get_src_file(self) -> bytes:
        if self.src_hash is None:
            raise ValueError("Source object does not exist")
        return await git_ops.get_blob(self.src_hash)

    async def get_dest_file(self) -> bytes:
        if self.dest_hash is None:
            raise ValueError("Destination object does not exist")
        return await git_ops.get_blob(self.dest_hash)

    @classmethod
    def _read_diff_output(cls, cwd: Path, data: bytes) -> Iterable[ChangedFile]:
        for record in data.split(b":"):
            if len(record) == 0:
                continue

            parts = record.split(b"\x00")[:-1]
            main_line = parts[0].split(b" ")
            src_path = parts[1]

            if len(parts) == 3:
                dst_path = parts[2]
            else:
                dst_path = None

            yield cls(cwd, main_line, src_path, dst_path)

    @classmethod
    async def get_tree_diff(
        cls,
        tree_1: str,
        tree_2: str,
        *,
        merge_base: bool = False,
        cwd: Union[str, Path, None] = None,
        filter_paths: Union[str, Path, Iterable[Union[str, Path]], None] = None
    ) -> AsyncGenerator[ChangedFile, None]:
        if cwd is None:
            cwd = config.ops_repo_dir

        if merge_base:
            # Get the merge base for tree_1 vs tree_2
            _, base_commit = await git_ops.run_git_command(
                ["merge-base", tree_1, tree_2], cwd=cwd
            )
            tree_1 = base_commit.strip()

        cmd_args = ["diff-tree", "-r", "-z"]
        cmd_args.append(tree_1)
        cmd_args.append(tree_2)

        if filter_paths is not None:
            cmd_args.append("--")
            if isinstance(filter_paths, Path) or isinstance(filter_paths, str):
                cmd_args.append(str(filter_paths))
            else:
                cmd_args.extend(map(str, filter_paths))

        _, data = await git_ops.run_git_command(cmd_args, cwd=cwd, decode=False)
        for elem in cls._read_diff_output(cwd, data):
            yield elem


code:
from __future__ import annotations

import asyncio
import aioredis
import os.path as osp
import json
from typing import Tuple, Union

from . import info
from ..config import config

GIT_LOCK = asyncio.Lock()

class CommandError(Exception):
    def __init__(
        self, msg: str, stdout: str, stderr: str, command: str, return_code: int
    ) -> CommandError:
        super().__init__(msg, stdout, stderr, command, return_code)
        self.msg: str = msg
        self.stdout: str = stdout
        self.stderr: str = stderr
        self.command: str = command
        self.return_code: int = return_code

async def run_git_command(
    args, raise_on_nonzero=True, stdin=None, cwd=None, decode=True
) -> Tuple[int, Union[str, bytes]]:
    if cwd is None:
        cwd = config.ops_repo_dir

    proc = await asyncio.create_subprocess_exec(
        "git",
        *args,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=cwd
    )

    stdout, stderr = await asyncio.wait_for(proc.communicate(stdin), timeout=300.0)
    if decode:
        stdout = stdout.decode("utf-8")

    stderr = stderr.decode("utf-8")

    if raise_on_nonzero and proc.returncode != 0:
        raise CommandError(
            "git command `{}` exited abnormally with return code {}".format(
                " ".join(args), proc.returncode
            ),
            stdout,
            stderr,
            args[0],
            proc.returncode,
        )

    return proc.returncode, stdout

async def sync_mirror_repo():
    proc = await asyncio.create_subprocess_exec(
        "git",
        "fetch",
        "origin",
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=config.mirror_repo_dir,
    )

    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300.0)
    stdout = stderr.decode("utf-8")
    stderr = stderr.decode("utf-8")

    if proc.returncode != 0:
        raise CommandError(
            "git command `fetch origin` exited abnormally with return code {}".format(
                proc.returncode
            ),
            stdout,
            stderr,
            "fetch origin",
            proc.returncode,
        )

async def get_file(ref: str, path: str, *, from_ops: bool = False) -> bytes:
    """Get the contents at the file at the given ref."""

    if from_ops:
        cwd = config.ops_repo_dir
    else:
        cwd = config.mirror_repo_dir

    proc = await asyncio.create_subprocess_exec(
        "git",
        "cat-file",
        "--filters",
        "{}:{}".format(ref, str(path).strip()),
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=cwd,
    )

    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300.0)
    stderr = stderr.decode("utf-8")

    if proc.returncode != 0:
        raise CommandError(
            "git command `cat-file --filters {}:{}` exited abnormally with return code {}".format(
                ref, str(path), proc.returncode
            ),
            "",
            stderr,
            "cat-file",
            proc.returncode,
        )

    return stdout

async def get_blob(oid: str, *, from_ops: bool = False) -> bytes:
    if from_ops:
        cwd = config.ops_repo_dir
    else:
        cwd = config.mirror_repo_dir

    # First get the raw blob contents with `git cat-file`.
    # Then pass it through `git lfs smudge` to convert any pointers to binary files.
    # (lfs smudge ignores non-pointer data)
    _, blob_data = await run_git_command(
        ["cat-file", "blob", oid], cwd=cwd, decode=False
    )
    _, smudged_data = await run_git_command(
        ["lfs", "smudge"], stdin=blob_data, cwd=cwd, decode=False
    )

    return smudged_data

async def reset_ops_repo():
    await run_git_command(
        [
            "fetch",
            config.upstream_remote_name,
            "+refs/heads/*:refs/remotes/" + config.upstream_remote_name + "/*",
            "+refs/tags/*:refs/remotes/tags/*",
            "+refs/merge-requests/*/head:refs/remotes/"
            + config.upstream_remote_name
            + "/merge-requests/*",
        ]
    )

    # NOTE: explicitly switch to `master` before resetting it to upstream's `master`, to
    # ensure that Git LFS uses the upstream remote when pulling LFS objects.
    #
    # If we did this in one step with `git checkout -fB`, LFS might try to pull objects
    # using the staging remote instead of the upstream project's remote, leading to smudge filter
    # issues if the staging repo's LFS object store is out-of-sync with the upstream repo's LFS objects
    # (which might happen if updates are submitted shortly after a commit to upstream that adds new LFS objects).
    await run_git_command(["checkout", "-f", "master"])
    await run_git_command(["reset", "--hard", config.upstream_remote_name + "/master"])

    await run_git_command(["clean", "-ffdx"])

async def branch_exists(branch):
    retcode, _ = await run_git_command(
        ["show-ref", "--verify", "--quiet", "refs/heads/" + branch],
        raise_on_nonzero=False,
    )
    return retcode == 0

async def prepare_character_update(client, character):
    await reset_ops_repo()

    update_branch = "bot-update-" + character

    open_mrs = await info.get_merge_requests(
        client, config.upstream_project_id, "opened"
    )

    for mr in open_mrs:
        if (mr["source_project_id"] == config.staging_project_id) and (
            mr["source_branch"] == update_branch
        ):
            await run_git_command(["checkout", update_branch])
            break
    else:
        await run_git_command(
            ["checkout", "-B", update_branch, config.upstream_remote_name + "/master"]
        )

async def get_latest_tag(ref) -> str:
    """Get the latest tag reachable from a ref."""
    return await run_git_command(["describe", "--abbrev=0", "--tags", ref])[1]

async def commit_with_msg(msg):
    await run_git_command(["commit", "--file=-"], stdin=msg.encode("utf-8"))

async def tag_with_msg(tag_name, msg):
    await run_git_command(
        ["tag", "-u", "DB779289B9011EF5", "--file=-", tag_name],
        stdin=msg.encode("utf-8"),
    )

async def finalize_character_update(client, author, msg):
    await run_git_command(
        ["commit", "--author=" + author, "--file=-"], stdin=msg.encode("utf-8")
    )


code:
import asyncio
import aiohttp
import urllib.parse

from ..config import config
from ..utils import get_http_session

commit_cache = {}

class GitlabUserNotFound(Exception):
    pass

class GitlabProjectNotFound(Exception):
    pass

GIT_API_BASE_URL = "https://gitgud.io/api/v4"

async def get_commit_data(commit_sha):
    global commit_cache

    if commit_sha in commit_cache:
        return commit_cache[commit_sha]

    session = get_http_session()
    async with session.get(
        GIT_API_BASE_URL
        + "/projects/{}/repository/commits/{}".format(
            config.upstream_project_id, commit_sha
        ),
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        if resp.status < 200 or resp.status >= 300:
            return None
        data = await resp.json()

        commit_cache[commit_sha] = data
        return data

async def find_user(client, username):
    session = get_http_session()
    async with session.get(
        GIT_API_BASE_URL + "/users?username=" + urllib.parse.quote_plus(username),
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        data = await resp.json()

        for user in data:
            if user["username"] == username:
                return user

        raise GitlabUserNotFound()

async def find_user_project(client, user_id, project_name="spnati"):
    session = get_http_session()
    async with session.get(
        GIT_API_BASE_URL
        + "/users/"
        + str(int(user_id))
        + "/projects?search="
        + project_name,
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        data = await resp.json()

        for project in data:
            if project["name"] == project_name:
                return project

        raise GitlabProjectNotFound()

async def remote_branch_info(client, project_id, branch):
    session = get_http_session()
    async with session.get(
        gitlab_api_url(project_id, "/repository/branches/" + branch),
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        if resp.status >= 200 and resp.status <= 299:
            return await resp.json()
        else:
            return None

async def remote_branch_exists(client, project_id, branch):
    branch_info = await remote_branch_info(client, project_id, branch)
    return branch_info is not None

async def get_merge_requests(client, project_id, state="opened"):
    session = get_http_session()
    async with session.get(
        gitlab_api_url(project_id, "/merge_requests?state=" + state),
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        if resp.status >= 200 and resp.status <= 299:
            return await resp.json()
        else:
            data = await resp.json()
            raise OSError(
                "Failed to retrieve Gitlab merge requests - error {} ({})".format(
                    resp.status, str(data["message"])
                )
            )

async def get_incomplete_jobs(client, project_id):
    session = get_http_session()
    async with session.get(
        gitlab_api_url(
            project_id, "/jobs?scope[]=created&scope[]=pending&scope[]=running"
        ),
        headers={"Private-Token": config.git_apikey},
    ) as resp:
        if resp.status >= 200 and resp.status <= 299:
            return await resp.json()
        else:
            data = await resp.json()
            raise OSError(
                "Failed to retrieve Gitlab CI/CD jobs listing - error {} ({})".format(
                    resp.status, str(data["message"])
                )
            )

def gitlab_api_url(project_id, endpoint):
    return GIT_API_BASE_URL + "/projects/" + str(project_id) + endpoint


code:
from __future__ import annotations

from aiohttp import web
import asyncio
import discord
import json
import logging
from nacl.signing import VerifyKey
from nacl.exceptions import BadSignatureError
from typing import Dict, Any, Callable, Awaitable, Optional

from . import main, utils
from .config import config

class InteractionContext(object):
    def __init__(
        self,
        client: main.NotifierClient,
        command: str,
        token: str,
        guild: discord.Guild,
        channel: discord.TextChannel,
        user: discord.Member,
    ):
        self.client: main.NotifierClient = client
        self.guild: discord.Guild = guild
        self.channel: discord.TextChannel = channel
        self.user: discord.Member = user
        self.command = command
        self.token = token

    def reply(self, reply_text: Optional[str] = None, ephemeral=False, suppress_mentions=False) -> web.Response:
        if reply_text is None:
            return web.json_response({"type": 5})

        resp = {
            "type": 4,
            "data": {"content": reply_text},
        }

        if ephemeral:
            resp["data"]["flags"] = (1 << 6)

        if suppress_mentions:
            resp["data"]["allowed_mentions"] = {"parse": []}

        return web.json_response(resp)

InteractionHandler = Callable[[InteractionContext], Awaitable[web.Response]]

class Interactions(object):
    def __init__(self, client: main.NotifierClient):
        self.client: main.NotifierClient = client
        self.verify_key = VerifyKey(bytes.fromhex(config.interactions_public_key))
        self.app = web.Application()
        self.app.add_routes([web.post("/interaction", self.handler)])
        self.runner: web.AppRunner = None
        self.site: web.TCPSite = None
        self.interaction_handlers: Dict[str, InteractionHandler] = {}

    def register_handler(self, cmd: str, cb: InteractionHandler):
        self.interaction_handlers[cmd] = cb

    async def handler(self, request: web.Request):
        """Handler for incoming Interactions."""

        # Verify interaction signatures:
        try:
            signature: str = request.headers["X-Signature-Ed25519"]
            timestamp: str = request.headers["X-Signature-Timestamp"]
        except KeyError:
            raise web.HTTPUnauthorized(text="missing request signature")

        body: str = await request.text()
        verify_payload = (timestamp + body).encode("utf-8")
        try:
            self.verify_key.verify(verify_payload, bytes.fromhex(signature))
        except BadSignatureError:
            logging.error("Received interaction with invalid request signature")
            raise web.HTTPUnauthorized(text="invalid request signature")

        data = json.loads(body)
        if data["type"] == 1:
            return web.json_response({"type": 1})
        elif data["type"] != 2:
            logging.error("Received non-Command interaction")
            raise web.HTTPBadRequest(
                text="unknown interaction type " + str(data["type"])
            )

        cmd_data = data["data"]
        cmd = cmd_data["name"]
        try:
            cmd_handler = self.interaction_handlers[cmd]
        except KeyError:
            logging.error("Received unknown interaction command '" + cmd + "'")
            return web.json_response(
                {
                    "type": 4,
                    "data": {
                        "content": "<@!"
                        + data["member"]["user"]["id"]
                        + "> | `"
                        + cmd
                        + "` isn't a valid command. Try `b!help`?"
                    },
                }
            )

        logging.info("Processing slash command: " + cmd)

        guild: discord.Guild = self.client.get_guild(int(data["guild_id"]))
        channel: discord.TextChannel = self.client.get_channel(int(data["channel_id"]))
        user: discord.Member = guild.get_member(int(data["member"]["user"]["id"]))
        ctx = InteractionContext(self.client, cmd, data["token"], guild, channel, user)

        return await cmd_handler(ctx, cmd_data)

    async def start(self):
        self.runner = web.AppRunner(self.app)
        await self.runner.setup()
        self.site = web.TCPSite(self.runner, "localhost", config.interactions_port)
        await self.site.start()

    async def stop(self):
        if self.runner is not None:
            await self.runner.cleanup()


code:
import aiohttp
from bs4 import BeautifulSoup

from .config import config
from .utils import get_http_session

LINECOUNTD_URL = "http://localhost:8042"

class LinecountdError(Exception):
    def __init__(self, resp):
        self.endpoint = resp.url.path
        self.method = resp.method
        self.status = resp.status

        super().__init__(resp.method, self.endpoint, resp.status)

    def __str__(self):
        return "{} request to linecountd {} failed with error {}".format(
            self.method, self.endpoint, self.status
        )

async def linecountd_request(method, endpoint, *, headers=None, json=None, data=None):
    session = get_http_session()

    async with session.request(
        method, LINECOUNTD_URL + endpoint, headers=headers, json=json, data=data
    ) as resp:
        if resp.status < 200 or resp.status > 299:
            raise LinecountdError(resp)

        return await resp.json()

async def get_roster(ref: str):
    return await linecountd_request("GET", "/roster/" + ref)

async def get_linecount(roster_id: str, ref: str):
    return await linecountd_request(
        "GET", "/count/{}/{}".format(roster_id.lower(), ref)
    )

async def get_inbound_targets(entity: str, ref: str):
    return await linecountd_request(
        "GET", "/inbound_targeting/{}/{}".format(entity.lower(), ref)
    )

async def get_lineset_diff(xml_data, tags_xml_data, roster_id: str, ref: str):
    if tags_xml_data is not None:
        if isinstance(xml_data, bytes):
            xml_data = xml_data.decode("utf-8")

        if isinstance(tags_xml_data, bytes):
            tags_xml_data = tags_xml_data.decode("utf-8")

        return await linecountd_request(
            "POST",
            "/diff/{}/{}".format(roster_id.lower(), ref),
            json={
                "behaviour.xml": xml_data,
                "tags.xml": tags_xml_data
            },
            headers={"Content-Type": "application/json"},
        )
    else:
        if isinstance(xml_data, str):
            xml_data = xml_data.encode("utf-8")

        return await linecountd_request(
            "POST",
            "/diff/{}/{}".format(roster_id.lower(), ref),
            data=xml_data,
            headers={"Content-Type": "application/xml"},
        )

async def get_character_tags(roster_id: str, ref):
    return await linecountd_request(
        "GET", "/tags/character/{}/{}".format(roster_id.lower(), ref)
    )

async def get_commit_tags(ref: str):
    return await linecountd_request("GET", "/tags/commit/{}".format(ref))

async def get_tagged_characters(ref: str, tag: str):
    return await linecountd_request(
        "GET", "/tags/commit/{}/{}".format(ref, tag.lower())
    )