-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
- self.messages[sender]["links"].add(furl)\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 not re.match(sed_args[1], 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
+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.kalli.st"]
+
+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
+ )
+ self.messages[sender]["links"].add(furl)
+ 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.search(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 not re.match(sed_args[1], 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)
projects / angel.git / commitdiff