boi what the hell is this update
This commit is contained in:
4 files changed
+286
-162
No files matched your search
+185
-104
@@ -7,29 +7,29 @@ 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("""
|
||||
randomlda = {APP_NAME: (1) // int("""
|
||||
.valueOf(), // cool!
|
||||
};
|
||||
const fn =
|
||||
/* """ * 0 + "1")} # fmt: skip
|
||||
# this is LOVELY
|
||||
|
||||
# ruff: noqa: E402
|
||||
# flake8: noqa: E402
|
||||
|
||||
# autopep8: off
|
||||
import asyncio
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
import sys
|
||||
from itertools import count
|
||||
from pathlib import Path
|
||||
|
||||
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,
|
||||
)
|
||||
@@ -44,9 +44,13 @@ CHROME_DATA_DIR.mkdir(exist_ok=True)
|
||||
|
||||
EF_ICON_PATH = Path(__file__).parent / "resources" / "ef_icon.ico"
|
||||
|
||||
LOGIN_TIMEOUT_MS = 4 * 60 * 1000
|
||||
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))
|
||||
|
||||
|
||||
async def launch_context(p: Playwright, *, headless: bool):
|
||||
return await p.chromium.launch_persistent_context(
|
||||
@@ -55,144 +59,221 @@ async def launch_context(p: Playwright, *, headless: bool):
|
||||
)
|
||||
|
||||
|
||||
async def notify(msg: str, title: str = "Login"):
|
||||
async def notify(msg: str, title: str = "Login", *, notifier=DESKTOP_NOTIFIER):
|
||||
"""Send a desktop notification.
|
||||
|
||||
desktop_notifier can't make callbacks that rerun the script... im crine.
|
||||
"""
|
||||
print(f"{title}: {msg}")
|
||||
|
||||
notifier = DesktopNotifier(app_name=APP_NAME, app_icon=Icon(EF_ICON_PATH))
|
||||
# lets see if this blocks
|
||||
await notifier.send(title=title, message=msg, timeout=5)
|
||||
# oops this spawns another event loop.
|
||||
|
||||
|
||||
async def needs_manual_login(page):
|
||||
return await page.locator("form").first.is_visible()
|
||||
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)
|
||||
|
||||
|
||||
async def already_claimed(container: Locator):
|
||||
# language=javascript
|
||||
fn = (
|
||||
"""
|
||||
// */
|
||||
fn = """// */
|
||||
/** @param {HTMLElement} el */
|
||||
(el) => {
|
||||
return [...el.parentElement.children].some(
|
||||
(x) => x !== el && x.id === "completed-overlay" && x.checkVisibility?.(),
|
||||
);
|
||||
};
|
||||
/* *"""
|
||||
+ "/"
|
||||
) # yes sir
|
||||
/* *\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)
|
||||
|
||||
|
||||
async def program():
|
||||
async with async_playwright() as p:
|
||||
context = await launch_context(p, headless=True)
|
||||
class LoginNeededError(Exception):
|
||||
"""Raised when manual login is required."""
|
||||
|
||||
page = await context.new_page()
|
||||
await page.goto(URL)
|
||||
|
||||
login_form = page.locator("form").first
|
||||
container: Locator = page.locator("#lottie-container").first
|
||||
async def click_loop(page: Page):
|
||||
"""Loop that attempts to click the claim button until successful or already claimed.
|
||||
|
||||
deadline = time.time() + 60
|
||||
May run indefinitely if the claim button is not visible or the page is not ready."""
|
||||
container: Locator = page.locator("#lottie-container").first
|
||||
|
||||
while time.time() < deadline:
|
||||
# auth expired
|
||||
if await login_form.is_visible():
|
||||
await notify(
|
||||
"Please log in manually and close the browser. You have 4 minutes.",
|
||||
title="Manual Login Required",
|
||||
)
|
||||
await container.wait_for(timeout=0) # wait for visible
|
||||
|
||||
await context.close()
|
||||
# at this point, the page is ready.
|
||||
if await already_claimed(container):
|
||||
await notify(
|
||||
"Work already completed elsewhere.",
|
||||
title="Already Claimed",
|
||||
)
|
||||
return
|
||||
|
||||
headful = await launch_context(
|
||||
p,
|
||||
headless=False,
|
||||
)
|
||||
|
||||
page = headful.pages[0] if headful.pages else await headful.new_page()
|
||||
|
||||
await page.goto(URL)
|
||||
|
||||
try:
|
||||
await headful.wait_for_event(
|
||||
"close",
|
||||
timeout=LOGIN_TIMEOUT_MS,
|
||||
)
|
||||
|
||||
await notify(
|
||||
"Browser closed. Please rerun the script.",
|
||||
title="Browser Closed",
|
||||
)
|
||||
|
||||
except PlaywrightTimeoutError:
|
||||
await notify(
|
||||
"Login timeout reached. Please rerun the script.",
|
||||
title="Login Timeout",
|
||||
)
|
||||
|
||||
try:
|
||||
await headful.close()
|
||||
except PlaywrightError:
|
||||
pass
|
||||
|
||||
return
|
||||
|
||||
# page not ready yet
|
||||
if not await container.is_visible():
|
||||
await page.wait_for_timeout(200)
|
||||
continue
|
||||
|
||||
# already claimed
|
||||
if (not await login_form.is_visible()) and (
|
||||
await already_claimed(container)
|
||||
): # toctou here is crazy
|
||||
await notify(
|
||||
"Work already completed elsewhere.",
|
||||
title="Already Claimed",
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
await container.click(
|
||||
timeout=3000,
|
||||
delay=random.randint(100, 300),
|
||||
# we checked for visibility manually. This will probs work.
|
||||
force=True,
|
||||
)
|
||||
for t in count(1): # can be interrupted by timeout at any await point
|
||||
try:
|
||||
await container.click(
|
||||
timeout=3000,
|
||||
delay=random.randint(100, 300),
|
||||
# we checked for visibility manually. This will probs work.
|
||||
force=True,
|
||||
)
|
||||
except (
|
||||
PlaywrightTimeoutError,
|
||||
PlaywrightError,
|
||||
):
|
||||
pass
|
||||
|
||||
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",
|
||||
)
|
||||
return
|
||||
|
||||
return
|
||||
await asyncio.sleep(2)
|
||||
|
||||
except (
|
||||
PlaywrightTimeoutError,
|
||||
PlaywrightError,
|
||||
):
|
||||
pass
|
||||
|
||||
await page.wait_for_timeout(1000)
|
||||
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)
|
||||
# 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(
|
||||
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:
|
||||
return False
|
||||
|
||||
if tasks[0] in done:
|
||||
tasks[0].result()
|
||||
raise LoginNeededError("Manual login required.")
|
||||
if tasks[1] in done:
|
||||
tasks[1].result()
|
||||
return True
|
||||
if tasks[2] in done:
|
||||
return False # page closed
|
||||
|
||||
return False # should never happen, but just in case
|
||||
|
||||
|
||||
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)
|
||||
|
||||
await notify(
|
||||
"Timed out waiting for claim state.",
|
||||
title="Timeout",
|
||||
f"Please log in manually and close the browser. You have {LOGIN_TIMEOUT_MS // (60 * 1000)} minute(s).",
|
||||
title="Manual Login Required",
|
||||
)
|
||||
|
||||
try:
|
||||
await context.close()
|
||||
except PlaywrightError:
|
||||
pass
|
||||
await context.wait_for_event(
|
||||
"close",
|
||||
timeout=LOGIN_TIMEOUT_MS,
|
||||
)
|
||||
except PlaywrightTimeoutError, TimeoutError:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def program2():
|
||||
async with async_playwright() as p:
|
||||
for i in range(2):
|
||||
try:
|
||||
if await claim_loop(p):
|
||||
return
|
||||
|
||||
await notify(
|
||||
"Timed out trying to claim the reward. Please rerun the script.",
|
||||
title="Timeout",
|
||||
)
|
||||
raise SystemExit(1)
|
||||
except LoginNeededError:
|
||||
if i > 0:
|
||||
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",
|
||||
)
|
||||
if not await manual_login(p):
|
||||
await notify(
|
||||
"Login timeout reached. Please rerun the script.",
|
||||
title="Login Timeout",
|
||||
)
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
def main(): # uv no main fix
|
||||
asyncio.run(program())
|
||||
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:
|
||||
# pondering using BaseException
|
||||
await notify(
|
||||
f"An error occurred\n{e!s}",
|
||||
title="Error",
|
||||
)
|
||||
raise
|
||||
|
||||
return asyncio.run(run_program())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
raise SystemExit(main())
|
||||
# async, also this file is valid js. */ fn;
|
||||
Reference in new issue
Block a user