284 lines
10 KiB
Python
284 lines
10 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import typing
|
|
from typing import TYPE_CHECKING
|
|
|
|
import httpx
|
|
from github import Github, Issue, PaginatedList
|
|
from pathlib import Path
|
|
import niobot
|
|
|
|
if typing.TYPE_CHECKING:
|
|
from ..main import NonsenseBot
|
|
|
|
|
|
class MSCGetter(niobot.Module):
|
|
if TYPE_CHECKING:
|
|
bot: "NonsenseBot"
|
|
log = logging.getLogger(__name__)
|
|
|
|
def __init__(self, bot):
|
|
super().__init__(bot)
|
|
self.latest_msc = None
|
|
self.msc_cache = Path.cwd() / ".msc-cache"
|
|
self.msc_cache.mkdir(parents=True, exist_ok=True)
|
|
self.gh = Github()
|
|
|
|
async def get_msc_with_cache(self, number: int) -> dict:
|
|
if number not in range(1, 10_000):
|
|
return {"error": "Invalid MSC ID"}
|
|
|
|
file = self.msc_cache / ("%d.json" % number)
|
|
content = None
|
|
if file.exists():
|
|
try:
|
|
content = json.loads(file.read_text("utf-8", "replace"))
|
|
except json.JSONDecodeError:
|
|
file.unlink()
|
|
|
|
if content:
|
|
return content
|
|
|
|
self.log.debug("Requesting MSC: %d", number)
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
"https://api.github.com/repos/matrix-org/matrix-spec-proposals/issues/%d"
|
|
% number
|
|
)
|
|
if response.status_code != 200:
|
|
return {
|
|
"error": "Failed to fetch issue from GitHub (HTTP {!s}),"
|
|
"and no cached version was available.".format(response.status_code)
|
|
}
|
|
content = response.json()
|
|
file.write_text(json.dumps(content))
|
|
return content
|
|
|
|
async def search_for_msc(self, query: str) -> list[dict]:
|
|
force_fetch = "+fetch" in query
|
|
query = query.replace("+fetch", "")
|
|
found = []
|
|
for cached_msc in self.msc_cache.glob("*.json"):
|
|
data = json.loads(cached_msc.read_text())
|
|
if query.casefold() in data["title"].casefold():
|
|
found.append(data)
|
|
|
|
if len(found) < 10 or force_fetch:
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(
|
|
"https://api.github.com/search/issues",
|
|
params={
|
|
"q": "%s+is:pull-request+repo:matrix.org/matrix-spec-proposals" % query
|
|
}
|
|
)
|
|
response.raise_for_status()
|
|
issues = response.json()["items"]
|
|
except httpx.HTTPStatusError as err:
|
|
issue = {
|
|
"html_url": "https://http.cat/%d" % err.response.status_code,
|
|
"title": "There was an issue contacting GitHub's search."
|
|
}
|
|
if err.response.status_code == 429:
|
|
issue["title"] = "Ratelimited while querying github. Try again later."
|
|
found.append(issue)
|
|
except Exception:
|
|
found.append(
|
|
{
|
|
"title": "Error querying GitHub.",
|
|
"html_url": "https://http.cat/500"
|
|
}
|
|
)
|
|
else:
|
|
for issue in issues:
|
|
file = self.msc_cache / ("%d.json" % issue["number"])
|
|
file.write_text(json.dumps(issue))
|
|
found += issues
|
|
return found
|
|
|
|
@niobot.event("message")
|
|
async def on_message(self, room: niobot.MatrixRoom, message: niobot.RoomMessage):
|
|
if self.bot.is_old(message):
|
|
return
|
|
if message.sender == self.bot.user:
|
|
return
|
|
if not isinstance(message, niobot.RoomMessageText):
|
|
return
|
|
if "m.in_reply_to" in message.source.get("m.relates_to", []):
|
|
return
|
|
if await self.bot.redis.get(
|
|
self.bot.redis_key(room.room_id, "auto_msc.enabled")
|
|
):
|
|
matches = re.finditer(r"((msc)\W?)([0-9]{1,4})", message.body, re.IGNORECASE)
|
|
|
|
lines = []
|
|
for m in matches:
|
|
no = m.group(3)
|
|
if no:
|
|
data = await self.get_msc_with_cache(int(no))
|
|
if data.get("error"):
|
|
continue
|
|
lines.append(f"[{data['title']}]({data['html_url']})")
|
|
if lines:
|
|
return await self.bot.send_message(
|
|
room, "\n".join((f"* {ln}" for ln in lines)), reply_to=message
|
|
)
|
|
|
|
@staticmethod
|
|
def pr_to_display(data: dict) -> str:
|
|
return f"* [{data['title']}]({data['html_url']})"
|
|
|
|
@niobot.command()
|
|
async def msc(
|
|
self,
|
|
ctx: niobot.Context,
|
|
number_or_query
|
|
):
|
|
"""Fetches the given MSC"""
|
|
if number_or_query.startswith("?"): # search
|
|
msg = await ctx.respond("Searching for relevant MSCs with query %r..." % number_or_query[1:])
|
|
results = await self.search_for_msc(number_or_query[1:])
|
|
if not results:
|
|
return await msg.edit("No MSCs matched your query.")
|
|
|
|
lines = []
|
|
for pr in results:
|
|
lines.append(self.pr_to_display(pr))
|
|
lines_formatted = "\n".join(lines)
|
|
|
|
if len(lines) > 3:
|
|
new_lines = [
|
|
"<ul>"
|
|
]
|
|
for pr in results[:3]:
|
|
new_lines.append('<li><a href="{0[html_url]}" target="_blank">{0[title]}</a></li>'.format(pr))
|
|
new_lines += [
|
|
"</ul>",
|
|
"<details>",
|
|
"<summary>And %d more...</summary>" % (len(results) - 3),
|
|
"<ul>"
|
|
]
|
|
for pr in results[3:]:
|
|
new_lines.append('<li><a href="{0[html_url]}" target="_blank">{0[title]}</a></li>'.format(pr))
|
|
new_lines += [
|
|
"</ul>",
|
|
"</details>"
|
|
]
|
|
|
|
await msg.edit(
|
|
content="\n".join(new_lines),
|
|
content_type="html.raw",
|
|
override={
|
|
"body": lines_formatted
|
|
}
|
|
)
|
|
else:
|
|
await msg.edit(content=lines_formatted)
|
|
return
|
|
if number_or_query.startswith("msc"):
|
|
number_or_query = number_or_query[3:]
|
|
elif number_or_query.startswith("#"):
|
|
number_or_query = number_or_query[1:]
|
|
if not number_or_query.isdigit() or len(number_or_query) != 4:
|
|
return await ctx.respond("Invalid MXC number.")
|
|
|
|
msg = await ctx.respond("Fetching MSC #{:0>4}...".format(number_or_query))
|
|
data: dict = await self.get_msc_with_cache(int(number_or_query))
|
|
if data.get("error"):
|
|
return await msg.edit(data["error"])
|
|
return await msg.edit(self.pr_to_display(data))
|
|
|
|
@niobot.command("msc.cache")
|
|
@niobot.is_owner()
|
|
async def msc_cache_manager(self, ctx: niobot.Context, operation, arg1=None):
|
|
"""
|
|
MSC Cache manager
|
|
|
|
Available commands: get, download, remove, clear, list
|
|
"""
|
|
operation: str
|
|
arg1: str | None
|
|
if operation == "list":
|
|
known = []
|
|
total_size = 0
|
|
for file in self.msc_cache.glob("*.json"):
|
|
known.append(int(file.name.split(".")[0]))
|
|
total_size += (await asyncio.to_thread(
|
|
file.stat
|
|
)).st_size
|
|
known.sort()
|
|
|
|
units = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"]
|
|
size = total_size
|
|
while size > 1024:
|
|
units.pop(0)
|
|
size /= 1024
|
|
|
|
return await ctx.respond(
|
|
"{:,} cached entries (amounting to {:,.2f}{}):\n{}".format(
|
|
len(known),
|
|
size,
|
|
units[0],
|
|
", ".join(map(str, known))
|
|
)
|
|
)
|
|
elif operation in ["get", "view"]:
|
|
file = self.msc_cache / ("{:0>4}.json".format(int(arg1)))
|
|
if file.exists():
|
|
data = json.loads(file.read_text())
|
|
out = json.dumps(data, indent=4)
|
|
return await ctx.respond(
|
|
"```json\n%s\n```" % out
|
|
)
|
|
return await ctx.respond("%s does not exist." % file.name)
|
|
elif operation in ["download", "dl", "fetch"]:
|
|
x = await self.get_msc_with_cache(int(arg1))
|
|
return await ctx.respond(
|
|
"```json\n%s\n```" % json.dumps(x, indent=4)
|
|
)
|
|
elif operation in ["delete", "remove", "rm", "del"]:
|
|
file = self.msc_cache / ("{:0>4}.json".format(int(arg1)))
|
|
if file.exists():
|
|
file.remove()
|
|
return await ctx.respond("Removed cached file %s." % file.name)
|
|
return await ctx.respond("No cached file to remove")
|
|
elif operation == "clear":
|
|
shutil.rmtree(self.msc_cache)
|
|
self.msc_cache.mkdir()
|
|
return await ctx.respond("Cleared all cached MSCs.")
|
|
else:
|
|
return await ctx.respond("Unknown operation.")
|
|
|
|
@niobot.command("automsc", usage='<action: enable|disable|status>')
|
|
async def automsc_manager(self, ctx: niobot.Context, action: str):
|
|
"""
|
|
Enables or disables automatic MSC linking.
|
|
"""
|
|
action = action.casefold()
|
|
key = self.bot.redis_key(ctx.room.room_id, "auto_msc.enabled")
|
|
if action == "enable":
|
|
exists = await self.bot.redis.get(key)
|
|
if exists:
|
|
return await ctx.respond("AutoMSC is already enabled in this room.")
|
|
await self.bot.redis.set(key, 1)
|
|
await self.bot.redis.save()
|
|
return await self.bot.add_reaction(
|
|
ctx.room, ctx.message, "\N{WHITE HEAVY CHECK MARK}"
|
|
)
|
|
elif action == "status":
|
|
enabled = bool(await self.bot.redis.get(key))
|
|
text = "AutoMSC detection & linking is **{}**."
|
|
text = text.format({True: "**enabled**", False: "**disabled**"}[enabled])
|
|
return await ctx.respond(text)
|
|
elif action == "delete":
|
|
await self.bot.redis.delete(key)
|
|
await self.bot.redis.save()
|
|
await self.bot.add_reaction(
|
|
ctx.room, ctx.message, "\N{WHITE HEAVY CHECK MARK}"
|
|
)
|
|
else:
|
|
return await ctx.respond("Unrecognised action.")
|