"""Rate limiting to prevent IP bans from YouTube scraping."""
import asyncio
from datetime import datetime, timedelta
from typing import Callable, TypeVar
from functools import wraps
T = TypeVar("T")
class RateLimiter:
"""
Global rate limiter to ensure safe delays between YouTube requests.
This prevents IP bans by enforcing a minimum delay between consecutive
API calls to YouTube's infrastructure.
Default: 0.75 seconds between requests (conservative for scraping).
"""
def __init__(self, delay_seconds: float = 0.75):
self.delay_seconds = delay_seconds
self.last_request_time: datetime | None = None
self._lock = asyncio.Lock()
async def acquire(self) -> None:
"""Acquire the rate limiter, blocking if necessary to enforce the delay."""
async with self._lock:
if self.last_request_time is not None:
elapsed = (datetime.now() - self.last_request_time).total_seconds()
wait_time = self.delay_seconds - elapsed
if wait_time > 0:
await asyncio.sleep(wait_time)
self.last_request_time = datetime.now()
def __call__(self, func: Callable[..., T]) -> Callable[..., T]:
"""Decorator to apply rate limiting to async functions."""
@wraps(func)
async def wrapper(*args, **kwargs) -> T:
await self.acquire()
return await func(*args, **kwargs)
return wrapper
# Global instance
rate_limiter = RateLimiter(delay_seconds=0.75)
