This Python script is a bot designed to monitor a...

July 3, 2025 at 05:24 PM

import asyncio
import getpass
import json
import os
import random
import re
from datetime import datetime, timedelta

import aiohttp
import discord
from tweety import TwitterAsync
from tweety.exceptions_ import UserNotFound
from tweety.filters import SearchFilters

MARKED_TWEETS_FOLDER = "marked_tweets"
SESSION_FILE = "session.tw_session"
CONFIGS_FILE = "configs.json"

app = TwitterAsync("session")


def load_configs():
    if not os.path.exists(CONFIGS_FILE):
        return {}
    with open(CONFIGS_FILE, "r", encoding="utf-8") as file:
        return json.load(file)


def save_configs(configs):
    with open(CONFIGS_FILE, "w", encoding="utf-8") as file:
        json.dump(configs, file, indent=4)


def get_user_config(username):
    configs = load_configs()
    if username not in configs:
        print(f"No configuration found for user: @{username}")
        discord_webhook_urls = input(
            "Enter Discord webhook URLs (comma-separated): "
        ).strip()
        discord_webhook_urls = [url.strip() for url in discord_webhook_urls.split(",")]
        configs[username] = {
            "discord_webhook_urls": discord_webhook_urls,
            "last_notification_webhook_index": 0,
        }
        save_configs(configs)
    return configs[username]


def save_marked_tweets(username, tweet_ids):
    if not os.path.exists(MARKED_TWEETS_FOLDER):
        os.makedirs(MARKED_TWEETS_FOLDER)

    file_path = os.path.join(MARKED_TWEETS_FOLDER, f"{username}.txt")
    existing_ids = load_marked_tweets(username)

    combined_ids = list(set(existing_ids + tweet_ids))
    combined_ids.sort()

    with open(file_path, "w", encoding="utf-8") as file:
        for tweet_id in combined_ids:
            file.write(f"{tweet_id}\n")


def load_marked_tweets(username):
    file_path = os.path.join(MARKED_TWEETS_FOLDER, f"{username}.txt")
    try:
        with open(file_path, "r", encoding="utf-8") as file:
            return file.read().splitlines()
    except FileNotFoundError:
        return []


def format_utc_to_jakarta_local(utc_datetime):
    day_names = {
        "Monday": "Senin",
        "Tuesday": "Selasa",
        "Wednesday": "Rabu",
        "Thursday": "Kamis",
        "Friday": "Jumat",
        "Saturday": "Sabtu",
        "Sunday": "Minggu",
    }

    jakarta_offset = timedelta(hours=7)
    jakarta_datetime = utc_datetime + jakarta_offset

    formatted_datetime = jakarta_datetime.strftime("%A, %d %B %Y %H:%M:%S")
    english_day_name = jakarta_datetime.strftime("%A")
    indonesian_day_name = day_names.get(english_day_name, english_day_name)

    formatted_datetime = formatted_datetime.replace(
        english_day_name, indonesian_day_name
    )

    return formatted_datetime


def get_tweet_type(tweet):
    if tweet.is_reply:
        first_word = tweet.text.split()[0] if tweet.text else ""
        if first_word.startswith("@"):
            mentioned_user = first_word[1:]
            return f"Replying to @{mentioned_user}"
        else:
            return "Posted"
    elif tweet.is_retweet:
        return "Retweet"
    elif tweet.is_quoted:
        return "Quote Retweet"
    else:
        return "Posted"


def process_tweet(tweet):
    if hasattr(tweet, "tweets"):
        return [(t.id, t.text, get_tweet_type(tweet)) for t in tweet.tweets][::-1]
    else:
        return [(tweet.id, tweet.text, get_tweet_type(tweet))]


async def extract_original_links(tweet_text, tweet_urls):
    # Replace t.co URLs with their original URLs
    for url in tweet_urls:
        tweet_text = tweet_text.replace(url.url, url.expanded_url)

    # Remove any remaining t.co URLs (if any)
    tweet_text = re.sub(r"https://t\.co/[a-zA-Z0-9]+", "", tweet_text).strip()

    return tweet_text


async def post_discord_notification(
    author_name,
    author_profile_image_url,
    author_username,
    created_on,
    media_urls,
    text,
    type,
    url,
    discord_webhook_urls,
    notification_webhook_index,
):
    async with aiohttp.ClientSession() as session:
        webhook = discord.Webhook.from_url(
            discord_webhook_urls[notification_webhook_index], session=session
        )

        embeds = []
        main_embed = discord.Embed(
            url=url,
            description=text,
            color=discord.Color.blue(),
        )
        main_embed.set_author(
            name=f"{author_name} ({author_username})",
            icon_url=author_profile_image_url,
        )
        main_embed.set_footer(text=created_on)
        embeds.append(main_embed)

        if media_urls:
            for i, media_url in enumerate(media_urls):
                if i == 0:
                    main_embed.set_image(url=media_url)
                else:
                    image_embed = discord.Embed(url=url)
                    image_embed.set_image(url=media_url)
                    embeds.append(image_embed)

        content = f"[{type}]({url})"

        await webhook.send(
            content=content,
            username=author_name,
            avatar_url=author_profile_image_url,
            embeds=embeds,
        )

    # Add a 2-second delay after each successful request to Discord
    await asyncio.sleep(2)

    # Update the last notification webhook index
    configs = load_configs()
    configs[author_username]["last_notification_webhook_index"] = (
        notification_webhook_index + 1
    ) % len(discord_webhook_urls)
    save_configs(configs)


async def validate_twitter_account(username):
    try:
        await app.get_user_info(username)
        return True
    except UserNotFound:
        return False


async def get_all_tweets_from_page(username, pages=1):
    keyword = f"(from:{username})"
    tweets = await app.search(
        keyword=keyword, filter_=SearchFilters.Latest(), pages=pages, wait_time=10
    )
    return tweets


async def monitor_twitter_account(username):
    print(f"Monitoring Twitter account: @{username}")
    marked_tweet_ids = load_marked_tweets(username)

    user_config = get_user_config(username)
    discord_webhook_urls = user_config["discord_webhook_urls"]
    notification_webhook_index = user_config["last_notification_webhook_index"]

    if not marked_tweet_ids:
        print("First run. Fetching and marking tweets...")
        tweets = await get_all_tweets_from_page(username, pages=10)
        tweet_ids = []
        for tweet in tweets:
            tweet_ids.extend([str(t[0]) for t in process_tweet(tweet)])
        save_marked_tweets(username, tweet_ids)
        print(f"Tweet IDs saved to {MARKED_TWEETS_FOLDER}/{username}.txt.")
        print("Skipping sending tweets to Discord on first run.")
    else:
        print("Resuming monitoring with existing tweet IDs.")

    while True:
        print("Checking for new tweets...")
        latest_tweets = await get_all_tweets_from_page(username)
        latest_tweet_ids = []
        for tweet in latest_tweets:
            latest_tweet_ids.extend([str(t[0]) for t in process_tweet(tweet)])

        new_tweets = []
        for tweet in latest_tweets:
            processed_tweets = process_tweet(tweet)
            for t_id, t_text, tweet_type in processed_tweets:
                if str(t_id) not in marked_tweet_ids:
                    new_tweets.append((t_id, t_text, tweet_type, tweet))

        if new_tweets:
            print(f"Found {len(new_tweets)} new tweet(s):")
            for t_id, t_text, tweet_type, tweet in reversed(new_tweets):
                print(f"- Type: {tweet_type}, Tweet ID: {t_id}, Tweet: {t_text}")

                author_name = tweet.author.name
                author_profile_image_url = tweet.author.profile_image_url_https.replace(
                    "_normal", ""
                )
                author_username = tweet.author.username
                created_on = format_utc_to_jakarta_local(tweet.created_on)
                media_urls = [url.media_url_https + "?name=orig" for url in tweet.media]
                text = await extract_original_links(tweet.text, tweet.urls)
                url = tweet.url

                await post_discord_notification(
                    author_name,
                    author_profile_image_url,
                    author_username,
                    created_on,
                    media_urls,
                    text,
                    tweet_type,
                    url,
                    discord_webhook_urls,
                    notification_webhook_index,
                )

                # Update the notification_webhook_index for the next tweet
                notification_webhook_index = (notification_webhook_index + 1) % len(
                    discord_webhook_urls
                )

            marked_tweet_ids = latest_tweet_ids + marked_tweet_ids
            save_marked_tweets(username, marked_tweet_ids)
        else:
            print("No new tweets found.")

        delay = random.randint(60, 120)
        print(f"Waiting for {delay} seconds before the next check...\n")
        await asyncio.sleep(delay)


async def login():
    username = input("Enter your Twitter username (email or handle): ").strip()
    password = getpass.getpass("Enter your Twitter password: ")

    await app.sign_in(username, password)
    print("Logged in successfully.")


async def main():
    if not os.path.exists(SESSION_FILE):
        print("No session file found. Please log in.")
        await login()
    else:
        print("Session file found. Connecting to existing session...")
        await app.connect()

    username = input("Enter the Twitter username to monitor (without @): ").strip()

    if not await validate_twitter_account(username):
        print(f"Error: The account @{username} does not exist or is invalid.")
        return

    await monitor_twitter_account(username)


if __name__ == "__main__":
    asyncio.run(main())

This Python script is a bot designed to monitor a specific Twitter account for new tweets and send notifications about them to Discord webhook channels. Here's an overview of its functionality:

Highlights of the Code:

Twitter Integration:
- The script uses the tweety library to interact with Twitter, allowing the bot to search for tweets and gather information about a specific user's activity (tweets).
- It handles session authentication, loading a previous session from a file or logging in when no session exists (login function).
Discord Notifications:
- The bot sends new tweet notifications to specified Discord webhook URLs using the discord library.
- The tweet information (text, media, links, and type) is extracted and formatted into rich Discord message embeds for better presentation.
Tracking and Persistence:
- It maintains a record of already "marked" tweets (those that have been processed) to avoid duplicate notifications. These are stored in the marked_tweets folder as files named after the monitored username.
- Configuration details like Discord webhook URLs and a pointer to track the last-used webhook URL are stored in a configs.json file for persistence across sessions.
Functionality Summary:
- Authentication: Authenticates with Twitter (either via existing session or login credentials).
- Monitoring: Monitors a specific Twitter account for new tweets using a loop with random delay (between 60–120 seconds for throttling).
- Processing: Processes each new tweet to format its details, resolve URLs, and categorize the type of interaction (e.g., posted, reply, retweet, quote retweet).
- Notifications: Sends a Discord notification for new tweets, attaching any images, text, and metadata for clarity.
- Localization: The script localizes UTC tweet timestamps to Jakarta's time zone (WIB) and translates the day names to Indonesian.

Breakdown of the Code Functions:

load_configs / save_configs:
- Load or save the configs.json file storing user configuration (e.g., webhook URLs and their most recent usage index).
get_user_config:
- Retrieves or initializes the configuration for a specified username, prompting the user to enter Discord webhook URLs if none exist.
load_marked_tweets / save_marked_tweets:
- Manages local storage of processed tweet IDs in a file to prevent duplicate notifications.
get_tweet_type / process_tweet:
- Determines the type of a tweet (post, reply, retweet, etc.) based on its attributes, and organizes tweet details into a structured tuple for further processing.
format_utc_to_jakarta_local:
- Converts UTC timestamps to Jakarta local time (WIB) and translates English day names to Indonesian.
extract_original_links:
- Replaces all shortened Twitter t.co URLs with their expanded forms in the tweet's text.
post_discord_notification:
- Formats and sends a new tweet notification to the configured Discord webhook URLs, using Discord embeds.
validate_twitter_account:
- Validates whether the provided username corresponds to an actual Twitter account.
get_all_tweets_from_page:
- Fetches tweets from a specified account using tweety with an option to retrieve multiple pages.
monitor_twitter_account:
- The main monitoring loop. Fetches tweets at regular intervals, compares them with already processed tweets, and triggers Discord notifications for new tweets.
login / main:
- Manages authentication and entry point for monitoring (via main). If no session exists, it prompts for login credentials.

Overall Purpose:

The script is primarily designed to monitor a Twitter account and notify Discord channels (via webhooks) of new tweets. It ensures that:

Only new tweets are processed and notified.
The system can handle media attachments and distinguish tweet types.
Notifications are sent to multiple Discord hooks in a balanced manner (using a round-robin approach).
It localizes time information and adheres to Jakarta timezone formatting.

This script can be useful for organizations, fans, or community managers who want to broadcast updates from a Twitter account to their Discord communities automatically.

Generate your own explanations

Download our vscode extension

Read other generated explanations

Built by @thebuilderjr