playwright-demo/playwright_demo.py

"\
Tool to help with endfield sign-in rewards.\n\n\
It will notify you if you need to log in manually, and then \
wait for you to close the browser after logging in. \n\
If the rewards are claimed successfully, it will notify you as well. \n\
If the rewards have already been claimed, it will let you know.\
"

APP_NAME = "lda's ef sign-in helper"
randomlda = {APP_NAME: (1) // int("""
    .valueOf(), // cool!
};
const fn =
  /* """ * 0 + "1")}  # fmt: skip
# this is LOVELY

# autopep8: off
import asyncio
import logging
import os
import random
import sys
from collections.abc import Awaitable, Callable
from inspect import isawaitable
from itertools import count
from math import ceil
from pathlib import Path
from typing import Any

os.environ["PLAYWRIGHT_SKIP_BROWSER_DOWNLOAD"] = "1"
from desktop_notifier import DesktopNotifier, Icon
from playwright.async_api import (
    Error as PlaywrightError,
)
from playwright.async_api import (
    Locator,
    Page,
    Playwright,
    async_playwright,
)
from playwright.async_api import (
    TimeoutError as PlaywrightTimeoutError,
)

# autopep8: on

CHROME_DATA_DIR = Path(__file__).parent / "chrome-data-dir"
CHROME_DATA_DIR.mkdir(exist_ok=True)

EF_ICON_PATH = Path(__file__).parent / "resources" / "ef_icon.ico"

LOGIN_TIMEOUT_MS = 30 * 60 * 1000
CLAIM_TIMEOUT_MS = 90 * 1000

URL = "https://game.skport.com/endfield/sign-in"

DESKTOP_NOTIFIER = DesktopNotifier(app_name=APP_NAME, app_icon=Icon(EF_ICON_PATH))

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s.%(msecs)03d %(levelname)-8s %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger(__name__)


async def launch_context(p: Playwright, *, headless: bool):
    return await p.chromium.launch_persistent_context(
        user_data_dir=CHROME_DATA_DIR,
        headless=headless,
    )


async def notify(
    msg: str,
    title: str = "Login",
    *,
    notifier=DESKTOP_NOTIFIER,
    on_clicked: Callable[[], Any] | None = None,
):
    """Send a desktop notification.

    desktop_notifier can't make callbacks that rerun the script... im crine.
    """
    # print(f"{title}: {msg}")
    logger.info("%s: %s", title, msg)
    await notifier.send(title=title, message=msg, timeout=5, on_clicked=on_clicked)


async def wait_for_manual_login(page: Page, *, timeout: int = LOGIN_TIMEOUT_MS):
    "suitable as a task, does raise PlaywrightTimeoutError if timeout reached. Set to 0 for no timeout."
    await page.locator("form").first.wait_for(timeout=timeout)
    logger.info("Manual login detected, form is visible.")


async def already_claimed(container: Locator):
    # language=javascript
    fn = """// */
  /** @param {HTMLElement} el */
  (el) => {
    return [...el.parentElement.children].some(
      (x) => x !== el && x.id === "completed-overlay" && x.checkVisibility?.(),
    );
  };
/* *\x2f"""  # yes sir
    # python: 2f -> /, below call would receive:
    # - first line: commented out end block comment
    # - the function (thanks chatgpt)
    # - last line: starts and ends a block comment

    # js: middle of the file.
    # - ends the block comment full of python
    # - the function (thanks chatgpt)
    # - starts another block comment full of python
    # - note the next asterisk slash is at the end of the file, and nothing in between could have those two characters together.
    # - unless you immediately start a new block comment. Or have valid js code in between.
    # return await container.evaluate(Path(__file__).read_text()) # works btw
    return await container.evaluate(fn)


class LoginNeededError(Exception):
    """Raised when manual login is required."""


async def click_loop(page: Page):
    """Loop that attempts to click the claim button until successful or already claimed.

    May run indefinitely if the claim button is not visible or the page is not ready."""
    container: Locator = page.locator("#lottie-container").first

    logger.debug("Waiting for claim button to be visible.")
    await container.wait_for(timeout=0)  # wait for visible

    # at this point, the page is ready.
    logger.debug("Claim button is visible, starting click loop.")
    if await already_claimed(container):
        await notify(
            "Work already completed elsewhere.",
            title="Already Claimed",
        )
        logger.info("Already claimed, exiting click loop.")
        return

    for t in count(1):  # can be interrupted by timeout at any await point
        logger.debug("Attempt %d: Clicking claim button.", t)
        try:
            await container.click(
                timeout=3000,
                delay=random.randint(100, 300),
                # we checked for visibility manually. This will probs work.
                force=True,
            )
        except PlaywrightTimeoutError:
            logger.debug("Attempt %d: Click timed out.", t)
        except PlaywrightError:
            logger.debug("Attempt %d: Playwright error during click.", t, exc_info=True)

        logger.debug("Attempt %d: Checking claim state.", t)

        if await already_claimed(container):  # what a fucking mess
            if t == 1:
                await notify(
                    "Claimed successfully.",
                    title="Success",
                )
            else:
                await notify(
                    f"Claimed successfully after {t} attempts.",
                    title="Success",
                )

            logger.info("Claimed successfully after %d attempts.", t)
            return

        await asyncio.sleep(2)


async def claim_loop(p: Playwright) -> bool:
    """Loop that attempts to claim the reward until successful or already claimed.

    Does have a timeout.

    Raises:
        LoginNeededError: If manual login is required.
        other exceptions may be raised if there are unexpected errors during the claim process.

    Returns:
        True: If the reward is claimed successfully or already claimed.
        False: If the reward is not claimed and the timeout is reached.
    """
    async with await launch_context(p, headless=True) as context:
        page = await context.new_page()
        await page.goto(URL)
        logger.debug("Page loaded.")

        # the following code is BUNS
        tasks: list[asyncio.Task] = [
            asyncio.create_task(wait_for_manual_login(page, timeout=0)),
            asyncio.create_task(click_loop(page)),
            asyncio.create_task(page.wait_for_event("close")),
        ]

        done, _ = await asyncio.wait(
            tasks,
            # important line below, no tasks cancels normally, as in they run forever.
            timeout=CLAIM_TIMEOUT_MS / 1000,  # deadline in seconds
            return_when=asyncio.FIRST_COMPLETED,
        )

    # chromium context closed; drain remaining tasks.
    await asyncio.gather(*tasks, return_exceptions=True)  # we dont care

    if not done:
        logger.warning("Claim attempt timed out after %ds.", CLAIM_TIMEOUT_MS // 1000)
        return False

    if tasks[0] in done:
        tasks[0].result()
        logger.info("Manual login required.")
        raise LoginNeededError("Manual login required.")
    if tasks[1] in done:
        tasks[1].result()
        logger.debug("Claim worker finished.")
        return True
    if tasks[2] in done:
        logger.warning("Browser page closed before claim completed.")
        return False  # page closed

    return False  # should never happen, but just in case


async def focus_page(page: Page):
    """Focus the page by bringing it to the front."""
    if page.is_closed():
        return
    try:
        await page.bring_to_front()
        logger.debug("Brought page to front.")
    except PlaywrightError:
        pass  # Page/context closed in the meantime.


async def notify_schedule(
    duration: float,
    callback: Callable[[float], None | Awaitable[None]] | None = None,
    *,
    checkpoints: tuple[float, ...] = (1800, 900, 600, 300, 180, 120, 60),
):
    """Run callback at selected remaining-time checkpoints.

    Args:
        duration (float): Total duration in seconds.
        callback (Callable[[float], None | Awaitable[None]] | None): Callback function to be called at each checkpoint. It receives the remaining time in seconds as an argument.
        checkpoints (tuple[float, ...]): Checkpoints in seconds at which the callback will be called. Default is (1800, 900, 600, 300, 180, 120, 60).
    """

    loop = asyncio.get_running_loop()
    deadline = loop.time() + duration

    for target_remaining in sorted(checkpoints, reverse=True):
        if duration - target_remaining < 1:  # give it a second
            continue

        await asyncio.sleep(max(0, deadline - target_remaining - loop.time()))

        if callback is not None:
            remaining = max(0, deadline - loop.time())
            result = callback(remaining)
            if isawaitable(result):
                await result
    await asyncio.sleep(max(0, deadline - loop.time()))


async def manual_login(p: Playwright) -> bool:
    """Handle manual login process.

    Returns:
        bool: True if browser was closed manually, False if timeout occurred.
    """
    async with await launch_context(p, headless=False) as context:
        page = context.pages[0] if context.pages else await context.new_page()
        await page.goto(URL)

        logger.info(
            "Manual login required. Waiting for user to log in and close the browser."
        )
        await notify(
            f"Please log in manually and close the browser. You have {LOGIN_TIMEOUT_MS // (60 * 1000)} minute(s).",
            title="Manual Login Required",
            on_clicked=lambda: asyncio.create_task(focus_page(page)),
        )

        close_task = asyncio.create_task(
            context.wait_for_event(
                "close",
                timeout=LOGIN_TIMEOUT_MS,
            )
        )

        async def notify_schedule_callback(remaining: float):
            minutes, seconds = divmod(ceil(remaining), 60)
            logger.info("Manual login timeout in %dm %ds.", minutes, seconds)
            await notify(
                f"Manual login timeout in {minutes}m {seconds}s. Click to return.",
                title="Manual Login Timeout",
                on_clicked=lambda: asyncio.create_task(focus_page(page)),
            )

        wake_task = asyncio.create_task(
            notify_schedule(LOGIN_TIMEOUT_MS / 1000, callback=notify_schedule_callback)
        )

        done, _ = await asyncio.wait(
            {close_task, wake_task},
            return_when=asyncio.FIRST_COMPLETED,
        )

        close_task_finished_first = close_task in done

        if close_task_finished_first and not wake_task.done():
            wake_task.cancel()

    # Context has now closed. Drain both tasks.
    await asyncio.gather(
        close_task,
        wake_task,
        return_exceptions=True,
    )

    try:
        close_task.result()
    except PlaywrightTimeoutError:
        logger.warning("Manual login timed out.")
        return False

    if close_task_finished_first:
        logger.info("Manual login browser closed by user.")
        return True

    wake_task.result()
    logger.warning("Manual login timed out.")
    return False


async def program2():
    logger.info("Starting sign-in automation.")
    async with async_playwright() as p:
        for i in range(2):
            try:
                logger.info("Attempting to claim reward (%d)", i + 1)
                if await claim_loop(p):
                    logger.info("Sign-in reward workflow completed.")
                    return

                await notify(
                    "Timed out trying to claim the reward. Please rerun the script.",
                    title="Timeout",
                )
                raise SystemExit(1)
            except LoginNeededError:
                if i > 0:
                    logger.warning(
                        "Manual authentication failed; login is still required."
                    )
                    await notify(
                        "Still not logged in. Please check your session.",
                        title="Login Failed",
                    )
                    raise SystemExit(1)
                logger.info("Opening browser for manual authentication.")
                if not await manual_login(p):
                    await notify(
                        "Login timeout reached. Please rerun the script.",
                        title="Login Timeout",
                    )
                    raise SystemExit(1)
                logger.info("Manual authentication finished. Retrying claim.")


def main():  # uv no main fix
    async def run_program():
        try:
            if sys.argv[1:] and sys.argv[1] == "--manual-login":
                async with async_playwright() as p:
                    return not await manual_login(p)
            else:
                await program2()
        except Exception as e:
            logger.exception("Unhandled exception in sign-in helper.")
            await notify(
                f"An error occurred\n{e!s}",
                title="Error",
            )
            raise

    return asyncio.run(run_program())


if __name__ == "__main__":
    raise SystemExit(main())
    # async, also this file is valid js. */ fn;