> ## Documentation Index
> Fetch the complete documentation index at: https://docs.sorsa.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Monitoramento em tempo real

Detecte novas publicações de contas, monitore palavras-chave e alimente sua aplicação com dados do X. Este guia usa consultas periódicas (polling) para montar um pipeline quase em tempo real.

A latência depende do intervalo de consulta, da resposta da API e de quando o conteúdo aparece no feed ou índice de busca. Use checkpoints, paginação e deduplicação; não suponha que todo post novo estará na próxima resposta.

> **Prototipe grátis:** as primeiras 100 requisições, sem cartão e sem validade, permitem validar os monitores e o encaminhamento a Slack ou Discord. Para uso contínuo, dimensione o plano pelo intervalo de polling.

> Veja outras arquiteturas e exemplos no [guia do blog](https://api.sorsa.io/blog/real-time-twitter-monitoring).

## Como funciona o polling

Ao contrário de streaming ou webhooks que enviam eventos, a Sorsa usa consultas periódicas:

1. **Consulte** um endpoint a cada 1–30 segundos.
2. **Compare** os IDs com os já vistos.
3. **Processe** as novidades: armazene, envie alertas ou encaminhe a outros serviços.
4. **Repita.**

IDs codificam o horário de criação e podem ser comparados com inteiros no Python ou BigInt no JavaScript. O maior ID é um checkpoint útil, mas resultados podem chegar atrasados ou fora de ordem. Em produção, consulte uma janela com sobreposição e deduplique. Persista e carregue o checkpoint explicitamente após reiniciar.

### Escolha do endpoint

| O que monitorar          | Endpoint         | Método | Motivo                                 |
| :----------------------- | :--------------- | :----- | :------------------------------------- |
| Uma conta                | `/user-tweets`   | POST   | Timeline recente de um usuário         |
| Até 5.000 contas         | `/list-tweets`   | GET    | Uma chamada para os membros da lista   |
| Palavra-chave ou hashtag | `/search-tweets` | POST   | Operadores completos e `order: latest` |
| @menções de uma conta    | `/mentions`      | POST   | Filtros próprios de engajamento        |

## Nível 1: monitorar uma conta

O loop consulta a primeira página de `/user-tweets` e emite IDs posteriores ao último visto. Na primeira consulta bem-sucedida, define a referência sem emitir posts antigos. É um exemplo de desenvolvimento: pode perder dados se chegar mais de uma página entre consultas ou durante uma parada.

### Python

```python theme={null}
import requests
import time

API_KEY = "YOUR_API_KEY"
USERNAME = "elonmusk"
POLL_INTERVAL = 5  # seconds

URL = "https://api.sorsa.io/v3/user-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}

last_seen_id = None

print(f"Monitoring @{USERNAME}...")

while True:
    try:
        resp = requests.post(URL, headers=HEADERS, json={"username": USERNAME}, timeout=30)
        resp.raise_for_status()
        tweets = resp.json().get("tweets", [])

        if tweets:
            # Snowflake IDs arrive as strings. Use the highest (newest) ID in the
            # batch; this stays correct even if a pinned tweet appears first.
            top_id = max(int(t["id"]) for t in tweets)

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                for tweet in reversed(new_tweets):  # oldest first
                    print(f"[NEW] @{USERNAME}: {tweet['full_text'][:140]}")
                if new_tweets:
                    last_seen_id = top_id

    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(POLL_INTERVAL * 2)
        continue

    time.sleep(POLL_INTERVAL)
```

### JavaScript

```javascript theme={null}
const API_KEY = "YOUR_API_KEY";
const USERNAME = "elonmusk";
const POLL_INTERVAL = 5000;

let lastSeenId = null;
console.log(`Monitoring @${USERNAME}...`);

while (true) {
  try {
    const resp = await fetch("https://api.sorsa.io/v3/user-tweets", {
      method: "POST",
      headers: { "ApiKey": API_KEY, "Content-Type": "application/json" },
      body: JSON.stringify({ username: USERNAME }),
    });
    if (!resp.ok) throw new Error(`HTTP ${resp.status}`);

    const tweets = (await resp.json()).tweets || [];

    if (tweets.length > 0) {
      // BigInt avoids precision loss on 64-bit Snowflake IDs.
      // Use the highest ID in the batch (robust if a pinned tweet appears first).
      let topId = 0n;
      for (const t of tweets) {
        const id = BigInt(t.id);
        if (id > topId) topId = id;
      }

      if (lastSeenId === null) {
        lastSeenId = topId;
        console.log(`Baseline set: ${lastSeenId}`);
      } else {
        const newTweets = tweets.filter((t) => BigInt(t.id) > lastSeenId);
        for (const t of [...newTweets].reverse()) {
          console.log(`[NEW] @${USERNAME}: ${t.full_text.slice(0, 140)}`);
        }
        if (newTweets.length) lastSeenId = topId;
      }
    }
  } catch (err) {
    console.error(`Error: ${err.message}`);
    await new Promise((r) => setTimeout(r, POLL_INTERVAL * 2));
    continue;
  }
  await new Promise((r) => setTimeout(r, POLL_INTERVAL));
}
```

Para 50 contas, seriam 50 loops e 50 vezes mais chamadas. Use listas para reduzir esse custo.

## Nível 2: várias contas em uma chamada

Listas do X reúnem até 5.000 contas. `/list-tweets` retorna o feed recente combinado. É o padrão recomendado para várias contas; veja [listas e comunidades](https://docs.sorsa.io/pt-BR/lists-and-communities).

### Etapa 1: crie uma lista pública

1. Acesse [X Lists](https://x.com/i/lists) e crie uma lista.
2. Adicione até 5.000 contas.
3. Defina como **Public**. Listas privadas não são acessíveis.
4. Copie o ID da URL: em `https://x.com/i/lists/1234567890`, o ID é `1234567890`.

### Etapa 2: consulte a lista

```python theme={null}
import requests
import time

API_KEY = "YOUR_API_KEY"
LIST_ID = "YOUR_LIST_ID"
POLL_INTERVAL = 5

URL = f"https://api.sorsa.io/v3/list-tweets?list_id={LIST_ID}"
HEADERS = {"ApiKey": API_KEY, "Accept": "application/json"}


def monitor_list(callback, interval=POLL_INTERVAL):
    """Poll an X List and call `callback` for each new tweet detected."""
    last_seen_id = None
    print(f"Monitoring List {LIST_ID} (interval: {interval}s)")

    while True:
        try:
            resp = requests.get(URL, headers=HEADERS, timeout=10)
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if not tweets:
                time.sleep(interval)
                continue

            top_id = max(int(t["id"]) for t in tweets)

            if last_seen_id is None:
                last_seen_id = top_id
                print(f"Baseline set: {last_seen_id}")
            else:
                new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                if new_tweets:
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Request error: {e}. Retrying in {interval * 2}s")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


def on_new_tweet(tweet):
    user = tweet["user"]
    print(f"[NEW] @{user['username']}: {tweet['full_text'][:120]}")
    print(
        f"       Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
    )


if __name__ == "__main__":
    monitor_list(on_new_tweet)
```

Consultar 50 contas a cada 10 segundos custa 50 × 8.640 = 432.000 chamadas/dia. Uma lista com as mesmas contas custa 8.640, redução de 50 vezes. Veja [otimização](https://docs.sorsa.io/pt-BR/optimizing-api-usage).

> Cada página traz até 20 publicações. Se o volume exceder isso por ciclo, reduza o intervalo para 2–3 segundos ou percorra `next_cursor` até cobrir os dados já vistos.

## Nível 3: palavra-chave ou hashtag

Consulte `/search-tweets` com `order: "latest"`.

```python theme={null}
import requests
import time

API_KEY = "YOUR_API_KEY"
QUERY = '("your brand" OR @yourbrand) lang:en'
POLL_INTERVAL = 10

URL = "https://api.sorsa.io/v3/search-tweets"
HEADERS = {"ApiKey": API_KEY, "Content-Type": "application/json"}


def monitor_keyword(query, callback, interval=10):
    last_seen_id = None
    print(f"Monitoring: {query} (interval: {interval}s)")

    while True:
        try:
            resp = requests.post(
                URL,
                headers=HEADERS,
                json={"query": query, "order": "latest"},
                timeout=10,
            )
            resp.raise_for_status()
            tweets = resp.json().get("tweets", [])

            if tweets:
                top_id = max(int(t["id"]) for t in tweets)
                if last_seen_id is None:
                    last_seen_id = top_id
                    print(f"Baseline set: {last_seen_id}")
                else:
                    new_tweets = [t for t in tweets if int(t["id"]) > last_seen_id]
                    for tweet in reversed(new_tweets):
                        callback(tweet)
                    if new_tweets:
                        last_seen_id = top_id

        except requests.exceptions.RequestException as e:
            print(f"Error: {e}")
            time.sleep(interval * 2)
            continue

        time.sleep(interval)


monitor_keyword(QUERY, on_new_tweet, interval=10)
```

Use os [operadores de busca](https://docs.sorsa.io/pt-BR/search-operators). Este exemplo busca menções em inglês com engajamento e exclui repostagens:

```python theme={null}
monitor_keyword('"your brand" min_faves:10 lang:en -filter:retweets', on_new_tweet)
```

## Encaminhar publicações a outros serviços

O loop produz eventos; o callback decide o destino. Ele pode chamar qualquer serviço HTTP.

### Slack com Incoming Webhook

```python theme={null}
import requests

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"


def send_to_slack(tweet):
    user = tweet["user"]
    text = (
        f"*New tweet from @{user['username']}*\n"
        f"{tweet['full_text']}\n"
        f"Likes: {tweet.get('likes_count', 0)} | "
        f"RTs: {tweet.get('retweet_count', 0)} | "
        f"Views: {tweet.get('view_count', 'N/A')}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(SLACK_WEBHOOK_URL, json={"text": text})


# Plug into any monitor:
monitor_list(send_to_slack)
# or: monitor_keyword("bitcoin lang:en min_faves:50", send_to_slack)
```

### Discord

```python theme={null}
DISCORD_WEBHOOK_URL = "https://discord.com/api/webhooks/YOUR/WEBHOOK"


def send_to_discord(tweet):
    user = tweet["user"]
    content = (
        f"**@{user['username']}** just tweeted:\n"
        f"{tweet['full_text']}\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(DISCORD_WEBHOOK_URL, json={"content": content})
```

### Telegram

```python theme={null}
TELEGRAM_BOT_TOKEN = "YOUR_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_CHAT_ID"


def send_to_telegram(tweet):
    user = tweet["user"]
    text = (
        f"New tweet from @{user['username']}\n\n"
        f"{tweet['full_text']}\n\n"
        f"https://x.com/{user['username']}/status/{tweet['id']}"
    )
    requests.post(
        f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage",
        json={"chat_id": TELEGRAM_CHAT_ID, "text": text},
    )
```

### Endpoint HTTP próprio

```python theme={null}
def send_to_internal_api(tweet):
    requests.post(
        "https://internal.example.com/events/twitter",
        json={
            "tweet_id": tweet["id"],
            "username": tweet["user"]["username"],
            "text": tweet["full_text"],
            "metrics": {
                "likes": tweet.get("likes_count", 0),
                "retweets": tweet.get("retweet_count", 0),
                "views": tweet.get("view_count", 0),
            },
            "url": f"https://x.com/{tweet['user']['username']}/status/{tweet['id']}",
        },
        headers={"Authorization": "Bearer YOUR_INTERNAL_TOKEN"},
        timeout=5,
    )
```

## Estimativa de consumo

A tabela considera uma página por ciclo, horário fixo e nenhuma nova tentativa. Multiplique pela quantidade de monitores e acrescente páginas extras e repetições. Os exemplos esperam após cada resposta, então seus ciclos também incluem rede e processamento.

| Intervalo   | Req/hora | Req/dia | Req/mês (30 dias) |
| :---------- | :------- | :------ | :---------------- |
| 1 segundo   | 3.600    | 86.400  | 2.592.000         |
| 5 segundos  | 720      | 17.280  | 518.400           |
| 10 segundos | 360      | 8.640   | 259.200           |
| 30 segundos | 120      | 2.880   | 86.400            |
| 1 minuto    | 60       | 1.440   | 43.200            |

Escolha pela latência tolerada e atividade do feed. Intervalos menores aumentam chamadas, mas não garantem disponibilidade imediata na busca.

As 100 chamadas gratuitas servem para prototipar. Um monitor a cada 30–60 segundos cabe no Pro (100.000/mês); a cada 10 segundos, no Enterprise (500.000/mês). Vários monitores multiplicam o consumo. Veja [preços](https://api.sorsa.io/pricing).

> Para cotas ou frequência acima dos planos padrão, [fale com vendas](https://api.sorsa.io/talk-to-sales) ou use o [Discord](https://discord.com/invite/uwAefKCj7X).

## Preparação para produção

### 1. Persista last\_seen\_id

Sem checkpoint, reiniciar pode repetir alertas ou pular o intervalo de parada. Use arquivo, banco ou Redis.

```python theme={null}
import json
import os

STATE_FILE = "monitor_state.json"


def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE) as f:
            return json.load(f).get("last_seen_id")
    return None


def save_state(last_seen_id):
    with open(STATE_FILE, "w") as f:
        json.dump({"last_seen_id": last_seen_id}, f)
```

Substitua `last_seen_id = None` por `last_seen_id = load_state()`. Salve o novo checkpoint só depois de processar ou enfileirar de forma durável todas as páginas. Não avance após falha de página ou entrega. Ao reiniciar, percorra o intervalo pendente antes de aceitar um checkpoint mais recente.

### 2. Espera exponencial em erros

Para falhas de rede, 429 e erros transitórios, aumente a espera gradualmente, com um teto. Veja [códigos de erro](https://docs.sorsa.io/pt-BR/error-codes).

```python theme={null}
retry_delay = POLL_INTERVAL
MAX_DELAY = 60

while True:
    try:
        resp = requests.get(URL, headers=HEADERS, timeout=10)
        if resp.status_code == 429:
            print(f"Rate limited. Backing off {retry_delay}s")
            time.sleep(retry_delay)
            retry_delay = min(retry_delay * 2, MAX_DELAY)
            continue
        resp.raise_for_status()
        retry_delay = POLL_INTERVAL  # reset on success
        # process tweets
    except requests.exceptions.RequestException as e:
        print(f"Error: {e}")
        time.sleep(retry_delay)
        retry_delay = min(retry_delay * 2, MAX_DELAY)
        continue

    time.sleep(POLL_INTERVAL)
```

### 3. Separe consulta e processamento

NLP, gravações e chamadas externas demoradas não devem bloquear o loop. Enfileire as publicações para outro worker.

```python theme={null}
from collections import deque
import threading

tweet_queue = deque()


def polling_loop():
    """Fast loop: poll and enqueue. No heavy work here."""
    # Standard polling code, but instead of calling callback(tweet):
    # tweet_queue.append(tweet)
    pass


def processing_worker():
    """Separate thread: dequeue and dispatch."""
    while True:
        if tweet_queue:
            tweet = tweet_queue.popleft()
            send_to_slack(tweet)
            save_to_database(tweet)
        else:
            time.sleep(0.1)


threading.Thread(target=processing_worker, daemon=True).start()
polling_loop()
```

Para cargas maiores, substitua a deque em memória por Redis, RabbitMQ, SQS ou outro broker.

### 4. Monitore o próprio monitor

Registre horário, quantidade de novidades, tempo de resposta e erros. Gere alerta se não houver uma consulta bem-sucedida nos últimos N minutos. Consulte também o [status da Sorsa](https://uptime.sorsa.io/status/v3).

### 5. Trate casos especiais

**Páginas excedentes e atrasos:** percorra `next_cursor` até cobrir o período desde o checkpoint. Mantenha sobreposição e deduplique para não descartar resultados atrasados só porque têm IDs menores. Os exemplos de primeira página não implementam essa recuperação.

**Entrega do callback:** confira o status dos webhooks e use tentativas limitadas ou fila durável. Sucesso na Sorsa não significa aceitação pelo Slack, Discord ou banco.

* **Posts excluídos:** links podem retornar 404 entre consulta e processamento; trate como esperado.
* **Contas protegidas:** se uma conta se tornar privada, `/user-tweets` retorna lista vazia; registre e continue.
* **Posts fixados:** `tweets[0]` pode ser o fixado. Use `max(int(t["id"]) for t in tweets)` ou ordene por `created_at`.
* **Repostagens:** `tweet["retweeted_status"]` é preenchido. Decida se deve incluí-las.
* **Respostas restritas:** `is_replies_limited` indica restrição pelo autor.

## Próximos passos

* [Operadores](https://docs.sorsa.io/pt-BR/search-operators): reduza ruído.
* [Menções](https://docs.sorsa.io/pt-BR/search-mentions): filtros de engajamento.
* [Limites](https://docs.sorsa.io/pt-BR/rate-limits): trate 429.
* [Paginação](https://docs.sorsa.io/pt-BR/pagination): combine recuperação histórica e monitoramento.
* [Referência da API](https://docs.sorsa.io/pt-BR/api-reference-guide): especificação completa.