"""Instagram posting flow (media required). 1. navigate instagram.com 2. click svg[aria-label="New post"] 3. wait for the "Create new post" dialog 4. setFileInputFiles on the dialog's file input (bypasses the picker) 5. crop screen -> Next 6. filter screen -> Next 7. caption screen -> type caption 8. Share 9. dialog closes = success """ import asyncio import json from ..cdp import CDPError, human_delay from . import guard, require_media from ..selectors import INSTAGRAM as S PLATFORM = "ig" async def post(cdp_client, text, media_path=None, options=None): options = options or {} require_media(PLATFORM, media_path) tab, reused = await cdp_client.get_or_open_tab(S["url"], "instagram.com") if reused: await tab.eval("window.onbeforeunload = null") await tab.navigate(S["url"]) return await guard(PLATFORM, tab, _flow(tab, text or "", media_path)) async def _flow(tab, caption, media_path): await tab.wait_for(S["new_post_svg"]) await human_delay() # JS click — walks up to the clickable parent of the SVG icon await tab.eval(S["new_post_js"]) await human_delay() await tab.wait_for(S["dialog"]) await tab.wait_for_text(S["create_dialog_text"]) # The file input is present but hidden — visible=False on purpose. await tab.wait_for(S["file_input"], visible=False) await tab.set_file_input(S["file_input"], media_path) # TODO: VALIDATE LIVE — how long IG takes to move from picker to crop, and # whether the crop step is skipped for already-square media. await human_delay(2.0, 4.0) # Crop -> Next, filter -> Next. Same button text on both screens, so we # click it twice with a settle between; if a screen is skipped the second # Next simply won't be there and we fail loudly rather than guessing. for step in ("crop", "filter"): await _click_next(tab, step) await human_delay(1.0, 2.0) await tab.wait_for(S["caption_box"]) if caption: await tab.type_text(S["caption_box"], caption) await tab.click_js(S["share_button_js"]) # TODO: VALIDATE LIVE — IG shows "Your post has been shared" before the # dialog closes; either signal is success, whichever lands first. await _wait_for_share(tab) async def _click_next(tab, step): deadline = asyncio.get_running_loop().time() + 20.0 while asyncio.get_running_loop().time() < deadline: found = await tab.eval("!!(%s)" % S["next_button_js"]) if found: await tab.click_js(S["next_button_js"]) return await asyncio.sleep(0.25) raise CDPError(f"ig: no 'Next' button on the {step} screen after 20s") async def _wait_for_share(tab): deadline = asyncio.get_running_loop().time() + 90.0 while asyncio.get_running_loop().time() < deadline: if await tab.eval( "document.body.innerText.includes(%s)" % json.dumps(S["shared_text"]) ): # Click "Done" to dismiss the confirmation dialog await human_delay(0.5, 1.0) try: await tab.click_js(S["done_button_js"]) except Exception: pass return if not await tab.exists(S["dialog"]): return await asyncio.sleep(0.5) raise CDPError("ig: no share confirmation and the dialog stayed open for 90s") async def check_login(cdp_client): tab, reused = await cdp_client.get_or_open_tab(S["url"], "instagram.com", activate=False) if reused: await tab.eval("window.onbeforeunload = null") await tab.navigate(S["url"]) if await tab.exists(S["login_marker"]): return False # JS state rather than the new-post SVG. The icon does survive a background # tab today (measured 2026-09-12), but it is an icon — it moves when the nav # is redesigned. The session cookie is the session. state = await tab.eval( "(() => { const m = document.cookie.match(/(?:^|;\\s*)ds_user_id=(\\d+)/);" " return m ? 'in:' + m[1] : (document.cookie ? 'out' : 'unavailable'); })()" ) if state.startswith("in:"): return True if state == "out": return False raise CDPError("ig: no cookies readable on instagram.com — cannot tell signed-in from signed-out")