migrate Instagram integration to Graph API v26 with new test account

- switch host to graph.instagram.com (was graph.facebook.com) and bump v21.0 -> v26.0
- use header-based auth for metrics collector, matching the poster
- poll media container status instead of a blind sleep before publishing
- derive post_url from the authenticated account instead of the old hardcoded handle
- add INSTAGRAM_TEST_* secrets so dev CI posts to the test account, not prod
- correct .env.example/README var names; add instagram_manual_test.py

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: 2 people

.env.example

@@ -1,6 +1,6 @@
 CUTEPETSBOSTON_RESCUEGROUPS_API_KEY=
-INSTAGRAM_USERNAME=
-INSTAGRAM_PASSWORD=
+INSTAGRAM_BUSINESS_ACCOUNT_ID=
+INSTAGRAM_PAGE_ACCESS_TOKEN=
 BLUESKY_HANDLE=
 BLUESKY_PASSWORD=
 MASTODON_TOKEN=

.github/workflows/dev.yml

@@ -70,8 +70,8 @@ jobs:
       - name: Call RescueGroups API
         env:
           CUTEPETSBOSTON_RESCUEGROUPS_API_KEY: ${{ secrets.CUTEPETSBOSTON_RESCUEGROUPS_API_KEY }}
-          INSTAGRAM_BUSINESS_ACCOUNT_ID: ${{ secrets.INSTAGRAM_BUSINESS_ACCOUNT_ID }}
-          INSTAGRAM_PAGE_ACCESS_TOKEN: ${{ secrets.INSTAGRAM_PAGE_ACCESS_TOKEN }}
+          INSTAGRAM_BUSINESS_ACCOUNT_ID: ${{ secrets.INSTAGRAM_TEST_BUSINESS_ACCOUNT_ID }}
+          INSTAGRAM_PAGE_ACCESS_TOKEN: ${{ secrets.INSTAGRAM_TEST_PAGE_ACCESS_TOKEN }}
           BLUESKY_HANDLE: ${{ secrets.BLUESKY_TEST_HANDLE }}
           BLUESKY_PASSWORD: ${{ secrets.BLUESKY_TEST_PASSWORD }}
           MASTODON_TOKEN: ${{ secrets.MASTODON_TEST_TOKEN }}

README.md

@@ -16,8 +16,8 @@ Required:
 - `CUTEPETSBOSTON_RESCUEGROUPS_API_KEY`
 
 Optional for Instagram posting:
-- `INSTAGRAM_HANDLE`
-- `INSTAGRAM_PASSWORD`
+- `INSTAGRAM_BUSINESS_ACCOUNT_ID` (or `INSTAGRAM_TEST_BUSINESS_ACCOUNT_ID`)
+- `INSTAGRAM_PAGE_ACCESS_TOKEN` (or `INSTAGRAM_TEST_PAGE_ACCESS_TOKEN`)
 
 Optional for Bluesky posting:
 - `BLUESKY_HANDLE` (or `BLUESKY_TEST_HANDLE`)

manual_testing/instagram_manual_test.py

@@ -0,0 +1,81 @@
+import argparse
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from abstractions import AdoptablePet
+from social_posters.instagram import PosterInstagram
+
+
+def sample_pet():
+    return AdoptablePet(
+        name="Brian",
+        species="dog",
+        breed="Labrador Retriever",
+        location="Boston, MA",
+        description="Brian is a laid-back lab mix who loves a good nap and a good book.",
+        adoption_url="https://example.org/adopt/brian",
+        image_url="https://static.wikia.nocookie.net/familyguy/images/c/c2/FamilyGuy_Single_BrianWriter_R7.jpg/revision/latest?cb=20230807152447",
+        age_string="4 years",
+        sex="Male",
+        size_group="Large",
+        pet_id="manual-test-brian",
+    )
+
+
+testing_cases = [sample_pet]
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description="Manually exercise the Instagram poster against a real account."
+    )
+    parser.add_argument(
+        "--dry-run",
+        action="store_true",
+        help="Format the post without authenticating or publishing.",
+    )
+    parser.add_argument(
+        "--image-url",
+        help="Override the sample pet's image URL with a different publicly accessible image.",
+    )
+    args = parser.parse_args()
+
+    poster = PosterInstagram()
+
+    if not args.dry_run and not poster.authenticate():
+        print("Authentication failed!")
+        sys.exit(1)
+
+    if not args.dry_run:
+        print(f"Authenticated to Instagram as @{poster.username}")
+
+    for make_pet in testing_cases:
+        pet = make_pet()
+        if args.image_url:
+            pet.image_url = args.image_url
+
+        post = poster.format_post(pet)
+        print(f"\nPost preview:\n{post.text}")
+        print(f"\nTags: {post.tags}")
+        print(f"Alt text: {post.alt_text}")
+
+        if args.dry_run:
+            continue
+
+        print(
+            "\nPublishing (this polls Instagram until the image finishes "
+            "processing, up to 60s)..."
+        )
+        result = poster.publish(post)
+
+        if result.success:
+            print(f"\nPosted successfully! Media ID: {result.post_id}, URL: {result.post_url}")
+        else:
+            print(f"\nPost failed: {result.error_message}")
+            sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

metric_collectors/instagram.py

@@ -32,10 +32,8 @@ def fetch_metrics(
         try:
             response = requests.get(
                 f"{GRAPH_API_BASE}/{post_id}",
-                params={
-                    "fields": "like_count,comments_count",
-                    "access_token": self.access_token,
-                },
+                params={"fields": "like_count,comments_count"},
+                headers={"Authorization": f"Bearer {self.access_token}"},
                 timeout=20,
             )
             response.raise_for_status()

social_posters/instagram.py

@@ -6,8 +6,13 @@
 from abstractions import Post, PostResult, SocialPoster
 
 
-GRAPH_API_VERSION = "v21.0"
-GRAPH_API_BASE = f"https://graph.facebook.com/{GRAPH_API_VERSION}"
+GRAPH_API_VERSION = "v26.0"
+GRAPH_API_BASE = f"https://graph.instagram.com/{GRAPH_API_VERSION}"
+
+# Images typically finish container processing in seconds; 
+# video would need Meta's suggested ~1-minute cadence
+CONTAINER_POLL_INTERVAL_SECONDS = 5
+CONTAINER_POLL_TIMEOUT_SECONDS = 60
 
 
 class PosterInstagram(SocialPoster):
@@ -16,6 +21,7 @@ def __init__(self):
         self.access_token = os.environ.get("INSTAGRAM_PAGE_ACCESS_TOKEN")
         self._is_available = bool(self.account_id and self.access_token)
         self._authenticated = False
+        self.username = None
 
     @property
     def platform_name(self) -> str:
@@ -37,6 +43,7 @@ def authenticate(self) -> bool:
                 timeout=10,
             )
             response.raise_for_status()
+            self.username = response.json().get("username")
             self._authenticated = True
             return True
         except requests.exceptions.HTTPError as exc:
@@ -64,15 +71,16 @@ def publish(self, post: Post) -> PostResult:
 
         try:
             container_id = self._create_media_container(post)
-            # Instagram needs time to process the uploaded image before publishing.
-            # Publishing immediately returns "Media ID is not available" (error 9007).
-            time.sleep(10)
-    
+            self._wait_for_container_ready(container_id)
+
             media_id = self._publish_media(container_id)
+            post_url = (
+                f"https://www.instagram.com/{self.username}/" if self.username else None
+            )
             return PostResult(
                 success=True,
                 post_id=media_id,
-                post_url="https://www.instagram.com/cute.pets.boston/",
+                post_url=post_url,
             )
         except requests.exceptions.HTTPError as exc:
             body = exc.response.text if exc.response is not None else "no response body"
@@ -99,7 +107,37 @@ def _create_media_container(self, post: Post) -> str:
         response.raise_for_status()
         return response.json()["id"]
 
-  
+    def _wait_for_container_ready(self, container_id: str) -> None:
+        """Poll the container until Instagram finishes processing the image.
+
+        Publishing before the container is FINISHED returns "Media ID is not
+        available" (error 9007).
+        """
+        deadline = time.monotonic() + CONTAINER_POLL_TIMEOUT_SECONDS
+        while True:
+            response = requests.get(
+                f"{GRAPH_API_BASE}/{container_id}",
+                params={"fields": "status_code"},
+                headers=self._authorization_headers,
+                timeout=10,
+            )
+            response.raise_for_status()
+            status = response.json().get("status_code")
+
+            if status == "FINISHED":
+                return
+            if status in ("ERROR", "EXPIRED"):
+                raise RuntimeError(
+                    f"Instagram media container {container_id} failed with status {status}"
+                )
+            if time.monotonic() >= deadline:
+                raise RuntimeError(
+                    f"Instagram media container {container_id} did not finish "
+                    f"processing within {CONTAINER_POLL_TIMEOUT_SECONDS}s "
+                    f"(last status: {status})"
+                )
+            time.sleep(CONTAINER_POLL_INTERVAL_SECONDS)
+
     def _publish_media(self, container_id: str) -> str:
         response = requests.post(
             f"{GRAPH_API_BASE}/{self.account_id}/media_publish",

tests/test_instagram.py

@@ -17,6 +17,7 @@ def build_poster(monkeypatch) -> PosterInstagram:
 def test_authenticate_keeps_access_token_out_of_query_params(monkeypatch):
     poster = build_poster(monkeypatch)
     response = Mock()
+    response.json.return_value = {"id": ACCOUNT_ID, "username": "cutepetsboston2026_test"}
 
     with patch(
         "social_posters.instagram.requests.get",
@@ -31,6 +32,7 @@ def test_authenticate_keeps_access_token_out_of_query_params(monkeypatch):
         timeout=10,
     )
     response.raise_for_status.assert_called_once_with()
+    assert poster.username == "cutepetsboston2026_test"
 
 
 def test_create_media_container_uses_authorization_header(monkeypatch):
@@ -77,3 +79,60 @@ def test_publish_media_uses_authorization_header(monkeypatch):
         timeout=30,
     )
     response.raise_for_status.assert_called_once_with()
+
+
+def test_wait_for_container_ready_returns_once_finished(monkeypatch):
+    poster = build_poster(monkeypatch)
+    in_progress = Mock()
+    in_progress.json.return_value = {"status_code": "IN_PROGRESS"}
+    finished = Mock()
+    finished.json.return_value = {"status_code": "FINISHED"}
+
+    with (
+        patch(
+            "social_posters.instagram.requests.get",
+            side_effect=[in_progress, finished],
+        ) as request_get,
+        patch("social_posters.instagram.time.sleep") as mock_sleep,
+    ):
+        poster._wait_for_container_ready("container-id")
+
+    assert request_get.call_count == 2
+    request_get.assert_called_with(
+        f"{GRAPH_API_BASE}/container-id",
+        params={"fields": "status_code"},
+        headers={"Authorization": f"Bearer {ACCESS_TOKEN}"},
+        timeout=10,
+    )
+    mock_sleep.assert_called_once()
+
+
+def test_wait_for_container_ready_raises_on_error_status(monkeypatch):
+    poster = build_poster(monkeypatch)
+    response = Mock()
+    response.json.return_value = {"status_code": "ERROR"}
+
+    with patch("social_posters.instagram.requests.get", return_value=response):
+        try:
+            poster._wait_for_container_ready("container-id")
+            assert False, "expected RuntimeError"
+        except RuntimeError as exc:
+            assert "ERROR" in str(exc)
+
+
+def test_publish_builds_post_url_from_authenticated_username(monkeypatch):
+    poster = build_poster(monkeypatch)
+    poster._authenticated = True
+    poster.username = "cutepetsboston2026_test"
+    post = Post(text="Meet Poppy!", image_url="https://example.com/poppy.jpg")
+
+    with (
+        patch.object(poster, "_create_media_container", return_value="container-id"),
+        patch.object(poster, "_wait_for_container_ready"),
+        patch.object(poster, "_publish_media", return_value="media-id"),
+    ):
+        result = poster.publish(post)
+
+    assert result.success is True
+    assert result.post_id == "media-id"
+    assert result.post_url == "https://www.instagram.com/cutepetsboston2026_test/"

tests/test_metric_collector_instagram.py

@@ -20,10 +20,8 @@ def test_maps_media_counts_and_marks_reposts_not_applicable(self, mock_get):
         assert metrics.comments == 5
         mock_get.assert_called_once_with(
             f"{GRAPH_API_BASE}/media-123",
-            params={
-                "fields": "like_count,comments_count",
-                "access_token": "token",
-            },
+            params={"fields": "like_count,comments_count"},
+            headers={"Authorization": "Bearer token"},
             timeout=20,
         )
         response.raise_for_status.assert_called_once_with()
@@ -41,7 +39,8 @@ def test_returns_none_on_http_error(self, mock_get):
         assert metrics is None
 
     @patch("metric_collectors.instagram.requests.get")
-    def test_returns_none_without_access_token(self, mock_get):
+    def test_returns_none_without_access_token(self, mock_get, monkeypatch):
+        monkeypatch.delenv("INSTAGRAM_PAGE_ACCESS_TOKEN", raising=False)
         metrics = CollectorInstagram(access_token="").fetch_metrics("media-123")
 
         assert metrics is None