-import requests
-import bs4
-import yt_dlp as youtube_dl
-import random
-import configparser
-import re
-import io
-import os
-import asyncio
-from collections import defaultdict
-from PythonSed import Sed
-from slixmpp import ClientXMPP
-from urllib.parse import urlparse, parse_qs, urlunparse
-from pantomime import normalize_mimetype
-import cgi
-
-sed_parse = re.compile("(?<!\\\\)[/#]")
-sed_cmd = re.compile("^s[/#].*[/#].*[/#]")
-
-parser = "html.parser"
-user_agent = "Mozilla/5.0 (X11; Linux x86_64; rv:10.0)"
-" Gecko/20100101 Firefox/10.0"
-accept_lang = "en-US"
-data_limit = 100000000 # 100MB
-
-headers = {
- "user-agent": user_agent,
- "Accept-Language": accept_lang,
- "Cache-Control": "no-cache",
-}
-
-youtube_links = ["www.youtube.com", "m.youtube.com"]
-
-youtube_link = "youtu.be"
-
-ydl = youtube_dl.YoutubeDL()
-
-invidious_instances = ["invidious.snopyta.org"]
-
-block_list = ("localhost", "127.0.0.1", "0.0.0.0")
-
-req_list = ("http://", "https://")
-
-html_files = ("text/html", "application/xhtml+xml")
-
-
-class Lifo(list):
- """Limited size LIFO array to store messages and urls."""
-
- def __init__(self, size):
- """Initialize the LIFO array."""
- super().__init__()
- self.size = size
-
- def add(self, item):
- """Add an item to the LIFO array."""
- self.insert(0, item)
- if len(self) > self.size:
- self.pop()
-
-
-def get_youtube_title(url):
- """Get the title of a youtube video."""
- try:
- info = ydl.extract_info(url, download=False)
- return info["title"]
- except Exception as e:
- print(e)
- return None
-
-
-def get_invidious_link(yurl):
- """Get an invidious link from a youtube link."""
- video = yurl.split("/")[-1]
- instance = random.choice(invidious_instances)
- return f"https://{instance}/watch?v={video}"
-
-
-def get_yurl(path):
- """Get a youtube link from a path."""
- yurl = f"https://youtu.be/{path}"
- return yurl
-
-
-class AngelBot(ClientXMPP):
- """AngelBot class."""
-
- messages = defaultdict(
- lambda: {
- "messages": Lifo(100),
- "links": Lifo(10),
- "previews": Lifo(10),
- }
- )
-
- def get_urls(self, msg):
- """Get urls from a message."""
- str_list = msg["body"].strip().split()
- urls = [u for u in str_list if any(r in u for r in req_list)]
- return urls
-
- def send_youtube_info(self, uri, sender, mtype):
- """Send youtube info to the sender."""
- yurl = None
- if uri.netloc == youtube_link:
- yurl = get_yurl(uri.path)
- elif "v" in (query := parse_qs(uri.query)):
- if v := query["v"]:
- yurl = get_yurl(v[0])
- else:
- return
-
- invidious = get_invidious_link(yurl)
-
- if output := get_youtube_title(invidious):
- if output in self.messages[sender]["previews"]:
- return
- self.messages[sender]["previews"].add(output)
-
- self.send_message(mto=sender, mbody=f"*{output}*", mtype=mtype)
- self.send_message(mto=sender, mbody=invidious, mtype=mtype)
-
- async def parse_uri(self, uri, sender, mtype):
- """Parse a uri and send the result to the sender."""
- netloc = uri.netloc
- if netloc in (youtube_links + [youtube_link]):
- self.send_youtube_info(uri, sender, mtype)
- elif netloc.split(":")[0] in block_list:
- return
- else:
- await self.process_link(uri, sender, mtype)
-
- async def process_link(self, uri, sender, mtype):
- """Process a link and send the result to the sender."""
- url = urlunparse(uri)
- 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
-
- if ftype in html_files:
- data = ""
- for i in r.iter_content(chunk_size=1024, decode_unicode=False):
- data += i.decode("utf-8", errors="ignore")
- if len(data) > data_limit or "</head>" in data.lower():
- break
- soup = bs4.BeautifulSoup(data, parser)
- if title := soup.find("title"):
- output = title.text.strip()
- if output:
- output = f"*{output}*" if ("\n" not in output) else output
- if output in self.messages[sender]["previews"]:
- return
-
- self.messages[sender]["previews"].add(output)
- if r.history:
- self.send_message(mto=sender, mbody=r.url, mtype=mtype)
- self.send_message(mto=sender, mbody=output, mtype=mtype)
-
- else:
- try:
- lenght = 0
- outfile = io.BytesIO()
- for chunk in r.iter_content(
- chunk_size=512,
- decode_unicode=False,
- ):
- lenght += 512
- if lenght >= data_limit:
- return
- outfile.write(chunk)
-
- content_disposition = r.headers.get("content-disposition")
- filename = None
- if content_disposition:
- _, params = cgi.parse_header(content_disposition)
- 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)
- except Exception as e:
- print(e)
-
- async def embed_file(self, url, sender, mtype, ftype, fname, outfile):
- """Embed a file and send the result to the sender."""
- furl = await self.plugin["xep_0363"].upload_file(
- fname, content_type=ftype, input_file=outfile
- )
- message = self.make_message(sender)
- message["body"] = furl
- message["type"] = mtype
- message["oob"]["url"] = furl
- message.send()
-
- async def parse_urls(self, msg, urls, sender, mtype):
- """Parse urls and send the result to the sender."""
- body = msg["body"].lower()
- if "nsfw" in body: return
- if "nsfl" in body: return
- for u in urls:
- if u in self.messages[sender]["links"]:
- continue
- else:
- self.messages[sender]["links"].add(u)
-
- uri = urlparse(u)
- await self.parse_uri(uri, sender, mtype)
-
- def sed_command(self, msg, sender, mtype):
- """Process sed command."""
- try:
- text = msg["body"]
- if not sed_cmd.match(text):
- self.messages[sender]["messages"].add(text)
- return
- sed_args = sed_parse.split(text)
-
- if len(sed_args) < 4:
- return
-
- sed = Sed()
- sed.load_string(text)
-
- for message in self.messages[sender]["messages"]:
- if sed_args[1] not in message:
- continue
- msg = io.StringIO(message)
- res = "\n".join(sed.apply(msg, None))
- self.messages[sender]["messages"].add(res)
- return self.send_message(
- mto=sender,
- mbody=res,
- mtype=mtype,
- )
-
- except Exception as e:
- print(e)
-
- def __init__(self, jid, password, nick="angel", autojoin=None):
- """Initialize the bot."""
- ClientXMPP.__init__(self, jid, password)
- self.jid = jid
- self.nick = nick
- self.autojoin = autojoin or []
- self.register_plugin("xep_0030")
- self.register_plugin("xep_0060")
- self.register_plugin("xep_0054")
- self.register_plugin("xep_0045")
- self.register_plugin("xep_0066")
- self.register_plugin("xep_0084")
- self.register_plugin("xep_0153")
- self.register_plugin("xep_0363")
-
- self.add_event_handler("session_start", self.session_start)
- self.add_event_handler("message", self.message)
- self.add_event_handler("groupchat_message", self.muc_message)
- # self.add_event_handler("vcard_avatar_update", self.debug_event)
- # self.add_event_handler("stream_error", self.debug_event)
- self.add_event_handler("disconnected", lambda _: self.connect())
-
- async def session_start(self, event):
- """Start the bot."""
- self.send_presence()
- await self.get_roster()
- await self.update_info()
- for channel in self.autojoin:
- try:
- self.plugin["xep_0045"].join_muc(channel, self.nick)
- except Exception as e:
- print(e)
-
- async def update_info(self):
- """Update the bot info."""
- with open("angel.png", "rb") as avatar_file:
- avatar = avatar_file.read()
-
- avatar_type = "image/png"
- 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"
-
- asyncio.gather(self.plugin["xep_0054"].publish_vcard(vcard))
-
- 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
-
- try:
- if not msg["oob"]["url"]:
- if urls := self.get_urls(msg):
- await self.parse_urls(msg, urls, sender, mtype)
- except Exception as e:
- print(e)
-
- self.sed_command(msg, sender, mtype)
-
- async def muc_message(self, msg):
- """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"]:
- if urls := self.get_urls(msg):
- await self.parse_urls(msg, urls, sender, mtype)
- except Exception as e:
- print(e)
-
- self.sed_command(msg, sender, mtype)
-
-
-if __name__ == "__main__":
- config = configparser.ConfigParser()
- config.read("config.ini")
- jid = config["angel"]["jid"]
- password = config["angel"]["password"]
- autojoin = config["angel"]["autojoin"].split()
- nick = config["angel"]["nick"]
- bot = AngelBot(jid, password, nick=nick, autojoin=autojoin)
-
- bot.connect()
- bot.process(forever=True)
+import requests\r
+import bs4\r
+import yt_dlp as youtube_dl\r
+import random\r
+import configparser\r
+import re\r
+import io\r
+import os\r
+import asyncio\r
+from collections import defaultdict\r
+from PythonSed import Sed\r
+from slixmpp import ClientXMPP\r
+from urllib.parse import urlparse, parse_qs, urlunparse\r
+from pantomime import normalize_mimetype\r
+import cgi\r
+\r
+sed_parse = re.compile("(?<!\\\\)[/#]")\r
+sed_cmd = re.compile("^s[/#].*[/#].*[/#]")\r
+\r
+parser = "html.parser"\r
+user_agent = "Mozilla/5.0 (X11; Linux x86_64; rv:10.0)"\r
+" Gecko/20100101 Firefox/10.0"\r
+accept_lang = "en-US"\r
+data_limit = 100000000 # 100MB\r
+\r
+headers = {\r
+ "user-agent": user_agent,\r
+ "Accept-Language": accept_lang,\r
+ "Cache-Control": "no-cache",\r
+}\r
+\r
+youtube_links = ["www.youtube.com", "m.youtube.com"]\r
+\r
+youtube_link = "youtu.be"\r
+\r
+ydl = youtube_dl.YoutubeDL()\r
+\r
+invidious_instances = ["invidious.kalli.st"]\r
+\r
+block_list = ("localhost", "127.0.0.1", "0.0.0.0")\r
+\r
+req_list = ("http://", "https://")\r
+\r
+html_files = ("text/html", "application/xhtml+xml")\r
+\r
+\r
+class Lifo(list):\r
+ """Limited size LIFO array to store messages and urls."""\r
+\r
+ def __init__(self, size):\r
+ """Initialize the LIFO array."""\r
+ super().__init__()\r
+ self.size = size\r
+\r
+ def add(self, item):\r
+ """Add an item to the LIFO array."""\r
+ self.insert(0, item)\r
+ if len(self) > self.size:\r
+ self.pop()\r
+\r
+\r
+def get_youtube_title(url):\r
+ """Get the title of a youtube video."""\r
+ try:\r
+ info = ydl.extract_info(url, download=False)\r
+ return info["title"]\r
+ except Exception as e:\r
+ print(e)\r
+ return None\r
+\r
+\r
+def get_invidious_link(yurl):\r
+ """Get an invidious link from a youtube link."""\r
+ video = yurl.split("/")[-1]\r
+ instance = random.choice(invidious_instances)\r
+ return f"https://{instance}/watch?v={video}"\r
+\r
+\r
+def get_yurl(path):\r
+ """Get a youtube link from a path."""\r
+ yurl = f"https://youtu.be/{path}"\r
+ return yurl\r
+\r
+\r
+class AngelBot(ClientXMPP):\r
+ """AngelBot class."""\r
+\r
+ messages = defaultdict(\r
+ lambda: {\r
+ "messages": Lifo(100),\r
+ "links": Lifo(10),\r
+ "previews": Lifo(10),\r
+ }\r
+ )\r
+\r
+ def get_urls(self, msg):\r
+ """Get urls from a message."""\r
+ str_list = msg["body"].strip().split()\r
+ urls = [u for u in str_list if any(r in u for r in req_list)]\r
+ return urls\r
+\r
+ def send_youtube_info(self, uri, sender, mtype):\r
+ """Send youtube info to the sender."""\r
+ yurl = None\r
+ if uri.netloc == youtube_link:\r
+ yurl = get_yurl(uri.path)\r
+ elif "v" in (query := parse_qs(uri.query)):\r
+ if v := query["v"]:\r
+ yurl = get_yurl(v[0])\r
+ else:\r
+ return\r
+\r
+ invidious = get_invidious_link(yurl)\r
+\r
+ if output := get_youtube_title(invidious):\r
+ if output in self.messages[sender]["previews"]:\r
+ return\r
+ self.messages[sender]["previews"].add(output)\r
+\r
+ self.send_message(mto=sender, mbody=f"*{output}*", mtype=mtype)\r
+ self.send_message(mto=sender, mbody=invidious, mtype=mtype)\r
+\r
+ async def parse_uri(self, uri, sender, mtype):\r
+ """Parse a uri and send the result to the sender."""\r
+ netloc = uri.netloc\r
+ if netloc in (youtube_links + [youtube_link]):\r
+ self.send_youtube_info(uri, sender, mtype)\r
+ elif netloc.split(":")[0] in block_list:\r
+ return\r
+ else:\r
+ await self.process_link(uri, sender, mtype)\r
+\r
+ async def process_link(self, uri, sender, mtype):\r
+ """Process a link and send the result to the sender."""\r
+ url = urlunparse(uri)\r
+ r = requests.get(url, stream=True, headers=headers, timeout=6)\r
+ if not r.ok:\r
+ return\r
+\r
+ ftype = normalize_mimetype(r.headers.get("content-type"))\r
+\r
+ if not ftype:\r
+ return\r
+\r
+ if ftype in html_files:\r
+ data = ""\r
+ for i in r.iter_content(chunk_size=1024, decode_unicode=False):\r
+ data += i.decode("utf-8", errors="ignore")\r
+ if len(data) > data_limit or "</head>" in data.lower():\r
+ break\r
+ soup = bs4.BeautifulSoup(data, parser)\r
+ if title := soup.find("title"):\r
+ output = title.text.strip()\r
+ if output:\r
+ output = f"*{output}*" if ("\n" not in output) else output\r
+ if output in self.messages[sender]["previews"]:\r
+ return\r
+\r
+ self.messages[sender]["previews"].add(output)\r
+ if r.history:\r
+ self.send_message(mto=sender, mbody=r.url, mtype=mtype)\r
+ self.send_message(mto=sender, mbody=output, mtype=mtype)\r
+\r
+ else:\r
+ try:\r
+ lenght = 0\r
+ outfile = io.BytesIO()\r
+ for chunk in r.iter_content(\r
+ chunk_size=512,\r
+ decode_unicode=False,\r
+ ):\r
+ lenght += 512\r
+ if lenght >= data_limit:\r
+ return\r
+ outfile.write(chunk)\r
+\r
+ content_disposition = r.headers.get("content-disposition")\r
+ filename = None\r
+ if content_disposition:\r
+ _, params = cgi.parse_header(content_disposition)\r
+ filename = params.get("filename")\r
+ else:\r
+ filename = os.path.basename(uri.path)\r
+\r
+ ext = os.path.splitext(filename)[1] if filename else ".txt"\r
+ fname = filename if filename else f"file{ext}"\r
+ await self.embed_file(url, sender, mtype, ftype, fname, outfile)\r
+ except Exception as e:\r
+ print(e)\r
+\r
+ async def embed_file(self, url, sender, mtype, ftype, fname, outfile):\r
+ """Embed a file and send the result to the sender."""\r
+ furl = await self.plugin["xep_0363"].upload_file(\r
+ fname, content_type=ftype, input_file=outfile\r
+ )\r
+ message = self.make_message(sender)\r
+ message["body"] = furl\r
+ message["type"] = mtype\r
+ message["oob"]["url"] = furl\r
+ message.send()\r
+\r
+ async def parse_urls(self, msg, urls, sender, mtype):\r
+ """Parse urls and send the result to the sender."""\r
+ body = msg["body"].lower()\r
+ if "nsfw" in body: return\r
+ if "nsfl" in body: return\r
+ for u in urls:\r
+ if u in self.messages[sender]["links"]:\r
+ continue\r
+ else:\r
+ self.messages[sender]["links"].add(u)\r
+\r
+ uri = urlparse(u)\r
+ await self.parse_uri(uri, sender, mtype)\r
+\r
+ def sed_command(self, msg, sender, mtype):\r
+ """Process sed command."""\r
+ try:\r
+ text = msg["body"]\r
+ if not sed_cmd.match(text):\r
+ self.messages[sender]["messages"].add(text)\r
+ return\r
+ sed_args = sed_parse.split(text)\r
+\r
+ if len(sed_args) < 4:\r
+ return\r
+\r
+ sed = Sed()\r
+ sed.load_string(text)\r
+\r
+ for message in self.messages[sender]["messages"]:\r
+ if sed_args[1] not in message:\r
+ continue\r
+ msg = io.StringIO(message)\r
+ res = "\n".join(sed.apply(msg, None))\r
+ self.messages[sender]["messages"].add(res)\r
+ return self.send_message(\r
+ mto=sender,\r
+ mbody=res,\r
+ mtype=mtype,\r
+ )\r
+\r
+ except Exception as e:\r
+ print(e)\r
+\r
+ def __init__(self, jid, password, nick="angel", autojoin=None):\r
+ """Initialize the bot."""\r
+ ClientXMPP.__init__(self, jid, password)\r
+ self.jid = jid\r
+ self.nick = nick\r
+ self.autojoin = autojoin or []\r
+ self.register_plugin("xep_0030")\r
+ self.register_plugin("xep_0060")\r
+ self.register_plugin("xep_0054")\r
+ self.register_plugin("xep_0045")\r
+ self.register_plugin("xep_0066")\r
+ self.register_plugin("xep_0084")\r
+ self.register_plugin("xep_0153")\r
+ self.register_plugin("xep_0363")\r
+\r
+ self.add_event_handler("session_start", self.session_start)\r
+ self.add_event_handler("message", self.message)\r
+ self.add_event_handler("groupchat_message", self.muc_message)\r
+ # self.add_event_handler("vcard_avatar_update", self.debug_event)\r
+ # self.add_event_handler("stream_error", self.debug_event)\r
+ self.add_event_handler("disconnected", lambda _: self.connect())\r
+\r
+ async def session_start(self, event):\r
+ """Start the bot."""\r
+ self.send_presence()\r
+ await self.get_roster()\r
+ await self.update_info()\r
+ for channel in self.autojoin:\r
+ try:\r
+ self.plugin["xep_0045"].join_muc(channel, self.nick)\r
+ except Exception as e:\r
+ print(e)\r
+\r
+ async def update_info(self):\r
+ """Update the bot info."""\r
+ with open("angel.png", "rb") as avatar_file:\r
+ avatar = avatar_file.read()\r
+\r
+ avatar_type = "image/png"\r
+ avatar_id = self.plugin["xep_0084"].generate_id(avatar)\r
+ avatar_bytes = len(avatar)\r
+\r
+ asyncio.gather(self.plugin["xep_0084"].publish_avatar(avatar))\r
+\r
+ asyncio.gather(\r
+ self.plugin["xep_0153"].set_avatar(\r
+ avatar=avatar,\r
+ mtype=avatar_type,\r
+ )\r
+ )\r
+\r
+ info = {\r
+ "id": avatar_id,\r
+ "type": avatar_type,\r
+ "bytes": avatar_bytes,\r
+ }\r
+\r
+ asyncio.gather(self.plugin["xep_0084"].publish_avatar_metadata([info]))\r
+\r
+ vcard = self.plugin["xep_0054"].make_vcard()\r
+\r
+ vcard["URL"] = "https://wiki.kalli.st/Angel"\r
+ vcard["DESC"] = "Angel is a bot that can do link previews and embeds."\r
+ vcard["NICKNAME"] = "Angel"\r
+ vcard["FN"] = "Angel"\r
+\r
+ asyncio.gather(self.plugin["xep_0054"].publish_vcard(vcard))\r
+\r
+ async def message(self, msg):\r
+ """Process a message."""\r
+ if msg["type"] in ("chat", "normal"):\r
+ edit = "urn:xmpp:message-correct:0" in str(msg)\r
+ if edit:\r
+ return\r
+\r
+ mtype = msg["type"]\r
+ sender = msg["from"].bare\r
+\r
+ try:\r
+ if not msg["oob"]["url"]:\r
+ if urls := self.get_urls(msg):\r
+ await self.parse_urls(msg, urls, sender, mtype)\r
+ except Exception as e:\r
+ print(e)\r
+\r
+ self.sed_command(msg, sender, mtype)\r
+\r
+ async def muc_message(self, msg):\r
+ """Process a groupchat message."""\r
+ if msg["type"] in ("groupchat", "normal"):\r
+ edit = "urn:xmpp:message-correct:0" in str(msg)\r
+\r
+ if edit:\r
+ return\r
+\r
+ if msg["mucnick"] == self.nick:\r
+ return\r
+\r
+ mtype = msg["type"]\r
+ sender = msg["from"].bare\r
+\r
+\r
+ try:\r
+ if not msg["oob"]["url"]:\r
+ if urls := self.get_urls(msg):\r
+ await self.parse_urls(msg, urls, sender, mtype)\r
+ except Exception as e:\r
+ print(e)\r
+\r
+ self.sed_command(msg, sender, mtype)\r
+\r
+\r
+if __name__ == "__main__":\r
+ config = configparser.ConfigParser()\r
+ config.read("config.ini")\r
+ jid = config["angel"]["jid"]\r
+ password = config["angel"]["password"]\r
+ autojoin = config["angel"]["autojoin"].split()\r
+ nick = config["angel"]["nick"]\r
+ bot = AngelBot(jid, password, nick=nick, autojoin=autojoin)\r
+\r
+ bot.connect()\r
+ bot.process(forever=True)\r
projects / angel.git / commitdiff