Consolidate Mastodon into shared prod workflow

.github/workflows/prod.yml

@@ -36,4 +36,5 @@ jobs:
           INSTAGRAM_PASSWORD: ${{ secrets.INSTAGRAM_PASSWORD }}
           BLUESKY_HANDLE: ${{ secrets.BLUESKY_HANDLE }}
           BLUESKY_PASSWORD: ${{ secrets.BLUESKY_PASSWORD }}
+          MASTODON_TOKEN: ${{ secrets.MASTODON_TOKEN }}
         run: python ./main.py

README.md

@@ -13,16 +13,23 @@ This Project runs on github actions and runs periodically.
 ## Set up your environment variables
 
 Required:
-- `RESCUEGROUPS_API_KEY`
+- `CUTEPETSBOSTON_RESCUEGROUPS_API_KEY`
 
 Optional for Instagram posting:
-- `INSTAGRAM_USERNAME`
+- `INSTAGRAM_HANDLE`
 - `INSTAGRAM_PASSWORD`
 
 Optional for Bluesky posting:
 - `BLUESKY_HANDLE` (or `BLUESKY_TEST_HANDLE`)
 - `BLUESKY_PASSWORD` (or `BLUESKY_TEST_PASSWORD`)
 
+Optional for Mastodon posting:
+- `MASTODON_TOKEN` or `MASTODON_TEST_TOKEN`
+- `MASTODON_API_BASE_URL` (defaults to `https://mastodon.social`)
+
+Optional platform selection:
+- `POSTER_PLATFORMS` to limit posting to specific platforms, for example `mastodon` or `bluesky,mastodon`
+
 ## File organization
 
 - `main.py`: orchestrates fetching pets and publishing posts.
@@ -35,6 +42,10 @@ Optional for Bluesky posting:
 
     python main.py
 
+To run only the Mastodon poster locally or in GitHub Actions:
+
+    POSTER_PLATFORMS=mastodon python main.py
+
 # History
 
 This project was originally started by [Becky Boone](https://github.com/boonrs) and [Drew](https://github.com/drewrwilson) during their fellowship at Code for America in 2014.

main.py

@@ -1,3 +1,4 @@
+import os
 import random
 import argparse
 
@@ -15,18 +16,26 @@ def main():
 
 
 def create_posters(debug=False):
-    from social_posters import PosterDebug
-
     if debug:
+        from social_posters.debug import PosterDebug
+
         return [PosterDebug()]
 
-    from social_posters.instagram import PosterInstagram
-    from social_posters.bluesky import PosterBluesky
+    requested_platforms = _requested_platforms()
+    poster_factories = {
+        "bluesky": _load_bluesky_poster,
+        "instagram": _load_instagram_poster,
+        "mastodon": _load_mastodon_poster,
+    }
+
+    if requested_platforms:
+        return [
+            poster_factories[platform_name]()
+            for platform_name in poster_factories
+            if platform_name in requested_platforms
+        ]
 
-    posters = []
-    posters.append(PosterBluesky())
-    posters.append(PosterInstagram())
-    return posters
+    return [factory() for factory in poster_factories.values()]
 
 
 def create_sources(debug=False):
@@ -80,5 +89,35 @@ def pick_pet(pets):
     return random.choice(eligible)
 
 
+def _requested_platforms():
+    raw_value = os.environ.get("POSTER_PLATFORMS", "")
+    if not raw_value.strip():
+        return set()
+
+    return {
+        platform.strip().lower()
+        for platform in raw_value.split(",")
+        if platform.strip()
+    }
+
+
+def _load_bluesky_poster():
+    from social_posters.bluesky import PosterBluesky
+
+    return PosterBluesky()
+
+
+def _load_instagram_poster():
+    from social_posters.instagram import PosterInstagram
+
+    return PosterInstagram()
+
+
+def _load_mastodon_poster():
+    from social_posters.mastodon import PosterMastodon
+
+    return PosterMastodon()
+
+
 if __name__ == "__main__":
     main()

manual_testing/mastodon_manual_test.py

@@ -0,0 +1,51 @@
+#curl -X POST "https://mastodon.social/api/v1/statuses" \
+#  -H "Authorization: Bearer h_o6jBz37M5322Mb8a1PYNTA9ALjfKL15_XMY2dYwAs" \
+#  -H "Content-Type: application/json" \
+#  -d '{"status": "Hello from my dev app! 🚀"}'
+
+# RESET TOKEN LATER!!!
+
+import sys
+import os
+import random
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+from adoption_sources import SourceRescueGroups
+from social_posters.mastodon import PosterMastodon
+
+def main():
+    poster = PosterMastodon()
+
+    if not poster.authenticate():
+        print("Authentication failed!")
+        exit(1)
+
+    print("Authenticated to Mastodon!")
+
+    source = SourceRescueGroups()
+    pets = list(source.fetch_pets())
+    print(f"Fetched {len(pets)} pets")
+
+    with_images = [p for p in pets if p.image_url]
+    if not with_images:
+        print("No pets with images found.")
+        exit(1)
+
+    pet = random.choice(with_images)
+    print(f"Selected: {pet.name}")
+
+    post = poster.format_post(pet)
+    print(f"\nPost preview:\n{post.text}")
+    print(f"\nTags: {post.tags}")
+
+    result = poster.publish(post)
+
+    if result.success:
+        print(f"\nPosted successfully! URL: {result.post_url}")
+    else:
+        print(f"\nPost failed: {result.error_message}")
+
+
+if __name__ == "__main__":
+    main()

manual_testing/mastodon_simple_test.py

@@ -0,0 +1,12 @@
+from mastodon import Mastodon
+import os
+from datetime import datetime
+
+client = Mastodon(
+    access_token=os.environ.get("MASTODON_TOKEN") or os.environ["MASTODON_TEST_TOKEN"],
+    api_base_url=os.environ.get("MASTODON_API_BASE_URL", "https://mastodon.social"),
+)
+
+client.account_verify_credentials()
+client.status_post(f"Simple Test at {datetime.now()}")
+print("Success")

requirements.txt

@@ -1,5 +1,52 @@
+anyio==4.12.1
+api-display-purposes==0.0.3
+attrs==25.4.0
+beautifulsoup4==4.14.3
+blurhash==1.1.5
+certifi==2026.2.25
+chardet==3.0.4
+charset-normalizer==3.4.4
 instagrapi>=2.3.0
 clarifai==2.6.2
+configparser==3.8.1
+decorator==4.4.2
+EasyProcess==1.1
 emoji==1.7.0
-requests>=2.28.0
-setuptools>=70.0
+future==1.0.0
+googleapis-common-protos==1.72.0
+grpcio==1.78.0
+h11==0.16.0
+httpcore==1.0.9
+httpx==0.28.1
+idna==2.10
+instapy==0.6.16
+jsonschema==2.6.0
+Mastodon.py==2.1.4
+MeaningCloud-python==2.0.0
+outcome==1.3.0.post0
+plyer==2.1.0
+protobuf==3.20.3
+PySocks==1.7.1
+python-dateutil==2.9.0.post0
+python-magic==0.4.27
+python-telegram-bot==22.6
+PyVirtualDisplay==3.0
+PyYAML==6.0.3
+regex==2026.2.28
+requests==2.32.5
+selenium==4.41.0
+semantic-version==2.10.0
+setuptools==82.0.0
+setuptools-rust==1.12.0
+six==1.17.0
+sniffio==1.3.1
+sortedcontainers==2.4.0
+soupsieve==2.8.3
+tqdm==4.67.3
+trio==0.33.0
+trio-websocket==0.12.2
+typing_extensions==4.15.0
+urllib3==2.6.3
+webdriverdownloader==1.1.0.4
+websocket-client==1.9.0
+wsproto==1.3.2

social_posters/__init__.py

@@ -1,5 +1,23 @@
 """Social media poster implementations implementing the SocialPoster interface."""
 
-from social_posters.debug import PosterDebug
+__all__ = ["PosterBluesky", "PosterDebug", "PosterMastodon", "PosterInstagram"]
 
-__all__ = ["PosterBluesky", "PosterDebug", "PosterInstagram"]
+
+def __getattr__(name):
+    if name == "PosterBluesky":
+        from social_posters.bluesky import PosterBluesky
+
+        return PosterBluesky
+    if name == "PosterDebug":
+        from social_posters.debug import PosterDebug
+
+        return PosterDebug
+    if name == "PosterInstagram":
+        from social_posters.instagram import PosterInstagram
+
+        return PosterInstagram
+    if name == "PosterMastodon":
+        from social_posters.mastodon import PosterMastodon
+
+        return PosterMastodon
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

social_posters/debug.py

@@ -22,6 +22,7 @@ def publish(self, post: Post) -> PostResult:
             f"Link: {post.link}\n"
             f"Alt: {post.alt_text}\n"
             f"Tags: {post.tags}\n"
+            f"Url: {post.link}\n"
         )
         if self.stream:
             self.stream.write(output)

social_posters/mastodon.py

@@ -0,0 +1,111 @@
+import os
+from urllib.parse import urlparse
+import tempfile
+
+import requests
+from mastodon import Mastodon
+
+from abstractions import Post, PostResult, SocialPoster
+
+MASTODON_CHARACTER_LIMIT = 500
+ELLIPSIS = "..."
+
+
+class PosterMastodon(SocialPoster):
+    def __init__(self):
+        raw_token = os.environ.get("MASTODON_TOKEN")
+        self.token = raw_token.strip() if raw_token else None
+        self.api_base_url = "https://mastodon.social"
+        self._session = None
+        self._is_available = bool(self.token)
+        self._auth_error = None
+
+    @property
+    def platform_name(self) -> str:
+        return "Mastodon"
+
+    def authenticate(self) -> bool:
+        try:
+            self._session = Mastodon(
+                access_token=self.token,
+                api_base_url=self.api_base_url,
+            )
+            self._session.account_verify_credentials()
+            self._auth_error = None
+            return True
+        except Exception as exc:
+            self._session = None
+            self._auth_error = f"{type(exc).__name__}: {exc}"
+            return False
+
+    def publish(self, post: Post) -> PostResult:
+        if not self._is_available:
+            return PostResult(
+                success=False,
+                error_message="Mastodon credentials not available.",
+            )
+
+        if not post.image_url:
+            return PostResult(
+                success=False,
+                error_message="Mastodon posts require an image URL.",
+            )
+
+        if not self._session and not self.authenticate():
+            return PostResult(
+                success=False,
+                error_message=(
+                    "Mastodon authentication failed."
+                    if not self._auth_error
+                    else f"Mastodon authentication failed: {self._auth_error}"
+                ),
+            )
+
+        image_path = None
+        try:
+            image_path = self._download_image(post.image_url)
+            media = self._session.media_post(
+                image_path,
+                description=post.alt_text or "Photo of an adoptable pet",
+            )
+            status = self._session.status_post(
+                self._format_caption(post),
+                media_ids=[media["id"]],
+            )
+            return PostResult(
+                success=True,
+                post_id=str(status["id"]),
+                post_url=status.get("url"),
+            )
+        except Exception as exc:
+            return PostResult(success=False, error_message=str(exc))
+        finally:
+            self._session = None
+            if image_path and os.path.exists(image_path):
+                os.unlink(image_path)
+
+    def _format_caption(self, post: Post) -> str:
+        tags = " ".join(f"#{tag}" for tag in post.tags if tag)
+        suffix = f"\n\n{tags}" if tags else ""
+        available_text_length = MASTODON_CHARACTER_LIMIT - len(suffix)
+
+        if available_text_length <= len(ELLIPSIS):
+            return (suffix[-MASTODON_CHARACTER_LIMIT:]).strip()
+
+        caption_text = post.text.strip()
+        if len(caption_text) > available_text_length:
+            caption_text = caption_text[: available_text_length - len(ELLIPSIS)].rstrip()
+            caption_text = f"{caption_text}{ELLIPSIS}"
+
+        return f"{caption_text}{suffix}"
+
+    def _download_image(self, image_url: str) -> str:
+        parsed = urlparse(image_url)
+        ext = os.path.splitext(parsed.path)[1] or ".jpg"
+        with tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
+            response = requests.get(image_url, stream=True, timeout=20)
+            response.raise_for_status()
+            for chunk in response.iter_content(chunk_size=1024 * 128):
+                if chunk:
+                    tmp.write(chunk)
+            return tmp.name