Powerstream, streaming API de baixa latência na X API v2

Powerstream é nossa streaming API de menor latência para acessar dados públicos do X em tempo real. Ao contrário de outros endpoints de streaming que priorizam a hidratação e a entrega de dados (com latência P99 de ~6-7 segundos), o Powerstream é otimizado para velocidade e entrega dados com atraso mínimo. Semelhante à legada GNIP Powertrack API, ele usa regras para filtrar Posts com base em palavras-chave, operadores e metadados. Após uma conexão HTTP persistente ser estabelecida com o endpoint Powerstream, você pode começar a receber Posts correspondentes em tempo real. Atualmente, o Powerstream suporta até 1.000 regras e cada regra pode ter 2048 caracteres.

Recursos principais

Entrega de dados em tempo real: Opção de menor latência para transmitir Posts assim que são publicados.
Filtragem precisa: Filtre exatamente os dados que você procura usando consultas booleanas com operadores.
Entrega: Resposta JSON via HTTP/1.1 chunked transfer encoding.
Suporte a datacenter local: Busque Posts apenas do datacenter local para reduzir ainda mais a latência, evitando o atraso de replicação.

A Powerstream API é uma oferta premium disponível em planos Enterprise selecionados.Se você estiver interessado em acessar o Powerstream ou saber mais sobre nossas ofertas Enterprise, entre em contato com nosso time de Vendas enviando o Enterprise Request Form. Teremos prazer em discutir como o Powerstream pode atender às suas necessidades.

Início rápido

Esta seção mostra como começar rapidamente com os endpoints do PowerStream usando Python com a biblioteca requests. Instale-a via pip install requests. Todos os exemplos usam autenticação OAuth 2.0 Bearer Token. Substitua YOUR_BEARER_TOKEN pelo seu token real (armazene-o com segurança, por exemplo, via os.getenv('BEARER_TOKEN')). Vamos cobrir cada endpoint com trechos de código. Assuma estas importações no topo:

import requests
import json
import time
import sys
import os  # For env vars

Configuração

bearer_token = os.getenv('BEARER_TOKEN') or "YOUR_BEARER_TOKEN"  # Use env var for security
base_url = "https://api.x.com/2/powerstream"
rules_url = f"{base_url}/rules"  # For rule management
headers = {
   "Authorization": f"Bearer {bearer_token}",
   "Content-Type": "application/json"
}

1. Criar regras (POST /rules)

Adicione regras para filtrar seu stream.

Exemplo

data = {
   "rules": [
       {
           "value": "(cat OR dog) lang:en -is:retweet",
           "tag": "pet-monitor"
       },
       # Add more rules as needed (up to 100)
   ]
}

response = requests.post(rules_url, headers=headers, json=data)
if response.status_code == 201:
   rules_added = response.json().get("data", {}).get("rules", [])
   print("Rules added:")
   for rule in rules_added:
       print(f"ID: {rule['id']}, Value: {rule['value']}, Tag: {rule.get('tag', 'N/A')}")
else:
   print(f"Error {response.status_code}: {response.text}")

2. Excluir regras (POST /rules)

Remova regras por ID (recomendado) ou por valor.

Exemplo

data = {
   "rules": [
       {
           "value": "(cat OR dog) lang:en -is:retweet",
           "tag": "pet-monitor"
       },
       # Add more rules as needed (up to 100)
   ]
}

response = requests.delete(rules_url, headers=headers, json=data)
if response.status_code == 200:
   deleted = response.json().get("data", {})
   print(f"Deleted count: {deleted.get('deleted', 'N/A')}")
   if 'not_deleted' in deleted:
       print("Not deleted:", deleted['not_deleted'])
else:
   print(f"Error {response.status_code}: {response.text}")

Dica: Para excluir todas as regras, primeiro faça um GET, extraia os IDs e, em seguida, exclua em lote.

3. Obter regras (GET /rules)

Busque todas as regras ativas.

response = requests.get(rules_url, headers=headers)
if response.status_code == 200:
   rules = response.json().get("data", {}).get("rules", [])
   if rules:
       print("Active rules:")
       for rule in rules:
           print(f"ID: {rule['id']}, Value: {rule['value']}, Tag: {rule.get('tag', 'N/A')}")
   else:
       print("No active rules.")
else:
   print(f"Error {response.status_code}: {response.text}")

4. PowerStream (GET /stream)

Conecte-se ao stream para receber Posts em tempo real e com baixa latência. Use stream=True para leitura linha a linha. Implemente lógica de reconexão para maior robustez.

Exemplo

stream_url = base_url

def main():
   while True:
       response = requests.request("GET", stream_url, headers=headers, stream=True)
       print(response.status_code)
       for response_line in response.iter_lines():
           if response_line:
               json_response = json.loads(response_line)
               print(json.dumps(json_response, indent=4, sort_keys=True))
               if response.status_code != 200:
                   print(response.headers)
                   raise Exception(
                       "Request returned an error: {} {}".format(
                           response.status_code, response.text
                       )
                   )

Suporte a datacenter local

Para otimização de latência, o Powerstream oferece uma opção para buscar apenas posts que se originaram ou foram criados no datacenter local onde uma conexão é estabelecida. Isso evita o atraso de replicação, resultando em entrega mais rápida em comparação com posts de outros datacenters. Para habilitar isso, adicione o parâmetro de consulta ?localDcOnly=true ao endpoint do stream (por exemplo, /2/powerstream?localDcOnly=true). O datacenter ao qual você está conectado será indicado tanto no payload de dados inicial do stream quanto como um header HTTP na resposta. Para usar no código:

# For local datacenter only:
stream_url = "https://api.x.com/2/powerstream?localDcOnly=true"

Se o parâmetro localDcOnly estiver habilitado, quando o stream se conectar pela primeira vez, ele incluirá os seguintes headers de resposta indicando qual datacenter local está sendo usado:

'x-powerstream-datacenter': 'atla',
'x-powerstream-localdconly': 'true'

Além disso, também enviará um payload inicial especificando o datacenter:

{
    "type": "connection_metadata",
    "datacenter": "atla",
    "timestamp": 1762557264155
}

Dica: Para otimizar a latência, configure conexões a partir de diferentes locais geográficos (por exemplo, uma próxima a Atlanta na costa leste dos EUA e outra próxima a Portland na costa oeste dos EUA), habilitando localDcOnly=true para cada uma. Isso fornece acesso mais rápido a posts de cada datacenter respectivo. Agregue os streams do seu lado para combinar dados entre datacenters.

Operadores

Para definir regras de filtragem, você pode usar palavras-chave e operadores.

Operadores do Powerstream

Lista completa dos operadores disponíveis

Respostas

O payload da Powerstream API tem o mesmo formato da legada GNIP Powertrack API. Um exemplo de resposta JSON é semelhante a:

Exemplo de resposta

[
   {
       "created_at": "Tue Mar 21 20:50:14 +0000 2006",
       "id": 20,
       "id_str": "20",
       "text": "just setting up my twttr",
       "truncated": false,
       "entities": {
           "hashtags": [],
           "symbols": [],
           "user_mentions": [],
           "urls": []
       },
       "source": "<a href=\"http://x.com\" rel=\"nofollow\">X Web Client</a>",
       "in_reply_to_status_id": null,
       "in_reply_to_status_id_str": null,
       "in_reply_to_user_id": null,
       "in_reply_to_user_id_str": null,
       "in_reply_to_screen_name": null,
       "user": {
           "id": 12,
           "id_str": "12",
           "name": "jack",
           "screen_name": "jack",
           "location": "",
           "description": "no state is the best state",
           "url": "https://t.co/ZEpOg6rn5L",
           "entities": {
               "url": {
                   "urls": [
                       {
                           "url": "https://t.co/ZEpOg6rn5L",
                           "expanded_url": "http://primal.net/jack",
                           "display_url": "primal.net/jack",
                           "indices": [
                               0,
                               23
                           ]
                       }
                   ]
               },
               "description": {
                   "urls": []
               }
           },
           "protected": false,
           "followers_count": 6427829,
           "friends_count": 3,
           "listed_count": 32968,
           "created_at": "Tue Mar 21 20:50:14 +0000 2006",
           "favourites_count": 36306,
           "utc_offset": null,
           "time_zone": null,
           "geo_enabled": true,
           "verified": false,
           "statuses_count": 30134,
           "lang": null,
           "contributors_enabled": false,
           "is_translator": false,
           "is_translation_enabled": false,
           "profile_background_color": "EBEBEB",
           "profile_background_image_url": "http://abs.twimg.com/images/themes/theme7/bg.gif",
           "profile_background_image_url_https": "https://abs.twimg.com/images/themes/theme7/bg.gif",
           "profile_background_tile": false,
           "profile_image_url": "http://pbs.twimg.com/profile_images/1661201415899951105/azNjKOSH_normal.jpg",
           "profile_image_url_https": "https://pbs.twimg.com/profile_images/1661201415899951105/azNjKOSH_normal.jpg",
           "profile_banner_url": "https://pbs.twimg.com/profile_banners/12/1742427520",
           "profile_link_color": "990000",
           "profile_sidebar_border_color": "DFDFDF",
           "profile_sidebar_fill_color": "F3F3F3",
           "profile_text_color": "333333",
           "profile_use_background_image": true,
           "has_extended_profile": true,
           "default_profile": false,
           "default_profile_image": false,
           "following": null,
           "follow_request_sent": null,
           "notifications": null,
           "translator_type": "regular",
           "withheld_in_countries": []
       },
       "geo": null,
       "coordinates": null,
       "place": null,
       "contributors": null,
       "is_quote_status": false,
       "retweet_count": 122086,
       "favorite_count": 263321,
       "favorited": false,
       "retweeted": false,
       "lang": "en"
   }
]

Limites e melhores práticas

Rate Limits: 50 requisições/24h para gerenciamento de regras; sem limite em streams (mas se aplicam limites de conexão).
Reconexão: Backoff exponencial em desconexões.
Monitoramento: Use os headers Connection: keep-alive.

Fundamentos de streaming

Consumindo dados de streaming

Melhores práticas para clientes de streaming

Lidando com desconexões

Reconecte-se de forma elegante

Capacidade de alto volume

Lide com alta vazão

Recuperação e redundância

Construa aplicações resilientes

Getting Started

Products

Fundamentals

Partners & Customers

Resources

Powerstream, streaming API de baixa latência na X API v2

Recursos principais

Início rápido

Configuração

1. Criar regras (POST /rules)

2. Excluir regras (POST /rules)

3. Obter regras (GET /rules)

4. PowerStream (GET /stream)

Suporte a datacenter local

Operadores

Operadores do Powerstream

Respostas

Limites e melhores práticas

Fundamentos de streaming

Consumindo dados de streaming

Lidando com desconexões

Capacidade de alto volume

Recuperação e redundância

​Recursos principais

​Início rápido

​Configuração

​1. Criar regras (POST /rules)

​2. Excluir regras (POST /rules)

​3. Obter regras (GET /rules)

​4. PowerStream (GET /stream)

​Suporte a datacenter local

​Operadores

Operadores do Powerstream

​Respostas

​Limites e melhores práticas

​Fundamentos de streaming

Consumindo dados de streaming

Lidando com desconexões

Capacidade de alto volume

Recuperação e redundância

Recursos principais

Início rápido

Configuração

1. Criar regras (POST /rules)

2. Excluir regras (POST /rules)

3. Obter regras (GET /rules)

4. PowerStream (GET /stream)

Suporte a datacenter local

Operadores

Respostas

Limites e melhores práticas

Fundamentos de streaming