Build a Twitter Brand Monitoring Tool with Python

A useful Twitter brand monitor needs more than a search request. It must remember what it has seen, avoid flooding you with old posts on the first run, retry failed notifications, and run on a predictable request budget.

This tutorial builds that complete loop in Python. The finished monitor searches one or more queries, stores results in SQLite, queues one alert record for each newly discovered post, and optionally delivers alerts to Telegram.

Last verified: August 3, 2026.

Disclaimer: TweetAPI is a third-party service. Not affiliated with X Corp.

What the Monitor Does

Each scheduled run:

1. calls the Twitter search endpoint with type="Latest";
2. stores returned post IDs in a local SQLite database;
3. treats the first successful result set for each query as a silent baseline;
4. queues posts first discovered on later runs;
5. prints alerts or sends them through Telegram; and
6. marks an alert sent only after delivery succeeds.

The post ID is the global deduplication key. If overlapping queries find the same post, the monitor queues one alert record rather than one record per query.

At the start of a run, the script snapshots which queries already have a baseline. It processes those initialized queries first, preserving command-line order within the initialized and new groups. This prevents a newly added overlapping query from storing a post as a silent baseline before an established query can queue it. The single-row schema still records only the first matching query; use the join-table extension later in the article when every match must be retained.

Design a Search Query Before Writing Code

Start with the brand name, its account, and distinctive product terms. For a fictional Acme Cloud account, a useful first query could be:

("Acme Cloud" OR @acmecloud) -from:acmecloud

The exact phrase reduces unrelated matches, OR includes direct mentions, and -from: removes the brand's own posts. Add ambiguous terms one at a time and inspect the noise they introduce.

Useful query groups include:

• the brand name and account handle;
• product names that are distinctive without the brand;
• common misspellings;
• a campaign name or hashtag; and
• a high-intent problem phrase near the brand name.

Do not build one giant query before validating smaller ones. Separate queries are easier to tune and give you clearer request-budget math.

Prerequisites

Install the TweetAPI Python SDK:

python -m pip install tweetapi

Set the API key in the process environment:

export TWEETAPI_KEY="your_api_key"

Telegram is optional. Without Telegram variables, alerts print to standard output. To send messages, create a bot, obtain its chat ID, and set both values:

export TELEGRAM_BOT_TOKEN="your_bot_token"
export TELEGRAM_CHAT_ID="your_chat_id"

Telegram documents the HTTPS request format and required chat_id and text fields in its official Bot API sendMessage reference. Keep the bot token in a secret store just like the TweetAPI key.

Complete Python Brand Monitor

Save this as brand_monitor.py:

import argparse
import json
import os
import sqlite3
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from tweetapi import (
    AuthenticationError,
    NetworkError,
    RateLimitError,
    TweetAPI,
    TweetAPIError,
)


SCHEMA = """
CREATE TABLE IF NOT EXISTS seen_posts (
    post_id TEXT PRIMARY KEY,
    matched_query TEXT NOT NULL,
    author_username TEXT NOT NULL,
    post_text TEXT NOT NULL,
    post_created_at TEXT,
    discovered_at TEXT NOT NULL,
    alert_status TEXT NOT NULL CHECK (
        alert_status IN ('baseline', 'pending', 'sent')
    ),
    alerted_at TEXT
);

CREATE TABLE IF NOT EXISTS query_state (
    query TEXT PRIMARY KEY,
    initialized_at TEXT NOT NULL,
    last_checked_at TEXT NOT NULL
);
"""


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def open_database(path: Path) -> sqlite3.Connection:
    path.parent.mkdir(parents=True, exist_ok=True)
    connection = sqlite3.connect(path)
    connection.row_factory = sqlite3.Row
    connection.executescript(SCHEMA)
    return connection


def query_is_initialized(connection: sqlite3.Connection, query: str) -> bool:
    row = connection.execute(
        "SELECT 1 FROM query_state WHERE query = ?",
        (query,),
    ).fetchone()
    return row is not None


def remember_results(
    connection: sqlite3.Connection,
    query: str,
    posts: list[dict[str, Any]],
) -> tuple[int, bool]:
    initialized = query_is_initialized(connection, query)
    discovered_at = utc_now()
    new_pending = 0

    with connection:
        for post in posts:
            post_id = str(post.get("id", ""))
            if not post_id:
                continue

            author = post.get("author") or {}
            status = "pending" if initialized else "baseline"
            cursor = connection.execute(
                """
                INSERT OR IGNORE INTO seen_posts (
                    post_id,
                    matched_query,
                    author_username,
                    post_text,
                    post_created_at,
                    discovered_at,
                    alert_status
                ) VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    post_id,
                    query,
                    str(author.get("username", "unknown")),
                    str(post.get("text", "")),
                    post.get("createdAt"),
                    discovered_at,
                    status,
                ),
            )
            if cursor.rowcount == 1 and status == "pending":
                new_pending += 1

        connection.execute(
            """
            INSERT INTO query_state (query, initialized_at, last_checked_at)
            VALUES (?, ?, ?)
            ON CONFLICT(query) DO UPDATE SET last_checked_at = excluded.last_checked_at
            """,
            (query, discovered_at, discovered_at),
        )

    return new_pending, not initialized


def format_alert(row: sqlite3.Row) -> str:
    username = row["author_username"]
    text = row["post_text"].replace("\x00", "").strip()
    if len(text) > 700:
        text = text[:697] + "..."

    post_url = f"https://x.com/{username}/status/{row['post_id']}"
    return (
        "New Twitter brand mention\n"
        f"Query: {row['matched_query']}\n"
        f"Author: @{username}\n"
        f"Post: {text}\n"
        f"Link: {post_url}"
    )


def send_telegram_message(token: str, chat_id: str, text: str) -> None:
    request = urllib.request.Request(
        f"https://api.telegram.org/bot{token}/sendMessage",
        data=json.dumps({"chat_id": chat_id, "text": text}).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(request, timeout=15) as response:
        result = json.loads(response.read().decode("utf-8"))
    if not result.get("ok"):
        raise RuntimeError(result.get("description", "Telegram rejected the alert"))


def deliver_pending_alerts(
    connection: sqlite3.Connection,
    alert_limit: int,
) -> int:
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    chat_id = os.environ.get("TELEGRAM_CHAT_ID")
    if bool(token) != bool(chat_id):
        raise ValueError(
            "Set both TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID, or neither."
        )

    pending = connection.execute(
        """
        SELECT * FROM seen_posts
        WHERE alert_status = 'pending'
        ORDER BY discovered_at ASC
        LIMIT ?
        """,
        (alert_limit,),
    ).fetchall()

    delivered = 0
    for row in pending:
        message = format_alert(row)
        if token and chat_id:
            send_telegram_message(token, chat_id, message)
        else:
            print("\n" + message + "\n")

        with connection:
            connection.execute(
                """
                UPDATE seen_posts
                SET alert_status = 'sent', alerted_at = ?
                WHERE post_id = ?
                """,
                (utc_now(), row["post_id"]),
            )
        delivered += 1

    return delivered


def run_monitor(
    queries: list[str],
    database_path: Path,
    alert_limit: int,
) -> None:
    client = TweetAPI(api_key=os.environ["TWEETAPI_KEY"])
    connection = open_database(database_path)

    try:
        query_states = [
            (query, query_is_initialized(connection, query))
            for query in queries
        ]
        ordered_queries = [
            query for query, initialized in query_states if initialized
        ] + [
            query for query, initialized in query_states if not initialized
        ]

        for query in ordered_queries:
            response = client.explore.search(query=query, type="Latest")
            posts = response.get("data", [])
            pending_count, was_baseline = remember_results(
                connection,
                query,
                posts,
            )

            if was_baseline:
                print(f"Baseline stored for {query!r}: {len(posts)} results")
            else:
                print(f"Checked {query!r}: {pending_count} new posts queued")

        delivered = deliver_pending_alerts(connection, alert_limit)
        print(f"Delivered {delivered} alerts")
    finally:
        connection.close()


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Search Twitter brand queries and queue new-post alerts."
    )
    parser.add_argument(
        "--query",
        action="append",
        required=True,
        help="Repeat --query to monitor multiple searches",
    )
    parser.add_argument("--database", default="brand-monitor.db")
    parser.add_argument("--alert-limit", type=int, default=20)
    args = parser.parse_args()

    if "TWEETAPI_KEY" not in os.environ:
        parser.error("set TWEETAPI_KEY before running the monitor")
    if args.alert_limit < 1:
        parser.error("--alert-limit must be at least 1")

    queries = [query.strip() for query in args.query if query.strip()]
    if not queries:
        parser.error("at least one non-empty --query is required")

    try:
        run_monitor(
            queries=queries,
            database_path=Path(args.database),
            alert_limit=args.alert_limit,
        )
    except AuthenticationError:
        raise SystemExit("Authentication failed. Check TWEETAPI_KEY.")
    except RateLimitError as error:
        retry_after = getattr(error, "retry_after", None)
        detail = f" Retry after about {retry_after} seconds." if retry_after else ""
        raise SystemExit("Rate limit reached." + detail)
    except NetworkError:
        raise SystemExit("Network request failed after SDK retries.")
    except (OSError, RuntimeError, ValueError, sqlite3.Error) as error:
        raise SystemExit(f"Monitor delivery or storage error: {error}")
    except TweetAPIError as error:
        raise SystemExit(
            f"TweetAPI error {error.status_code}: {error.message}"
        )


if __name__ == "__main__":
    main()

Run the Baseline

Run the script once with your real query:

python brand_monitor.py \
  --query '("Acme Cloud" OR @acmecloud) -from:acmecloud'

The first successful run stores current results as baseline. It intentionally sends no alerts for those posts. This prevents an initial flood that looks like new activity but predates the monitor.

Run it again later with the same database. Posts with IDs not already stored become pending, then sent after the console or Telegram delivery succeeds.

Monitor multiple independent searches by repeating the option:

python brand_monitor.py \
  --query '("Acme Cloud" OR @acmecloud) -from:acmecloud' \
  --query '"Acme Deploy" -from:acmecloud' \
  --database ./data/brand-monitor.db

Why Failed Alerts Stay Pending

The script writes a new post as pending before sending the notification. It changes that status to sent only after Telegram returns a successful response or the message prints successfully.

If Telegram is temporarily unavailable, the program exits and leaves the row pending. The next scheduled run tries that alert again. This is a small durable outbox: it is safer than marking a notification sent before delivery, and safer than keeping unsent alerts only in memory.

The delivery guarantee is at least once, not exactly once. If Telegram accepts a message but the process stops before SQLite commits the sent update, the row remains pending and a later run can deliver a duplicate. Removing that narrow ambiguous-delivery window requires an idempotency mechanism shared with the destination; the Telegram sendMessage call used here does not provide one.

The default --alert-limit 20 also prevents one run from sending an unbounded burst. A backlog drains across later runs.

Schedule the Monitor

This script performs one check and exits, which makes it suitable for cron, a systemd timer, GitHub Actions, a serverless scheduler, or a worker platform.

For cron on a server where the required environment variables are already available to the job:

*/5 * * * * cd /opt/brand-monitor && .venv/bin/python brand_monitor.py --query '("Acme Cloud" OR @acmecloud) -from:acmecloud' >> monitor.log 2>&1

Cron usually does not inherit variables from your interactive shell. Inject secrets through the host's secret manager or a root-readable environment file, and test the exact scheduled command before relying on it. Restrict access to the SQLite database and logs because both can contain post text.

Do not run two copies against the same SQLite file at the same time. For multiple workers, replace SQLite with a server database and claim pending alerts transactionally.

Calculate the Monthly Request Budget

Each query makes one search request per scheduled run before any SDK retry attempts:

monthly search requests = queries × runs per hour × 24 × 30

At a five-minute interval:

1 query  × 12 × 24 × 30 = 8,640 requests/month
5 queries × 12 × 24 × 30 = 43,200 requests/month
10 queries × 12 × 24 × 30 = 86,400 requests/month

Leave capacity for retries and every other API workflow on the same account. Compare the resulting total with the request quota and per-minute limit on the current pricing page, not with an assumed cost per returned post.

If the interval is one minute, one query alone produces 43,200 scheduled search requests in a 30-day month. Faster polling is not automatically better; tune the interval to the query volume and the delay your team can tolerate.

Monitoring Limits You Should State Clearly

This starter checks the latest result page returned when the scheduler runs. It is not a firehose, a streaming connection, or a guarantee that every matching post will be collected.

A high-volume query can produce more new posts between checks than fit in that page. For those workloads, add cursor pagination with a bounded page limit, shorten the interval only after recalculating quota, or choose an access product designed for the required completeness and latency.

Search relevance can also change. Exact phrases may miss spelling variants, broad terms create noise, and deleted or unavailable posts may disappear before a scheduled check. Review false positives and false negatives using saved examples rather than assuming one query is permanent.

This monitor does not perform sentiment analysis. Route and deduplicate reliable mentions first; add a classifier only when you have labeled examples and a way to measure its errors.

Production Extensions

Once the basic loop is reliable, useful additions include:

• store which query terms matched each post in a separate join table;
• add allowlists and blocklists for known noisy accounts or phrases;
• send daily summaries in addition to urgent alerts;
• route high-value queries to separate chats;
• record request duration, result count, alert latency, and failures;
• add a review state for analyst triage; and
• implement retention and deletion handling appropriate to the application.

Keep collection, classification, and notification as separate stages. That makes it possible to change a classifier or alert destination without recollecting the same posts.

FAQ

Why not alert on the first run?

The first result set existed before the monitor established its state. Treating it as a baseline avoids presenting old posts as newly discovered activity.

Will overlapping queries send duplicate alerts?

seen_posts.post_id is the primary key, so overlapping queries queue one alert record for a post. Initialized queries run before newly added query baselines, which prevents a new query from suppressing an established query's alert. A delivery can still be repeated in the narrow at-least-once failure window described above.

Can I use Slack or email instead of Telegram?

Yes. Replace send_telegram_message with a delivery function for the destination, but preserve the pending to sent transition so failed deliveries can be retried.

Does the monitor require a Twitter developer account?

It uses a TweetAPI key rather than official X API credentials. The Twitter data access methods guide explains how official access, third-party APIs, and manual exports differ.

Establish the Baseline Today

Create a TweetAPI account, test one narrow query, and inspect its silent baseline. The 100-request trial is enough to validate the code, result quality, deduplication, and alert delivery before choosing a longer schedule.

For a one-time dataset rather than continuous monitoring, use the resumable Python follower-to-CSV exporter. For client configuration, pagination helpers, retry behavior, and TypeScript equivalents, read the complete SDK guide.