mirror of
https://github.com/User0332/rewards-farmer.git
synced 2026-09-16 01:31:36 +00:00
normalize indentation to tabs
This commit is contained in:
Carl Furtado
+84
-84
@@ -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())
|
||||
|
||||
@@ -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()
|
||||
get_random_image()
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user