deepinfra_client.py

import requests
import json
import os
import random
import time
from typing import Any, Dict, Optional, Generator, Union, List
import warnings
from fake_useragent import UserAgent
from requests_ip_rotator import ApiGateway
from proxy_finder import ProxyFinder

class IPRotator:
    """Manages AWS API Gateway rotation for multiple regions"""
    
    def __init__(self, target_url="deepinfra.com", regions=None):
        """Initialize with target URL and regions"""
        self.target_url = target_url
        self.regions = regions or ["us-east-1", "us-west-1", "eu-west-1", "ap-southeast-1"]
        self.gateways = {}
        
    def setup(self):
        """Set up API gateways for each region"""
        for region in self.regions:
            try:
                gateway = ApiGateway(self.target_url, region=region)
                gateway.start()
                self.gateways[region] = gateway
            except Exception as e:
                print(f"Failed to set up gateway in {region}: {str(e)}")
        
        if not self.gateways:
            raise Exception("Failed to set up any API gateways for IP rotation")
    
    def get_session(self):
        """Get a random session from a gateway"""
        if not self.gateways:
            return requests.Session()
            
        # Choose a random gateway
        region = random.choice(list(self.gateways.keys()))
        gateway = self.gateways[region]
        return gateway.get_session()
    
    def shutdown(self):
        """Clean up all gateways"""
        for gateway in self.gateways.values():
            try:
                gateway.shutdown()
            except:
                pass

class ProxyManager:
    """Manages proxy rotation for HTTP requests"""
    
    def __init__(self, proxies=None):
        """Initialize with a list of proxies or an empty list"""
        self.proxies = proxies or []
        
    def add_proxy(self, proxy):
        """Add a proxy to the list"""
        self.proxies.append(proxy)
        
    def get_random(self):
        """Return a random proxy if available, otherwise None"""
        if not self.proxies:
            return None
        return random.choice(self.proxies)

class DeepInfraClient:
    """

    A client for DeepInfra API with OpenAI-compatible interface and enhanced features

    """

    AVAILABLE_MODELS = [
        "deepseek-ai/DeepSeek-R1-Turbo",
        "deepseek-ai/DeepSeek-R1",
        "deepseek-ai/DeepSeek-R1-Distill-Llama-70B",
        "deepseek-ai/DeepSeek-V3",
        "meta-llama/Llama-3.3-70B-Instruct-Turbo",
        "mistralai/Mistral-Small-24B-Instruct-2501",
        "deepseek-ai/DeepSeek-R1-Distill-Qwen-32B",
        "microsoft/phi-4",
        "meta-llama/Meta-Llama-3.1-70B-Instruct",
        "meta-llama/Meta-Llama-3.1-8B-Instruct",
        "meta-llama/Meta-Llama-3.1-405B-Instruct",
        "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo",
        "meta-llama/Meta-Llama-3.1-70B-Instruct-Turbo",
        "Qwen/Qwen2.5-Coder-32B-Instruct",
        "nvidia/Llama-3.1-Nemotron-70B-Instruct",
        "Qwen/Qwen2.5-72B-Instruct",
        "meta-llama/Llama-3.2-90B-Vision-Instruct",
        "meta-llama/Llama-3.2-11B-Vision-Instruct",
        "Gryphe/MythoMax-L2-13b",
        "NousResearch/Hermes-3-Llama-3.1-405B",
        "NovaSky-AI/Sky-T1-32B-Preview",
        "Qwen/Qwen2.5-7B-Instruct",
        "Sao10K/L3.1-70B-Euryale-v2.2",
        "Sao10K/L3.3-70B-Euryale-v2.3",
        "google/gemma-2-27b-it",
        "google/gemma-2-9b-it",
        "meta-llama/Llama-3.2-1B-Instruct",
        "meta-llama/Llama-3.2-3B-Instruct",
        "meta-llama/Meta-Llama-3-70B-Instruct",
        "meta-llama/Meta-Llama-3-8B-Instruct",
        "mistralai/Mistral-Nemo-Instruct-2407",
        "mistralai/Mistral-7B-Instruct-v0.3",
        "mistralai/Mixtral-8x7B-Instruct-v0.1"
    ]

    def __init__(

        self,

        api_key: Optional[str] = None,

        base_url: str = "https://api.deepinfra.com/v1",

        timeout: int = 30,

        max_tokens: int = 2049,

        model: str = "meta-llama/Llama-3.3-70B-Instruct-Turbo",

        use_random_user_agent: bool = True,

        use_proxy_rotation: bool = True,

        use_ip_rotation: bool = True,

        proxy_types: List[str] = None

    ):
        """Initialize the DeepInfraClient"""
        self.base_url = base_url
        self.api_key = api_key
        self.model = model
        self.timeout = timeout
        self.max_tokens = max_tokens
        self.use_random_user_agent = use_random_user_agent
        self.use_ip_rotation = use_ip_rotation
        self.use_proxy_rotation = use_proxy_rotation
        self.proxy_types = proxy_types or ['http', 'socks5'] # Default proxy types
        
        # Initialize user agent generator
        self.user_agent = UserAgent()
        
        # Set up proxy finder and get initial proxies if proxy rotation is enabled
        self.proxy_finder = None
        if self.use_proxy_rotation:
            self.proxy_finder = ProxyFinder(verbose=False)
            self.proxy_finder.get_proxies(self.proxy_types)
        
        # Set up IP rotator if enabled
        self.ip_rotator = None
        if use_ip_rotation:
            try:
                self.ip_rotator = IPRotator(target_url="deepinfra.com")
                self.ip_rotator.setup()
            except Exception as e:
                print(f"Failed to set up IP rotation: {e}. Continuing without IP rotation.")
                self.ip_rotator = None
            
        # Set up headers with random or fixed user agent
        self.headers = self._create_headers()
        
        # Initialize session based on available rotation methods
        if self.use_ip_rotation and self.ip_rotator:
            self.session = self.ip_rotator.get_session()
        else:
            self.session = requests.Session()
            
        self.session.headers.update(self.headers)
        
        # Apply proxy if proxy rotation is enabled
        if self.use_proxy_rotation and self.proxy_finder:
            self._apply_random_proxy()
        
        # Resources
        self.models = Models(self)
        self.chat = ChatCompletions(self)
        self.completions = Completions(self)
        
    def _create_headers(self) -> Dict[str, str]:
        """Create headers for the HTTP request, optionally with a random user agent"""
        user_agent = self.user_agent.random if self.use_random_user_agent else \
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36'
        
        headers = {
            'User-Agent': user_agent,
            'Accept-Language': 'en-US,en;q=0.9',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'Content-Type': 'application/json',
            'Origin': 'https://deepinfra.com',
            'Referer': 'https://deepinfra.com/',
            'Sec-Fetch-Dest': 'empty',
            'Sec-Fetch-Mode': 'cors',
            'Sec-Fetch-Site': 'same-site',
            'X-Deepinfra-Source': 'web-embed',
            'accept': 'text/event-stream',
            'sec-ch-ua-mobile': '?0',
            'sec-ch-ua-platform': '"macOS"'
        }
        
        if self.api_key:
            headers['Authorization'] = f"Bearer {self.api_key}"
            
        return headers
    
    def _apply_random_proxy(self):
        """Apply a random proxy to the current session"""
        if not self.proxy_finder:
            return False
            
        # First try to get a proxy of preferred type (http/https first, then socks5)
        for proxy_type in self.proxy_types:
            proxy = self.proxy_finder.get_random_proxy(proxy_type)
            if proxy:
                if proxy_type in ['http', 'https']:
                    self.session.proxies.update({
                        "http": f"http://{proxy}",
                        "https": f"http://{proxy}"
                    })
                    return True
                elif proxy_type == 'socks4':
                    self.session.proxies.update({
                        "http": f"socks4://{proxy}",
                        "https": f"socks4://{proxy}"
                    })
                    return True
                elif proxy_type == 'socks5':
                    self.session.proxies.update({
                        "http": f"socks5://{proxy}",
                        "https": f"socks5://{proxy}"
                    })
                    return True
        
        # If no proxy found, return False
        return False
    
    def refresh_session(self):
        """Refresh the session with new headers and possibly a new proxy or IP"""
        if self.use_random_user_agent:
            self.headers['User-Agent'] = self.user_agent.random
            
        # Apply a random proxy if proxy rotation is enabled
        if self.use_proxy_rotation and self.proxy_finder:
            proxy_applied = self._apply_random_proxy()
            # If no proxy was applied, try to get new proxies
            if not proxy_applied:
                self.proxy_finder.get_proxies(self.proxy_types)
                self._apply_random_proxy()
        
        # Rotate IP if enabled
        if self.use_ip_rotation and self.ip_rotator:
            self.session = self.ip_rotator.get_session()
            
        self.session.headers.update(self.headers)
        
    def _request(self, method: str, endpoint: str, **kwargs) -> requests.Response:
        """Make an HTTP request with automatic retry and proxy/user-agent rotation"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        max_retries = 3
        retry_delay = 1
        
        for attempt in range(max_retries):
            try:
                response = self.session.request(method, url, **kwargs)
                response.raise_for_status()
                return response
            except requests.RequestException as e:
                if attempt < max_retries - 1:
                    self.refresh_session()
                    time.sleep(retry_delay * (attempt + 1))
                    continue
                raise e
    
    def refresh_proxies(self):
        """Refresh all proxies by fetching new ones"""
        if self.proxy_finder:
            self.proxy_finder.get_proxies(self.proxy_types)
            self._apply_random_proxy()
            return True
        return False
        
    def __del__(self):
        """Clean up resources on deletion"""
        if self.ip_rotator:
            try:
                self.ip_rotator.shutdown()
            except:
                pass

class Models:
    def __init__(self, client: DeepInfraClient):
        self.client = client
        
    def list(self) -> Dict[str, Any]:
        """Get available models, similar to OpenAI's /v1/models endpoint"""
        model_data = []
        for model_id in self.client.AVAILABLE_MODELS:
            model_data.append({
                "id": model_id,
                "object": "model",
                "created": 1677610602,
                "owned_by": "deepinfra"
            })
        
        return {
            "object": "list",
            "data": model_data
        }

class ChatCompletions:
    def __init__(self, client: DeepInfraClient):
        self.client = client
    
    def create(

        self,

        messages: List[Dict[str, str]],

        model: str = None,

        temperature: float = 0.7,

        max_tokens: int = None,

        stream: bool = False,

        **kwargs

    ) -> Union[Dict[str, Any], Generator[Dict[str, Any], None, None]]:
        """Create a chat completion, similar to OpenAI's chat/completions endpoint"""
        model = model or self.client.model
        max_tokens = max_tokens or self.client.max_tokens
        
        url = "openai/chat/completions"
        
        # Prepare the payload for the API request
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "stream": stream
        }
        
        # Add any additional parameters
        payload.update({k: v for k, v in kwargs.items() if v is not None})
        
        if stream:
            return self._handle_stream(url, payload)
        else:
            return self._handle_request(url, payload)
    
    def _handle_request(self, url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
        """Handle non-streaming requests"""
        try:
            response = self.client._request(
                "POST",
                url,
                json=payload,
                timeout=self.client.timeout
            )
            return response.json()
        except requests.RequestException as e:
            error_message = f"Request failed: {str(e)}"
            if hasattr(e, 'response') and e.response is not None:
                try:
                    error_data = e.response.json()
                    if 'error' in error_data:
                        error_message = f"API error: {error_data['error']}"
                except:
                    error_message = f"API error: {e.response.text}"
            
            raise Exception(error_message)
    
    def _handle_stream(self, url: str, payload: Dict[str, Any]) -> Generator[Dict[str, Any], None, None]:
        """Handle streaming requests"""
        try:
            response = self.client._request(
                "POST", 
                url,
                json=payload,
                stream=True,
                timeout=self.client.timeout
            )
            
            for line in response.iter_lines(decode_unicode=True):
                if line:
                    line = line.strip()
                    if line.startswith("data: "):
                        json_str = line[6:]
                        if json_str == "[DONE]":
                            break
                        try:
                            json_data = json.loads(json_str)
                            yield json_data
                        except json.JSONDecodeError:
                            continue
        except requests.RequestException as e:
            error_message = f"Stream request failed: {str(e)}"
            if hasattr(e, 'response') and e.response is not None:
                try:
                    error_data = e.response.json()
                    if 'error' in error_data:
                        error_message = f"API error: {error_data['error']}"
                except:
                    error_message = f"API error: {e.response.text}"
            
            raise Exception(error_message)

class Completions:
    def __init__(self, client: DeepInfraClient):
        self.client = client
    
    def create(

        self,

        prompt: str,

        model: str = None,

        temperature: float = 0.7,

        max_tokens: int = None,

        stream: bool = False,

        **kwargs

    ) -> Union[Dict[str, Any], Generator[Dict[str, Any], None, None]]:
        """Create a completion, similar to OpenAI's completions endpoint"""
        # Convert prompt to messages format for chat models
        messages = [{"role": "user", "content": prompt}]
        
        return self.client.chat.create(
            messages=messages,
            model=model,
            temperature=temperature,
            max_tokens=max_tokens,
            stream=stream,
            **kwargs
        )

if __name__ == "__main__":
    import time
    from rich import print
    
    # Example with random user agent, proxy rotation and IP rotation
    client = DeepInfraClient(
        use_random_user_agent=True,
        use_ip_rotation=True,
        use_proxy_rotation=True,
        proxy_types=['http', 'socks5']
    )
    
    # Get available models
    models_response = client.models.list()
    print("Available models:")
    for model in models_response["data"][:5]:  # Print first 5 models
        print(f"- {model['id']}")
    print("...")
    
    # Non-streaming chat completion
    chat_response = client.chat.create(
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Write a short poem about AI"}
        ]
    )
    print("\nNon-streaming response:")
    print(chat_response["choices"][0]["message"]["content"])
    
    # Refresh proxies and try again with another request
    print("\nRefreshing proxies and making another request...")
    client.refresh_proxies()
    client.refresh_session()
    
    streaming_response = client.chat.create(
        messages=[
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "Tell me about the future of AI in 3 sentences."}
        ],
        stream=True
    )
    
    print("\nStreaming response:")
    full_response = ""
    for chunk in streaming_response:
        if 'choices' in chunk and len(chunk['choices']) > 0:
            delta = chunk['choices'][0].get('delta', {})
            if 'content' in delta:
                content = delta['content']
                print(content, end='', flush=True)
                full_response += content
    print("\n")