juxtapoandClaude Opus 5
5419b08264
get_or_open_tab(activate=True) — added, not replaced. Read paths pass activate=False; post flows keep the foreground for dialogs and file inputs. check_login moved off DOM signals onto session state: fb c_user, ig/threads ds_user_id, twitter twid, yt ytcfg.LOGGED_IN. Three platforms were reporting OUT on a signed-in account because their composer never mounts in a hidden tab. The fixed sleep(4)/sleep(5) before each check goes with them — every signal reads present the moment navigate() returns. Unreadable cookies raise instead of returning False: "cannot tell" is not "signed out". tiktok untouched and still activated — it reported visibilityState "visible" from a sibling-activated tab, so it has never been read from a genuine background. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""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")
|