Exemplos Python

Exemplos Python para interagir com a API External do Efflux.

Configuração

Instalação

pip install requests python-dotenv

Variáveis de Ambiente

Crie um arquivo .env:

EFFLUX_API_KEY=efflux_live_suaempresa_abc123...

Classe Cliente da API

import os
from dataclasses import dataclass
from datetime import datetime
from typing import Optional, List, Dict, Any
import requests
from dotenv import load_dotenv

load_dotenv()


class EffluxApiError(Exception):
    """Exceção customizada para erros da API Efflux."""

    def __init__(self, status_code: int, code: str, message: str):
        self.status_code = status_code
        self.code = code
        self.message = message
        super().__init__(f"{code}: {message}")


@dataclass
class ContentItem:
    id: str
    name: str
    type: str
    mime_type: str
    status: str
    storage_url: str
    thumbnail_url: Optional[str]
    duration_seconds: Optional[int]
    created_at: str
    updated_at: str
    description: Optional[str] = None

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'ContentItem':
        return cls(
            id=data['id'],
            name=data['name'],
            type=data['type'],
            mime_type=data.get('mimeType', ''),
            status=data['status'],
            storage_url=data.get('storageUrl', ''),
            thumbnail_url=data.get('thumbnailUrl'),
            duration_seconds=data.get('durationSeconds'),
            created_at=data['createdAt'],
            updated_at=data['updatedAt'],
            description=data.get('description'),
        )


@dataclass
class Schedule:
    id: int
    name: str
    status: str
    content_count: int
    player_ids: List[str]
    content_ids: List[str]
    created_at: str
    updated_at: str
    description: Optional[str] = None

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'Schedule':
        return cls(
            id=data['id'],
            name=data['name'],
            status=data.get('status', ''),
            content_count=data.get('contentCount', 0),
            player_ids=data.get('playerIds', []),
            content_ids=data.get('contentIds', []),
            created_at=data['createdAt'],
            updated_at=data['updatedAt'],
            description=data.get('description'),
        )


@dataclass
class Player:
    id: str
    name: str
    status: str
    platform: str
    health_status: str
    last_seen_at: str
    created_at: str

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'Player':
        return cls(
            id=data['id'],
            name=data['name'],
            status=data.get('status', ''),
            platform=data.get('platform', ''),
            health_status=data.get('healthStatus', ''),
            last_seen_at=data.get('lastSeenAt', ''),
            created_at=data['createdAt'],
        )


@dataclass
class AnalyticsSummary:
    total_content: int
    active_content: int
    total_players: int
    total_plays: int
    total_duration_seconds: int
    last_played_at: Optional[str]

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'AnalyticsSummary':
        return cls(
            total_content=data['totalContent'],
            active_content=data['activeContent'],
            total_players=data['totalPlayers'],
            total_plays=data['totalPlays'],
            total_duration_seconds=data['totalDurationSeconds'],
            last_played_at=data.get('lastPlayedAt'),
        )


class EffluxClient:
    """Cliente para a API External do Efflux."""

    def __init__(self, api_key: Optional[str] = None, base_url: Optional[str] = None):
        self.api_key = api_key or os.environ.get('EFFLUX_API_KEY')
        if not self.api_key:
            raise ValueError("API key é obrigatória")

        self.base_url = base_url or 'https://api.efflux.media/api/external'
        self.session = requests.Session()
        self.session.headers.update({
            'X-API-Key': self.api_key,
            'Accept': 'application/json',
        })

    def _request(self, endpoint: str, params: Optional[Dict] = None) -> Any:
        """Faz uma requisição à API."""
        url = f"{self.base_url}{endpoint}"
        response = self.session.get(url, params=params)

        if not response.ok:
            try:
                error = response.json()
                raise EffluxApiError(
                    response.status_code,
                    error.get('code', 'UNKNOWN'),
                    error.get('error', 'Erro desconhecido'),
                )
            except ValueError:
                raise EffluxApiError(
                    response.status_code,
                    'UNKNOWN',
                    response.text,
                )

        return response.json()

    # Métodos de Conteúdo
    def list_content(
        self,
        status: Optional[str] = None,
        limit: Optional[int] = None,
    ) -> List[ContentItem]:
        """Lista itens de conteúdo."""
        params = {}
        if status:
            params['status'] = status
        if limit:
            params['limit'] = limit

        data = self._request('/content', params)
        return [ContentItem.from_dict(item) for item in data]

    def get_content(self, content_id: str) -> ContentItem:
        """Obtém um item de conteúdo específico por ID."""
        data = self._request(f'/content/{content_id}')
        return ContentItem.from_dict(data)

    # Métodos de Programação
    def list_schedules(self, status: Optional[str] = None) -> List[Schedule]:
        """Lista programações."""
        params = {}
        if status:
            params['status'] = status

        data = self._request('/schedules', params)
        return [Schedule.from_dict(item) for item in data]

    # Métodos de Players
    def list_players(self) -> List[Player]:
        """Lista players."""
        data = self._request('/players')
        return [Player.from_dict(item) for item in data]

    # Métodos de Analytics
    def get_analytics_summary(self) -> AnalyticsSummary:
        """Obtém resumo de analytics."""
        data = self._request('/analytics/summary')
        return AnalyticsSummary.from_dict(data)

Exemplos de Uso

Uso Básico

from efflux_client import EffluxClient, EffluxApiError

# Inicializa o cliente (usa EFFLUX_API_KEY do ambiente)
client = EffluxClient()

# Listar todo o conteúdo
conteudo = client.list_content()
print(f"Total de conteúdos: {len(conteudo)}")

# Listar conteúdo ativo com limite
conteudo_ativo = client.list_content(status='active', limit=10)
for item in conteudo_ativo:
    print(f"- {item.name} ({item.type})")

Obter Conteúdo Específico

try:
    content = client.get_content('550e8400-e29b-41d4-a716-446655440000')
    print(f"Nome: {content.name}")
    print(f"Tipo: {content.type}")
    print(f"Status: {content.status}")
    print(f"URL: {content.storage_url}")
except EffluxApiError as e:
    if e.status_code == 404:
        print("Conteúdo não encontrado")
    else:
        print(f"Erro: {e.message}")

Monitorar Players

from datetime import datetime, timedelta, timezone

players = client.list_players()

# Agrupar por status de saúde
saudaveis = [p for p in players if p.health_status == 'healthy']
offline = [p for p in players if p.health_status == 'offline']

print(f"Saudáveis: {len(saudaveis)}, Offline: {len(offline)}")

# Encontrar players inativos (não vistos em 15 minutos)
limite = datetime.now(timezone.utc) - timedelta(minutes=15)

for player in players:
    if player.last_seen_at:
        last_seen = datetime.fromisoformat(player.last_seen_at.replace('Z', '+00:00'))
        if last_seen < limite:
            print(f"Aviso: {player.name} último acesso em {player.last_seen_at}")

Dashboard de Analytics

summary = client.get_analytics_summary()

print("=== Resumo de Analytics ===")
print(f"Conteúdo: {summary.active_content}/{summary.total_content} ativos")
print(f"Players: {summary.total_players}")
print(f"Total de reproduções: {summary.total_plays:,}")
print(f"Tempo total: {summary.total_duration_seconds // 3600} horas")
if summary.last_played_at:
    print(f"Última atividade: {summary.last_played_at}")

Aplicação Web Flask

from flask import Flask, jsonify
from efflux_client import EffluxClient, EffluxApiError
import time

app = Flask(__name__)
client = EffluxClient()


# Decorador simples de cache
def timed_cache(seconds: int):
    def decorator(func):
        cache = {}

        def wrapper(*args, **kwargs):
            key = str(args) + str(kwargs)
            if key in cache:
                result, timestamp = cache[key]
                if time.time() - timestamp < seconds:
                    return result

            result = func(*args, **kwargs)
            cache[key] = (result, time.time())
            return result

        return wrapper
    return decorator


@timed_cache(300)  # Cache por 5 minutos
def get_content_cached(status=None):
    return client.list_content(status=status)


@app.route('/api/content')
def list_content():
    try:
        content = get_content_cached(status='active')
        return jsonify([{
            'id': c.id,
            'name': c.name,
            'type': c.type,
            'status': c.status,
        } for c in content])
    except EffluxApiError as e:
        return jsonify({'error': e.message, 'code': e.code}), e.status_code


@app.route('/api/dashboard')
def dashboard():
    try:
        content = client.list_content()
        players = client.list_players()
        analytics = client.get_analytics_summary()

        return jsonify({
            'conteudo': {
                'total': len(content),
                'ativos': len([c for c in content if c.status == 'active']),
            },
            'players': {
                'total': len(players),
                'saudaveis': len([p for p in players if p.health_status == 'healthy']),
            },
            'analytics': {
                'totalReproducoes': analytics.total_plays,
                'totalHoras': analytics.total_duration_seconds // 3600,
            },
        })
    except EffluxApiError as e:
        return jsonify({'error': e.message}), e.status_code


if __name__ == '__main__':
    app.run(debug=True)

Ferramenta de Linha de Comando

#!/usr/bin/env python3
import argparse
import json
from efflux_client import EffluxClient, EffluxApiError


def main():
    parser = argparse.ArgumentParser(description='CLI da API Efflux')
    subparsers = parser.add_subparsers(dest='command', help='Comandos')

    # Comandos de Conteúdo
    content_parser = subparsers.add_parser('content', help='Operações de conteúdo')
    content_parser.add_argument('--status', help='Filtrar por status')
    content_parser.add_argument('--limit', type=int, help='Limitar resultados')
    content_parser.add_argument('--id', help='Obter conteúdo específico por ID')

    # Comando de Players
    subparsers.add_parser('players', help='Listar players')

    # Comando de Programações
    schedules_parser = subparsers.add_parser('schedules', help='Listar programações')
    schedules_parser.add_argument('--status', help='Filtrar por status')

    # Comando de Analytics
    subparsers.add_parser('analytics', help='Obter resumo de analytics')

    args = parser.parse_args()

    try:
        client = EffluxClient()

        if args.command == 'content':
            if args.id:
                content = client.get_content(args.id)
                print(json.dumps({
                    'id': content.id,
                    'name': content.name,
                    'type': content.type,
                    'status': content.status,
                    'url': content.storage_url,
                }, indent=2, ensure_ascii=False))
            else:
                content = client.list_content(status=args.status, limit=args.limit)
                for item in content:
                    print(f"{item.id}\t{item.name}\t{item.type}\t{item.status}")

        elif args.command == 'players':
            players = client.list_players()
            for p in players:
                print(f"{p.id}\t{p.name}\t{p.health_status}\t{p.last_seen_at}")

        elif args.command == 'schedules':
            schedules = client.list_schedules(status=args.status)
            for s in schedules:
                print(f"{s.id}\t{s.name}\t{s.status}\t{s.content_count} itens")

        elif args.command == 'analytics':
            summary = client.get_analytics_summary()
            print(f"Conteúdo: {summary.active_content}/{summary.total_content} ativos")
            print(f"Players: {summary.total_players}")
            print(f"Total de reproduções: {summary.total_plays:,}")
            print(f"Tempo total: {summary.total_duration_seconds // 3600} horas")

        else:
            parser.print_help()

    except EffluxApiError as e:
        print(f"Erro: {e.message} ({e.code})")
        exit(1)


if __name__ == '__main__':
    main()

Uso do CLI

# Listar todo o conteúdo
python efflux_cli.py content

# Listar conteúdo ativo
python efflux_cli.py content --status active --limit 10

# Obter conteúdo específico
python efflux_cli.py content --id 550e8400-e29b-41d4-a716-446655440000

# Listar players
python efflux_cli.py players

# Obter analytics
python efflux_cli.py analytics

Tratamento de Erros

from efflux_client import EffluxClient, EffluxApiError

client = EffluxClient()

def safe_api_call(func, *args, **kwargs):
    """Wrapper para chamadas seguras à API com tratamento de erros."""
    try:
        return func(*args, **kwargs)
    except EffluxApiError as e:
        if e.status_code == 401:
            print(f"Autenticação falhou: {e.code}")
            # Tratar erro de auth (notificar admin, etc.)
        elif e.status_code == 403:
            print(f"Permissão negada: {e.message}")
        elif e.status_code == 404:
            print("Recurso não encontrado")
        elif e.status_code == 429:
            print("Rate limit - por favor aguarde")
            # Implementar lógica de retry
        else:
            print(f"Erro da API: {e.message}")
        return None

# Uso
content = safe_api_call(client.list_content, status='active')
if content:
    print(f"Encontrados {len(content)} itens")