import re
import io
import os
-import mimetypes
import asyncio
from collections import defaultdict
from PythonSed import Sed
html_files = ("text/html", "application/xhtml+xml")
+
class Lifo(list):
"""Limited size LIFO array to store messages and urls."""
class AngelBot(ClientXMPP):
"""AngelBot class."""
-
+
messages = defaultdict(
lambda: {
"messages": Lifo(100),
r = requests.get(url, stream=True, headers=headers, timeout=6)
if not r.ok:
return
-
+
ftype = normalize_mimetype(r.headers.get("content-type"))
if not ftype:
- return
-
+ return
+
if ftype in html_files:
data = ""
for i in r.iter_content(chunk_size=1024, decode_unicode=False):
filename = params.get("filename")
else:
filename = os.path.basename(uri.path)
-
+
ext = os.path.splitext(filename)[1] if filename else ".txt"
fname = filename if filename else f"file{ext}"
await self.embed_file(url, sender, mtype, ftype, fname, outfile)
print(e)
async def update_info(self):
- """Update the bot info."""
+ """Update the bot info."""
with open("angel.png", "rb") as avatar_file:
avatar = avatar_file.read()
avatar_id = self.plugin["xep_0084"].generate_id(avatar)
avatar_bytes = len(avatar)
+ asyncio.gather(self.plugin["xep_0084"].publish_avatar(avatar))
+
+ asyncio.gather(
+ self.plugin["xep_0153"].set_avatar(
+ avatar=avatar,
+ mtype=avatar_type,
+ )
+ )
+
info = {
"id": avatar_id,
"type": avatar_type,
"bytes": avatar_bytes,
}
+ asyncio.gather(self.plugin["xep_0084"].publish_avatar_metadata([info]))
+
vcard = self.plugin["xep_0054"].make_vcard()
vcard["URL"] = "https://wiki.kalli.st/Angel"
vcard["DESC"] = "Angel is a bot that can do link previews and embeds."
vcard["NICKNAME"] = "Angel"
vcard["FN"] = "Angel"
- vcard["PHOTO"] = info
asyncio.gather(self.plugin["xep_0054"].publish_vcard(vcard))
- asyncio.gather(self.plugin["xep_0084"].publish_avatar(avatar))
-
- asyncio.gather(
- self.plugin["xep_0153"].set_avatar(
- avatar=avatar,
- mtype=avatar_type,
- )
- )
-
- asyncio.gather(self.plugin["xep_0084"].publish_avatar_metadata([info]))
-
async def message(self, msg):
"""Process a message."""
if msg["type"] in ("chat", "normal"):
edit = "urn:xmpp:message-correct:0" in str(msg)
if edit:
return
-
+
mtype = msg["type"]
sender = msg["from"].bare
"""Process a groupchat message."""
if msg["type"] in ("groupchat", "normal"):
edit = "urn:xmpp:message-correct:0" in str(msg)
-
+
if edit:
return
if msg["mucnick"] == self.nick:
return
-
+
mtype = msg["type"]
sender = msg["from"].bare
-
+
try:
if not msg["oob"]["url"]: