nifty-coder/stemsplit-backend
0
1"""2Rate Limiter implementation for the voice control optimization system.3 4This module implements sophisticated rate limiting and quota management5with sliding window tracking, provider-specific limits, and automatic6provider switching capabilities.7"""8 9import asyncio10import logging11import time12from typing import Dict, List, Optional, Set, Tuple, Any13from datetime import datetime, timedelta, timezone14from collections import defaultdict, deque15from dataclasses import dataclass, field16import json17 18from .interfaces import RateLimiterInterface19from .models import QuotaStatus, QuotaType, UsageStats, ProviderConfig20from .exceptions import QuotaExceededError, RateLimitExceededError, ConfigurationError21 22 23logger = logging.getLogger(__name__)24 25 26@dataclass27class TimeWindow:28 """Represents a time window for rate limiting."""29 duration: int # Duration in seconds30 max_requests: int31 max_audio_duration: float32 reset_schedule: Optional[str] = None # "hourly", "daily", "monthly", or None for sliding33 34 35@dataclass36class WindowUsage:37 """Tracks usage within a time window."""38 requests: deque = field(default_factory=deque) # Timestamps of requests39 audio_durations: deque = field(default_factory=deque) # (timestamp, duration) tuples40 total_requests: int = 041 total_audio_duration: float = 0.042 last_reset: datetime = field(default_factory=lambda: datetime.now(timezone.utc))43 44 45class RateLimiter(RateLimiterInterface):46 """47 Implements multi-window rate limiting with sliding window tracking.48 49 Features:50 - Sliding window rate limiting for requests per minute/hour/day51 - Audio duration quota tracking across multiple time windows52 - Provider-specific limit enforcement53 - Automatic quota resets based on provider schedules54 - Exponential backoff for rate limit violations55 """56 57 def __init__(self, redis_url: Optional[str] = None, cleanup_interval: int = 300):58 """59 Initialize the rate limiter.60 61 Args:62 redis_url: Redis URL for distributed rate limiting (None for in-memory)63 cleanup_interval: Interval in seconds for cleaning up old data64 """65 self._redis_url = redis_url66 self._cleanup_interval = cleanup_interval67 68 # In-memory storage (used when Redis is not available)69 self._provider_windows: Dict[str, Dict[str, TimeWindow]] = {}70 self._provider_usage: Dict[str, Dict[str, WindowUsage]] = defaultdict(lambda: defaultdict(WindowUsage))71 72 # Provider configurations73 self._provider_configs: Dict[str, ProviderConfig] = {}74 75 # Backoff tracking76 self._backoff_state: Dict[str, Dict[str, Any]] = defaultdict(dict)77 78 # Background tasks79 self._cleanup_task: Optional[asyncio.Task] = None80 self._reset_scheduler_task: Optional[asyncio.Task] = None81 82 # Lock for thread safety83 self._lock = asyncio.Lock()84 85 # Redis client (will be initialized if redis_url is provided)86 self._redis_client = None87 88 # Reset scheduler configuration89 self._reset_check_interval = 60 # Check for resets every minute90 91 logger.info(f"RateLimiter initialized (redis_url: {redis_url is not None})")92 93 async def initialize(self) -> None:94 """Initialize the rate limiter and start background tasks."""95 if self._redis_url:96 await self._initialize_redis()97 98 # Start background tasks99 self._cleanup_task = asyncio.create_task(self._cleanup_loop())100 self._reset_scheduler_task = asyncio.create_task(self._reset_scheduler_loop())101 102 logger.info("RateLimiter initialization complete")103 104 async def _initialize_redis(self) -> None:105 """Initialize Redis client for distributed rate limiting."""106 try:107 import redis.asyncio as redis108 self._redis_client = redis.from_url(self._redis_url)109 await self._redis_client.ping()110 logger.info("Redis client initialized successfully")111 except ImportError:112 logger.warning("Redis not available, falling back to in-memory rate limiting")113 self._redis_client = None114 except Exception as e:115 logger.error(f"Failed to initialize Redis client: {e}")116 self._redis_client = None117 118 async def configure_provider(self, provider_config: ProviderConfig) -> None:119 """120 Configure rate limits for a provider.121 122 Args:123 provider_config: Provider configuration with rate limits124 """125 async with self._lock:126 provider_name = provider_config.name127 self._provider_configs[provider_name] = provider_config128 129 # Initialize time windows based on provider configuration130 windows = {}131 132 # Standard rate limit windows133 rate_limits = provider_config.rate_limits134 135 if "requests_per_minute" in rate_limits:136 windows["requests_per_minute"] = TimeWindow(137 duration=60,138 max_requests=rate_limits["requests_per_minute"],139 max_audio_duration=float('inf')140 )141 142 if "requests_per_hour" in rate_limits:143 windows["requests_per_hour"] = TimeWindow(144 duration=3600,145 max_requests=rate_limits["requests_per_hour"],146 max_audio_duration=float('inf')147 )148 149 if "requests_per_day" in rate_limits:150 windows["requests_per_day"] = TimeWindow(151 duration=86400,152 max_requests=rate_limits["requests_per_day"],153 max_audio_duration=float('inf'),154 reset_schedule="daily"155 )156 157 # Audio duration limits from free tier limits158 free_tier = provider_config.free_tier_limits159 160 if "audio_minutes_per_day" in free_tier:161 windows["audio_minutes_per_day"] = TimeWindow(162 duration=86400,163 max_requests=float('inf'),164 max_audio_duration=free_tier["audio_minutes_per_day"] * 60.0, # Convert to seconds165 reset_schedule="daily"166 )167 168 if "audio_minutes_per_month" in free_tier:169 windows["audio_minutes_per_month"] = TimeWindow(170 duration=30 * 86400, # 30 days171 max_requests=float('inf'),172 max_audio_duration=free_tier["audio_minutes_per_month"] * 60.0, # Convert to seconds173 reset_schedule="monthly"174 )175 176 self._provider_windows[provider_name] = windows177 178 # Initialize usage tracking for each window179 for window_name in windows:180 if window_name not in self._provider_usage[provider_name]:181 self._provider_usage[provider_name][window_name] = WindowUsage()182 183 logger.info(f"Configured rate limits for provider {provider_name}: {list(windows.keys())}")184 185 async def check_quota(186 self, 187 provider: str, 188 audio_duration: float189 ) -> QuotaStatus:190 """191 Check if a request would exceed quota limits.192 193 Args:194 provider: Provider name195 audio_duration: Duration of audio in seconds196 197 Returns:198 QuotaStatus indicating current usage and limits199 """200 if provider not in self._provider_windows:201 # No limits configured for this provider202 return QuotaStatus(203 provider=provider,204 quota_type=QuotaType.REQUESTS_PER_MINUTE,205 current_usage=0.0,206 limit=float('inf'),207 remaining=float('inf'),208 reset_time=datetime.now(timezone.utc) + timedelta(hours=1),209 percentage_used=0.0210 )211 212 async with self._lock:213 # Check all configured windows for this provider214 most_restrictive_status = None215 highest_usage_percentage = 0.0216 217 for window_name, window in self._provider_windows[provider].items():218 usage = self._provider_usage[provider][window_name]219 220 # Clean up old entries for sliding windows221 if window.reset_schedule is None:222 await self._cleanup_sliding_window(provider, window_name, window.duration)223 224 # Calculate current usage225 current_requests = len(usage.requests)226 current_audio_duration = sum(duration for _, duration in usage.audio_durations)227 228 # Check request limits229 if window.max_requests != float('inf'):230 requests_percentage = current_requests / window.max_requests231 if requests_percentage > highest_usage_percentage:232 highest_usage_percentage = requests_percentage233 quota_type = self._window_name_to_quota_type(window_name)234 most_restrictive_status = QuotaStatus(235 provider=provider,236 quota_type=quota_type,237 current_usage=current_requests,238 limit=window.max_requests,239 remaining=max(0, window.max_requests - current_requests),240 reset_time=self._calculate_reset_time(window),241 percentage_used=requests_percentage242 )243 244 # Check audio duration limits245 if window.max_audio_duration != float('inf'):246 # Check if adding this request would exceed the limit247 projected_duration = current_audio_duration + audio_duration248 duration_percentage = projected_duration / window.max_audio_duration249 250 if duration_percentage > highest_usage_percentage:251 highest_usage_percentage = duration_percentage252 quota_type = self._window_name_to_quota_type(window_name)253 254 # Create QuotaStatus with projected usage to properly reflect if limit would be exceeded255 most_restrictive_status = QuotaStatus(256 provider=provider,257 quota_type=quota_type,258 current_usage=projected_duration / 60.0, # Use projected duration in minutes259 limit=window.max_audio_duration / 60.0, # Convert to minutes260 remaining=max(0, (window.max_audio_duration - projected_duration) / 60.0),261 reset_time=self._calculate_reset_time(window),262 percentage_used=0.0 # Will be recalculated in __post_init__263 )264 265 # Return the most restrictive quota status266 if most_restrictive_status is None:267 # No limits configured268 return QuotaStatus(269 provider=provider,270 quota_type=QuotaType.REQUESTS_PER_MINUTE,271 current_usage=0.0,272 limit=float('inf'),273 remaining=float('inf'),274 reset_time=datetime.now(timezone.utc) + timedelta(hours=1),275 percentage_used=0.0276 )277 278 return most_restrictive_status279 280 async def consume_quota(281 self, 282 provider: str, 283 audio_duration: float,284 request_count: int = 1285 ) -> bool:286 """287 Consume quota for a request.288 289 Args:290 provider: Provider name291 audio_duration: Duration of audio in seconds292 request_count: Number of requests (default 1)293 294 Returns:295 True if quota was consumed successfully, False if exceeded296 297 Raises:298 QuotaExceededError: If quota would be exceeded299 """300 # First check if the request would exceed limits301 quota_status = await self.check_quota(provider, audio_duration)302 303 if quota_status.is_exceeded:304 # Check if we should apply backoff (only on repeated attempts)305 backoff_key = f"{provider}:{quota_status.quota_type.value}"306 if backoff_key in self._backoff_state:307 await self._apply_backoff(provider, quota_status.quota_type.value)308 else:309 # First time hitting the limit, just record it for future backoff310 self._backoff_state[backoff_key] = {311 "attempts": 1,312 "last_attempt": time.time(),313 "backoff_duration": 1.0314 }315 316 raise QuotaExceededError(provider, quota_status.quota_type.value, quota_status.current_usage, quota_status.limit)317 318 # Check if we're near the limit and should warn319 if quota_status.is_near_limit(threshold=0.9):320 logger.warning(f"Provider {provider} approaching quota limit: {quota_status.percentage_used:.1%} used")321 322 # Consume the quota323 async with self._lock:324 if provider not in self._provider_windows:325 return True # No limits configured326 327 current_time = time.time()328 329 for window_name in self._provider_windows[provider]:330 usage = self._provider_usage[provider][window_name]331 332 # Add request timestamps333 for _ in range(request_count):334 usage.requests.append(current_time)335 336 # Add audio duration337 if audio_duration > 0:338 usage.audio_durations.append((current_time, audio_duration))339 340 # Update totals341 usage.total_requests += request_count342 usage.total_audio_duration += audio_duration343 344 # Reset backoff on successful consumption345 if provider in self._backoff_state:346 # Clear backoff for this provider347 keys_to_remove = [key for key in self._backoff_state.keys() if key.startswith(f"{provider}:")]348 for key in keys_to_remove:349 del self._backoff_state[key]350 351 logger.debug(f"Consumed quota for {provider}: {request_count} requests, {audio_duration:.2f}s audio")352 return True353 354 async def get_usage_stats(355 self, 356 provider: str, 357 time_window: str358 ) -> UsageStats:359 """360 Get usage statistics for a provider and time window.361 362 Args:363 provider: Provider name364 time_window: Time window ("minute", "hour", "day", "month")365 366 Returns:367 UsageStats for the specified period368 """369 # Map time window to actual window name used in tracking370 window_mapping = {371 "minute": "requests_per_minute",372 "hour": "requests_per_hour", 373 "day": "requests_per_day",374 "month": "audio_minutes_per_month"375 }376 377 window_name = window_mapping.get(time_window, "requests_per_minute")378 379 async with self._lock:380 if provider not in self._provider_usage or window_name not in self._provider_usage[provider]:381 # No usage data available382 now = datetime.now(timezone.utc)383 return UsageStats(384 provider=provider,385 requests_count=0,386 audio_minutes=0.0,387 estimated_cost=0.0,388 success_rate=1.0,389 average_latency=0.0,390 time_window=time_window,391 window_start=now,392 window_end=now393 )394 395 usage = self._provider_usage[provider][window_name]396 397 # Calculate window boundaries using current time398 current_time = time.time()399 400 if time_window == "minute":401 window_start_timestamp = current_time - 60402 elif time_window == "hour":403 window_start_timestamp = current_time - 3600404 elif time_window == "day":405 window_start_timestamp = current_time - 86400406 elif time_window == "month":407 window_start_timestamp = current_time - (30 * 86400)408 else:409 window_start_timestamp = current_time - 3600410 411 # Count requests and audio duration in the window412 requests_in_window = sum(1 for timestamp in usage.requests if timestamp >= window_start_timestamp)413 audio_in_window = sum(duration for timestamp, duration in usage.audio_durations if timestamp >= window_start_timestamp)414 415 # Convert back to datetime for return value416 now = datetime.now(timezone.utc)417 window_start = datetime.fromtimestamp(window_start_timestamp, tz=timezone.utc)418 419 # Estimate cost420 provider_config = self._provider_configs.get(provider)421 estimated_cost = 0.0422 if provider_config:423 estimated_cost = (audio_in_window / 60.0) * provider_config.cost_per_minute424 425 return UsageStats(426 provider=provider,427 requests_count=requests_in_window,428 audio_minutes=audio_in_window / 60.0,429 estimated_cost=estimated_cost,430 success_rate=1.0, # We don't track failures here431 average_latency=0.0, # We don't track latency here432 time_window=time_window,433 window_start=window_start,434 window_end=now435 )436 437 async def reset_quota(self, provider: str, quota_type: str) -> None:438 """439 Reset quota counters for a provider.440 441 Args:442 provider: Provider name443 quota_type: Type of quota to reset444 """445 async with self._lock:446 if provider not in self._provider_usage:447 return448 449 # Find matching windows450 windows_to_reset = []451 for window_name in self._provider_usage[provider]:452 if quota_type in window_name:453 windows_to_reset.append(window_name)454 455 # Reset usage for matching windows456 for window_name in windows_to_reset:457 usage = self._provider_usage[provider][window_name]458 usage.requests.clear()459 usage.audio_durations.clear()460 usage.total_requests = 0461 usage.total_audio_duration = 0.0462 usage.last_reset = datetime.now(timezone.utc)463 464 logger.info(f"Reset quota for provider {provider}, type {quota_type}")465 466 async def _cleanup_sliding_window(self, provider: str, window_name: str, window_duration: int) -> None:467 """468 Clean up old entries from a sliding window.469 470 Args:471 provider: Provider name472 window_name: Name of the window473 window_duration: Duration of the window in seconds474 """475 usage = self._provider_usage[provider][window_name]476 current_time = time.time()477 cutoff_time = current_time - window_duration478 479 # Remove old request timestamps480 while usage.requests and usage.requests[0] < cutoff_time:481 usage.requests.popleft()482 483 # Remove old audio duration entries484 while usage.audio_durations and usage.audio_durations[0][0] < cutoff_time:485 usage.audio_durations.popleft()486 487 def _window_name_to_quota_type(self, window_name: str) -> QuotaType:488 """Convert window name to QuotaType enum."""489 if "requests_per_minute" in window_name:490 return QuotaType.REQUESTS_PER_MINUTE491 elif "requests_per_hour" in window_name:492 return QuotaType.REQUESTS_PER_HOUR493 elif "requests_per_day" in window_name:494 return QuotaType.REQUESTS_PER_DAY495 elif "audio_minutes_per_day" in window_name:496 return QuotaType.AUDIO_MINUTES_PER_DAY497 elif "audio_minutes_per_month" in window_name:498 return QuotaType.AUDIO_MINUTES_PER_MONTH499 else:500 return QuotaType.REQUESTS_PER_MINUTE501 502 def _calculate_reset_time(self, window: TimeWindow) -> datetime:503 """Calculate when the window will reset."""504 now = datetime.now(timezone.utc)505 506 if window.reset_schedule == "hourly":507 # Reset at the top of the next hour508 next_reset = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1)509 elif window.reset_schedule == "daily":510 # Reset at midnight511 next_reset = now.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=1)512 elif window.reset_schedule == "weekly":513 # Reset at the beginning of next week (Monday)514 days_until_monday = (7 - now.weekday()) % 7515 if days_until_monday == 0: # If today is Monday, reset next Monday516 days_until_monday = 7517 next_reset = now.replace(hour=0, minute=0, second=0, microsecond=0) + timedelta(days=days_until_monday)518 elif window.reset_schedule == "monthly":519 # Reset at the beginning of next month520 if now.month == 12:521 next_reset = now.replace(year=now.year + 1, month=1, day=1, hour=0, minute=0, second=0, microsecond=0)522 else:523 next_reset = now.replace(month=now.month + 1, day=1, hour=0, minute=0, second=0, microsecond=0)524 else:525 # Sliding window - reset continuously526 next_reset = now + timedelta(seconds=window.duration)527 528 return next_reset529 530 async def _apply_backoff(self, provider: str, quota_type: str) -> None:531 """532 Apply exponential backoff for rate limit violations.533 534 Args:535 provider: Provider name536 quota_type: Type of quota that was exceeded537 """538 backoff_key = f"{provider}:{quota_type}"539 540 if backoff_key not in self._backoff_state:541 self._backoff_state[backoff_key] = {542 "attempts": 0,543 "last_attempt": time.time(),544 "backoff_duration": 1.0 # Start with 1 second545 }546 547 backoff_info = self._backoff_state[backoff_key]548 current_time = time.time()549 550 # Check if enough time has passed since last attempt551 time_since_last = current_time - backoff_info["last_attempt"]552 if time_since_last < backoff_info["backoff_duration"]:553 remaining_wait = backoff_info["backoff_duration"] - time_since_last554 raise RateLimitExceededError(provider, quota_type, remaining_wait)555 556 # Update backoff state557 backoff_info["attempts"] += 1558 backoff_info["last_attempt"] = current_time559 backoff_info["backoff_duration"] = min(300, backoff_info["backoff_duration"] * 2) # Max 5 minutes560 561 logger.warning(f"Applied exponential backoff for {provider}:{quota_type}, next attempt in {backoff_info['backoff_duration']}s")562 563 async def _cleanup_loop(self) -> None:564 """Background task for periodic cleanup of old data."""565 while True:566 try:567 await asyncio.sleep(self._cleanup_interval)568 await self._perform_cleanup()569 except asyncio.CancelledError:570 logger.info("Rate limiter cleanup loop cancelled")571 break572 except Exception as e:573 logger.error(f"Error in rate limiter cleanup loop: {e}")574 575 async def _reset_scheduler_loop(self) -> None:576 """Background task for scheduled quota resets."""577 while True:578 try:579 await asyncio.sleep(self._reset_check_interval)580 await self._check_and_perform_resets()581 except asyncio.CancelledError:582 logger.info("Rate limiter reset scheduler loop cancelled")583 break584 except Exception as e:585 logger.error(f"Error in rate limiter reset scheduler loop: {e}")586 587 async def _perform_cleanup(self) -> None:588 """Perform cleanup of old rate limiting data."""589 async with self._lock:590 current_time = time.time()591 592 for provider in list(self._provider_usage.keys()):593 for window_name in list(self._provider_usage[provider].keys()):594 if provider in self._provider_windows and window_name in self._provider_windows[provider]:595 window = self._provider_windows[provider][window_name]596 597 # Only cleanup sliding windows598 if window.reset_schedule is None:599 await self._cleanup_sliding_window(provider, window_name, window.duration)600 601 # Cleanup old backoff state602 for backoff_key in list(self._backoff_state.keys()):603 backoff_info = self._backoff_state[backoff_key]604 if current_time - backoff_info["last_attempt"] > 3600: # 1 hour605 del self._backoff_state[backoff_key]606 607 logger.debug("Rate limiter cleanup completed")608 609 async def _check_and_perform_resets(self) -> None:610 """Check for scheduled resets and perform them if needed."""611 async with self._lock:612 current_time = datetime.now(timezone.utc)613 614 for provider in list(self._provider_usage.keys()):615 if provider not in self._provider_windows:616 continue617 618 for window_name, window in self._provider_windows[provider].items():619 # Only process windows with scheduled resets620 if window.reset_schedule is None:621 continue622 623 usage = self._provider_usage[provider][window_name]624 625 # Check if reset is due626 if self._is_reset_due(window, usage.last_reset, current_time):627 # Perform the reset628 usage.requests.clear()629 usage.audio_durations.clear()630 usage.total_requests = 0631 usage.total_audio_duration = 0.0632 usage.last_reset = current_time633 634 logger.info(f"Scheduled reset performed for {provider}:{window_name} ({window.reset_schedule})")635 636 logger.debug("Reset scheduler check completed")637 638 def _is_reset_due(self, window: TimeWindow, last_reset: datetime, current_time: datetime) -> bool:639 """640 Check if a scheduled reset is due for a window.641 642 Args:643 window: The time window configuration644 last_reset: When the window was last reset645 current_time: Current time646 647 Returns:648 True if reset is due, False otherwise649 """650 if window.reset_schedule == "hourly":651 # Reset at the top of each hour652 last_reset_hour = last_reset.replace(minute=0, second=0, microsecond=0)653 current_hour = current_time.replace(minute=0, second=0, microsecond=0)654 return current_hour > last_reset_hour655 656 elif window.reset_schedule == "daily":657 # Reset at midnight each day658 last_reset_day = last_reset.replace(hour=0, minute=0, second=0, microsecond=0)659 current_day = current_time.replace(hour=0, minute=0, second=0, microsecond=0)660 return current_day > last_reset_day661 662 elif window.reset_schedule == "monthly":663 # Reset at the beginning of each month664 last_reset_month = last_reset.replace(day=1, hour=0, minute=0, second=0, microsecond=0)665 current_month = current_time.replace(day=1, hour=0, minute=0, second=0, microsecond=0)666 return current_month > last_reset_month667 668 elif window.reset_schedule == "weekly":669 # Reset at the beginning of each week (Monday)670 days_since_monday = current_time.weekday()671 current_week_start = current_time.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=days_since_monday)672 673 last_reset_days_since_monday = last_reset.weekday()674 last_reset_week_start = last_reset.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=last_reset_days_since_monday)675 676 return current_week_start > last_reset_week_start677 678 # Unknown schedule type, don't reset679 return False680 681 async def get_provider_quota_status(self, provider: str) -> Dict[str, QuotaStatus]:682 """683 Get quota status for all windows of a provider.684 685 Args:686 provider: Provider name687 688 Returns:689 Dictionary mapping window names to quota status690 """691 if provider not in self._provider_windows:692 return {}693 694 quota_statuses = {}695 696 for window_name, window in self._provider_windows[provider].items():697 # Use a small audio duration for checking current status698 quota_status = await self.check_quota(provider, 0.1)699 quota_statuses[window_name] = quota_status700 701 return quota_statuses702 703 async def is_provider_available(self, provider: str, audio_duration: float = 0.0) -> bool:704 """705 Check if a provider is available (not rate limited).706 707 Args:708 provider: Provider name709 audio_duration: Duration of audio to check (optional)710 711 Returns:712 True if provider is available, False if rate limited713 """714 try:715 quota_status = await self.check_quota(provider, audio_duration)716 return not quota_status.is_exceeded717 except (QuotaExceededError, RateLimitExceededError):718 return False719 720 async def is_provider_approaching_limit(self, provider: str, audio_duration: float = 0.0, threshold: float = 0.8) -> bool:721 """722 Check if a provider is approaching its quota limits.723 724 Args:725 provider: Provider name726 audio_duration: Duration of audio to check (optional)727 threshold: Percentage threshold to consider "approaching" (default 0.8 = 80%)728 729 Returns:730 True if provider is approaching limits, False otherwise731 """732 try:733 quota_status = await self.check_quota(provider, audio_duration)734 return quota_status.is_near_limit(threshold)735 except (QuotaExceededError, RateLimitExceededError):736 return True # Already at limit737 738 async def get_providers_by_availability(self, audio_duration: float = 0.0) -> Dict[str, bool]:739 """740 Get availability status for all configured providers.741 742 Args:743 audio_duration: Duration of audio to check (optional)744 745 Returns:746 Dictionary mapping provider names to availability status747 """748 availability = {}749 750 for provider in self._provider_configs.keys():751 availability[provider] = await self.is_provider_available(provider, audio_duration)752 753 return availability754 755 async def configure_reset_schedule(self, provider: str, window_name: str, reset_schedule: str) -> None:756 """757 Configure a custom reset schedule for a specific provider window.758 759 Args:760 provider: Provider name761 window_name: Name of the window to configure762 reset_schedule: Reset schedule ("hourly", "daily", "weekly", "monthly", or None for sliding)763 """764 async with self._lock:765 if provider not in self._provider_windows:766 raise ValueError(f"Provider {provider} not configured")767 768 if window_name not in self._provider_windows[provider]:769 raise ValueError(f"Window {window_name} not found for provider {provider}")770 771 # Validate reset schedule772 valid_schedules = ["hourly", "daily", "weekly", "monthly", None]773 if reset_schedule not in valid_schedules:774 raise ValueError(f"Invalid reset schedule: {reset_schedule}. Must be one of {valid_schedules}")775 776 # Update the window configuration777 window = self._provider_windows[provider][window_name]778 old_schedule = window.reset_schedule779 window.reset_schedule = reset_schedule780 781 # Reset the usage tracking to start fresh with new schedule782 if window_name in self._provider_usage[provider]:783 usage = self._provider_usage[provider][window_name]784 usage.requests.clear()785 usage.audio_durations.clear()786 usage.total_requests = 0787 usage.total_audio_duration = 0.0788 usage.last_reset = datetime.now(timezone.utc)789 790 logger.info(f"Updated reset schedule for {provider}:{window_name} from {old_schedule} to {reset_schedule}")791 792 async def get_reset_schedule_info(self, provider: str) -> Dict[str, Dict[str, Any]]:793 """794 Get reset schedule information for all windows of a provider.795 796 Args:797 provider: Provider name798 799 Returns:800 Dictionary with window names as keys and reset info as values801 """802 if provider not in self._provider_windows:803 return {}804 805 reset_info = {}806 current_time = datetime.now(timezone.utc)807 808 for window_name, window in self._provider_windows[provider].items():809 usage = self._provider_usage[provider].get(window_name)810 811 info = {812 "reset_schedule": window.reset_schedule,813 "last_reset": usage.last_reset if usage else None,814 "next_reset": self._calculate_reset_time(window) if window.reset_schedule else None,815 "is_sliding_window": window.reset_schedule is None816 }817 818 # Add time until next reset819 if info["next_reset"]:820 time_until_reset = info["next_reset"] - current_time821 info["seconds_until_reset"] = max(0, int(time_until_reset.total_seconds()))822 else:823 info["seconds_until_reset"] = None824 825 reset_info[window_name] = info826 827 return reset_info828 829 async def should_switch_provider(self, provider: str, audio_duration: float = 0.0, threshold: float = 0.9) -> bool:830 """831 Determine if we should switch away from a provider due to quota concerns.832 833 Args:834 provider: Current provider name835 audio_duration: Duration of audio to process836 threshold: Percentage threshold for switching (default 0.9 = 90%)837 838 Returns:839 True if should switch to alternative provider, False otherwise840 """841 if provider not in self._provider_configs:842 return False843 844 # Check if provider is approaching limits845 if await self.is_provider_approaching_limit(provider, audio_duration, threshold):846 logger.info(f"Provider {provider} approaching quota limit, recommending switch")847 return True848 849 # Check if provider is in backoff state850 for backoff_key in self._backoff_state.keys():851 if backoff_key.startswith(f"{provider}:"):852 backoff_info = self._backoff_state[backoff_key]853 current_time = time.time()854 time_since_last = current_time - backoff_info["last_attempt"]855 856 if time_since_last < backoff_info["backoff_duration"]:857 logger.info(f"Provider {provider} in backoff state, recommending switch")858 return True859 860 return False861 862 async def shutdown(self) -> None:863 """Shutdown the rate limiter and cleanup resources."""864 logger.info("Shutting down RateLimiter")865 866 # Cancel background tasks867 if self._cleanup_task:868 self._cleanup_task.cancel()869 try:870 await self._cleanup_task871 except asyncio.CancelledError:872 pass873 874 if self._reset_scheduler_task:875 self._reset_scheduler_task.cancel()876 try:877 await self._reset_scheduler_task878 except asyncio.CancelledError:879 pass880 881 # Close Redis connection if available882 if self._redis_client:883 await self._redis_client.close()884 885 # Clear all data886 async with self._lock:887 self._provider_windows.clear()888 self._provider_usage.clear()889 self._provider_configs.clear()890 self._backoff_state.clear()891 892 logger.info("RateLimiter shutdown complete")