~nova/fletcher

fletcher/github.py -rw-r--r-- 5.9 KiB
ece10afa — Novalinium f-string issue 9 days ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
from sys import exc_info
import aiohttp
import discord
import logging
import messagefuncs
import os

logger = logging.getLogger("fletcher")

base_url = "https://api.github.com"


async def github_search_function(message, client, args):
    try:
        if "Guild " + str(message.guild.id) in config:
            scoped_config = config["Guild " + str(message.guild.id)]
        else:
            await messagefuncs.sendWrappedMessage(
                "You can't search issues on GitHub until the server admin sets the associated `github-repo` key in the server configuration at https://fletcher.fun",
                message.author,
            )
            raise Exception(
                "No guild-specific configuration for source code management on guild "
                + str(message.guild)
            )
        get_url = "/search/issues"
        async with aiohttp.ClientSession(
            headers={
                "Authorization": f"token {config['github']['personalAccessToken']}"
            }
        ) as session:
            url = base_url + get_url
            async with session.get(
                url,
                params={"q": " ".join(args) + " repo:" + scoped_config["github-repo"]},
            ) as response:
                response_body = await response.json()
                if response_body["total_count"] == 0:
                    return await messagefuncs.sendWrappedMessage(
                        "No results found.", message.channel
                    )
                elif response_body["total_count"] <= 5:
                    for issue in response_body["items"]:
                        await messagefuncs.sendWrappedMessage(
                            target=message.channel,
                            embed=issue_to_embed(issue, issue["user"]["login"]),
                        )
                else:
                    for issue in response_body["items"][:5]:
                        await messagefuncs.sendWrappedMessage(
                            target=message.channel,
                            embed=issue_to_embed(issue, issue["user"]["login"]),
                        )
                    await messagefuncs.sendWrappedMessage(
                        f"Results truncated, {response_body['total_count']} total responses",
                        message.channel,
                    )
    except Exception as e:
        exc_type, exc_obj, exc_tb = exc_info()
        logger.error(f"GSF[{exc_tb.tb_lineno}]: {type(e).__name__} {e}")


async def github_report_function(message, client, args):
    try:
        if "Guild " + str(message.guild.id) in config:
            scoped_config = config["Guild " + str(message.guild.id)]
        else:
            await messagefuncs.sendWrappedMessage(
                "You can't report issues to GitHub until the server admin sets the `github-repo` key in the server configuration at https://fletcher.fun",
                message.author,
            )
            raise Exception(
                "No guild-specific configuration for source code management on guild "
                + str(message.guild)
            )
        post_url = "/repos/" + scoped_config["github-repo"] + "/issues"
        async with aiohttp.ClientSession(
            headers={
                "Authorization": "token " + config["github"]["personalAccessToken"]
            }
        ) as session:
            url = base_url + post_url
            content = "\n".join(message.content.splitlines(True)[1:])
            content = (
                content
                + "\n reported on behalf of "
                + message.author.display_name
                + " via [Fletcher](fletcher.fun) on Discord ("
                + message.guild.name
                + ":"
                + message.channel.name
                + ")"
            )
            title = " ".join(message.content.splitlines()[0].split(" ")[1:])
            async with session.post(
                url, json={"title": title, "body": content}
            ) as response:
                response_body = await response.json()
                await messagefuncs.sendWrappedMessage(
                    target=message.channel,
                    embed=issue_to_embed(response_body, message.author.display_name),
                )
    except Exception as e:
        exc_type, exc_obj, exc_tb = exc_info()
        logger.error(f"GRF[{exc_tb.tb_lineno}]: {type(e).__name__} {e}")


def issue_to_embed(issue, author_name):
    if not issue["body"]:
        issue["body"] = "*Empty*"
        body_inline = True
    else:
        body_inline = False
        if len(issue["body"]) > 1024:
            issue["body"] = issue["body"][:1023] + "…"
    return (
        discord.Embed(
            title="#" + str(issue["number"]) + ": " + issue["title"],
            url=issue["html_url"],
        )
        .add_field(name="Status", value=issue["state"].capitalize(), inline=True)
        .add_field(name="Content", value=issue["body"], inline=body_inline)
        .set_footer(
            icon_url="https://download.nova.anticlack.com/fletcher/github.favicon.png",
            text=f"On behalf of {author_name}",
        )
    )


def autoload(ch):
    ch.add_command(
        {
            "trigger": ["!ghreport"],
            "function": github_report_function,
            "async": True,
            "args_num": 1,
            "args_name": ["Line 1: Title", "[Line 2: Body]"],
            "description": "GitHub Issue Reporter (if server has repository configured)",
            "long_run": True,
        }
    )
    ch.add_command(
        {
            "trigger": ["!ghsearch"],
            "function": github_search_function,
            "async": True,
            "args_num": 1,
            "args_name": [
                "Query String (https://help.github.com/en/github/searching-for-information-on-github/understanding-the-search-syntax)"
            ],
            "description": "GitHub Issue Search",
            "long_run": True,
        }
    )


async def autounload(ch):
    pass