Spaces:
Sleeping
Sleeping
File size: 5,633 Bytes
014ab37 2132d78 014ab37 5b002a8 014ab37 5b002a8 014ab37 5b002a8 039dda4 5b002a8 9c09838 5b002a8 9c09838 5b002a8 014ab37 2132d78 014ab37 2132d78 014ab37 2132d78 014ab37 2132d78 014ab37 2132d78 014ab37 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 | """
Image handling and fetching from Unsplash.
"""
import requests
import json
from io import BytesIO
from PIL import Image
from logger import logger
from config import UNSPLASH_API_KEY, IMAGE_FEATURED_WIDTH, IMAGE_FEATURED_HEIGHT
class ImageHandler:
"""Handles image fetching from Unsplash API."""
def __init__(self, api_key=UNSPLASH_API_KEY):
"""Initialize image handler."""
if not api_key:
raise ValueError("UNSPLASH_API_KEY not set in environment")
self.api_key = api_key
self.base_url = "https://api.unsplash.com/photos/random"
logger.debug("ImageHandler initialized")
def fetch_image(self, tags):
"""
Fetch a random image from Unsplash based on tags. If no image is found, retry by removing one word at a time.
Args:
tags (list): List of search tags
Returns:
dict: Contains 'url', 'credit', 'filename' keys
Raises:
Exception: If API call fails or response is invalid
"""
if not tags:
raise ValueError("At least one tag is required")
tried_queries = set()
def try_fetch(tags_subset):
search_query = " ".join(tags_subset)
if not search_query or search_query in tried_queries:
return None
tried_queries.add(search_query)
logger.info(f"Searching for image with query: '{search_query}'")
params = {
"query": search_query,
"orientation": "landscape",
"w": IMAGE_FEATURED_WIDTH,
"h": IMAGE_FEATURED_HEIGHT,
}
headers = {"Authorization": f"Client-ID {self.api_key}"}
logger.debug(f"Unsplash API request: {self.base_url}")
try:
response = requests.get(self.base_url, params=params, headers=headers)
logger.debug(f"Unsplash response status: {response.status_code}")
if response.status_code != 200:
logger.warning(f"No image found for query '{search_query}' (status {response.status_code})")
return None
photo = response.json()
if "urls" not in photo or "user" not in photo:
logger.warning(f"Invalid response structure for query '{search_query}': {json.dumps(photo, indent=2)}")
return None
image_url = photo["urls"].get("regular")
image_credit = photo["user"].get("name", "")
image_credit_url = photo["user"].get("links", {}).get("html", "https://unsplash.com")
filename = f"featured_{search_query.replace(' ', '_')}.avif"
logger.info(f"Image found by {image_credit}: {image_url}")
return {
"url": image_url,
"credit": image_credit,
"credit_url": image_credit_url,
"filename": filename,
}
except requests.RequestException as e:
logger.error(f"Request failed: {e}")
return None
except Exception as e:
logger.error(f"Failed to fetch image: {e}")
return None
# Try with all tags first
result = try_fetch(tags)
if result:
return result
# Try by removing one word at a time
for i in range(len(tags)):
reduced_tags = tags[:i] + tags[i+1:]
if not reduced_tags:
continue
result = try_fetch(reduced_tags)
if result:
return result
# Try each individual tag
for tag in tags:
result = try_fetch([tag])
if result:
return result
raise Exception("No image found for any tag combination.")
def download_image(self, image_url):
"""
Download image from URL and convert to AVIF format.
Args:
image_url (str): URL of the imageF
Returns:
bytes: Image data in AVIF format
Raises:
Exception: If download or conversion fails
"""
try:
logger.info("Downloading image...")
response = requests.get(image_url)
response.raise_for_status()
image_data = response.content
logger.info(f"Downloaded image: {len(image_data) // 1024} KB")
# Convert to AVIF format
logger.info("Converting image to AVIF format...")
try:
img = Image.open(BytesIO(image_data))
# Convert RGBA to RGB if needed
if img.mode in ("RGBA", "LA", "P"):
rgb_img = Image.new("RGB", img.size, (255, 255, 255))
rgb_img.paste(img, mask=img.split()[-1] if img.mode == "RGBA" else None)
img = rgb_img
# Save as AVIF
avif_buffer = BytesIO()
img.save(avif_buffer, format="AVIF", quality=85)
avif_data = avif_buffer.getvalue()
logger.info(f"Converted to AVIF: {len(avif_data) // 1024} KB")
return avif_data
except Exception as e:
logger.error(f"Failed to convert image to AVIF: {e}")
logger.info("Returning original image data")
return image_data
except requests.RequestException as e:
logger.error(f"Failed to download image: {e}")
raise
|