From f3e43ede927a10dc6c98106da3de5bb16a229b4c Mon Sep 17 00:00:00 2001 From: lda Date: Sun, 1 Nov 2026 06:12:56 +0700 Subject: [PATCH] logging, timeout warning --- playwright_demo.py | 164 +++++++++++++++++++++++++++++++++++++++------ 1 file changed, 143 insertions(+), 21 deletions(-) diff --git a/playwright_demo.py b/playwright_demo.py index 399bab9..e839283 100644 --- a/playwright_demo.py +++ b/playwright_demo.py @@ -16,11 +16,16 @@ const fn = # 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 @@ -51,6 +56,13 @@ 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( @@ -59,19 +71,26 @@ async def launch_context(p: Playwright, *, headless: bool): ) -async def notify(msg: str, title: str = "Login", *, notifier=DESKTOP_NOTIFIER): +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}") - - await notifier.send(title=title, message=msg, timeout=5) + # 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." - return await page.locator("form").first.wait_for(timeout=timeout) + await page.locator("form").first.wait_for(timeout=timeout) + logger.info("Manual login detected, form is visible.") async def already_claimed(container: Locator): @@ -109,17 +128,21 @@ async def click_loop(page: Page): 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, @@ -127,11 +150,12 @@ async def click_loop(page: Page): # we checked for visibility manually. This will probs work. force=True, ) - except ( - PlaywrightTimeoutError, - PlaywrightError, - ): - pass + 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: @@ -144,6 +168,8 @@ async def click_loop(page: Page): f"Claimed successfully after {t} attempts.", title="Success", ) + + logger.info("Claimed successfully after %d attempts.", t) return await asyncio.sleep(2) @@ -165,15 +191,16 @@ async def claim_loop(p: Playwright) -> bool: async with await launch_context(p, headless=True) as context: page = await context.new_page() await page.goto(URL) - # the following code is BUNS + 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, _pending = await asyncio.wait( + 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 @@ -184,20 +211,66 @@ async def claim_loop(p: Playwright) -> bool: 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. @@ -208,27 +281,75 @@ async def manual_login(p: Playwright) -> bool: 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)), ) - try: - await context.wait_for_event( + close_task = asyncio.create_task( + context.wait_for_event( "close", timeout=LOGIN_TIMEOUT_MS, ) - except PlaywrightTimeoutError, TimeoutError: - return False + ) + 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( @@ -238,21 +359,22 @@ async def program2(): 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) - await notify( - "Manual login required. A new browser window will open.", - title="Manual Login Required", - ) + 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 @@ -264,7 +386,7 @@ def main(): # uv no main fix else: await program2() except Exception as e: - # pondering using BaseException + logger.exception("Unhandled exception in sign-in helper.") await notify( f"An error occurred\n{e!s}", title="Error",