API Reference

boost_mentions

Module to boost mentions that tag the community bots

BoostMentions

Class to handle boosting mentions of the community bots.

Source code in src/boost_mentions.py

class BoostMentions():
    """
    Class to handle boosting mentions of the community bots.
    """
    def __init__(self, config_dict=None, no_dry_run=True):
        self.logger = logging.getLogger(__name__)
        logging.basicConfig(level=logging.INFO)

        self.process_images = False
        self.no_dry_run = no_dry_run
        self.config_dict = config_dict

    def boost_mentions(self):
        """
        Method to boost mentions on social media platforms.
        """
        self.set_up_config_dict()

        self.logger.info("==========================")
        client_name = self.config_dict.get("client_name")
        self.logger.info('Initializing %s Bot', client_name)
        self.logger.info("=================%s", "=" * len(client_name or ""))
        self.logger.info(' > Connecting to %s',
                         self.config_dict['api_base_url'])

        if self.config_dict["platform"] == "mastodon":
            account, client = login_mastodon(self.config_dict)
            notifications = client.notifications(types=['mention'])
            self.logger.info(' > Fetched account data for %s',
                             account.acct)

            self.logger.info(
                ' > Beginning search-loop and toot and boost toots'
            )
            self.logger.info('------------------------')

            self.logger.info(" > Reading statuses to identify tootable status")
            for notification in notifications:
                if not notification.status.favourited and \
                        notification.status.account.acct != account.acct:
                    # Boost and favorite the new status
                    try:
                        self.logger.info(
                            "   * Boosting new toot by %s viewable at: %s",
                            notification.account.username,
                            notification.status.url,
                        )
                        client.status_reblog(notification.status.id)
                        client.status_favourite(notification.status.id)
                    except Exception as e:
                        self.logger.info(
                            "   * Boosting new toot by %s did not work: %s ",
                            notification.account.username,
                            e,
                        )
        elif self.config_dict["platform"] == "bluesky":
            client = login_bluesky(self.config_dict)
            self.logger.info(" > Fetched account data")

            self.logger.info(" > Beginning search-loop and repost posts")
            self.logger.info("------------------------")

            self.logger.info(
                " > Reading statuses to identify postable statuses"
            )
            last_seen_at = client.get_current_time_iso()
            response = client.app.bsky.notification.list_notifications()
            timeline = client.get_timeline(algorithm='reverse-chronological')
            cids = [post.post.cid for post in timeline.feed]

            for notification in response.notifications:
                if (
                    notification.reason == "mention"
                    and notification.cid not in cids
                ):
                    try:
                        self.logger.info(
                            "   * Reposted post reference: %s",
                            client.repost(
                                uri=notification.uri,
                                cid=notification.cid
                            )
                        )
                    except Exception as e:
                        self.logger.info(
                            """
                            * Reposting new post with URI %s
                            and CID %s did not work because of %s -
                            going to the next post.
                            """,
                            notification.uri,
                            notification.cid,
                            e,
                        )

            client.app.bsky.notification.update_seen({'seen_at': last_seen_at})
            self.logger.info(
                'Successfully process notification. Last seen at: %s',
                last_seen_at
            )

    def set_up_config_dict(self):
        """
        Method to set up the config dictionary with the required parameters
        """
        self.config_dict = {
            "platform": os.getenv("PLATFORM"),
            "password": os.getenv("PASSWORD"),
            "username": os.getenv("USERNAME"),
            "client_name": os.getenv("CLIENT_NAME")
        }
        if self.config_dict["platform"] == "mastodon":
            self.config_dict["mastodon_visiblity"] = config.MASTODON_VISIBILITY
            self.config_dict["api_base_url"] = config.API_BASE_URL
            self.config_dict["access_token"] = os.getenv("ACCESS_TOKEN")
            self.config_dict["client_cred_file"] = os.getenv(
                'BOT_CLIENTCRED_SECRET'
            )
            self.config_dict["timeline_depth_limit"] = 40
        else:
            self.config_dict["api_base_url"] = "bluesky"

boost_mentions()

Method to boost mentions on social media platforms.

Source code in src/boost_mentions.py

def boost_mentions(self):
    """
    Method to boost mentions on social media platforms.
    """
    self.set_up_config_dict()

    self.logger.info("==========================")
    client_name = self.config_dict.get("client_name")
    self.logger.info('Initializing %s Bot', client_name)
    self.logger.info("=================%s", "=" * len(client_name or ""))
    self.logger.info(' > Connecting to %s',
                     self.config_dict['api_base_url'])

    if self.config_dict["platform"] == "mastodon":
        account, client = login_mastodon(self.config_dict)
        notifications = client.notifications(types=['mention'])
        self.logger.info(' > Fetched account data for %s',
                         account.acct)

        self.logger.info(
            ' > Beginning search-loop and toot and boost toots'
        )
        self.logger.info('------------------------')

        self.logger.info(" > Reading statuses to identify tootable status")
        for notification in notifications:
            if not notification.status.favourited and \
                    notification.status.account.acct != account.acct:
                # Boost and favorite the new status
                try:
                    self.logger.info(
                        "   * Boosting new toot by %s viewable at: %s",
                        notification.account.username,
                        notification.status.url,
                    )
                    client.status_reblog(notification.status.id)
                    client.status_favourite(notification.status.id)
                except Exception as e:
                    self.logger.info(
                        "   * Boosting new toot by %s did not work: %s ",
                        notification.account.username,
                        e,
                    )
    elif self.config_dict["platform"] == "bluesky":
        client = login_bluesky(self.config_dict)
        self.logger.info(" > Fetched account data")

        self.logger.info(" > Beginning search-loop and repost posts")
        self.logger.info("------------------------")

        self.logger.info(
            " > Reading statuses to identify postable statuses"
        )
        last_seen_at = client.get_current_time_iso()
        response = client.app.bsky.notification.list_notifications()
        timeline = client.get_timeline(algorithm='reverse-chronological')
        cids = [post.post.cid for post in timeline.feed]

        for notification in response.notifications:
            if (
                notification.reason == "mention"
                and notification.cid not in cids
            ):
                try:
                    self.logger.info(
                        "   * Reposted post reference: %s",
                        client.repost(
                            uri=notification.uri,
                            cid=notification.cid
                        )
                    )
                except Exception as e:
                    self.logger.info(
                        """
                        * Reposting new post with URI %s
                        and CID %s did not work because of %s -
                        going to the next post.
                        """,
                        notification.uri,
                        notification.cid,
                        e,
                    )

        client.app.bsky.notification.update_seen({'seen_at': last_seen_at})
        self.logger.info(
            'Successfully process notification. Last seen at: %s',
            last_seen_at
        )

set_up_config_dict()

Method to set up the config dictionary with the required parameters

Source code in src/boost_mentions.py

def set_up_config_dict(self):
    """
    Method to set up the config dictionary with the required parameters
    """
    self.config_dict = {
        "platform": os.getenv("PLATFORM"),
        "password": os.getenv("PASSWORD"),
        "username": os.getenv("USERNAME"),
        "client_name": os.getenv("CLIENT_NAME")
    }
    if self.config_dict["platform"] == "mastodon":
        self.config_dict["mastodon_visiblity"] = config.MASTODON_VISIBILITY
        self.config_dict["api_base_url"] = config.API_BASE_URL
        self.config_dict["access_token"] = os.getenv("ACCESS_TOKEN")
        self.config_dict["client_cred_file"] = os.getenv(
            'BOT_CLIENTCRED_SECRET'
        )
        self.config_dict["timeline_depth_limit"] = 40
    else:
        self.config_dict["api_base_url"] = "bluesky"

boost_tags

Module to boost posts containing specific tags using community bots.

BoostTags

Handles boosting of posts containing specified tags across different platforms. Currently supports Bluesky. Mastodon support is stubbed.

Source code in src/boost_tags.py

class BoostTags:
    """
    Handles boosting of posts containing specified tags across different platforms.
    Currently supports Bluesky. Mastodon support is stubbed.
    """

    def __init__(self, config_dict: dict | None = None, no_dry_run: bool = True) -> None:
        """
        Initialize the BoostTags handler.

        Args:
            config_dict (dict | None): Configuration dictionary for the bot.
                If None, values will be loaded from environment variables.
            no_dry_run (bool): If True, actually perform reposts instead of dry-run.
        """
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)

        self.config_dict = config_dict
        self.no_dry_run = no_dry_run

    def repost_tags_mastodon(self, client) -> None:
        """
        Repost Mastodon statuses containing the configured tags.

        Args:
            client: Authenticated Mastodon client instance.

        Notes:
            Currently non-functional since account fetching is commented out.
        """
        if "tags" not in self.config_dict:
            self.logger.warning("No tags configured for Mastodon reposts.")
            return

        for tag in self.config_dict["tags"]:
            tag = tag.lower().strip("# ")
            self.logger.info("Reading timeline for new toots tagged #%s", tag)

            try:
                statuses = client.timeline_hashtag(
                    tag,
                    limit=self.config_dict.get("timeline_depth_limit", 40),
                )
            except (
                MastodonNetworkError,
                MastodonAPIError,
                ConnectionError,
                TimeoutError
            ) as e:
                # NOTE: Replace/extend with library-specific 
                # exceptions as needed.
                self.logger.error(
                    "Network/API error when fetching statuses: %s. Retrying...",
                    e
                )
                time.sleep(30)
                continue

            time.sleep(0.1)  # rate limiting

            for status in statuses:
                domain = urlparse(status.url).netloc
                if (
                    not getattr(status, "favourited", False)
                    and domain not in config.IGNORE_SERVERS
                    and getattr(status.account, "acct", None) != self.config_dict.get("username")
                ):
                    self.logger.info(
                        "Boosting toot by %s tagged #%s (%s)",
                        status.account.username,
                        tag,
                        status.url,
                    )
                    client.status_reblog(status.id)
                    client.status_favourite(status.id)

    def boost_tags(self) -> None:
        """
        Main entrypoint to start boosting tags based on configuration.

        Loads configuration from environment variables if not provided.
        Handles platform-specific reposting logic.
        """
        if self.config_dict is None and self.no_dry_run:
            self._load_config_from_env()

        platform = self.config_dict.get("platform")
        client_name = self.config_dict.get("client_name", "Unknown")
        self.logger.info("========")
        self.logger.info("Initializing %s Bot", client_name)
        self.logger.info("=" * (20 + len(client_name)))
        self.logger.info("Connecting to %s", self.config_dict["api_base_url"])

        if platform == "mastodon":
            self._boost_tags_mastodon()
            self.logger.warning("Mastodon support is currently not implemented.")
        elif platform == "bluesky":
            self._boost_tags_bluesky()
        else:
            self.logger.error("Unsupported platform: %s", platform)

    def _load_config_from_env(self) -> None:
        """Load configuration values from environment variables into self.config_dict."""
        self.config_dict = {
            "platform": os.getenv("PLATFORM", "").lower(),
            "password": os.getenv("PASSWORD"),
            "username": os.getenv("USERNAME"),
            "client_name": os.getenv("CLIENT_NAME", "CommunityBot"),
            "tags": os.getenv("TAGS_TO_BOOST", "").split(","),
        }
        if self.config_dict["platform"] == "mastodon":
            self.config_dict.update({
                "mastodon_visibility": config.MASTODON_VISIBILITY,
                "api_base_url": config.API_BASE_URL,
                "access_token": os.getenv("ACCESS_TOKEN"),
                "client_cred_file": os.getenv("BOT_CLIENTCRED_SECRET"),
                "timeline_depth_limit": 40,
            })
        else:
            self.config_dict["api_base_url"] = "bluesky"

    def _boost_tags_mastodon(self) -> None:
        """Handle reposting tags on Mastodon."""
        # # Commented because it wasn't fully working

        # account, client = login_mastodon(config_dict)
        # self.logger.info(f" > Fetched account data for {account.acct}")

        # repost_tags_mastodon(client, config_dict)
        self.logger.info(
            """
            This feature currently doesn't work for Mastodon.
            It's deployed using AWS.
            """
        )

    def _boost_tags_bluesky(self) -> None:
        """Handle reposting tags on Bluesky."""
        if not self.no_dry_run:
            self.logger.info("Dry-run mode: no reposts will be made.")
            return

        client = login_bluesky(self.config_dict)
        self.logger.info("Fetched Bluesky account data.")
        self.logger.info("Starting search-loop for reposting.")

        timeline = client.get_timeline(algorithm="reverse-chronological")
        seen_cids = {post.post.cid for post in timeline.feed}

        for tag in self.config_dict["tags"]:
            response = client.app.bsky.feed.search_posts(
                params={"q": tag, "tag": [tag], "sort": "top", "limit": 50}
            )
            for post in response.posts:
                tags_in_post = {
                    t.strip("#").lower()
                    for t in post.record.text.split()
                    if t.startswith("#")
                }

                if tag.lower() in tags_in_post and post.cid not in seen_cids:
                    try:
                        result = client.repost(uri=post.uri, cid=post.cid)
                        self.logger.info(
                            "Reposted post by %s (ref: %s)",
                            post.author.handle, result
                        )
                    except AtProtocolError as e:
                        self.logger.error(
                            "Failed to repost URI %s, CID %s: %s",
                            post.uri,
                            post.cid,
                            e,
                        )
                    time.sleep(0.1)  # avoid hammering API

        self.logger.info("Finished processing Bluesky reposts.")

__init__(config_dict=None, no_dry_run=True)

Initialize the BoostTags handler.

Parameters:

Name	Type	Description	Default
config_dict	dict | None	Configuration dictionary for the bot. If None, values will be loaded from environment variables.	None
no_dry_run	bool	If True, actually perform reposts instead of dry-run.	True

Source code in src/boost_tags.py

def __init__(self, config_dict: dict | None = None, no_dry_run: bool = True) -> None:
    """
    Initialize the BoostTags handler.

    Args:
        config_dict (dict | None): Configuration dictionary for the bot.
            If None, values will be loaded from environment variables.
        no_dry_run (bool): If True, actually perform reposts instead of dry-run.
    """
    self.logger = logging.getLogger(__name__)
    self.logger.setLevel(logging.INFO)

    self.config_dict = config_dict
    self.no_dry_run = no_dry_run

boost_tags()

Main entrypoint to start boosting tags based on configuration.

Loads configuration from environment variables if not provided. Handles platform-specific reposting logic.

Source code in src/boost_tags.py

def boost_tags(self) -> None:
    """
    Main entrypoint to start boosting tags based on configuration.

    Loads configuration from environment variables if not provided.
    Handles platform-specific reposting logic.
    """
    if self.config_dict is None and self.no_dry_run:
        self._load_config_from_env()

    platform = self.config_dict.get("platform")
    client_name = self.config_dict.get("client_name", "Unknown")
    self.logger.info("========")
    self.logger.info("Initializing %s Bot", client_name)
    self.logger.info("=" * (20 + len(client_name)))
    self.logger.info("Connecting to %s", self.config_dict["api_base_url"])

    if platform == "mastodon":
        self._boost_tags_mastodon()
        self.logger.warning("Mastodon support is currently not implemented.")
    elif platform == "bluesky":
        self._boost_tags_bluesky()
    else:
        self.logger.error("Unsupported platform: %s", platform)

repost_tags_mastodon(client)

Repost Mastodon statuses containing the configured tags.

Parameters:

Name	Type	Description	Default
client		Authenticated Mastodon client instance.	required

Notes

Currently non-functional since account fetching is commented out.

Source code in src/boost_tags.py

def repost_tags_mastodon(self, client) -> None:
    """
    Repost Mastodon statuses containing the configured tags.

    Args:
        client: Authenticated Mastodon client instance.

    Notes:
        Currently non-functional since account fetching is commented out.
    """
    if "tags" not in self.config_dict:
        self.logger.warning("No tags configured for Mastodon reposts.")
        return

    for tag in self.config_dict["tags"]:
        tag = tag.lower().strip("# ")
        self.logger.info("Reading timeline for new toots tagged #%s", tag)

        try:
            statuses = client.timeline_hashtag(
                tag,
                limit=self.config_dict.get("timeline_depth_limit", 40),
            )
        except (
            MastodonNetworkError,
            MastodonAPIError,
            ConnectionError,
            TimeoutError
        ) as e:
            # NOTE: Replace/extend with library-specific 
            # exceptions as needed.
            self.logger.error(
                "Network/API error when fetching statuses: %s. Retrying...",
                e
            )
            time.sleep(30)
            continue

        time.sleep(0.1)  # rate limiting

        for status in statuses:
            domain = urlparse(status.url).netloc
            if (
                not getattr(status, "favourited", False)
                and domain not in config.IGNORE_SERVERS
                and getattr(status.account, "acct", None) != self.config_dict.get("username")
            ):
                self.logger.info(
                    "Boosting toot by %s tagged #%s (%s)",
                    status.account.username,
                    tag,
                    status.url,
                )
                client.status_reblog(status.id)
                client.status_favourite(status.id)

config

Config file for community bots

debug

This script aims at making debugging easier

DebugBots

Class to handle debugging of all modules.

Source code in src/debug.py

class DebugBots:
    """
    Class to handle debugging of all modules.
    """
    def __init__(self):
        self.bot = 'rladies'  # 'pyladies' or 'rladies'
        self.what_to_debug = 'blog'  # 'blog' or 'boost_tags' or 'rss' or 'anniversary
        self.platform = 'bluesky'  # 'bluesky' or 'mastodon'
        self.no_dry_run = False

    def start_debug(self):
        """Start debugging."""
        if self.what_to_debug == 'blog':
            config_dict = self.get_config_blog()
            promote_blog_post_handler = PromoteBlogPost(
                config_dict,
                self.no_dry_run
            )
            promote_blog_post_handler.promote_blog_post()

        elif self.what_to_debug == 'rss':
            config_dict = self.get_config_rss()
            rss_data_handler = RSSData(
                config_dict,
                self.no_dry_run
            )
            rss_data_handler.get_rss_data()

        elif self.what_to_debug == 'boost_tags':
            config_dict = self.get_config_boost()
            boost_tags_handler = BoostTags(
                config_dict,
                self.no_dry_run
            )
            boost_tags_handler.boost_tags()

        elif self.what_to_debug == 'boost_mentions':
            config_dict = self.get_config_boost()
            boost_tags_handler = BoostMentions(
                config_dict,
                self.no_dry_run
            )
            boost_tags_handler.boost_mentions()

        elif self.what_to_debug == 'anniversary':
            config_dict = self.get_config_anniversary()
            promote_anniversary_handler = PromoteAnniversary(
                config_dict,
                self.no_dry_run
            )
            promote_anniversary_handler.promote_anniversary()

    def get_config_blog(self):
        """Method to generate config for promoting blog posts"""
        if self.bot == 'pyladies':
            if self.platform == 'bluesky':
                return {
                    "archive": "pyladies_archive_directory_bluesky",
                    "counter": "metadata/pyladies_counter_bluesky.txt",
                    "json_file": "metadata/pyladies_meta_data.json",
                    "client_name": "pyladies_self.bot",
                    "images": "pyladies_images",
                    "api_base_url": self.platform,
                    "mastodon": None,
                    "gen_ai_support": True,
                    "gemini_model_name": "gemini-2.5-flash",
                    "password": os.getenv("PYLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("PYLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                }
            return {
                'archive': 'pyladies_archive_directory',
                'counter': 'pyladies_counter.txt',
                'json_file': 'metadata/pyladies_meta_data.json',
                'client_name': 'pyladies_self.bot',
                'mastodon': None,
            }

        if self.bot == 'rladies':
            if self.platform == 'bluesky':
                return {
                    "archive": "rladies_archive_directory_bluesky",
                    "counter": "../metadata/rladies_counter_bluesky.txt",
                    "json_file": "../metadata/rladies_meta_data.json",
                    "client_name": "rladies_self.bot",
                    "images": "rladies_images",
                    "api_base_url": self.platform,
                    "mastodon": None,
                    "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("RLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                }
            return {
                "archive": "rladies_archive_directory",
                "counter": "../metadata/rladies_counter.txt",
                "json_file": "../metadata/rladies_meta_data.json",
                "client_name": "rladies_self.bot",
                "mastodon": None,
            }

        return None

    def get_config_boost(self):
        """Method to generate config for boosting tags"""
        if self.bot == 'pyladies':
            return {"client_name": "pyladies_self.bot", "mastodon": None}

        if self.bot == 'rladies':
            if self.platform == "bluesky":
                return {
                    "client_name": "rladies_self.bot",
                    "api_base_url": self.platform,
                    "mastodon": None,
                    "password": os.getenv("PASSWORD"),
                    "username": os.getenv("USERNAME"),
                    "platform": self.platform,
                    "tags": "rladies",
                }
            return {"client_name": "rladies_self.bot", "mastodon": None}

        return None

    def get_config_anniversary(self):
        """Method to get config for promoting anniversaries"""
        if self.bot == 'pyladies':
            if self.platform == 'bluesky':
                return {
                    'client_name': 'pyladies_self.bot',
                    'api_base_url': self.platform,
                    'mastodon': None,
                    'password': os.getenv('PYLADIES_BSKY_PASSWORD'),
                    'username': os.getenv('PYLADIES_BSKY_USERNAME'),
                    'images': 'anniversary_images',
                    'platform': self.platform,
                }
            return {'client_name': 'pyladies_self.bot', 'mastodon': None}

        if self.bot == 'rladies':
            if self.platform == 'bluesky':
                return {
                    'client_name': 'rladies_self.bot',
                    'api_base_url': self.platform,
                    'mastodon': None,
                    'password': os.getenv('RLADIES_BSKY_PASSWORD'),
                    'username': os.getenv('RLADIES_BSKY_USERNAME'),
                    'images': 'anniversary_images',
                    'platform': self.platform,
                }
            return {'client_name': 'rladies_self.bot', 'mastodon': None}

        return None

get_config_anniversary()

Method to get config for promoting anniversaries

Source code in src/debug.py

def get_config_anniversary(self):
    """Method to get config for promoting anniversaries"""
    if self.bot == 'pyladies':
        if self.platform == 'bluesky':
            return {
                'client_name': 'pyladies_self.bot',
                'api_base_url': self.platform,
                'mastodon': None,
                'password': os.getenv('PYLADIES_BSKY_PASSWORD'),
                'username': os.getenv('PYLADIES_BSKY_USERNAME'),
                'images': 'anniversary_images',
                'platform': self.platform,
            }
        return {'client_name': 'pyladies_self.bot', 'mastodon': None}

    if self.bot == 'rladies':
        if self.platform == 'bluesky':
            return {
                'client_name': 'rladies_self.bot',
                'api_base_url': self.platform,
                'mastodon': None,
                'password': os.getenv('RLADIES_BSKY_PASSWORD'),
                'username': os.getenv('RLADIES_BSKY_USERNAME'),
                'images': 'anniversary_images',
                'platform': self.platform,
            }
        return {'client_name': 'rladies_self.bot', 'mastodon': None}

    return None

get_config_blog()

Method to generate config for promoting blog posts

Source code in src/debug.py

def get_config_blog(self):
    """Method to generate config for promoting blog posts"""
    if self.bot == 'pyladies':
        if self.platform == 'bluesky':
            return {
                "archive": "pyladies_archive_directory_bluesky",
                "counter": "metadata/pyladies_counter_bluesky.txt",
                "json_file": "metadata/pyladies_meta_data.json",
                "client_name": "pyladies_self.bot",
                "images": "pyladies_images",
                "api_base_url": self.platform,
                "mastodon": None,
                "gen_ai_support": True,
                "gemini_model_name": "gemini-2.5-flash",
                "password": os.getenv("PYLADIES_BSKY_PASSWORD"),
                "username": os.getenv("PYLADIES_BSKY_USERNAME"),
                "platform": self.platform,
            }
        return {
            'archive': 'pyladies_archive_directory',
            'counter': 'pyladies_counter.txt',
            'json_file': 'metadata/pyladies_meta_data.json',
            'client_name': 'pyladies_self.bot',
            'mastodon': None,
        }

    if self.bot == 'rladies':
        if self.platform == 'bluesky':
            return {
                "archive": "rladies_archive_directory_bluesky",
                "counter": "../metadata/rladies_counter_bluesky.txt",
                "json_file": "../metadata/rladies_meta_data.json",
                "client_name": "rladies_self.bot",
                "images": "rladies_images",
                "api_base_url": self.platform,
                "mastodon": None,
                "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                "username": os.getenv("RLADIES_BSKY_USERNAME"),
                "platform": self.platform,
            }
        return {
            "archive": "rladies_archive_directory",
            "counter": "../metadata/rladies_counter.txt",
            "json_file": "../metadata/rladies_meta_data.json",
            "client_name": "rladies_self.bot",
            "mastodon": None,
        }

    return None

get_config_boost()

Method to generate config for boosting tags

Source code in src/debug.py

def get_config_boost(self):
    """Method to generate config for boosting tags"""
    if self.bot == 'pyladies':
        return {"client_name": "pyladies_self.bot", "mastodon": None}

    if self.bot == 'rladies':
        if self.platform == "bluesky":
            return {
                "client_name": "rladies_self.bot",
                "api_base_url": self.platform,
                "mastodon": None,
                "password": os.getenv("PASSWORD"),
                "username": os.getenv("USERNAME"),
                "platform": self.platform,
                "tags": "rladies",
            }
        return {"client_name": "rladies_self.bot", "mastodon": None}

    return None

start_debug()

Start debugging.

Source code in src/debug.py

def start_debug(self):
    """Start debugging."""
    if self.what_to_debug == 'blog':
        config_dict = self.get_config_blog()
        promote_blog_post_handler = PromoteBlogPost(
            config_dict,
            self.no_dry_run
        )
        promote_blog_post_handler.promote_blog_post()

    elif self.what_to_debug == 'rss':
        config_dict = self.get_config_rss()
        rss_data_handler = RSSData(
            config_dict,
            self.no_dry_run
        )
        rss_data_handler.get_rss_data()

    elif self.what_to_debug == 'boost_tags':
        config_dict = self.get_config_boost()
        boost_tags_handler = BoostTags(
            config_dict,
            self.no_dry_run
        )
        boost_tags_handler.boost_tags()

    elif self.what_to_debug == 'boost_mentions':
        config_dict = self.get_config_boost()
        boost_tags_handler = BoostMentions(
            config_dict,
            self.no_dry_run
        )
        boost_tags_handler.boost_mentions()

    elif self.what_to_debug == 'anniversary':
        config_dict = self.get_config_anniversary()
        promote_anniversary_handler = PromoteAnniversary(
            config_dict,
            self.no_dry_run
        )
        promote_anniversary_handler.promote_anniversary()

get_rss_data

Module to get RSS metadata from JSON files.

RSSData

Handle gathering RSS data from JSON files.

Source code in src/get_rss_data.py

class RSSData:
    """
    Handle gathering RSS data from JSON files.
    """

    def __init__(self, config_dict=None, no_dry_run=True):
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)

        self.config_dict = config_dict or {}
        self.no_dry_run = no_dry_run

        if self.no_dry_run:
            self.base_url = os.getenv("BASE_URL")
            self.github_raw_url = os.getenv("GITHUB_RAW_URL")
            self.json_file = os.getenv("JSON_FILE")
        else:
            self.base_url = self.config_dict.get("api_base_url")
            self.github_raw_url = self.config_dict.get("github_raw_url")
            self.json_file = self.config_dict.get("json_file")

    def get_rss_data(self):
        """
        Retrieve and save RSS metadata.
        """
        contents_list = self.get_json_data()
        meta_data = self.get_meta_data(contents_list)

        if self.no_dry_run:
            with open(self.json_file, "w", encoding="utf-8") as fp:
                json.dump(meta_data, fp, ensure_ascii=False, indent=2)

            self.logger.info(
                "Meta data successfully saved to %s",
                self.json_file
            )

    @staticmethod
    def extract_elements(string: str, suffix: str) -> list[str]:
        """
        Extract matching substrings from a given string.

        The method searches for substrings enclosed in double quotes (`"`)
        that end with the provided suffix, excluding any that contain the word
        "blog".

        Args:
            string (str): Input text to search through.
            suffix (str): Suffix pattern to match at the end of elements.

        Returns:
            list[str]: A list of matched substrings.
        """
        pattern = rf'"((?!blog)[^"]*{suffix})"'
        return re.findall(pattern, string)

    def get_json_file_names(self) -> list[str]:
        """
        Retrieve available JSON file names from the configured base URL.

        The method loads the page at `self.base_url`, extracts embedded
        JavaScript data inside the `<react-app>` element, and constructs full
        raw GitHub URLs for each JSON file.

        Returns:
            list[str]: A list of JSON file URLs.

        Raises:
            requests.HTTPError: If the request to `self.base_url` fails.
            json.JSONDecodeError: If the embedded script cannot be parsed
                                    as JSON.
            AttributeError: If the expected DOM structure is missing.
        """
        response = requests.get(self.base_url, timeout=REQUEST_TIMEOUT)
        response.raise_for_status()

        soup = BeautifulSoup(response.content, "html.parser")
        script_tag = soup.find("react-app").find("script")

        payload = json.loads(script_tag.string)
        return [
            f"{self.github_raw_url}/{item['path'].split('/')[-1]}"
            for item in payload["payload"]["tree"]["items"]
        ]

    def get_json_data(self) -> list[dict]:
        """
        Download and parse JSON files from discovered file URLs.

        The method retrieves the list of JSON file URLs via
        `get_json_file_names()`, fetches each file, and loads it into memory.

        Returns:
            list[dict]: A list of parsed JSON objects.

        Raises:
            RuntimeError: If no JSON file URLs were found.
            requests.HTTPError: If fetching a JSON file fails with an HTTP
                                error.
            json.JSONDecodeError: If a response is not valid JSON.
        """
        json_files = self.get_json_file_names()
        if not json_files:
            raise RuntimeError("No JSON files found.")

        contents_list = []
        for json_file in json_files:
            try:
                response = requests.get(json_file, timeout=REQUEST_TIMEOUT)
                response.raise_for_status()
                contents_list.append(response.json())
            except (requests.RequestException, json.JSONDecodeError) as exc:
                self.logger.warning("Could not access %s. %s", json_file, exc)

        return contents_list

    @staticmethod
    def extract_info(content: dict) -> dict:
        """
        Extract metadata information from a single JSON content item.

        The method collects:
        - `name`: The author's name (first entry in `authors`).
        - `rss_feed`: RSS feed URL (prefers `rss_feed`, falls back to
            `rss_feed_youtube`).
        - `mastodon`: Author's Mastodon handle if available.
        - `bluesky`: Author's Bluesky handle if available.

        Args:
            content (dict): Parsed JSON object representing author and
                            feed data.

        Returns:
            dict: A dictionary containing metadata fields.
        """
        rss_feed = [content.get("rss_feed")]
        rss_feed_yt = [content.get("rss_feed_youtube")]

        rss_feed = [a or b for a, b in zip(rss_feed, rss_feed_yt)]
        rss_feed = "" if rss_feed == [None] else rss_feed

        author = content.get("authors", [{}])[0]
        name = author.get("name", "")

        social_media = author.get("social_media", [{}])[0]
        mastodon = social_media.get("mastodon", "")
        bluesky = social_media.get("bluesky", "")

        return {
            "name": name,
            "rss_feed": rss_feed,
            "mastodon": mastodon,
            "bluesky": bluesky,
        }

    def get_meta_data(self, contents_list: list[dict]) -> list[dict]:
        """
        Aggregate metadata from multiple JSON content items.

        Iterates through all content dictionaries, extracts metadata
        using `extract_info()`, and compiles the results into a list.

        Args:
            contents_list (list[dict]): List of parsed JSON content
                                        dictionaries.

        Returns:
            list[dict]: A list of metadata dictionaries.
        """
        meta_data = []
        for content in contents_list:
            content_data = self.extract_info(content)
            if content_data:
                meta_data.append(content_data)
        return meta_data

extract_elements(string, suffix) staticmethod

Extract matching substrings from a given string.

The method searches for substrings enclosed in double quotes (") that end with the provided suffix, excluding any that contain the word "blog".

Parameters:

Name	Type	Description	Default
string	str	Input text to search through.	required
suffix	str	Suffix pattern to match at the end of elements.	required

Returns:

Type	Description
list[str]	list[str]: A list of matched substrings.

Source code in src/get_rss_data.py

@staticmethod
def extract_elements(string: str, suffix: str) -> list[str]:
    """
    Extract matching substrings from a given string.

    The method searches for substrings enclosed in double quotes (`"`)
    that end with the provided suffix, excluding any that contain the word
    "blog".

    Args:
        string (str): Input text to search through.
        suffix (str): Suffix pattern to match at the end of elements.

    Returns:
        list[str]: A list of matched substrings.
    """
    pattern = rf'"((?!blog)[^"]*{suffix})"'
    return re.findall(pattern, string)

extract_info(content) staticmethod

Extract metadata information from a single JSON content item.

The method collects: - name: The author's name (first entry in authors). - rss_feed: RSS feed URL (prefers rss_feed, falls back to rss_feed_youtube). - mastodon: Author's Mastodon handle if available. - bluesky: Author's Bluesky handle if available.

Parameters:

Name	Type	Description	Default
content	dict	Parsed JSON object representing author and feed data.	required

Returns:

Name	Type	Description
dict	dict	A dictionary containing metadata fields.

Source code in src/get_rss_data.py

@staticmethod
def extract_info(content: dict) -> dict:
    """
    Extract metadata information from a single JSON content item.

    The method collects:
    - `name`: The author's name (first entry in `authors`).
    - `rss_feed`: RSS feed URL (prefers `rss_feed`, falls back to
        `rss_feed_youtube`).
    - `mastodon`: Author's Mastodon handle if available.
    - `bluesky`: Author's Bluesky handle if available.

    Args:
        content (dict): Parsed JSON object representing author and
                        feed data.

    Returns:
        dict: A dictionary containing metadata fields.
    """
    rss_feed = [content.get("rss_feed")]
    rss_feed_yt = [content.get("rss_feed_youtube")]

    rss_feed = [a or b for a, b in zip(rss_feed, rss_feed_yt)]
    rss_feed = "" if rss_feed == [None] else rss_feed

    author = content.get("authors", [{}])[0]
    name = author.get("name", "")

    social_media = author.get("social_media", [{}])[0]
    mastodon = social_media.get("mastodon", "")
    bluesky = social_media.get("bluesky", "")

    return {
        "name": name,
        "rss_feed": rss_feed,
        "mastodon": mastodon,
        "bluesky": bluesky,
    }

get_json_data()

Download and parse JSON files from discovered file URLs.

The method retrieves the list of JSON file URLs via get_json_file_names(), fetches each file, and loads it into memory.

Returns:

Type	Description
list[dict]	list[dict]: A list of parsed JSON objects.

Raises:

Type	Description
RuntimeError	If no JSON file URLs were found.
HTTPError	If fetching a JSON file fails with an HTTP error.
JSONDecodeError	If a response is not valid JSON.

Source code in src/get_rss_data.py

def get_json_data(self) -> list[dict]:
    """
    Download and parse JSON files from discovered file URLs.

    The method retrieves the list of JSON file URLs via
    `get_json_file_names()`, fetches each file, and loads it into memory.

    Returns:
        list[dict]: A list of parsed JSON objects.

    Raises:
        RuntimeError: If no JSON file URLs were found.
        requests.HTTPError: If fetching a JSON file fails with an HTTP
                            error.
        json.JSONDecodeError: If a response is not valid JSON.
    """
    json_files = self.get_json_file_names()
    if not json_files:
        raise RuntimeError("No JSON files found.")

    contents_list = []
    for json_file in json_files:
        try:
            response = requests.get(json_file, timeout=REQUEST_TIMEOUT)
            response.raise_for_status()
            contents_list.append(response.json())
        except (requests.RequestException, json.JSONDecodeError) as exc:
            self.logger.warning("Could not access %s. %s", json_file, exc)

    return contents_list

get_json_file_names()

Retrieve available JSON file names from the configured base URL.

The method loads the page at self.base_url, extracts embedded JavaScript data inside the <react-app> element, and constructs full raw GitHub URLs for each JSON file.

Returns:

Type	Description
list[str]	list[str]: A list of JSON file URLs.

Raises:

Type	Description
HTTPError	If the request to self.base_url fails.
JSONDecodeError	If the embedded script cannot be parsed as JSON.
AttributeError	If the expected DOM structure is missing.

Source code in src/get_rss_data.py

def get_json_file_names(self) -> list[str]:
    """
    Retrieve available JSON file names from the configured base URL.

    The method loads the page at `self.base_url`, extracts embedded
    JavaScript data inside the `<react-app>` element, and constructs full
    raw GitHub URLs for each JSON file.

    Returns:
        list[str]: A list of JSON file URLs.

    Raises:
        requests.HTTPError: If the request to `self.base_url` fails.
        json.JSONDecodeError: If the embedded script cannot be parsed
                                as JSON.
        AttributeError: If the expected DOM structure is missing.
    """
    response = requests.get(self.base_url, timeout=REQUEST_TIMEOUT)
    response.raise_for_status()

    soup = BeautifulSoup(response.content, "html.parser")
    script_tag = soup.find("react-app").find("script")

    payload = json.loads(script_tag.string)
    return [
        f"{self.github_raw_url}/{item['path'].split('/')[-1]}"
        for item in payload["payload"]["tree"]["items"]
    ]

get_meta_data(contents_list)

Aggregate metadata from multiple JSON content items.

Iterates through all content dictionaries, extracts metadata using extract_info(), and compiles the results into a list.

Parameters:

Name	Type	Description	Default
contents_list	list[dict]	List of parsed JSON content dictionaries.	required

Returns:

Type	Description
list[dict]	list[dict]: A list of metadata dictionaries.

Source code in src/get_rss_data.py

def get_meta_data(self, contents_list: list[dict]) -> list[dict]:
    """
    Aggregate metadata from multiple JSON content items.

    Iterates through all content dictionaries, extracts metadata
    using `extract_info()`, and compiles the results into a list.

    Args:
        contents_list (list[dict]): List of parsed JSON content
                                    dictionaries.

    Returns:
        list[dict]: A list of metadata dictionaries.
    """
    meta_data = []
    for content in contents_list:
        content_data = self.extract_info(content)
        if content_data:
            meta_data.append(content_data)
    return meta_data

get_rss_data()

Retrieve and save RSS metadata.

Source code in src/get_rss_data.py

def get_rss_data(self):
    """
    Retrieve and save RSS metadata.
    """
    contents_list = self.get_json_data()
    meta_data = self.get_meta_data(contents_list)

    if self.no_dry_run:
        with open(self.json_file, "w", encoding="utf-8") as fp:
            json.dump(meta_data, fp, ensure_ascii=False, indent=2)

        self.logger.info(
            "Meta data successfully saved to %s",
            self.json_file
        )

helper

check_length_anniversary

Script to check for length of content

check_entries(data)

Function to check if the combined length of name, description, and wiki_link exceeds 500 characters

Source code in src/helper/check_length_anniversary.py

def check_entries(data):
    """
    Function to check if the combined length of name, description, and wiki_link exceeds 500 characters
    """
    if not data:
        return

    for entry in data:
        # Combine name, description, and wiki_link fields
        combined_text = ""
        combined_text += f"Let's meet {entry.get('name', '')} ✨\n\n{entry.get('description', '')}\n\n🔗 {entry.get('wiki_link', '')}"
        combined_text += f"\n\n#amazingwomeninstem #womeninstem #womenalsoknow #impactthefuture"

        # Check the length of the combined text
        if len(combined_text) > 500:
            print(f"🚨 Alert: The combined text for '{entry.get('name', 'Unknown')}' exceeds 500 characters!")
            print(f"Combined length: {len(combined_text)} characters.")
            print(combined_text)
            print(f"Length of description: {len(entry.get('description', ''))}.")
            sys.exit(1)  # Exit with an error code to indicate failure

load_json(filename)

Function to load JSON data from a file.

Source code in src/helper/check_length_anniversary.py

def load_json(filename):
    """
    Function to load JSON data from a file.
    """
    try:
        with open(filename, 'r') as file:
            return json.load(file)
    except FileNotFoundError:
        print(f"Error: The file '{filename}' was not found.")
        return None
    except json.JSONDecodeError:
        print(f"Error: The file '{filename}' contains invalid JSON.")
        return None

login_bluesky

Module to login to Bluesky

login_mastodon

Module to log into Mastodon

promote_anniversaries

Module to promote anniversaries on Mastodon and Bluesky. Handles fetching events, building posts, and posting to platforms.

PromoteAnniversary

Handles fetching event data and posting anniversary messages to social platforms.

Source code in src/promote_anniversaries.py

class PromoteAnniversary:
    """
    Handles fetching event data and posting anniversary messages
    to social platforms.
    """
    def __init__(self, config_dict=None, no_dry_run=True):
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger(__name__)
        self.config_dict = config_dict
        self.no_dry_run = no_dry_run

    def promote_anniversary(self):
        """
        Method to promote anniversaries on social media.
        """
        if (self.config_dict is None) and (self.no_dry_run):
            self.config_dict = {
                "platform": os.getenv("PLATFORM"),
                "images": os.getenv("IMAGES"),
                "password": os.getenv("PASSWORD"),
                "username": os.getenv("USERNAME"),
                "client_name": os.getenv("CLIENT_NAME")
            }
            if self.config_dict["platform"] == "mastodon":
                self.config_dict["api_base_url"] = config.API_BASE_URL
                self.config_dict[
                    "mastodon_visibility"
                ] = config.MASTODON_VISIBILITY
                self.config_dict["client_id"] = os.getenv("CLIENT_ID")
                self.config_dict["client_secret"] = os.getenv("CLIENT_SECRET")
                self.config_dict["access_token"] = os.getenv("ACCESS_TOKEN")
                self.config_dict[
                    "client_cred_file"
                ] = os.getenv('BOT_CLIENTCRED_SECRET')
            else:
                self.config_dict["api_base_url"] = "bluesky"

        if self.no_dry_run:
            self.logger.info("")
            self.logger.info(
                "Initializing %s Bot",
                self.config_dict["client_name"]
            )
            self.logger.info(
                "%s",
                "=" * (len(self.config_dict["client_name"]) + 17)
            )
            self.logger.info(
                " > Connecting to %s",
                self.config_dict["api_base_url"]
            )

            if self.config_dict["platform"] == "mastodon":
                _, client = login_mastodon(self.config_dict)
            elif self.config_dict["platform"] == "bluesky":
                client = login_bluesky(self.config_dict)
        else:
            client = None

        with open('metadata/events.json', encoding='utf-8') as f:
            events = json.load(f)

        if self.no_dry_run:
            for event in events:
                if self.is_matching_current_date(event["date"]):
                    self.send_post(event, client)
                    continue
                    # if  self.config_dict["platform"] == "mastodon":
                    #     send_post_to_mastodon(
                    #         event,
                    #         self.config_dict,
                    #         client
                    #     )
                    #     continue
                    # elif  self.config_dict["platform"] == "bluesky":
                    #     send_post_to_bluesky(
                    #         event,
                    #         self.config_dict,
                    #         client
                    #     )
                    #    continue

    @staticmethod
    def is_matching_current_date(date_str: str, date_format='%m-%d') -> bool:
        """
        Method to define if the event matches the current date and
        should be posted.

        Args:
            date_str (str): Date taken from event dictionary
            date_format (str, optional): _description_. Defaults to '%m-%d'.

        Returns:
            bool: Defines whether the date matches the current date
                (True if yes)
        """
        current_date = datetime.now().strftime(date_format)
        return date_str == current_date

    def download_image(self, url: str) -> str:
        """
        Method downloads images. It's heavily inspired by:
        https://github.com/zeratax/mastodon-img-bot/blob/master/bot.py

        Args:
            url: string with the url to the image

        Returns:
            string with the path to the saved image
        """
        path = urlsplit(url).path
        filename = posixpath.basename(path)

        file_path = f"{self.config_dict['images']}/{filename}"
        if not os.path.isfile(file_path):
            if not os.path.isdir(self.config_dict['images']):
                os.makedirs(self.config_dict['images'])

            headers = {
                'User-Agent': (
                    'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:20.0) '
                    'Gecko/20100101 Firefox/20.0'
                )
            }
            response = requests.get(
                url,
                headers=headers,
                stream=True,
                timeout=REQUEST_TIMEOUT
            )

            with open(file_path, 'wb') as out_file:
                shutil.copyfileobj(
                    response.raw, out_file)
            del response
        else:
            print("Image already downloaded")
        return file_path

    def build_post(self, event: dict):
        """Method to build the toot

        Args:
            event (dict): Dictionary with information

        Returns:
            Toot text
        """
        tags = "\n\n#amazingwomenintech #womenalsoknow #impactthefuture"

        if self.config_dict["platform"] == "mastodon":
            toot_str = ""
            toot_str += (
                f"Let's meet {event['name']} ✨\n\n"
                f"{event['description_mastodon']}\n\n"
                f"🔗 {event['wiki_link']}"
            )
            toot_str += tags
            return toot_str
        if self.config_dict["platform"] == "bluesky":
            text_builder = client_utils.TextBuilder()
            if event["bluesky"]:
                did = self.get_bluesky_did(event["bluesky"])
                text_builder.text("Let's meet ")
                text_builder.mention(f"{event['bluesky']}", did)
                text_builder.text(" ⭐️\n\n")
            else:
                text_builder.text(f"Let's meet {event['name']} ⭐️\n\n")
            split_text = re.split(r'(#\w+)', event["description_bluesky"])
            split_text = [
                item.rstrip(' ')
                for item in split_text
                if item.strip()
            ]
            for text_chunk in split_text:
                if text_chunk.startswith('#'):
                    for tag in text_chunk.split("#"):
                        tag_clean = tag.strip()
                    if tag_clean:
                        text_builder.tag(f"#{tag_clean}", tag_clean)
                else:
                    text_chunk_clean = self.add_whitespace_if_needed(
                        text_chunk
                    )
                    text_builder.text(text_chunk_clean)
            text_builder.text('\n\n🔗 ')
            text_builder.link(event["wiki_link"], event["wiki_link"])
            text_builder.text('\n\n')
            for tag in tags.split("#"):
                tag_clean = tag.strip()
                if tag_clean:
                    text_builder.tag(f"#{tag_clean} ", tag_clean)
            return text_builder

    def send_post(self, event, client):
        """Send a post to the configured platform (Mastodon or Bluesky)."""

        self.logger.info(
            """
            Preparing the post on %s (%s) ...
            """,
            self.config_dict['client_name'],
            self.config_dict['platform']
        )

        post_txt = self.build_post(event)
        if self.config_dict["platform"] == "mastodon":
            self.send_post_to_mastodon(event, client, post_txt)
        elif self.config_dict["platform"] == "bluesky":
            embed_external = self.build_embed_external(event, client)
            self.send_post_to_bluesky(event, client, post_txt, embed_external)

    def build_embed_external(self, event, client):
        """Build external embed object for Bluesky posts."""
        repo_url = (
            "https://raw.githubusercontent.com/cosimameyer/illustrations/main"
        )
        base_path = f"{repo_url}/amazing-women"
        url = f"{base_path}/{event['img']}"
        filename = self.download_image(url)
        with open(filename, 'rb') as f:
            img_data = f.read()

        thumb = client.upload_blob(img_data)

        return models.AppBskyEmbedExternal.Main(
            external=models.AppBskyEmbedExternal.External(
                title=f"Image of {event['name']}",
                description=event["alt"],
                uri=url,
                thumb=thumb.blob,
            )
        )

    @staticmethod
    def get_bluesky_did(platform_user_handle: str):
        """
        Method to extract the Bluesky specific unique user ID (`did`).

        Args:
            platform_user_handle (str): User handle of bluesky

        Returns:
            str: did
        """
        url = (
            f"https://bsky.social/xrpc/com.atproto.identity.resolveHandle?"
            f"handle={platform_user_handle.lstrip('@')}"
        )
        try:
            response = requests.get(
                url,
                timeout=REQUEST_TIMEOUT
            )

            if response.status_code == 200:
                data = response.json()

                did = data.get("did")

                if did:
                    return did
                print("The 'did' field was not found in the response.")
            print(
                f"Failed to retrieve data. Status code: {response.status_code}"
            )

        except requests.RequestException as e:
            print("An error occurred:", e)

    def send_post_to_bluesky(self, event, client, post_txt, embed_external):
        """Send a post to Bluesky with optional media embed."""
        self.logger.info(
            'Preview your post...\n\n%s',
            post_txt._buffer.getvalue().decode('utf-8')
        )
        try:
            client.send_post(text=post_txt, embed=embed_external)
            self.logger.info("Posted 🎉")
        except Exception as e:
            self.logger.exception("Urg, exception %s for %s", e, event['name'])

    @staticmethod
    def add_whitespace_if_needed(text_chunk):
        if not text_chunk.endswith(('(', '{', '[')):
            text_chunk += ' '
        return text_chunk

    def send_post_to_mastodon(self, event, client, post_txt):
        """Send a post to Mastodon, with media if available."""
        if event['img']:
            try:
                print("Uploading media to mastodon")
                base_path = (
                    "https://raw.githubusercontent.com/"
                    "cosimameyer/illustrations/main/"
                    "amazing-women"
                )
                url = f"{base_path}/{event['img']}"

                filename = self.download_image(url)
                media_upload_mastodon = client.media_post(filename)

                print("adding description")
                if event["alt"]:
                    client.media_update(media_upload_mastodon,
                                        description=event["alt"])
                else:
                    client.media_update(media_upload_mastodon,
                                        description=str(event["name"]))

                print("ready to post")
                client.status_post(
                    post_txt,
                    media_ids=media_upload_mastodon
                )

                print("posted")
            except Exception as e:
                self.logger.info(
                    """
                    Urg, media could not be printed.\n
                    Exception %s because of %s
                    """,
                    event['name'],
                    e
                )
                client.status_post(post_txt)
                self.logger.info("Posted toot without image.")
        else:
            try:
                client.status_post(post_txt)
                self.logger.info("posted")
            except Exception as e:
                self.logger.info(
                    "Urg, exception %s. The reason was %s",
                    event['toot'],
                    e
                )

build_embed_external(event, client)

Build external embed object for Bluesky posts.

Source code in src/promote_anniversaries.py

def build_embed_external(self, event, client):
    """Build external embed object for Bluesky posts."""
    repo_url = (
        "https://raw.githubusercontent.com/cosimameyer/illustrations/main"
    )
    base_path = f"{repo_url}/amazing-women"
    url = f"{base_path}/{event['img']}"
    filename = self.download_image(url)
    with open(filename, 'rb') as f:
        img_data = f.read()

    thumb = client.upload_blob(img_data)

    return models.AppBskyEmbedExternal.Main(
        external=models.AppBskyEmbedExternal.External(
            title=f"Image of {event['name']}",
            description=event["alt"],
            uri=url,
            thumb=thumb.blob,
        )
    )

build_post(event)

Method to build the toot

Parameters:

Name	Type	Description	Default
event	dict	Dictionary with information	required

Returns:

Type	Description
	Toot text

Source code in src/promote_anniversaries.py

def build_post(self, event: dict):
    """Method to build the toot

    Args:
        event (dict): Dictionary with information

    Returns:
        Toot text
    """
    tags = "\n\n#amazingwomenintech #womenalsoknow #impactthefuture"

    if self.config_dict["platform"] == "mastodon":
        toot_str = ""
        toot_str += (
            f"Let's meet {event['name']} ✨\n\n"
            f"{event['description_mastodon']}\n\n"
            f"🔗 {event['wiki_link']}"
        )
        toot_str += tags
        return toot_str
    if self.config_dict["platform"] == "bluesky":
        text_builder = client_utils.TextBuilder()
        if event["bluesky"]:
            did = self.get_bluesky_did(event["bluesky"])
            text_builder.text("Let's meet ")
            text_builder.mention(f"{event['bluesky']}", did)
            text_builder.text(" ⭐️\n\n")
        else:
            text_builder.text(f"Let's meet {event['name']} ⭐️\n\n")
        split_text = re.split(r'(#\w+)', event["description_bluesky"])
        split_text = [
            item.rstrip(' ')
            for item in split_text
            if item.strip()
        ]
        for text_chunk in split_text:
            if text_chunk.startswith('#'):
                for tag in text_chunk.split("#"):
                    tag_clean = tag.strip()
                if tag_clean:
                    text_builder.tag(f"#{tag_clean}", tag_clean)
            else:
                text_chunk_clean = self.add_whitespace_if_needed(
                    text_chunk
                )
                text_builder.text(text_chunk_clean)
        text_builder.text('\n\n🔗 ')
        text_builder.link(event["wiki_link"], event["wiki_link"])
        text_builder.text('\n\n')
        for tag in tags.split("#"):
            tag_clean = tag.strip()
            if tag_clean:
                text_builder.tag(f"#{tag_clean} ", tag_clean)
        return text_builder

download_image(url)

Method downloads images. It's heavily inspired by: https://github.com/zeratax/mastodon-img-bot/blob/master/bot.py

Parameters:

Name	Type	Description	Default
url	str	string with the url to the image	required

Returns:

Type	Description
str	string with the path to the saved image

Source code in src/promote_anniversaries.py

def download_image(self, url: str) -> str:
    """
    Method downloads images. It's heavily inspired by:
    https://github.com/zeratax/mastodon-img-bot/blob/master/bot.py

    Args:
        url: string with the url to the image

    Returns:
        string with the path to the saved image
    """
    path = urlsplit(url).path
    filename = posixpath.basename(path)

    file_path = f"{self.config_dict['images']}/{filename}"
    if not os.path.isfile(file_path):
        if not os.path.isdir(self.config_dict['images']):
            os.makedirs(self.config_dict['images'])

        headers = {
            'User-Agent': (
                'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:20.0) '
                'Gecko/20100101 Firefox/20.0'
            )
        }
        response = requests.get(
            url,
            headers=headers,
            stream=True,
            timeout=REQUEST_TIMEOUT
        )

        with open(file_path, 'wb') as out_file:
            shutil.copyfileobj(
                response.raw, out_file)
        del response
    else:
        print("Image already downloaded")
    return file_path

get_bluesky_did(platform_user_handle) staticmethod

Method to extract the Bluesky specific unique user ID (did).

Parameters:

Name	Type	Description	Default
platform_user_handle	str	User handle of bluesky	required

Returns:

Name	Type	Description
str		did

Source code in src/promote_anniversaries.py

@staticmethod
def get_bluesky_did(platform_user_handle: str):
    """
    Method to extract the Bluesky specific unique user ID (`did`).

    Args:
        platform_user_handle (str): User handle of bluesky

    Returns:
        str: did
    """
    url = (
        f"https://bsky.social/xrpc/com.atproto.identity.resolveHandle?"
        f"handle={platform_user_handle.lstrip('@')}"
    )
    try:
        response = requests.get(
            url,
            timeout=REQUEST_TIMEOUT
        )

        if response.status_code == 200:
            data = response.json()

            did = data.get("did")

            if did:
                return did
            print("The 'did' field was not found in the response.")
        print(
            f"Failed to retrieve data. Status code: {response.status_code}"
        )

    except requests.RequestException as e:
        print("An error occurred:", e)

is_matching_current_date(date_str, date_format='%m-%d') staticmethod

Method to define if the event matches the current date and should be posted.

Parameters:

Name	Type	Description	Default
date_str	str	Date taken from event dictionary	required
date_format	str	description. Defaults to '%m-%d'.	'%m-%d'

Returns:

Name	Type	Description
bool	bool	Defines whether the date matches the current date (True if yes)

Source code in src/promote_anniversaries.py

@staticmethod
def is_matching_current_date(date_str: str, date_format='%m-%d') -> bool:
    """
    Method to define if the event matches the current date and
    should be posted.

    Args:
        date_str (str): Date taken from event dictionary
        date_format (str, optional): _description_. Defaults to '%m-%d'.

    Returns:
        bool: Defines whether the date matches the current date
            (True if yes)
    """
    current_date = datetime.now().strftime(date_format)
    return date_str == current_date

promote_anniversary()

Method to promote anniversaries on social media.

Source code in src/promote_anniversaries.py

def promote_anniversary(self):
    """
    Method to promote anniversaries on social media.
    """
    if (self.config_dict is None) and (self.no_dry_run):
        self.config_dict = {
            "platform": os.getenv("PLATFORM"),
            "images": os.getenv("IMAGES"),
            "password": os.getenv("PASSWORD"),
            "username": os.getenv("USERNAME"),
            "client_name": os.getenv("CLIENT_NAME")
        }
        if self.config_dict["platform"] == "mastodon":
            self.config_dict["api_base_url"] = config.API_BASE_URL
            self.config_dict[
                "mastodon_visibility"
            ] = config.MASTODON_VISIBILITY
            self.config_dict["client_id"] = os.getenv("CLIENT_ID")
            self.config_dict["client_secret"] = os.getenv("CLIENT_SECRET")
            self.config_dict["access_token"] = os.getenv("ACCESS_TOKEN")
            self.config_dict[
                "client_cred_file"
            ] = os.getenv('BOT_CLIENTCRED_SECRET')
        else:
            self.config_dict["api_base_url"] = "bluesky"

    if self.no_dry_run:
        self.logger.info("")
        self.logger.info(
            "Initializing %s Bot",
            self.config_dict["client_name"]
        )
        self.logger.info(
            "%s",
            "=" * (len(self.config_dict["client_name"]) + 17)
        )
        self.logger.info(
            " > Connecting to %s",
            self.config_dict["api_base_url"]
        )

        if self.config_dict["platform"] == "mastodon":
            _, client = login_mastodon(self.config_dict)
        elif self.config_dict["platform"] == "bluesky":
            client = login_bluesky(self.config_dict)
    else:
        client = None

    with open('metadata/events.json', encoding='utf-8') as f:
        events = json.load(f)

    if self.no_dry_run:
        for event in events:
            if self.is_matching_current_date(event["date"]):
                self.send_post(event, client)
                continue

send_post(event, client)

Send a post to the configured platform (Mastodon or Bluesky).

Source code in src/promote_anniversaries.py

def send_post(self, event, client):
    """Send a post to the configured platform (Mastodon or Bluesky)."""

    self.logger.info(
        """
        Preparing the post on %s (%s) ...
        """,
        self.config_dict['client_name'],
        self.config_dict['platform']
    )

    post_txt = self.build_post(event)
    if self.config_dict["platform"] == "mastodon":
        self.send_post_to_mastodon(event, client, post_txt)
    elif self.config_dict["platform"] == "bluesky":
        embed_external = self.build_embed_external(event, client)
        self.send_post_to_bluesky(event, client, post_txt, embed_external)

send_post_to_bluesky(event, client, post_txt, embed_external)

Send a post to Bluesky with optional media embed.

Source code in src/promote_anniversaries.py

def send_post_to_bluesky(self, event, client, post_txt, embed_external):
    """Send a post to Bluesky with optional media embed."""
    self.logger.info(
        'Preview your post...\n\n%s',
        post_txt._buffer.getvalue().decode('utf-8')
    )
    try:
        client.send_post(text=post_txt, embed=embed_external)
        self.logger.info("Posted 🎉")
    except Exception as e:
        self.logger.exception("Urg, exception %s for %s", e, event['name'])

send_post_to_mastodon(event, client, post_txt)

Send a post to Mastodon, with media if available.

Source code in src/promote_anniversaries.py

def send_post_to_mastodon(self, event, client, post_txt):
    """Send a post to Mastodon, with media if available."""
    if event['img']:
        try:
            print("Uploading media to mastodon")
            base_path = (
                "https://raw.githubusercontent.com/"
                "cosimameyer/illustrations/main/"
                "amazing-women"
            )
            url = f"{base_path}/{event['img']}"

            filename = self.download_image(url)
            media_upload_mastodon = client.media_post(filename)

            print("adding description")
            if event["alt"]:
                client.media_update(media_upload_mastodon,
                                    description=event["alt"])
            else:
                client.media_update(media_upload_mastodon,
                                    description=str(event["name"]))

            print("ready to post")
            client.status_post(
                post_txt,
                media_ids=media_upload_mastodon
            )

            print("posted")
        except Exception as e:
            self.logger.info(
                """
                Urg, media could not be printed.\n
                Exception %s because of %s
                """,
                event['name'],
                e
            )
            client.status_post(post_txt)
            self.logger.info("Posted toot without image.")
    else:
        try:
            client.status_post(post_txt)
            self.logger.info("posted")
        except Exception as e:
            self.logger.info(
                "Urg, exception %s. The reason was %s",
                event['toot'],
                e
            )

promote_blog_post

Promote blog posts

PromoteBlogPost

Class to handle promoting blog posts by the community bots.

Source code in src/promote_blog_post.py

class PromoteBlogPost():
    """
    Class to handle promoting blog posts by the community bots.
    """
    def __init__(self, config_dict=None, no_dry_run=True):
        self.logger = logging.getLogger(__name__)
        logging.basicConfig(level=logging.INFO)

        self.process_images = False
        self.no_dry_run = no_dry_run
        self.config_dict = config_dict

    def get_config(self):
        """
        Get config file
        """
        if (self.config_dict is None) and (self.no_dry_run):
            self.config_dict = {
                "platform": os.getenv("PLATFORM"),
                "archive": os.getenv("ARCHIVE_DIRECTORY"),
                "images": os.getenv("IMAGES"),
                "counter": self._ensure_metadata_prefix(
                    os.getenv("COUNTER", "")
                ),
                "password": os.getenv("PASSWORD"),
                "username": os.getenv("USERNAME"),
                "client_name": os.getenv("CLIENT_NAME"),
                "json_file": self._ensure_metadata_prefix(
                    os.getenv("JSON_FILE", "")
                ),
                "gen_ai_support": True,
                "gemini_api_key": os.getenv("GEMINI_API_KEY"),
                "gemini_model_name": "gemini-2.5-flash"
            }
            if self.config_dict["platform"] == "mastodon":
                self.config_dict["api_base_url"] = config.API_BASE_URL
                self.config_dict["mastodon_visibility"] = (
                    config.MASTODON_VISIBILITY
                )
                self.config_dict["client_id"] = os.getenv("CLIENT_ID")
                self.config_dict["client_secret"] = os.getenv("CLIENT_SECRET")
                self.config_dict["access_token"] = os.getenv("ACCESS_TOKEN")
                self.config_dict["client_cred_file"] = os.getenv(
                    'BOT_CLIENTCRED_SECRET'
                )
            else:
                self.config_dict["api_base_url"] = "bluesky"

            if self.config_dict["gen_ai_support"]:
                genai.configure(api_key=self.config_dict["gemini_api_key"])
        else:
            self.config_dict['json_file'] = self._ensure_metadata_prefix(
                self.config_dict.get('json_file')
            )
            self.config_dict['counter'] = self._ensure_metadata_prefix(
                self.config_dict.get('counter')
            )

    def promote_blog_post(self):
        """Core method to promote blog post"""

        self.get_config()

        if self.no_dry_run:
            client_name = self.config_dict.get('client_name', 'unknown')
            self.logger.info("")
            self.logger.info(
                'Initializing %s Bot',
                client_name
            )
            separator = "%s", "=" * (len(client_name) + 17)
            self.logger.info(separator)
            self.logger.info(
                " > Connecting to %s",
                self.config_dict.get('api_base_url', '')
            )

            if self.config_dict["platform"] == "mastodon":
                _, client = login_mastodon(self.config_dict)
            elif self.config_dict["platform"] == "bluesky":
                client = login_bluesky(self.config_dict)
        else:
            client = None

        feeds = self.read_metadata_json()
        counter_name = self.read_counter_name()

        # Initiate count to post a maximum of 2 posts per run
        count_post = 0

        # Drop empty rss_feeds
        feeds = [x for x in feeds if x['rss_feed'] != '']

        if self.no_dry_run:
            self.process_feeds(feeds, counter_name, count_post, client)
        else:
            for feed in feeds:
                count_post = self.process_feed(
                    feed,
                    count_post,
                    client
                )

    def process_feeds(self, feeds, counter_name, count_post, client):
        """
        Method to handle processing of all feeds.
        """
        for feed in feeds:
            if counter_name not in (feed['name'], '\n', ''):
                continue
            if len(feed['rss_feed']) == 0 or feed['rss_feed'] == [None]:
                continue

            is_last_feed = feed['name'] == feeds[-1]['name']

            if count_post == 0 and is_last_feed:
                count_post = self.process_feed(
                    feed,
                    count_post,
                    client
                )

                # Add the counter_name
                if is_last_feed:
                    new_feed = feeds[0]
                    count_post = self.process_feed(
                        new_feed,
                        count_post,
                        client
                    )

                    self.logger.info(
                        "Successfully promoted blog posts. "
                        "Thank you and see you next time!")
                    self.update_counter(feeds[1]['name'])
                    break

            elif count_post < 2:
                count_post = self.process_feed(
                    feed,
                    count_post,
                    client
                )
                counter_name = ''
                if is_last_feed:
                    self.update_counter(feed['name'])
                self.logger.info(
                    "=========================================")

            else:
                self.logger.info(
                    "Successfully promoted blog posts. "
                    "Thank you and see you next time!")
                self.update_counter(feed['name'])
                break

    def update_counter(self, counter_name):
        """
        Update counter name
        """
        with open(
            self.config_dict["counter"],
            'w',
            encoding='utf-8'
        ) as txt_file:
            txt_file.write(counter_name)

    def read_counter_name(self):
        """
        Read counter name from txt file
        """
        with open(self.config_dict["counter"], 'r', encoding='utf-8') as f:
            return f.read()

    def read_metadata_json(self):
        """
        Read metadata JSON file
        """
        with open(self.config_dict["json_file"], 'rb') as fp:
            self.logger.info(
                "============================================="
            )
            feeds = json.load(fp)
            self.logger.info('Meta data was successfully loaded')
            self.logger.info(
                "============================================="
            )
            return feeds

    @staticmethod
    def _ensure_metadata_prefix(value: str, prefix="metadata/") -> str:
        """
        Ensures that a string has the prefix "metadata/". If it does not
        have this, update it.
        """
        if not value.startswith(prefix):
            return prefix + value
        return value

    def download_image(self, url: str):
        """
        Downloads an image from the given URL and saves it locally,
        organizing files by domain name.
        """
        try:
            filename = ''
            # Parse the URL components
            if self.config_dict["platform"] == "bluesky":
                domain = urlsplit(url).path
                filename = posixpath.basename(domain)
            elif self.config_dict["platform"] == "mastodon":
                domain = urlsplit(url).netloc
                filename = posixpath.basename(urlsplit(url).path)

            # Create folder structure based on the domain name
            domain_dir = Path(self.config_dict['images']) / domain
            domain_dir.mkdir(parents=True, exist_ok=True)

            # Full file path for the image
            file_path = domain_dir / filename

            if file_path.is_file():
                self.logger.info("Image already downloaded: %s", file_path)
                return str(file_path)

            # Set user-agent headers for the request
            headers = {
                'User-Agent': (
                    'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:20.0) '
                    'Gecko/20100101 Firefox/20.0'
                )
            }

            # Download the image
            self.logger.info("Downloading image from %s...", url)
            response = requests.get(
                url,
                headers=headers,
                stream=True,
                timeout=15
            )
            response.raise_for_status()  # Raises an exception for HTTP errors

            # Save the image to the designated path
            with open(file_path, 'wb') as out_file:
                shutil.copyfileobj(response.raw, out_file)

            self.logger.info("Image successfully downloaded: %s", file_path)
            return str(file_path)

        except requests.exceptions.RequestException as e:
            self.logger.error("Failed to download image from %s: %e", url, e)
            return None
        except OSError as e:
            self.logger.error("File system error while saving image: %s", e)
            return None
        finally:
            if 'response' in locals():
                response.close()

    def parse_pub_date(self, entry):
        """Method to parse the publication date"""
        date_formats = [
            "%a, %d %b %Y %H:%M:%S %z",  # Format 1
            "%a, %d %b %Y %H:%M:%S %Z",  # Format 2
            "%Y-%m-%d",                  # Format 3
            "%Y-%m-%dT%H:%M:%S.%f%Z"     # Format 4
        ]

        pub_date_str = entry.get('pub_date', '')

        for date_format in date_formats:
            try:
                pub_date = datetime.strptime(
                    pub_date_str, date_format).replace(tzinfo=None)
                return pub_date  # Return as soon as a valid format is found
            except ValueError:
                self.logger.info(
                    "Failed to parse date with format: %s",
                    date_format
                )

        # If none of the formats match, use the current date as a fallback
        self.logger.warning(
            "No matching date format found. Using current date."
        )
        return datetime.now()  # Fallback value

    def define_tags(self, entry):
        """
        Define tags that will be posted along the posts.
        """
        if self.config_dict.get('client_name', '') == 'pyladies_bot':
            tags = '#pyladies #python '
        elif self.config_dict.get('client_name', '') == 'rladies_bot':
            tags = '#rladies #rstats '
        else:
            self.logger.info('Bot name not found')
            tags = ''

        pub_date = self.parse_pub_date(entry)

        age_of_post = datetime.now() - pub_date

        if age_of_post.days > 730:
            tags += '#oldiebutgoodie '

        if len(entry['tags']) > 0:
            for tag in entry['tags']:
                if tag.lower() in ['pyladies', 'python', 'rstats', 'rladies']:
                    pass
                else:
                    tags += (
                        f"#{tag.replace(' ', '').replace('-', '').lower()} "
                    )

        return tags

    def get_bluesky_did(self, platform_user_handle):
        """
        Method to get Bluesky DID to uniquely identify (and tag) user.
        """
        url = (
            f"https://bsky.social/xrpc/com.atproto.identity.resolveHandle?"
            f"handle={platform_user_handle.lstrip('@')}"
        )
        try:
            response = requests.get(url)

            if response.status_code == 200:
                data = response.json()
                did = data.get('did', None)

                if did:
                    return did
                else:
                    self.logger.info(
                        'The "did" field was not found in the response.'
                    )
            else:
                self.logger.info(
                    'Failed to retrieve data. Status code: %s',
                    response.status_code
                )

        except requests.RequestException as e:
            self.logger.info('An error occurred: %s', e)

    def build_post_mastodon(
        self, basis_text, platform_user_handle, tags, entry
    ):
        """
        Build Mastodon post.
        """
        platform_user_handle = self.check_platform_handle(platform_user_handle)

        if platform_user_handle:
            basis_text += f" ({platform_user_handle}) "
        if self.config_dict.get('gen_ai_support', None):
            summarized_blog_post = self.summarize_text(entry)
            if summarized_blog_post:
                basis_text.text('\n\n📖 ')
                basis_text.text(summarized_blog_post)
        basis_text += f"\n\n🔗 {entry.get('link', '')}\n\n{tags}"

        self.logger.info('*****************************')
        self.logger.info(basis_text)
        self.logger.info('*****************************')

        return basis_text

    @staticmethod
    def generate_text_to_summarize(entry):
        """
        Generate text to summarize.
        """
        text = (
            f"Title: {entry.get('title', '')}\n"
            f"Summary: {entry.get('summary', '')}"
        )
        if len(text.split()) > 700:
            words = text.split()[:700]
            return ' '.join(words)
        return text

    @staticmethod
    def clean_response(response):
        """
        Clean response.
        """
        return ' '.join(response.text.replace('\n', ' ').split())

    def summarize_text(self, entry):
        """
        Summarize text using LLMs.
        """
        text = self.generate_text_to_summarize(entry)
        model = genai.GenerativeModel(
            self.config_dict.get('gemini_model_name', '')
        )
        prompt_parts = [
            'Summarize the content of the post in maximum 60 characters.',
            'Be as concise as possible and be engaging.',
            'Don\'t repeat the title.',
            text
        ]
        response = model.generate_content(prompt_parts)
        response_cleaned = self.clean_response(response)
        safety_ratings = response.candidates[0].safety_ratings
        if all(
            rating.probability.name == 'NEGLIGIBLE'
            for rating in safety_ratings
        ):
            return response_cleaned
        return ''

    @staticmethod
    def check_platform_handle(platform_user_handle):
        """
        Check platform handle.
        """
        if (len(platform_user_handle) > 1
                and not platform_user_handle.startswith('@')):
            return f"@{platform_user_handle}"
        return platform_user_handle

    def build_post_bluesky(
        self,
        basis_text,
        platform_user_handle,
        tags,
        entry
    ):
        """
        Build post for Bluesky.
        """
        text_builder = client_utils.TextBuilder()
        text_builder.text(basis_text)

        platform_user_handle = self.check_platform_handle(platform_user_handle)

        if platform_user_handle:
            did = self.get_bluesky_did(platform_user_handle)
            text_builder.mention(f" ({platform_user_handle})", did)
        if self.config_dict.get('gen_ai_support', None):
            summarized_blog_post = self.summarize_text(entry)
            if summarized_blog_post:
                text_builder.text('\n\n📖 ')
                text_builder.text(summarized_blog_post)
        text_builder.text('\n\n🔗 ')
        link = entry.get('link', '')
        text_builder.link(link, link)
        text_builder.text('\n\n')
        for tag in tags.split('#'):
            tag_clean = tag.strip()
            if tag_clean:
                text_builder.tag(f"#{tag_clean} ", tag_clean)
        return text_builder

    def build_post(self, entry, feed):
        """Take the entry dict and build a post"""

        tags = self.define_tags(entry)
        platform = self.config_dict.get('platform', '')
        platform_user_handle = feed.get(platform)

        title = entry.get('title', '')
        name = feed.get('name', '')

        basis_text = ""

        if title:
            basis_text += f"📝 '{title}'\n\n"

        if name:
            basis_text += f"👤 {name}"

        if self.config_dict.get('platform', '') == 'mastodon':
            return self.build_post_mastodon(
                basis_text,
                platform_user_handle,
                tags,
                entry
            )
        elif self.config_dict.get('platform', '') == 'bluesky':
            return self.build_post_bluesky(
                basis_text,
                platform_user_handle,
                tags,
                entry
            )

    def send_post_to_mastodon(self, en, client, post_txt):
        """
        Send post to Mastodon.
        """
        media_content = en.get('media_content', None)
        alt_text = en.get('alt_text', None)

        if media_content:
            try:
                self.logger.info('Uploading media to mastodon')
                filename = self.download_image(media_content)
                media_upload_mastodon = client.media_post(filename)

                if alt_text:
                    self.logger.info('Adding description')
                    client.media_update(media_upload_mastodon,
                                        description=alt_text)

                self.logger.info('Now ready to post... ⏳')
                client.status_post(post_txt, media_ids=media_upload_mastodon)

                self.logger.info('Posted 🎉')
                return 'success'
            except Exception as e:
                self.logger.exception(
                    'Urg, media could not be printed for %s. Exception: %s',
                    en.get('link', 'unknown link'),
                    e)
                client.status_post(post_txt)
                self.logger.info('Posted post without image.')
                return 'failed'
        else:
            try:
                client.status_post(post_txt)
                self.logger.info('Posted 🎉')
                return 'success'
            except Exception as e:
                self.logger.exception(
                    'Urg, exception %s for %s',
                    e,
                    en.get('link', 'unknown link')
                )
                return 'failed'

    def send_post_to_bluesky(self, en, client, post_txt, embed_external):
        """
        Send post to Bluesky.
        """
        try:
            if embed_external:
                client.send_post(text=post_txt, embed=embed_external)
            else:
                client.send_post(text=post_txt)
            self.logger.info("Posted 🎉")
            return 'success'
        except Exception as e:
            self.logger.exception("Urg, exception %s for %s", e, en['link'])
            return 'failed'

    def build_embed_external(self, en, client):
        """
        Build embed external. This is a speciality of Bluesky's protocol.
        """
        if en['media_content']:
            filename = self.download_image(en['media_content'])
            with open(filename, 'rb') as f:
                img_data = f.read()

            thumb = client.upload_blob(img_data)

            return models.AppBskyEmbedExternal.Main(
                external=models.AppBskyEmbedExternal.External(
                    title=en['title'],
                    description=en['title'],
                    uri=en['link'],
                    thumb=thumb.blob,
                )
            )
        return None

    def send_post(self, en, feed, client):
        """Turn the dict into post text and send the post"""
        result = None
        self.logger.info(
            "Preparing the post on %s "
            "(%s) ...",
            self.config_dict['client_name'],
            {self.config_dict['platform']}
        )

        post_txt = self.build_post(
            en,
            feed
        )
        if self.config_dict["platform"] == "mastodon":
            result = self.send_post_to_mastodon(
                en,
                client,
                post_txt
            )
        elif self.config_dict["platform"] == "bluesky":
            embed_external = self.build_embed_external(
                en,
                client
            )
            result = self.send_post_to_bluesky(
                en,
                client,
                post_txt,
                embed_external
            )
        return result

    @staticmethod
    def load_feed(feed_path, d):
        """Method to load RSS feed"""
        full_fpd = feedparser.parse(feed_path)
        return d + full_fpd.entries

    @staticmethod
    def get_rss_feed_archive(feed):
        """Method to get RSS feed archive content"""
        archive_path = Path(feed['ARCHIVE'][0])
        archive_file = archive_path / 'file.json'

        if archive_path.exists():
            try:
                with archive_file.open('rb') as fp:
                    rss_feed_archive = json.load(fp)
            except (FileNotFoundError, json.JSONDecodeError):
                rss_feed_archive = {'link': []}
        else:
            if any(
                domain in feed['ARCHIVE'][0]
                for domain in ["www.youtube.com", "medium.com"]
            ):
                archive_path = archive_path / \
                    feed['name'].lower().replace(' ', '-')

            archive_path.mkdir(parents=True, exist_ok=True)
            rss_feed_archive = {'link': []}

        return rss_feed_archive

    @staticmethod
    def get_number_of_archive_entries(d, rss_feed_archive):
        """
        Calculate the number of entries in the feed and archive,
        ensuring archive structure is correct.
        """
        number_of_entries_feed = len(d)

        if 'link' in rss_feed_archive and isinstance(
            rss_feed_archive['link'],
            list
        ):
            number_of_entries_archive = len(set(rss_feed_archive['link']))
        else:
            # Fix the archive structure if 'link' key is missing or incorrect
            rss_feed_archive = {'link': list(set(rss_feed_archive))}
            number_of_entries_archive = len(rss_feed_archive['link'])

        return (
            rss_feed_archive,
            number_of_entries_archive,
            number_of_entries_feed,
        )

    @staticmethod
    def adjust_archive_path(base_path, domain, counter_name):
        """
        Helper function to clean up path construction for
        YouTube and Medium
        """
        feed_name_slug = counter_name.lower().replace(' ', '-')
        if "www.youtube.com" in domain or "medium.com" in domain:
            return base_path / feed_name_slug / feed_name_slug
        return base_path

    def get_folder_path(self, feed):
        """Method to identify folder path"""

        rss_feeds = feed.get('rss_feed', [])
        archive_paths = []
        archive = f"archive/{self.config_dict.get('archive', '')}"

        if len(rss_feeds) > 1:
            for rss_feed in rss_feeds:
                domain = urlsplit(rss_feed).netloc
                folder_path = Path(archive) / domain
                archive_paths.append(str(folder_path))

        elif len(rss_feeds) == 1:
            domain = urlsplit(rss_feeds[0]).netloc
            folder_path = Path(archive) / domain
            folder_path = self.adjust_archive_path(
                folder_path,
                domain,
                feed['name']
            )
            archive_paths.append(str(folder_path))

        feed['ARCHIVE'] = archive_paths
        return feed

    def process_feed(self, feed, count_post, client):
        """
        Process the RSS feed and generate a post for any entry
        we haven't yet seen.
        """
        name = feed.get('name', 'unknown name')
        rss_feed = feed.get('rss_feed', 'unknown feed')
        self.logger.info("=========================================")
        self.logger.info(
            'Begin processing of feeds from %s (%s)',
            name,
            rss_feed
        )

        feed = self.get_folder_path(feed)

        d = []

        for feed_path in rss_feed:
            # if "medium.com" in feed_path:
            #     parsed_url = urlparse(feed_path)
            #     subdomain = parsed_url.hostname.split('.')[0]
            #     feed_path = f"https://medium.com/feed/@{subdomain}"
            # # Load the feed
            try:
                d = self.load_feed(feed_path, d)
                rss_feed_archive = self.get_rss_feed_archive(feed)
                # Identify number of entries
                (
                    rss_feed_archive,
                    number_of_entries_archive,
                    number_of_entries_feed
                ) = self.get_number_of_archive_entries(d, rss_feed_archive)
                # If there are more entries, go through the list:

                feed_config = {
                    'rss_feed_archive': rss_feed_archive,
                    'number_of_entries_feed': number_of_entries_feed,
                    'feed': feed,
                    'd': d
                }

                if number_of_entries_feed > number_of_entries_archive:
                    count_post = self._process_feed(
                        client,
                        count_post,
                        feed_config
                    )
                    self.logger.info(
                        'New RSS feeds are successfully loaded and '
                        'processed.'
                    )
                    return count_post
                self.logger.info('Maximum number of posts is already posted.')
                return count_post
            except Exception as e:
                self.logger.info(
                    '🚨 Feed for %s not available because %s',
                    feed_path,
                    e
                )
                return count_post

    def _save_rss_feed_archive(self, feed, rss_feed_archive):
        """ Save RSS feed archive to a file """
        archive_path = os.path.join(feed['ARCHIVE'][0], 'file.json')
        with open(archive_path, 'wb') as fp:
            json.dump(rss_feed_archive, fp)
        self.logger.info("Archive for %s updated successfully.", feed['name'])

    @staticmethod
    def _get_media_content(entry):
        """ Extract media content from an RSS entry """
        en = {}
        if 'www.youtube.com' in entry.link:
            en['media_content'] = (
                f"http://img.youtube.com/vi/"
                f"{entry.id.replace('yt:video:', '')}/hqdefault.jpg"
            )
        elif 'media_content' in entry:
            en['media_content'] = entry.media_content[0]['url']
        else:
            soup = BeautifulSoup(entry.summary, "html.parser")
            img_url = [
                img['src']
                for img in soup.find_all('img')
                if img.has_attr('src')
            ]
            alt_text = [
                img['alt']
                for img in soup.find_all('img')
                if img.has_attr('alt')
            ]
            if img_url:
                en['media_content'] = img_url[0]
            if alt_text:
                en['alt_text'] = alt_text[0] if alt_text else ''
        return en

    def _process_feed(
        self,
        client,
        count_post,
        feed_config
    ):
        """ Process RSS feed entries and send posts """
        count = 0
        count_fails = 0
        result = None
        for _, entry in enumerate(feed_config['d']):
            if count >= 1:  # Limit to 1 post per run
                break
            elif count_fails >= 1:
                break

            en = {
                'title': entry.title,
                'link': entry.link,
                'pub_date': entry.published,
                'tags': [tag['term'] for tag in getattr(entry, 'tags', [])],
                'media_content': [],
                'summary': entry.summary
            }

            if not en['tags'] and 'category' in entry:
                en['tags'].append(entry.category)

            if self.process_images:
                en.update(self._get_media_content(entry))

            if en['link'] not in feed_config['rss_feed_archive']['link']:
                feed_config['rss_feed_archive']['link'].append(en['link'])
                if self.no_dry_run:
                    result = self.send_post(en, feed_config['feed'], client)
                if result == 'success':
                    count_post += 1
                    count += 1
                    time.sleep(1)
                elif result == 'failed':
                    count_fails += 1
                    time.sleep(1)

        if self.no_dry_run:
            if result == 'success':
                self._save_rss_feed_archive(
                    feed_config['feed'],
                    feed_config['rss_feed_archive']
                )

        return count_post

adjust_archive_path(base_path, domain, counter_name) staticmethod

Helper function to clean up path construction for YouTube and Medium

Source code in src/promote_blog_post.py

@staticmethod
def adjust_archive_path(base_path, domain, counter_name):
    """
    Helper function to clean up path construction for
    YouTube and Medium
    """
    feed_name_slug = counter_name.lower().replace(' ', '-')
    if "www.youtube.com" in domain or "medium.com" in domain:
        return base_path / feed_name_slug / feed_name_slug
    return base_path

build_embed_external(en, client)

Build embed external. This is a speciality of Bluesky's protocol.

Source code in src/promote_blog_post.py

def build_embed_external(self, en, client):
    """
    Build embed external. This is a speciality of Bluesky's protocol.
    """
    if en['media_content']:
        filename = self.download_image(en['media_content'])
        with open(filename, 'rb') as f:
            img_data = f.read()

        thumb = client.upload_blob(img_data)

        return models.AppBskyEmbedExternal.Main(
            external=models.AppBskyEmbedExternal.External(
                title=en['title'],
                description=en['title'],
                uri=en['link'],
                thumb=thumb.blob,
            )
        )
    return None

build_post(entry, feed)

Take the entry dict and build a post

Source code in src/promote_blog_post.py

def build_post(self, entry, feed):
    """Take the entry dict and build a post"""

    tags = self.define_tags(entry)
    platform = self.config_dict.get('platform', '')
    platform_user_handle = feed.get(platform)

    title = entry.get('title', '')
    name = feed.get('name', '')

    basis_text = ""

    if title:
        basis_text += f"📝 '{title}'\n\n"

    if name:
        basis_text += f"👤 {name}"

    if self.config_dict.get('platform', '') == 'mastodon':
        return self.build_post_mastodon(
            basis_text,
            platform_user_handle,
            tags,
            entry
        )
    elif self.config_dict.get('platform', '') == 'bluesky':
        return self.build_post_bluesky(
            basis_text,
            platform_user_handle,
            tags,
            entry
        )

build_post_bluesky(basis_text, platform_user_handle, tags, entry)

Build post for Bluesky.

Source code in src/promote_blog_post.py

def build_post_bluesky(
    self,
    basis_text,
    platform_user_handle,
    tags,
    entry
):
    """
    Build post for Bluesky.
    """
    text_builder = client_utils.TextBuilder()
    text_builder.text(basis_text)

    platform_user_handle = self.check_platform_handle(platform_user_handle)

    if platform_user_handle:
        did = self.get_bluesky_did(platform_user_handle)
        text_builder.mention(f" ({platform_user_handle})", did)
    if self.config_dict.get('gen_ai_support', None):
        summarized_blog_post = self.summarize_text(entry)
        if summarized_blog_post:
            text_builder.text('\n\n📖 ')
            text_builder.text(summarized_blog_post)
    text_builder.text('\n\n🔗 ')
    link = entry.get('link', '')
    text_builder.link(link, link)
    text_builder.text('\n\n')
    for tag in tags.split('#'):
        tag_clean = tag.strip()
        if tag_clean:
            text_builder.tag(f"#{tag_clean} ", tag_clean)
    return text_builder

build_post_mastodon(basis_text, platform_user_handle, tags, entry)

Build Mastodon post.

Source code in src/promote_blog_post.py

def build_post_mastodon(
    self, basis_text, platform_user_handle, tags, entry
):
    """
    Build Mastodon post.
    """
    platform_user_handle = self.check_platform_handle(platform_user_handle)

    if platform_user_handle:
        basis_text += f" ({platform_user_handle}) "
    if self.config_dict.get('gen_ai_support', None):
        summarized_blog_post = self.summarize_text(entry)
        if summarized_blog_post:
            basis_text.text('\n\n📖 ')
            basis_text.text(summarized_blog_post)
    basis_text += f"\n\n🔗 {entry.get('link', '')}\n\n{tags}"

    self.logger.info('*****************************')
    self.logger.info(basis_text)
    self.logger.info('*****************************')

    return basis_text

check_platform_handle(platform_user_handle) staticmethod

Check platform handle.

Source code in src/promote_blog_post.py

@staticmethod
def check_platform_handle(platform_user_handle):
    """
    Check platform handle.
    """
    if (len(platform_user_handle) > 1
            and not platform_user_handle.startswith('@')):
        return f"@{platform_user_handle}"
    return platform_user_handle

clean_response(response) staticmethod

Clean response.

Source code in src/promote_blog_post.py

@staticmethod
def clean_response(response):
    """
    Clean response.
    """
    return ' '.join(response.text.replace('\n', ' ').split())

define_tags(entry)

Define tags that will be posted along the posts.

Source code in src/promote_blog_post.py

def define_tags(self, entry):
    """
    Define tags that will be posted along the posts.
    """
    if self.config_dict.get('client_name', '') == 'pyladies_bot':
        tags = '#pyladies #python '
    elif self.config_dict.get('client_name', '') == 'rladies_bot':
        tags = '#rladies #rstats '
    else:
        self.logger.info('Bot name not found')
        tags = ''

    pub_date = self.parse_pub_date(entry)

    age_of_post = datetime.now() - pub_date

    if age_of_post.days > 730:
        tags += '#oldiebutgoodie '

    if len(entry['tags']) > 0:
        for tag in entry['tags']:
            if tag.lower() in ['pyladies', 'python', 'rstats', 'rladies']:
                pass
            else:
                tags += (
                    f"#{tag.replace(' ', '').replace('-', '').lower()} "
                )

    return tags

download_image(url)

Downloads an image from the given URL and saves it locally, organizing files by domain name.

Source code in src/promote_blog_post.py

def download_image(self, url: str):
    """
    Downloads an image from the given URL and saves it locally,
    organizing files by domain name.
    """
    try:
        filename = ''
        # Parse the URL components
        if self.config_dict["platform"] == "bluesky":
            domain = urlsplit(url).path
            filename = posixpath.basename(domain)
        elif self.config_dict["platform"] == "mastodon":
            domain = urlsplit(url).netloc
            filename = posixpath.basename(urlsplit(url).path)

        # Create folder structure based on the domain name
        domain_dir = Path(self.config_dict['images']) / domain
        domain_dir.mkdir(parents=True, exist_ok=True)

        # Full file path for the image
        file_path = domain_dir / filename

        if file_path.is_file():
            self.logger.info("Image already downloaded: %s", file_path)
            return str(file_path)

        # Set user-agent headers for the request
        headers = {
            'User-Agent': (
                'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:20.0) '
                'Gecko/20100101 Firefox/20.0'
            )
        }

        # Download the image
        self.logger.info("Downloading image from %s...", url)
        response = requests.get(
            url,
            headers=headers,
            stream=True,
            timeout=15
        )
        response.raise_for_status()  # Raises an exception for HTTP errors

        # Save the image to the designated path
        with open(file_path, 'wb') as out_file:
            shutil.copyfileobj(response.raw, out_file)

        self.logger.info("Image successfully downloaded: %s", file_path)
        return str(file_path)

    except requests.exceptions.RequestException as e:
        self.logger.error("Failed to download image from %s: %e", url, e)
        return None
    except OSError as e:
        self.logger.error("File system error while saving image: %s", e)
        return None
    finally:
        if 'response' in locals():
            response.close()

generate_text_to_summarize(entry) staticmethod

Generate text to summarize.

Source code in src/promote_blog_post.py

@staticmethod
def generate_text_to_summarize(entry):
    """
    Generate text to summarize.
    """
    text = (
        f"Title: {entry.get('title', '')}\n"
        f"Summary: {entry.get('summary', '')}"
    )
    if len(text.split()) > 700:
        words = text.split()[:700]
        return ' '.join(words)
    return text

get_bluesky_did(platform_user_handle)

Method to get Bluesky DID to uniquely identify (and tag) user.

Source code in src/promote_blog_post.py

def get_bluesky_did(self, platform_user_handle):
    """
    Method to get Bluesky DID to uniquely identify (and tag) user.
    """
    url = (
        f"https://bsky.social/xrpc/com.atproto.identity.resolveHandle?"
        f"handle={platform_user_handle.lstrip('@')}"
    )
    try:
        response = requests.get(url)

        if response.status_code == 200:
            data = response.json()
            did = data.get('did', None)

            if did:
                return did
            else:
                self.logger.info(
                    'The "did" field was not found in the response.'
                )
        else:
            self.logger.info(
                'Failed to retrieve data. Status code: %s',
                response.status_code
            )

    except requests.RequestException as e:
        self.logger.info('An error occurred: %s', e)

get_config()

Get config file

Source code in src/promote_blog_post.py

def get_config(self):
    """
    Get config file
    """
    if (self.config_dict is None) and (self.no_dry_run):
        self.config_dict = {
            "platform": os.getenv("PLATFORM"),
            "archive": os.getenv("ARCHIVE_DIRECTORY"),
            "images": os.getenv("IMAGES"),
            "counter": self._ensure_metadata_prefix(
                os.getenv("COUNTER", "")
            ),
            "password": os.getenv("PASSWORD"),
            "username": os.getenv("USERNAME"),
            "client_name": os.getenv("CLIENT_NAME"),
            "json_file": self._ensure_metadata_prefix(
                os.getenv("JSON_FILE", "")
            ),
            "gen_ai_support": True,
            "gemini_api_key": os.getenv("GEMINI_API_KEY"),
            "gemini_model_name": "gemini-2.5-flash"
        }
        if self.config_dict["platform"] == "mastodon":
            self.config_dict["api_base_url"] = config.API_BASE_URL
            self.config_dict["mastodon_visibility"] = (
                config.MASTODON_VISIBILITY
            )
            self.config_dict["client_id"] = os.getenv("CLIENT_ID")
            self.config_dict["client_secret"] = os.getenv("CLIENT_SECRET")
            self.config_dict["access_token"] = os.getenv("ACCESS_TOKEN")
            self.config_dict["client_cred_file"] = os.getenv(
                'BOT_CLIENTCRED_SECRET'
            )
        else:
            self.config_dict["api_base_url"] = "bluesky"

        if self.config_dict["gen_ai_support"]:
            genai.configure(api_key=self.config_dict["gemini_api_key"])
    else:
        self.config_dict['json_file'] = self._ensure_metadata_prefix(
            self.config_dict.get('json_file')
        )
        self.config_dict['counter'] = self._ensure_metadata_prefix(
            self.config_dict.get('counter')
        )

get_folder_path(feed)

Method to identify folder path

Source code in src/promote_blog_post.py

def get_folder_path(self, feed):
    """Method to identify folder path"""

    rss_feeds = feed.get('rss_feed', [])
    archive_paths = []
    archive = f"archive/{self.config_dict.get('archive', '')}"

    if len(rss_feeds) > 1:
        for rss_feed in rss_feeds:
            domain = urlsplit(rss_feed).netloc
            folder_path = Path(archive) / domain
            archive_paths.append(str(folder_path))

    elif len(rss_feeds) == 1:
        domain = urlsplit(rss_feeds[0]).netloc
        folder_path = Path(archive) / domain
        folder_path = self.adjust_archive_path(
            folder_path,
            domain,
            feed['name']
        )
        archive_paths.append(str(folder_path))

    feed['ARCHIVE'] = archive_paths
    return feed

get_number_of_archive_entries(d, rss_feed_archive) staticmethod

Calculate the number of entries in the feed and archive, ensuring archive structure is correct.

Source code in src/promote_blog_post.py

@staticmethod
def get_number_of_archive_entries(d, rss_feed_archive):
    """
    Calculate the number of entries in the feed and archive,
    ensuring archive structure is correct.
    """
    number_of_entries_feed = len(d)

    if 'link' in rss_feed_archive and isinstance(
        rss_feed_archive['link'],
        list
    ):
        number_of_entries_archive = len(set(rss_feed_archive['link']))
    else:
        # Fix the archive structure if 'link' key is missing or incorrect
        rss_feed_archive = {'link': list(set(rss_feed_archive))}
        number_of_entries_archive = len(rss_feed_archive['link'])

    return (
        rss_feed_archive,
        number_of_entries_archive,
        number_of_entries_feed,
    )

get_rss_feed_archive(feed) staticmethod

Method to get RSS feed archive content

Source code in src/promote_blog_post.py

@staticmethod
def get_rss_feed_archive(feed):
    """Method to get RSS feed archive content"""
    archive_path = Path(feed['ARCHIVE'][0])
    archive_file = archive_path / 'file.json'

    if archive_path.exists():
        try:
            with archive_file.open('rb') as fp:
                rss_feed_archive = json.load(fp)
        except (FileNotFoundError, json.JSONDecodeError):
            rss_feed_archive = {'link': []}
    else:
        if any(
            domain in feed['ARCHIVE'][0]
            for domain in ["www.youtube.com", "medium.com"]
        ):
            archive_path = archive_path / \
                feed['name'].lower().replace(' ', '-')

        archive_path.mkdir(parents=True, exist_ok=True)
        rss_feed_archive = {'link': []}

    return rss_feed_archive

load_feed(feed_path, d) staticmethod

Method to load RSS feed

Source code in src/promote_blog_post.py

@staticmethod
def load_feed(feed_path, d):
    """Method to load RSS feed"""
    full_fpd = feedparser.parse(feed_path)
    return d + full_fpd.entries

parse_pub_date(entry)

Method to parse the publication date

Source code in src/promote_blog_post.py

def parse_pub_date(self, entry):
    """Method to parse the publication date"""
    date_formats = [
        "%a, %d %b %Y %H:%M:%S %z",  # Format 1
        "%a, %d %b %Y %H:%M:%S %Z",  # Format 2
        "%Y-%m-%d",                  # Format 3
        "%Y-%m-%dT%H:%M:%S.%f%Z"     # Format 4
    ]

    pub_date_str = entry.get('pub_date', '')

    for date_format in date_formats:
        try:
            pub_date = datetime.strptime(
                pub_date_str, date_format).replace(tzinfo=None)
            return pub_date  # Return as soon as a valid format is found
        except ValueError:
            self.logger.info(
                "Failed to parse date with format: %s",
                date_format
            )

    # If none of the formats match, use the current date as a fallback
    self.logger.warning(
        "No matching date format found. Using current date."
    )
    return datetime.now()  # Fallback value

process_feed(feed, count_post, client)

Process the RSS feed and generate a post for any entry we haven't yet seen.

Source code in src/promote_blog_post.py

def process_feed(self, feed, count_post, client):
    """
    Process the RSS feed and generate a post for any entry
    we haven't yet seen.
    """
    name = feed.get('name', 'unknown name')
    rss_feed = feed.get('rss_feed', 'unknown feed')
    self.logger.info("=========================================")
    self.logger.info(
        'Begin processing of feeds from %s (%s)',
        name,
        rss_feed
    )

    feed = self.get_folder_path(feed)

    d = []

    for feed_path in rss_feed:
        # if "medium.com" in feed_path:
        #     parsed_url = urlparse(feed_path)
        #     subdomain = parsed_url.hostname.split('.')[0]
        #     feed_path = f"https://medium.com/feed/@{subdomain}"
        # # Load the feed
        try:
            d = self.load_feed(feed_path, d)
            rss_feed_archive = self.get_rss_feed_archive(feed)
            # Identify number of entries
            (
                rss_feed_archive,
                number_of_entries_archive,
                number_of_entries_feed
            ) = self.get_number_of_archive_entries(d, rss_feed_archive)
            # If there are more entries, go through the list:

            feed_config = {
                'rss_feed_archive': rss_feed_archive,
                'number_of_entries_feed': number_of_entries_feed,
                'feed': feed,
                'd': d
            }

            if number_of_entries_feed > number_of_entries_archive:
                count_post = self._process_feed(
                    client,
                    count_post,
                    feed_config
                )
                self.logger.info(
                    'New RSS feeds are successfully loaded and '
                    'processed.'
                )
                return count_post
            self.logger.info('Maximum number of posts is already posted.')
            return count_post
        except Exception as e:
            self.logger.info(
                '🚨 Feed for %s not available because %s',
                feed_path,
                e
            )
            return count_post

process_feeds(feeds, counter_name, count_post, client)

Method to handle processing of all feeds.

Source code in src/promote_blog_post.py

def process_feeds(self, feeds, counter_name, count_post, client):
    """
    Method to handle processing of all feeds.
    """
    for feed in feeds:
        if counter_name not in (feed['name'], '\n', ''):
            continue
        if len(feed['rss_feed']) == 0 or feed['rss_feed'] == [None]:
            continue

        is_last_feed = feed['name'] == feeds[-1]['name']

        if count_post == 0 and is_last_feed:
            count_post = self.process_feed(
                feed,
                count_post,
                client
            )

            # Add the counter_name
            if is_last_feed:
                new_feed = feeds[0]
                count_post = self.process_feed(
                    new_feed,
                    count_post,
                    client
                )

                self.logger.info(
                    "Successfully promoted blog posts. "
                    "Thank you and see you next time!")
                self.update_counter(feeds[1]['name'])
                break

        elif count_post < 2:
            count_post = self.process_feed(
                feed,
                count_post,
                client
            )
            counter_name = ''
            if is_last_feed:
                self.update_counter(feed['name'])
            self.logger.info(
                "=========================================")

        else:
            self.logger.info(
                "Successfully promoted blog posts. "
                "Thank you and see you next time!")
            self.update_counter(feed['name'])
            break

promote_blog_post()

Core method to promote blog post

Source code in src/promote_blog_post.py

def promote_blog_post(self):
    """Core method to promote blog post"""

    self.get_config()

    if self.no_dry_run:
        client_name = self.config_dict.get('client_name', 'unknown')
        self.logger.info("")
        self.logger.info(
            'Initializing %s Bot',
            client_name
        )
        separator = "%s", "=" * (len(client_name) + 17)
        self.logger.info(separator)
        self.logger.info(
            " > Connecting to %s",
            self.config_dict.get('api_base_url', '')
        )

        if self.config_dict["platform"] == "mastodon":
            _, client = login_mastodon(self.config_dict)
        elif self.config_dict["platform"] == "bluesky":
            client = login_bluesky(self.config_dict)
    else:
        client = None

    feeds = self.read_metadata_json()
    counter_name = self.read_counter_name()

    # Initiate count to post a maximum of 2 posts per run
    count_post = 0

    # Drop empty rss_feeds
    feeds = [x for x in feeds if x['rss_feed'] != '']

    if self.no_dry_run:
        self.process_feeds(feeds, counter_name, count_post, client)
    else:
        for feed in feeds:
            count_post = self.process_feed(
                feed,
                count_post,
                client
            )

read_counter_name()

Read counter name from txt file

Source code in src/promote_blog_post.py

def read_counter_name(self):
    """
    Read counter name from txt file
    """
    with open(self.config_dict["counter"], 'r', encoding='utf-8') as f:
        return f.read()

read_metadata_json()

Read metadata JSON file

Source code in src/promote_blog_post.py

def read_metadata_json(self):
    """
    Read metadata JSON file
    """
    with open(self.config_dict["json_file"], 'rb') as fp:
        self.logger.info(
            "============================================="
        )
        feeds = json.load(fp)
        self.logger.info('Meta data was successfully loaded')
        self.logger.info(
            "============================================="
        )
        return feeds

send_post(en, feed, client)

Turn the dict into post text and send the post

Source code in src/promote_blog_post.py

def send_post(self, en, feed, client):
    """Turn the dict into post text and send the post"""
    result = None
    self.logger.info(
        "Preparing the post on %s "
        "(%s) ...",
        self.config_dict['client_name'],
        {self.config_dict['platform']}
    )

    post_txt = self.build_post(
        en,
        feed
    )
    if self.config_dict["platform"] == "mastodon":
        result = self.send_post_to_mastodon(
            en,
            client,
            post_txt
        )
    elif self.config_dict["platform"] == "bluesky":
        embed_external = self.build_embed_external(
            en,
            client
        )
        result = self.send_post_to_bluesky(
            en,
            client,
            post_txt,
            embed_external
        )
    return result

send_post_to_bluesky(en, client, post_txt, embed_external)

Send post to Bluesky.

Source code in src/promote_blog_post.py

def send_post_to_bluesky(self, en, client, post_txt, embed_external):
    """
    Send post to Bluesky.
    """
    try:
        if embed_external:
            client.send_post(text=post_txt, embed=embed_external)
        else:
            client.send_post(text=post_txt)
        self.logger.info("Posted 🎉")
        return 'success'
    except Exception as e:
        self.logger.exception("Urg, exception %s for %s", e, en['link'])
        return 'failed'

send_post_to_mastodon(en, client, post_txt)

Send post to Mastodon.

Source code in src/promote_blog_post.py

def send_post_to_mastodon(self, en, client, post_txt):
    """
    Send post to Mastodon.
    """
    media_content = en.get('media_content', None)
    alt_text = en.get('alt_text', None)

    if media_content:
        try:
            self.logger.info('Uploading media to mastodon')
            filename = self.download_image(media_content)
            media_upload_mastodon = client.media_post(filename)

            if alt_text:
                self.logger.info('Adding description')
                client.media_update(media_upload_mastodon,
                                    description=alt_text)

            self.logger.info('Now ready to post... ⏳')
            client.status_post(post_txt, media_ids=media_upload_mastodon)

            self.logger.info('Posted 🎉')
            return 'success'
        except Exception as e:
            self.logger.exception(
                'Urg, media could not be printed for %s. Exception: %s',
                en.get('link', 'unknown link'),
                e)
            client.status_post(post_txt)
            self.logger.info('Posted post without image.')
            return 'failed'
    else:
        try:
            client.status_post(post_txt)
            self.logger.info('Posted 🎉')
            return 'success'
        except Exception as e:
            self.logger.exception(
                'Urg, exception %s for %s',
                e,
                en.get('link', 'unknown link')
            )
            return 'failed'

summarize_text(entry)

Summarize text using LLMs.

Source code in src/promote_blog_post.py

def summarize_text(self, entry):
    """
    Summarize text using LLMs.
    """
    text = self.generate_text_to_summarize(entry)
    model = genai.GenerativeModel(
        self.config_dict.get('gemini_model_name', '')
    )
    prompt_parts = [
        'Summarize the content of the post in maximum 60 characters.',
        'Be as concise as possible and be engaging.',
        'Don\'t repeat the title.',
        text
    ]
    response = model.generate_content(prompt_parts)
    response_cleaned = self.clean_response(response)
    safety_ratings = response.candidates[0].safety_ratings
    if all(
        rating.probability.name == 'NEGLIGIBLE'
        for rating in safety_ratings
    ):
        return response_cleaned
    return ''

update_counter(counter_name)

Update counter name

Source code in src/promote_blog_post.py

def update_counter(self, counter_name):
    """
    Update counter name
    """
    with open(
        self.config_dict["counter"],
        'w',
        encoding='utf-8'
    ) as txt_file:
        txt_file.write(counter_name)