From afccf81f8f8e4f88a6628c0f9169b637d3c50969 Mon Sep 17 00:00:00 2001 From: Mally Dang Date: Wed, 2 Sep 2026 12:54:14 +0700 Subject: [PATCH 1/4] feat: spoof the rewards app headers --- src/main.py | 171 +++++++++++++++++++++++++++++----------------------- 1 file changed, 94 insertions(+), 77 deletions(-) diff --git a/src/main.py b/src/main.py index 1b9e362..fbae9f3 100644 --- a/src/main.py +++ b/src/main.py @@ -8,108 +8,125 @@ import rewards_tasks from selenium import webdriver from selenium.common.exceptions import SessionNotCreatedException -HEADLESS = os.environ.get("REWARDS_HEADLESS", "").strip().lower() in ("1", "true", "yes") +HEADLESS = os.environ.get("REWARDS_HEADLESS", "").strip().lower() in ( + "1", + "true", + "yes", +) logger = logging.getLogger(__name__) def build_options(account: accounts.Account) -> webdriver.EdgeOptions: - options = webdriver.EdgeOptions() + options = webdriver.EdgeOptions() - options.add_experimental_option("excludeSwitches", ["enable-automation"]) - options.add_experimental_option('useAutomationExtension', False) - options.add_argument("--disable-blink-features=AutomationControlled") - options.add_argument(f"--user-data-dir={account.user_data_dir}") - options.add_argument(f"--profile-directory={account.profile_name}") + options.add_experimental_option("excludeSwitches", ["enable-automation"]) + options.add_experimental_option("useAutomationExtension", False) + options.add_argument("--disable-blink-features=AutomationControlled") + options.add_argument(f"--user-data-dir={account.user_data_dir}") + options.add_argument(f"--profile-directory={account.profile_name}") - if HEADLESS: - # A container has no display. The window size is set explicitly because - # the pointer code works in viewport coordinates, and the default - # headless window is small enough to put cards out of reach. - options.add_argument("--headless=new") - options.add_argument("--window-size=1920,1080") - options.add_argument("--no-sandbox") - options.add_argument("--disable-dev-shm-usage") + if HEADLESS: + # A container has no display. The window size is set explicitly because + # the pointer code works in viewport coordinates, and the default + # headless window is small enough to put cards out of reach. + options.add_argument("--headless=new") + options.add_argument("--window-size=1920,1080") + options.add_argument("--no-sandbox") + options.add_argument("--disable-dev-shm-usage") - return options + return options def run_account(account: accounts.Account) -> bool: - """Work one account. Returns whether the browser started.""" - try: - driver = webdriver.Edge(options=build_options(account)) - except SessionNotCreatedException as exc: - # Chromium allows one process per user data directory. When the profile - # is already open the driver's copy exits during startup, and selenium - # reports it as the browser crashing with a message that names neither - # the profile nor the other window. - logger.error("[FAIL] %s: could not start Edge with this profile.", account.name) - logger.error(" profile directory: %s", account.user_data_dir) - logger.error(" The usual cause is that this profile is already open in another") - logger.error(" Edge window, including one left over from a previous run.") - logger.error(" driver said: %s", log_utils.exception_summary(exc)) + """Work one account. Returns whether the browser started.""" + try: + driver = webdriver.Edge(options=build_options(account)) + # Set headers to spoof the rewards app for the rewards only quests + driver.execute_cdp_cmd("Network.enable", {}) + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36 Edg/151.0.0.0 MSRewards/Desktop/1.1.0", + "X-Rewards-Source": "msrewards-desktop", + } - return False + driver.execute_cdp_cmd("Network.setExtraHTTPHeaders", {"headers": headers}) + except SessionNotCreatedException as exc: + # Chromium allows one process per user data directory. When the profile + # is already open the driver's copy exits during startup, and selenium + # reports it as the browser crashing with a message that names neither + # the profile nor the other window. + logger.error("[FAIL] %s: could not start Edge with this profile.", account.name) + logger.error(" profile directory: %s", account.user_data_dir) + logger.error( + " The usual cause is that this profile is already open in another" + ) + logger.error(" Edge window, including one left over from a previous run.") + logger.error(" driver said: %s", log_utils.exception_summary(exc)) - try: - rewards = rewards_tasks.RewardsTaskUtils(driver) - rewards.complete_all_tasks() - finally: - try: - driver.quit() - except Exception as exc: - # quit() raises when the browser is already gone. Letting it out - # here would replace whatever actually went wrong with the tidy-up's - # own error, and the process it is meant to end is dead anyway. - logger.warning( - "%s: the driver did not shut down cleanly: %s", - account.name, log_utils.exception_summary(exc) - ) + return False - return True + try: + rewards = rewards_tasks.RewardsTaskUtils(driver) + rewards.complete_all_tasks() + finally: + try: + driver.quit() + except Exception as exc: + # quit() raises when the browser is already gone. Letting it out + # here would replace whatever actually went wrong with the tidy-up's + # own error, and the process it is meant to end is dead anyway. + logger.warning( + "%s: the driver did not shut down cleanly: %s", + account.name, + log_utils.exception_summary(exc), + ) + + return True def main() -> int: - log_utils.setup_logging() + log_utils.setup_logging() - try: - configured = accounts.configured() - except ValueError as exc: - logger.error("[FAIL] %s", exc) + try: + configured = accounts.configured() + except ValueError as exc: + logger.error("[FAIL] %s", exc) - return 2 + return 2 - started = 0 + started = 0 - for account in configured: - if len(configured) > 1: - logger.info("=== account: %s ===", account.name) + for account in configured: + if len(configured) > 1: + logger.info("=== account: %s ===", account.name) - # One account must not be able to end the batch. complete_all_tasks - # already contains a task that fails, and run_account names the profile - # that is already open, but everything else - a driver that will not - # start for some other reason, the browser dying mid-run, a page that - # never loads - reached here and took the remaining accounts with it. - # KeyboardInterrupt is deliberately not caught: Ctrl-C means stop. - try: - if run_account(account): - started += 1 - except Exception as exc: - logger.error( - "[FAIL] %s: %s: %s", - account.name, type(exc).__name__, log_utils.exception_summary(exc), - exc_info=logger.isEnabledFor(logging.DEBUG) - ) + # One account must not be able to end the batch. complete_all_tasks + # already contains a task that fails, and run_account names the profile + # that is already open, but everything else - a driver that will not + # start for some other reason, the browser dying mid-run, a page that + # never loads - reached here and took the remaining accounts with it. + # KeyboardInterrupt is deliberately not caught: Ctrl-C means stop. + try: + if run_account(account): + started += 1 + except Exception as exc: + logger.error( + "[FAIL] %s: %s: %s", + account.name, + type(exc).__name__, + log_utils.exception_summary(exc), + exc_info=logger.isEnabledFor(logging.DEBUG), + ) - if len(configured) > 1: - logger.info("%s/%s accounts ran", started, len(configured)) + if len(configured) > 1: + logger.info("%s/%s accounts ran", started, len(configured)) - # Nothing is watching a container, and stdin is not a terminal there. - if not HEADLESS: - input("Press Enter to exit...") + # Nothing is watching a container, and stdin is not a terminal there. + if not HEADLESS: + input("Press Enter to exit...") - return 0 if started else 1 + return 0 if started else 1 if __name__ == "__main__": - sys.exit(main()) + sys.exit(main()) From 15782d183673ca7806f99b69900669f3657df23e Mon Sep 17 00:00:00 2001 From: mardausdennis <71312763+mardausdennis@users.noreply.github.com> Date: Thu, 3 Sep 2026 22:38:01 +0200 Subject: [PATCH 2/4] tell absence and a slow render apart when a task fails Every failed task printed the same reason, because the blanket catch in complete_all_tasks treated NoSuchElementException and TimeoutException as one finding. A wait that expires says nothing about whether the section was there, since wait_for_element swallowed whatever the getter raised and returned False. A market that does not ship visual search and a points panel that was on screen and slow both arrived as TimeoutException wait_for_element now keeps the last reason the getter gave and raises ElementNeverAppeared when nothing was in the DOM for the whole wait. _container_by_id raises ElementNotReady when the id is present but no visible copy has content yet, which is the hydrating case it used to report as missing. Both subclass what they refine, so claim_bonus_points and complete_bing_daily_set keep working untouched A task that was reached and then ran out of time is now a [FAIL] rather than a [SKIP], because it may have left points behind. Absence stays a [SKIP] 56 tests pass, 20 of them new. A mutation check backs them, forcing every expired wait to count as absence fails 2, and collapsing the two branches back into one [SKIP] fails 4 Refs #52 --- src/element_selectors.py | 25 ++- src/rewards_tasks.py | 87 ++++++++++- tests/test_element_selectors.py | 13 ++ tests/test_task_outcomes.py | 262 ++++++++++++++++++++++++++++++++ 4 files changed, 378 insertions(+), 9 deletions(-) create mode 100644 tests/test_task_outcomes.py diff --git a/src/element_selectors.py b/src/element_selectors.py index 0108590..a59ff82 100644 --- a/src/element_selectors.py +++ b/src/element_selectors.py @@ -6,6 +6,20 @@ from selenium.common.exceptions import NoSuchElementException, StaleElementRefer from selenium import webdriver +class ElementNotReady(NoSuchElementException): + """The element is in the page but not usable yet. + + A section this market does not ship and a section that has not finished + hydrating both reach the caller as NoSuchElementException, which is why a + run could report "not available in this UI variant" for something that was + on screen. They need different messages and different next steps, so the + second case gets its own type. + + Subclassed rather than separate, so every existing `except + NoSuchElementException` keeps catching it. + """ + + class Labels: """Visible labels the selectors match on. @@ -45,7 +59,9 @@ class ElementSelectionUtils: pick the copy that is visible and actually has content. Anything the current variant does not ship raises NoSuchElementException so - the caller can skip that task instead of aborting the whole run. + the caller can skip that task instead of aborting the whole run. Something + that is present but not usable yet raises ElementNotReady instead, because + skipping it is the wrong answer and so is the message that goes with it. """ def __init__(self, driver: webdriver.Edge): @@ -66,6 +82,11 @@ class ElementSelectionUtils: their `.text` is empty, so returning one produces silent no-ops further up. Raising instead lets the caller's WebDriverWait retry while the page finishes hydrating. + + The two failures are not the same finding. No element with the id means + this variant does not ship the section. An id that is there but has no + usable copy means it is still rendering, so that one raises + ElementNotReady. """ matches = self.driver.find_elements(By.ID, element_id) @@ -79,7 +100,7 @@ class ElementSelectionUtils: except StaleElementReferenceException: continue - raise NoSuchElementException( + raise ElementNotReady( f"{element_id!r} is present but no visible copy has content yet" ) diff --git a/src/rewards_tasks.py b/src/rewards_tasks.py index ffa1b42..c880c88 100644 --- a/src/rewards_tasks.py +++ b/src/rewards_tasks.py @@ -20,6 +20,48 @@ VISUAL_SEARCH_IMAGE_PATH = os.path.abspath("visual_search.jpg") logger = logging.getLogger(__name__) + +class ElementNeverAppeared(TimeoutException): + """A wait expired without the element ever being in the page. + + WebDriverWait reports only that the wait ran out, so a section this market + does not ship and a section that was on screen and slow arrived as the same + TimeoutException. Reporting both as "not available in this UI variant" was + wrong for the second one, which is what #52 describes. + + Subclassed from TimeoutException so the handlers that already wait on a + control being absent, claim_bonus_points and complete_bing_daily_set, keep + working unchanged. + """ + + +def task_failure_report(exc: BaseException) -> tuple[str, str]: + """The tag and the reason a failed task is reported with. + + Absence and an expired wait need different next steps. A section this market + does not ship is nothing to act on, so it stays a [SKIP]. A section that was + on the page and never became usable may have left points behind, so it is + reported as a failure instead of being folded into the same sentence. + + Ordered from the most specific case outwards, not by exception hierarchy: + ElementNeverAppeared is a TimeoutException and ElementNotReady is a + NoSuchElementException, so each has to be tested before the class it + refines. + """ + unavailable = f"not available in this UI variant ({type(exc).__name__})" + + if isinstance(exc, ElementNeverAppeared): + return "SKIP", unavailable + + if isinstance(exc, (element_selectors.ElementNotReady, TimeoutException)): + return "FAIL", f"on the page but not ready in time ({type(exc).__name__})" + + if isinstance(exc, NoSuchElementException): + return "SKIP", unavailable + + return "FAIL", f"{type(exc).__name__}: {log_utils.exception_summary(exc)}" + + class RewardsTaskUtils: def __init__(self, driver: webdriver.Edge): self.driver = driver @@ -37,15 +79,45 @@ class RewardsTaskUtils: return self.driver.find_element(By.XPATH, xpath) def wait_for_element(self, element_getter: Callable[[], WebElement | list[WebElement]], timeout: int = 10) -> WebElement | list[WebElement]: + # Keep the last reason the getter gave. Without it a wait that expires + # cannot say whether the element was missing the whole time or was on + # the page and not ready, and those are reported differently. + last_error: BaseException | None = None + def condition(_: webdriver.Edge): + nonlocal last_error + try: element_or_elements = element_getter() + except Exception as exc: + # Exception rather than a bare except, so Ctrl+C during a + # getter ends the run instead of being retried away. + last_error = exc - return element_or_elements - except: return False - return WebDriverWait(self.driver, timeout).until(condition) + last_error = None + + return element_or_elements + + try: + return WebDriverWait(self.driver, timeout).until(condition) + except TimeoutException: + # A falsy return means the getter found something and rejected it, + # and ElementNotReady means it was there but still rendering. Only + # a plain NoSuchElementException every time means it was never + # there at all. + never_there = ( + isinstance(last_error, NoSuchElementException) + and not isinstance(last_error, element_selectors.ElementNotReady) + ) + + if not never_there: + raise + + raise ElementNeverAppeared( + f"nothing matched during the {timeout}s wait: {log_utils.exception_summary(last_error)}" + ) from last_error def switch_to_earn_page(self): self.move_to_and_click(self.elements.get_earn_tab()) @@ -303,11 +375,12 @@ class RewardsTaskUtils: try: step() logger.info("[OK] %s", name) - except (NoSuchElementException, TimeoutException) as exc: - logger.warning("[SKIP] %s: not available in this UI variant (%s)", name, type(exc).__name__) except Exception as exc: - logger.error( - "[FAIL] %s: %s: %s", name, type(exc).__name__, log_utils.exception_summary(exc), + tag, reason = task_failure_report(exc) + + logger.log( + logging.WARNING if tag == "SKIP" else logging.ERROR, + "[%s] %s: %s", tag, name, reason, exc_info=logger.isEnabledFor(logging.DEBUG) ) diff --git a/tests/test_element_selectors.py b/tests/test_element_selectors.py index 3db8431..0af6da6 100644 --- a/tests/test_element_selectors.py +++ b/tests/test_element_selectors.py @@ -163,6 +163,19 @@ class DuplicatedContainer(unittest.TestCase): self.assertEqual(len(cards), 7) + def test_a_container_that_is_there_but_empty_is_not_reported_as_missing(self): + # Both failures used to raise NoSuchElementException, so a section that + # was on the page and still rendering got reported as one this market + # does not ship. Waiting is the answer to this one. + with self.assertRaises(element_selectors.ElementNotReady): + selectors_for(self._driver(visible_links=0, hidden_links=7)).get_all_misc_cards() + + def test_a_container_that_is_absent_is_reported_as_missing(self): + with self.assertRaises(NoSuchElementException) as caught: + selectors_for(FakeDriver()).get_all_misc_cards() + + self.assertNotIsInstance(caught.exception, element_selectors.ElementNotReady) + class DailySetOpener(unittest.TestCase): """The opener label has to be distinguished from the level up entry.""" diff --git a/tests/test_task_outcomes.py b/tests/test_task_outcomes.py new file mode 100644 index 0000000..7089ef5 --- /dev/null +++ b/tests/test_task_outcomes.py @@ -0,0 +1,262 @@ +"""Tests for what a run says about a task that did not complete. + +The reported reason used to be a guess. Every wait that expired and every +lookup that missed produced "not available in this UI variant", so a section +that was on the page and slow read exactly like one this market does not ship, +which is #52. These pin down which failures are absence and which are not. + +None of them need a browser. + + python -m unittest discover -s tests +""" + +import logging +import os +import sys +import types +import unittest + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from selenium.common.exceptions import ( + NoSuchElementException, + TimeoutException, + WebDriverException, +) + +import rewards_tasks +from element_selectors import ElementNotReady +from fakes import FakeDriver +from rewards_tasks import ElementNeverAppeared, task_failure_report + +# Long enough for one poll, short enough that the suite stays quick. +# WebDriverWait sleeps 0.5s between attempts, so a wait that expires costs +# about that regardless of the timeout asked for. +BRIEF = 0.05 + + +def make_tasks(): + """A RewardsTaskUtils without the browser its __init__ opens.""" + tasks = rewards_tasks.RewardsTaskUtils.__new__(rewards_tasks.RewardsTaskUtils) + + tasks.driver = FakeDriver() + tasks.tab_utils = types.SimpleNamespace(close_all_other_tabs=lambda: None) + + return tasks + + +class WaitClassification(unittest.TestCase): + """wait_for_element has to say why it gave up, not just that it did.""" + + def test_a_getter_that_never_finds_anything_is_absence(self): + def missing(): + raise NoSuchElementException("no button containing 'visual search streak'") + + with self.assertRaises(ElementNeverAppeared): + make_tasks().wait_for_element(missing, timeout=BRIEF) + + def test_a_section_that_is_still_rendering_is_not_absence(self): + # The id is in the page, no visible copy has content yet. Waiting + # longer is the answer here, skipping the task is not. + def not_ready(): + raise ElementNotReady("'moreactivities' is present but no visible copy has content yet") + + with self.assertRaises(TimeoutException) as caught: + make_tasks().wait_for_element(not_ready, timeout=BRIEF) + + self.assertNotIsInstance(caught.exception, ElementNeverAppeared) + + def test_a_getter_that_rejects_what_it_finds_is_not_absence(self): + # complete_bing_daily_set holds out for all three activities and + # returns False until they are there. The panel itself is open. + with self.assertRaises(TimeoutException) as caught: + make_tasks().wait_for_element(lambda: [], timeout=BRIEF) + + self.assertNotIsInstance(caught.exception, ElementNeverAppeared) + + def test_an_element_that_arrives_late_is_still_returned(self): + attempts = [] + + def slow(): + attempts.append(None) + + if len(attempts) < 2: + raise NoSuchElementException("not yet") + + return "the element" + + self.assertEqual(make_tasks().wait_for_element(slow, timeout=5), "the element") + # A single lucky first attempt would prove nothing about the retry. + self.assertGreater(len(attempts), 1) + + def test_the_getters_own_reason_survives(self): + def missing(): + raise NoSuchElementException("no button containing 'points breakdown'") + + with self.assertRaises(ElementNeverAppeared) as caught: + make_tasks().wait_for_element(missing, timeout=BRIEF) + + self.assertIn("points breakdown", str(caught.exception)) + + +class ExistingTimeoutHandlers(unittest.TestCase): + """The new type has to stay catchable where TimeoutException was.""" + + def test_it_is_still_a_timeout(self): + self.assertTrue(issubclass(ElementNeverAppeared, TimeoutException)) + + def test_having_no_bonus_points_is_still_only_a_warning(self): + # There is no Claim button when there is nothing to claim, so this + # path reaches the wait expecting to be disappointed. If the new type + # escaped its `except TimeoutException`, an ordinary run would start + # reporting a failed task every day. + def bonus_button(): + pass + + def claim_button(): + pass + + tasks = make_tasks() + tasks.switch_to_dashboard = lambda: None + tasks.elements = types.SimpleNamespace( + get_bonus_button_on_dashboard=bonus_button, + get_claim_bonus_points_button=claim_button, + ) + + def wait_for_then_click(getter, timeout=10): + if getter is claim_button: + raise ElementNeverAppeared("nothing matched during the 10s wait") + + tasks.wait_for_then_click = wait_for_then_click + + with self.assertLogs(rewards_tasks.logger, level=logging.WARNING) as captured: + tasks.claim_bonus_points() + + self.assertIn("no bonus points to claim", "\n".join(captured.output).lower()) + + +class FailureReport(unittest.TestCase): + def test_a_section_this_variant_does_not_ship_is_skipped(self): + tag, reason = task_failure_report( + NoSuchElementException("no element with id 'moreactivities'") + ) + + self.assertEqual(tag, "SKIP") + self.assertIn("not available in this UI variant", reason) + + def test_a_wait_that_never_saw_the_element_is_skipped(self): + tag, reason = task_failure_report(ElementNeverAppeared("nothing matched")) + + self.assertEqual(tag, "SKIP") + self.assertIn("not available in this UI variant", reason) + + def test_a_section_that_never_finished_rendering_is_not_skipped(self): + tag, reason = task_failure_report( + ElementNotReady("'moreactivities' is present but no visible copy has content yet") + ) + + self.assertEqual(tag, "FAIL") + self.assertNotIn("not available", reason) + + def test_an_expired_wait_is_not_skipped(self): + # The line in #52, reported for a panel that was on screen the whole + # time: "[SKIP] Required searches: not available in this UI variant + # (TimeoutException)". + tag, reason = task_failure_report(TimeoutException("Message: ")) + + self.assertEqual(tag, "FAIL") + self.assertNotIn("not available", reason) + + def test_the_exception_name_is_kept(self): + # It is the difference between a lookup that missed and a wait that + # expired, and someone pasting a log should not lose it. + self.assertIn("ElementNeverAppeared", task_failure_report(ElementNeverAppeared("x"))[1]) + self.assertIn("TimeoutException", task_failure_report(TimeoutException("x"))[1]) + + def test_anything_else_keeps_its_own_message(self): + tag, reason = task_failure_report(WebDriverException("chrome not reachable")) + + self.assertEqual(tag, "FAIL") + self.assertIn("WebDriverException", reason) + self.assertIn("chrome not reachable", reason) + + +class TaskLoop(unittest.TestCase): + """complete_all_tasks, with the six tasks replaced by recorded calls.""" + + STEPS = ( + ("Bing daily set", "complete_bing_daily_set"), + ("Explore on Bing", "complete_explore_on_bing_tasks"), + ("Visual search", "complete_visual_search"), + ("Misc cards", "complete_misc_cards"), + ("Required searches", "complete_required_searches"), + ("Bonus points", "claim_bonus_points"), + ) + + def setUp(self): + self.ran = [] + + def _tasks(self, failures=None): + failures = failures or {} + tasks = make_tasks() + + for _, attribute in self.STEPS: + def step(name=attribute): + self.ran.append(name) + + if name in failures: + raise failures[name] + + setattr(tasks, attribute, step) + + return tasks + + def _run(self, failures=None): + with self.assertLogs(rewards_tasks.logger, level=logging.INFO) as captured: + self._tasks(failures).complete_all_tasks() + + return "\n".join(captured.output) + + def test_a_failing_task_does_not_stop_the_ones_after_it(self): + self._run({"complete_visual_search": WebDriverException("chrome not reachable")}) + + self.assertEqual(self.ran, [attribute for _, attribute in self.STEPS]) + + def test_absence_and_an_expired_wait_read_differently(self): + output = self._run({ + "complete_visual_search": ElementNeverAppeared("nothing matched"), + "complete_required_searches": TimeoutException("Message: "), + }) + + self.assertIn("[SKIP] Visual search: not available in this UI variant", output) + self.assertIn("[FAIL] Required searches: on the page but not ready in time", output) + self.assertIn("[OK] Bing daily set", output) + + def test_a_section_that_never_rendered_is_not_called_unavailable(self): + output = self._run({ + "complete_misc_cards": ElementNotReady( + "'moreactivities' is present but no visible copy has content yet" + ), + }) + + self.assertIn("[FAIL] Misc cards: on the page but not ready in time", output) + self.assertNotIn("Misc cards: not available", output) + + def test_a_task_this_variant_does_not_ship_is_still_skipped(self): + output = self._run({ + "complete_explore_on_bing_tasks": NoSuchElementException( + "no Explore on Bing section in this UI variant" + ), + }) + + self.assertIn("[SKIP] Explore on Bing: not available in this UI variant", output) + + def test_an_unexpected_failure_still_reports_what_went_wrong(self): + output = self._run({"complete_misc_cards": WebDriverException("chrome not reachable")}) + + self.assertIn("[FAIL] Misc cards: WebDriverException", output) + self.assertIn("chrome not reachable", output) + + +if __name__ == "__main__": + unittest.main() From 3c6b29e8574b64b801344c7ba4f9e76d7ec4fa69 Mon Sep 17 00:00:00 2001 From: Mally Dang Date: Mon, 7 Sep 2026 18:06:43 +0700 Subject: [PATCH 3/4] chore: move header setting thing to `RewardsTaskUtils` --- src/main.py | 8 -------- src/rewards_tasks.py | 10 ++++++++++ 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/src/main.py b/src/main.py index fbae9f3..64cb5fc 100644 --- a/src/main.py +++ b/src/main.py @@ -42,14 +42,6 @@ def run_account(account: accounts.Account) -> bool: """Work one account. Returns whether the browser started.""" try: driver = webdriver.Edge(options=build_options(account)) - # Set headers to spoof the rewards app for the rewards only quests - driver.execute_cdp_cmd("Network.enable", {}) - headers = { - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36 Edg/151.0.0.0 MSRewards/Desktop/1.1.0", - "X-Rewards-Source": "msrewards-desktop", - } - - driver.execute_cdp_cmd("Network.setExtraHTTPHeaders", {"headers": headers}) except SessionNotCreatedException as exc: # Chromium allows one process per user data directory. When the profile # is already open the driver's copy exits during startup, and selenium diff --git a/src/rewards_tasks.py b/src/rewards_tasks.py index ffa1b42..595cf93 100644 --- a/src/rewards_tasks.py +++ b/src/rewards_tasks.py @@ -24,6 +24,16 @@ class RewardsTaskUtils: def __init__(self, driver: webdriver.Edge): self.driver = driver + # Set headers to spoof the rewards app for the rewards only quests + self.driver.execute_cdp_cmd("Network.enable", {}) + + headers = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36 Edg/151.0.0.0 MSRewards/Desktop/1.1.0", + "X-Rewards-Source": "msrewards-desktop", + } + + self.driver.execute_cdp_cmd("Network.setExtraHTTPHeaders", {"headers": headers}) + self.driver.get("https://rewards.bing.com/") self.tab_utils = tab_utils.TabUtils(driver) From de669da145321a585152e98039c90da5662208fe Mon Sep 17 00:00:00 2001 From: Carl Furtado Date: Mon, 7 Sep 2026 10:13:24 -0400 Subject: [PATCH 4/4] normalize indentation to tabs --- src/main.py | 168 +++---- src/random_image_for_visual_search.py | 676 +++++++++++++------------- src/visualize_bezier_distortions.py | 126 ++--- 3 files changed, 485 insertions(+), 485 deletions(-) diff --git a/src/main.py b/src/main.py index 64cb5fc..67e2db1 100644 --- a/src/main.py +++ b/src/main.py @@ -9,116 +9,116 @@ from selenium import webdriver from selenium.common.exceptions import SessionNotCreatedException HEADLESS = os.environ.get("REWARDS_HEADLESS", "").strip().lower() in ( - "1", - "true", - "yes", + "1", + "true", + "yes", ) logger = logging.getLogger(__name__) def build_options(account: accounts.Account) -> webdriver.EdgeOptions: - options = webdriver.EdgeOptions() + options = webdriver.EdgeOptions() - options.add_experimental_option("excludeSwitches", ["enable-automation"]) - options.add_experimental_option("useAutomationExtension", False) - options.add_argument("--disable-blink-features=AutomationControlled") - options.add_argument(f"--user-data-dir={account.user_data_dir}") - options.add_argument(f"--profile-directory={account.profile_name}") + options.add_experimental_option("excludeSwitches", ["enable-automation"]) + options.add_experimental_option("useAutomationExtension", False) + options.add_argument("--disable-blink-features=AutomationControlled") + options.add_argument(f"--user-data-dir={account.user_data_dir}") + options.add_argument(f"--profile-directory={account.profile_name}") - if HEADLESS: - # A container has no display. The window size is set explicitly because - # the pointer code works in viewport coordinates, and the default - # headless window is small enough to put cards out of reach. - options.add_argument("--headless=new") - options.add_argument("--window-size=1920,1080") - options.add_argument("--no-sandbox") - options.add_argument("--disable-dev-shm-usage") + if HEADLESS: + # A container has no display. The window size is set explicitly because + # the pointer code works in viewport coordinates, and the default + # headless window is small enough to put cards out of reach. + options.add_argument("--headless=new") + options.add_argument("--window-size=1920,1080") + options.add_argument("--no-sandbox") + options.add_argument("--disable-dev-shm-usage") - return options + return options def run_account(account: accounts.Account) -> bool: - """Work one account. Returns whether the browser started.""" - try: - driver = webdriver.Edge(options=build_options(account)) - except SessionNotCreatedException as exc: - # Chromium allows one process per user data directory. When the profile - # is already open the driver's copy exits during startup, and selenium - # reports it as the browser crashing with a message that names neither - # the profile nor the other window. - logger.error("[FAIL] %s: could not start Edge with this profile.", account.name) - logger.error(" profile directory: %s", account.user_data_dir) - logger.error( - " The usual cause is that this profile is already open in another" - ) - logger.error(" Edge window, including one left over from a previous run.") - logger.error(" driver said: %s", log_utils.exception_summary(exc)) + """Work one account. Returns whether the browser started.""" + try: + driver = webdriver.Edge(options=build_options(account)) + except SessionNotCreatedException as exc: + # Chromium allows one process per user data directory. When the profile + # is already open the driver's copy exits during startup, and selenium + # reports it as the browser crashing with a message that names neither + # the profile nor the other window. + logger.error("[FAIL] %s: could not start Edge with this profile.", account.name) + logger.error(" profile directory: %s", account.user_data_dir) + logger.error( + " The usual cause is that this profile is already open in another" + ) + logger.error(" Edge window, including one left over from a previous run.") + logger.error(" driver said: %s", log_utils.exception_summary(exc)) - return False + return False - try: - rewards = rewards_tasks.RewardsTaskUtils(driver) - rewards.complete_all_tasks() - finally: - try: - driver.quit() - except Exception as exc: - # quit() raises when the browser is already gone. Letting it out - # here would replace whatever actually went wrong with the tidy-up's - # own error, and the process it is meant to end is dead anyway. - logger.warning( - "%s: the driver did not shut down cleanly: %s", - account.name, - log_utils.exception_summary(exc), - ) + try: + rewards = rewards_tasks.RewardsTaskUtils(driver) + rewards.complete_all_tasks() + finally: + try: + driver.quit() + except Exception as exc: + # quit() raises when the browser is already gone. Letting it out + # here would replace whatever actually went wrong with the tidy-up's + # own error, and the process it is meant to end is dead anyway. + logger.warning( + "%s: the driver did not shut down cleanly: %s", + account.name, + log_utils.exception_summary(exc), + ) - return True + return True def main() -> int: - log_utils.setup_logging() + log_utils.setup_logging() - try: - configured = accounts.configured() - except ValueError as exc: - logger.error("[FAIL] %s", exc) + try: + configured = accounts.configured() + except ValueError as exc: + logger.error("[FAIL] %s", exc) - return 2 + return 2 - started = 0 + started = 0 - for account in configured: - if len(configured) > 1: - logger.info("=== account: %s ===", account.name) + for account in configured: + if len(configured) > 1: + logger.info("=== account: %s ===", account.name) - # One account must not be able to end the batch. complete_all_tasks - # already contains a task that fails, and run_account names the profile - # that is already open, but everything else - a driver that will not - # start for some other reason, the browser dying mid-run, a page that - # never loads - reached here and took the remaining accounts with it. - # KeyboardInterrupt is deliberately not caught: Ctrl-C means stop. - try: - if run_account(account): - started += 1 - except Exception as exc: - logger.error( - "[FAIL] %s: %s: %s", - account.name, - type(exc).__name__, - log_utils.exception_summary(exc), - exc_info=logger.isEnabledFor(logging.DEBUG), - ) + # One account must not be able to end the batch. complete_all_tasks + # already contains a task that fails, and run_account names the profile + # that is already open, but everything else - a driver that will not + # start for some other reason, the browser dying mid-run, a page that + # never loads - reached here and took the remaining accounts with it. + # KeyboardInterrupt is deliberately not caught: Ctrl-C means stop. + try: + if run_account(account): + started += 1 + except Exception as exc: + logger.error( + "[FAIL] %s: %s: %s", + account.name, + type(exc).__name__, + log_utils.exception_summary(exc), + exc_info=logger.isEnabledFor(logging.DEBUG), + ) - if len(configured) > 1: - logger.info("%s/%s accounts ran", started, len(configured)) + if len(configured) > 1: + logger.info("%s/%s accounts ran", started, len(configured)) - # Nothing is watching a container, and stdin is not a terminal there. - if not HEADLESS: - input("Press Enter to exit...") + # Nothing is watching a container, and stdin is not a terminal there. + if not HEADLESS: + input("Press Enter to exit...") - return 0 if started else 1 + return 0 if started else 1 if __name__ == "__main__": - sys.exit(main()) + sys.exit(main()) diff --git a/src/random_image_for_visual_search.py b/src/random_image_for_visual_search.py index 0cf22d2..06a0bee 100644 --- a/src/random_image_for_visual_search.py +++ b/src/random_image_for_visual_search.py @@ -43,10 +43,10 @@ MAX_BACKOFF = 300 session = requests.Session() session.headers.update({ - "User-Agent": ( - "RandomVisualSearchImage/1.1 " - "(contact: 12345rfdz@gmail.com)" - ) + "User-Agent": ( + "RandomVisualSearchImage/1.1 " + "(contact: 12345rfdz@gmail.com)" + ) }) @@ -55,17 +55,17 @@ session.headers.update({ # ============================================================ def clean_url(url): - """Remove query parameters from Wikimedia URLs.""" + """Remove query parameters from Wikimedia URLs.""" - parts = urlsplit(url) + parts = urlsplit(url) - return urlunsplit(( - parts.scheme, - parts.netloc, - parts.path, - "", - "", - )) + return urlunsplit(( + parts.scheme, + parts.netloc, + parts.path, + "", + "", + )) # ============================================================ @@ -73,32 +73,32 @@ def clean_url(url): # ============================================================ def wait_after_429(response, attempt): - """Wait according to Wikimedia's Retry-After header.""" + """Wait according to Wikimedia's Retry-After header.""" - retry_after = response.headers.get("Retry-After") + retry_after = response.headers.get("Retry-After") - if retry_after: - try: - wait_time = int(retry_after) - except ValueError: - wait_time = min( - 2 ** attempt, - MAX_BACKOFF, - ) - else: - wait_time = min( - 2 ** attempt, - MAX_BACKOFF, - ) + if retry_after: + try: + wait_time = int(retry_after) + except ValueError: + wait_time = min( + 2 ** attempt, + MAX_BACKOFF, + ) + else: + wait_time = min( + 2 ** attempt, + MAX_BACKOFF, + ) - wait_time = max(5, wait_time) + wait_time = max(5, wait_time) - print( - f"Rate limited. Waiting " - f"{wait_time} seconds..." - ) + print( + f"Rate limited. Waiting " + f"{wait_time} seconds..." + ) - time.sleep(wait_time) + time.sleep(wait_time) # ============================================================ @@ -106,51 +106,51 @@ def wait_after_429(response, attempt): # ============================================================ def download_image(url): - """Download image bytes from Wikimedia.""" + """Download image bytes from Wikimedia.""" - url = clean_url(url) + url = clean_url(url) - try: - response = session.get( - url, - timeout=30, - allow_redirects=True, - ) + try: + response = session.get( + url, + timeout=30, + allow_redirects=True, + ) - except requests.RequestException as e: - print(f"Download failed: {e}") - return None + except requests.RequestException as e: + print(f"Download failed: {e}") + return None - if response.status_code == 429: - wait_after_429(response, 1) - return None + if response.status_code == 429: + wait_after_429(response, 1) + return None - if response.status_code == 403: - print("Wikimedia returned 403 Forbidden.") - return None + if response.status_code == 403: + print("Wikimedia returned 403 Forbidden.") + return None - try: - response.raise_for_status() - except requests.RequestException as e: - print(f"HTTP error: {e}") - return None + try: + response.raise_for_status() + except requests.RequestException as e: + print(f"HTTP error: {e}") + return None - content_type = response.headers.get( - "Content-Type", - "", - ).lower() + content_type = response.headers.get( + "Content-Type", + "", + ).lower() - if not content_type.startswith("image/"): - print( - f"Not an image: {content_type}" - ) - return None + if not content_type.startswith("image/"): + print( + f"Not an image: {content_type}" + ) + return None - if not response.content: - print("Downloaded image is empty.") - return None + if not response.content: + print("Downloaded image is empty.") + return None - return response.content + return response.content # ============================================================ @@ -158,40 +158,40 @@ def download_image(url): # ============================================================ def convert_to_jpeg(image_data): - """Convert downloaded image bytes to JPEG.""" + """Convert downloaded image bytes to JPEG.""" - try: - with Image.open( - io.BytesIO(image_data) - ) as image: + try: + with Image.open( + io.BytesIO(image_data) + ) as image: - # JPEG does not support alpha (transparency). - # If the image has transparency (RGBA or LA), paste it over a white background. - if image.mode in ("RGBA", "LA") or (image.mode == "P" and "transparency" in image.info): - background = Image.new("RGB", image.size, (255, 255, 255)) - if image.mode == "P": - image = image.convert("RGBA") - background.paste(image, mask=image.split()[-1]) - jpeg_image = background - else: - jpeg_image = image.convert("RGB") + # JPEG does not support alpha (transparency). + # If the image has transparency (RGBA or LA), paste it over a white background. + if image.mode in ("RGBA", "LA") or (image.mode == "P" and "transparency" in image.info): + background = Image.new("RGB", image.size, (255, 255, 255)) + if image.mode == "P": + image = image.convert("RGBA") + background.paste(image, mask=image.split()[-1]) + jpeg_image = background + else: + jpeg_image = image.convert("RGB") - output = io.BytesIO() + output = io.BytesIO() - jpeg_image.save( - output, - format="JPEG", - quality=90, - optimize=True, - ) + jpeg_image.save( + output, + format="JPEG", + quality=90, + optimize=True, + ) - return output.getvalue() + return output.getvalue() - except Exception as e: - print( - f"JPEG conversion failed: {e}" - ) - return None + except Exception as e: + print( + f"JPEG conversion failed: {e}" + ) + return None # ============================================================ @@ -200,308 +200,308 @@ def convert_to_jpeg(image_data): def get_random_image(): - for attempt in range( - 1, - MAX_ATTEMPTS + 1, - ): + for attempt in range( + 1, + MAX_ATTEMPTS + 1, + ): - if attempt > 1: - time.sleep(REQUEST_DELAY) + if attempt > 1: + time.sleep(REQUEST_DELAY) - print( - f"\nAttempt " - f"{attempt}/{MAX_ATTEMPTS}" - ) + print( + f"\nAttempt " + f"{attempt}/{MAX_ATTEMPTS}" + ) - # ---------------------------------------------------- - # RANDOM FILE - # ---------------------------------------------------- + # ---------------------------------------------------- + # RANDOM FILE + # ---------------------------------------------------- - params = { - "action": "query", - "format": "json", + params = { + "action": "query", + "format": "json", - "generator": "random", - "grnnamespace": 6, - "grnlimit": 1, + "generator": "random", + "grnnamespace": 6, + "grnlimit": 1, - "prop": "imageinfo", + "prop": "imageinfo", - "iiprop": ( - "url|size|mime|dimensions" - ), + "iiprop": ( + "url|size|mime|dimensions" + ), - "iiurlwidth": THUMBNAIL_WIDTH, - } + "iiurlwidth": THUMBNAIL_WIDTH, + } - try: - response = session.get( - API_URL, - params=params, - timeout=20, - ) + try: + response = session.get( + API_URL, + params=params, + timeout=20, + ) - except requests.RequestException as e: - print(f"API request failed: {e}") - continue + except requests.RequestException as e: + print(f"API request failed: {e}") + continue - # ---------------------------------------------------- - # API RATE LIMIT - # ---------------------------------------------------- + # ---------------------------------------------------- + # API RATE LIMIT + # ---------------------------------------------------- - if response.status_code == 429: - wait_after_429( - response, - attempt, - ) - continue + if response.status_code == 429: + wait_after_429( + response, + attempt, + ) + continue - try: - response.raise_for_status() - data = response.json() + try: + response.raise_for_status() + data = response.json() - except ( - requests.RequestException, - ValueError, - ) as e: - print(f"API error: {e}") - continue + except ( + requests.RequestException, + ValueError, + ) as e: + print(f"API error: {e}") + continue - # ---------------------------------------------------- - # GET PAGE - # ---------------------------------------------------- + # ---------------------------------------------------- + # GET PAGE + # ---------------------------------------------------- - pages = ( - data - .get("query", {}) - .get("pages", {}) - ) + pages = ( + data + .get("query", {}) + .get("pages", {}) + ) - if not pages: - print("No page returned.") - continue + if not pages: + print("No page returned.") + continue - page = next( - iter(pages.values()) - ) + page = next( + iter(pages.values()) + ) - title = page.get( - "title", - "Unknown", - ) + title = page.get( + "title", + "Unknown", + ) - imageinfo = page.get( - "imageinfo" - ) + imageinfo = page.get( + "imageinfo" + ) - if not imageinfo: - print( - "No image information." - ) - continue + if not imageinfo: + print( + "No image information." + ) + continue - info = imageinfo[0] + info = imageinfo[0] - mime = info.get( - "mime", - "", - ) + mime = info.get( + "mime", + "", + ) - width = info.get( - "width", - 0, - ) + width = info.get( + "width", + 0, + ) - height = info.get( - "height", - 0, - ) + height = info.get( + "height", + 0, + ) - size = info.get( - "size", - 0, - ) + size = info.get( + "size", + 0, + ) - thumbnail_url = info.get( - "thumburl" - ) + thumbnail_url = info.get( + "thumburl" + ) - original_url = info.get( - "url" - ) + original_url = info.get( + "url" + ) - # ---------------------------------------------------- - # FILTER - # ---------------------------------------------------- + # ---------------------------------------------------- + # FILTER + # ---------------------------------------------------- - if mime not in { - "image/jpeg", - "image/png", - "image/webp", - }: - print( - f"Skipping unsupported type: " - f"{mime}" - ) - continue + if mime not in { + "image/jpeg", + "image/png", + "image/webp", + }: + print( + f"Skipping unsupported type: " + f"{mime}" + ) + continue - if width < MIN_WIDTH or height < MIN_HEIGHT: - print( - f"Skipping small image: " - f"{width}x{height}" - ) - continue + if width < MIN_WIDTH or height < MIN_HEIGHT: + print( + f"Skipping small image: " + f"{width}x{height}" + ) + continue - if size > MAX_FILE_SIZE: - print( - f"Skipping large image: " - f"{size / 1024 / 1024:.1f} MB" - ) - continue + if size > MAX_FILE_SIZE: + print( + f"Skipping large image: " + f"{size / 1024 / 1024:.1f} MB" + ) + continue - if not thumbnail_url: - print("No thumbnail URL.") - continue + if not thumbnail_url: + print("No thumbnail URL.") + continue - # ---------------------------------------------------- - # FOUND - # ---------------------------------------------------- + # ---------------------------------------------------- + # FOUND + # ---------------------------------------------------- - print(f"Found: {title}") - print( - f"Size: {width}x{height}" - ) + print(f"Found: {title}") + print( + f"Size: {width}x{height}" + ) - # ---------------------------------------------------- - # DOWNLOAD THUMBNAIL - # ---------------------------------------------------- + # ---------------------------------------------------- + # DOWNLOAD THUMBNAIL + # ---------------------------------------------------- - image_data = download_image( - thumbnail_url - ) + image_data = download_image( + thumbnail_url + ) - # ---------------------------------------------------- - # FALLBACK TO ORIGINAL - # ---------------------------------------------------- + # ---------------------------------------------------- + # FALLBACK TO ORIGINAL + # ---------------------------------------------------- - if image_data is None and original_url: - print( - "Trying original..." - ) + if image_data is None and original_url: + print( + "Trying original..." + ) - image_data = download_image( - original_url - ) + image_data = download_image( + original_url + ) - if image_data is None: - print( - "Couldn't download image." - ) - continue + if image_data is None: + print( + "Couldn't download image." + ) + continue - # ---------------------------------------------------- - # CONVERT TO JPEG - # ---------------------------------------------------- + # ---------------------------------------------------- + # CONVERT TO JPEG + # ---------------------------------------------------- - print("Converting to JPEG...") + print("Converting to JPEG...") - jpeg_data = convert_to_jpeg( - image_data - ) + jpeg_data = convert_to_jpeg( + image_data + ) - if jpeg_data is None: - continue + if jpeg_data is None: + continue - # ---------------------------------------------------- - # SAVE JPEG - # ---------------------------------------------------- + # ---------------------------------------------------- + # SAVE JPEG + # ---------------------------------------------------- - try: - OUTPUT_FILE.write_bytes( - jpeg_data - ) + try: + OUTPUT_FILE.write_bytes( + jpeg_data + ) - except OSError as e: - print( - f"Couldn't save image: {e}" - ) - continue + except OSError as e: + print( + f"Couldn't save image: {e}" + ) + continue - # ---------------------------------------------------- - # SAVE METADATA - # ---------------------------------------------------- + # ---------------------------------------------------- + # SAVE METADATA + # ---------------------------------------------------- - metadata = { - "title": title, - "source": "Wikimedia Commons", - "output_format": "JPEG", + metadata = { + "title": title, + "source": "Wikimedia Commons", + "output_format": "JPEG", - "width": width, - "height": height, + "width": width, + "height": height, - "original_mime": mime, + "original_mime": mime, - "original_size": size, + "original_size": size, - "jpeg_size": len( - jpeg_data - ), + "jpeg_size": len( + jpeg_data + ), - "original_url": ( - clean_url(original_url) - if original_url - else None - ), + "original_url": ( + clean_url(original_url) + if original_url + else None + ), - "thumbnail_url": ( - clean_url(thumbnail_url) - if thumbnail_url - else None - ), - } + "thumbnail_url": ( + clean_url(thumbnail_url) + if thumbnail_url + else None + ), + } - try: - METADATA_FILE.write_text( - json.dumps( - metadata, - indent=4, - ensure_ascii=False, - ), - encoding="utf-8", - ) + try: + METADATA_FILE.write_text( + json.dumps( + metadata, + indent=4, + ensure_ascii=False, + ), + encoding="utf-8", + ) - except OSError as e: - print( - f"Warning: couldn't save " - f"metadata: {e}" - ) + except OSError as e: + print( + f"Warning: couldn't save " + f"metadata: {e}" + ) - # ---------------------------------------------------- - # DONE - # ---------------------------------------------------- + # ---------------------------------------------------- + # DONE + # ---------------------------------------------------- - print() - print("=" * 50) - print("SUCCESS") - print("=" * 50) - print( - f"Image: " - f"{OUTPUT_FILE.absolute()}" - ) - print( - f"Size: " - f"{len(jpeg_data) / 1024:.1f} KB" - ) - print( - f"Source: {title}" - ) + print() + print("=" * 50) + print("SUCCESS") + print("=" * 50) + print( + f"Image: " + f"{OUTPUT_FILE.absolute()}" + ) + print( + f"Size: " + f"{len(jpeg_data) / 1024:.1f} KB" + ) + print( + f"Source: {title}" + ) - return metadata + return metadata - raise RuntimeError( - "Unable to obtain a suitable " - "Wikimedia image." - ) + raise RuntimeError( + "Unable to obtain a suitable " + "Wikimedia image." + ) # ============================================================ @@ -509,4 +509,4 @@ def get_random_image(): # ============================================================ if __name__ == "__main__": - get_random_image() \ No newline at end of file + get_random_image() \ No newline at end of file diff --git a/src/visualize_bezier_distortions.py b/src/visualize_bezier_distortions.py index ae7983d..a8d5350 100644 --- a/src/visualize_bezier_distortions.py +++ b/src/visualize_bezier_distortions.py @@ -19,87 +19,87 @@ Point = tuple[int, int] def make_paths() -> tuple[callable, callable]: - """Return independent paths with the same start and end points.""" - start = (150, 535) - end = (950, 535) - base_path = get_bezier_path(start, end, intermediate_radius_interval=(150, 210)) - distorted_path = get_distorted_bezier_path( - start, - end, - intermediate_radius_interval=(150, 210), - distortion_zone_time_length=0.08, - distortion_frequency=1.0, - deviation_interval=(10, 18), - ) + """Return independent paths with the same start and end points.""" + start = (150, 535) + end = (950, 535) + base_path = get_bezier_path(start, end, intermediate_radius_interval=(150, 210)) + distorted_path = get_distorted_bezier_path( + start, + end, + intermediate_radius_interval=(150, 210), + distortion_zone_time_length=0.08, + distortion_frequency=1.0, + deviation_interval=(10, 18), + ) - return base_path, distorted_path + return base_path, distorted_path def sample_path(path: callable, steps: int = 360) -> list[Point]: - return [ - (round(point[0]), round(point[1])) - for point in (path(index / (steps - 1)) for index in range(steps)) - ] + return [ + (round(point[0]), round(point[1])) + for point in (path(index / (steps - 1)) for index in range(steps)) + ] def draw_label(screen: pygame.Surface, font: pygame.font.Font, text: str, position: tuple[int, int], color: tuple[int, int, int]) -> None: - screen.blit(font.render(text, True, color), position) + screen.blit(font.render(text, True, color), position) def main() -> None: - pygame.init() - screen = pygame.display.set_mode(WINDOW_SIZE) - pygame.display.set_caption("Bezier Path Distortion") - clock = pygame.time.Clock() - title_font = pygame.font.SysFont("Segoe UI", 28, bold=True) - body_font = pygame.font.SysFont("Segoe UI", 19) - button_font = pygame.font.SysFont("Segoe UI", 18, bold=True) + pygame.init() + screen = pygame.display.set_mode(WINDOW_SIZE) + pygame.display.set_caption("Bezier Path Distortion") + clock = pygame.time.Clock() + title_font = pygame.font.SysFont("Segoe UI", 28, bold=True) + body_font = pygame.font.SysFont("Segoe UI", 19) + button_font = pygame.font.SysFont("Segoe UI", 18, bold=True) - base_path, distorted_path = make_paths() - base_points = sample_path(base_path) - distorted_points = sample_path(distorted_path) - show_distorted = False - button = pygame.Rect(405, 595, 290, 52) + base_path, distorted_path = make_paths() + base_points = sample_path(base_path) + distorted_points = sample_path(distorted_path) + show_distorted = False + button = pygame.Rect(405, 595, 290, 52) - running = True - while running: - for event in pygame.event.get(): - if event.type == pygame.QUIT: - running = False - elif event.type == pygame.MOUSEBUTTONDOWN and event.button == 1: - if button.collidepoint(event.pos): - show_distorted = not show_distorted + running = True + while running: + for event in pygame.event.get(): + if event.type == pygame.QUIT: + running = False + elif event.type == pygame.MOUSEBUTTONDOWN and event.button == 1: + if button.collidepoint(event.pos): + show_distorted = not show_distorted - screen.fill(BACKGROUND) - draw_label(screen, title_font, "Bezier path comparison", (38, 28), PATH_COLOR) - draw_label( - screen, - body_font, - "The red path adds temporary offsets to the same underlying curve.", - (40, 70), - (91, 97, 102), - ) + screen.fill(BACKGROUND) + draw_label(screen, title_font, "Bezier path comparison", (38, 28), PATH_COLOR) + draw_label( + screen, + body_font, + "The red path adds temporary offsets to the same underlying curve.", + (40, 70), + (91, 97, 102), + ) - pygame.draw.lines(screen, (184, 188, 190), False, base_points, 1) - pygame.draw.lines(screen, DISTORTED_PATH_COLOR if show_distorted else PATH_COLOR, False, distorted_points if show_distorted else base_points, 4) - pygame.draw.circle(screen, POINT_COLOR, base_points[0], 10) - pygame.draw.circle(screen, POINT_COLOR, base_points[-1], 10) + pygame.draw.lines(screen, (184, 188, 190), False, base_points, 1) + pygame.draw.lines(screen, DISTORTED_PATH_COLOR if show_distorted else PATH_COLOR, False, distorted_points if show_distorted else base_points, 4) + pygame.draw.circle(screen, POINT_COLOR, base_points[0], 10) + pygame.draw.circle(screen, POINT_COLOR, base_points[-1], 10) - draw_label(screen, body_font, "A", (base_points[0][0] - 8, base_points[0][1] + 18), PATH_COLOR) - draw_label(screen, body_font, "B", (base_points[-1][0] - 8, base_points[-1][1] + 18), PATH_COLOR) - draw_label(screen, body_font, "DISTORTED" if show_distorted else "UNDISTORTED", (20, 535), DISTORTED_PATH_COLOR if show_distorted else PATH_COLOR) + draw_label(screen, body_font, "A", (base_points[0][0] - 8, base_points[0][1] + 18), PATH_COLOR) + draw_label(screen, body_font, "B", (base_points[-1][0] - 8, base_points[-1][1] + 18), PATH_COLOR) + draw_label(screen, body_font, "DISTORTED" if show_distorted else "UNDISTORTED", (20, 535), DISTORTED_PATH_COLOR if show_distorted else PATH_COLOR) - button_color = BUTTON_HOVER_COLOR if button.collidepoint(pygame.mouse.get_pos()) else BUTTON_COLOR - pygame.draw.rect(screen, button_color, button, border_radius=7) - button_text = "Show undistorted path" if show_distorted else "Show distorted path" - text_surface = button_font.render(button_text, True, BUTTON_TEXT_COLOR) - screen.blit(text_surface, text_surface.get_rect(center=button.center)) + button_color = BUTTON_HOVER_COLOR if button.collidepoint(pygame.mouse.get_pos()) else BUTTON_COLOR + pygame.draw.rect(screen, button_color, button, border_radius=7) + button_text = "Show undistorted path" if show_distorted else "Show distorted path" + text_surface = button_font.render(button_text, True, BUTTON_TEXT_COLOR) + screen.blit(text_surface, text_surface.get_rect(center=button.center)) - pygame.display.flip() - clock.tick(60) + pygame.display.flip() + clock.tick(60) - pygame.quit() + pygame.quit() if __name__ == "__main__": - main() + main()