From a48a850db07c6306f52a80b2b993a876299ebfee Mon Sep 17 00:00:00 2001 From: break27 Date: Fri, 10 Jul 2026 15:22:59 +0800 Subject: [PATCH] update: tweaks on APIs --- pyproject.toml | 2 +- src/common/actionflow.py | 30 +++++++++++++++++++----------- src/common/jsonrpc2/client.js | 11 ++++------- src/common/jsonrpc2/server.py | 2 +- src/common/utils/__init__.py | 4 ++-- src/common/utils/selenium.py | 16 ++++++++-------- 6 files changed, 35 insertions(+), 30 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c2c30da..cd8a069 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta" [project] name = "common" description = "Commonly reusable code" -version = "0.1.17" +version = "0.1.18" requires-python = ">=3.13" authors = [ { name="BreakerBear", email="breakerbear@autistic.men" }, diff --git a/src/common/actionflow.py b/src/common/actionflow.py index 464c7be..bb91d74 100644 --- a/src/common/actionflow.py +++ b/src/common/actionflow.py @@ -20,15 +20,18 @@ class ActionFlow: self.indices: dict[str, int] = {} self.actions: list[type[Action]] = [] self.on: list[bool] = [] - - def queue(self, action: type[Action]): - name = action.__name__ - index = self.indices.get(name) - assert index is not None, "Action '%s' is not registered" % name - - if self.on[index]: raise Unavailable(name) - if self.on[index] is None: raise NotAllowed(name) - self.on[index] = bool(self.actions[index].prepare()) + + def __getitem__(self, key): + index = self.indices[key] + state = self.on[index] + + def inner(*args, **kwargs): + self.on[index] = self.actions[index].prepare(*args, **kwargs) + return self.on[index] + + if state: raise Unavailable(key) + if state is None: raise NotAllowed(key) + return inner def react(self, *stage: type[Action]): for Props in stage: @@ -39,21 +42,26 @@ class ActionFlow: for index in self.indices.values(): if (self.on[index]): self.on[index] = None - self.on[index] = bool(self.actions[index].perform()) + self.on[index] = self.actions[index].perform() def index(self, action: type[Action]) -> int: name = action.__name__ index = self.indices.get(name) assert index is not None, "Action '%s' is not registered" % name return index - + def allow(self, action: type[Action], value=True): index = self.index(action) self.on[index] = False if value else None + def stage(self, *preset: type[Action]): + for Props in preset: + self.append(Props) + def append(self, action: type[Action]): name = action.__name__ index = len(self.indices) + assert name not in self.indices, "Action '%s' is already registered" % name self.indices[name] = index self.actions.insert(index, action) diff --git a/src/common/jsonrpc2/client.js b/src/common/jsonrpc2/client.js index ae33853..6c12678 100644 --- a/src/common/jsonrpc2/client.js +++ b/src/common/jsonrpc2/client.js @@ -20,19 +20,16 @@ export const LogRecord = { }; export const Rpc2 = { - notify: async (method, ...args) => { - let request = { method }; - if (args.length > 0) request.params = args; + notify: async (method, options) => { + let request = { ...options, method }; let body = JSON.stringify(request); if (!navigator.sendBeacon('/', body)) { throw new Error('Data transmission failed'); } }, - invoke: async (method, ...args) => { + invoke: async (method, options) => { let id = Math.floor(Math.random() * 1000000000); - let request = { method, id }; - if (args.length > 0) request.params = args; - + let request = { ...options, method, id }; let body = JSON.stringify(request); let response = await fetch('/', { method: 'POST', body }); let json = await response.json(); diff --git a/src/common/jsonrpc2/server.py b/src/common/jsonrpc2/server.py index 033d60e..ce2a845 100644 --- a/src/common/jsonrpc2/server.py +++ b/src/common/jsonrpc2/server.py @@ -41,7 +41,7 @@ class Request[T]: if args.keys() != set(argnames): raise self.ParamsError(args) return handler(**args) - raise self.Invalid('Arguments unacceptable') + raise self.Invalid(type(args)) class Response[T]: def __init__(self, id: str|int|None, inner: T): diff --git a/src/common/utils/__init__.py b/src/common/utils/__init__.py index c2fa5c8..d49de6d 100644 --- a/src/common/utils/__init__.py +++ b/src/common/utils/__init__.py @@ -3,7 +3,7 @@ from common.utils.selenium import ( sleep, locate, click, - short, + identity, setup, ) @@ -12,6 +12,6 @@ __all__ = [ 'sleep', 'locate', 'click', - 'short', + 'identity', 'setup', ] diff --git a/src/common/utils/selenium.py b/src/common/utils/selenium.py index a831ca3..9291c0d 100644 --- a/src/common/utils/selenium.py +++ b/src/common/utils/selenium.py @@ -10,7 +10,7 @@ from selenium.webdriver.support.wait import WebDriverWait from selenium.common.exceptions import StaleElementReferenceException, NoAlertPresentException, TimeoutException driver: WebDriver|None = None -identity, parameters = None, dict() +unique, parameters = None, dict() def until(condition: typing.Callable[[WebDriver], bool], watch=True): try: @@ -50,10 +50,10 @@ def click(selector: str|WebElement, wait=True, condition=False): error = False element = locate(selector, wait, predicate) if isinstance(selector, str) else selector - counter = lambda: int(element.get_attribute(identity) or '0') + counter = lambda: int(element.get_attribute(unique) or '0') value = counter() - driver.execute_script("window.__%s__ = () => { arguments[0].setAttribute('%s', arguments[1] + 1) };" % ((identity,) * 2), element, value) - driver.execute_script("arguments[0].addEventListener('click', __%s__);" % identity, element) + driver.execute_script("window.__%s__ = () => { arguments[0].setAttribute('%s', arguments[1] + 1) };" % ((unique,) * 2), element, value) + driver.execute_script("arguments[0].addEventListener('click', __%s__);" % unique, element) for _ in range(parameters.get('attempts', 0)): try: @@ -70,14 +70,14 @@ def click(selector: str|WebElement, wait=True, condition=False): except TimeoutException: continue except: break - try: driver.execute_script("arguments[0].removeEventListener('click', __%s__);" % identity, element) + try: driver.execute_script("arguments[0].removeEventListener('click', __%s__);" % unique, element) except: pass -def short(): +def identity(): return ''.join(random.choices(string.digits + string.ascii_uppercase + string.ascii_lowercase, k=8)) def setup(a: WebDriver, b: dict): - global identity, driver - identity = short() + global unique, driver + unique = identity() driver = a parameters.update(b)