API Reference

This page documents the public interface of all bot modules, auto-generated from their docstrings. Each class, its constructor arguments, and public methods are listed below.

boost_mentions

Module to boost mentions that tag the community bots.

BoostMentions

Handle boosting mentions of the community bots across platforms.

This class connects to either Mastodon or Bluesky depending on configuration and boosts or reposts mentions accordingly.

Source code in src/boost_mentions.py

class BoostMentions:
    """
    Handle boosting mentions of the community bots across platforms.

    This class connects to either Mastodon or Bluesky depending on
    configuration and boosts or reposts mentions accordingly.
    """

    def __init__(
        self,
        config_dict: Optional[Dict[str, Any]] = None,
        no_dry_run: bool = True
    ) -> None:
        """
        Initialize the BoostMentions handler.

        Args:
            config_dict: Optional configuration dictionary.
            no_dry_run: If True, perform actual boosts; if False, dry run.
        """
        self.logger = logging.getLogger(__name__)

        self.no_dry_run = no_dry_run
        self.config_dict = config_dict

    @property
    def cfg(self) -> Dict[str, Any]:
        """Property to ensure that the dictionary is initialized."""
        if self.config_dict is None:
            raise RuntimeError(
                "config_dict is not set; call boost_mentions() or pass "
                "config_dict to the constructor before accessing cfg"
            )
        return self.config_dict

    def boost_mentions(self) -> None:
        """
        Boost mentions on the configured social media platform.

        Connects to the appropriate API (Mastodon or Bluesky) and processes
        notifications to identify mentions. When found, the mention is
        boosted, favorited, or reposted depending on platform.
        """
        self.set_up_config_dict()

        self.logger.info("==========================")
        client_name = self.cfg.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.cfg["api_base_url"]
        )

        if self.cfg["platform"] == "mastodon":
            self._boost_mentions_mastodon()
        elif self.cfg["platform"] == "bluesky":
            self._boost_mentions_bluesky()

    def _boost_mentions_mastodon(self) -> None:
        """Fetch and boost Mastodon mentions."""
        account, client = login_mastodon(cast(MastodonConfig, 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")

        max_boosts = self.cfg.get("max_boosts_per_run", 5)
        boost_count = 0

        for notification in notifications:
            if boost_count >= max_boosts:
                self.logger.info(
                    " > Reached max boosts per run (%d), stopping.", max_boosts
                )
                break
            if (
                not notification.status.reblogged
                and notification.status.account.acct != account.acct
            ):
                if not self.no_dry_run:
                    self.logger.info(
                        "   * [DRY RUN] Would boost toot by %s viewable at: %s",
                        notification.account.username,
                        notification.status.url,
                    )
                    boost_count += 1
                else:
                    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)
                        boost_count += 1
                    except Exception as e:
                        self.logger.error(
                            "   * Boosting new toot by %s did not work: %s",
                            notification.account.username,
                            e,
                        )

    def _boost_mentions_bluesky(self) -> None:
        """Fetch and repost Bluesky mentions."""
        client = login_bluesky(cast(BlueskyConfig, 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()
        max_boosts = self.cfg.get("max_boosts_per_run", 5)
        boost_count = 0

        for notification in response.notifications:
            if boost_count >= max_boosts:
                self.logger.info(
                    " > Reached max boosts per run (%d), stopping.", max_boosts
                )
                break
            if notification.reason == "mention" and not notification.is_read:
                if not self.no_dry_run:
                    self.logger.info(
                        "   * [DRY RUN] Would repost URI %s CID %s",
                        notification.uri,
                        notification.cid,
                    )
                    boost_count += 1
                else:
                    try:
                        self.logger.info(
                            "   * Reposted post reference: %s",
                            client.repost(
                                uri=notification.uri,
                                cid=notification.cid,
                            ),
                        )
                        boost_count += 1
                    except Exception as e:
                        self.logger.error(
                            "   * Reposting new post with URI %s and "
                            "CID %s did not work: %s",
                            notification.uri,
                            notification.cid,
                            e,
                        )

        if self.no_dry_run:
            client.app.bsky.notification.update_seen({"seen_at": last_seen_at})
        self.logger.info(
            "Successfully processed notifications. Last seen at: %s",
            last_seen_at,
        )

    def set_up_config_dict(self) -> None:
        """
        Populate the configuration dictionary with required parameters.

        Loads environment variables and values from `config` to prepare
        platform-specific settings.
        """
        if self.config_dict is None:
            self.config_dict = {}
        self.config_dict.setdefault("platform", os.getenv("PLATFORM"))
        self.config_dict.setdefault("password", os.getenv("PASSWORD"))
        self.config_dict.setdefault("username", os.getenv("USERNAME"))
        self.config_dict.setdefault("client_name", os.getenv("CLIENT_NAME"))
        self.config_dict.setdefault("max_boosts_per_run", 5)
        if self.config_dict["platform"] == "mastodon":
            self.config_dict.setdefault("mastodon_visibility", config.MASTODON_VISIBILITY)
            self.config_dict.setdefault("api_base_url", config.API_BASE_URL)
            self.config_dict.setdefault("access_token", os.getenv("ACCESS_TOKEN"))
            self.config_dict.setdefault(
                "client_cred_file", os.getenv("BOT_CLIENTCRED_SECRET")
            )
            self.config_dict.setdefault("timeline_depth_limit", 40)
        elif self.config_dict["platform"] == "bluesky":
            self.config_dict.setdefault("api_base_url", "bluesky")
        else:
            raise ValueError(
                f"Unknown platform: {self.config_dict['platform']!r}. "
                "Expected 'mastodon' or 'bluesky'."
            )

cfg property

Property to ensure that the dictionary is initialized.

__init__(config_dict=None, no_dry_run=True)

Initialize the BoostMentions handler.

Parameters:

Name	Type	Description	Default
config_dict	Optional[Dict[str, Any]]	Optional configuration dictionary.	None
no_dry_run	bool	If True, perform actual boosts; if False, dry run.	True

Source code in src/boost_mentions.py

def __init__(
    self,
    config_dict: Optional[Dict[str, Any]] = None,
    no_dry_run: bool = True
) -> None:
    """
    Initialize the BoostMentions handler.

    Args:
        config_dict: Optional configuration dictionary.
        no_dry_run: If True, perform actual boosts; if False, dry run.
    """
    self.logger = logging.getLogger(__name__)

    self.no_dry_run = no_dry_run
    self.config_dict = config_dict

boost_mentions()

Boost mentions on the configured social media platform.

Connects to the appropriate API (Mastodon or Bluesky) and processes notifications to identify mentions. When found, the mention is boosted, favorited, or reposted depending on platform.

Source code in src/boost_mentions.py

def boost_mentions(self) -> None:
    """
    Boost mentions on the configured social media platform.

    Connects to the appropriate API (Mastodon or Bluesky) and processes
    notifications to identify mentions. When found, the mention is
    boosted, favorited, or reposted depending on platform.
    """
    self.set_up_config_dict()

    self.logger.info("==========================")
    client_name = self.cfg.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.cfg["api_base_url"]
    )

    if self.cfg["platform"] == "mastodon":
        self._boost_mentions_mastodon()
    elif self.cfg["platform"] == "bluesky":
        self._boost_mentions_bluesky()

set_up_config_dict()

Populate the configuration dictionary with required parameters.

Loads environment variables and values from config to prepare platform-specific settings.

Source code in src/boost_mentions.py

def set_up_config_dict(self) -> None:
    """
    Populate the configuration dictionary with required parameters.

    Loads environment variables and values from `config` to prepare
    platform-specific settings.
    """
    if self.config_dict is None:
        self.config_dict = {}
    self.config_dict.setdefault("platform", os.getenv("PLATFORM"))
    self.config_dict.setdefault("password", os.getenv("PASSWORD"))
    self.config_dict.setdefault("username", os.getenv("USERNAME"))
    self.config_dict.setdefault("client_name", os.getenv("CLIENT_NAME"))
    self.config_dict.setdefault("max_boosts_per_run", 5)
    if self.config_dict["platform"] == "mastodon":
        self.config_dict.setdefault("mastodon_visibility", config.MASTODON_VISIBILITY)
        self.config_dict.setdefault("api_base_url", config.API_BASE_URL)
        self.config_dict.setdefault("access_token", os.getenv("ACCESS_TOKEN"))
        self.config_dict.setdefault(
            "client_cred_file", os.getenv("BOT_CLIENTCRED_SECRET")
        )
        self.config_dict.setdefault("timeline_depth_limit", 40)
    elif self.config_dict["platform"] == "bluesky":
        self.config_dict.setdefault("api_base_url", "bluesky")
    else:
        raise ValueError(
            f"Unknown platform: {self.config_dict['platform']!r}. "
            "Expected 'mastodon' or '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 = 'pyladies'  # 'pyladies' or 'rladies'
        self.what_to_debug = 'blog'  # 'blog', 'boost_tags', 'rss', 'anniversary', 'boost_mentions'
        self.platform = 'bluesky'  # 'bluesky' or 'mastodon'
        self.no_dry_run = False  # True to actually post

    def start_debug(self):
        """Start debugging."""
        logger = logging.getLogger(__name__)

        if self.what_to_debug == 'blog':
            config_dict = self.get_config_blog()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r platform=%r — check bot/platform settings.",
                    self.bot, self.platform
                )
                return
            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()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r platform=%r — check bot/platform settings.",
                    self.bot, self.platform
                )
                return
            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()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r platform=%r — check bot/platform settings.",
                    self.bot, self.platform
                )
                return
            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()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r platform=%r — check bot/platform settings.",
                    self.bot, self.platform
                )
                return
            boost_mentions_handler = BoostMentions(
                config_dict,
                self.no_dry_run
            )
            boost_mentions_handler.boost_mentions()

        elif self.what_to_debug == 'anniversary':
            config_dict = self.get_config_anniversary()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r platform=%r — check bot/platform settings.",
                    self.bot, self.platform
                )
                return
            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_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,
                }
            if self.platform == 'mastodon':
                return {
                    "archive": "pyladies_archive_directory",
                    "counter": "metadata/pyladies_counter.txt",
                    "json_file": "metadata/pyladies_meta_data.json",
                    "client_name": "pyladies_bot",
                    "images": "pyladies_images",
                    "api_base_url": config.API_BASE_URL,
                    "mastodon": None,
                    "password": os.getenv("PYLADIES_MASTODON_PASSWORD"),
                    "username": os.getenv("PYLADIES_MASTODON_USERNAME"),
                    "access_token": os.getenv("PYLADIES_MASTODON_ACCESS_TOKEN"),
                    "client_cred_file": os.getenv("PYLADIES_BOT_CLIENTCRED_SECRET"),
                    "mastodon_visibility": config.MASTODON_VISIBILITY,
                    "platform": self.platform,
                }

        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_bot",
                    "images": "rladies_images",
                    "api_base_url": self.platform,
                    "mastodon": None,
                    "gen_ai_support": True,
                    "gemini_model_name": "gemini-2.5-flash",
                    "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("RLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                }
            if self.platform == 'mastodon':
                return {
                    "archive": "rladies_archive_directory",
                    "counter": "../metadata/rladies_counter.txt",
                    "json_file": "../metadata/rladies_meta_data.json",
                    "client_name": "rladies_bot",
                    "images": "rladies_images",
                    "api_base_url": config.API_BASE_URL,
                    "mastodon": None,
                    "password": os.getenv("RLADIES_MASTODON_PASSWORD"),
                    "username": os.getenv("RLADIES_MASTODON_USERNAME"),
                    "access_token": os.getenv("RLADIES_MASTODON_ACCESS_TOKEN"),
                    "client_cred_file": os.getenv("RLADIES_BOT_CLIENTCRED_SECRET"),
                    "mastodon_visibility": config.MASTODON_VISIBILITY,
                    "platform": self.platform,
                }

        return None

    def get_config_boost(self):
        """Method to generate config for boosting tags"""
        if self.bot == 'pyladies':
            if self.platform == 'bluesky':
                return {
                    "client_name": "pyladies_bot",
                    "api_base_url": self.platform,
                    "mastodon": None,
                    "password": os.getenv("PYLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("PYLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                    "tags": "pyladies",
                }
            if self.platform == 'mastodon':
                return {
                    "client_name": "pyladies_bot",
                    "api_base_url": config.API_BASE_URL,
                    "mastodon": None,
                    "password": os.getenv("PYLADIES_MASTODON_PASSWORD"),
                    "username": os.getenv("PYLADIES_MASTODON_USERNAME"),
                    "access_token": os.getenv("PYLADIES_MASTODON_ACCESS_TOKEN"),
                    "client_cred_file": os.getenv("PYLADIES_BOT_CLIENTCRED_SECRET"),
                    "platform": self.platform,
                    "mastodon_visibility": config.MASTODON_VISIBILITY,
                    "tags": "pyladies",
                }

        if self.bot == 'rladies':
            if self.platform == "bluesky":
                return {
                    "client_name": "rladies_bot",
                    "api_base_url": self.platform,
                    "mastodon": None,
                    "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("RLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                    "tags": "rladies",
                }
            if self.platform == 'mastodon':
                return {
                    "client_name": "rladies_bot",
                    "api_base_url": config.API_BASE_URL,
                    "mastodon": None,
                    "password": os.getenv("RLADIES_MASTODON_PASSWORD"),
                    "username": os.getenv("RLADIES_MASTODON_USERNAME"),
                    "access_token": os.getenv("RLADIES_MASTODON_ACCESS_TOKEN"),
                    "client_cred_file": os.getenv("RLADIES_BOT_CLIENTCRED_SECRET"),
                    "platform": self.platform,
                    "mastodon_visibility": config.MASTODON_VISIBILITY,
                    "tags": "rladies",
                }

        return None

    def get_config_rss(self):
        """Method to generate config for fetching RSS data"""
        if self.bot == 'pyladies':
            return {
                "json_file": "metadata/pyladies_meta_data.json",
                "api_base_url": (
                    "https://github.com/cosimameyer/"
                    "awesome-pyladies-blogs/tree/main/blogs"
                ),
                "github_raw_url": (
                    "https://raw.githubusercontent.com/cosimameyer/"
                    "awesome-pyladies-blogs/main/blogs"
                ),
            }
        if self.bot == 'rladies':
            return {
                "json_file": "../metadata/rladies_meta_data.json",
                "api_base_url": (
                    "https://github.com/rladies/"
                    "awesome-rladies-blogs/tree/main/blogs"
                ),
                "github_raw_url": (
                    "https://raw.githubusercontent.com/rladies/"
                    "awesome-rladies-blogs/main/blogs"
                ),
            }
        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_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,
                }
            if self.platform == 'mastodon':
                return {
                    'client_name': 'pyladies_bot',
                    'api_base_url': config.API_BASE_URL,
                    'mastodon': None,
                    'password': os.getenv('PYLADIES_MASTODON_PASSWORD'),
                    'username': os.getenv('PYLADIES_MASTODON_USERNAME'),
                    'access_token': os.getenv('PYLADIES_MASTODON_ACCESS_TOKEN'),
                    'client_cred_file': os.getenv('PYLADIES_BOT_CLIENTCRED_SECRET'),
                    'images': 'anniversary_images',
                    'platform': self.platform,
                    'mastodon_visibility': config.MASTODON_VISIBILITY,
                }

        if self.bot == 'rladies':
            if self.platform == 'bluesky':
                return {
                    'client_name': 'rladies_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,
                }
            if self.platform == 'mastodon':
                return {
                    'client_name': 'rladies_bot',
                    'api_base_url': config.API_BASE_URL,
                    'mastodon': None,
                    'password': os.getenv('RLADIES_MASTODON_PASSWORD'),
                    'username': os.getenv('RLADIES_MASTODON_USERNAME'),
                    'access_token': os.getenv('RLADIES_MASTODON_ACCESS_TOKEN'),
                    'client_cred_file': os.getenv('RLADIES_BOT_CLIENTCRED_SECRET'),
                    'images': 'anniversary_images',
                    'platform': self.platform,
                    'mastodon_visibility': config.MASTODON_VISIBILITY,
                }

        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_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,
            }
        if self.platform == 'mastodon':
            return {
                'client_name': 'pyladies_bot',
                'api_base_url': config.API_BASE_URL,
                'mastodon': None,
                'password': os.getenv('PYLADIES_MASTODON_PASSWORD'),
                'username': os.getenv('PYLADIES_MASTODON_USERNAME'),
                'access_token': os.getenv('PYLADIES_MASTODON_ACCESS_TOKEN'),
                'client_cred_file': os.getenv('PYLADIES_BOT_CLIENTCRED_SECRET'),
                'images': 'anniversary_images',
                'platform': self.platform,
                'mastodon_visibility': config.MASTODON_VISIBILITY,
            }

    if self.bot == 'rladies':
        if self.platform == 'bluesky':
            return {
                'client_name': 'rladies_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,
            }
        if self.platform == 'mastodon':
            return {
                'client_name': 'rladies_bot',
                'api_base_url': config.API_BASE_URL,
                'mastodon': None,
                'password': os.getenv('RLADIES_MASTODON_PASSWORD'),
                'username': os.getenv('RLADIES_MASTODON_USERNAME'),
                'access_token': os.getenv('RLADIES_MASTODON_ACCESS_TOKEN'),
                'client_cred_file': os.getenv('RLADIES_BOT_CLIENTCRED_SECRET'),
                'images': 'anniversary_images',
                'platform': self.platform,
                'mastodon_visibility': config.MASTODON_VISIBILITY,
            }

    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_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,
            }
        if self.platform == 'mastodon':
            return {
                "archive": "pyladies_archive_directory",
                "counter": "metadata/pyladies_counter.txt",
                "json_file": "metadata/pyladies_meta_data.json",
                "client_name": "pyladies_bot",
                "images": "pyladies_images",
                "api_base_url": config.API_BASE_URL,
                "mastodon": None,
                "password": os.getenv("PYLADIES_MASTODON_PASSWORD"),
                "username": os.getenv("PYLADIES_MASTODON_USERNAME"),
                "access_token": os.getenv("PYLADIES_MASTODON_ACCESS_TOKEN"),
                "client_cred_file": os.getenv("PYLADIES_BOT_CLIENTCRED_SECRET"),
                "mastodon_visibility": config.MASTODON_VISIBILITY,
                "platform": self.platform,
            }

    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_bot",
                "images": "rladies_images",
                "api_base_url": self.platform,
                "mastodon": None,
                "gen_ai_support": True,
                "gemini_model_name": "gemini-2.5-flash",
                "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                "username": os.getenv("RLADIES_BSKY_USERNAME"),
                "platform": self.platform,
            }
        if self.platform == 'mastodon':
            return {
                "archive": "rladies_archive_directory",
                "counter": "../metadata/rladies_counter.txt",
                "json_file": "../metadata/rladies_meta_data.json",
                "client_name": "rladies_bot",
                "images": "rladies_images",
                "api_base_url": config.API_BASE_URL,
                "mastodon": None,
                "password": os.getenv("RLADIES_MASTODON_PASSWORD"),
                "username": os.getenv("RLADIES_MASTODON_USERNAME"),
                "access_token": os.getenv("RLADIES_MASTODON_ACCESS_TOKEN"),
                "client_cred_file": os.getenv("RLADIES_BOT_CLIENTCRED_SECRET"),
                "mastodon_visibility": config.MASTODON_VISIBILITY,
                "platform": self.platform,
            }

    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':
        if self.platform == 'bluesky':
            return {
                "client_name": "pyladies_bot",
                "api_base_url": self.platform,
                "mastodon": None,
                "password": os.getenv("PYLADIES_BSKY_PASSWORD"),
                "username": os.getenv("PYLADIES_BSKY_USERNAME"),
                "platform": self.platform,
                "tags": "pyladies",
            }
        if self.platform == 'mastodon':
            return {
                "client_name": "pyladies_bot",
                "api_base_url": config.API_BASE_URL,
                "mastodon": None,
                "password": os.getenv("PYLADIES_MASTODON_PASSWORD"),
                "username": os.getenv("PYLADIES_MASTODON_USERNAME"),
                "access_token": os.getenv("PYLADIES_MASTODON_ACCESS_TOKEN"),
                "client_cred_file": os.getenv("PYLADIES_BOT_CLIENTCRED_SECRET"),
                "platform": self.platform,
                "mastodon_visibility": config.MASTODON_VISIBILITY,
                "tags": "pyladies",
            }

    if self.bot == 'rladies':
        if self.platform == "bluesky":
            return {
                "client_name": "rladies_bot",
                "api_base_url": self.platform,
                "mastodon": None,
                "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                "username": os.getenv("RLADIES_BSKY_USERNAME"),
                "platform": self.platform,
                "tags": "rladies",
            }
        if self.platform == 'mastodon':
            return {
                "client_name": "rladies_bot",
                "api_base_url": config.API_BASE_URL,
                "mastodon": None,
                "password": os.getenv("RLADIES_MASTODON_PASSWORD"),
                "username": os.getenv("RLADIES_MASTODON_USERNAME"),
                "access_token": os.getenv("RLADIES_MASTODON_ACCESS_TOKEN"),
                "client_cred_file": os.getenv("RLADIES_BOT_CLIENTCRED_SECRET"),
                "platform": self.platform,
                "mastodon_visibility": config.MASTODON_VISIBILITY,
                "tags": "rladies",
            }

    return None

get_config_rss()

Method to generate config for fetching RSS data

Source code in src/debug.py

def get_config_rss(self):
    """Method to generate config for fetching RSS data"""
    if self.bot == 'pyladies':
        return {
            "json_file": "metadata/pyladies_meta_data.json",
            "api_base_url": (
                "https://github.com/cosimameyer/"
                "awesome-pyladies-blogs/tree/main/blogs"
            ),
            "github_raw_url": (
                "https://raw.githubusercontent.com/cosimameyer/"
                "awesome-pyladies-blogs/main/blogs"
            ),
        }
    if self.bot == 'rladies':
        return {
            "json_file": "../metadata/rladies_meta_data.json",
            "api_base_url": (
                "https://github.com/rladies/"
                "awesome-rladies-blogs/tree/main/blogs"
            ),
            "github_raw_url": (
                "https://raw.githubusercontent.com/rladies/"
                "awesome-rladies-blogs/main/blogs"
            ),
        }
    return None

start_debug()

Start debugging.

Source code in src/debug.py

def start_debug(self):
    """Start debugging."""
    logger = logging.getLogger(__name__)

    if self.what_to_debug == 'blog':
        config_dict = self.get_config_blog()
        if config_dict is None:
            logger.error(
                "No config for bot=%r platform=%r — check bot/platform settings.",
                self.bot, self.platform
            )
            return
        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()
        if config_dict is None:
            logger.error(
                "No config for bot=%r platform=%r — check bot/platform settings.",
                self.bot, self.platform
            )
            return
        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()
        if config_dict is None:
            logger.error(
                "No config for bot=%r platform=%r — check bot/platform settings.",
                self.bot, self.platform
            )
            return
        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()
        if config_dict is None:
            logger.error(
                "No config for bot=%r platform=%r — check bot/platform settings.",
                self.bot, self.platform
            )
            return
        boost_mentions_handler = BoostMentions(
            config_dict,
            self.no_dry_run
        )
        boost_mentions_handler.boost_mentions()

    elif self.what_to_debug == 'anniversary':
        config_dict = self.get_config_anniversary()
        if config_dict is None:
            logger.error(
                "No config for bot=%r platform=%r — check bot/platform settings.",
                self.bot, self.platform
            )
            return
        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 that content length does not exceed 500 characters.

check_entries(data)

Check if combined length of name, description, and wiki_link exceeds 500 characters for any entry.

Parameters:

Name	Type	Description	Default
data	List[Dict[str, Any]]	List of entries to validate.	required

Raises:

Type	Description
SystemExit	If any entry exceeds 500 characters.

Source code in src/helper/check_length_anniversary.py

def check_entries(data: List[Dict[str, Any]]) -> None:
    """
    Check if combined length of `name`, `description`, and `wiki_link`
    exceeds 500 characters for any entry.

    Args:
        data: List of entries to validate.

    Raises:
        SystemExit: If any entry exceeds 500 characters.
    """
    if not data:
        logger.warning("Warning: No entries found to check.")
        return
    for entry in data:
        name = entry.get('name', '')
        combined_text = (
            f"Let's meet {name} ✨\n\n"
            f"{entry.get('description', '')}\n\n"
            f"🔗 {entry.get('wiki_link', '')}\n\n"
            "#amazingwomeninstem #womeninstem "
            "#womenalsoknow #impactthefuture"
        )

        if len(combined_text) > MAX_POST_LENGTH:
            logger.warning(
                "🚨 Alert: The combined text for '%s' exceeds %s characters!",
                name, MAX_POST_LENGTH
            )
            logger.info("Combined length: %s characters.", len(combined_text))
            logger.info(combined_text)
            logger.info(
                "Length of description: %s",
                len(entry.get('description', ''))
            )
            sys.exit(1)  # Exit with error code to indicate failure

load_json(filename)

Load JSON data from a file.

Parameters:

Name	Type	Description	Default
filename	str	Path to the JSON file.	required

Returns:

Type	Description
Optional[List[Dict[str, Any]]]	Parsed JSON data as a list of dictionaries,
Optional[List[Dict[str, Any]]]	or None if the file is missing or invalid.

Source code in src/helper/check_length_anniversary.py

def load_json(filename: str) -> Optional[List[Dict[str, Any]]]:
    """
    Load JSON data from a file.

    Args:
        filename: Path to the JSON file.

    Returns:
        Parsed JSON data as a list of dictionaries,
        or None if the file is missing or invalid.
    """
    try:
        with open(filename, "r", encoding="utf-8") as file:
            return json.load(file)
    except FileNotFoundError:
        logger.error("Error: The file '%s' was not found.", filename)
        return None
    except json.JSONDecodeError:
        logger.error(
            "Error: The file '%s' contains invalid JSON.",
            filename
        )
        return None

main()

Load the JSON file and validate content length for each entry.

Source code in src/helper/check_length_anniversary.py

def main() -> None:
    """
    Load the JSON file and validate content length for each entry.
    """
    filename = sys.argv[1] if len(sys.argv) > 1 else "events.json"
    data = load_json(filename)

    if data is None:
        sys.exit(1)

    check_entries(data)
    logger.info("All good! 🎉")

login_bluesky

Module to log into Bluesky.

BlueskyConfig

Bases: TypedDict

Typed configuration for Bluesky login.

Source code in src/helper/login_bluesky.py

class BlueskyConfig(TypedDict):
    """Typed configuration for Bluesky login."""
    username: str
    password: str

login_bluesky(config_dict)

Log in to Bluesky and return the client instance.

Parameters:

Name	Type	Description	Default
config_dict	BlueskyConfig	Configuration required for Bluesky login.	required

Returns:

Type	Description
Client	A logged-in Bluesky Client instance.

Source code in src/helper/login_bluesky.py

def login_bluesky(config_dict: BlueskyConfig) -> Client:
    """
    Log in to Bluesky and return the client instance.

    Args:
        config_dict: Configuration required for Bluesky login.

    Returns:
        A logged-in Bluesky `Client` instance.
    """
    logger.info(" > Logging in as %s", config_dict["username"])

    client = Client()
    profile = client.login(
        config_dict["username"],
        config_dict["password"],
    )

    logger.info(" > Successfully logged in as @%s", profile.handle)

    return client

login_mastodon

Module to log into Mastodon.

MastodonConfig

Bases: TypedDict

Typed configuration for Mastodon login.

Source code in src/helper/login_mastodon.py

class MastodonConfig(TypedDict):
    """Typed configuration for Mastodon login."""
    api_base_url: str
    access_token: str

login_mastodon(config_dict)

Log in to Mastodon and return the account and client.

Parameters:

Name	Type	Description	Default
config_dict	MastodonConfig	Configuration required for Mastodon login.	required

Returns:

Type	Description
tuple[dict, Mastodon]	A tuple containing: - account: The Mastodon account object. - client: The Mastodon client instance.

Source code in src/helper/login_mastodon.py

def login_mastodon(config_dict: MastodonConfig) -> tuple[dict, Mastodon]:
    """
    Log in to Mastodon and return the account and client.

    Args:
        config_dict: Configuration required for Mastodon login.

    Returns:
        A tuple containing:
            - account: The Mastodon account object.
            - client: The Mastodon client instance.
    """
    client = Mastodon(
        access_token=config_dict["access_token"],
        api_base_url=config_dict["api_base_url"],
    )

    logger.info(" > Logging in as access token holder on %s", config_dict["api_base_url"])

    account = client.me()

    logger.info(" > Successfully logged in as @%s", account["username"])

    return account, client

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: Optional[Dict[str, Any]] = None,
        no_dry_run: bool = True
    ) -> None:
        """
        Initialize a PromoteAnniversary handler.

        Args:
            config_dict: Optional configuration dictionary.
            no_dry_run: Whether to actually execute posting (True)
                or just simulate actions (False).
        """
        self.logger = logging.getLogger(__name__)
        self.config_dict = config_dict
        self.no_dry_run = no_dry_run
        self.base_path = (
            "https://raw.githubusercontent.com/cosimameyer/"
            "illustrations/main/amazing-women"
        )

    @property
    def cfg(self) -> Dict[str, Any]:
        """Property to ensure that the dictionary is initialized."""
        if self.config_dict is None:
            raise RuntimeError(
                "config_dict is not set; call promote_anniversary() or pass "
                "config_dict to the constructor before accessing cfg"
            )
        return self.config_dict

    def promote_anniversary(self) -> None:
        """Main entry point. Loads configuration, fetches events, and posts if applicable."""
        if self.config_dict is None and self.no_dry_run:
            self._setup_config_from_env()

        if self.config_dict is None and self.no_dry_run:
            self.logger.error("No config_dict provided — cannot run")
            return

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

        client = self._connect_client() if self.no_dry_run else None
        if client is None and self.no_dry_run:
            self.logger.error("Failed to connect to %s", self.cfg["platform"])
            return

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

        for event in events:
            if self.is_matching_current_date(event["date"]):
                if not self.no_dry_run:
                    self.logger.info(
                        "[DRY RUN] Would post anniversary for %s on %s",
                        event.get("name"),
                        event.get("date"),
                    )
                else:
                    self.send_post(event, client)

    def _setup_config_from_env(self) -> None:
        """Populate config_dict from environment variables (used in GitHub Actions)."""
        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"] = "https://bsky.social"

    def _connect_client(self):
        """Connect to the configured platform and return the client."""
        if self.cfg["platform"] == "mastodon":
            _, client = login_mastodon(cast(MastodonConfig, self.config_dict))
            return client
        if self.cfg["platform"] == "bluesky":
            return login_bluesky(cast(BlueskyConfig, self.config_dict))
        self.logger.error("Unsupported platform: %s", self.cfg["platform"])
        return None

    @staticmethod
    def is_matching_current_date(
        date_str: str, date_format: str = "%m-%d"
    ) -> bool:
        """
        Check whether the given date matches today's date.

        Args:
            date_str: Date string to compare (e.g., "08-30").
            date_format: Format of the provided date string. 
                         Defaults to "%m-%d".

        Returns:
            True if the date matches today's date, False otherwise.
        """
        current_date = datetime.now().strftime(date_format)
        return date_str == current_date

    def download_image(self, url: str) -> str:
        """
        Download an image from a URL if not already cached locally.

        Args:
            url: URL to the image.

        Returns:
            Path to the downloaded image file.
        """
        path = urlsplit(url).path
        filename = posixpath.basename(path)
        file_path = os.path.join(self.cfg["images"], filename)

        if not os.path.isfile(file_path):
            os.makedirs(self.cfg["images"], exist_ok=True)
            headers = {
                "User-Agent": (
                    "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:20.0) "
                    "Gecko/20100101 Firefox/20.0"
                )
            }
            with requests.get(
                url,
                headers=headers,
                stream=True,
                timeout=REQUEST_TIMEOUT
            ) as response:
                with open(file_path, "wb") as out_file:
                    shutil.copyfileobj(response.raw, out_file)
        else:
            self.logger.info("Image already downloaded: %s", file_path)

        return file_path

    def build_post(
        self,
        event: Dict[str, Any]
    ) -> Union[str, client_utils.TextBuilder]:
        """
        Build the post text for Mastodon or Bluesky.

        Args:
            event: Dictionary containing event data.

        Returns:
            A formatted post string (Mastodon) or TextBuilder object (Bluesky).
        """
        tags = "\n\n#amazingwomenintech #womenalsoknow #impactthefuture"

        if self.cfg["platform"] == "mastodon":
            return (
                f"Let's meet {event['name']} ✨\n\n"
                f"{event['description_mastodon']}\n\n"
                f"🔗 {event['wiki_link']}{tags}"
            )

        if self.cfg["platform"] == "bluesky":
            bluesky_max_graphemes = 300
            tag_list = [t.strip() for t in tags.split("#") if t.strip()]
            did = self.get_bluesky_did(event["bluesky"]) if event.get("bluesky") else None

            def _build(tag_subset, desc_override=None):
                desc = desc_override if desc_override is not None else event["description_bluesky"]
                tb = client_utils.TextBuilder()
                if event.get("bluesky"):
                    tb.text("Let's meet ")
                    tb.mention(event["bluesky"], did)
                    tb.text(" ⭐️\n\n")
                else:
                    tb.text(f"Let's meet {event['name']} ⭐️\n\n")
                split_text = [
                    item.rstrip(" ")
                    for item in re.split(r"(#\w+)", desc)
                    if item.strip()
                ]
                for text_chunk in split_text:
                    if text_chunk.startswith("#"):
                        for tag in text_chunk.split("#"):
                            if tag.strip():
                                tb.tag(f"#{tag.strip()}", tag.strip())
                    else:
                        tb.text(self.add_whitespace_if_needed(text_chunk))
                tb.text("\n\n🔗 ")
                tb.link(event["wiki_link"], event["wiki_link"])
                if tag_subset:
                    tb.text("\n\n")
                    for i, tag in enumerate(tag_subset):
                        display = f"#{tag}" if i == len(tag_subset) - 1 else f"#{tag} "
                        tb.tag(display, tag)
                return tb

            # Drop trailing tags one by one until within limit
            for count in range(len(tag_list), -1, -1):
                text_builder = _build(tag_list[:count])
                if len(text_builder.build_text()) <= bluesky_max_graphemes:
                    return text_builder

            # Still over limit: trim description to fit
            overhead = len(_build([], desc_override="").build_text())
            available = bluesky_max_graphemes - overhead - 1  # -1 for "…"
            desc_trimmed = event["description_bluesky"][:available].rstrip() + "…"
            return _build([], desc_override=desc_trimmed)

        raise ValueError(
            f"Unsupported platform: {self.cfg['platform']}"
        )

    def send_post(self, event: Dict[str, Any], client: Any) -> None:
        """Send a post to the configured platform (Mastodon or Bluesky)."""
        self.logger.info(
            "Preparing the post on %s (%s)...",
            self.cfg["client_name"],
            self.cfg["platform"]
        )
        post_txt = self.build_post(event)

        if self.cfg["platform"] == "mastodon":
            self.send_post_to_mastodon(event, client, post_txt)
        elif self.cfg["platform"] == "bluesky":
            embed_external = (
                self.build_embed_external(event, client) if event.get("img") else None
            )
            self.send_post_to_bluesky(event, client, post_txt, embed_external)

    def build_embed_external(
        self,
        event: Dict[str, Any],
        client: Any
    ) -> models.AppBskyEmbedExternal.Main:
        """
        Build an external embed object for Bluesky posts.

        Args:
            event: Event data dictionary.
            client: Authenticated Bluesky client.

        Returns:
            A Bluesky external embed object.
        """
        url = f"{self.base_path}/{event['img']}"
        filename = self.download_image(url)

        with open(filename, "rb") as f:
            img_data = f.read()

        img_data = self._compress_for_bluesky(img_data)
        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,
            )
        )

    _BLUESKY_MAX_BLOB_BYTES = 1_000_000

    @staticmethod
    def _compress_for_bluesky(img_data: bytes) -> bytes:
        """Return img_data compressed to under 1 MB for Bluesky, converting to JPEG if needed."""
        if len(img_data) <= PromoteAnniversary._BLUESKY_MAX_BLOB_BYTES:
            return img_data
        image = Image.open(io.BytesIO(img_data)).convert("RGB")
        for quality in (85, 70, 55, 40):
            buf = io.BytesIO()
            image.save(buf, format="JPEG", quality=quality, optimize=True)
            if buf.tell() <= PromoteAnniversary._BLUESKY_MAX_BLOB_BYTES:
                return buf.getvalue()
        return buf.getvalue()

    @staticmethod
    def get_bluesky_did(platform_user_handle: str) -> Optional[str]:
        """
        Resolve a Bluesky handle into a DID.

        Args:
            platform_user_handle: User handle on Bluesky (with or without '@').

        Returns:
            The DID string if found, otherwise None.
        """
        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()
                return data.get("did")
            logger.warning(
                "Failed to retrieve data. Status code: %s",
                response.status_code,
            )
        except requests.RequestException as e:
            logger.warning("An error occurred: %s", e)
        return None

    @staticmethod
    def add_whitespace_if_needed(text_chunk: str) -> str:
        """Ensure spacing consistency for Bluesky text chunks."""
        return text_chunk + " " if not text_chunk.endswith(("(", "{", "[")) else text_chunk

    def send_post_to_bluesky(
        self,
        event: Dict[str, Any],
        client: Any,
        post_txt: client_utils.TextBuilder,
        embed_external: Any
    ) -> None:
        """Send a post to Bluesky with optional media embed."""
        self.logger.info(
            "Preview your post...\n\n%s",
            post_txt.build_text()
        )
        try:
            client.send_post(text=post_txt, embed=embed_external)
            self.logger.info("Posted 🎉")
        except Exception as e:
            self.logger.exception("Exception %s for %s", e, event["name"])

    def send_post_to_mastodon(
        self,
        event: Dict[str, Any],
        client: Any,
        post_txt: str
    ) -> None:
        """Send a post to Mastodon, with media if available."""
        if event.get("img"):
            try:
                self.logger.info("Uploading media to Mastodon")
                url = f"{self.base_path}/{event['img']}"
                filename = self.download_image(url)

                media_upload = client.media_post(filename)
                description = event.get("alt") or str(event["name"])
                client.media_update(media_upload, description=description)

                client.status_post(post_txt, media_ids=[media_upload])
                self.logger.info("Posted with image 🎉")
                return
            except Exception as e:
                self.logger.exception(
                    "Media upload failed for %s: %s — falling back to text-only",
                    event.get("name"),
                    e,
                )

        try:
            client.status_post(post_txt)
            self.logger.info("Posted without image 🎉")
        except Exception as e:
            self.logger.exception("Exception %s for %s", e, event.get("name"))

cfg property

Property to ensure that the dictionary is initialized.

__init__(config_dict=None, no_dry_run=True)

Initialize a PromoteAnniversary handler.

Parameters:

Name	Type	Description	Default
config_dict	Optional[Dict[str, Any]]	Optional configuration dictionary.	None
no_dry_run	bool	Whether to actually execute posting (True) or just simulate actions (False).	True

Source code in src/promote_anniversaries.py

def __init__(
    self,
    config_dict: Optional[Dict[str, Any]] = None,
    no_dry_run: bool = True
) -> None:
    """
    Initialize a PromoteAnniversary handler.

    Args:
        config_dict: Optional configuration dictionary.
        no_dry_run: Whether to actually execute posting (True)
            or just simulate actions (False).
    """
    self.logger = logging.getLogger(__name__)
    self.config_dict = config_dict
    self.no_dry_run = no_dry_run
    self.base_path = (
        "https://raw.githubusercontent.com/cosimameyer/"
        "illustrations/main/amazing-women"
    )

add_whitespace_if_needed(text_chunk) staticmethod

Ensure spacing consistency for Bluesky text chunks.

Source code in src/promote_anniversaries.py

@staticmethod
def add_whitespace_if_needed(text_chunk: str) -> str:
    """Ensure spacing consistency for Bluesky text chunks."""
    return text_chunk + " " if not text_chunk.endswith(("(", "{", "[")) else text_chunk

build_embed_external(event, client)

Build an external embed object for Bluesky posts.

Parameters:

Name	Type	Description	Default
event	Dict[str, Any]	Event data dictionary.	required
client	Any	Authenticated Bluesky client.	required

Returns:

Type	Description
Main	A Bluesky external embed object.

Source code in src/promote_anniversaries.py

def build_embed_external(
    self,
    event: Dict[str, Any],
    client: Any
) -> models.AppBskyEmbedExternal.Main:
    """
    Build an external embed object for Bluesky posts.

    Args:
        event: Event data dictionary.
        client: Authenticated Bluesky client.

    Returns:
        A Bluesky external embed object.
    """
    url = f"{self.base_path}/{event['img']}"
    filename = self.download_image(url)

    with open(filename, "rb") as f:
        img_data = f.read()

    img_data = self._compress_for_bluesky(img_data)
    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)

Build the post text for Mastodon or Bluesky.

Parameters:

Name	Type	Description	Default
event	Dict[str, Any]	Dictionary containing event data.	required

Returns:

Type	Description
Union[str, TextBuilder]	A formatted post string (Mastodon) or TextBuilder object (Bluesky).

Source code in src/promote_anniversaries.py

def build_post(
    self,
    event: Dict[str, Any]
) -> Union[str, client_utils.TextBuilder]:
    """
    Build the post text for Mastodon or Bluesky.

    Args:
        event: Dictionary containing event data.

    Returns:
        A formatted post string (Mastodon) or TextBuilder object (Bluesky).
    """
    tags = "\n\n#amazingwomenintech #womenalsoknow #impactthefuture"

    if self.cfg["platform"] == "mastodon":
        return (
            f"Let's meet {event['name']} ✨\n\n"
            f"{event['description_mastodon']}\n\n"
            f"🔗 {event['wiki_link']}{tags}"
        )

    if self.cfg["platform"] == "bluesky":
        bluesky_max_graphemes = 300
        tag_list = [t.strip() for t in tags.split("#") if t.strip()]
        did = self.get_bluesky_did(event["bluesky"]) if event.get("bluesky") else None

        def _build(tag_subset, desc_override=None):
            desc = desc_override if desc_override is not None else event["description_bluesky"]
            tb = client_utils.TextBuilder()
            if event.get("bluesky"):
                tb.text("Let's meet ")
                tb.mention(event["bluesky"], did)
                tb.text(" ⭐️\n\n")
            else:
                tb.text(f"Let's meet {event['name']} ⭐️\n\n")
            split_text = [
                item.rstrip(" ")
                for item in re.split(r"(#\w+)", desc)
                if item.strip()
            ]
            for text_chunk in split_text:
                if text_chunk.startswith("#"):
                    for tag in text_chunk.split("#"):
                        if tag.strip():
                            tb.tag(f"#{tag.strip()}", tag.strip())
                else:
                    tb.text(self.add_whitespace_if_needed(text_chunk))
            tb.text("\n\n🔗 ")
            tb.link(event["wiki_link"], event["wiki_link"])
            if tag_subset:
                tb.text("\n\n")
                for i, tag in enumerate(tag_subset):
                    display = f"#{tag}" if i == len(tag_subset) - 1 else f"#{tag} "
                    tb.tag(display, tag)
            return tb

        # Drop trailing tags one by one until within limit
        for count in range(len(tag_list), -1, -1):
            text_builder = _build(tag_list[:count])
            if len(text_builder.build_text()) <= bluesky_max_graphemes:
                return text_builder

        # Still over limit: trim description to fit
        overhead = len(_build([], desc_override="").build_text())
        available = bluesky_max_graphemes - overhead - 1  # -1 for "…"
        desc_trimmed = event["description_bluesky"][:available].rstrip() + "…"
        return _build([], desc_override=desc_trimmed)

    raise ValueError(
        f"Unsupported platform: {self.cfg['platform']}"
    )

download_image(url)

Download an image from a URL if not already cached locally.

Parameters:

Name	Type	Description	Default
url	str	URL to the image.	required

Returns:

Type	Description
str	Path to the downloaded image file.

Source code in src/promote_anniversaries.py

def download_image(self, url: str) -> str:
    """
    Download an image from a URL if not already cached locally.

    Args:
        url: URL to the image.

    Returns:
        Path to the downloaded image file.
    """
    path = urlsplit(url).path
    filename = posixpath.basename(path)
    file_path = os.path.join(self.cfg["images"], filename)

    if not os.path.isfile(file_path):
        os.makedirs(self.cfg["images"], exist_ok=True)
        headers = {
            "User-Agent": (
                "Mozilla/5.0 (Windows NT 6.1; WOW64; rv:20.0) "
                "Gecko/20100101 Firefox/20.0"
            )
        }
        with requests.get(
            url,
            headers=headers,
            stream=True,
            timeout=REQUEST_TIMEOUT
        ) as response:
            with open(file_path, "wb") as out_file:
                shutil.copyfileobj(response.raw, out_file)
    else:
        self.logger.info("Image already downloaded: %s", file_path)

    return file_path

get_bluesky_did(platform_user_handle) staticmethod

Resolve a Bluesky handle into a DID.

Parameters:

Name	Type	Description	Default
platform_user_handle	str	User handle on Bluesky (with or without '@').	required

Returns:

Type	Description
Optional[str]	The DID string if found, otherwise None.

Source code in src/promote_anniversaries.py

@staticmethod
def get_bluesky_did(platform_user_handle: str) -> Optional[str]:
    """
    Resolve a Bluesky handle into a DID.

    Args:
        platform_user_handle: User handle on Bluesky (with or without '@').

    Returns:
        The DID string if found, otherwise None.
    """
    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()
            return data.get("did")
        logger.warning(
            "Failed to retrieve data. Status code: %s",
            response.status_code,
        )
    except requests.RequestException as e:
        logger.warning("An error occurred: %s", e)
    return None

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

Check whether the given date matches today's date.

Parameters:

Name	Type	Description	Default
date_str	str	Date string to compare (e.g., "08-30").	required
date_format	str	Format of the provided date string. Defaults to "%m-%d".	'%m-%d'

Returns:

Type	Description
bool	True if the date matches today's date, False otherwise.

Source code in src/promote_anniversaries.py

@staticmethod
def is_matching_current_date(
    date_str: str, date_format: str = "%m-%d"
) -> bool:
    """
    Check whether the given date matches today's date.

    Args:
        date_str: Date string to compare (e.g., "08-30").
        date_format: Format of the provided date string. 
                     Defaults to "%m-%d".

    Returns:
        True if the date matches today's date, False otherwise.
    """
    current_date = datetime.now().strftime(date_format)
    return date_str == current_date

promote_anniversary()

Main entry point. Loads configuration, fetches events, and posts if applicable.

Source code in src/promote_anniversaries.py

def promote_anniversary(self) -> None:
    """Main entry point. Loads configuration, fetches events, and posts if applicable."""
    if self.config_dict is None and self.no_dry_run:
        self._setup_config_from_env()

    if self.config_dict is None and self.no_dry_run:
        self.logger.error("No config_dict provided — cannot run")
        return

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

    client = self._connect_client() if self.no_dry_run else None
    if client is None and self.no_dry_run:
        self.logger.error("Failed to connect to %s", self.cfg["platform"])
        return

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

    for event in events:
        if self.is_matching_current_date(event["date"]):
            if not self.no_dry_run:
                self.logger.info(
                    "[DRY RUN] Would post anniversary for %s on %s",
                    event.get("name"),
                    event.get("date"),
                )
            else:
                self.send_post(event, client)

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: Dict[str, Any], client: Any) -> None:
    """Send a post to the configured platform (Mastodon or Bluesky)."""
    self.logger.info(
        "Preparing the post on %s (%s)...",
        self.cfg["client_name"],
        self.cfg["platform"]
    )
    post_txt = self.build_post(event)

    if self.cfg["platform"] == "mastodon":
        self.send_post_to_mastodon(event, client, post_txt)
    elif self.cfg["platform"] == "bluesky":
        embed_external = (
            self.build_embed_external(event, client) if event.get("img") else None
        )
        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: Dict[str, Any],
    client: Any,
    post_txt: client_utils.TextBuilder,
    embed_external: Any
) -> None:
    """Send a post to Bluesky with optional media embed."""
    self.logger.info(
        "Preview your post...\n\n%s",
        post_txt.build_text()
    )
    try:
        client.send_post(text=post_txt, embed=embed_external)
        self.logger.info("Posted 🎉")
    except Exception as e:
        self.logger.exception("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: Dict[str, Any],
    client: Any,
    post_txt: str
) -> None:
    """Send a post to Mastodon, with media if available."""
    if event.get("img"):
        try:
            self.logger.info("Uploading media to Mastodon")
            url = f"{self.base_path}/{event['img']}"
            filename = self.download_image(url)

            media_upload = client.media_post(filename)
            description = event.get("alt") or str(event["name"])
            client.media_update(media_upload, description=description)

            client.status_post(post_txt, media_ids=[media_upload])
            self.logger.info("Posted with image 🎉")
            return
        except Exception as e:
            self.logger.exception(
                "Media upload failed for %s: %s — falling back to text-only",
                event.get("name"),
                e,
            )

    try:
        client.status_post(post_txt)
        self.logger.info("Posted without image 🎉")
    except Exception as e:
        self.logger.exception("Exception %s for %s", e, event.get("name"))

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"

        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')
            )

        if self.config_dict.get('gen_ai_support'):
            self.genai_client = genai.Client(api_key=self.config_dict.get('gemini_api_key'))

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

        self.get_config()

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

        if self.no_dry_run:
            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
        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:
                if count_post >= 2:
                    break
                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.
        """
        n = len(feeds)
        if n == 0:
            return

        start_index = 0
        for i, f in enumerate(feeds):
            if counter_name in (f['name'], '\n', ''):
                start_index = i
                break

        next_index = start_index

        for offset in range(n):
            idx = (start_index + offset) % n
            feed = feeds[idx]

            if len(feed['rss_feed']) == 0 or feed['rss_feed'] == [None]:
                continue

            if count_post >= 2:
                next_index = idx
                self.logger.info(
                    "Successfully promoted blog posts. "
                    "Thank you and see you next time!")
                break

            count_post = self.process_feed(feed, count_post, client)
            next_index = (idx + 1) % n
            self.logger.info("=========================================")
        else:
            if count_post >= 2:
                self.logger.info(
                    "Successfully promoted blog posts. "
                    "Thank you and see you next time!")

        self.update_counter(feeds[next_index]['name'])

    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"""
        pub_date_str = entry.get('pub_date', '')
        if pub_date_str:
            try:
                return dateutil_parser.parse(pub_date_str).replace(tzinfo=None)
            except (ValueError, OverflowError):
                pass
        self.logger.warning("No matching date format found. Using current date.")
        return datetime.now()

    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
                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)

        return None

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

        post = f'📝 "{title}"\n\n' if title else ''

        if self.config_dict.get('gen_ai_support', None):
            summarized_blog_post = self.summarize_text(entry)
            if summarized_blog_post:
                post += summarized_blog_post + '\n\n'

        if name:
            post += f'👤 {name}'
        if platform_user_handle:
            post += f' ({platform_user_handle})'
        if name or platform_user_handle:
            post += '\n\n'

        post += f"🔗 {entry.get('link', '')}\n\n{tags}"

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

        return post

    @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)
        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
        ]
        _retryable_codes = ("429", "503")
        _max_attempts = 3
        _retry_wait = 30
        response = None
        for attempt in range(_max_attempts):
            try:
                response = self.genai_client.models.generate_content(
                    model=self.config_dict.get('gemini_model_name', ''),
                    contents=prompt_parts
                )
                break
            except Exception as e:  # pylint: disable=broad-except
                if attempt < _max_attempts - 1 and any(
                    code in str(e) for code in _retryable_codes
                ):
                    self.logger.info(
                        "Gemini API transient error (attempt %d/%d), "
                        "retrying in %ds: %s",
                        attempt + 1, _max_attempts, _retry_wait, e
                    )
                    time.sleep(_retry_wait)
                else:
                    raise
        response_cleaned = self.clean_response(response)
        safety_ratings = response.candidates[0].safety_ratings
        if safety_ratings and 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,
        title,
        name,
        platform_user_handle,
        tags,
        entry
    ):
        """
        Build post for Bluesky.
        """
        bluesky_max_graphemes = 300
        link = entry.get('link', '')
        platform_user_handle = self.check_platform_handle(platform_user_handle)

        summarized_blog_post = ''
        if self.config_dict.get('gen_ai_support', None):
            summarized_blog_post = self.summarize_text(entry) or ''

        # Resolve DID once so _build() never makes a duplicate HTTP call
        did = self.get_bluesky_did(platform_user_handle) if platform_user_handle else None

        tag_list = [t.strip() for t in tags.split('#') if t.strip()]

        def _build(tag_subset):
            tb = client_utils.TextBuilder()
            if title:
                tb.text(f'📝 "{title}"\n\n')
            if summarized_blog_post:
                tb.text(summarized_blog_post)
                tb.text('\n\n')
            if name:
                tb.text(f'👤 {name}')
            if platform_user_handle:
                tb.mention(f' ({platform_user_handle})', did)
            if name or platform_user_handle:
                tb.text('\n\n')
            tb.text('🔗 ')
            tb.link(link, link)
            tb.text('\n\n')
            for tag_clean in tag_subset:
                tb.tag(f'#{tag_clean} ', tag_clean)
            return tb

        # Try with all tags; drop from the end one by one until within limit
        for count in range(len(tag_list), -1, -1):
            text_builder = _build(tag_list[:count])
            if len(text_builder.build_text()) <= bluesky_max_graphemes:
                return text_builder

        return _build([])

    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', '')

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

    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:
                    prev_count = count_post
                    count_post = self._process_feed(
                        client,
                        count_post,
                        feed_config
                    )
                    if count_post > prev_count:
                        self.logger.info(
                            'New RSS feeds are successfully loaded and '
                            'processed.'
                        )
                    else:
                        self.logger.info(
                            'Feed has new entries but all are already '
                            'in the archive — nothing to post.'
                        )
                    return count_post
                self.logger.info(
                    'Archive is up to date with the feed — '
                    'no new entries since last run.'
                )
                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, 'w', encoding='utf-8') 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
            if count_fails >= 1:
                self.logger.warning(
                    "Stopping feed after post failure — skipping remaining entries."
                )
                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']:
                if self.no_dry_run:
                    feed_config['rss_feed_archive']['link'].append(en['link'])
                    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)
                else:
                    self.logger.info(
                        "[DRY RUN] Would post: '%s' from %s",
                        en.get('title', 'unknown'),
                        en.get('link', 'unknown'),
                    )
                    count_post += 1
                    count += 1

        if self.no_dry_run and result == 'success':
            try:
                self._save_rss_feed_archive(
                    feed_config['feed'],
                    feed_config['rss_feed_archive']
                )
            except OSError as e:
                self.logger.error(
                    "Failed to save archive for %s: %s",
                    feed_config['feed'].get('name', 'unknown'),
                    e,
                )

        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', '')

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

build_post_bluesky(title, name, platform_user_handle, tags, entry)

Build post for Bluesky.

Source code in src/promote_blog_post.py

def build_post_bluesky(
    self,
    title,
    name,
    platform_user_handle,
    tags,
    entry
):
    """
    Build post for Bluesky.
    """
    bluesky_max_graphemes = 300
    link = entry.get('link', '')
    platform_user_handle = self.check_platform_handle(platform_user_handle)

    summarized_blog_post = ''
    if self.config_dict.get('gen_ai_support', None):
        summarized_blog_post = self.summarize_text(entry) or ''

    # Resolve DID once so _build() never makes a duplicate HTTP call
    did = self.get_bluesky_did(platform_user_handle) if platform_user_handle else None

    tag_list = [t.strip() for t in tags.split('#') if t.strip()]

    def _build(tag_subset):
        tb = client_utils.TextBuilder()
        if title:
            tb.text(f'📝 "{title}"\n\n')
        if summarized_blog_post:
            tb.text(summarized_blog_post)
            tb.text('\n\n')
        if name:
            tb.text(f'👤 {name}')
        if platform_user_handle:
            tb.mention(f' ({platform_user_handle})', did)
        if name or platform_user_handle:
            tb.text('\n\n')
        tb.text('🔗 ')
        tb.link(link, link)
        tb.text('\n\n')
        for tag_clean in tag_subset:
            tb.tag(f'#{tag_clean} ', tag_clean)
        return tb

    # Try with all tags; drop from the end one by one until within limit
    for count in range(len(tag_list), -1, -1):
        text_builder = _build(tag_list[:count])
        if len(text_builder.build_text()) <= bluesky_max_graphemes:
            return text_builder

    return _build([])

build_post_mastodon(title, name, platform_user_handle, tags, entry)

Build Mastodon post.

Source code in src/promote_blog_post.py

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

    post = f'📝 "{title}"\n\n' if title else ''

    if self.config_dict.get('gen_ai_support', None):
        summarized_blog_post = self.summarize_text(entry)
        if summarized_blog_post:
            post += summarized_blog_post + '\n\n'

    if name:
        post += f'👤 {name}'
    if platform_user_handle:
        post += f' ({platform_user_handle})'
    if name or platform_user_handle:
        post += '\n\n'

    post += f"🔗 {entry.get('link', '')}\n\n{tags}"

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

    return post

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
            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)

    return None

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"

    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')
        )

    if self.config_dict.get('gen_ai_support'):
        self.genai_client = genai.Client(api_key=self.config_dict.get('gemini_api_key'))

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"""
    pub_date_str = entry.get('pub_date', '')
    if pub_date_str:
        try:
            return dateutil_parser.parse(pub_date_str).replace(tzinfo=None)
        except (ValueError, OverflowError):
            pass
    self.logger.warning("No matching date format found. Using current date.")
    return datetime.now()

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:
                prev_count = count_post
                count_post = self._process_feed(
                    client,
                    count_post,
                    feed_config
                )
                if count_post > prev_count:
                    self.logger.info(
                        'New RSS feeds are successfully loaded and '
                        'processed.'
                    )
                else:
                    self.logger.info(
                        'Feed has new entries but all are already '
                        'in the archive — nothing to post.'
                    )
                return count_post
            self.logger.info(
                'Archive is up to date with the feed — '
                'no new entries since last run.'
            )
            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.
    """
    n = len(feeds)
    if n == 0:
        return

    start_index = 0
    for i, f in enumerate(feeds):
        if counter_name in (f['name'], '\n', ''):
            start_index = i
            break

    next_index = start_index

    for offset in range(n):
        idx = (start_index + offset) % n
        feed = feeds[idx]

        if len(feed['rss_feed']) == 0 or feed['rss_feed'] == [None]:
            continue

        if count_post >= 2:
            next_index = idx
            self.logger.info(
                "Successfully promoted blog posts. "
                "Thank you and see you next time!")
            break

        count_post = self.process_feed(feed, count_post, client)
        next_index = (idx + 1) % n
        self.logger.info("=========================================")
    else:
        if count_post >= 2:
            self.logger.info(
                "Successfully promoted blog posts. "
                "Thank you and see you next time!")

    self.update_counter(feeds[next_index]['name'])

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()

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

    if self.no_dry_run:
        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
    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:
            if count_post >= 2:
                break
            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)
    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
    ]
    _retryable_codes = ("429", "503")
    _max_attempts = 3
    _retry_wait = 30
    response = None
    for attempt in range(_max_attempts):
        try:
            response = self.genai_client.models.generate_content(
                model=self.config_dict.get('gemini_model_name', ''),
                contents=prompt_parts
            )
            break
        except Exception as e:  # pylint: disable=broad-except
            if attempt < _max_attempts - 1 and any(
                code in str(e) for code in _retryable_codes
            ):
                self.logger.info(
                    "Gemini API transient error (attempt %d/%d), "
                    "retrying in %ds: %s",
                    attempt + 1, _max_attempts, _retry_wait, e
                )
                time.sleep(_retry_wait)
            else:
                raise
    response_cleaned = self.clean_response(response)
    safety_ratings = response.candidates[0].safety_ratings
    if safety_ratings and 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)