Skip to content

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 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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 = {}
        platform = (os.getenv("PLATFORM") or "").lower()
        if not platform and not self.config_dict.get("platform"):
            raise ValueError("PLATFORM environment variable is not set.")
        self.config_dict.setdefault("platform", 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", "https://bsky.social")
        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 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
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 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
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 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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 = {}
    platform = (os.getenv("PLATFORM") or "").lower()
    if not platform and not self.config_dict.get("platform"):
        raise ValueError("PLATFORM environment variable is not set.")
    self.config_dict.setdefault("platform", 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", "https://bsky.social")
    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 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
class BoostTags:
    """
    Handles boosting of posts containing specified tags across different platforms.
    Currently supports Bluesky. Mastodon support is stubbed.
    """

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

        Args:
            config_dict: Optional configuration dictionary for the bot.
            no_dry_run: If True, actually perform reposts; 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_tags() or pass "
                "config_dict to the constructor before accessing cfg"
            )
        return self.config_dict

    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.
        """
        self.set_up_config_dict()

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

        if self.cfg["platform"] == "mastodon":
            self._boost_tags_mastodon()
        elif self.cfg["platform"] == "bluesky":
            self._boost_tags_bluesky()

    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 = {}
        platform = (os.getenv("PLATFORM") or "").lower()
        if not platform and not self.config_dict.get("platform"):
            raise ValueError("PLATFORM environment variable is not set.")
        self.config_dict.setdefault("platform", 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", "CommunityBot"))
        self.config_dict.setdefault(
            "tags",
            [t.strip() for t in os.getenv("TAGS_TO_BOOST", "").split(",") if t.strip()],
        )
        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", "https://bsky.social")
        else:
            raise ValueError(
                f"Unknown platform: {self.config_dict['platform']!r}. "
                "Expected 'mastodon' or 'bluesky'."
            )

    def _boost_tags_mastodon(self) -> None:
        """Handle reposting tags on Mastodon."""
        raise NotImplementedError("Mastodon tag boosting is not yet implemented.")

    _SEEN_CIDS_FILE = Path("metadata/bluesky_seen_cids.json")
    _SEEN_CIDS_MAX_AGE_DAYS = 7
    _VALID_TAG_PATTERN = re.compile(r'^[a-zA-Z0-9_]{1,100}$')

    def _load_seen_cids(self) -> dict:
        """Load persistent seen-CIDs from file, pruning entries older than 7 days."""
        if self._SEEN_CIDS_FILE.exists():
            try:
                with self._SEEN_CIDS_FILE.open("r", encoding="utf-8") as f:
                    data = json.load(f)
            except (json.JSONDecodeError, OSError):
                data = {}
        else:
            data = {}
        cutoff = datetime.now(timezone.utc) - timedelta(days=self._SEEN_CIDS_MAX_AGE_DAYS)
        return {
            cid: ts
            for cid, ts in data.items()
            if datetime.fromisoformat(ts) > cutoff
        }

    def _save_seen_cids(self, seen_cids: dict) -> None:
        """Save persistent seen-CIDs to file."""
        try:
            self._SEEN_CIDS_FILE.parent.mkdir(parents=True, exist_ok=True)
            with self._SEEN_CIDS_FILE.open("w", encoding="utf-8") as f:
                json.dump(seen_cids, f, ensure_ascii=False, indent=2)
        except OSError as e:
            self.logger.warning("Could not save seen CIDs file: %s", e)

    def _boost_tags_bluesky(self) -> None:
        """Handle reposting tags on Bluesky."""
        # Fix 1: validate tags from config before proceeding
        raw_tags = self.cfg.get("tags", [])
        if isinstance(raw_tags, str):
            raw_tag_list = [t.strip().lstrip('#') for t in raw_tags.split(',')]
        else:
            raw_tag_list = [t.strip().lstrip('#') for t in raw_tags]
        valid_tags = [t for t in raw_tag_list if self._VALID_TAG_PATTERN.match(t)]
        if not valid_tags:
            self.logger.warning("No valid tags after validation — skipping boost.")
            return

        try:
            client = login_bluesky(cast(BlueskyConfig, self.config_dict))
        except InvokeTimeoutError:
            self.logger.error("Timed out while logging in to Bluesky. Aborting.")
            return
        self.logger.info(" > Fetched Bluesky account data.")
        self.logger.info(" > Starting search-loop for reposting.")

        # Fix 9: use persistent seen-CIDs file instead of single-page timeline dedup
        seen_cids_dict = self._load_seen_cids()
        try:
            timeline = client.get_timeline(algorithm="reverse-chronological")
            now_ts = datetime.now(timezone.utc).isoformat()
            for post in timeline.feed:
                seen_cids_dict[post.post.cid] = now_ts
        except InvokeTimeoutError:
            self.logger.error("Timed out fetching timeline. Aborting.")
            return
        seen_cids = set(seen_cids_dict.keys())

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

        for tag in valid_tags:
            tag = tag.lower().strip("# ")
            self.logger.info(" > Searching for tag #%s", tag)
            if boost_count >= max_boosts:
                self.logger.info(
                    " > Reached max boosts per run (%d), stopping.", max_boosts
                )
                break

            try:
                response = client.app.bsky.feed.search_posts(
                    params={"q": tag, "tag": [tag], "sort": "top", "limit": 50}
                )
            except InvokeTimeoutError:
                self.logger.error("Timed out searching posts for tag #%s. Skipping.", tag)
                continue
            for post in response.posts:
                if boost_count >= max_boosts:
                    break

                # Prefer facets (structured ATProto tags) over text parsing.
                tags_in_post: set = set()
                for facet in (getattr(post.record, "facets", None) or []):
                    for feature in facet.features:
                        if hasattr(feature, "tag"):
                            tags_in_post.add(feature.tag.lower())
                if not tags_in_post:
                    tags_in_post = {
                        t.strip("#.,!?").lower()
                        for t in post.record.text.split()
                        if t.startswith("#")
                    }

                own_handle = (self.cfg.get("username") or "").lower()
                if (
                    tag in tags_in_post
                    and post.cid not in seen_cids
                    and post.author.handle.lower() != own_handle
                ):
                    if not self.no_dry_run:
                        self.logger.info(
                            "   * [DRY RUN] Would repost URI %s CID %s by %s",
                            post.uri,
                            post.cid,
                            post.author.handle,
                        )
                        seen_cids.add(post.cid)
                        seen_cids_dict[post.cid] = datetime.now(timezone.utc).isoformat()
                        boost_count += 1
                    else:
                        try:
                            result = client.repost(uri=post.uri, cid=post.cid)
                            self.logger.info(
                                "   * Reposted post by %s (ref: %s)",
                                post.author.handle,
                                result,
                            )
                            seen_cids.add(post.cid)
                            seen_cids_dict[post.cid] = datetime.now(timezone.utc).isoformat()
                            boost_count += 1
                        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._save_seen_cids(seen_cids_dict)
        self.logger.info("Finished processing Bluesky reposts.")
cfg property

Property to ensure that the dictionary is initialized.

__init__(config_dict=None, no_dry_run=True)

Initialize the BoostTags handler.

Parameters:

Name Type Description Default
config_dict Optional[Dict[str, Any]]

Optional configuration dictionary for the bot.

 None
no_dry_run bool

If True, actually perform reposts; if False, dry run.

 True
Source code in src/boost_tags.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self,
    config_dict: Optional[Dict[str, Any]] = None,
    no_dry_run: bool = True,
) -> None:
    """
    Initialize the BoostTags handler.

    Args:
        config_dict: Optional configuration dictionary for the bot.
        no_dry_run: If True, actually perform reposts; if False, dry run.
    """
    self.logger = logging.getLogger(__name__)

    self.no_dry_run = no_dry_run
    self.config_dict = config_dict
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 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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.
    """
    self.set_up_config_dict()

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

    if self.cfg["platform"] == "mastodon":
        self._boost_tags_mastodon()
    elif self.cfg["platform"] == "bluesky":
        self._boost_tags_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_tags.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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 = {}
    platform = (os.getenv("PLATFORM") or "").lower()
    if not platform and not self.config_dict.get("platform"):
        raise ValueError("PLATFORM environment variable is not set.")
    self.config_dict.setdefault("platform", 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", "CommunityBot"))
    self.config_dict.setdefault(
        "tags",
        [t.strip() for t in os.getenv("TAGS_TO_BOOST", "").split(",") if t.strip()],
    )
    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", "https://bsky.social")
    else:
        raise ValueError(
            f"Unknown platform: {self.config_dict['platform']!r}. "
            "Expected 'mastodon' or 'bluesky'."
        )

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 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
class DebugBots:
    """
    Class to handle debugging of all modules.
    """
    def __init__(self):
        self.bot = 'pyladies'  # 'pyladies' or 'rladies'
        self.what_to_debug = 'package'  # 'blog', 'boost_tags', 'rss', 'anniversary', 'boost_mentions', 'get_packages', 'package'
        self.platform = 'bluesky'  # 'bluesky' or 'mastodon'
        self.no_dry_run = True  # 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()

        elif self.what_to_debug == 'get_packages':
            config_dict = self.get_config_packages()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r — check bot settings.",
                    self.bot
                )
                return
            packages_data_handler = PackagesData(
                config_dict,
                self.no_dry_run
            )
            packages_data_handler.get_packages_data()

        elif self.what_to_debug == 'package':
            config_dict = self.get_config_package()
            if config_dict is None:
                logger.error(
                    "No config for bot=%r platform=%r — check bot/platform settings.",
                    self.bot, self.platform
                )
                return
            promote_package_handler = PromotePackage(
                config_dict,
                self.no_dry_run
            )
            promote_package_handler.promote_package()

    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": os.path.join(_METADATA_DIR, "rladies_counter_bluesky.txt"),
                    "json_file": os.path.join(_METADATA_DIR, "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": os.path.join(_METADATA_DIR, "rladies_counter.txt"),
                    "json_file": os.path.join(_METADATA_DIR, "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",
                    "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",
                    "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

    def get_config_packages(self):
        """Method to generate config for fetching package metadata."""
        if self.bot == 'pyladies':
            return {
                "base_url": (
                    "https://github.com/cosimameyer/"
                    "awesome-pyladies-creations/tree/main/data/packages"
                ),
                "github_raw_url": (
                    "https://raw.githubusercontent.com/cosimameyer/"
                    "awesome-pyladies-creations/main/data/packages"
                ),
                "json_file": os.path.join(_METADATA_DIR, "pyladies_packages_meta_data.json"),
            }
        if self.bot == 'rladies':
            return {
                "base_url": (
                    "https://github.com/rladies/"
                    "awesome-rladies-creations/tree/main/data/packages"
                ),
                "github_raw_url": (
                    "https://raw.githubusercontent.com/rladies/"
                    "awesome-rladies-creations/main/data/packages"
                ),
                "json_file": os.path.join(_METADATA_DIR, "rladies_packages_meta_data.json"),
            }
        return None

    def get_config_package(self):
        """Method to generate config for promoting packages."""
        if self.bot == 'pyladies':
            if self.platform == 'bluesky':
                return {
                    "counter": os.path.join(_METADATA_DIR, "pyladies_packages_counter_bluesky.txt"),
                    "json_file": os.path.join(_METADATA_DIR, "pyladies_packages_meta_data.json"),
                    "archive_file": os.path.join(_METADATA_DIR, "pyladies_packages_archive.json"),
                    "client_name": "pyladies_bot",
                    "api_base_url": self.platform,
                    "password": os.getenv("PYLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("PYLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                }
            if self.platform == 'mastodon':
                return {
                    "counter": os.path.join(_METADATA_DIR, "pyladies_packages_counter_mastodon.txt"),
                    "json_file": os.path.join(_METADATA_DIR, "pyladies_packages_meta_data.json"),
                    "archive_file": os.path.join(_METADATA_DIR, "pyladies_packages_archive.json"),
                    "client_name": "pyladies_bot",
                    "api_base_url": config.API_BASE_URL,
                    "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 {
                    "counter": os.path.join(_METADATA_DIR, "rladies_packages_counter_bluesky.txt"),
                    "json_file": os.path.join(_METADATA_DIR, "rladies_packages_meta_data.json"),
                    "archive_file": os.path.join(_METADATA_DIR, "rladies_packages_archive.json"),
                    "client_name": "rladies_bot",
                    "api_base_url": self.platform,
                    "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                    "username": os.getenv("RLADIES_BSKY_USERNAME"),
                    "platform": self.platform,
                }
            if self.platform == 'mastodon':
                return {
                    "counter": os.path.join(_METADATA_DIR, "rladies_packages_counter_mastodon.txt"),
                    "json_file": os.path.join(_METADATA_DIR, "rladies_packages_meta_data.json"),
                    "archive_file": os.path.join(_METADATA_DIR, "rladies_packages_archive.json"),
                    "client_name": "rladies_bot",
                    "api_base_url": config.API_BASE_URL,
                    "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_anniversary()

Method to get config for promoting anniversaries

Source code in src/debug.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
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 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
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": os.path.join(_METADATA_DIR, "rladies_counter_bluesky.txt"),
                "json_file": os.path.join(_METADATA_DIR, "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": os.path.join(_METADATA_DIR, "rladies_counter.txt"),
                "json_file": os.path.join(_METADATA_DIR, "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 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
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",
                "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",
                "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_package()

Method to generate config for promoting packages.

Source code in src/debug.py 
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
def get_config_package(self):
    """Method to generate config for promoting packages."""
    if self.bot == 'pyladies':
        if self.platform == 'bluesky':
            return {
                "counter": os.path.join(_METADATA_DIR, "pyladies_packages_counter_bluesky.txt"),
                "json_file": os.path.join(_METADATA_DIR, "pyladies_packages_meta_data.json"),
                "archive_file": os.path.join(_METADATA_DIR, "pyladies_packages_archive.json"),
                "client_name": "pyladies_bot",
                "api_base_url": self.platform,
                "password": os.getenv("PYLADIES_BSKY_PASSWORD"),
                "username": os.getenv("PYLADIES_BSKY_USERNAME"),
                "platform": self.platform,
            }
        if self.platform == 'mastodon':
            return {
                "counter": os.path.join(_METADATA_DIR, "pyladies_packages_counter_mastodon.txt"),
                "json_file": os.path.join(_METADATA_DIR, "pyladies_packages_meta_data.json"),
                "archive_file": os.path.join(_METADATA_DIR, "pyladies_packages_archive.json"),
                "client_name": "pyladies_bot",
                "api_base_url": config.API_BASE_URL,
                "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 {
                "counter": os.path.join(_METADATA_DIR, "rladies_packages_counter_bluesky.txt"),
                "json_file": os.path.join(_METADATA_DIR, "rladies_packages_meta_data.json"),
                "archive_file": os.path.join(_METADATA_DIR, "rladies_packages_archive.json"),
                "client_name": "rladies_bot",
                "api_base_url": self.platform,
                "password": os.getenv("RLADIES_BSKY_PASSWORD"),
                "username": os.getenv("RLADIES_BSKY_USERNAME"),
                "platform": self.platform,
            }
        if self.platform == 'mastodon':
            return {
                "counter": os.path.join(_METADATA_DIR, "rladies_packages_counter_mastodon.txt"),
                "json_file": os.path.join(_METADATA_DIR, "rladies_packages_meta_data.json"),
                "archive_file": os.path.join(_METADATA_DIR, "rladies_packages_archive.json"),
                "client_name": "rladies_bot",
                "api_base_url": config.API_BASE_URL,
                "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_packages()

Method to generate config for fetching package metadata.

Source code in src/debug.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
def get_config_packages(self):
    """Method to generate config for fetching package metadata."""
    if self.bot == 'pyladies':
        return {
            "base_url": (
                "https://github.com/cosimameyer/"
                "awesome-pyladies-creations/tree/main/data/packages"
            ),
            "github_raw_url": (
                "https://raw.githubusercontent.com/cosimameyer/"
                "awesome-pyladies-creations/main/data/packages"
            ),
            "json_file": os.path.join(_METADATA_DIR, "pyladies_packages_meta_data.json"),
        }
    if self.bot == 'rladies':
        return {
            "base_url": (
                "https://github.com/rladies/"
                "awesome-rladies-creations/tree/main/data/packages"
            ),
            "github_raw_url": (
                "https://raw.githubusercontent.com/rladies/"
                "awesome-rladies-creations/main/data/packages"
            ),
            "json_file": os.path.join(_METADATA_DIR, "rladies_packages_meta_data.json"),
        }
    return None
get_config_rss()

Method to generate config for fetching RSS data

Source code in src/debug.py 
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
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 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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()

    elif self.what_to_debug == 'get_packages':
        config_dict = self.get_config_packages()
        if config_dict is None:
            logger.error(
                "No config for bot=%r — check bot settings.",
                self.bot
            )
            return
        packages_data_handler = PackagesData(
            config_dict,
            self.no_dry_run
        )
        packages_data_handler.get_packages_data()

    elif self.what_to_debug == 'package':
        config_dict = self.get_config_package()
        if config_dict is None:
            logger.error(
                "No config for bot=%r platform=%r — check bot/platform settings.",
                self.bot, self.platform
            )
            return
        promote_package_handler = PromotePackage(
            config_dict,
            self.no_dry_run
        )
        promote_package_handler.promote_package()

get_packages_data

Module to get package metadata from awesome-*-creations repos.

PackagesData

Handle gathering package metadata from awesome-*-creations repos.

Source code in src/get_packages_data.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class PackagesData:
    """
    Handle gathering package metadata from awesome-*-creations repos.
    """

    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.config_dict:
            self.base_url = self.config_dict.get("base_url")
            self.github_raw_url = self.config_dict.get("github_raw_url")
            self.json_file = self.config_dict.get("json_file")
        else:
            self.base_url = os.getenv("BASE_URL")
            self.github_raw_url = os.getenv("GITHUB_RAW_URL")
            self.json_file = os.getenv("JSON_FILE")

    _ALLOWED_URL_PREFIXES = (
        "https://github.com/",
        "https://raw.githubusercontent.com/",
    )

    def _validate_urls(self) -> None:
        """Validate that base_url and github_raw_url are allowed GitHub URLs."""
        for attr_name in ("base_url", "github_raw_url"):
            url = getattr(self, attr_name, None) or ""
            if not any(url.startswith(prefix) for prefix in self._ALLOWED_URL_PREFIXES):
                raise ValueError(
                    f"{attr_name} {url!r} is not an allowed URL — must start with "
                    f"'https://github.com/' or 'https://raw.githubusercontent.com/'"
                )

    def _validate_json_file_path(self) -> None:
        """Validate that json_file is within the project root."""
        safe_root = Path.cwd().resolve()
        target = Path(self.json_file).resolve()
        if not str(target).startswith(str(safe_root)):
            raise ValueError(
                f"json_file path {self.json_file!r} escapes the project root — refusing to write."
            )

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

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

            self.logger.info(
                "Package meta data successfully saved to %s",
                self.json_file
            )
        else:
            self.logger.info(
                "[DRY RUN] Would write %d entries to %s:\n%s",
                len(meta_data),
                self.json_file,
                json.dumps(meta_data, ensure_ascii=False, indent=2),
            )

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

        Returns:
            list[str]: A list of JSON file URLs.
        """
        self._validate_urls()
        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)
        try:
            items = (
                payload["payload"]["codeViewTreeRoute"]["tree"]["items"]
            )
        except KeyError as exc:
            top_keys = list(payload.get("payload", {}).keys())
            raise RuntimeError(
                f"Unexpected GitHub payload structure — missing key {exc}. "
                f"Available keys under 'payload': {top_keys}. "
                "Update the key path in get_json_file_names()."
            ) from exc
        return [
            f"{self.github_raw_url}/{item['path'].split('/')[-1]}"
            for item in items
            if item["path"].endswith(".json")
        ]

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

        Returns:
            list[dict]: A list of parsed JSON objects.
        """
        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 from a single package JSON.

        Handles both source formats:
        - PyLadies: maintainers[] + pypi_url + docs_url; social nested under
          maintainers[].social_media[].
        - RLadies: authors[] + pkdown_url + bug_reports_url; no social in
          package JSON. Authors with a directory_id are R-Ladies members.

        PyLadies docs_url is mapped to pkdown_url (semantically equivalent).

        Returns:
            dict: Normalised metadata dictionary with a contributors list.
                  R-Ladies community members carry a directory_id field;
                  other contributors and all PyLadies maintainers do not.
        """
        maintainers = content.get("maintainers", [])
        authors = content.get("authors", [])

        contributors = [
            _extract_contributor(p)
            for p in (maintainers or authors)
        ]

        return {
            "name": content.get("name", ""),
            "title": content.get("title", ""),
            "description": content.get("description", ""),
            "repo_url": content.get("repo_url", ""),
            "pypi_url": content.get("pypi_url", ""),
            "pkdown_url": content.get("pkdown_url") or content.get("docs_url") or "",
            "bug_reports_url": content.get("bug_reports_url", ""),
            "logo_url": content.get("logo_url", ""),
            "last_updated": content.get("last_updated", ""),
            "contributors": contributors,
        }

    def get_meta_data(self, contents_list: list[dict]) -> list[dict]:
        """
        Aggregate metadata from all package JSON files.

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

Extract metadata from a single package JSON.

Handles both source formats: - PyLadies: maintainers[] + pypi_url + docs_url; social nested under maintainers[].social_media[]. - RLadies: authors[] + pkdown_url + bug_reports_url; no social in package JSON. Authors with a directory_id are R-Ladies members.

PyLadies docs_url is mapped to pkdown_url (semantically equivalent).

Returns:

Name Type Description
dict dict

Normalised metadata dictionary with a contributors list. R-Ladies community members carry a directory_id field; other contributors and all PyLadies maintainers do not.
Source code in src/get_packages_data.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
@staticmethod
def extract_info(content: dict) -> dict:
    """
    Extract metadata from a single package JSON.

    Handles both source formats:
    - PyLadies: maintainers[] + pypi_url + docs_url; social nested under
      maintainers[].social_media[].
    - RLadies: authors[] + pkdown_url + bug_reports_url; no social in
      package JSON. Authors with a directory_id are R-Ladies members.

    PyLadies docs_url is mapped to pkdown_url (semantically equivalent).

    Returns:
        dict: Normalised metadata dictionary with a contributors list.
              R-Ladies community members carry a directory_id field;
              other contributors and all PyLadies maintainers do not.
    """
    maintainers = content.get("maintainers", [])
    authors = content.get("authors", [])

    contributors = [
        _extract_contributor(p)
        for p in (maintainers or authors)
    ]

    return {
        "name": content.get("name", ""),
        "title": content.get("title", ""),
        "description": content.get("description", ""),
        "repo_url": content.get("repo_url", ""),
        "pypi_url": content.get("pypi_url", ""),
        "pkdown_url": content.get("pkdown_url") or content.get("docs_url") or "",
        "bug_reports_url": content.get("bug_reports_url", ""),
        "logo_url": content.get("logo_url", ""),
        "last_updated": content.get("last_updated", ""),
        "contributors": contributors,
    }
get_json_data()

Download and parse JSON files from discovered file URLs.

Returns:

Type Description
list[dict]

list[dict]: A list of parsed JSON objects.
Source code in src/get_packages_data.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def get_json_data(self) -> list[dict]:
    """
    Download and parse JSON files from discovered file URLs.

    Returns:
        list[dict]: A list of parsed JSON objects.
    """
    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.

Returns:

Type Description
list[str]

list[str]: A list of JSON file URLs.
Source code in src/get_packages_data.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def get_json_file_names(self) -> list[str]:
    """
    Retrieve available JSON file names from the configured base URL.

    Returns:
        list[str]: A list of JSON file URLs.
    """
    self._validate_urls()
    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)
    try:
        items = (
            payload["payload"]["codeViewTreeRoute"]["tree"]["items"]
        )
    except KeyError as exc:
        top_keys = list(payload.get("payload", {}).keys())
        raise RuntimeError(
            f"Unexpected GitHub payload structure — missing key {exc}. "
            f"Available keys under 'payload': {top_keys}. "
            "Update the key path in get_json_file_names()."
        ) from exc
    return [
        f"{self.github_raw_url}/{item['path'].split('/')[-1]}"
        for item in items
        if item["path"].endswith(".json")
    ]
get_meta_data(contents_list)

Aggregate metadata from all package JSON files.

Returns:

Type Description
list[dict]

list[dict]: A list of metadata dictionaries.
Source code in src/get_packages_data.py 
195
196
197
198
199
200
201
202
203
204
205
def get_meta_data(self, contents_list: list[dict]) -> list[dict]:
    """
    Aggregate metadata from all package JSON files.

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

Retrieve and save package metadata.

Source code in src/get_packages_data.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def get_packages_data(self):
    """
    Retrieve and save package metadata.
    """
    contents_list = self.get_json_data()
    meta_data = self.get_meta_data(contents_list)

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

        self.logger.info(
            "Package meta data successfully saved to %s",
            self.json_file
        )
    else:
        self.logger.info(
            "[DRY RUN] Would write %d entries to %s:\n%s",
            len(meta_data),
            self.json_file,
            json.dumps(meta_data, ensure_ascii=False, indent=2),
        )

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 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
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.config_dict:
            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")
        else:
            self.base_url = os.getenv("BASE_URL")
            self.github_raw_url = os.getenv("GITHUB_RAW_URL")
            self.json_file = os.getenv("JSON_FILE")

    _ALLOWED_URL_PREFIXES = (
        "https://github.com/",
        "https://raw.githubusercontent.com/",
    )

    def _validate_urls(self) -> None:
        """Validate that base_url and github_raw_url are allowed GitHub URLs."""
        for attr_name in ("base_url", "github_raw_url"):
            url = getattr(self, attr_name, None) or ""
            if not any(url.startswith(prefix) for prefix in self._ALLOWED_URL_PREFIXES):
                raise ValueError(
                    f"{attr_name} {url!r} is not an allowed URL — must start with "
                    f"'https://github.com/' or 'https://raw.githubusercontent.com/'"
                )

    def _validate_json_file_path(self) -> None:
        """Validate that json_file is within the project root."""
        safe_root = Path.cwd().resolve()
        target = Path(self.json_file).resolve()
        if not str(target).startswith(str(safe_root)):
            raise ValueError(
                f"json_file path {self.json_file!r} escapes the project root — refusing to write."
            )

    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:
            self._validate_json_file_path()
            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
            )
        else:
            self.logger.info(
                "[DRY RUN] Would write %d entries to %s:\n%s",
                len(meta_data),
                self.json_file,
                json.dumps(meta_data, ensure_ascii=False, indent=2),
            )

    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.
        """
        self._validate_urls()
        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)
        try:
            items = (
                payload["payload"]["codeViewTreeRoute"]["tree"]["items"]
            )
        except KeyError as exc:
            top_keys = list(payload.get("payload", {}).keys())
            raise RuntimeError(
                f"Unexpected GitHub payload structure — missing key {exc}. "
                f"Available keys under 'payload': {top_keys}. "
                "Update the key path in get_json_file_names()."
            ) from exc
        return [
            f"{self.github_raw_url}/{item['path'].split('/')[-1]}"
            for item in items
            if item["path"].endswith(".json")
        ]

    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`).
        - `content_type`: Content type string (`"blog"`, `"youtube"`, or
            `"podcast"`). Defaults to `"blog"` when absent.
        - `rss_feed`: List containing the `rss_feed` URL, or empty list if
            not set.
        - `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 = [url for url in [content.get("rss_feed")] if url]
        content_type = content.get("type", "blog")

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

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

        return {
            "name": name,
            "content_type": content_type,
            "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:
            meta_data.append(self.extract_info(content))
        return meta_data
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). - content_type: Content type string ("blog", "youtube", or "podcast"). Defaults to "blog" when absent. - rss_feed: List containing the rss_feed URL, or empty list if not set. - 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 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
@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`).
    - `content_type`: Content type string (`"blog"`, `"youtube"`, or
        `"podcast"`). Defaults to `"blog"` when absent.
    - `rss_feed`: List containing the `rss_feed` URL, or empty list if
        not set.
    - `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 = [url for url in [content.get("rss_feed")] if url]
    content_type = content.get("type", "blog")

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

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

    return {
        "name": name,
        "content_type": content_type,
        "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 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
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 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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.
    """
    self._validate_urls()
    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)
    try:
        items = (
            payload["payload"]["codeViewTreeRoute"]["tree"]["items"]
        )
    except KeyError as exc:
        top_keys = list(payload.get("payload", {}).keys())
        raise RuntimeError(
            f"Unexpected GitHub payload structure — missing key {exc}. "
            f"Available keys under 'payload': {top_keys}. "
            "Update the key path in get_json_file_names()."
        ) from exc
    return [
        f"{self.github_raw_url}/{item['path'].split('/')[-1]}"
        for item in items
        if item["path"].endswith(".json")
    ]
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 
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
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:
        meta_data.append(self.extract_info(content))
    return meta_data
get_rss_data()

Retrieve and save RSS metadata.

Source code in src/get_rss_data.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
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:
        self._validate_json_file_path()
        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
        )
    else:
        self.logger.info(
            "[DRY RUN] Would write %d entries to %s:\n%s",
            len(meta_data),
            self.json_file,
            json.dumps(meta_data, ensure_ascii=False, indent=2),
        )

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 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
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 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
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 
76
77
78
79
80
81
82
83
84
85
86
87
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 
16
17
18
19
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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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 
16
17
18
19
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 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
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 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
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

        events_file = self.cfg.get("events_file", "metadata/events.json") if self.config_dict else "metadata/events.json"
        with open(events_file, 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"
        self.config_dict.setdefault(
            "events_file", os.getenv("EVENTS_FILE", "metadata/events.json")
        )

    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)
        if len(img_data) > self._BLUESKY_MAX_BLOB_BYTES:
            self.logger.warning(
                "Image still exceeds Bluesky blob limit (%d bytes) after compression — upload may fail.",
                len(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 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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 
360
361
362
363
@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 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
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)
    if len(img_data) > self._BLUESKY_MAX_BLOB_BYTES:
        self.logger.warning(
            "Image still exceeds Bluesky blob limit (%d bytes) after compression — upload may fail.",
            len(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 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
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 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
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 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
@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 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@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 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
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

    events_file = self.cfg.get("events_file", "metadata/events.json") if self.config_dict else "metadata/events.json"
    with open(events_file, 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 
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
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 
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
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 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
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 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
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": bool(os.getenv("GEMINI_API_KEY")),
                "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!")

        if count_post > 0:
            self.update_counter(feeds[next_index]['name'])
        else:
            self.logger.info(
                "No posts made this run — counter left unchanged."
            )

    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
        """
        try:
            with open(self.config_dict["counter"], 'r', encoding='utf-8') as f:
                return f.read()
        except FileNotFoundError:
            return ""

    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 contains "metadata/" as a proper path segment.
        Handles bare names ("counter.txt" → "metadata/counter.txt") while
        leaving already-prefixed paths intact, including relative ones
        ("../metadata/counter.txt" is returned unchanged).
        """
        if not value:
            return value
        segments = value.replace("\\", "/").split("/")
        if prefix.rstrip("/") in segments:
            return value
        return prefix + 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
            parsed = urlsplit(url)
            domain = parsed.netloc
            safe_filename = posixpath.basename(parsed.path) or "image"
            if self.config_dict["platform"] == "bluesky":
                filename = safe_filename
            elif self.config_dict["platform"] == "mastodon":
                filename = safe_filename

            # 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 (always under images/domain/)
            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:
                    tag_clean = tag.replace(' ', '').replace('-', '').lower()[:50]
                    tags += f"#{tag_clean} "

        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, timeout=10)

            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, content_type="blog"
    ):
        """
        Build Mastodon post.
        """
        platform_user_handle = self.check_platform_handle(platform_user_handle)

        emoji = CONTENT_TYPE_EMOJI.get(content_type, "📝")
        post = f'{emoji} "{title}"\n\n' if title else ''

        if self.config_dict.get('gen_ai_support', None):
            try:
                summarized_blog_post = self.summarize_text(entry)
            except Exception as e:  # pylint: disable=broad-except
                self.logger.warning("Gemini summarization failed, falling back to no summary: %s", e)
                summarized_blog_post = ""
            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 not platform_user_handle:
            return ""
        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,
        content_type="blog"
    ):
        """
        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):
            try:
                summarized_blog_post = self.summarize_text(entry) or ''
            except Exception as e:  # pylint: disable=broad-except
                self.logger.warning("Gemini summarization failed, falling back to no summary: %s", e)
                summarized_blog_post = ''

        # 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

        emoji = CONTENT_TYPE_EMOJI.get(content_type, "📝")
        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'{emoji} "{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', '')
        content_type = feed.get('content_type', 'blog')

        if self.config_dict.get('platform', '') == 'mastodon':
            return self.build_post_mastodon(
                title,
                name,
                platform_user_handle,
                tags,
                entry,
                content_type,
            )
        if self.config_dict.get('platform', '') == 'bluesky':
            return self.build_post_bluesky(
                title,
                name,
                platform_user_handle,
                tags,
                entry,
                content_type,
            )
        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'])
            if filename is None:
                return None
            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"""
        if not feed.get('ARCHIVE'):
            return {'link': []}
        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')
        safe_root = Path.cwd().resolve()
        target = Path(archive_path).resolve()
        if not str(target).startswith(str(safe_root)):
            raise ValueError(
                f"Archive path {archive_path!r} escapes the project root — refusing to write."
            )
        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:
                    result = self.send_post(en, feed_config['feed'], client)
                    if result == 'success':
                        feed_config['rss_feed_archive']['link'].append(en['link'])
                        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 
735
736
737
738
739
740
741
742
743
744
@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 
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
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'])
        if filename is None:
            return None
        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 
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
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', '')
    content_type = feed.get('content_type', 'blog')

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

Build post for Bluesky.

Source code in src/promote_blog_post.py 
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
def build_post_bluesky(
    self,
    title,
    name,
    platform_user_handle,
    tags,
    entry,
    content_type="blog"
):
    """
    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):
        try:
            summarized_blog_post = self.summarize_text(entry) or ''
        except Exception as e:  # pylint: disable=broad-except
            self.logger.warning("Gemini summarization failed, falling back to no summary: %s", e)
            summarized_blog_post = ''

    # 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

    emoji = CONTENT_TYPE_EMOJI.get(content_type, "📝")
    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'{emoji} "{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, content_type='blog')

Build Mastodon post.

Source code in src/promote_blog_post.py 
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
def build_post_mastodon(
    self, title, name, platform_user_handle, tags, entry, content_type="blog"
):
    """
    Build Mastodon post.
    """
    platform_user_handle = self.check_platform_handle(platform_user_handle)

    emoji = CONTENT_TYPE_EMOJI.get(content_type, "📝")
    post = f'{emoji} "{title}"\n\n' if title else ''

    if self.config_dict.get('gen_ai_support', None):
        try:
            summarized_blog_post = self.summarize_text(entry)
        except Exception as e:  # pylint: disable=broad-except
            self.logger.warning("Gemini summarization failed, falling back to no summary: %s", e)
            summarized_blog_post = ""
        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 
462
463
464
465
466
467
468
469
470
471
472
@staticmethod
def check_platform_handle(platform_user_handle):
    """
    Check platform handle.
    """
    if not platform_user_handle:
        return ""
    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 
412
413
414
415
416
417
@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 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
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:
                tag_clean = tag.replace(' ', '').replace('-', '').lower()[:50]
                tags += f"#{tag_clean} "

    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 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
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
        parsed = urlsplit(url)
        domain = parsed.netloc
        safe_filename = posixpath.basename(parsed.path) or "image"
        if self.config_dict["platform"] == "bluesky":
            filename = safe_filename
        elif self.config_dict["platform"] == "mastodon":
            filename = safe_filename

        # 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 (always under images/domain/)
        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 
398
399
400
401
402
403
404
405
406
407
408
409
410
@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 
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
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, timeout=10)

        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 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
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": bool(os.getenv("GEMINI_API_KEY")),
            "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 
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
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 
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
@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 
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
@staticmethod
def get_rss_feed_archive(feed):
    """Method to get RSS feed archive content"""
    if not feed.get('ARCHIVE'):
        return {'link': []}
    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 
678
679
680
681
682
@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 
292
293
294
295
296
297
298
299
300
301
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 
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
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 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
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!")

    if count_post > 0:
        self.update_counter(feeds[next_index]['name'])
    else:
        self.logger.info(
            "No posts made this run — counter left unchanged."
        )
promote_blog_post()

Core method to promote blog post

Source code in src/promote_blog_post.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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 
190
191
192
193
194
195
196
197
198
def read_counter_name(self):
    """
    Read counter name from txt file
    """
    try:
        with open(self.config_dict["counter"], 'r', encoding='utf-8') as f:
            return f.read()
    except FileNotFoundError:
        return ""
read_metadata_json()

Read metadata JSON file

Source code in src/promote_blog_post.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
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 
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
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 
607
608
609
610
611
612
613
614
615
616
617
618
619
620
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 
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
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 
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
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 
179
180
181
182
183
184
185
186
187
188
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)

promote_package

Promote community libraries (packages) from awesome-*-creations repos.

PromotePackage

Cycle through package metadata and promote one library per run.

Skips packages that have already been promoted at the same version. Re-promotes when a newer version is detected (PyPI version for PyLadies packages; last_updated timestamp for RLadies packages).

Source code in src/promote_package.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
class PromotePackage():
    """
    Cycle through package metadata and promote one library per run.

    Skips packages that have already been promoted at the same version.
    Re-promotes when a newer version is detected (PyPI version for PyLadies
    packages; ``last_updated`` timestamp for RLadies packages).
    """

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

        self.no_dry_run = no_dry_run
        self.config_dict = config_dict

    def get_config(self):
        """Load configuration from environment or provided dict."""
        if (self.config_dict is None) and (self.no_dry_run):
            self.config_dict = {
                "platform": os.getenv("PLATFORM"),
                "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", "")
                ),
                "archive_file": self._ensure_metadata_prefix(
                    os.getenv("ARCHIVE_FILE", "")
                ),
            }
            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:
            if self.config_dict:
                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", "")
                )
                self.config_dict["archive_file"] = self._ensure_metadata_prefix(
                    self.config_dict.get("archive_file", "")
                )

    @staticmethod
    def _ensure_metadata_prefix(value: str, prefix: str = "metadata/") -> str:
        if not value:
            return value
        # Check whether the path already contains 'metadata' as a proper segment
        # (e.g. "metadata/file.txt" or "../metadata/file.txt"), not just as a
        # substring (e.g. "my-metadata/file.txt" would previously be a false positive).
        segments = value.replace("\\", "/").split("/")
        if prefix.rstrip("/") in segments:
            return value
        return prefix + value

    def promote_package(self):
        """Core method: read metadata, pick next library, post about it."""
        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

        packages = self.read_metadata_json()
        if not packages:
            self.logger.info("No packages found in metadata — nothing to do.")
            return

        counter_name = self.read_counter_name()
        self.process_packages(packages, counter_name, client)

    def process_packages(self, packages, counter_name, client):
        """Find the next package to promote and post it.

        Starting from the package after ``counter_name``, iterates through the
        full list (wrapping around) until it finds one that is due for promotion.
        A package is skipped when:
          - Its version is known and matches the archived version (version-tracked).
          - Its version cannot be determined but it has been promoted before
            (no-version sentinel: archived value is "").
        If every package is already up-to-date, nothing is posted and the
        counter is left unchanged.
        """
        n = len(packages)

        start_index = 0
        for i, pkg in enumerate(packages):
            if pkg.get("name") == counter_name:
                start_index = i
                break

        archive = self.read_archive()

        for offset in range(1, n + 1):
            candidate_index = (start_index + offset) % n
            package = packages[candidate_index]
            package_name = package.get("name", "unknown")

            self.logger.info("Considering package: %s", package_name)

            current_version = self.get_current_version(package)
            archived_value = archive.get(package_name)  # None = not yet promoted

            # Decide whether to skip this package.
            # Case A: version is deterministic and unchanged.
            if current_version is not None and current_version == archived_value:
                self.logger.info(
                    "Skipping %s — already promoted at version %s.",
                    package_name,
                    current_version,
                )
                continue
            # Case B: version cannot be determined but package was promoted before
            # (sentinel "" stored in archive to signal "promoted, no version info").
            if current_version is None and archived_value is not None:
                self.logger.info(
                    "Skipping %s — already promoted (no version info available).",
                    package_name,
                )
                continue

            # ── Found a package to promote ──────────────────────────────────
            self.logger.info("Promoting package: %s", package_name)
            if current_version:
                self.logger.info("  Version: %s", current_version)

            if self.no_dry_run:
                result = self.send_post(package, client)
                if result == "success":
                    self.logger.info(
                        "Successfully promoted %s.", package_name
                    )
                    # Store version string, or "" as a sentinel when unavailable.
                    archive[package_name] = (
                        current_version if current_version is not None else ""
                    )
                    self.write_archive(archive)
                    self.update_counter(package_name)
                else:
                    self.logger.warning(
                        "Post failed for %s — counter not advanced, will retry next run.",
                        package_name
                    )
            else:
                # Dry run: preview only, do not touch counter or archive.
                platform = self.config_dict.get("platform", "")
                if platform == "mastodon":
                    post_text = self.build_post_mastodon(package)
                elif platform == "bluesky":
                    post_text = self.build_post_bluesky(package).build_text()
                else:
                    post_text = ""

                self.logger.info(
                    "[DRY RUN] Would promote: %s (%s)\n%s",
                    package_name,
                    package.get("repo_url", ""),
                    post_text,
                )

            return

        # All packages are already up-to-date — nothing to post this run.
        self.logger.info(
            "All packages already promoted at their current version — nothing to post."
        )

    # ------------------------------------------------------------------
    # Archive helpers
    # ------------------------------------------------------------------

    def read_archive(self) -> dict:
        """Read the promotion archive. Returns {} if not found."""
        archive_file = self.config_dict.get("archive_file", "")
        if not archive_file:
            self.logger.warning(
                "ARCHIVE_FILE not configured — version tracking disabled. "
                "Packages may be re-promoted every cycle."
            )
            return {}
        try:
            with open(archive_file, "r", encoding="utf-8") as f:
                return json.load(f)
        except (FileNotFoundError, json.JSONDecodeError):
            return {}

    def write_archive(self, archive: dict):
        """Persist the promotion archive."""
        archive_file = self.config_dict.get("archive_file", "")
        if not archive_file:
            return
        with open(archive_file, "w", encoding="utf-8") as f:
            json.dump(archive, f, ensure_ascii=False, indent=2)

    # ------------------------------------------------------------------
    # Version helpers
    # ------------------------------------------------------------------

    def get_pypi_version(self, pypi_url: str) -> str | None:
        """Fetch the latest version of a package from the PyPI JSON API."""
        if not pypi_url:
            return None
        # Extract the package name from the canonical PyPI URL pattern
        # "/project/<name>/", guarding against version segments like "/project/foo/1.0/".
        match = re.search(r"/project/([^/]+)", pypi_url)
        package_name = match.group(1) if match else pypi_url.rstrip("/").split("/")[-1]
        url = f"https://pypi.org/pypi/{package_name}/json"
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                return response.json().get("info", {}).get("version")
        except requests.RequestException as e:
            self.logger.info("Could not fetch PyPI version for %s: %s", package_name, e)
        return None

    def get_current_version(self, package: dict) -> str | None:
        """
        Return the current version string for a package.

        - PyLadies packages: latest version from PyPI (via pypi_url).
        - RLadies packages: last_updated timestamp from metadata.
        """
        client_name = self.config_dict.get("client_name", "")
        if client_name == "pyladies_bot":
            return self.get_pypi_version(package.get("pypi_url", ""))
        if client_name == "rladies_bot":
            last_updated = package.get("last_updated", "")
            return last_updated if last_updated else None
        return None

    # ------------------------------------------------------------------
    # Counter helpers
    # ------------------------------------------------------------------

    def update_counter(self, counter_name):
        """Write the name of the just-promoted library to the counter file."""
        with open(
            self.config_dict["counter"], "w", encoding="utf-8"
        ) as txt_file:
            txt_file.write(counter_name)

    def read_counter_name(self) -> str:
        """Read the last-promoted library name from the counter file."""
        try:
            with open(
                self.config_dict["counter"], "r", encoding="utf-8"
            ) as f:
                return f.read().strip()
        except FileNotFoundError:
            return ""

    def read_metadata_json(self) -> list[dict]:
        """Read package metadata JSON. Returns [] if the file doesn't exist yet."""
        try:
            with open(self.config_dict["json_file"], "rb") as fp:
                self.logger.info("=============================================")
                packages = json.load(fp)
                self.logger.info("Package meta data successfully loaded.")
                self.logger.info("=============================================")
                return packages
        except FileNotFoundError:
            self.logger.warning(
                "Metadata file %s not found — run the packages data workflow first.",
                self.config_dict["json_file"],
            )
            return []

    def define_tags(self) -> str:
        """Return community-specific hashtags."""
        client_name = self.config_dict.get("client_name", "")
        if client_name == "pyladies_bot":
            return "#pyladies #python #opensource "
        if client_name == "rladies_bot":
            return "#rladies #rstats #opensource "
        return "#opensource "

    def get_bluesky_did(self, handle: str):
        """Resolve a Bluesky handle to a DID."""
        url = (
            f"https://bsky.social/xrpc/com.atproto.identity.resolveHandle?"
            f"handle={handle.lstrip('@')}"
        )
        try:
            response = requests.get(url, timeout=10)
            if response.status_code == 200:
                return response.json().get("did")
        except requests.RequestException as e:
            self.logger.info("Could not resolve Bluesky DID: %s", e)
        return None

    @staticmethod
    def _check_handle(handle: str) -> str:
        if handle and len(handle) > 1 and not handle.startswith("@"):
            return f"@{handle}"
        return handle

    @staticmethod
    def _join_names(parts: list[str]) -> str:
        """Join a list of names as 'A', 'A & B', or 'A, B & C'."""
        if not parts:
            return ""
        if len(parts) == 1:
            return parts[0]
        return ", ".join(parts[:-1]) + " & " + parts[-1]

    def _is_community_member(self, contributor: dict) -> bool:
        """Return True if this contributor is a community member.

        R-Ladies: community membership is marked by a directory_id in the
        contributor entry (present only for R-Ladies members).
        PyLadies: all listed maintainers are treated as community members
        (no directory_id equivalent exists in the PyLadies schema).
        """
        client_name = self.config_dict.get("client_name", "")
        if "rladies" in client_name.lower():
            return bool(contributor.get("directory_id"))
        return True

    @staticmethod
    def _fit_description(description: str, budget: int) -> str:
        """Truncate description to fit within budget characters, adding '…'."""
        if budget <= 0:
            return ""
        if len(description) <= budget:
            return description
        return description[:budget - 1] + "…"

    def build_post_mastodon(self, package: dict) -> str:
        """Build a Mastodon post for the given package."""
        mastodon_max_chars = 500
        name = package.get("name", "")
        description = package.get("description", "")
        repo_url = package.get("repo_url", "")
        contributors = package.get("contributors", [])
        tags = self.define_tags()

        contributor_line = ""
        if contributors:
            parts = []
            for c in contributors:
                entry = c.get("name", "")
                if self._is_community_member(c):
                    handle = self._check_handle(c.get("mastodon", ""))
                    if handle:
                        entry += f" ({handle})"
                parts.append(entry)
            contributor_line = f"👤 {self._join_names(parts)}\n\n"

        suffix = f"🔗 {repo_url}\n\n{tags}"
        prefix = f"📦 {name}\n\n"
        # "\n\n" separates description from what follows
        overhead = len(prefix) + len(contributor_line) + len(suffix) + 2
        description = self._fit_description(description, mastodon_max_chars - overhead)

        post = prefix
        if description:
            post += f"{description}\n\n"
        post += contributor_line + suffix

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

        return post

    def build_post_bluesky(self, package: dict):
        """Build a Bluesky TextBuilder post for the given package."""
        bluesky_max_graphemes = 300
        name = package.get("name", "")
        description = package.get("description", "")
        repo_url = package.get("repo_url", "")
        contributors = package.get("contributors", [])
        tags = self.define_tags()
        tag_list = [t.strip() for t in tags.split("#") if t.strip()]

        # Resolve DIDs only for community members, once before the tag-trimming loop.
        resolved = []
        for c in contributors:
            if self._is_community_member(c):
                handle = self._check_handle(c.get("bluesky", ""))
                did = self.get_bluesky_did(handle) if handle else None
            else:
                handle, did = "", None
            resolved.append((c.get("name", ""), handle, did))

        def _build(desc, tag_subset):
            tb = client_utils.TextBuilder()
            tb.text(f"📦 {name}\n\n")
            if desc:
                tb.text(desc)
                tb.text("\n\n")
            if resolved:
                tb.text("👤 ")
                for i, (cname, handle, did) in enumerate(resolved):
                    if i > 0:
                        tb.text(", " if i < len(resolved) - 1 else " & ")
                    tb.text(cname)
                    if handle and did:
                        tb.mention(f" ({handle})", did)
                    elif handle:
                        tb.text(f" ({handle})")
                tb.text("\n\n")
            tb.text("🔗 ")
            tb.link(repo_url, repo_url)
            tb.text("\n\n")
            for tag_clean in tag_subset:
                tb.tag(f"#{tag_clean} ", tag_clean)
            return tb

        # First trim tags, then trim description if still over limit.
        for count in range(len(tag_list), -1, -1):
            text_builder = _build(description, tag_list[:count])
            if len(text_builder.build_text()) <= bluesky_max_graphemes:
                return text_builder

        # Tags exhausted — trim description to fit.
        no_desc_len = len(_build("", []).build_text())
        desc_budget = bluesky_max_graphemes - no_desc_len - 2  # 2 for "\n\n"
        trimmed = self._fit_description(description, desc_budget)
        return _build(trimmed, [])

    def send_post(self, package: dict, client) -> str:
        """Build and send the post for a package."""
        self.logger.info(
            "Preparing post on %s (%s) ...",
            self.config_dict["client_name"],
            self.config_dict["platform"],
        )

        platform = self.config_dict.get("platform", "")

        if platform == "mastodon":
            post_txt = self.build_post_mastodon(package)
            try:
                client.status_post(post_txt)
                self.logger.info("Posted 🎉")
                return "success"
            except Exception as e:
                self.logger.exception("Post failed: %s", e)
                return "failed"

        if platform == "bluesky":
            post_txt = self.build_post_bluesky(package)
            try:
                client.send_post(text=post_txt)
                self.logger.info("Posted 🎉")
                return "success"
            except Exception as e:
                self.logger.exception("Post failed: %s", e)
                return "failed"

        return "failed"
build_post_bluesky(package)

Build a Bluesky TextBuilder post for the given package.

Source code in src/promote_package.py 
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
def build_post_bluesky(self, package: dict):
    """Build a Bluesky TextBuilder post for the given package."""
    bluesky_max_graphemes = 300
    name = package.get("name", "")
    description = package.get("description", "")
    repo_url = package.get("repo_url", "")
    contributors = package.get("contributors", [])
    tags = self.define_tags()
    tag_list = [t.strip() for t in tags.split("#") if t.strip()]

    # Resolve DIDs only for community members, once before the tag-trimming loop.
    resolved = []
    for c in contributors:
        if self._is_community_member(c):
            handle = self._check_handle(c.get("bluesky", ""))
            did = self.get_bluesky_did(handle) if handle else None
        else:
            handle, did = "", None
        resolved.append((c.get("name", ""), handle, did))

    def _build(desc, tag_subset):
        tb = client_utils.TextBuilder()
        tb.text(f"📦 {name}\n\n")
        if desc:
            tb.text(desc)
            tb.text("\n\n")
        if resolved:
            tb.text("👤 ")
            for i, (cname, handle, did) in enumerate(resolved):
                if i > 0:
                    tb.text(", " if i < len(resolved) - 1 else " & ")
                tb.text(cname)
                if handle and did:
                    tb.mention(f" ({handle})", did)
                elif handle:
                    tb.text(f" ({handle})")
            tb.text("\n\n")
        tb.text("🔗 ")
        tb.link(repo_url, repo_url)
        tb.text("\n\n")
        for tag_clean in tag_subset:
            tb.tag(f"#{tag_clean} ", tag_clean)
        return tb

    # First trim tags, then trim description if still over limit.
    for count in range(len(tag_list), -1, -1):
        text_builder = _build(description, tag_list[:count])
        if len(text_builder.build_text()) <= bluesky_max_graphemes:
            return text_builder

    # Tags exhausted — trim description to fit.
    no_desc_len = len(_build("", []).build_text())
    desc_budget = bluesky_max_graphemes - no_desc_len - 2  # 2 for "\n\n"
    trimmed = self._fit_description(description, desc_budget)
    return _build(trimmed, [])
build_post_mastodon(package)

Build a Mastodon post for the given package.

Source code in src/promote_package.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def build_post_mastodon(self, package: dict) -> str:
    """Build a Mastodon post for the given package."""
    mastodon_max_chars = 500
    name = package.get("name", "")
    description = package.get("description", "")
    repo_url = package.get("repo_url", "")
    contributors = package.get("contributors", [])
    tags = self.define_tags()

    contributor_line = ""
    if contributors:
        parts = []
        for c in contributors:
            entry = c.get("name", "")
            if self._is_community_member(c):
                handle = self._check_handle(c.get("mastodon", ""))
                if handle:
                    entry += f" ({handle})"
            parts.append(entry)
        contributor_line = f"👤 {self._join_names(parts)}\n\n"

    suffix = f"🔗 {repo_url}\n\n{tags}"
    prefix = f"📦 {name}\n\n"
    # "\n\n" separates description from what follows
    overhead = len(prefix) + len(contributor_line) + len(suffix) + 2
    description = self._fit_description(description, mastodon_max_chars - overhead)

    post = prefix
    if description:
        post += f"{description}\n\n"
    post += contributor_line + suffix

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

    return post
define_tags()

Return community-specific hashtags.

Source code in src/promote_package.py 
312
313
314
315
316
317
318
319
def define_tags(self) -> str:
    """Return community-specific hashtags."""
    client_name = self.config_dict.get("client_name", "")
    if client_name == "pyladies_bot":
        return "#pyladies #python #opensource "
    if client_name == "rladies_bot":
        return "#rladies #rstats #opensource "
    return "#opensource "
get_bluesky_did(handle)

Resolve a Bluesky handle to a DID.

Source code in src/promote_package.py 
321
322
323
324
325
326
327
328
329
330
331
332
333
def get_bluesky_did(self, handle: str):
    """Resolve a Bluesky handle to a DID."""
    url = (
        f"https://bsky.social/xrpc/com.atproto.identity.resolveHandle?"
        f"handle={handle.lstrip('@')}"
    )
    try:
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            return response.json().get("did")
    except requests.RequestException as e:
        self.logger.info("Could not resolve Bluesky DID: %s", e)
    return None
get_config()

Load configuration from environment or provided dict.

Source code in src/promote_package.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_config(self):
    """Load configuration from environment or provided dict."""
    if (self.config_dict is None) and (self.no_dry_run):
        self.config_dict = {
            "platform": os.getenv("PLATFORM"),
            "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", "")
            ),
            "archive_file": self._ensure_metadata_prefix(
                os.getenv("ARCHIVE_FILE", "")
            ),
        }
        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:
        if self.config_dict:
            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", "")
            )
            self.config_dict["archive_file"] = self._ensure_metadata_prefix(
                self.config_dict.get("archive_file", "")
            )
get_current_version(package)

Return the current version string for a package.

  • PyLadies packages: latest version from PyPI (via pypi_url).
  • RLadies packages: last_updated timestamp from metadata.
Source code in src/promote_package.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def get_current_version(self, package: dict) -> str | None:
    """
    Return the current version string for a package.

    - PyLadies packages: latest version from PyPI (via pypi_url).
    - RLadies packages: last_updated timestamp from metadata.
    """
    client_name = self.config_dict.get("client_name", "")
    if client_name == "pyladies_bot":
        return self.get_pypi_version(package.get("pypi_url", ""))
    if client_name == "rladies_bot":
        last_updated = package.get("last_updated", "")
        return last_updated if last_updated else None
    return None
get_pypi_version(pypi_url)

Fetch the latest version of a package from the PyPI JSON API.

Source code in src/promote_package.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def get_pypi_version(self, pypi_url: str) -> str | None:
    """Fetch the latest version of a package from the PyPI JSON API."""
    if not pypi_url:
        return None
    # Extract the package name from the canonical PyPI URL pattern
    # "/project/<name>/", guarding against version segments like "/project/foo/1.0/".
    match = re.search(r"/project/([^/]+)", pypi_url)
    package_name = match.group(1) if match else pypi_url.rstrip("/").split("/")[-1]
    url = f"https://pypi.org/pypi/{package_name}/json"
    try:
        response = requests.get(url, timeout=10)
        if response.status_code == 200:
            return response.json().get("info", {}).get("version")
    except requests.RequestException as e:
        self.logger.info("Could not fetch PyPI version for %s: %s", package_name, e)
    return None
process_packages(packages, counter_name, client)

Find the next package to promote and post it.

Starting from the package after counter_name, iterates through the full list (wrapping around) until it finds one that is due for promotion. A package is skipped when: - Its version is known and matches the archived version (version-tracked). - Its version cannot be determined but it has been promoted before (no-version sentinel: archived value is ""). If every package is already up-to-date, nothing is posted and the counter is left unchanged.

Source code in src/promote_package.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def process_packages(self, packages, counter_name, client):
    """Find the next package to promote and post it.

    Starting from the package after ``counter_name``, iterates through the
    full list (wrapping around) until it finds one that is due for promotion.
    A package is skipped when:
      - Its version is known and matches the archived version (version-tracked).
      - Its version cannot be determined but it has been promoted before
        (no-version sentinel: archived value is "").
    If every package is already up-to-date, nothing is posted and the
    counter is left unchanged.
    """
    n = len(packages)

    start_index = 0
    for i, pkg in enumerate(packages):
        if pkg.get("name") == counter_name:
            start_index = i
            break

    archive = self.read_archive()

    for offset in range(1, n + 1):
        candidate_index = (start_index + offset) % n
        package = packages[candidate_index]
        package_name = package.get("name", "unknown")

        self.logger.info("Considering package: %s", package_name)

        current_version = self.get_current_version(package)
        archived_value = archive.get(package_name)  # None = not yet promoted

        # Decide whether to skip this package.
        # Case A: version is deterministic and unchanged.
        if current_version is not None and current_version == archived_value:
            self.logger.info(
                "Skipping %s — already promoted at version %s.",
                package_name,
                current_version,
            )
            continue
        # Case B: version cannot be determined but package was promoted before
        # (sentinel "" stored in archive to signal "promoted, no version info").
        if current_version is None and archived_value is not None:
            self.logger.info(
                "Skipping %s — already promoted (no version info available).",
                package_name,
            )
            continue

        # ── Found a package to promote ──────────────────────────────────
        self.logger.info("Promoting package: %s", package_name)
        if current_version:
            self.logger.info("  Version: %s", current_version)

        if self.no_dry_run:
            result = self.send_post(package, client)
            if result == "success":
                self.logger.info(
                    "Successfully promoted %s.", package_name
                )
                # Store version string, or "" as a sentinel when unavailable.
                archive[package_name] = (
                    current_version if current_version is not None else ""
                )
                self.write_archive(archive)
                self.update_counter(package_name)
            else:
                self.logger.warning(
                    "Post failed for %s — counter not advanced, will retry next run.",
                    package_name
                )
        else:
            # Dry run: preview only, do not touch counter or archive.
            platform = self.config_dict.get("platform", "")
            if platform == "mastodon":
                post_text = self.build_post_mastodon(package)
            elif platform == "bluesky":
                post_text = self.build_post_bluesky(package).build_text()
            else:
                post_text = ""

            self.logger.info(
                "[DRY RUN] Would promote: %s (%s)\n%s",
                package_name,
                package.get("repo_url", ""),
                post_text,
            )

        return

    # All packages are already up-to-date — nothing to post this run.
    self.logger.info(
        "All packages already promoted at their current version — nothing to post."
    )
promote_package()

Core method: read metadata, pick next library, post about it.

Source code in src/promote_package.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def promote_package(self):
    """Core method: read metadata, pick next library, post about it."""
    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

    packages = self.read_metadata_json()
    if not packages:
        self.logger.info("No packages found in metadata — nothing to do.")
        return

    counter_name = self.read_counter_name()
    self.process_packages(packages, counter_name, client)
read_archive()

Read the promotion archive. Returns {} if not found.

Source code in src/promote_package.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def read_archive(self) -> dict:
    """Read the promotion archive. Returns {} if not found."""
    archive_file = self.config_dict.get("archive_file", "")
    if not archive_file:
        self.logger.warning(
            "ARCHIVE_FILE not configured — version tracking disabled. "
            "Packages may be re-promoted every cycle."
        )
        return {}
    try:
        with open(archive_file, "r", encoding="utf-8") as f:
            return json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        return {}
read_counter_name()

Read the last-promoted library name from the counter file.

Source code in src/promote_package.py 
286
287
288
289
290
291
292
293
294
def read_counter_name(self) -> str:
    """Read the last-promoted library name from the counter file."""
    try:
        with open(
            self.config_dict["counter"], "r", encoding="utf-8"
        ) as f:
            return f.read().strip()
    except FileNotFoundError:
        return ""
read_metadata_json()

Read package metadata JSON. Returns [] if the file doesn't exist yet.

Source code in src/promote_package.py 
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def read_metadata_json(self) -> list[dict]:
    """Read package metadata JSON. Returns [] if the file doesn't exist yet."""
    try:
        with open(self.config_dict["json_file"], "rb") as fp:
            self.logger.info("=============================================")
            packages = json.load(fp)
            self.logger.info("Package meta data successfully loaded.")
            self.logger.info("=============================================")
            return packages
    except FileNotFoundError:
        self.logger.warning(
            "Metadata file %s not found — run the packages data workflow first.",
            self.config_dict["json_file"],
        )
        return []
send_post(package, client)

Build and send the post for a package.

Source code in src/promote_package.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
def send_post(self, package: dict, client) -> str:
    """Build and send the post for a package."""
    self.logger.info(
        "Preparing post on %s (%s) ...",
        self.config_dict["client_name"],
        self.config_dict["platform"],
    )

    platform = self.config_dict.get("platform", "")

    if platform == "mastodon":
        post_txt = self.build_post_mastodon(package)
        try:
            client.status_post(post_txt)
            self.logger.info("Posted 🎉")
            return "success"
        except Exception as e:
            self.logger.exception("Post failed: %s", e)
            return "failed"

    if platform == "bluesky":
        post_txt = self.build_post_bluesky(package)
        try:
            client.send_post(text=post_txt)
            self.logger.info("Posted 🎉")
            return "success"
        except Exception as e:
            self.logger.exception("Post failed: %s", e)
            return "failed"

    return "failed"
update_counter(counter_name)

Write the name of the just-promoted library to the counter file.

Source code in src/promote_package.py 
279
280
281
282
283
284
def update_counter(self, counter_name):
    """Write the name of the just-promoted library to the counter file."""
    with open(
        self.config_dict["counter"], "w", encoding="utf-8"
    ) as txt_file:
        txt_file.write(counter_name)
write_archive(archive)

Persist the promotion archive.

Source code in src/promote_package.py 
231
232
233
234
235
236
237
def write_archive(self, archive: dict):
    """Persist the promotion archive."""
    archive_file = self.config_dict.get("archive_file", "")
    if not archive_file:
        return
    with open(archive_file, "w", encoding="utf-8") as f:
        json.dump(archive, f, ensure_ascii=False, indent=2)