Grab the avatar from the RSS feed

discord_twitter_webhooks/send_embed.py

@@ -1,9 +1,12 @@
+from functools import lru_cache
 from random import randint
 from typing import TYPE_CHECKING
+from xml.etree import ElementTree
 
 from discord_webhook import DiscordEmbed, DiscordWebhook
 from loguru import logger
 from reader import Entry
+from requests import request
 
 from discord_twitter_webhooks.dataclasses import Settings
 from discord_twitter_webhooks.get_tweet_text import get_tweet_text
@@ -33,6 +36,32 @@ def get_color(settings: Settings) -> int:
     return int(embed_color[1:], 16)
 
 
+# TODO: Should we use requests-cache?
+# TODO: Add a way to clear the cache in the web interface or on a timer?
+@lru_cache(maxsize=128)
+def get_avatar(rss_feed: str) -> str:
+    """Get the avatar of the embed.
+
+    Returns:
+        The avatar of the embed as an int.
+    """
+    # Go to the rss feed and get the avatar
+    response: Response = request("GET", rss_feed)
+    default_avatar: str = "https://pbs.twimg.com/profile_images/1354479643882004483/Btnfm47p_400x400.jpg"
+    if response.ok:
+        # Parse XML and get the avatar
+        xml_data: str = response.content.decode("utf-8")
+
+        # TODO: This is a security risk, we should use https://github.com/tiran/defusedxml
+        root: ElementTree.Element = ElementTree.fromstring(xml_data)  # noqa: S314
+        found: ElementTree.Element | None = root.find("channel/image/url")
+
+        return found.text or default_avatar if found is not None else default_avatar
+
+    logger.error(f"Got {response.status_code} from {rss_feed}. Response: {response.text}")
+    return default_avatar
+
+
 def send_embed(entry: Entry, settings: Settings) -> None:
     """Send an embed to Discord.
 
@@ -52,18 +81,20 @@ def send_embed(entry: Entry, settings: Settings) -> None:
 
     tweet_text: str = get_tweet_text(entry, settings)
     embed = DiscordEmbed(description=tweet_text)
+    entry_link: str = entry.link or ""
 
     if settings.embed_show_title:
         if entry.title:
             embed.set_title(entry.title)
         else:
-            logger.error("No title for tweet https://twitter.com/i/status/{}", entry.link)
+            logger.error("No title for {}", entry_link)
 
     if settings.embed_show_author:
         if entry.author:
-            embed.set_author(name=entry.author)
+            avatar: str = get_avatar(entry.feed_url)
+            embed.set_author(name=entry.author, url=entry_link, icon_url=avatar)
         else:
-            logger.error("No author for tweet https://twitter.com/i/status/{}", entry.link)
+            logger.error("No author for {}", entry_link)
 
     if settings.embed_timestamp:
         embed.set_timestamp()
@@ -81,3 +112,4 @@ def send_embed(entry: Entry, settings: Settings) -> None:
             logger.info("Webhook posted for tweet https://twitter.com/i/status/{}", entry.link)
         else:
             logger.error(f"Got {response.status_code} from {webhook}. Response: {response.text}")
+            logger.error(f"Got {response.status_code} from {webhook}. Response: {response.text}")