> ## 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.

# Supervisión en tiempo real

Detecta nuevas publicaciones de cuentas concretas, sigue palabras clave y alimenta tu aplicación con datos actuales de X. Esta guía explica cómo crear un flujo de supervisión casi en tiempo real con Sorsa API mediante consultas periódicas.

La latencia depende del intervalo entre consultas, el tiempo de respuesta y cuándo aparece la publicación en el feed o índice elegido. Diseña el proceso con puntos de control, paginación y deduplicación: no supongas que cada publicación nueva estará en la siguiente respuesta.

> **Prototipos gratuitos:** las primeras 100 solicitudes permiten probar todos los endpoints, sin tarjeta ni caducidad. Puedes validar la detección de publicaciones y su envío a Slack o Discord antes de elegir un plan. Las consultas periódicas consumen muchas solicitudes; dimensiona el plan con la tabla de esta guía.

> **Nota:** consulta otras arquitecturas y ejemplos completos en la [guía de supervisión con REST API](https://api.sorsa.io/blog/real-time-twitter-monitoring).

***

## Cómo funciona la supervisión mediante consultas periódicas

Los datos pueden llegar por envío del servidor (streaming o webhooks) o recuperarse con consultas periódicas. Sorsa utiliza este segundo método:

1. **Consulta un endpoint** a intervalos regulares de 1 a 30 segundos.
2. **Compara los resultados** con los IDs ya vistos.
3. **Procesa las publicaciones nuevas:** guarda datos o envía alertas a Slack, Discord u otros destinos.
4. **Repite.**

Los IDs codifican la fecha de creación y pueden compararse con enteros de Python o BigInt de JavaScript. El mayor ID visto sirve como punto de control, pero los resultados pueden llegar tarde o desordenados. En producción, vuelve a consultar un intervalo solapado y elimina duplicados por ID. Para recuperarte de una caída, debes guardar y cargar explícitamente el punto de control.

### Elegir el endpoint

| Qué supervisar          | Endpoint         | Método | Motivo                                                         |
| :---------------------- | :--------------- | :----- | :------------------------------------------------------------- |
| Una cuenta              | `/user-tweets`   | POST   | Publicaciones recientes de su cronología                       |
| Hasta 5.000 cuentas     | `/list-tweets`   | GET    | Una solicitud cubre los miembros de una lista de X             |
| Palabra clave o hashtag | `/search-tweets` | POST   | Operadores de búsqueda y orden cronológico con `order: latest` |
| Menciones de una cuenta | `/mentions`      | POST   | Seguimiento de menciones con filtros de interacción            |

***

## Nivel 1: Supervisar una cuenta

El bucle consulta la primera página de `/user-tweets` y emite publicaciones posteriores al último ID visto. En la primera consulta correcta establece el punto inicial sin emitir las publicaciones existentes. Es un ejemplo de desarrollo: puede omitir resultados si llega más de una página entre consultas o mientras el proceso está detenido.

### 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));
}
```

Este método escala mal: supervisar 50 cuentas implica 50 bucles y 50 veces más solicitudes. Para eso sirven las listas de X.

***

## Nivel 2: Supervisar varias cuentas con una solicitud

Las listas agrupan hasta 5.000 cuentas. `/list-tweets` devuelve sus publicaciones recientes combinadas en una llamada. Es el patrón habitual para supervisar varias cuentas en producción. Consulta [Listas y comunidades](https://docs.sorsa.io/es/lists-and-communities).

### Paso 1: Crea una lista pública de X

1. Abre [Listas de X](https://x.com/i/lists) y crea una lista.
2. Añade las cuentas que quieras supervisar, hasta 5.000.
3. Establécela como **pública**: la API no accede a listas privadas.
4. Copia el **ID** de la URL. En `https://x.com/i/lists/1234567890`, es `1234567890`.

### Paso 2: Consulta la 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)
```

**Ahorro de solicitudes.** Consultar 50 cuentas por separado cada 10 segundos requiere 50 × 8.640 = 432.000 solicitudes al día. Consultar una lista con las mismas cuentas requiere 8.640: 50 veces menos. Consulta [Optimización del uso de la API](https://docs.sorsa.io/es/optimizing-api-usage).

> `/list-tweets` devuelve hasta 20 publicaciones por página. Si se publican más durante un intervalo, redúcelo a 2–3 segundos o recorre `next_cursor` hasta alcanzar un ID ya visto.

***

## Nivel 3: Supervisar una palabra clave o hashtag

Consulta `/search-tweets` con `order: "latest"` para obtener coincidencias cronológicas.

```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)
```

Puedes utilizar los [operadores de búsqueda](https://docs.sorsa.io/es/search-operators). Por ejemplo, para seguir menciones en inglés con alta interacción y excluir retuits:

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

***

## Enviar publicaciones a Slack, Discord o cualquier endpoint HTTP

El bucle produce los datos; la función de callback decide qué hacer con cada publicación. Puede enviarla a cualquier servicio HTTP.

### Slack mediante 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 propio

```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,
    )
```

***

## Calcular el consumo de API

La tabla supone una página por ciclo, una frecuencia fija y ningún reintento. Multiplica por el número de procesos de supervisión y añade páginas y reintentos adicionales. Los ejemplos esperan después de recibir la respuesta, por lo que cada ciclo también incluye tiempo de red y procesamiento.

| Intervalo   | Solicitudes/hora | Solicitudes/día | Solicitudes/mes (30 días) |
| :---------- | :--------------- | :-------------- | :------------------------ |
| 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                    |

Elige el intervalo según la latencia tolerable y la actividad del feed. Un intervalo menor aumenta las solicitudes, pero no garantiza que una publicación aparezca inmediatamente en la búsqueda.

Las 100 solicitudes gratuitas permiten validar un prototipo completo. Para uso continuo, un bucle cada 30–60 segundos cabe en Pro (100.000 al mes) y uno cada 10 segundos en Enterprise (500.000). Son cifras por proceso; varios en paralelo multiplican el consumo. Dimensiona el plan según el volumen combinado y consulta los [precios](https://api.sorsa.io/pricing).

> Para límites o volúmenes superiores, [contacta con ventas](https://api.sorsa.io/talk-to-sales) o pregunta en [Discord](https://discord.com/invite/uwAefKCj7X).

***

## Preparación para producción

Los ejemplos anteriores sirven para desarrollo. En producción, resuelve estos cinco puntos.

### 1. Guarda `last_seen_id` entre reinicios

Sin un punto de control persistente, una caída puede provocar alertas duplicadas u omitir el intervalo pendiente. Guarda el último ID en un archivo, base de datos o 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)
```

Sustituye `last_seen_id = None` por `last_seen_id = load_state()`. Guarda el nuevo punto de control después de procesar o encolar de forma persistente todas las páginas del ciclo. No lo avances si falla una página o entrega. Al reiniciar, recupera el intervalo pendiente antes de aceptar un punto más reciente.

### 2. Esperas exponenciales ante errores

Habrá fallos de red, límites HTTP 429 y errores transitorios. Aumenta gradualmente la espera hasta un máximo, en lugar de reintentar inmediatamente. Consulta [Códigos de error](https://docs.sorsa.io/es/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. Separa las consultas del procesamiento

No ejecutes operaciones costosas, como NLP, escrituras de base de datos o llamadas externas, de forma síncrona en el bucle. Un destino lento retrasaría las consultas. Encola las publicaciones y procésalas en otro proceso.

```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 mayores, sustituye la cola en memoria por Redis, RabbitMQ, SQS u otro intermediario de mensajes de tu infraestructura.

### 4. Supervisa el propio proceso

Registra cada ciclo: hora, nuevas publicaciones, latencia y errores. Genera una alerta si no se completa una consulta correcta durante N minutos: los fallos silenciosos provocan lagunas de datos. Consulta el estado del servicio en [Sorsa Status](https://uptime.sorsa.io/status/v3).

### 5. Gestiona los casos especiales

**Más de una página y resultados tardíos:** recorre `next_cursor` hasta cubrir el intervalo desde el último punto de control. Mantén solapamiento y deduplicación para no descartar resultados tardíos por tener IDs más antiguos. Los ejemplos de primera página no implementan esta recuperación.

**Entrega mediante callback:** comprueba el estado de respuesta del webhook y utiliza reintentos limitados o una cola persistente. Una solicitud correcta a Sorsa no significa que Slack, Discord o tu base de datos hayan aceptado el evento.

* **Publicaciones eliminadas:** si se eliminan entre la consulta y el callback, su URL devolverá 404. Es un caso esperado.
* **Cuentas protegidas:** si una cuenta pasa a privada, `/user-tweets` devuelve una lista vacía. Regístralo y continúa.
* **Publicaciones fijadas:** el primer resultado puede ser el fijado, no el más reciente. No uses `tweets[0]` como mayor ID; calcula `max(int(t["id"]) for t in tweets)`, como en los ejemplos, u ordena por `created_at`.
* **Retuits:** `tweet["retweeted_status"]` contiene datos cuando es un retuit. Decide si incluirlos.
* **Respuestas restringidas:** `is_replies_limited` indica restricciones del autor y puede ser una señal útil.

***

## Próximos pasos

* [Operadores de búsqueda](https://docs.sorsa.io/es/search-operators): reduce el ruido.
* [Seguimiento de menciones](https://docs.sorsa.io/es/search-mentions): menciones con filtros de interacción.
* [Límites de solicitudes](https://docs.sorsa.io/es/rate-limits): gestión de 429.
* [Paginación](https://docs.sorsa.io/es/pagination): combina recuperación histórica y supervisión.
* [Referencia de la API](https://docs.sorsa.io/es/api-reference-guide): especificaciones de los endpoints.