home/links
2
0
Fork 0
mirror of https://github.com/wahyd4/links.git synced 2026-08-09 05:06:16 +10:00
Code Issues Packages Projects Releases Wiki Activity

Image api performance improvement

Browse Source
This commit is contained in:
junv
2025-05-30 15:16:36 +10:00
parent 287fe1a64b
commit 9c89b77006
1 changed files with 93 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
links/image_api.py
+93 -49
View File
@@ -25,6 +25,7 @@ IMAGES_FOLDER = os.getenv('IMAGES_FOLDER', '/Users/junv/Downloads')
CACHE_TIMEOUT = 600 # 10 minutes
MAX_FILE_SIZE = 3 * 1024 * 1024 # 3MB in bytes
SUPPORTED_FORMATS = {'.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.tiff'}
MAX_RANDOM_ATTEMPTS = 30 # Maximum number of attempts to find a valid random image
image_paths_cache_key = 'random_image_paths'
executor = ThreadPoolExecutor(max_workers=4)
@@ -50,26 +51,43 @@ def is_valid_image_fast(file_path: Path) -> bool:
if len(header) < 4:
return False
# Check if file contains valid image data
signature_valid = False
# JPEG signature
if header[:2] == b'\xff\xd8':
return True
signature_valid = True
# PNG signature
if header[:8] == b'\x89PNG\r\n\x1a\n':
return True
elif header[:8] == b'\x89PNG\r\n\x1a\n':
signature_valid = True
# GIF signature
if header[:6] in (b'GIF87a', b'GIF89a'):
return True
elif header[:6] in (b'GIF87a', b'GIF89a'):
signature_valid = True
# WebP signature
if header[:4] == b'RIFF' and header[8:12] == b'WEBP':
return True
elif header[:4] == b'RIFF' and header[8:12] == b'WEBP':
signature_valid = True
# BMP signature
if header[:2] == b'BM':
return True
elif header[:2] == b'BM':
signature_valid = True
# TIFF signatures
if header[:4] in (b'II*\x00', b'MM\x00*'):
return True
elif header[:4] in (b'II*\x00', b'MM\x00*'):
signature_valid = True
return False
if not signature_valid:
return False
# Check image dimensions to ensure both width and height are at least 512px
# This requires opening the image, but it's necessary to check dimensions
try:
with Image.open(file_path) as img:
width, height = img.size
if width < 400 or height < 400:
logger.debug(f"Image too small: {file_path.name} ({width}x{height})")
return False
except Exception:
return False
return True
except (OSError, IOError, Exception):
return False
@@ -140,6 +158,7 @@ def get_image_paths() -> List[Path]:
def invalidate_image_cache():
"""Force invalidation of the image paths cache."""
cache.delete(image_paths_cache_key)
cache.delete('all_image_files')
logger.info("Image paths cache invalidated")
def get_image_stats() -> dict:
@@ -222,29 +241,70 @@ def resize_image(image: Image.Image, width: Optional[int], height: Optional[int]
ratio = height / orig_height
return image.resize((int(orig_width * ratio), height), Image.Resampling.LANCZOS)
def get_random_image() -> Optional[Path]:
"""Get a truly random image path with proper distribution across all folders."""
image_paths = get_image_paths()
def get_all_image_files() -> List[Path]:
"""Get all files with image extensions from the images folder."""
if not IMAGES_FOLDER or not os.path.exists(IMAGES_FOLDER):
logger.warning(f"Images folder not found or not configured: {IMAGES_FOLDER}")
return []
if not image_paths:
logger.warning("No valid images available for selection")
# Cache the file listing to avoid repeating directory traversal
all_image_files = cache.get('all_image_files')
if all_image_files is not None:
return all_image_files
# Get a list of all potential image files by extension only (no validation)
all_files = []
base_path = Path(IMAGES_FOLDER)
# Using os.walk instead of Path.rglob for better performance on large directories
for root, _, files in os.walk(str(base_path)):
for file in files:
# Check extension before adding to the list
ext = os.path.splitext(file)[1].lower()
if ext in SUPPORTED_FORMATS:
all_files.append(Path(os.path.join(root, file)))
logger.info(f"Found {len(all_files)} files with image extensions in {IMAGES_FOLDER}")
cache.set('all_image_files', all_files, CACHE_TIMEOUT)
return all_files
def get_random_image() -> Optional[Path]:
"""Get a random valid image path by sampling and validating on-the-fly."""
# Get all potential image files (only filtered by extension)
all_files = get_all_image_files()
if not all_files:
logger.warning("No image files found in the configured folder")
return None
# Ensure truly random selection by re-seeding with current time and system entropy
random.seed(time.time() + hash(str(os.urandom(8))))
# Try up to MAX_RANDOM_ATTEMPTS times to find a valid image
attempts = 0
while attempts < MAX_RANDOM_ATTEMPTS:
attempts += 1
# Use cryptographically secure random choice for better distribution
selected_image = random.SystemRandom().choice(image_paths)
# Ensure truly random selection by re-seeding with current time and system entropy
random.seed(time.time() + hash(str(os.urandom(8))))
# Log the selected image with folder and filename details
relative_path = selected_image.relative_to(Path(IMAGES_FOLDER))
folder_name = relative_path.parent if relative_path.parent != Path('.') else 'root'
file_size_mb = selected_image.stat().st_size / (1024 * 1024)
# Use cryptographically secure random choice
selected_file = random.SystemRandom().choice(all_files)
logger.info(f"Selected random image: folder='{folder_name}', file='{selected_image.name}', "
f"size={file_size_mb:.2f}MB, total_pool={len(image_paths)}")
# Check if file exists and passes fast validation
if selected_file.exists() and is_valid_image_fast(selected_file):
# Log the selected image with folder and filename details
relative_path = selected_file.relative_to(Path(IMAGES_FOLDER))
folder_name = relative_path.parent if relative_path.parent != Path('.') else 'root'
file_size_mb = selected_file.stat().st_size / (1024 * 1024)
return selected_image
logger.info(f"Selected random image: folder='{folder_name}', file='{selected_file.name}', "
f"size={file_size_mb:.2f}MB, attempt={attempts}/{MAX_RANDOM_ATTEMPTS}")
return selected_file
if attempts % 5 == 0:
logger.debug(f"Still searching for valid image, attempt {attempts}/{MAX_RANDOM_ATTEMPTS}")
logger.warning(f"Failed to find valid image after {MAX_RANDOM_ATTEMPTS} attempts")
return None
def process_image(image_path: Path, width: Optional[int] = None, height: Optional[int] = None,
fit: FitMode = 'scale') -> Tuple[bytes, str]:
@@ -276,7 +336,8 @@ def warm_image_cache_background():
def _warm_cache():
try:
logger.info("Starting background image cache warmup...")
get_image_paths()
# Only cache the file listing, not the full validation
get_all_image_files()
logger.info("Background image cache warmup completed")
except Exception as e:
logger.error(f"Error during background cache warmup: {e}")
@@ -329,28 +390,11 @@ def random_image(request, width: Optional[int] = None, height: Optional[int] = N
return HttpResponseNotFound("Height too large (max 4096px)")
try:
# Get random image with retry mechanism
max_retries = 5
image_path = None
for attempt in range(max_retries):
candidate_path = get_random_image()
if not candidate_path:
logger.warning("No images available in the configured folder")
return HttpResponseNotFound("No images available")
# Quick check if file still exists and is accessible
if candidate_path.exists() and candidate_path.stat().st_size <= MAX_FILE_SIZE:
image_path = candidate_path
break
else:
logger.warning(f"Selected image no longer valid: {candidate_path}, retrying... ({attempt + 1}/{max_retries})")
# Invalidate cache if we're getting bad files
if attempt >= 2:
invalidate_image_cache()
# Get a random valid image directly - the new get_random_image already has retry logic
image_path = get_random_image()
if not image_path:
logger.error("Failed to find valid image after multiple attempts")
logger.error("Could not find a suitable image")
return HttpResponseNotFound("No valid images available")
# Log the request details