socialmag/platforms/tiktok.py

"""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):
    # The only platform still activated for a login check. TikTok could not be
    # backgrounded on 2026-09-12: a sibling tab in the *same* window was
    # activated (Browser.getWindowForTarget agreed on windowId) and TikTok still
    # reported visibilityState "visible" — own window, or it overrides the
    # property to keep video playing. Its DOM signal has never lied, but it has
    # never been read from a genuine background either. Three states, not two:
    # this stays foreground until there is evidence, not until it looks fine.
    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"])