diff --git a/discord_twitter_webhooks/send_embed.py b/discord_twitter_webhooks/send_embed.py index 70ccac9..c5d837f 100644 --- a/discord_twitter_webhooks/send_embed.py +++ b/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}")