Add reddit and threads support

snsconverter/commandHandlers.py

@@ -14,7 +14,7 @@ async def _grpSns(self, ctx: Context):
     async def _cmdToggle(self, ctx: Context):
         """Toggle SNSConverter replacements on the server
 
-        This will toggle the auto-reply of any Twitter, Instagram or Tiktok links with
-        embeds, and replace them with vxtwitter, ddinstagram or vxtiktok, respectively.
+        This will toggle the auto-reply of any Twitter, Instagram, Threads, Tiktok, or Reddit links with
+        embeds, and replace them with vxtwitter, ddinstagram, vxthreads, vxtiktok, or fxreddit respectively.
         """
         await self.cmdToggle(ctx)

snsconverter/constants.py

@@ -3,14 +3,21 @@
 
 KEY_ENABLED = "enabled"
 DEFAULT_GUILD = {KEY_ENABLED: False}
-INSTA_REGEX_PATTERN = re.compile(r"https://(?:www\.)?(instagram.com)")
-TIKTOK_REGEX_PATTERN = re.compile(r"https://(www\.|vm\.)?(tiktok.com)")
-TWITTER_REGEX_PATTERN = re.compile(r"https://(?:www\.|)twitter\.com(/[^/]+/status/\d+)")
-X_REGEX_PATTERN = re.compile(r"https://(?:www\.|)x\.com(/[^/]+/status/\d+)")
+INSTA_REGEX_PATTERN = re.compile(r"http(?:s)?://(?:www\.)?(instagram\.com)")
+TIKTOK_REGEX_PATTERN = re.compile(r"http(?:s)?://(www\.|vm\.)?(tiktok\.com)")
+TWITTER_REGEX_PATTERN = re.compile(
+    r"http(?:s)?://(?:www\.)?twitter\.com(/[^/]+/status/\d+)"
+)
+X_REGEX_PATTERN = re.compile(r"http(?:s)?://(?:www\.)?x\.com(/[^/]+/status/\d+)")
+# Match any reddit subdomain, too many to list (old, np, de, us, etc)
+REDDIT_REGEX_PATTERN = re.compile(r"http(?:s)?://(?:[\w-]+?\.)?reddit\.com")
+THREADS_REGEX_PATTERN = re.compile(r"http(?:s)?://(?:www\.)?(threads\.net)")
 
 
 class SocialMedia(enum.Enum):
     INSTAGRAM = "Instagram"
     # I'm not calling it f****ng "x" lol
     TWITTER = "Twitter"
     TIKTOK = "TikTok"
+    REDDIT = "Reddit"
+    THREADS = "Threads"

snsconverter/eventHandlers.py

@@ -10,9 +10,13 @@ async def twit_replacer(self, message: Message):
         await self._on_message_twit_replacer(message)
         await self._on_message_insta_replacer(message)
         await self._on_message_tik_replacer(message)
+        await self._on_message_reddit_replacer(message)
+        await self._on_message_threads_replacer(message)
 
     @commands.Cog.listener("on_message_edit")
     async def twit_edit_replacer(self, message_before: Message, message_after):
         await self._on_edit_twit_replacer(message_before, message_after)
         await self._on_edit_insta_replacer(message_before, message_after)
         await self._on_edit_tik_replacer(message_before, message_after)
+        await self._on_edit_reddit_replacer(message_before, message_after)
+        await self._on_edit_threads_replacer(message_before, message_after)

snsconverter/eventsCore.py

@@ -5,6 +5,8 @@
 from .helpers import (
     convert_to_ddinsta_url,
     convert_to_fx_twitter_url,
+    convert_to_rxddit_url,
+    convert_to_vx_threads_url,
     convert_to_vx_tiktok_url,
     urls_to_string,
     valid,
@@ -200,3 +202,147 @@ async def _on_edit_tik_replacer(
         # Remove embeds from user message if reply is successful
         if ok:
             await message_after.edit(suppress=True)
+
+    async def _on_message_reddit_replacer(self, message: Message):
+        if not valid(message):
+            return
+
+        if not message.embeds:
+            return
+
+        if not await self.config.guild(message.guild).get_attr(KEY_ENABLED)():
+            self.logger.debug(
+                "SNSConverter disabled for guild %s (%s), skipping",
+                message.guild.name,
+                message.guild.id,
+            )
+            return
+
+        # the actual code part
+        rxddit_urls = convert_to_rxddit_url(message.embeds)
+
+        # no changed urls detected
+        if not rxddit_urls:
+            return
+
+        # constructs the message and replies with a mention
+        ok = await message.reply(urls_to_string(rxddit_urls, SocialMedia.REDDIT))
+
+        # Remove embeds from user message if reply is successful
+        if ok:
+            await message.edit(suppress=True)
+
+    async def _on_edit_reddit_replacer(
+        self, message_before: Message, message_after: Message
+    ):
+        # skips if the message is sent by any bot
+        if not valid(message_after):
+            return
+
+        # skips if the message has no embeds
+        if not message_after.embeds:
+            return
+
+        if not await self.config.guild(message_after.guild).get_attr(KEY_ENABLED)():
+            self.logger.debug(
+                "SNSConverter disabled for guild %s (%s), skipping",
+                message_after.guild.name,
+                message_after.guild.id,
+            )
+            return
+
+        video_embed_before = [embed for embed in message_before.embeds if embed.video]
+        video_embed_after = [embed for embed in message_after.embeds if embed.video]
+        new_video_embeds = [
+            embed for embed in video_embed_after if embed not in video_embed_before
+        ]
+
+        # skips if the message has no new embeds
+        if not new_video_embeds:
+            return
+
+        rxddit_urls = convert_to_rxddit_url(new_video_embeds)
+
+        # no changed urls detected
+        if not rxddit_urls:
+            return
+
+        # constructs the message and replies with a mention
+        ok = await message_after.reply(urls_to_string(rxddit_urls, SocialMedia.REDDIT))
+
+        # Remove embeds from user message if reply is successful
+        if ok:
+            await message_after.edit(suppress=True)
+
+    async def _on_message_threads_replacer(self, message: Message):
+        if not valid(message):
+            return
+
+        if not message.embeds:
+            return
+
+        if not await self.config.guild(message.guild).get_attr(KEY_ENABLED)():
+            self.logger.debug(
+                "SNSConverter disabled for guild %s (%s), skipping",
+                message.guild.name,
+                message.guild.id,
+            )
+            return
+
+        # the actual code part
+        vx_threads_urls = convert_to_vx_threads_url(message.embeds)
+
+        # no changed urls detected
+        if not vx_threads_urls:
+            return
+
+        # constructs the message and replies with a mention
+        ok = await message.reply(urls_to_string(vx_threads_urls, SocialMedia.THREADS))
+
+        # Remove embeds from user message if reply is successful
+        if ok:
+            await message.edit(suppress=True)
+
+    async def _on_edit_threads_replacer(
+        self, message_before: Message, message_after: Message
+    ):
+        # skips if the message is sent by any bot
+        if not valid(message_after):
+            return
+
+        # skips if the message has no embeds
+        if not message_after.embeds:
+            return
+
+        if not await self.config.guild(message_after.guild).get_attr(KEY_ENABLED)():
+            self.logger.debug(
+                "SNSConverter disabled for guild %s (%s), skipping",
+                message_after.guild.name,
+                message_after.guild.id,
+            )
+            return
+
+        video_embed_before = [embed for embed in message_before.embeds if embed.video]
+        video_embed_after = [embed for embed in message_after.embeds if embed.video]
+        new_video_embeds = [
+            embed for embed in video_embed_after if embed not in video_embed_before
+        ]
+
+        # skips if the message has no new embeds
+        if not new_video_embeds:
+            return
+
+        vx_threads_urls = convert_to_vx_threads_url(new_video_embeds)
+
+        # no changed urls detected
+        if not vx_threads_urls:
+            return
+
+        # constructs the message and replies with a mention
+        ok = await message_after.reply(
+            urls_to_string(vx_threads_urls, SocialMedia.THREADS)
+        )
+
+        # Remove embeds from user message if reply is successful
+        if ok:
+            await message_after.edit(suppress=True)

snsconverter/helpers.py

@@ -4,6 +4,8 @@
 
 from .constants import (
     INSTA_REGEX_PATTERN,
+    REDDIT_REGEX_PATTERN,
+    THREADS_REGEX_PATTERN,
     TIKTOK_REGEX_PATTERN,
     TWITTER_REGEX_PATTERN,
     X_REGEX_PATTERN,
@@ -72,7 +74,6 @@ def convert_to_fx_twitter_url(message_content: str):
 
     fixed_urls = []
 
-    # Extracts all twitter urls from message content, and converts them to fxtwitter, returning a list of urls
     for word in message_split:
         # I don't think @everyone will work anyway, but just incase...
         if "@" in word:
@@ -87,6 +88,52 @@ def convert_to_fx_twitter_url(message_content: str):
     return fixed_urls
 
 
+def convert_to_rxddit_url(embeds: list[Embed]):
+    """
+    Parameters
+    ----------
+    embeds: list of Discord embeds
+
+    Returns
+    -------
+        filtered list of Reddit URLs that have been converted to rxddit
+    """
+
+    # pulls only video embeds from list of embeds
+    urls = [entry.url for entry in embeds]
+
+    rxddit_urls = [
+        re.sub(REDDIT_REGEX_PATTERN, r"https://rxddit.com", result)
+        for result in urls
+        if re.match(REDDIT_REGEX_PATTERN, result)
+    ]
+
+    return rxddit_urls
+
+
+def convert_to_vx_threads_url(embeds: list[Embed]):
+    """
+    Parameters
+    ----------
+    embeds: list of Discord embeds
+
+    Returns
+    -------
+        filtered list of Threads URLs that have been converted to vxthreads
+    """
+
+    # pulls only video embeds from list of embeds
+    urls = [entry.url for entry in embeds]
+
+    vxthreads_urls = [
+        re.sub(THREADS_REGEX_PATTERN, r"https://vx\1", result)
+        for result in urls
+        if re.match(THREADS_REGEX_PATTERN, result)
+    ]
+
+    return vxthreads_urls
+
+
 def urls_to_string(links: list[str], socialMedia: SocialMedia):
     """
     Parameters

snsconverter/snsconverter.py

@@ -5,6 +5,6 @@
 
 
 class SNSConverter(commands.Cog, CommandHandlers, EventHandlers):
-    """Converts Twitter, Instagram & Tiktok links for better video embeds"""
+    """Converts Twitter, Instagram, Threads, Tiktok & Reddit links for better embeds"""
 
     pass