API Limits and Constraints

HTTP/1.1 202 Accepted
X-RateLimit-Limit: 600000
X-RateLimit-Remaining: 599950
X-RateLimit-Reset: 1640995260
Content-Type: application/json

# Convert to supported format with FFmpeg
ffmpeg -i input.aac -f mp3 output.mp3  # Convert AAC to MP3
ffmpeg -i input.m4a -f wav output.wav  # Convert M4A to WAV

# Optimize MP3 file
ffmpeg -i input.wav \
  -ar 16000 \          # Sample rate: 16kHz
  -ab 128k \           # Bit rate: 128kbps
  -ac 1 \              # Mono channel
  -f mp3 \             # MP3 format
  output.mp3

# Check file info
ffprobe -v quiet -show_entries format=duration,size,bit_rate input.mp3

import time
import random
from typing import Optional

class APIRetryHandler:
    """Robust retry handler for LigueLead API calls."""
    
    def __init__(self, max_retries: int = 3, base_delay: float = 1.0):
        self.max_retries = max_retries
        self.base_delay = base_delay
    
    def exponential_backoff(self, attempt: int) -> float:
        """Calculate delay with exponential backoff and jitter."""
        delay = self.base_delay * (2 ** attempt)
        jitter = random.uniform(0.1, 0.3) * delay
        return delay + jitter
    
    async def retry_request(self, request_func, *args, **kwargs):
        """Execute request with retry logic."""
        
        last_exception = None
        
        for attempt in range(self.max_retries + 1):
            try:
                response = await request_func(*args, **kwargs)
                
                # Handle rate limiting
                if response.status_code == 429:
                    if attempt < self.max_retries:
                        print(f"Rate limit hit, waiting 60 seconds... (attempt {attempt + 1})")
                        await asyncio.sleep(60)
                        continue
                    else:
                        raise Exception("Rate limit exceeded after max retries")
                
                # Handle server errors (5xx)
                elif 500 <= response.status_code < 600:
                    if attempt < self.max_retries:
                        delay = self.exponential_backoff(attempt)
                        print(f"Server error, retrying in {delay:.1f}s... (attempt {attempt + 1})")
                        await asyncio.sleep(delay)
                        continue
                    else:
                        raise Exception(f"Server error after {self.max_retries} retries")
                
                # Success or client error (don't retry client errors)
                return response
                
            except Exception as e:
                last_exception = e
                if attempt < self.max_retries:
                    delay = self.exponential_backoff(attempt)
                    print(f"Request failed, retrying in {delay:.1f}s... (attempt {attempt + 1})")
                    await asyncio.sleep(delay)
                else:
                    break
        
        raise last_exception

# Usage
retry_handler = APIRetryHandler(max_retries=3)
response = await retry_handler.retry_request(api_call_function, data)

import asyncio
import aiohttp
from asyncio import Semaphore

class ConcurrentAPIClient:
    """Manage concurrent API operations efficiently."""
    
    def __init__(self, upload_limit: int = 3, send_limit: int = 5):
        self.upload_semaphore = Semaphore(upload_limit)
        self.send_semaphore = Semaphore(send_limit)
    
    async def upload_audio_file(self, session, file_path, title):
        """Upload audio file with concurrency control."""
        async with self.upload_semaphore:
            # Your upload logic here
            return await self._do_upload(session, file_path, title)
    
    async def send_voice_message(self, session, voice_id, phones, title):
        """Send voice message with concurrency control."""
        async with self.send_semaphore:
            # Your send logic here
            return await self._do_send(session, voice_id, phones, title)
    
    async def batch_operations(self, operations):
        """Execute multiple operations concurrently."""
        async with aiohttp.ClientSession() as session:
            tasks = []
            
            for op in operations:
                if op['type'] == 'upload':
                    task = self.upload_audio_file(session, op['file'], op['title'])
                elif op['type'] == 'send':
                    task = self.send_voice_message(session, op['voice_id'], op['phones'], op['title'])
                
                tasks.append(task)
            
            results = await asyncio.gather(*tasks, return_exceptions=True)
            return results

# ✅ Efficient: Single request with multiple recipients
payload = {
    "title": "Monthly Newsletter",
    "message": "Check out our latest updates!",
    "phones": ["11999999999", "21888888888", "31777777777"]  # Up to 10k numbers
}

# ❌ Inefficient: Multiple requests for individual recipients
for phone in phones:
    payload = {"message": "Hello", "phones": [phone]}  # Wastes rate limit

class AudioFileManager:
    """Optimize audio file usage and storage."""
    
    def __init__(self):
        self.uploaded_files = {}  # Cache uploaded file IDs
    
    def upload_once_use_many(self, file_path, title, api_token, app_id):
        """Upload audio file once and reuse across campaigns."""
        
        # Check if already uploaded
        file_key = f"{title}_{hash(file_path)}"
        if file_key in self.uploaded_files:
            print(f"♻️  Reusing existing upload: {self.uploaded_files[file_key]}")
            return self.uploaded_files[file_key]
        
        # Upload new file
        with open(file_path, 'rb') as audio_file:
            files = {'file': audio_file}
            data = {'title': title}
            headers = {'api-token': api_token, 'app-id': app_id}
            
            response = requests.post(
                "https://api.liguelead.com.br/v1/voice/uploads",
                headers=headers,
                files=files,
                data=data
            )
            
            if response.status_code == 201:
                voice_id = response.json()['data']['id']
                self.uploaded_files[file_key] = voice_id
                print(f"✅ New upload successful: {voice_id}")
                return voice_id
            else:
                raise Exception(f"Upload failed: {response.text}")

# Usage
audio_manager = AudioFileManager()

# Upload once
voice_id = audio_manager.upload_once_use_many(
    "promo.mp3", 
    "Black Friday Promotion", 
    api_token, 
    app_id
)

# Use in multiple campaigns
campaigns = [
    {"title": "VIP Customers", "phones": vip_numbers},
    {"title": "Regular Customers", "phones": regular_numbers},
    {"title": "New Prospects", "phones": prospect_numbers}
]

for campaign in campaigns:
    send_voice_message(voice_id, campaign["phones"], campaign["title"])

import time
from collections import deque

class RateLimitManager:
    """Smart rate limit management to maximize throughput."""
    
    def __init__(self, limit_per_minute: int = 600000):
        self.limit = limit_per_minute
        self.requests = deque()
    
    def can_make_request(self) -> bool:
        """Check if we can make a request without hitting rate limit."""
        now = time.time()
        
        # Remove requests older than 1 minute
        while self.requests and self.requests[0] <= now - 60:
            self.requests.popleft()
        
        return len(self.requests) < self.limit
    
    def wait_for_slot(self):
        """Wait until we can make a request."""
        while not self.can_make_request():
            time.sleep(1)
    
    def record_request(self):
        """Record that a request was made."""
        self.requests.append(time.time())
    
    def make_controlled_request(self, request_func, *args, **kwargs):
        """Make request with automatic rate limiting."""
        self.wait_for_slot()
        result = request_func(*args, **kwargs)
        self.record_request()
        return result

# Usage
rate_manager = RateLimitManager()

for campaign in large_campaign_list:
    response = rate_manager.make_controlled_request(
        send_sms_message,
        campaign_data
    )