afspiegeling/rewards-farmer
1
0
Fork 0
mirror of https://github.com/User0332/rewards-farmer.git synced 2026-09-17 01:41:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Address code review: refactor move_to_and_click retries, simplify wait_for_then_click, resolve conflicts with main

Browse Source
This commit is contained in:
abhijitgorde5-crypto
2026-09-07 23:55:04 +05:30
parent 4c67f3f46e 4121c42552
commit 7686537dc1
7 changed files with 804 additions and 424 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/element_selectors.py
+23 -2
View File
@@ -6,6 +6,20 @@ from selenium.common.exceptions import NoSuchElementException, StaleElementRefer
from selenium import webdriver 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: class Labels:
"""Visible labels the selectors match on. """Visible labels the selectors match on.
@@ -45,7 +59,9 @@ class ElementSelectionUtils:
pick the copy that is visible and actually has content. pick the copy that is visible and actually has content.
Anything the current variant does not ship raises NoSuchElementException so 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): def __init__(self, driver: webdriver.Edge):
@@ -66,6 +82,11 @@ class ElementSelectionUtils:
their `.text` is empty, so returning one produces silent no-ops further their `.text` is empty, so returning one produces silent no-ops further
up. Raising instead lets the caller's WebDriverWait retry while the page up. Raising instead lets the caller's WebDriverWait retry while the page
finishes hydrating. 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) matches = self.driver.find_elements(By.ID, element_id)
@@ -79,7 +100,7 @@ class ElementSelectionUtils:
except StaleElementReferenceException: except StaleElementReferenceException:
continue continue
raise NoSuchElementException( raise ElementNotReady(
f"{element_id!r} is present but no visible copy has content yet" f"{element_id!r} is present but no visible copy has content yet"
) )
src/main.py
+15 -6
View File
@@ -8,7 +8,11 @@ import rewards_tasks
from selenium import webdriver from selenium import webdriver
from selenium.common.exceptions import SessionNotCreatedException 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__) logger = logging.getLogger(__name__)
@@ -17,7 +21,7 @@ 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("excludeSwitches", ["enable-automation"])
options.add_experimental_option('useAutomationExtension', False) options.add_experimental_option("useAutomationExtension", False)
options.add_argument("--disable-blink-features=AutomationControlled") options.add_argument("--disable-blink-features=AutomationControlled")
options.add_argument(f"--user-data-dir={account.user_data_dir}") options.add_argument(f"--user-data-dir={account.user_data_dir}")
options.add_argument(f"--profile-directory={account.profile_name}") options.add_argument(f"--profile-directory={account.profile_name}")
@@ -45,7 +49,9 @@ def run_account(account: accounts.Account) -> bool:
# the profile nor the other window. # the profile nor the other window.
logger.error("[FAIL] %s: could not start Edge with this profile.", account.name) logger.error("[FAIL] %s: could not start Edge with this profile.", account.name)
logger.error(" profile directory: %s", account.user_data_dir) 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(
" 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(" Edge window, including one left over from a previous run.")
logger.error(" driver said: %s", log_utils.exception_summary(exc)) logger.error(" driver said: %s", log_utils.exception_summary(exc))
@@ -63,7 +69,8 @@ def run_account(account: accounts.Account) -> bool:
# own error, and the process it is meant to end is dead anyway. # own error, and the process it is meant to end is dead anyway.
logger.warning( logger.warning(
"%s: the driver did not shut down cleanly: %s", "%s: the driver did not shut down cleanly: %s",
account.name, log_utils.exception_summary(exc) account.name,
log_utils.exception_summary(exc),
) )
return True return True
@@ -97,8 +104,10 @@ def main() -> int:
except Exception as exc: except Exception as exc:
logger.error( logger.error(
"[FAIL] %s: %s: %s", "[FAIL] %s: %s: %s",
account.name, type(exc).__name__, log_utils.exception_summary(exc), account.name,
exc_info=logger.isEnabledFor(logging.DEBUG) type(exc).__name__,
log_utils.exception_summary(exc),
exc_info=logger.isEnabledFor(logging.DEBUG),
) )
if len(configured) > 1: if len(configured) > 1:
src/random_image_for_visual_search.py
+322 -330
View File
@@ -43,10 +43,10 @@ MAX_BACKOFF = 300
session = requests.Session() session = requests.Session()
session.headers.update({ session.headers.update({
"User-Agent": ( "User-Agent": (
"RandomVisualSearchImage/1.1 " "RandomVisualSearchImage/1.1 "
"(contact: 12345rfdz@gmail.com)" "(contact: 12345rfdz@gmail.com)"
) )
}) })
@@ -55,17 +55,17 @@ session.headers.update({
# ============================================================ # ============================================================
def clean_url(url): def clean_url(url):
"""Remove query parameters from Wikimedia URLs.""" """Remove query parameters from Wikimedia URLs."""
parts = urlsplit(url) parts = urlsplit(url)
return urlunsplit(( return urlunsplit((
parts.scheme, parts.scheme,
parts.netloc, parts.netloc,
parts.path, parts.path,
"", "",
"", "",
)) ))
# ============================================================ # ============================================================
@@ -73,32 +73,32 @@ def clean_url(url):
# ============================================================ # ============================================================
def wait_after_429(response, attempt): 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: if retry_after:
try: try:
wait_time = int(retry_after) wait_time = int(retry_after)
except ValueError: except ValueError:
wait_time = min( wait_time = min(
2 ** attempt, 2 ** attempt,
MAX_BACKOFF, MAX_BACKOFF,
) )
else: else:
wait_time = min( wait_time = min(
2 ** attempt, 2 ** attempt,
MAX_BACKOFF, MAX_BACKOFF,
) )
wait_time = max(5, wait_time) wait_time = max(5, wait_time)
print( print(
f"Rate limited. Waiting " f"Rate limited. Waiting "
f"{wait_time} seconds..." 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): def download_image(url):
"""Download image bytes from Wikimedia.""" """Download image bytes from Wikimedia."""
url = clean_url(url) url = clean_url(url)
try: try:
response = session.get( response = session.get(
url, url,
timeout=30, timeout=30,
allow_redirects=True, allow_redirects=True,
) )
except (requests.RequestException, ConnectionResetError, OSError) as e: except (requests.RequestException, ConnectionResetError, OSError) as e:
print(f"Download failed: {e}") print(f"Download failed: {e}")
return None return None
if response.status_code == 429: if response.status_code == 429:
wait_after_429(response, 1) wait_after_429(response, 1)
return None return None
if response.status_code == 403: if response.status_code == 403:
print("Wikimedia returned 403 Forbidden.") print("Wikimedia returned 403 Forbidden.")
return None return None
try: try:
response.raise_for_status() response.raise_for_status()
except requests.RequestException as e: except requests.RequestException as e:
print(f"HTTP error: {e}") print(f"HTTP error: {e}")
return None return None
content_type = response.headers.get( content_type = response.headers.get(
"Content-Type", "Content-Type",
"", "",
).lower() ).lower()
if not content_type.startswith("image/"): if not content_type.startswith("image/"):
print( print(
f"Not an image: {content_type}" f"Not an image: {content_type}"
) )
return None return None
if not response.content: if not response.content:
print("Downloaded image is empty.") print("Downloaded image is empty.")
return None return None
return response.content return response.content
# ============================================================ # ============================================================
@@ -158,65 +158,65 @@ def download_image(url):
# ============================================================ # ============================================================
def convert_to_jpeg(image_data): def convert_to_jpeg(image_data):
"""Convert downloaded image bytes to JPEG.""" """Convert downloaded image bytes to JPEG."""
try: try:
with Image.open( with Image.open(
io.BytesIO(image_data) io.BytesIO(image_data)
) as image: ) as image:
# JPEG does not support alpha (transparency). # JPEG does not support alpha (transparency).
# If the image has transparency (RGBA or LA), paste it over a white background. # 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): if image.mode in ("RGBA", "LA") or (image.mode == "P" and "transparency" in image.info):
background = Image.new("RGB", image.size, (255, 255, 255)) background = Image.new("RGB", image.size, (255, 255, 255))
if image.mode == "P": if image.mode == "P":
image = image.convert("RGBA") image = image.convert("RGBA")
background.paste(image, mask=image.split()[-1]) background.paste(image, mask=image.split()[-1])
jpeg_image = background jpeg_image = background
else: else:
jpeg_image = image.convert("RGB") jpeg_image = image.convert("RGB")
output = io.BytesIO() output = io.BytesIO()
jpeg_image.save( jpeg_image.save(
output, output,
format="JPEG", format="JPEG",
quality=90, quality=90,
optimize=True, optimize=True,
) )
return output.getvalue() return output.getvalue()
except Exception as e: except Exception as e:
print( print(
f"JPEG conversion failed: {e}" f"JPEG conversion failed: {e}"
) )
return None return None
def generate_fallback_image(): def generate_fallback_image():
"""Generate a synthetic local JPEG image using PIL as a fallback.""" """Generate a synthetic local JPEG image using PIL as a fallback."""
print("[INFO] Generating synthetic local fallback image for visual search...") print("[INFO] Generating synthetic local fallback image for visual search...")
from PIL import ImageDraw from PIL import ImageDraw
import random import random
img = Image.new("RGB", (800, 600), color=(random.randint(50, 200), random.randint(50, 200), random.randint(50, 200))) img = Image.new("RGB", (800, 600), color=(random.randint(50, 200), random.randint(50, 200), random.randint(50, 200)))
draw = ImageDraw.Draw(img) draw = ImageDraw.Draw(img)
for _ in range(10): for _ in range(10):
x0 = random.randint(0, 700) x0 = random.randint(0, 700)
y0 = random.randint(0, 500) y0 = random.randint(0, 500)
x1 = x0 + random.randint(50, 200) x1 = x0 + random.randint(50, 200)
y1 = y0 + random.randint(50, 200) y1 = y0 + random.randint(50, 200)
fill = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)) fill = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255))
draw.rectangle([x0, y0, x1, y1], fill=fill) draw.rectangle([x0, y0, x1, y1], fill=fill)
output = io.BytesIO() output = io.BytesIO()
img.save(output, format="JPEG", quality=90) img.save(output, format="JPEG", quality=90)
jpeg_data = output.getvalue() jpeg_data = output.getvalue()
OUTPUT_FILE.write_bytes(jpeg_data) OUTPUT_FILE.write_bytes(jpeg_data)
print(f"Saved fallback image to {OUTPUT_FILE.absolute()}") print(f"Saved fallback image to {OUTPUT_FILE.absolute()}")
return {"title": "Fallback Synthetic Image", "width": 800, "height": 600} return {"title": "Fallback Synthetic Image", "width": 800, "height": 600}
# ============================================================ # ============================================================
@@ -225,262 +225,262 @@ def generate_fallback_image():
def get_random_image(): def get_random_image():
for attempt in range( for attempt in range(
1, 1,
MAX_ATTEMPTS + 1, MAX_ATTEMPTS + 1,
): ):
if attempt > 1: if attempt > 1:
time.sleep(REQUEST_DELAY) time.sleep(REQUEST_DELAY)
print( print(
f"\nAttempt " f"\nAttempt "
f"{attempt}/{MAX_ATTEMPTS}" f"{attempt}/{MAX_ATTEMPTS}"
) )
params = { params = {
"action": "query", "action": "query",
"format": "json", "format": "json",
"generator": "random", "generator": "random",
"grnnamespace": 6, "grnnamespace": 6,
"grnlimit": 1, "grnlimit": 1,
"prop": "imageinfo", "prop": "imageinfo",
"iiprop": ( "iiprop": (
"url|size|mime|dimensions" "url|size|mime|dimensions"
), ),
"iiurlwidth": THUMBNAIL_WIDTH, "iiurlwidth": THUMBNAIL_WIDTH,
} }
try: try:
response = session.get( response = session.get(
API_URL, API_URL,
params=params, params=params,
timeout=20, timeout=20,
) )
except (requests.RequestException, ConnectionResetError, OSError) as e: except (requests.RequestException, ConnectionResetError, OSError) as e:
print(f"API request failed: {e}") print(f"API request failed: {e}")
continue continue
if response.status_code == 429: if response.status_code == 429:
wait_after_429( wait_after_429(
response, response,
attempt, attempt,
) )
continue continue
try: try:
response.raise_for_status() response.raise_for_status()
data = response.json() data = response.json()
except ( except (
requests.RequestException, requests.RequestException,
ValueError, ValueError,
) as e: ) as e:
print(f"API error: {e}") print(f"API error: {e}")
continue continue
pages = ( pages = (
data data
.get("query", {}) .get("query", {})
.get("pages", {}) .get("pages", {})
) )
if not pages: if not pages:
print("No page returned.") print("No page returned.")
continue continue
page = next( page = next(
iter(pages.values()) iter(pages.values())
) )
title = page.get( title = page.get(
"title", "title",
"Unknown", "Unknown",
) )
imageinfo = page.get( imageinfo = page.get(
"imageinfo" "imageinfo"
) )
if not imageinfo: if not imageinfo:
print( print(
"No image information." "No image information."
) )
continue continue
info = imageinfo[0] info = imageinfo[0]
mime = info.get( mime = info.get(
"mime", "mime",
"", "",
) )
width = info.get( width = info.get(
"width", "width",
0, 0,
) )
height = info.get( height = info.get(
"height", "height",
0, 0,
) )
size = info.get( size = info.get(
"size", "size",
0, 0,
) )
thumbnail_url = info.get( thumbnail_url = info.get(
"thumburl" "thumburl"
) )
original_url = info.get( original_url = info.get(
"url" "url"
) )
if mime not in { if mime not in {
"image/jpeg", "image/jpeg",
"image/png", "image/png",
"image/webp", "image/webp",
}: }:
print( print(
f"Skipping unsupported type: " f"Skipping unsupported type: "
f"{mime}" f"{mime}"
) )
continue continue
if width < MIN_WIDTH or height < MIN_HEIGHT: if width < MIN_WIDTH or height < MIN_HEIGHT:
print( print(
f"Skipping small image: " f"Skipping small image: "
f"{width}x{height}" f"{width}x{height}"
) )
continue continue
if size > MAX_FILE_SIZE: if size > MAX_FILE_SIZE:
print( print(
f"Skipping large image: " f"Skipping large image: "
f"{size / 1024 / 1024:.1f} MB" f"{size / 1024 / 1024:.1f} MB"
) )
continue continue
if not thumbnail_url: if not thumbnail_url:
print("No thumbnail URL.") print("No thumbnail URL.")
continue continue
print(f"Found: {title}") print(f"Found: {title}")
print( print(
f"Size: {width}x{height}" f"Size: {width}x{height}"
) )
image_data = download_image( image_data = download_image(
thumbnail_url thumbnail_url
) )
if image_data is None and original_url: if image_data is None and original_url:
print( print(
"Trying original..." "Trying original..."
) )
image_data = download_image( image_data = download_image(
original_url original_url
) )
if image_data is None: if image_data is None:
print( print(
"Couldn't download image." "Couldn't download image."
) )
continue continue
print("Converting to JPEG...") print("Converting to JPEG...")
jpeg_data = convert_to_jpeg( jpeg_data = convert_to_jpeg(
image_data image_data
) )
if jpeg_data is None: if jpeg_data is None:
continue continue
try: try:
OUTPUT_FILE.write_bytes( OUTPUT_FILE.write_bytes(
jpeg_data jpeg_data
) )
except OSError as e: except OSError as e:
print( print(
f"Couldn't save image: {e}" f"Couldn't save image: {e}"
) )
continue continue
metadata = { metadata = {
"title": title, "title": title,
"source": "Wikimedia Commons", "source": "Wikimedia Commons",
"output_format": "JPEG", "output_format": "JPEG",
"width": width, "width": width,
"height": height, "height": height,
"original_mime": mime, "original_mime": mime,
"original_size": size, "original_size": size,
"jpeg_size": len( "jpeg_size": len(
jpeg_data jpeg_data
), ),
"original_url": ( "original_url": (
clean_url(original_url) clean_url(original_url)
if original_url if original_url
else None else None
), ),
"thumbnail_url": ( "thumbnail_url": (
clean_url(thumbnail_url) clean_url(thumbnail_url)
if thumbnail_url if thumbnail_url
else None else None
), ),
} }
try: try:
METADATA_FILE.write_text( METADATA_FILE.write_text(
json.dumps( json.dumps(
metadata, metadata,
indent=4, indent=4,
ensure_ascii=False, ensure_ascii=False,
), ),
encoding="utf-8", encoding="utf-8",
) )
except OSError as e: except OSError as e:
print( print(
f"Warning: couldn't save " f"Warning: couldn't save "
f"metadata: {e}" f"metadata: {e}"
) )
print() print()
print("=" * 50) print("=" * 50)
print("SUCCESS") print("SUCCESS")
print("=" * 50) print("=" * 50)
print( print(
f"Image: " f"Image: "
f"{OUTPUT_FILE.absolute()}" f"{OUTPUT_FILE.absolute()}"
) )
print( print(
f"Size: " f"Size: "
f"{len(jpeg_data) / 1024:.1f} KB" f"{len(jpeg_data) / 1024:.1f} KB"
) )
print( print(
f"Source: {title}" f"Source: {title}"
) )
return metadata return metadata
print("[WARNING] Could not download image from Wikimedia Commons. Generating local fallback image.") print("[WARNING] Could not download image from Wikimedia Commons. Generating local fallback image.")
return generate_fallback_image() return generate_fallback_image()
# ============================================================ # ============================================================
@@ -488,12 +488,4 @@ def get_random_image():
# ============================================================ # ============================================================
if __name__ == "__main__": if __name__ == "__main__":
get_random_image() get_random_image()
# ============================================================
# MAIN
# ============================================================
if __name__ == "__main__":
get_random_image()
src/rewards_tasks.py
+106 -23
View File
@@ -20,10 +20,62 @@ VISUAL_SEARCH_IMAGE_PATH = os.path.abspath("visual_search.jpg")
logger = logging.getLogger(__name__) 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: class RewardsTaskUtils:
def __init__(self, driver: webdriver.Edge): def __init__(self, driver: webdriver.Edge):
self.driver = driver 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.driver.get("https://rewards.bing.com/")
self.tab_utils = tab_utils.TabUtils(driver) self.tab_utils = tab_utils.TabUtils(driver)
@@ -48,15 +100,45 @@ class RewardsTaskUtils:
return self.driver.find_element(By.XPATH, xpath) 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]: 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): def condition(_: webdriver.Edge):
nonlocal last_error
try: try:
element_or_elements = element_getter() 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 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): def switch_to_earn_page(self):
self.move_to_and_click(self.elements.get_earn_tab()) self.move_to_and_click(self.elements.get_earn_tab())
@@ -65,25 +147,25 @@ class RewardsTaskUtils:
self.move_to_and_click(self.elements.get_dashboard_tab()) self.move_to_and_click(self.elements.get_dashboard_tab())
def move_to_and_click(self, elem_or_getter: WebElement | Callable[[], WebElement], retries: int = 3): def move_to_and_click(self, elem_or_getter: WebElement | Callable[[], WebElement], retries: int = 3):
for attempt in range(retries): if callable(elem_or_getter):
try: for attempt in range(retries):
if callable(elem_or_getter): try:
target_elem = elem_or_getter() target_elem = elem_or_getter()
else: self.mouse.move_to_element(target_elem)
target_elem = elem_or_getter self.mouse.human_like_click()
return
self.mouse.move_to_element(target_elem) except StaleElementReferenceException as exc:
self.mouse.human_like_click() if attempt == retries - 1:
return raise exc
except StaleElementReferenceException as exc: logger.warning("StaleElementReferenceException during click attempt %d/%d, retrying...", attempt + 1, retries)
if attempt == retries - 1: time.sleep(0.5)
raise exc else:
print(f"[WARNING] StaleElementReferenceException during click attempt {attempt + 1}/{retries}, retrying...") self.mouse.move_to_element(elem_or_getter)
time.sleep(0.5) self.mouse.human_like_click()
def wait_for_then_click(self, element_getter: Callable[[], WebElement], timeout: int = 10): def wait_for_then_click(self, element_getter: Callable[[], WebElement], timeout: int = 10):
elem = self.wait_for_element(element_getter, timeout) self.wait_for_element(element_getter, timeout)
self.move_to_and_click(element_getter if callable(element_getter) else elem) self.move_to_and_click(element_getter)
def complete_bing_daily_set(self, expected_activities: int = 3): def complete_bing_daily_set(self, expected_activities: int = 3):
self.switch_to_earn_page() self.switch_to_earn_page()
@@ -347,11 +429,12 @@ class RewardsTaskUtils:
try: try:
step() step()
logger.info("[OK] %s", name) 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: except Exception as exc:
logger.error( tag, reason = task_failure_report(exc)
"[FAIL] %s: %s: %s", name, type(exc).__name__, log_utils.exception_summary(exc),
logger.log(
logging.WARNING if tag == "SKIP" else logging.ERROR,
"[%s] %s: %s", tag, name, reason,
exc_info=logger.isEnabledFor(logging.DEBUG) exc_info=logger.isEnabledFor(logging.DEBUG)
) )
src/visualize_bezier_distortions.py
+63 -63
View File
@@ -19,87 +19,87 @@ Point = tuple[int, int]
def make_paths() -> tuple[callable, callable]: def make_paths() -> tuple[callable, callable]:
"""Return independent paths with the same start and end points.""" """Return independent paths with the same start and end points."""
start = (150, 535) start = (150, 535)
end = (950, 535) end = (950, 535)
base_path = get_bezier_path(start, end, intermediate_radius_interval=(150, 210)) base_path = get_bezier_path(start, end, intermediate_radius_interval=(150, 210))
distorted_path = get_distorted_bezier_path( distorted_path = get_distorted_bezier_path(
start, start,
end, end,
intermediate_radius_interval=(150, 210), intermediate_radius_interval=(150, 210),
distortion_zone_time_length=0.08, distortion_zone_time_length=0.08,
distortion_frequency=1.0, distortion_frequency=1.0,
deviation_interval=(10, 18), deviation_interval=(10, 18),
) )
return base_path, distorted_path return base_path, distorted_path
def sample_path(path: callable, steps: int = 360) -> list[Point]: def sample_path(path: callable, steps: int = 360) -> list[Point]:
return [ return [
(round(point[0]), round(point[1])) (round(point[0]), round(point[1]))
for point in (path(index / (steps - 1)) for index in range(steps)) 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: 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: def main() -> None:
pygame.init() pygame.init()
screen = pygame.display.set_mode(WINDOW_SIZE) screen = pygame.display.set_mode(WINDOW_SIZE)
pygame.display.set_caption("Bezier Path Distortion") pygame.display.set_caption("Bezier Path Distortion")
clock = pygame.time.Clock() clock = pygame.time.Clock()
title_font = pygame.font.SysFont("Segoe UI", 28, bold=True) title_font = pygame.font.SysFont("Segoe UI", 28, bold=True)
body_font = pygame.font.SysFont("Segoe UI", 19) body_font = pygame.font.SysFont("Segoe UI", 19)
button_font = pygame.font.SysFont("Segoe UI", 18, bold=True) button_font = pygame.font.SysFont("Segoe UI", 18, bold=True)
base_path, distorted_path = make_paths() base_path, distorted_path = make_paths()
base_points = sample_path(base_path) base_points = sample_path(base_path)
distorted_points = sample_path(distorted_path) distorted_points = sample_path(distorted_path)
show_distorted = False show_distorted = False
button = pygame.Rect(405, 595, 290, 52) button = pygame.Rect(405, 595, 290, 52)
running = True running = True
while running: while running:
for event in pygame.event.get(): for event in pygame.event.get():
if event.type == pygame.QUIT: if event.type == pygame.QUIT:
running = False running = False
elif event.type == pygame.MOUSEBUTTONDOWN and event.button == 1: elif event.type == pygame.MOUSEBUTTONDOWN and event.button == 1:
if button.collidepoint(event.pos): if button.collidepoint(event.pos):
show_distorted = not show_distorted show_distorted = not show_distorted
screen.fill(BACKGROUND) screen.fill(BACKGROUND)
draw_label(screen, title_font, "Bezier path comparison", (38, 28), PATH_COLOR) draw_label(screen, title_font, "Bezier path comparison", (38, 28), PATH_COLOR)
draw_label( draw_label(
screen, screen,
body_font, body_font,
"The red path adds temporary offsets to the same underlying curve.", "The red path adds temporary offsets to the same underlying curve.",
(40, 70), (40, 70),
(91, 97, 102), (91, 97, 102),
) )
pygame.draw.lines(screen, (184, 188, 190), False, base_points, 1) 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.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[0], 10)
pygame.draw.circle(screen, POINT_COLOR, base_points[-1], 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, "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, "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, "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 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) pygame.draw.rect(screen, button_color, button, border_radius=7)
button_text = "Show undistorted path" if show_distorted else "Show distorted path" button_text = "Show undistorted path" if show_distorted else "Show distorted path"
text_surface = button_font.render(button_text, True, BUTTON_TEXT_COLOR) text_surface = button_font.render(button_text, True, BUTTON_TEXT_COLOR)
screen.blit(text_surface, text_surface.get_rect(center=button.center)) screen.blit(text_surface, text_surface.get_rect(center=button.center))
pygame.display.flip() pygame.display.flip()
clock.tick(60) clock.tick(60)
pygame.quit() pygame.quit()
if __name__ == "__main__": if __name__ == "__main__":
main() main()
tests/test_element_selectors.py
+13
View File
@@ -163,6 +163,19 @@ class DuplicatedContainer(unittest.TestCase):
self.assertEqual(len(cards), 7) 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): class DailySetOpener(unittest.TestCase):
"""The opener label has to be distinguished from the level up entry.""" """The opener label has to be distinguished from the level up entry."""
tests/test_task_outcomes.py
+262
View File
@@ -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()