rewards-farmer/src/random_image_for_visual_search.py

import io
import json
import logging
import time
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit

import requests
from PIL import Image

logger = logging.getLogger(__name__)


# ============================================================
# CONFIG
# ============================================================

API_URL = "https://commons.wikimedia.org/w/api.php"

OUTPUT_FILE = Path("visual_search.jpg")
METADATA_FILE = Path("visual_search.json")

MAX_ATTEMPTS = 10

# Loose filtering
MIN_WIDTH = 300
MIN_HEIGHT = 300

# Maximum original file size we'll accept
MAX_FILE_SIZE = 20 * 1024 * 1024  # 20 MB

# Wikimedia will generate a thumbnail around this width
THUMBNAIL_WIDTH = 1280

# Normal delay between failed attempts
REQUEST_DELAY = 3

# Maximum rate-limit wait
MAX_BACKOFF = 300


# ============================================================
# HTTP SESSION
# ============================================================

session = requests.Session()

session.headers.update({
	"User-Agent": (
		"RandomVisualSearchImage/1.1 "
		"(contact: 12345rfdz@gmail.com)"
	)
})


# ============================================================
# URL HELPERS
# ============================================================

def clean_url(url):
	"""Remove query parameters from Wikimedia URLs."""

	parts = urlsplit(url)

	return urlunsplit((
		parts.scheme,
		parts.netloc,
		parts.path,
		"",
		"",
	))


# ============================================================
# RATE LIMIT HANDLING
# ============================================================

def wait_after_429(response, attempt):
	"""Wait according to Wikimedia's Retry-After header."""

	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,
		)

	wait_time = max(5, wait_time)

	logger.warning("Rate limited. Waiting %d seconds...", wait_time)

	time.sleep(wait_time)


# ============================================================
# DOWNLOAD
# ============================================================

def download_image(url):
	"""Download image bytes from Wikimedia."""

	url = clean_url(url)

	try:
		response = session.get(
			url,
			timeout=30,
			allow_redirects=True,
		)

	except (requests.RequestException, ConnectionResetError, OSError) as e:
		logger.warning("Download failed: %s", e)
		return None

	if response.status_code == 429:
		wait_after_429(response, 1)
		return None

	if response.status_code == 403:
		logger.warning("Wikimedia returned 403 Forbidden.")
		return None

	try:
		response.raise_for_status()
	except requests.RequestException as e:
		logger.warning("HTTP error: %s", e)
		return None

	content_type = response.headers.get(
		"Content-Type",
		"",
	).lower()

	if not content_type.startswith("image/"):
		logger.warning("Not an image: %s", content_type)
		return None

	if not response.content:
		logger.warning("Downloaded image is empty.")
		return None

	return response.content


# ============================================================
# JPEG CONVERSION
# ============================================================

def convert_to_jpeg(image_data):
	"""Convert downloaded image bytes to JPEG."""

	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")

			output = io.BytesIO()

			jpeg_image.save(
				output,
				format="JPEG",
				quality=90,
				optimize=True,
			)

			return output.getvalue()

	except Exception as e:
		logger.warning("JPEG conversion failed: %s", e)
		return None


def generate_fallback_image():
	"""Generate a synthetic local JPEG image using PIL as a fallback."""
	logger.info("Generating synthetic local fallback image for visual search...")
	from PIL import ImageDraw
	import random

	img = Image.new("RGB", (800, 600), color=(random.randint(50, 200), random.randint(50, 200), random.randint(50, 200)))
	draw = ImageDraw.Draw(img)
	for _ in range(10):
		x0 = random.randint(0, 700)
		y0 = random.randint(0, 500)
		x1 = x0 + random.randint(50, 200)
		y1 = y0 + random.randint(50, 200)
		fill = (random.randint(0, 255), random.randint(0, 255), random.randint(0, 255))
		draw.rectangle([x0, y0, x1, y1], fill=fill)

	output = io.BytesIO()
	img.save(output, format="JPEG", quality=90)
	jpeg_data = output.getvalue()

	OUTPUT_FILE.write_bytes(jpeg_data)
	logger.info("Saved fallback image to %s", OUTPUT_FILE.absolute())
	return {"title": "Fallback Synthetic Image", "width": 800, "height": 600}


# ============================================================
# RANDOM IMAGE
# ============================================================

def get_random_image():

	for attempt in range(
		1,
		MAX_ATTEMPTS + 1,
	):

		if attempt > 1:
			time.sleep(REQUEST_DELAY)

		logger.debug("Attempt %d/%d", attempt, MAX_ATTEMPTS)

		params = {
			"action": "query",
			"format": "json",

			"generator": "random",
			"grnnamespace": 6,
			"grnlimit": 1,

			"prop": "imageinfo",

			"iiprop": (
				"url|size|mime|dimensions"
			),

			"iiurlwidth": THUMBNAIL_WIDTH,
		}

		try:
			response = session.get(
				API_URL,
				params=params,
				timeout=20,
			)

		except (requests.RequestException, ConnectionResetError, OSError) as e:
			logger.warning("API request failed: %s", e)
			continue

		if response.status_code == 429:
			wait_after_429(
				response,
				attempt,
			)
			continue

		try:
			response.raise_for_status()
			data = response.json()

		except (
			requests.RequestException,
			ValueError,
		) as e:
			logger.warning("API error: %s", e)
			continue

		pages = (
			data
			.get("query", {})
			.get("pages", {})
		)

		if not pages:
			logger.debug("No page returned.")
			continue

		page = next(
			iter(pages.values())
		)

		title = page.get(
			"title",
			"Unknown",
		)

		imageinfo = page.get(
			"imageinfo"
		)

		if not imageinfo:
			logger.debug("No image information.")
			continue

		info = imageinfo[0]

		mime = info.get(
			"mime",
			"",
		)

		width = info.get(
			"width",
			0,
		)

		height = info.get(
			"height",
			0,
		)

		size = info.get(
			"size",
			0,
		)

		thumbnail_url = info.get(
			"thumburl"
		)

		original_url = info.get(
			"url"
		)

		if mime not in {
			"image/jpeg",
			"image/png",
			"image/webp",
		}:
			logger.debug("Skipping unsupported type: %s", mime)
			continue

		if width < MIN_WIDTH or height < MIN_HEIGHT:
			logger.debug("Skipping small image: %dx%d", width, height)
			continue

		if size > MAX_FILE_SIZE:
			logger.debug("Skipping large image: %.1f MB", size / 1024 / 1024)
			continue

		if not thumbnail_url:
			logger.debug("No thumbnail URL.")
			continue

		logger.debug("Found: %s (%dx%d)", title, width, height)

		image_data = download_image(
			thumbnail_url
		)

		if image_data is None and original_url:
			logger.debug("Thumbnail download failed, trying original URL...")

			image_data = download_image(
				original_url
			)

		if image_data is None:
			logger.debug("Couldn't download image.")
			continue

		logger.debug("Converting to JPEG...")

		jpeg_data = convert_to_jpeg(
			image_data
		)

		if jpeg_data is None:
			continue

		try:
			OUTPUT_FILE.write_bytes(
				jpeg_data
			)

		except OSError as e:
			logger.warning("Couldn't save image: %s", e)
			continue

		metadata = {
			"title": title,
			"source": "Wikimedia Commons",
			"output_format": "JPEG",

			"width": width,
			"height": height,

			"original_mime": mime,

			"original_size": size,

			"jpeg_size": len(
				jpeg_data
			),

			"original_url": (
				clean_url(original_url)
				if original_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",
			)

		except OSError as e:
			logger.warning("Couldn't save metadata: %s", e)

		logger.info(
			"Visual search image saved: %s (%s, %.1f KB, source: %s)",
			OUTPUT_FILE.absolute(),
			f"{width}x{height}",
			len(jpeg_data) / 1024,
			title,
		)

		return metadata

	logger.warning("Could not download image from Wikimedia Commons. Generating local fallback image.")
	return generate_fallback_image()


# ============================================================
# MAIN
# ============================================================

if __name__ == "__main__":
	get_random_image()