# Add to InfinityClient class
async def get_bot_votes(self) -> Dict[str, Any]:
"""Get voting statistics for the bot."""
return await self._make_request("GET", f"/bots/{self.bot_id}/votes")
async def get_user_votes(self, user_id: str) -> Dict[str, Any]:
"""Get votes by a specific user."""
return await self._make_request("GET", f"/users/{user_id}/votes")
async def check_user_voted(self, user_id: str) -> bool:
"""Check if a user has voted for this bot today."""
try:
votes = await self._make_request("GET", f"/bots/{self.bot_id}/votes/{user_id}")
return votes.get("voted", False)
except InfinityAPIError:
return Falseasync def search_bots(self, query: str, category: str = None, limit: int = 10) -> Dict[str, Any]:
"""Search for bots by name, description, or tags."""
params = {"q": query, "limit": limit}
if category:
params["category"] = category
return await self._make_request("GET", "/bots/search", data=params)
async def get_featured_bots(self) -> Dict[str, Any]:
"""Get list of featured bots."""
return await self._make_request("GET", "/bots/featured")
async def get_random_bot(self) -> Dict[str, Any]:
"""Get a random bot for discovery."""
return await self._make_request("GET", "/bots/random")async def get_bot_analytics(self, period: str = "7d") -> Dict[str, Any]:
"""Get detailed bot analytics for a time period."""
return await self._make_request("GET", f"/bots/{self.bot_id}/analytics",
data={"period": period})
async def get_growth_stats(self) -> Dict[str, Any]:
"""Get bot growth statistics over time."""
return await self._make_request("GET", f"/bots/{self.bot_id}/growth")import asyncio
from functools import wraps
def retry_with_backoff(max_retries: int = 3, base_delay: float = 1.0):
"""Decorator for automatic retry with exponential backoff."""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except InfinityAPIError as e:
if e.status_code == 429 and attempt < max_retries - 1: # Rate limited
delay = base_delay * (2 ** attempt)
await asyncio.sleep(delay)
continue
raise
return None
return wrapper
return decorator
# Apply to API methods
@retry_with_backoff(max_retries=3)
async def post_bot_stats(self, server_count: int, user_count: Optional[int] = None,
shard_count: Optional[int] = None, shard_list: Optional[List[int]] = None) -> Dict[str, Any]:
# ... existing implementationimport time
from typing import Optional
class SimpleCache:
def __init__(self, ttl: int = 300): # 5 minutes default
self.cache = {}
self.ttl = ttl
def get(self, key: str) -> Optional[Any]:
if key in self.cache:
value, timestamp = self.cache[key]
if time.time() - timestamp < self.ttl:
return value
else:
del self.cache[key]
return None
def set(self, key: str, value: Any):
self.cache[key] = (value, time.time())
# Add to InfinityClient
def __init__(self, bot_token: str, bot_id: str, enable_cache: bool = True):
# ... existing init
self.cache = SimpleCache() if enable_cache else None
async def get_bot_info(self) -> Dict[str, Any]:
"""Get bot information with caching."""
cache_key = f"bot_info_{self.bot_id}"
if self.cache:
cached = self.cache.get(cache_key)
if cached:
return cached
result = await self._make_request("GET", f"/bots/{self.bot_id}")
if self.cache:
self.cache.set(cache_key, result)
return resultfrom typing import Callable
import hmac
import hashlib
class WebhookHandler:
def __init__(self, secret: str):
self.secret = secret.encode()
self.handlers = {}
def register_handler(self, event_type: str, handler: Callable):
"""Register a handler for a specific webhook event."""
self.handlers[event_type] = handler
def verify_signature(self, payload: bytes, signature: str) -> bool:
"""Verify webhook signature."""
expected = hmac.new(self.secret, payload, hashlib.sha256).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)
async def handle_webhook(self, payload: dict, signature: str):
"""Handle incoming webhook."""
event_type = payload.get("type")
if event_type in self.handlers:
await self.handlers[event_type](payload)
# Usage example
webhook_handler = WebhookHandler("your_webhook_secret")
@webhook_handler.register_handler("vote")
async def handle_vote(payload):
user_id = payload["user_id"]
bot_id = payload["bot_id"]
print(f"User {user_id} voted for bot {bot_id}!")
@webhook_handler.register_handler("review")
async def handle_review(payload):
user_id = payload["user_id"]
rating = payload["rating"]
print(f"New {rating}-star review from user {user_id}!")import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class PyfinityConfig:
bot_token: str
bot_id: str
base_url: str = "https://spider.infinitybots.gg/api"
timeout: float = 30.0
max_retries: int = 3
enable_cache: bool = True
cache_ttl: int = 300
@classmethod
def from_env(cls) -> 'PyfinityConfig':
"""Create config from environment variables."""
return cls(
bot_token=os.getenv("INFINITY_BOT_TOKEN", ""),
bot_id=os.getenv("BOT_ID", ""),
base_url=os.getenv("INFINITY_API_URL", "https://spider.infinitybots.gg/api"),
timeout=float(os.getenv("INFINITY_TIMEOUT", "30.0")),
max_retries=int(os.getenv("INFINITY_MAX_RETRIES", "3")),
enable_cache=os.getenv("INFINITY_ENABLE_CACHE", "true").lower() == "true",
cache_ttl=int(os.getenv("INFINITY_CACHE_TTL", "300"))
)
# Enhanced InfinityClient
class InfinityClient:
def __init__(self, config: PyfinityConfig = None, **kwargs):
if config is None:
# Use individual parameters for backward compatibility
config = PyfinityConfig(**kwargs)
self.config = config
self.bot_token = config.bot_token
self.bot_id = config.bot_id
# ... rest of initThese examples show how you can quickly implement some of the high-priority features. Each can be added incrementally to your existing codebase!