feat: support linked reels

docs/usage-guide/media.md

@@ -38,6 +38,7 @@ In terms of Instagram, this is called Media, usually users call it publications
 | media_info(media_pk: str, use_cache: bool = True) | Media | Return media info |
 | media_delete(media_id: str) | bool | Delete media |
 | media_edit(media_id: str, caption: str, title: str = "", usertags: List[Usertag] = [], location: Location = None) | dict | Edit caption, optional IGTV title, usertags, or location |
+| media_link_reel(media_id: str, target_media_id: str, link_name: str = "Watch Next") | bool | Link one Reel to another Reel so Instagram can show a navigation button |
 | media_user(media_pk: str) | UserShort | Get author of media |
 | media_oembed(url: str) | dict | Return short oEmbed-style media info by URL |
 | media_like(media_id: str) | bool | Like media |
@@ -275,6 +276,7 @@ Notes:
 * `media_pk_from_url()` now also resolves `share/p/...` URLs before extracting the canonical shortcode.
 * Accepted Instagram Collabs/coauthor users from private media payloads are available as `media.coauthor_producers`. This is separate from upload-time `coauthor_user_ids`, which only sends collaborator invitations.
 * `media_edit()` uses `caption` and optional `location`/`usertags`; for IGTV posts it can also derive or send a separate `title`.
+* `media_link_reel()` uses Instagram's private `media/{media_id}/edit_media/` endpoint. Pass full media IDs when you have them; plain media PKs are normalized with `media_id()` before the request.
 
 Extended `Media` metadata fields:
 

instagrapi/mixins/media.py

@@ -787,6 +787,39 @@ def media_edit(
         )
         return result
 
+    def media_link_reel(self, media_id: str, target_media_id: str, link_name: str = "Watch Next") -> bool:
+        """
+        Link one Reel to another Reel.
+
+        Parameters
+        ----------
+        media_id: str
+            Origin media id that receives the linked Reel navigation button
+        target_media_id: str
+            Destination media id opened by the linked Reel navigation button
+        link_name: str, optional
+            Text shown by Instagram for the link, default value is "Watch Next"
+
+        Returns
+        -------
+        bool
+            A boolean value
+        """
+        assert self.user_id, "Login required"
+        media_id = self.media_id(media_id)
+        target_media_id = self.media_id(target_media_id)
+        data = {
+            "_uid": str(self.user_id),
+            "_uuid": self.uuid,
+            "linked_media_info": dumps({"media_id": target_media_id, "link_name": link_name}),
+        }
+        self._medias_cache.pop(self.media_pk(media_id), None)
+        result = self.private_request(
+            f"media/{media_id}/edit_media/",
+            self.with_action_data(data),
+        )
+        return result.get("status") == "ok"
+
     def media_user(self, media_pk: str) -> UserShort:
         """
         Get author of the media

tests/helpers.py

@@ -211,11 +211,16 @@ def build_test_accounts_url(count=None):
 def client_from_test_account(acc):
     settings = dict(acc["client_settings"])
     totp_seed = settings.pop("totp_seed", None)
+    authorization_data = settings.get("authorization_data") or {}
+    account_user_id = acc.get("user_id") or authorization_data.get("ds_user_id")
+    has_authorized_session = bool(authorization_data)
     cl = Client(settings=settings, proxy=os.getenv("IG_PROXY") or acc["proxy"])
+    if has_authorized_session and account_user_id:
+        cl._user_id = str(account_user_id)
     login_kwargs = {
         "username": acc["username"],
         "password": acc["password"],
-        "relogin": True,
+        "relogin": not has_authorized_session,
     }
     if totp_seed:
         totp_code = cl.totp_generate_code(totp_seed)
@@ -224,7 +229,7 @@ def client_from_test_account(acc):
         login_kwargs["verification_code"] = totp_code
     with fresh_account_login_timeout():
         cl.login(**login_kwargs)
-    cl._user_id = acc.get("user_id")
+    cl._user_id = account_user_id
     return cl
 
 

tests/live/test_media.py

@@ -1,4 +1,11 @@
-from instagrapi.exceptions import ClientForbiddenError
+from instagrapi.exceptions import (
+    ClientForbiddenError,
+    ClientThrottledError,
+    ClipNotUpload,
+    LoginRequired,
+    PhotoNotUpload,
+    PleaseWaitFewMinutes,
+)
 from tests import helpers as _helpers
 from tests.helpers import *
 
@@ -186,6 +193,98 @@ def test_media_note_create_and_delete(self):
                 self.assertTrue(self.cl.media_note_delete(note["id"]))
 
 
+class ClientLinkedReelLiveTestCase(_helpers.ClientPrivateTestCase):
+    def __init__(self, *args, **kwargs):
+        self.cl = None
+        self.clients = []
+        return unittest.TestCase.__init__(self, *args, **kwargs)
+
+    def setup_method(self, *args, **kwargs):
+        return None
+
+    def setUp(self):
+        if not TEST_ACCOUNTS_URL:
+            self.skipTest("TEST_ACCOUNTS_URL is required for linked Reel live tests")
+        try:
+            self.clients = self.fresh_accounts(10)
+        except RuntimeError as exc:
+            self.skipTest(str(exc))
+
+    def make_cover_fixture(self, color):
+        try:
+            from PIL import Image
+        except ImportError:
+            self.skipTest("Pillow is required to generate a linked Reel cover fixture")
+
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as tmp:
+            path = Path(tmp.name)
+        self.addCleanup(lambda: path.unlink(missing_ok=True))
+        Image.new("RGB", (720, 1280), color).save(path, quality=95)
+        return path
+
+    def cleanup_uploaded_media(self, media):
+        if not media:
+            return
+        try:
+            self.assertTrue(self.cl.media_delete(media.id))
+        except Exception as exc:
+            print(f"Linked Reel cleanup media_delete failed: {exc.__class__.__name__} {exc}")
+
+    def run_media_link_reel(self, client, origin_path, target_path, origin_cover, target_cover):
+        self.cl = client
+        origin = None
+        target = None
+        try:
+            timestamp = int(time.time())
+            target = self.cl.clip_upload(target_path, f"Linked Reel target {timestamp}", thumbnail=target_cover)
+            self.assertUploadedMediaAccessible(
+                target,
+                media_type=2,
+                product_type="clips",
+                caption_text=f"Linked Reel target {timestamp}",
+            )
+            origin = self.cl.clip_upload(origin_path, f"Linked Reel origin {timestamp}", thumbnail=origin_cover)
+            self.assertUploadedMediaAccessible(
+                origin,
+                media_type=2,
+                product_type="clips",
+                caption_text=f"Linked Reel origin {timestamp}",
+            )
+
+            self.assertTrue(self.cl.media_link_reel(origin.id, target.id, link_name="Watch Next"))
+        finally:
+            for media in (origin, target):
+                self.cleanup_uploaded_media(media)
+
+    def test_media_link_reel(self):
+        origin_path = self.make_video_fixture(label="linked Reel origin fixture", color="blue")
+        target_path = self.make_video_fixture(label="linked Reel target fixture", color="red")
+        origin_cover = self.make_cover_fixture("blue")
+        target_cover = self.make_cover_fixture("red")
+        self.assertIsInstance(origin_path, Path)
+        self.assertIsInstance(target_path, Path)
+        self.assertIsInstance(origin_cover, Path)
+        self.assertIsInstance(target_cover, Path)
+
+        upload_failures = {}
+        for client in self.clients:
+            try:
+                self.run_media_link_reel(client, origin_path, target_path, origin_cover, target_cover)
+                return
+            except (
+                ClipNotUpload,
+                PhotoNotUpload,
+                LoginRequired,
+                PleaseWaitFewMinutes,
+                ClientThrottledError,
+                RetryError,
+            ) as exc:
+                upload_failures[exc.__class__.__name__] = upload_failures.get(exc.__class__.__name__, 0) + 1
+                continue
+
+        self.skipTest(f"No linked Reel upload-capable test account was available (upload_failures={upload_failures})")
+
+
 class ClientCompareExtractTestCase(_helpers.ClientPrivateTestCase):
     def assertLocation(self, v1, gql):
         if not isinstance(v1, dict):

tests/regression/test_helpers.py

@@ -65,6 +65,40 @@ def test_client_from_test_account_times_out_hung_login(self):
                 with self.assertRaises(helper_module.FreshAccountLoginTimeout):
                     helper_module.client_from_test_account(self._account_payload())
 
+    def test_client_from_test_account_reuses_authorized_session_settings(self):
+        account = self._account_payload()
+        account["client_settings"] = {
+            "authorization_data": {
+                "ds_user_id": "123",
+                "sessionid": "sessionid",
+                "should_use_header_over_cookies": True,
+            },
+        }
+        client = Mock()
+
+        def assert_user_id_set_before_login(**kwargs):
+            self.assertEqual(client._user_id, "123")
+            return True
+
+        client.login.side_effect = assert_user_id_set_before_login
+
+        with mock.patch.dict(helper_module.os.environ, {"IG_PROXY": ""}):
+            with mock.patch.object(helper_module, "Client", return_value=client):
+                result = helper_module.client_from_test_account(account)
+
+        self.assertIs(result, client)
+        client.login.assert_called_once_with(username="fresh.account", password="password", relogin=False)
+
+    def test_client_from_test_account_relogs_without_authorized_session_settings(self):
+        client = Mock()
+
+        with mock.patch.dict(helper_module.os.environ, {"IG_PROXY": ""}):
+            with mock.patch.object(helper_module, "Client", return_value=client):
+                result = helper_module.client_from_test_account(self._account_payload())
+
+        self.assertIs(result, client)
+        client.login.assert_called_once_with(username="fresh.account", password="password", relogin=True)
+
     def test_fresh_test_account_tries_next_account_after_login_timeout(self):
         accounts = [self._account_payload("first.account"), self._account_payload("second.account")]
         client = object()

tests/regression/test_media.py

@@ -910,6 +910,46 @@ def test_media_note_delete_posts_current_v2_payload(self):
             with_signature=False,
         )
 
+    def test_media_link_reel_posts_linked_media_info(self):
+        client = self._build_logged_in_client()
+        client._medias_cache = {"111": object()}
+
+        with mock.patch.object(client, "private_request", return_value={"status": "ok"}) as private_request:
+            result = client.media_link_reel("111_222", "333_444", link_name="Watch Part 1")
+
+        self.assertTrue(result)
+        endpoint, data = private_request.call_args.args
+        self.assertEqual(endpoint, "media/111_222/edit_media/")
+        self.assertEqual(data["_uid"], "1")
+        self.assertEqual(data["_uuid"], "uuid")
+        self.assertEqual(data["device_id"], "android-device")
+        self.assertEqual(data["radio_type"], "wifi-none")
+        self.assertEqual(
+            json.loads(data["linked_media_info"]),
+            {"media_id": "333_444", "link_name": "Watch Part 1"},
+        )
+        self.assertNotIn("111", client._medias_cache)
+
+    def test_media_link_reel_normalizes_origin_and_target_media_ids(self):
+        client = self._build_logged_in_client()
+
+        with (
+            mock.patch.object(
+                client,
+                "media_user",
+                side_effect=[
+                    UserShort(pk="222", username="origin", profile_pic_url="https://example.com/origin.jpg"),
+                    UserShort(pk="444", username="target", profile_pic_url="https://example.com/target.jpg"),
+                ],
+            ),
+            mock.patch.object(client, "private_request", return_value={"status": "ok"}) as private_request,
+        ):
+            self.assertTrue(client.media_link_reel("111", "333"))
+
+        endpoint, data = private_request.call_args.args
+        self.assertEqual(endpoint, "media/111_222/edit_media/")
+        self.assertEqual(json.loads(data["linked_media_info"])["media_id"], "333_444")
+
 
 class UsertagMediasPaginationRegressionTestCase(unittest.TestCase):
     def _media_v1_payload(self, pk="1"):