110 lines
3.7 KiB
Python
110 lines
3.7 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")
|
|
if reused:
|
|
await tab.eval("window.onbeforeunload = null")
|
|
await tab.navigate(S["url"])
|
|
await asyncio.sleep(5)
|
|
if await tab.exists(S["login_marker"]):
|
|
return False
|
|
return await tab.exists(S["new_post_svg"])
|