Python SDK

Python SDK

SDK moderno para integração com API Xtream UI usando Python

Instalação

Instale as dependências necessárias usando pip:

Terminal 
pip install requests aiohttp asyncio
requirements.txt 
requests>=2.28.0
aiohttp>=3.8.0
asyncio>=3.4.3

Configuração

Crie a classe principal do SDK:

xtream_api.py 
import requests
import json
import asyncio
import aiohttp
from typing import Dict, List, Optional, Any
from urllib.parse import urljoin

class XtreamAPI:
    def __init__(self, base_url: str, username: Optional[str] = None, password: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.username = username
        self.password = password
        self.session = requests.Session()
        self.session.verify = False  # Disable SSL verification if needed
        
    def _make_request(self, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
        """Make GET request to API endpoint"""
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"Request failed: {str(e)}")
        except json.JSONDecodeError as e:
            raise Exception(f"Invalid JSON response: {str(e)}")
    
    def _make_post_request(self, endpoint: str, data: Dict[str, Any] = None) -> Dict[str, Any]:
        """Make POST request to API endpoint"""
        url = f"{self.base_url}{endpoint}"
        
        try:
            response = self.session.post(url, data=data, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            raise Exception(f"Request failed: {str(e)}")
        except json.JSONDecodeError as e:
            raise Exception(f"Invalid JSON response: {str(e)}")

Autenticação

Métodos para autenticação e verificação de credenciais:

Métodos de Autenticação 
def authenticate(self, username: str, password: str) -> Dict[str, Any]:
    """Authenticate user and get user info"""
    params = {
        'username': username,
        'password': password
    }
    
    response = self._make_request('/player_api.php', params)
    
    if 'user_info' in response:
        self.username = username
        self.password = password
        return response
    
    raise Exception('Invalid credentials')

def get_user_info(self) -> Dict[str, Any]:
    """Get authenticated user information"""
    if not self.username or not self.password:
        raise Exception('User not authenticated')
    
    params = {
        'username': self.username,
        'password': self.password
    }
    
    return self._make_request('/player_api.php', params)

def panel_login(self, username: str, password: str) -> str:
    """Login to admin panel and get token"""
    data = {
        'username': username,
        'password': password
    }
    
    response = self._make_post_request('/api.php?action=login', data)
    
    if response.get('status') == 'success':
        return response['token']
    
    raise Exception('Panel login failed')

Gerenciamento de Usuários

Métodos para criar, editar e gerenciar usuários:

Métodos de Usuários 
def create_user(self, token: str, user_data: Dict[str, Any]) -> Dict[str, Any]:
    """Create new user"""
    data = {
        'action': 'user',
        'sub': 'create',
        'token': token,
        **user_data
    }
    
    return self._make_post_request('/api.php', data)

def edit_user(self, token: str, user_id: int, user_data: Dict[str, Any]) -> Dict[str, Any]:
    """Edit existing user"""
    data = {
        'action': 'user',
        'sub': 'edit',
        'user_id': user_id,
        'token': token,
        **user_data
    }
    
    return self._make_post_request('/api.php', data)

def delete_user(self, token: str, user_id: int) -> Dict[str, Any]:
    """Delete user"""
    data = {
        'action': 'user',
        'sub': 'delete',
        'user_id': user_id,
        'token': token
    }
    
    return self._make_post_request('/api.php', data)

def get_user_details(self, token: str, user_id: int) -> Dict[str, Any]:
    """Get user details"""
    params = {
        'action': 'user',
        'sub': 'info',
        'user_id': user_id,
        'token': token
    }
    
    return self._make_request('/api.php', params)

def get_all_users(self, token: str) -> List[Dict[str, Any]]:
    """Get all users"""
    params = {
        'action': 'user',
        'sub': 'list',
        'token': token
    }
    
    return self._make_request('/api.php', params)

Canais e Conteúdo

Métodos para gerenciar canais, VOD e séries:

Métodos de Conteúdo 
def get_live_streams(self) -> List[Dict[str, Any]]:
    """Get all live streams"""
    params = {
        'username': self.username,
        'password': self.password,
        'action': 'get_live_streams'
    }
    
    return self._make_request('/player_api.php', params)

def get_vod_streams(self) -> List[Dict[str, Any]]:
    """Get all VOD streams"""
    params = {
        'username': self.username,
        'password': self.password,
        'action': 'get_vod_streams'
    }
    
    return self._make_request('/player_api.php', params)

def get_series(self) -> List[Dict[str, Any]]:
    """Get all series"""
    params = {
        'username': self.username,
        'password': self.password,
        'action': 'get_series'
    }
    
    return self._make_request('/player_api.php', params)

def get_stream_info(self, stream_id: int, stream_type: str = 'live') -> Dict[str, Any]:
    """Get stream information"""
    params = {
        'username': self.username,
        'password': self.password,
        'action': f'get_{stream_type}_info',
        'stream_id': stream_id
    }
    
    return self._make_request('/player_api.php', params)

def get_categories(self, category_type: str = 'live') -> List[Dict[str, Any]]:
    """Get categories by type"""
    action_map = {
        'live': 'get_live_categories',
        'vod': 'get_vod_categories',
        'series': 'get_series_categories'
    }
    
    params = {
        'username': self.username,
        'password': self.password,
        'action': action_map.get(category_type, 'get_live_categories')
    }
    
    return self._make_request('/player_api.php', params)

Relatórios

Métodos para gerar relatórios e estatísticas:

Métodos de Relatórios 
def get_server_stats(self, token: str) -> Dict[str, Any]:
    """Get server statistics"""
    params = {
        'action': 'server_stats',
        'token': token
    }
    
    return self._make_request('/api.php', params)

def get_user_connections(self, token: str, user_id: Optional[int] = None) -> Dict[str, Any]:
    """Get user connection activity"""
    params = {
        'action': 'user_activity',
        'token': token
    }
    
    if user_id:
        params['user_id'] = user_id
    
    return self._make_request('/api.php', params)

def get_stream_stats(self, token: str, stream_id: int, stream_type: str = 'live') -> Dict[str, Any]:
    """Get stream statistics"""
    params = {
        'action': 'stream_stats',
        'token': token,
        'stream_id': stream_id,
        'type': stream_type
    }
    
    return self._make_request('/api.php', params)

Async/Await

Versão assíncrona do SDK para melhor performance:

XtreamAsyncAPI 
class XtreamAsyncAPI:
    def __init__(self, base_url: str, username: Optional[str] = None, password: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.username = username
        self.password = password
        
    async def _make_async_request(self, endpoint: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
        """Make async GET request"""
        url = f"{self.base_url}{endpoint}"
        
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(url, params=params, ssl=False, timeout=30) as response:
                    response.raise_for_status()
                    return await response.json()
            except aiohttp.ClientError as e:
                raise Exception(f"Request failed: {str(e)}")
    
    async def _make_async_post_request(self, endpoint: str, data: Dict[str, Any] = None) -> Dict[str, Any]:
        """Make async POST request"""
        url = f"{self.base_url}{endpoint}"
        
        async with aiohttp.ClientSession() as session:
            try:
                async with session.post(url, data=data, ssl=False, timeout=30) as response:
                    response.raise_for_status()
                    return await response.json()
            except aiohttp.ClientError as e:
                raise Exception(f"Request failed: {str(e)}")
    
    async def authenticate_async(self, username: str, password: str) -> Dict[str, Any]:
        """Async authentication"""
        params = {
            'username': username,
            'password': password
        }
        
        response = await self._make_async_request('/player_api.php', params)
        
        if 'user_info' in response:
            self.username = username
            self.password = password
            return response
        
        raise Exception('Invalid credentials')
    
    async def get_all_content_async(self) -> Dict[str, List[Dict[str, Any]]]:
        """Get all content types concurrently"""
        tasks = [
            self.get_live_streams_async(),
            self.get_vod_streams_async(),
            self.get_series_async()
        ]
        
        live, vod, series = await asyncio.gather(*tasks)
        
        return {
            'live_streams': live,
            'vod_streams': vod,
            'series': series
        }
    
    async def get_live_streams_async(self) -> List[Dict[str, Any]]:
        """Get live streams async"""
        params = {
            'username': self.username,
            'password': self.password,
            'action': 'get_live_streams'
        }
        
        return await self._make_async_request('/player_api.php', params)

Tratamento de Erros

Exemplo completo de uso com tratamento de erros:

exemplo_completo.py 
#!/usr/bin/env python3
import asyncio
import logging
from datetime import datetime, timedelta
from xtream_api import XtreamAPI, XtreamAsyncAPI

# Configurar logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

async def exemplo_completo():
    try:
        # Inicializar SDK síncrono
        api = XtreamAPI('http://dominio_da_instancia:8080')
        
        # Autenticar usuário
        user_auth = api.authenticate('usuario_teste', 'senha123')
        logger.info(f"Usuário autenticado: {user_auth['user_info']['username']}")
        
        # Login no painel
        token = api.panel_login('admin', 'admin_password')
        logger.info(f"Token do painel obtido: {token[:10]}...")
        
        # Criar novo usuário
        novo_usuario = {
            'username': f'novo_usuario_{int(datetime.now().timestamp())}',
            'password': 'senha_forte_123',
            'email': 'usuario@email.com',
            'exp_date': (datetime.now() + timedelta(days=30)).strftime('%Y-%m-%d'),
            'max_connections': 2,
            'is_trial': 0,
            'enabled': 1
        }
        
        resultado = api.create_user(token, novo_usuario)
        
        if resultado.get('status') == 'success':
            user_id = resultado['user_id']
            logger.info(f"Usuário criado com sucesso! ID: {user_id}")
            
            # Obter informações do usuário criado
            info_usuario = api.get_user_details(token, user_id)
            logger.info(f"Detalhes do usuário: {info_usuario}")
            
            # Usar SDK assíncrono para operações paralelas
            async_api = XtreamAsyncAPI('http://dominio_da_instancia:8080', 
                                     'usuario_teste', 'senha123')
            
            # Obter todo o conteúdo de forma assíncrona
            conteudo = await async_api.get_all_content_async()
            
            logger.info(f"Total de canais ao vivo: {len(conteudo['live_streams'])}")
            logger.info(f"Total de VODs: {len(conteudo['vod_streams'])}")
            logger.info(f"Total de séries: {len(conteudo['series'])}")
            
            # Obter estatísticas do servidor
            stats = api.get_server_stats(token)
            logger.info(f"Usuários online: {stats.get('total_users_online', 'N/A')}")
            
        else:
            logger.error(f"Erro ao criar usuário: {resultado.get('message', 'Erro desconhecido')}")
            
    except Exception as e:
        logger.error(f"Erro: {str(e)}")
        
        # Salvar erro em arquivo de log
        with open('/tmp/xtream_sdk_errors.log', 'a') as f:
            f.write(f"{datetime.now().isoformat()} - ERRO: {str(e)}\n")

def exemplo_sincrono():
    """Exemplo usando apenas métodos síncronos"""
    try:
        api = XtreamAPI('http://dominio_da_instancia:8080')
        
        # Autenticar
        api.authenticate('usuario_teste', 'senha123')
        
        # Obter canais
        canais = api.get_live_streams()
        print(f"Encontrados {len(canais)} canais")
        
        # Filtrar canais por categoria
        categorias = api.get_categories('live')
        for categoria in categorias[:5]:  # Primeiras 5 categorias
            print(f"Categoria: {categoria.get('category_name', 'N/A')}")
        
        # Obter informações de um canal específico
        if canais:
            primeiro_canal = canais[0]
            info_canal = api.get_stream_info(primeiro_canal['stream_id'], 'live')
            print(f"Canal: {info_canal.get('info', {}).get('name', 'N/A')}")
            
    except Exception as e:
        print(f"Erro no exemplo síncrono: {str(e)}")

if __name__ == "__main__":
    print("=== Exemplo Assíncrono ===")
    asyncio.run(exemplo_completo())
    
    print("\n=== Exemplo Síncrono ===")
    exemplo_sincrono()
Dica: Use a versão assíncrona para operações que requerem múltiplas requisições simultâneas, melhorando significativamente a performance.