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>
114 lines
4.1 KiB
Python
114 lines
4.1 KiB
Python
"""YouTube community post flow.
|
|
|
|
1. navigate youtube.com
|
|
2. click button[aria-label="Create"]
|
|
3. click "Create post" in the dropdown
|
|
4. wait for the redirect to the channel posts page
|
|
5. type into #contenteditable-root
|
|
6. click Post
|
|
7. the editor emptying / the post appearing = success
|
|
|
|
Community posts require a channel that has them enabled. Video upload
|
|
(post_youtube_video in the proposal) is NOT built here — the brief scoped this
|
|
module to the community post.
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from ..cdp import CDPError, human_delay
|
|
from . import guard, require_text, validate_media
|
|
from ..selectors import YOUTUBE as S
|
|
|
|
PLATFORM = "yt"
|
|
|
|
|
|
async def post(cdp_client, text, media_path=None, options=None):
|
|
options = options or {}
|
|
require_text(PLATFORM, text)
|
|
validate_media(media_path)
|
|
if media_path:
|
|
# Community posts do take images, but that flow is unmapped. Refuse
|
|
# rather than drop the media on the floor.
|
|
raise CDPError(
|
|
"yt: media on community posts is not implemented (the image-attach "
|
|
"flow has not been mapped live). Post text-only, or use a platform "
|
|
"whose media flow is built."
|
|
)
|
|
|
|
tab, reused = await cdp_client.get_or_open_tab(S["url"], "youtube.com")
|
|
if reused:
|
|
await tab.eval("window.onbeforeunload = null")
|
|
await tab.navigate(S["url"])
|
|
return await guard(PLATFORM, tab, _flow(tab, text))
|
|
|
|
|
|
async def _flow(tab, text):
|
|
await tab.wait_for(S["create_button"])
|
|
await human_delay()
|
|
await tab.click(S["create_button"])
|
|
|
|
# TODO: VALIDATE LIVE — the dropdown item text and element type.
|
|
await _wait_for_js(tab, S["create_post_item_js"], "the 'Create post' menu item")
|
|
await tab.click_js(S["create_post_item_js"])
|
|
|
|
# Step 4: YouTube navigates to the channel's posts page.
|
|
await tab.wait_for_load(timeout=30.0)
|
|
await tab.wait_for(S["editor"], timeout=30.0)
|
|
|
|
# #contenteditable-root is a YouTube custom editor: it listens to real key
|
|
# events, so type per-char rather than insertText.
|
|
# TODO: VALIDATE LIVE — whether insertText alone enables the Post button.
|
|
await tab.type_text(S["editor"], text, per_char=True)
|
|
|
|
if await tab.exists(S["post_button"]):
|
|
await tab.click(S["post_button"])
|
|
else:
|
|
await tab.click_js(S["post_button_js"])
|
|
|
|
# TODO: VALIDATE LIVE — success marker. The editor going empty is the
|
|
# cheapest observable signal; the proposal says "post appears".
|
|
await _wait_for_editor_empty(tab)
|
|
|
|
|
|
async def _wait_for_js(tab, expression, what, timeout=20.0):
|
|
deadline = asyncio.get_running_loop().time() + timeout
|
|
while asyncio.get_running_loop().time() < deadline:
|
|
if await tab.eval("!!(%s)" % expression):
|
|
return
|
|
await asyncio.sleep(0.25)
|
|
raise CDPError(f"yt: {what} did not appear within {timeout}s")
|
|
|
|
|
|
async def _wait_for_editor_empty(tab, timeout=30.0):
|
|
deadline = asyncio.get_running_loop().time() + timeout
|
|
while asyncio.get_running_loop().time() < deadline:
|
|
text = await tab.eval(
|
|
"(() => { const e = document.querySelector('%s');"
|
|
" return e ? e.innerText.trim() : null; })()" % S["editor"]
|
|
)
|
|
if text == "":
|
|
return
|
|
await asyncio.sleep(0.5)
|
|
raise CDPError("yt: editor still holds text 30s after clicking Post")
|
|
|
|
|
|
async def check_login(cdp_client):
|
|
tab, reused = await cdp_client.get_or_open_tab(S["url"], "youtube.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, not the masthead DOM: #avatar-btn needs a fully rendered
|
|
# foreground tab and reads absent for a signed-in account otherwise.
|
|
# ytcfg is set by the page's own bootstrap and does not depend on render.
|
|
state = await tab.eval(
|
|
"(() => { try { return String(ytcfg.get('LOGGED_IN')); }"
|
|
" catch (e) { return 'unavailable'; } })()"
|
|
)
|
|
if state == "true":
|
|
return True
|
|
if state == "false":
|
|
return False
|
|
raise CDPError("yt: ytcfg.LOGGED_IN unavailable — cannot tell signed-in from signed-out")
|