code:
import csv
import math
import re
import sys
import unicodedata

import numpy as np
from sklearn.feature_extraction import DictVectorizer

import classifier

def get_features():
    with open(sys.argv[1], "r", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            text = row["text"]
            features = classifier.extract_features(text)

            if features is None:
                continue

            yield (row["spam"] == "True", features)

def count_features():
    n_spam = 0
    n_ham = 0

    spam_counts = {}
    ham_counts = {}

    print("Counting features...")
    for spam, features in get_features():
        if spam:
            n_spam += 1
        else:
            n_ham += 1

        for feature in features:
            if spam:
                spam_counts[feature] = spam_counts.get(feature, 0) + 1
            else:
                ham_counts[feature] = ham_counts.get(feature, 0) + 1

    features = set(spam_counts.keys()).union(ham_counts.keys())
    total_counts = {}

    for feature in features:
        total_counts[feature] = (
            spam_counts.get(feature, 0),
            ham_counts.get(feature, 0),
        )

    print("Counted {} spam reports and {} ham reports.".format(n_spam, n_ham))

    return total_counts, n_spam, n_ham

def main():
    total_counts, total_spam, total_ham = count_features()
    scores = []

    print("Computing scores...")
    for feature, counts in total_counts.items():
        n_spam, n_ham = counts
        n_total = n_spam + n_ham

        # Compute raw p(spam | feature):
        p_spam = n_spam / total_spam
        p_ham = n_ham / total_ham
        p = p_spam / (p_ham + p_spam)

        # Compute adjusted p'(spam|feature), to deal with rare features:
        s = 3
        pr_s = 0.5
        p_prime = ((s * pr_s) + (n_total * p)) / (s + n_total)

        # Convert to a logit:
        score = math.log(1 - p_prime) - math.log(p_prime)

        scores.append((feature, score))

    with open(sys.argv[2], "w", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(("feature", "score"))

        print("Writing output...")
        writer.writerows(sorted(scores, key=lambda t: t[1], reverse=True))

if __name__ == "__main__":
    main()

repo_name: spnati/serve

license: mit


code:
#!/usr/bin/env bash

# Compute feature scores from corpus
pipenv run python3 ./do_train.py ./corpus.csv ./scores.csv

# Display ROC of classifier for threshold tuning
pipenv run python3 ./analyze.py ./scores.csv ./corpus.csv


code:
INCLUDE_WORDS = [
    "i'm",
    "masturbation",
    "dialog",
    "glitch",
    "i've",
    "i'd",
    "sprite",
    "pls",
    "boob",
    "cum",
    "cumming",
    "masterbate",
    "masturbate",
    "anime",
    "misspelled",
    "pokemon",
    "glitching",
    "rng",
    "collectable",
    "i'll",
    "plural",
    "spnati",
    "slutty",
    "browser",
    "masturbates",
    "dialouge",
    "glitched",
    "offline",
    "user's",
    "resubmitting",
    "valentine's",
    "ddlc",
    "cheerleader",
    "icon",
    "placeholder",
    "doki",
    "vibrator",
    "cuming",
    "fucked",
    "shantae",
    "pubic",
    "christmas",
    "sph",
    "parenthesis",
    "ifplural",
    "alt",
    "glitchy",
    "striping",
    "mlp",
    "tsundere",
    "masterbating",
    "footjob",
    "slut",
    "devs",
    "character's",
    "manga",
    "idk",
    "vampire",
    "futa",
    "masterbation",
    "freezed",
    "rephrase",
    "kazuma",
    "valentine",
    "dingaling",
    "titty",
    "misspelling",
    "clipping",
    "avatar",
    "youtube",
    "snide",
    "reskin",
    "mime",
    "cutscene",
    "postforfeit",
    "fullscreen",
    "kudos",
    "android",
    "zoom",
    "yandere",
    "asap",
    "reloading",
]


code:
import asyncio
import csv
import sys
from typing import AsyncGenerator, Tuple

from motor.motor_asyncio import (
    AsyncIOMotorClient,
    AsyncIOMotorDatabase,
    AsyncIOMotorCollection,
)

async def read_spam(
    coll: AsyncIOMotorCollection,
) -> AsyncGenerator[Tuple[str, str], None]:
    cursor = coll.find({"status": "spam", "type": {"$ne": "auto"}})
    cursor.sort("_id", -1)

    async for doc in cursor:
        try:
            text = doc["description"]
            doc_id = doc["_id"]
        except KeyError:
            continue

        yield (doc_id, text)

async def read_ham(
    coll: AsyncIOMotorCollection, n: int
) -> AsyncGenerator[Tuple[str, str], None]:
    cursor = coll.find(
        {
            "type": {"$ne": "auto"},
            "$or": [{"status": "open"}, {"status": {"$exists": False}}],
        }
    )
    cursor.sort("_id", -1)

    cur = 0
    async for doc in cursor:
        try:
            text = doc["description"].strip()
            doc_id = doc["_id"]
        except (KeyError, AttributeError):
            continue

        if len(text) > 500:
            continue

        cur += 1
        yield (doc_id, text)

        if cur >= n:
            return

async def main():
    db_client = AsyncIOMotorClient("localhost", 27017)

    db: AsyncIOMotorDatabase = db_client.spnati_usage_stats
    coll: AsyncIOMotorCollection = db.bug_reports

    with open(sys.argv[1], "w", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(("spam", "doc_id", "text"))

        print("Reading spam...")

        n_spam = 0
        async for doc_id, text in read_spam(coll):
            writer.writerow((True, doc_id, text))
            n_spam += 1

        print("Read {} spam reports.\nReading ham...".format(n_spam))

        async for doc_id, text in read_ham(coll, n_spam * 5):
            writer.writerow((False, doc_id, text))

        print("Read {} ham reports.".format(n_spam * 5))

if __name__ == "__main__":
    asyncio.run(main())


code:
import asyncio
import csv
import sys
from typing import AsyncGenerator, Tuple

from motor.motor_asyncio import (
    AsyncIOMotorClient,
    AsyncIOMotorDatabase,
    AsyncIOMotorCollection,
)

async def read(
    coll: AsyncIOMotorCollection, n: int, skip=0
) -> AsyncGenerator[Tuple[str, str], None]:
    cursor = coll.find({"type": {"$ne": "auto"},})
    cursor.sort("_id", -1)
    cursor.skip(skip)

    cur = 0
    async for doc in cursor:
        if "status" in doc and doc["status"] != "open":
            continue

        try:
            text = doc["description"].strip()
            doc_id = doc["_id"]
        except (KeyError, AttributeError):
            continue

        if len(text) > 500:
            continue

        cur += 1
        yield (doc_id, text)

        print(cur)

        if cur >= n:
            return

async def main():
    db_client = AsyncIOMotorClient("localhost", 27017)

    db: AsyncIOMotorDatabase = db_client.spnati_usage_stats
    coll: AsyncIOMotorCollection = db.bug_reports

    with open(sys.argv[1], "w", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(("doc_id", "text"))

        n = int(sys.argv[2])
        async for doc_id, text in read(coll, n, 6000):
            writer.writerow((False, doc_id, text))

        print("Read {} testing reports.".format(n))

if __name__ == "__main__":
    asyncio.run(main())


code:
import csv
import math
import sys

import numpy as np
from matplotlib import pyplot as plt

import classifier

def get_scores(corpus_file, scores):
    print("Computing scores...")
    with open(corpus_file, "r", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            score = classifier.classify(row["text"], scores)
            if score is None:
                continue

            yield (row["doc_id"], score, row["text"])

def do_classification(corpus_file, scores, threshold):
    print("Classifying...")

    n_spam = 0
    n_ham = 0
    t = str.maketrans("\n", " ")

    for doc_id, score, text in get_scores(corpus_file, scores):
        # text = classifier.clean(text)
        text = text.translate(t)

        if len(text) == 0:
            continue

        if score < threshold:
            n_spam += 1
        else:
            n_ham += 1

        yield (score < threshold, doc_id, text, score)

    print("Classification:")
    print("    {: 4d} spam".format(n_spam))
    print("    {: 4d} ham".format(n_ham))
    print("    {: 4d} total".format(n_spam + n_ham))

if __name__ == "__main__":
    # p = 0.800  # threshold probability
    t = 0.5
    print("Threshold: {:.3f}".format(t))

    scores = classifier.load_scores(sys.argv[1])
    data = sorted(
        do_classification(sys.argv[2], scores, t), key=lambda t: t[-1], reverse=True,
    )
    data = sorted(data, key=lambda t: t[0], reverse=True)

    with open(sys.argv[3], "w", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow(("spam", "doc_id", "text", "score"))
        writer.writerows(data)


code:
import asyncio
import re
from typing import Optional, Any, Tuple, List, Iterable
from pathlib import PurePath, Path
from functools import partial

from .git import git_ops
from .config import config
from . import linecount
from .utils import format_name

TREE_DIFF_REGEX = b":([0-7]{6}) ([0-7]{6}) ([0-9a-f.]{1,40}) ([0-9a-f.]{1,40}) (?:([ADMTUX])(\\d{0,3})\x00([^\x00]+)\x00)|(?:([CR])(\\d{1,3})\x00([^\x00]+)\x00([^\x00]+)\x00)"

class TreeDiffLine(object):
    def __init__(self, match: re.Match):
        self.src_mode: str = match[1].decode("utf-8")
        self.src_oid: str = match[2].decode("utf-8")
        self.dst_mode: str = match[3].decode("utf-8")
        self.dst_oid: str = match[4].decode("utf-8")
        self.dst_path: Optional[PurePath] = None

        if match[5] is not None:
            self.status: str = match[5].decode("utf-8")
            if len(match[6]) > 0:
                self.score = int(match[6].decode("utf-8"))
            else:
                self.score = None

            self.src_path: PurePath = PurePath(match[7].decode("utf-8"))
        else:
            self.status = match[8].decode("utf-8")
            self.score = int(match[9].decode("utf-8"))
            self.src_path = PurePath(match[10].decode("utf-8"))
            self.dst_path = PurePath(match[11].decode("utf-8"))

class Change(object):
    def __init__(self, change_type: str, character: str, data: Any):
        self.type = change_type
        self.character = character
        self.data = data

async def examine_change(start_ref: str, end_ref: str, file: TreeDiffLine):
    if file.src_path.parts[0] != "opponents":
        return

    if file.src_path.name == "behaviour.xml":
        opponent_name = file.src_path.parts[1]
        new_data = await git_ops.get_file(end_ref, str(file.src_path))

        try:
            new_tags_data = await git_ops.get_file(end_ref, "opponents/" + str(file.src_path.parts[1]) + "/tags.xml")
        except git_ops.CommandError:
            new_tags_data = None

        try:
            diff = await linecount.get_lineset_diff(new_data, new_tags_data, opponent_name, start_ref)
        except linecount.LinecountdError:
            return

        try:
            line_changes = diff["all_lines"]["net"]
        except KeyError:
            line_changes = 0

        try:
            pose_changes = diff["all_lines"]["net"]
        except KeyError:
            pose_changes = 0

        if line_changes == 0 and pose_changes == 0:
            return

        net_changes = {
            "lines": line_changes,
            "poses": pose_changes,
        }

        yield Change("character", opponent_name, net_changes)
    elif file.src_path.name == "listing.xml" and len(file.src_path.parts) == 2:
        old_roster: dict = await linecount.get_roster(start_ref)
        new_roster: dict = await linecount.get_roster(end_ref)

        old_ids = set(old_roster.keys())
        new_ids = set(new_roster.keys())

        added = new_ids.difference(old_ids)
        removed = old_ids.difference(new_ids)
        roster_changes = added.union(removed)

        for common_id in old_ids.intersection(new_ids):
            if old_roster[common_id] != new_roster[common_id]:
                roster_changes.add(common_id)

        for changed_id in roster_changes:
            change = {
                "from": old_roster.get(changed_id, None),
                "to": new_roster.get(changed_id, None),
            }

            yield Change("roster_change", changed_id, change)

async def compile_changes(start_ref: str, end_ref: str):
    _, diff_output = await git_ops.run_git_command(
        ["diff", "-z", "--raw", start_ref, end_ref],
        cwd=config.mirror_repo_dir,
        decode=False,
    )

    # _, diff_readable = await git_ops.run_git_command(
    #     ["diff", "--raw", start_ref, end_ref], cwd=config.mirror_repo_dir,
    # )

    # print(diff_readable)

    for match in re.finditer(TREE_DIFF_REGEX, diff_output):
        try:
            file = TreeDiffLine(match)
        except AttributeError:
            continue

        async for change in examine_change(start_ref, end_ref, file):
            # print("change: {} - {}".format(change.type, change.character))
            yield change

async def prepare_version(start_ref: str, end_ref: str) -> Tuple[bool, str]:
    # keyed by move destinations
    roster_moves = {}
    roster_additions = set()

    # sets of character ids
    dialogue_updates = set()
    art_updates = set()

    async for change in compile_changes(start_ref, end_ref):
        change: Change = change

        if change.type == "character":
            if change.data["lines"] > 0:
                dialogue_updates.add(change.character)

            if change.data["poses"] > 0:
                art_updates.add(change.character)
        elif change.type == "roster_change":
            if change.data["from"] is None and change.data["to"] == "testing":
                roster_additions.add(change.character)
            else:
                dest = change.data.get("to", "offline")
                if dest not in roster_moves:
                    roster_moves[dest] = set()
                roster_moves[dest].add(change.character)

    minor_bump = len(roster_additions) > 0 or len(roster_moves) > 0
    notes = []

    if "online" in roster_moves:
        notes.append(
            format_note(
                "{characters} {were} added to the main roster.", roster_moves["online"],
            )
        )

    if len(roster_additions) > 0:
        notes.append(
            format_note(
                "{characters} {were} added to the Testing Tables.", roster_additions
            )
        )

    if "testing" in roster_moves:
        notes.append(
            format_note(
                "{characters} {were} moved to the Testing Tables.",
                roster_moves["testing"],
            )
        )

    offline_characters: set = roster_moves.get("offline", set())
    incomplete_characters: set = roster_moves.get("incomplete", set())
    offlined = offline_characters.union(incomplete_characters)

    if len(offlined) > 0:
        notes.append(format_note("{characters} {were} moved offline.", offlined))

    if (
        len(dialogue_updates) > 0
        and len(art_updates) > 0
        and dialogue_updates == art_updates
    ):
        notes.append(
            format_note(
                "Updated dialogue and artwork for {characters}.", dialogue_updates
            )
        )
    else:
        if len(dialogue_updates) > 0:
            notes.append(
                format_note("Updated dialogue for {characters}.", dialogue_updates)
            )

        if len(art_updates) > 0:
            notes.append(format_note("Updated artwork for {characters}.", art_updates))

    return minor_bump, " ".join(notes)

def format_note(template: str, characters: Iterable[str]) -> str:
    were = "were"
    if len(characters) <= 1:
        were = "was"

    return template.format(
        characters=format_list(sorted(map(format_name, characters))), were=were
    )

def format_list(l: List[str]) -> str:
    if len(l) == 0:
        return ""
    elif len(l) == 1:
        return l[0]
    elif len(l) == 2:
        return l[0] + " and " + l[1]
    else:
        part = ", ".join(l[:-1])
        return part + ", and " + l[-1]


code:
from .handler import handle_cmd_string
from .slash_cmd import slash_commands

from . import character_info
from . import character_update
from . import deploy_testing
from . import graphing
from . import playstats
from . import purge_cache
from . import reports
from . import silence
from . import status
from . import sync
from . import version
from . import masquerade
from . import permissions
from . import update_alerts

async def initialize(client):
    await sync.init_ipc(client)

    for cmd_group in slash_commands:
        client.interactions.register_handler(cmd_group.name, cmd_group.handler)


code:
import asyncio
import csv
from datetime import datetime, timedelta
import io
import urllib.parse

import aiohttp
from bs4 import BeautifulSoup
import discord

from .handler import command
from ..config import config
from .. import linecount
from ..utils import get_http_session, format_name, split_long_message

@command("pairs", summary="Get pairing information for a given character.")
async def cmd_pairs(client, msg, args):
    if len(args) < 1:
        return await client.reply(msg, "Usage: `b!pairs [character ID]`")

    character = args[0].strip().lower()
    now = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
    cutoff = now - timedelta(days=7)

    roster = await linecount.get_roster("master")
    async with msg.channel.typing():
        n_total = 0
        cursor = client.db.usage_reports.find(
            {
                "date": {"$gte": cutoff, "$lt": now},
                "type": "start_game",
                "$or": [
                    {"table.1": character},
                    {"table.2": character},
                    {"table.3": character},
                    {"table.4": character},
                ],
            }
        )

        paired = {}

        async for report in cursor:
            for i in range(1, 5):
                slot = str(i)
                table_char = report["table"].get(slot, None)
                if (
                    table_char is not None
                    and table_char != character
                    and table_char in roster
                ):
                    paired[table_char] = paired.get(table_char, 0) + 1
            n_total += 1

        out = []
        idx = 1
        for char_id, pair_count in sorted(
            paired.items(), key=lambda c: c[1], reverse=True
        )[:10]:
            print("{} / {} pairing count: {}".format(character, char_id, pair_count))
            out.append(
                "**{}.** {} ({:.1%})".format(
                    idx, format_name(char_id), pair_count / n_total
                )
            )
            idx += 1

        agg_cursor = client.db.usage_reports.aggregate(
            [
                {
                    "$match": {
                        "date": {"$gte": cutoff, "$lt": now},
                        "type": "start_game",
                    }
                },
                {"$group": {"_id": "$ip"}},
                {"$group": {"_id": None, "total": {"$sum": 1}}},
            ]
        )

        ip_doc = await agg_cursor.to_list(1)
        n_ips = "<unknown>"
        if len(ip_doc) > 0:
            n_ips = "{:,d}".format(ip_doc[0]["total"])

        total_game_count = await client.db.usage_reports.count_documents(
            {"date": {"$gte": cutoff, "$lt": now}, "type": "start_game"}
        )

        return await msg.channel.send(
            "Top 10 paired characters for **{}:**\n{}\nData collected from **{:,d} games** and **{} players** in total over the past 7 days.".format(
                format_name(character), "\n".join(out), total_game_count, n_ips
            )
        )

@command(
    "targeting-all",
    summary="Get a spreadsheet of all outbound targets between characters.",
)
async def cmd_get_targeting_spreadsheet(client, msg, args):
    roster = await linecount.get_roster("master")

    filter_status = ["online", "testing"]

    if "include-offline" in args:
        filter_status.append("offline")

    if "include-incomplete" in args:
        filter_status.append("incomplete")

    if "exclude-testing" in args:
        filter_status.remove("testing")

    chars = list(sorted(filter(lambda c: roster[c] in filter_status, roster.keys())))
    with io.StringIO() as sio:
        fieldnames = ["character", "total"]
        fieldnames.extend(chars)

        writer = csv.DictWriter(sio, fieldnames)
        writer.writeheader()

        for character in chars:
            data = await linecount.get_linecount(character, "master")
            row = {"character": character, "total": data["targeted_lines"]}

            for target_char in chars:
                row[target_char] = data["targets"].get(target_char, 0)

            writer.writerow(row)

        with io.BytesIO(sio.getvalue().encode("utf-8")) as bio:
            await client.reply(
                msg,
                "Here is your outbound targeting spreadsheet:",
                file=discord.File(bio, "targeting.csv"),
            )

@command("linecount-all", summary="Get a spreadsheet of linecounts for all characters.")
async def cmd_get_linecount_spreadsheet(client, msg, args):
    roster = await linecount.get_roster("master")

    filter_status = ["online", "testing"]

    if "include-offline" in args:
        filter_status.append("offline")

    if "include-incomplete" in args:
        filter_status.append("incomplete")

    if "exclude-testing" in args:
        filter_status.remove("testing")

    chars = list(sorted(filter(lambda c: roster[c] in filter_status, roster.keys())))
    with io.StringIO() as sio:
        fieldnames = [
            "character",
            "total",
            "generic",
            "targeted",
            "filtered",
            "poses",
        ]
        writer = csv.DictWriter(sio, fieldnames)
        writer.writeheader()

        for character in chars:
            data = await linecount.get_linecount(character, "master")
            writer.writerow(
                {
                    "character": character,
                    "total": data["all_lines"],
                    "generic": data["generic_lines"],
                    "targeted": data["targeted_lines"],
                    "filtered": data["filtered_lines"],
                    "poses": data["poses"],
                }
            )

        with io.BytesIO(sio.getvalue().encode("utf-8")) as bio:
            await client.reply(
                msg,
                "Here is your linecount spreadsheet:",
                file=discord.File(bio, "linecounts.csv"),
            )

@command("linecount", summary="Get linecount statistics for a character.")
async def cmd_get_linecount(client, msg, args):
    if len(args) < 1:
        return await client.reply(msg, "USAGE: `b!linecount [opponent folder names]`")

    for character_id in args:
        try:
            linecounts = await linecount.get_linecount(character_id, "master")
        except linecount.LinecountdError as err:
            if err.status == 404:
                await msg.channel.send(
                    "❌  Could not find character `{}`!".format(character_id)
                )
            else:
                await msg.channel.send(
                    "❌  Could not get linecount for `{}`: error {}".format(
                        character_id, err.status
                    )
                )
            continue

        resp = "**Linecount for {} (`{}`):**\n".format(
            format_name(character_id), character_id
        )
        resp += "    - **Total Linecount:** {:d}\n".format(linecounts["all_lines"])
        resp += "    - **Generic Linecount:** {:d}\n".format(
            linecounts["generic_lines"]
        )

        resp += "    - **Targeted Linecount:** {:d}\n".format(
            linecounts["targeted_lines"]
        )
        resp += "        - **Opponents Targeted:** {:d}\n".format(
            len(linecounts["targeted_characters"])
        )

        resp += "    - **Filtered Linecount:** {:d}\n".format(
            linecounts["filtered_lines"]
        )
        resp += "        - **Tags Targeted:** {:d}\n".format(
            len(linecounts["targeted_tags"])
        )
        resp += "    - **Total Pose Count:** {:d}".format(linecounts["poses"])

        await client.reply(msg, resp)

@command("target_info", summary="List opponents that this character targets.")
async def cmd_target_info(client, msg, args):
    if len(args) < 1:
        return await client.reply(
            msg, "USAGE: `b!target_info [opponent folder name] [count|'all']`"
        )

    try:
        n_lines = int(args[1])
    except ValueError:
        if args[1].lower() == "all":
            n_lines = None
    except IndexError:
        n_lines = 10

    try:
        linecounts = await linecount.get_linecount(args[0], "master")
    except linecount.LinecountdError as err:
        if err.status == 404:
            await msg.channel.send("❌  Could not find character `{}`!".format(args[0]))
        else:
            await msg.channel.send(
                "❌  Could not get target info for `{}`: error {}".format(
                    args[0], err.status
                )
            )
        return

    target_counts = {}
    for target in linecounts["targeted_characters"]:
        target_counts[target] = linecounts["targets"][target]

    resp = "**Characters Targeted by** `{}`:\n".format(args[0])
    i = 0

    for target, count in sorted(target_counts.items(), key=lambda kv: (-kv[1], kv[0])):
        comp = "    - `{}`: {:d} lines\n".format(target, count)
        if len(resp) + len(comp) > 1900:
            await msg.channel.send(resp)
            resp = comp
        else:
            resp += comp

        i += 1
        if n_lines is not None and i >= n_lines:
            break

    await msg.channel.send(resp)

@command("filter_info", summary="List tags that this character targets.")
async def cmd_filter_info(client, msg, args):
    if len(args) < 1:
        return await client.reply(
            msg, "USAGE: `b!filter_info [opponent folder name] [count|'all']`"
        )

    try:
        n_lines = int(args[1])
    except ValueError:
        if args[1].lower() == "all":
            n_lines = None
    except IndexError:
        n_lines = 10

    try:
        linecounts = await linecount.get_linecount(args[0], "master")
    except linecount.LinecountdError as err:
        if err.status == 404:
            await msg.channel.send("❌  Could not find character `{}`!".format(args[0]))
        else:
            await msg.channel.send(
                "❌  Could not get target info for `{}`: error {}".format(
                    args[0], err.status
                )
            )
        return

    filter_counts = {}
    for target in linecounts["targeted_tags"]:
        filter_counts[target] = linecounts["targets"][target]

    resp = "**Tags Targeted by {} (`{}`):**\n".format(format_name(args[0]), args[0])
    i = 0

    for tag, count in sorted(filter_counts.items(), key=lambda kv: (-kv[1], kv[0])):
        comp = "    - `{}`: {:d} lines\n".format(tag, count)
        if len(resp) + len(comp) > 1900:
            await msg.channel.send(resp)
            resp = comp
        else:
            resp += comp

        i += 1
        if n_lines is not None and i >= n_lines:
            break

    await msg.channel.send(resp)

@command("inbound", summary="List opponents that target a given character.")
async def cmd_inbound_target_info(client, msg, args):
    if len(args) < 1:
        return await client.reply(
            msg, "USAGE: `b!inbound [opponent folder name] [count|'all']`"
        )

    try:
        n_lines = int(args[1])
    except ValueError:
        if args[1].lower() == "all":
            n_lines = None
    except IndexError:
        n_lines = 10

    try:
        inbound_counts = await linecount.get_inbound_targets(args[0], "master")
    except linecount.LinecountdError as err:
        if err.status == 404:
            await msg.channel.send("❌  Could not find character `{}`!".format(args[0]))
        else:
            await msg.channel.send(
                "❌  Could not get inbound target info for `{}`: error {}".format(
                    args[0], err.status
                )
            )
        return

    resp1 = "**Characters Directly Targeting {} (`{}`):**\n".format(
        format_name(args[0]), args[0]
    )
    i = 0

    for source, count in sorted(inbound_counts.items(), key=lambda kv: (-kv[1], kv[0])):
        comp = "    - `{}`: {:d} lines\n".format(source, count)
        if len(resp1) + len(comp) > 1900:
            await msg.channel.send(resp1)
            resp1 = comp
        else:
            resp1 += comp

        i += 1
        if n_lines is not None and i >= n_lines:
            break

    await msg.channel.send(resp1)

def _format_stage_set(stages, n_layers) -> str:
    stages = sorted(set(stages))
    if len(stages) == 0:
        return None
    elif len(stages) == 1:
        return "stage " + str(stages[0])
    elif (
        min(stages) == 0
        and max(stages) >= (n_layers + 2)
        and len(stages) >= (n_layers + 3)
    ):
        return "all stages"

    intervals = []
    prev_stage = stages[0]
    start_stage = prev_stage
    for stage in stages[1:]:
        if stage != prev_stage + 1:
            intervals.append((start_stage, prev_stage))
            start_stage = stage
        prev_stage = stage
    intervals.append((start_stage, prev_stage))

    formatted = []
    for start, end in intervals:
        if start == end:
            formatted.append(str(start))
        else:
            formatted.append("{:d}-{:d}".format(start, end))

    if len(formatted) == 1:
        return "stages " + formatted[0]
    elif len(formatted) == 2:
        return "stages " + formatted[0] + " and " + formatted[1]
    else:
        return "stages " + (", ".join(formatted[:-1])) + ", and " + formatted[-1]

def _format_tag_details(tag_data, n_layers) -> str:
    add_remove_fmt = ""
    if "case_added" in tag_data:
        fmt = _format_stage_set(tag_data["case_added"], n_layers)
        if fmt is not None:
            add_remove_fmt = "added in " + fmt

    if "case_removed" in tag_data:
        fmt = _format_stage_set(tag_data["case_removed"], n_layers)
        if fmt is not None:
            if len(add_remove_fmt) > 0:
                add_remove_fmt += " and "
            add_remove_fmt += "removed in " + fmt

    if "stages" in tag_data:
        ret = _format_stage_set(tag_data["stages"], n_layers)
        if len(add_remove_fmt) > 0:
            ret += " (possibly " + add_remove_fmt + ")"
        return ret
    elif len(add_remove_fmt) > 0:
        return "possibly " + add_remove_fmt
    else:
        return ""

@command("tags", summary="List all tags for a given character.")
async def cmd_tag_info(client, msg, args):
    if len(args) < 1:
        return await client.reply(msg, "USAGE: `b!tags [opponent folder names]`")

    for character_id in args:
        try:
            data = await linecount.get_character_tags(character_id, "master")
        except linecount.LinecountdError as err:
            if err.status == 404:
                await msg.channel.send(
                    "❌  Could not find character `{}`!".format(character_id)
                )
            else:
                await msg.channel.send(
                    "❌  Could not get tags for `{}`: error {}".format(
                        character_id, err.status
                    )
                )
            continue

        n_layers = data["n_layers"]
        lines = [
            "**Character Tags for {} (`{}`):**".format(
                format_name(character_id), character_id
            )
        ]

        max_tag_len = max(len(tag) for tag in data["tags"].keys())
        for tag_name, tag_data in sorted(data["tags"].items(), key=lambda kv: kv[0]):
            fmt_details = _format_tag_details(tag_data, n_layers)
            if len(fmt_details) == 0:
                continue  # shouldn't happen, but...

            lines.append("`{}` : {}".format(tag_name.ljust(max_tag_len), fmt_details))

        for page in split_long_message(lines):
            await msg.channel.send(page)


code:
import asyncio
from pathlib import Path, PurePath
import os
import os.path as osp
import random
import re
import sys
import tempfile
import traceback
import zipfile
from typing import Tuple
import urllib.parse

import aiohttp

from .handler import command
from ..config import config
from ..git import git_ops
from ..git.info import gitlab_api_url, get_merge_requests
from ..masquerade import get_character_mask
from ..reporting.routing import find_character_channel
from ..utils import (
    get_http_session,
    format_name,
    character_permissions_check,
)
from .. import linecount

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

diff_set_names = {
    "all_lines": "Total Linecount",
    "generic_lines": "Generic Linecount",
    "special_lines": "Special-Conditions Linecount",
    "targeted_lines": "Targeted Linecount",
    "filtered_lines": "Filtered Linecount",
    "poses": "Referenced Poses",
    "targeted_characters": "Total Targeted Characters",
    "targeted_tags": "Total Targeted Tags",
}

def format_diff_info(info: dict, include_modified: bool = True) -> str:
    ret = "{:d} -> {:d}".format(info["old"], info["new"])

    if not include_modified:
        return ret + " ({:+d})".format(info["net"])

    info_parts = []

    if info["added"] > 0:
        info_parts.append("added {:d}".format(info["added"]))

    if info["removed"] > 0:
        info_parts.append("removed {:d}".format(info["removed"]))

    if info["modified"] > 0:
        info_parts.append("modified {:d}".format(info["modified"]))

    if info["added"] > 0 and info["removed"] > 0:
        info_parts.append("net {:+d}".format(info["net"]))

    return "{:d} -> {:d} ({:s})".format(info["old"], info["new"], ", ".join(info_parts))

def _diff_info_sort_key(kv) -> Tuple[int, int, str]:
    target_key, info = kv
    # invert sign of net and modified counts to sort from most to least
    return (-info["net"], -info["modified"], target_key)

def format_linecount_updates(diff):
    commit_msg = ""

    for key, info in diff.items():
        try:
            header = diff_set_names[key]
        except KeyError:
            continue

        if key == "poses" and info["net"] == 0:
            continue

        commit_msg += "\n**{:s}: ** {:s}".format(
            header, format_diff_info(info, key != "poses")
        )

    if len(diff["character_targets"]) > 0:
        commit_msg += "\n\nTargeting Updates:"

        for target, info in sorted(
            diff["character_targets"].items(), key=_diff_info_sort_key
        ):
            commit_msg += "\n    - `{:s}`: {:s}".format(target, format_diff_info(info))

    if len(diff["tag_targets"]) > 0:
        commit_msg += "\n\nFilter Updates:"

        for target, info in sorted(
            diff["tag_targets"].items(), key=_diff_info_sort_key
        ):
            commit_msg += "\n    - `{:s}`: {:s}".format(target, format_diff_info(info))

    return commit_msg

async def compute_update_msgs(character: str) -> Tuple[str, str]:
    _, last_update_sha = await git_ops.run_git_command(["rev-parse", "HEAD"])
    last_update_sha = last_update_sha.strip()

    _, master_sha = await git_ops.run_git_command(
        ["rev-parse", config.upstream_remote_name + "/master"]
    )
    master_sha = master_sha.strip()

    behaviour_path = osp.join(
        config.ops_repo_dir, "opponents", character, "behaviour.xml"
    )
    with open(behaviour_path, "rb") as f:
        updated_data = f.read()

    tags_path = osp.join(config.ops_repo_dir, "opponents", character, "tags.xml")
    if osp.isfile(tags_path):
        with open(tags_path, "rb") as f:
            updated_tags_data = f.read()
    else:
        updated_tags_data = None

    if last_update_sha == master_sha:
        diff = await linecount.get_lineset_diff(
            updated_data, updated_tags_data, character, last_update_sha
        )
        formatted = format_linecount_updates(diff)
        return (formatted, formatted)
    else:
        update_diff, master_diff = await asyncio.gather(
            linecount.get_lineset_diff(
                updated_data, updated_tags_data, character, last_update_sha
            ),
            linecount.get_lineset_diff(
                updated_data, updated_tags_data, character, master_sha
            ),
        )
        return (
            format_linecount_updates(update_diff),
            format_linecount_updates(master_diff),
        )

# TODO: update later
names = [
    "Akashi",
    "Ooyodo",
    "GSh-18",
    "Amiya",
    "Closure",
    "Hibiscus",
    "Papascha",
    "Vis",
]
messages = []

def format_commit_msg(msg):
    global names

    def replace_numbers(match):
        start = 1
        end = 999

        try:
            if match.group(2) is None:
                end = int(match.group(1))
            else:
                start = int(match.group(1))
                end = int(match.group(2))
        except (IndexError, ValueError):
            pass

        if start > end:
            end = start * 2

        return str(random.randint(start, end))

    formatted = re.sub(r"XNUM(\d*)(?:,\s*(\d*))?X", replace_numbers, msg)
    formatted = formatted.replace("XNAMEX", random.choice(names))
    formatted = formatted.replace("XUPPERNAMEX", random.choice(names).upper())
    formatted = formatted.replace("XLOWERNAMEX", random.choice(names).lower())

    return formatted

class ExternalCommandError(Exception):
    pass

async def run_cmd(command, *args):
    proc = await asyncio.create_subprocess_exec(
        command,
        *args,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        cwd=config.ops_repo_dir
    )

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

    return stdout, stderr

def normalize_zip_path(in_path: str, character: str) -> PurePath:
    in_path = PurePath(in_path)
    out_segments = []

    # Strip '.' and '..' segments from path
    for segment in in_path.parts:
        if segment == ".":
            continue
        elif segment == "..":
            try:
                out_segments.pop()
            except IndexError:
                continue
        else:
            out_segments.append(segment)

    # Ensure output path is relative
    out_path = PurePath(*out_segments)
    out_path = out_path.relative_to(out_path.root)

    # Remove leading 'opponents/<id>/' and '<id>/' prefixes, if any
    try:
        out_path = out_path.relative_to("opponents", character)
    except ValueError:
        try:
            out_path = out_path.relative_to(character)
        except ValueError:
            pass

    return out_path

@command(
    "update",
    summary="Make or update a Merge Request for updating a character's game files.",
)
async def cmd_character_update(client, msg, args):
    if len(args) < 1:
        return await client.reply(
            msg,
            "**USAGE:** `b!update [character folder name] [update title]`\nUpdates must be provided as .zip files containing all files to update, or as single XML files.",
        )

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

    update_character = args[0].lower()
    update_title = " ".join(args[1:])

    if msg.author.bot:
        return

    # Ensure the target character exists first:
    listing = await linecount.get_roster("master")

    if update_character not in listing:
        return await client.reply(
            msg,
            "❌  That doesn't appear to be a valid character folder! Make sure to type it exactly.",
        )

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

    mask = await get_character_mask(client, update_character)

    # Check permissions:
    if not (await character_permissions_check(client, update_character, msg.author.id)):
        await client.error_notify(
            "Unauthorized user {}#{} attempted to update character {}".format(
                user.name, user.discriminator, update_character
            )
        )

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

    if len(msg.attachments) < 1:
        return await client.reply(
            msg,
            "❌  You didn't send an update zip! (Upload it like any other file, but add this command as the comment!)",
        )

    open_mrs = await get_merge_requests(client, config.upstream_project_id, "opened")
    update_mr = None
    creating_new_mr = True

    for mr in open_mrs:
        if (mr["source_project_id"] == config.staging_project_id) and (
            mr["source_branch"] == "bot-update-" + update_character
        ):
            update_mr = mr
            creating_new_mr = False
            break

    if len(update_title.strip()) <= 0:
        if creating_new_mr:
            return await client.reply(
                msg,
                "❌  You didn't give the commit a description! (Type one after your character's name.)",
            )
        else:
            update_title = "Corrections to the previous commit"

    update_attach = msg.attachments[0]
    _, ext = osp.splitext(update_attach.filename)

    if ext.lower() != ".zip":
        return await client.reply(
            msg, "❌  Sorry, I can only accept updates as .zip files."
        )

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

    sess = get_http_session()

    update_title = update_title.strip()

    updated_behaviour = False
    updated_meta = False
    updated_images = []

    async with git_ops.GIT_LOCK:
        await msg.channel.send("⌛  Preparing character update...")

        try:
            await git_ops.prepare_character_update(client, update_character)
        except asyncio.TimeoutError:
            await msg.channel.send("❌  Update initialization timed out!")
            raise

        n_actions = 0
        with tempfile.NamedTemporaryFile() as tp:
            await update_attach.save(tp.name)
            tp.seek(0)

            # Build actions list:
            try:
                with zipfile.ZipFile(tp, "r") as zf:
                    info = zf.infolist()
                    for zinfo in info:
                        if zinfo.is_dir():
                            continue

                        filename = normalize_zip_path(zinfo.filename, update_character)
                        basename = filename.name
                        ext = filename.suffix

                        # ignore dotfiles:
                        if basename.startswith(".") and basename != ".gitignore":
                            continue

                        if basename == "behaviour.xml":
                            updated_behaviour = True
                        elif basename == "meta.xml":
                            updated_meta = True
                        elif basename == "tags.xml":
                            await client.reply(
                                msg, "⚠  Skipping update to `tags.xml`..."
                            )
                            continue

                        file_path = Path(
                            config.ops_repo_dir,
                            "opponents",
                            update_character,
                            filename,
                        )
                        if ext.lower() == ".png":
                            updated_images.append(
                                osp.join("opponents", update_character, filename)
                            )

                        with zf.open(zinfo, "r") as inner_file:
                            data = inner_file.read()

                            # convert line-endings for text files
                            if ext == ".xml" or ext == ".txt":
                                data = data.replace(
                                    WINDOWS_LINE_ENDING, UNIX_LINE_ENDING
                                )

                            parent = file_path.parent
                            if not parent.is_dir():
                                parent.mkdir(parents=True, exist_ok=True)

                            with file_path.open("wb") as update_file:
                                update_file.write(data)

                        n_actions += 1
            except zipfile.BadZipFile:
                await client.log_exception(
                    "while opening zipfile from **{}#{}** for updating `{}`".format(
                        msg.author.name,
                        msg.author.discriminator,
                        update_character,
                    )
                )

                return await client.reply(
                    msg, "❌  You don't appear to have sent a valid zip file."
                )

        if n_actions <= 0:
            return await client.reply(
                msg, "❌  You don't appear to have sent any files to update."
            )

        if not updated_meta:
            return await client.reply(
                msg,
                "❌  You must supply an updated `meta.xml` as part of your update files.",
            )

        executed_by = ""
        if mask is None or not config.masquerade_enabled:
            executed_by = "{}#{}".format(msg.author.name, msg.author.discriminator)
        else:
            executed_by = await mask.get_name()

        commit_msg = format_name(update_character) + ": " + update_title
        comment_msg = "New commit added by {:s}: **{:s}**".format(
            executed_by, update_title
        )

        try:
            if updated_behaviour:
                await msg.author.send(
                    "Updates to `behaviour.xml` detected. Checking linecounts..."
                )

                new_update_msg, master_msg = await compute_update_msgs(update_character)

                commit_msg += "\n" + new_update_msg
                comment_msg += "\n\nNew update linecounts:" + master_msg
        except linecount.LinecountdError as e:
            await msg.channel.send(
                "    ⚠  Encountered error while checking linecounts:\n```{}```".format(
                    str(e)
                )
            )

        if len(updated_images) > 0:
            filesizes = {}
            prior_sz = 0
            for p in updated_images:
                sz = os.stat(osp.join(config.ops_repo_dir, p)).st_size
                filesizes[p] = sz
                prior_sz += sz

            await msg.author.send("Image updates detected. Compressing...")

            pngquant_compressed = 0
            pngquant_sz = prior_sz

            try:
                # Phase 1: PNGquant
                await run_cmd(
                    "pngquant",
                    "-f",
                    "--skip-if-larger",
                    "--ext",
                    ".png",
                    "--",
                    *updated_images
                )

                pngquant_sz = 0
                for p in updated_images:
                    new_sz = os.stat(osp.join(config.ops_repo_dir, p)).st_size
                    if new_sz < filesizes[p]:
                        pngquant_compressed += 1

                    filesizes[p] = new_sz
                    pngquant_sz += new_sz

                await msg.author.send(
                    "    - `pngquant`: Compressed {} / {} files ({:.1f} => {:.1f} KiB, {:.0%} reduction)".format(
                        pngquant_compressed,
                        len(updated_images),
                        prior_sz / 1024,
                        pngquant_sz / 1024,
                        (pngquant_sz - prior_sz) / prior_sz,
                    )
                )
            except Exception as e:
                await msg.channel.send(
                    "    ⚠  Encountered error while running `pngquant`:\n```{}```".format(
                        str(e)
                    )
                )

            try:
                # Phase 2: OptiPNG
                await run_cmd("optipng", *updated_images)

                optipng_compressed = 0
                optipng_sz = 0
                for p in updated_images:
                    new_sz = os.stat(osp.join(config.ops_repo_dir, p)).st_size

                    optipng_sz += new_sz
                    if new_sz < filesizes[p]:
                        optipng_compressed += 1

                await msg.author.send(
                    "    - `optipng`: Compressed {} / {} files ({:.1f} => {:.1f} KiB, {:.0%} reduction)".format(
                        optipng_compressed,
                        len(updated_images),
                        pngquant_sz / 1024,
                        optipng_sz / 1024,
                        (optipng_sz - pngquant_sz) / pngquant_sz,
                    )
                )
            except Exception as e:
                await msg.channel.send(
                    "    ⚠  Encountered error while running `optipng`:\n```{}```".format(
                        str(e)
                    )
                )

        await git_ops.run_git_command(
            ["add", "--", osp.join("opponents", update_character)]
        )

        commit_msg += "\n\nCommand executed by: {}\n".format(executed_by)

        await msg.author.send("Finalizing update commit...")

        try:
            await git_ops.finalize_character_update(
                client, executed_by + " <spnati.official@gmail.com>", commit_msg
            )
        except git_ops.CommandError as e:
            _, stdout, _, cmd, returncode = e.args
            if cmd == "commit" and "nothing to commit, working tree clean" in stdout:
                return await client.reply(
                    msg,
                    "❌  Could not find any files that need to be updated. (Did you upload an update twice?)",
                )
            else:
                await client.reply(
                    msg,
                    "❌  `{}` failed with error code {}.".format(cmd, returncode),
                )
                raise e

        _, commit_sha = await git_ops.run_git_command(["rev-parse", "HEAD"])
        _, commit_short_sha = await git_ops.run_git_command(
            ["rev-parse", "--short", "HEAD"]
        )

        await msg.author.send(
            "Created commit `{}` successfully!".format(commit_short_sha[:-1])
        )

        update_branch = "bot-update-" + update_character
        # await msg.author.send("Replicating to backup site...")

        # prev_ssh_cmd = os.environ.get("GIT_SSH_COMMAND", "ssh")
        # try:
        #     os.environ["GIT_SSH_COMMAND"] = "ssh -i " + os.path.expanduser(
        #         "~/.ssh/mirror_key_rsa"
        #     )
        #     await git_ops.run_git_command(
        #         [
        #             "push",
        #             "--porcelain",
        #             "--force",
        #             "backup-site",
        #             update_branch + ":" + update_branch,
        #         ]
        #     )
        # except asyncio.TimeoutError:
        #     await msg.channel.send("Error: replication push timed out")
        #     raise
        # finally:
        #     os.environ["GIT_SSH_COMMAND"] = prev_ssh_cmd

        # await client.log_notify(
        #     "Pushed commit `{}` to backup site: https://git.faraway-vision.io/spnati-utils-bot/spnati/commit/{}".format(
        #         commit_sha, commit_sha
        #     )
        # )

        await msg.author.send("Submitting to staging repository...")

        try:
            await git_ops.run_git_command(
                [
                    "push",
                    "--porcelain",
                    "--force",
                    "-u",
                    config.staging_remote_name,
                    update_branch + ":" + update_branch,
                ]
            )
        except asyncio.TimeoutError:
            await msg.channel.send("❌  Staging push timed out!")
            raise

        await client.log_notify(
            "Pushed commit `{}` to staging: https://gitgud.io/spnati.official/spnati/commit/{}".format(
                commit_sha, commit_sha
            )
        )

    if creating_new_mr:
        # Create a new MR:
        mr_payload = {
            "title": format_name(update_character) + ": " + update_title,
            "source_branch": "bot-update-" + update_character,
            "target_branch": "master",
            "target_project_id": config.upstream_project_id,
            "remove_source_branch": True,
            "allow_collaboration": True,
            "labels": "Awaiting Review",
            "description": "This merge request was created by {} via the SPNATI Utilities Bot.".format(
                executed_by
            ),
        }

        async with sess.post(
            gitlab_api_url(config.staging_project_id, "/merge_requests"),
            headers={
                "Private-Token": config.git_apikey,
                "Content-Type": "application/json",
            },
            json=mr_payload,
        ) as res:
            data = await res.json()
            if res.status < 200 or res.status > 299:
                print(str(data), file=sys.stderr)

                await client.error_notify(
                    "Error making MR for branch `{}`: status {}".format(
                        "bot-update-" + update_character, res.status
                    ),
                    ext_data=str(data),
                    ext_data_name="error.log",
                )

                if res.status == 500:
                    await client.reply(
                        msg,
                        "❌  Could not make MR for update. Please try again in a few minutes.",
                    )
                else:
                    await client.reply(
                        msg,
                        "❌  Could not make MR for update: error {}".format(res.status),
                    )

                return

            update_mr = data

    character_channel = find_character_channel(client, update_character)

    if not creating_new_mr:
        await client.log_notify(
            "User **{}** updated Merge Request **!{}.**".format(
                executed_by, update_mr["iid"]
            )
        )

        if character_channel is not None and msg.channel.id != character_channel.id:
            await character_channel.send(
                "🔄  Merge request **!{}** updated: {}".format(
                    update_mr["iid"], update_mr["web_url"]
                )
            )

        if updated_behaviour:
            if len(comment_msg) > 1000000:
                comment_msg = comment_msg[:1000000]

            note_url = (
                gitlab_api_url(
                    config.upstream_project_id,
                    "/merge_requests/{}/notes".format(update_mr["iid"]),
                )
                + "?body="
                + urllib.parse.quote_plus(comment_msg)
            )

            async with sess.post(
                note_url, headers={"Private-Token": config.git_apikey}
            ) as res:
                if res.status < 200 or res.status > 299:
                    data = await res.json()
                    print(str(data))

                    await client.error_notify(
                        "Error adding comment to MR !{}: status {}`".format(
                            update_mr["iid"], res.status
                        ),
                        ext_data=str(data),
                        ext_data_name="error.log",
                    )

                    if res.status == 500:
                        await client.reply(
                            msg,
                            "⚠  Could not add comment for update. Skipping comment...",
                        )
                    else:
                        await client.reply(
                            msg,
                            "⚠  Could not add comment for update: error {}. Skipping comment...".format(
                                res.status
                            ),
                        )

        return await client.reply(
            msg,
            "✅  Update pushed successfully!\nℹ  Your update merge request is still open at: "
            + update_mr["web_url"],
        )
    else:
        await client.log_notify(
            "User **{}#{}** opened Merge Request **!{}** via bot command: {}".format(
                msg.author.name,
                msg.author.discriminator,
                update_mr["iid"],
                update_mr["web_url"],
            )
        )

        if character_channel is not None and msg.channel.id != character_channel.id:
            await character_channel.send(
                "🔄  Update merge request **!{}** opened: {}".format(
                    update_mr["iid"], update_mr["web_url"]
                )
            )

        return await client.reply(
            msg,
            "✅  Update pushed successfully!\nℹ  A new merge request has automatically been opened for you at: "
            + update_mr["web_url"],
        )


code:
import asyncio
import sys

from .handler import command
from ..config import config
from ..git import git_ops
from ..git.info import (
    find_user,
    find_user_project,
    remote_branch_exists,
    GitlabProjectNotFound,
    GitlabUserNotFound,
)

@command("deploy-test", "Deploy a branch from a repo fork to a testing environment.")
async def cmd_deploy_testing(client, msg, args):
    if len(args) < 1:
        return await client.reply(
            msg, "**USAGE:** `b!deploy-test `[Git user] [branch]`"
        )

    if msg.author.bot:
        return

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

    src_user = args[0]
    src_branch = args[1]

    # Check roles:
    if msg.author.id not in config.authorized_users:
        if len(user.roles) <= 0:
            await client.error_notify(
                "Unauthorized user {}#{} attempted to deploy branch `{}` from Git user `{}`".format(
                    user.name, user.discriminator, src_branch, src_user
                )
            )
            return await client.reply(
                msg,
                "You need at least 1 development server role to access this command.",
            )

    if not (src_user.startswith("https://") and src_user.endswith(".git")):
        # Check to ensure source user and branch exist...
        try:
            user = await find_user(client, src_user)
        except GitlabUserNotFound:
            return await client.reply(
                msg,
                "I couldn't find any user known as `@{}` on GitGud.io.".format(
                    src_user
                ),
            )

        try:
            user_project = await find_user_project(client, user["id"])
        except GitlabProjectNotFound:
            return await client.reply(
                msg,
                "The given user on GitGud.io doesn't appear to have forked the SPNATI repository.",
            )

        if not await remote_branch_exists(client, user_project["id"], src_branch):
            return await client.reply(
                msg,
                "The `{}/spnati` repository doesn't appear to have any branch named `{}`.".format(
                    src_user, src_branch
                ),
            )

        fetch_from_url = user_project["http_url_to_repo"]
    else:
        parts = src_user.rsplit("/", maxsplit=2)
        fetch_from_url = src_user
        src_user = parts[1]

    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:
        deploy_slug = src_user + "-" + src_branch

        await msg.channel.send(
            "Deploying `{}/{}` to `test/{}`...".format(
                src_user, src_branch, deploy_slug
            )
        )

        await git_ops.reset_ops_repo()

        await msg.channel.send("    - Fetching source branch...")

        await git_ops.run_git_command(
            [
                "fetch",
                "-f",
                fetch_from_url,
                src_branch + ":" + deploy_slug,
            ]
        )
        await git_ops.run_git_command(["merge", deploy_slug])

        # run prepare-online.sh
        await msg.channel.send("    - Preparing commit for deployment...")
        proc = await asyncio.create_subprocess_exec(
            "bash",
            "./prepare-online.sh",
            cwd=config.ops_repo_dir,
            stderr=asyncio.subprocess.PIPE,
        )

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

            await client.error_notify(
                "Failed to deploy `{}/{}` - `prepare-online.sh` failed with error code {}".format(
                    src_user, src_branch, proc.returncode
                ),
                ext_data=stderr,
                ext_data_name="stderr.log",
            )

            return await client.reply(
                msg,
                "Failed to deploy - `prepare-online.sh` failed with error code {}".format(
                    proc.returncode
                ),
            )

        # run rsync
        await msg.channel.send("    - Copying files to deployment area...")
        proc = await asyncio.create_subprocess_exec(
            "rsync",
            "-rltz",
            "--delete",
            ".public/",
            "/mnt/disks/data/spnati-testing-environments/" + deploy_slug,
            cwd=config.ops_repo_dir,
            stderr=asyncio.subprocess.PIPE,
        )

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

            await client.error_notify(
                "Failed to deploy `{}/{}` - `rsync` failed with error code {}".format(
                    src_user, src_branch, proc.returncode
                ),
                ext_data=stderr,
                ext_data_name="stderr.log",
            )

            return await client.reply(
                msg,
                "Failed to deploy - `rsync` failed with error code {}".format(
                    proc.returncode
                ),
            )

        await client.log_notify(
            "<@!135165531908079616> User **{}#{}** deployed branch `{}/{}` to testing at https://spnati.faraway-vision.io/test/{}/index.html".format(
                msg.author.name,
                msg.author.discriminator,
                src_user,
                src_branch,
                deploy_slug,
            )
        )

        return await client.reply(
            msg,
            "Deployed `{}/{}` to testing at https://spnati.faraway-vision.io/test/{}/index.html".format(
                src_user, src_branch, deploy_slug
            ),
        )


code:
import asyncio
from datetime import datetime, timedelta
import io
import sys

import discord
from matplotlib import pyplot as plt

from .handler import command

@command("graph", summary="Generate graphs showing usage and bug report activity.")
async def cmd_graph(client, msg, args):
    n_days = args[0]
    cmd = [sys.executable, "./spnati_utils_bot/generate_report_graph.py", n_days]

    cmd.extend(args[1:])

    proc = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE)
    stdout_data, _ = await proc.communicate()

    if proc.returncode == 0:
        fig_io = io.BytesIO(stdout_data)
        await client.reply(
            msg,
            "Graph of usage and bug report counts over the past **{} days**:".format(
                n_days
            ),
            file=discord.File(fig_io, "graph.png"),
        )
        fig_io.close()

async def get_span_retention(db, start, end):
    query = [
        {
            "$match": {
                "type": {"$in": ["start_game"]},
                "date": {"$gte": start, "$lt": end},
            }
        },
        {"$group": {"_id": None, "ips": {"$addToSet": "$ip"}}},
    ]

    cursor = db.aggregate(query)
    doc = (await cursor.to_list(1))[0]
    return set(doc["ips"])

@command("retention", summary="Generate graphs showing weekly retention rates.")
async def cmd_graph_retention_rate(client, msg, args):
    db = client.db.usage_reports

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

    now = datetime.utcnow()
    offset_base = datetime(now.year, now.month, now.day)
    if offset_base.weekday() > 0:
        offset_base -= timedelta(days=offset_base.weekday())

    total = []
    retained = []
    unretained = []
    weeks = []
    week_starts = []

    last_week_ips = None

    for week_no in range(n_weeks, 0, -1):
        week_start = offset_base - timedelta(days=7 * week_no)
        week_end = week_start + timedelta(days=7)

        cur_ips = await get_span_retention(db, week_start, week_end)
        n_total = len(cur_ips)

        if last_week_ips is not None:
            n_retained = len(last_week_ips & cur_ips)

            retained.append(n_retained)
            unretained.append(n_total - n_retained)

            print(
                "Week {:2d} - total {:5d}, retained {:5d}, retention {:.2%}".format(
                    week_no, n_total, n_retained, n_retained / n_total
                )
            )
        else:
            retained.append(0)
            unretained.append(n_total)

            print("Week {:2d} - total {:5d}".format(week_no, n_total))

        last_week_ips = cur_ips

        total.append(n_total)
        weeks.append(week_no)
        week_starts.append(week_start.date())

    weeks.reverse()

    width = 0.35
    fig = plt.figure(figsize=(12.80, 7.20))
    ax = fig.add_subplot(111, title="Inter-Week Retention Rates")

    ax.bar(weeks, retained, width, label="Retained Users")
    ax.bar(weeks, unretained, width, bottom=retained, label="Unretained Users")

    for week_no, n_total, n_retained in zip(weeks[1:], total[1:], retained[1:]):
        ax.text(
            week_no,
            n_total + 250,
            "{:.2%}".format(n_retained / n_total),
            horizontalalignment="center",
        )

    ax.set_ylabel("Users")
    ax.set_xticks(weeks)
    ax.set_xticklabels(list(str(start_day) for start_day in week_starts))
    ax.legend()

    with io.BytesIO() as bio:
        fig.savefig(bio, format="png", bbox_inches="tight")
        bio.seek(0)

        outfile = discord.File(bio, filename="graph.png")
        await client.reply(
            msg,
            "Inter-week retention rates over the past **{:d} weeks:**".format(n_weeks),
            file=outfile,
        )

    plt.close(fig)


code:
import asyncio
import re

import discord
import sentry_sdk

from prometheus_client import Histogram
from prometheus_async.aio import time

from ..config import config
from ..tcnati_interface import handle_tcnati_message

commands = {}
CMD_REGEX = r"\"([^\"]+)\"|\'([^\']+)\'|\`\`\`([^\`]+)\`\`\`|\`([^\`]+)\`|(\S+)"

CMD_LATENCY = Histogram(
    "utilities_cmd_latency_seconds", "Time spent processing bot commands", ["command"]
)

class Command(object):
    def __init__(self, func, name, summary=None, authorized_only=False):
        self.func = func
        self.name = name
        self.summary = summary
        self.authorized_only = authorized_only

def command(name, summary=None, authorized_only=False):
    global commands

    def wrapper(func):
        commands[name] = Command(
            func, name, summary=summary, authorized_only=authorized_only
        )
        return func

    return wrapper

async def handle_cmd_string(client, msg, cmd_string):
    global commands

    args = []
    for m in re.finditer(CMD_REGEX, cmd_string):
        for group in m.groups():
            if group is not None:
                args.append(group)
                break

    cmd = args[0].lower().strip()
    args = args[1:]

    with sentry_sdk.configure_scope() as scope:
        scope.set_tag("command", cmd)
        scope.set_extra("args", args)

        if cmd in commands:
            cmd_obj = commands[cmd]

            if not cmd_obj.authorized_only:
                async with msg.channel.typing():
                    with CMD_LATENCY.labels(cmd).time():
                        return await cmd_obj.func(client, msg, args)

            if (
                (msg.author.id not in config.authorized_users)
                and not (
                    cmd == "version"
                    and msg.author.id in config.version_authorized_users
                )
                and (msg.channel.id not in config.listen_channels)
                and (not isinstance(msg.channel, discord.DMChannel))
            ):
                return await handle_tcnati_message(client, msg)

            async with msg.channel.typing():
                with CMD_LATENCY.labels(cmd).time():
                    return await cmd_obj.func(client, msg, args)
        else:
            return await client.reply(
                msg, "That doesn't appear to be a valid command. Try `b!help`?"
            )


code:
from __future__ import annotations

import discord
import hashlib
import re
import time
from typing import Optional

from .. import masquerade, main
from ..config import config
from .handler import command
from .slash_cmd import (
    slash_command,
    SlashCommandChannelOption,
    SlashCommandStringOption,
)
from ..interactions import InteractionContext
from .. import utils

MENTION_REGEX = r"<@!?(\d+)>"
CHANNEL_MENTION_REGEX = r"<#(\d+)>"

mask_cmd_group = slash_command("mask", "Perform operations with Masks.")

def parse_user_id(raw_uid):
    try:
        return int(raw_uid)
    except ValueError:
        m = re.match(MENTION_REGEX, raw_uid)
        if m is not None:
            return int(m.group(1))

        raise ValueError("Could not parse user ID: " + raw_uid)

def parse_channel_id(raw_cid):
    try:
        return int(raw_cid)
    except ValueError:
        m = re.match(CHANNEL_MENTION_REGEX, raw_cid)
        if m is not None:
            return int(m.group(1))

        raise ValueError("Could not parse channel ID: " + raw_cid)

async def format_mask_info(client, mask_id):
    mask = masquerade.Mask(client, mask_id)
    name = await mask.get_name()

    if name is None:
        return "**Mask {}:** (not configured)".format(mask_id)

    s = "**Mask {}**: _{}_".format(mask_id, name)
    character = await mask.get_character()

    if character is not None:
        s += "\n**Character:** " + utils.format_name(character)
    else:
        s += "\n**Character:** (none set)"

    channel_id = await mask.get_channel_id()
    if channel_id is not None:
        s += "\n**Channel:** <#{:d}>".format(channel_id)
    else:
        s += "\n**Channel:** (none set)"

    members = await mask.get_users()
    dev_server = client.get_guild(config.primary_server_id)
    names = []

    for user_id in members:
        user = await dev_server.fetch_member(user_id)
        names.append("{}#{}".format(user.name, user.discriminator))

    s += "\n**Members:** " + ", ".join(names)
    return s

@command("list-masks", "List all configured masks.", authorized_only=True)
async def cmd_list_masks(client, msg, args):
    if msg.author.id not in config.authorized_users:
        return await client.reply(
            msg, "Error: You are not authorized to access this command."
        )

    if not masquerade.private_channel_check(msg.channel):
        return await client.reply(msg, "Error: This is not a private channel.")

    s = "**Configured Masks:**"

    mask_ids = await masquerade.get_all_mask_ids(client)
    for mask_id in mask_ids:
        mask = masquerade.Mask(client, mask_id)
        name = await mask.get_name()

        if name is not None:
            s += "\nMask {:d}: **{:s}**".format(mask_id, name)
        else:
            s += "\nMask {:d}: (not configured)".format(mask_id)

    return await client.reply(msg, s)

@command("mask", "Configure mask settings.", authorized_only=True)
async def cmd_configure_mask(client, msg, args):
    if msg.author.id not in config.authorized_users:
        return await client.reply(
            msg, "Error: You are not authorized to access this command."
        )

    if not masquerade.private_channel_check(msg.channel):
        return await client.reply(msg, "Error: This is not a private channel.")

    logging_channel = client.get_channel(config.masquerade["logging_channel"])

    mask_id = int(args[0])
    operation = args[1].lower()

    if operation == "info":
        info = await format_mask_info(client, mask_id)
        return await client.reply(msg, info)
    elif operation == "enable":
        if await masquerade.is_mask_enabled(client, mask_id):
            return await client.reply(msg, "That mask is already enabled.")
        else:
            await masquerade.enable_mask_id(client, mask_id)
            return await client.reply(msg, "Mask {} enabled.".format(mask_id))
    elif operation == "disable":
        if not (await masquerade.is_mask_enabled(client, mask_id)):
            return await client.reply(msg, "That mask is already disabled.")
        else:
            await masquerade.disable_mask_id(client, mask_id)
            return await client.reply(msg, "Mask {} disabled.".format(mask_id))

    mask = masquerade.Mask(client, mask_id)

    if operation == "set-channel":
        channel_id = parse_channel_id(args[2])
        await mask.set_channel(channel_id)

        await logging_channel.send(
            "{}#{} configured primary channel for Mask {:d}: <#{:d}>".format(
                msg.author.name, msg.author.discriminator, mask_id, channel_id
            )
        )

        return await client.reply(
            msg, "Mask {:d} primary channel configured.".format(mask_id)
        )
    elif operation == "set-name":
        args = msg.content.split(" ", 3)
        name = args[3]
        await mask.set_name(name)

        await logging_channel.send(
            "{}#{} configured name for Mask {:d}: **{:s}**".format(
                msg.author.name, msg.author.discriminator, mask_id, name
            )
        )

        return await client.reply(
            msg, "Configured name for Mask {:d}: **{:s}**".format(mask_id, name)
        )
    elif operation == "set-users":
        users = list(map(parse_user_id, args[2:]))
        await mask.set_users(users)

        await logging_channel.send(
            "{}#{} configured users for Mask {:d}: {:s}".format(
                msg.author.name,
                msg.author.discriminator,
                mask_id,
                ",".join("<@!{:d}>".format(uid) for uid in users),
            )
        )

        return await client.reply(
            msg, "Configured {:d} members for Mask {:d}".format(len(users), mask_id)
        )
    elif operation == "set-character":
        await mask.set_character(args[2])

        await logging_channel.send(
            "{}#{} configured character for Mask {:d}: {:s}".format(
                msg.author.name, msg.author.discriminator, mask_id, args[2]
            )
        )

        return await client.reply(
            msg, "Assigned character `{:s}` to Mask {:d}".format(args[2], mask_id)
        )

def calculate_message_id(
    msg: str, author: discord.Member, dest: discord.TextChannel
) -> str:
    timestamp = int(time.time() * 1000)
    m = hashlib.sha1()

    m.update(timestamp.to_bytes(8, byteorder="little"))
    m.update(int(author.id).to_bytes(8, byteorder="little"))
    m.update(dest.id.to_bytes(8, byteorder="little"))
    m.update(msg.encode("utf-8"))

    return m.hexdigest()

async def send_mask_msg(
    client: main.NotifierClient,
    msg: str,
    author: discord.Member,
    mask: masquerade.Mask,
    dest: discord.TextChannel,
):
    msg = msg[:1850]  # truncate to fit total message length constraint
    msg_id = calculate_message_id(msg, author, dest)

    mask_name = await mask.get_name()
    if mask_name is not None:
        formatted_msg = "Mask **{:s}** says:\n".format(mask_name)
    else:
        formatted_msg = "Mask **#{:d}** says:\n".format(mask.id)
    formatted_msg += msg

    logging_channel = client.get_channel(config.masquerade["logging_channel"])
    await logging_channel.send(
        formatted_msg
        + "\nMessage sent by <@!{}> in <#{}>\nMessage ID: `{:s}`".format(
            author.id, dest.id, msg_id
        )
    )

    mask_channel_id = await mask.get_channel_id()
    mask_channel = client.get_channel(mask_channel_id)

    sent_msg: discord.Message = await dest.send(formatted_msg)
    await mask_channel.send(
        "<@!{:d}> | Sent message ID `{:s}` to <#{}>.".format(author.id, msg_id, dest.id)
    )

    await client.redis.hmset(
        "masquerade:messages:" + str(sent_msg.id),
        "author",
        str(author.id),
        "mask_channel",
        str(mask_channel_id),
    )

async def handle_mask_speak_reply(client: main.NotifierClient, msg: discord.Message, ref_id: int):
    msg_data = await client.redis.hgetall("masquerade:messages:" + str(ref_id), encoding="utf-8")
    notification_msg = "<@!{:s}> | **{:s}** replied to your message:\n".format(
        msg_data["author"],
        msg.author.display_name
    )

    notification_msg += ">>> " + msg.clean_content

    mask_channel = client.get_channel(int(msg_data["mask_channel"]))
    await mask_channel.send(notification_msg)

async def find_user_mask(
    client: main.NotifierClient, user: discord.Member
) -> Optional[masquerade.Mask]:
    # NOTE: This could probably be done better by just storing a direct mapping
    # from user IDs to masks, but this endpoint shouldn't be hit too often,
    # relatively speaking, so what's the harm?
    #
    # Plus, this obviates the need for a specific permissions check.

    mask_ids = await masquerade.get_all_mask_ids(client)
    for mask_id in mask_ids:
        mask = masquerade.Mask(client, mask_id)
        if await mask.user_in_mask(user):
            return mask
    else:
        return None

@command("mask-speak", "Speak from behind a Mask.")
async def cmd_mask_speak(client, msg, args):
    if not masquerade.private_channel_check(msg.channel):
        try:
            await msg.delete()
        except:
            pass
        return await msg.author.send(
            "You used `mask-speak` in a non-private channel! I tried to delete it.\nThe message was:\n"
            + msg.clean_content
        )

    for mask_id in await masquerade.get_all_mask_ids(client):
        mask = masquerade.Mask(client, mask_id)

        if (await mask.get_channel_id()) == msg.channel.id:
            break
    else:
        return await client.reply(
            msg, "Error: This command must be used from a Mask's channel."
        )

    if not (
        (await mask.user_in_mask(msg.author))
        or msg.author.id in config.authorized_users
    ):
        return await client.reply(msg, "Error: You are not a part of this Mask.")

    args = msg.content.split(" ", 2)

    try:
        dest_channel_id = parse_channel_id(args[1])
    except ValueError:
        return await client.reply(
            msg, "Error: that doesn't appear to be a valid channel."
        )

    dest_channel = client.get_channel(dest_channel_id)
    await send_mask_msg(client, args[2], msg.author, mask, dest_channel)

@mask_cmd_group.subcommand(
    "speak",
    "Speak from behind a Mask. Works in any channel; your command will be hidden.",
    SlashCommandChannelOption(
        "channel", "The channel to send the message to.", required=True
    ),
    SlashCommandStringOption("message", "The message to send.", required=True),
)
async def slash_mask_speak(
    ctx: InteractionContext, channel: discord.TextChannel, message: str
):
    mask = await find_user_mask(ctx.client, ctx.user)
    if mask is not None:
        await send_mask_msg(ctx.client, message, ctx.user, mask, channel)
    else:
        await ctx.user.send("Error: I could not find any valid Mask for you.")

    return ctx.reply(reply_text=None, ephemeral=True)

@mask_cmd_group.subcommand(
    "here",
    "Send a message as a mask to this channel. Your command will be hidden.",
    SlashCommandStringOption("message", "The message to send.", required=True),
)
async def slash_mask_here(ctx: InteractionContext, message: str):
    mask = await find_user_mask(ctx.client, ctx.user)
    if mask is not None:
        await send_mask_msg(ctx.client, message, ctx.user, mask, ctx.channel)
    else:
        await ctx.user.send("Error: I could not find any valid Mask for you.")

    return ctx.reply(reply_text=None, ephemeral=True)

@command(
    "update-mask-speak-cmd",
    "Update the mask-speak slash command.",
    authorized_only=True,
)
async def cmd_permission_slash_update(
    client: main.NotifierClient, msg: discord.Message, args
):
    ret = await mask_cmd_group.register_global(client)
    return await client.reply(msg, "Registered slash command {}.".format(ret["id"]),)


code:
from __future__ import annotations

from aiohttp import web
from aioredis import Redis
import asyncio
import discord
import io
import json
from typing import Union

from .handler import command
from .slash_cmd import slash_command, SlashCommandUserOption, SlashCommandStringOption
from ..interactions import InteractionContext
from .. import linecount, utils, main, permissions
from ..config import config

PERMISSION_SUBCMDS = ("check", "list", "add", "remove")

class UserNotFound(Exception):
    pass

def _try_resolve(client: main.NotifierClient, user_ref: str) -> discord.Member:
    try:
        resolved = client.resolve_user(user_ref)
    except ValueError:
        resolved = None

    if resolved is None:
        raise UserNotFound(
            "Could not identify `"
            + user_ref
            + "`. Are you sure they're a Dev Server member?"
        )

    return resolved

async def cmd_check_permission(
    client: main.NotifierClient, user_ref: Union[int, str], character: str
) -> str:
    user = _try_resolve(client, user_ref)
    db_status = await permissions.db_check(client.redis, user, character)
    adm_status = user.id in config.authorized_users

    if db_status or adm_status:
        if db_status and adm_status:
            reason = "Stored Permissions + Is Moderator"
        elif db_status:
            reason = "Stored Permissions"
        elif adm_status:
            reason = "Is Moderator"

        return "{}#{} **can** edit `{}` ({}).".format(
            user.name, user.discriminator, character, reason
        )
    else:
        return "{}#{} **cannot** edit `{}`.".format(
            user.name, user.discriminator, character
        )

async def cmd_list_developers(client: main.NotifierClient, character: str) -> str:
    dev_server: discord.Guild = client.get_guild(config.primary_server_id)
    devs = await permissions.get_developers(client.redis, character)

    reply_text = "Developers for `" + character + "`:\n"
    for dev_id in devs:
        user = await dev_server.fetch_member(dev_id)
        reply_text += "    - {}#{}\n".format(user.name, user.discriminator)

    return reply_text

async def cmd_list_characters(
    client: main.NotifierClient, user_ref: Union[int, str]
) -> str:
    user = _try_resolve(client, user_ref)
    chars = await permissions.get_characters(client.redis, user)

    reply_text = "Characters for **{}#{}**:\n".format(user.name, user.discriminator)
    for character in chars:
        reply_text += "    - `" + character + "`\n"
    return reply_text

async def cmd_add_binding(
    client: main.NotifierClient, user_ref: Union[int, str], character: str
) -> str:
    user = _try_resolve(client, user_ref)
    roster = await linecount.get_roster("master")
    if character not in roster:
        return "This character doesn't seem to be part of the current roster?"

    await permissions.bind(client.redis, user, character)
    return "Attached **{}#{}** to `{}`!".format(
        user.name, user.discriminator, character
    )

async def cmd_remove_binding(
    client: main.NotifierClient, user_ref: Union[int, str], character: str
) -> str:
    user = _try_resolve(client, user_ref)
    await permissions.unbind(client.redis, user, character)
    return "Detached **{}#{}** from `{}`!".format(
        user.name, user.discriminator, character
    )

@command("permissions", "Check character editing permissions.", authorized_only=True)
async def cmd_permissions(client: main.NotifierClient, msg: discord.Message, args):
    cmd = args[0].casefold()
    reply_text = ""

    if cmd not in PERMISSION_SUBCMDS:
        return await client.reply(
            msg, "USAGE: `b!permission [" + "|".join(PERMISSION_SUBCMDS) + "]`"
        )

    try:
        if cmd == "check":
            reply_text = await cmd_check_permission(client, args[1], args[2])
        elif cmd == "list":
            if args[1].startswith("character"):
                reply_text = await cmd_list_characters(client, args[2])
            elif args[1].startswith("dev") or args[1].startswith("user"):
                reply_text = await cmd_list_developers(client, args[2])
            else:
                reply_text = (
                    "Subcommand for `list` must be either `character` or `dev`."
                )
        elif cmd == "add":
            if msg.author.id in config.authorized_users:
                reply_text = await cmd_add_binding(client, args[1], args[2])
            else:
                reply_text = "You aren't authorized to access this command."
        elif cmd == "remove":
            if msg.author.id in config.authorized_users:
                reply_text = await cmd_remove_binding(client, args[1], args[2])
            else:
                reply_text = "You aren't authorized to access this command."
    except UserNotFound as e:
        reply_text = e.args[0]

    return await client.reply(msg, reply_text)

permissions_cmd = slash_command(
    "permissions", "Check or edit character editing permissions"
)
list_group = permissions_cmd.subcommand_group("list", "List characters or developers.")

@list_group.subcommand(
    "developers",
    "List developers associated with a given character.",
    SlashCommandStringOption("character", "The character to check", required=True),
)
async def slash_cmd_list_devs(ctx: InteractionContext, character: str) -> web.Response:
    reply_text = await cmd_list_developers(ctx.client, character)
    return ctx.reply(reply_text, ephemeral=True)

@list_group.subcommand(
    "characters",
    "List characters associated with a given developer.",
    SlashCommandUserOption("user", "The user to check", required=True),
)
async def slash_cmd_list_characters(
    ctx: InteractionContext, user: discord.Member
) -> web.Response:
    reply_text = await cmd_list_characters(ctx.client, user.id)
    return ctx.reply(reply_text, ephemeral=True)

@permissions_cmd.subcommand(
    "check",
    "Check if a user can work on a given character.",
    SlashCommandUserOption("user", "The user to check", required=True),
    SlashCommandStringOption("character", "The character to check", required=True),
)
async def slash_cmd_check_permissions(
    ctx: InteractionContext, user: discord.Member, character: str
) -> web.Response:
    reply_text = await cmd_check_permission(ctx.client, user.id, character)
    return ctx.reply(reply_text, ephemeral=True)

@permissions_cmd.subcommand(
    "add",
    "Add a user to a character permissions group.",
    SlashCommandUserOption("user", "The user to modify.", required=True),
    SlashCommandStringOption(
        "character", "The target character / permissions group.", required=True
    ),
    authorized_only=True,
)
async def slash_cmd_add_permissions(
    ctx: InteractionContext, user: discord.Member, character: str
) -> web.Response:
    roster = await linecount.get_roster("master")
    if character not in roster:
        return ctx.reply(
            "This character doesn't seem to be part of the current roster?",
            ephemeral=True,
        )
    await permissions.bind(ctx.client.redis, user, character)
    return ctx.reply(
        "Attached <@!{}> to `{}`!".format(user.id, character), suppress_mentions=True
    )

@permissions_cmd.subcommand(
    "remove",
    "Remove a user from a character permissions group.",
    SlashCommandUserOption("user", "The user to modify.", required=True),
    SlashCommandStringOption(
        "character", "The target character / permissions group.", required=True
    ),
    authorized_only=True,
)
async def slash_cmd_remove_permissions(
    ctx: InteractionContext, user: discord.Member, character: str
) -> web.Response:
    await permissions.unbind(ctx.client.redis, user, character)
    return ctx.reply(
        "Detached <@!{}> from `{}`!".format(user.id, character), suppress_mentions=True
    )

@permissions_cmd.subcommand(
    "sync",
    "Synchronize stored permissions with character roles.",
    authorized_only=True,
)
async def slash_cmd_sync_permissions(ctx: InteractionContext) -> web.Response:
    roster = await linecount.get_roster("master")
    dev_server: discord.Guild = ctx.client.get_guild(config.primary_server_id)

    normalized = {}
    for character in roster:
        normalized[utils.normalize_folder_name(character)] = character

    name_map = {}
    bindings = set()

    for role in dev_server.roles:
        if role.is_bot_managed() or role.is_default() or role.is_premium_subscriber():
            continue

        norm_name: str = utils.normalize_role_name(role.name)
        if (
            norm_name == "moderator"
            or norm_name == "workingonacharacter"
            or norm_name.startswith("made")
            or norm_name.startswith("coder")
        ):
            continue

        if norm_name in normalized:
            character = normalized[norm_name]
            name_map[role.name] = character

            for member in role.members:
                bindings.add((member.id, character))

    dev_map = {}  # Dev => Character
    char_map = {}  # Character => Dev

    for dev_id, char_id in bindings:
        dev_map.setdefault(dev_id, set()).add(char_id)
        char_map.setdefault(char_id, set()).add(dev_id)

    tr = ctx.client.redis.multi_exec()

    for char_id, dev_ids in char_map.items():
        tr.sadd("permissions:character:" + char_id, *map(str, dev_ids))

    for dev_id, chars in dev_map.items():
        tr.sadd("permissions:user:" + str(dev_id), *chars)

    await tr.execute()

    return ctx.reply(
        "Synchronized {} bindings for {} characters and {} developers.".format(
            len(bindings), len(char_map), len(dev_map)
        ),
    )

@command(
    "update-permissions-cmd",
    "Update the permissions slash command and its subcommands.",
    authorized_only=True,
)
async def cmd_permission_slash_update(
    client: main.NotifierClient, msg: discord.Message, args
):
    ret = await permissions_cmd.register_global(client)
    return await client.reply(
        msg,
        "Registered command {}.".format(ret["id"]),
    )