131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
"""TikTok upload flow (media required).
|
|
|
|
1. navigate tiktokstudio/upload
|
|
2. wait for the upload page
|
|
3. photos: switch to the Photos tab first
|
|
4. setFileInputFiles on the matching input[type=file]
|
|
5. wait for processing
|
|
6. type the caption
|
|
7. click Post/Publish
|
|
8. wait for the success confirmation
|
|
|
|
TikTok is the most bot-sensitive of the five (proposal: ⭐⭐⭐⭐). Delays here
|
|
are deliberately longer than elsewhere.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
|
|
from ..cdp import CDPError, human_delay
|
|
from . import guard, require_media
|
|
from ..selectors import TIKTOK as S
|
|
|
|
PLATFORM = "tiktok"
|
|
|
|
VIDEO_EXTS = {".mp4", ".mov", ".webm", ".avi", ".mkv"}
|
|
PHOTO_EXTS = {".jpg", ".jpeg", ".png", ".webp", ".heic"}
|
|
|
|
|
|
async def post(cdp_client, text, media_path=None, options=None):
|
|
options = options or {}
|
|
require_media(PLATFORM, media_path)
|
|
|
|
ext = os.path.splitext(media_path)[1].lower()
|
|
if ext in VIDEO_EXTS:
|
|
kind = "video"
|
|
elif ext in PHOTO_EXTS:
|
|
kind = "photo"
|
|
else:
|
|
raise CDPError(
|
|
f"tiktok: unrecognised media extension {ext!r} — "
|
|
f"videos {sorted(VIDEO_EXTS)}, photos {sorted(PHOTO_EXTS)}"
|
|
)
|
|
|
|
tab, reused = await cdp_client.get_or_open_tab(S["url"], "tiktok.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, kind))
|
|
|
|
|
|
async def _flow(tab, caption, media_path, kind):
|
|
await tab.wait_for_load(timeout=45.0)
|
|
await human_delay(1.5, 3.0)
|
|
|
|
if kind == "photo":
|
|
# TODO: VALIDATE LIVE — the Photos tab switch and the input it reveals.
|
|
await _wait_for_js(tab, S["photos_tab_js"], "the Photos tab", timeout=30.0)
|
|
await tab.click_js(S["photos_tab_js"])
|
|
await human_delay(1.0, 2.0)
|
|
file_input = S["photo_file_input"]
|
|
else:
|
|
file_input = S["video_file_input"]
|
|
|
|
await tab.wait_for(file_input, visible=False, timeout=45.0)
|
|
await tab.set_file_input(file_input, media_path)
|
|
|
|
# Step 5: TikTok transcodes before it will let you post.
|
|
# TODO: VALIDATE LIVE — the real "upload finished" marker. Right now we
|
|
# wait for the Post button to become enabled, which is the behaviour the
|
|
# flow actually depends on.
|
|
await _wait_for_js(
|
|
tab, S["post_button_js"], "an enabled Post button", timeout=300.0
|
|
)
|
|
|
|
if caption:
|
|
await tab.wait_for(S["caption_box"])
|
|
# DraftJS: type per-char so its keypress handlers fire.
|
|
# TODO: VALIDATE LIVE — whether insertText is accepted here.
|
|
await tab.type_text(S["caption_box"], caption, per_char=True)
|
|
await human_delay(1.0, 2.0)
|
|
|
|
await tab.click_js(S["post_button_js"])
|
|
|
|
# TikTok may show a copyright check dialog — dismiss it
|
|
await asyncio.sleep(3)
|
|
try:
|
|
has_continue = await tab.eval("!!(%s)" % S["continue_post_js"])
|
|
if has_continue:
|
|
await tab.click_js(S["continue_post_js"])
|
|
except Exception:
|
|
pass
|
|
|
|
await _wait_for_success(tab)
|
|
|
|
|
|
async def _wait_for_js(tab, expression, what, timeout=30.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.5)
|
|
raise CDPError(f"tiktok: {what} did not appear within {timeout}s")
|
|
|
|
|
|
async def _wait_for_success(tab, timeout=120.0):
|
|
# TODO: VALIDATE LIVE — the confirmation copy. TikTok also redirects to
|
|
# /tiktokstudio/content on success, which we accept as the same signal.
|
|
deadline = asyncio.get_running_loop().time() + timeout
|
|
marker = json.dumps(S["success_text"])
|
|
while asyncio.get_running_loop().time() < deadline:
|
|
if await tab.eval(f"document.body.innerText.includes({marker})"):
|
|
return
|
|
if "tiktokstudio/content" in (await tab.url() or ""):
|
|
return
|
|
await asyncio.sleep(1.0)
|
|
raise CDPError(
|
|
f"tiktok: no success confirmation within {timeout}s of clicking Post"
|
|
)
|
|
|
|
|
|
async def check_login(cdp_client):
|
|
tab, reused = await cdp_client.get_or_open_tab(S["url"], "tiktok.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["video_file_input"])
|