Fwd from Twitter to Telegram bot v 0.1.0 (deps: twint, aiogram, nest_asyncio)

app.py

import asyncio
import twint
from aiogram import Bot, types
from aiogram.dispatcher import Dispatcher

import nest_asyncio
nest_asyncio.apply()

TOKEN = 'YOUR_TOKEN_HERE'
bot = Bot(token=TOKEN)
dp = Dispatcher(bot)
loop = asyncio.get_event_loop()

# TODO: switching keys and values might be a good idea to improve performance
CLIENTS: dict[int, set[str]] = dict()


async def add_entity(id: int, who_to_follow: str) -> set[str]:
    if id not in CLIENTS.keys():
        CLIENTS[id] = set([who_to_follow])
        asyncio.create_task(tweets_scanner(who_to_follow))
    else:
        if who_to_follow not in CLIENTS[id]:
            CLIENTS[id].add(who_to_follow)
            asyncio.create_task(tweets_scanner(who_to_follow))
            
    return CLIENTS[id]

# TODO: finish all workers if there is no more clients follows this twitter account 
async def unfollow(id: int, who_to_unfollow: str) -> tuple[bool, str]:
    if id not in CLIENTS.keys():
        return False, "You are not following anyone"

    if who_to_unfollow not in CLIENTS[id]:
        return False, "You are not following this person"

    CLIENTS[id].remove(who_to_unfollow)
    return True, f"Unsubscribed from {who_to_unfollow}"


def twint_config() -> twint.Config:
    c = twint.Config()
    c.User_full = True
    c.Profile = True
    # ? even if we need only one tweet, we could not get less than 5, 
    # ?     because of the library bugs/won't work with less than ~~5
    c.Limit = 5
    c.Store_object = True
    c.Hide_output = True
    return c


async def last_tweet(conf: twint.Config) -> twint.tweet.tweet:
    tweets = list()         # prepare list to store tweets
    conf.Store_object_tweets_list = tweets
    # ! async call here
    twint.run.Search(conf)  # get tweets
    # TODO: check if tweets is empty, handle errors
    last = tweets[0]        # save last tweet
    tweets.clear()          # flush tweets
    return last             # return last tweet only


async def tweets_scanner(twitter_nickname: str, delay: float = 6.0):
    config = twint_config()
    config.Username = twitter_nickname
    last = await last_tweet(config)
    while True:
        # delay next check by specified X seconds
        await asyncio.sleep(delay)
        current = await last_tweet(config)
        if current.id != last.id:
            for tg_id, nicknames in CLIENTS.items():
                if twitter_nickname in nicknames:
                    await bot.send_message(tg_id,
                        f"{twitter_nickname} tweeted: {current.tweet}\n {current.link}")

            print("New tweet: ", current.datestamp,
                  "\t📝:", current.tweet,
                  "\t👉", current.link)

            last = current
        # else:
        #     print('No new tweets from: ', twitter_nickname)


def is_active(client_id: int) -> bool:
    return client_id in CLIENTS.keys()


@dp.message_handler(commands=['follow'])
async def process_follow_command(message: types.Message) -> None:
    args: str | None = message.get_args()
    if args is None:
        await message.reply(
            "No arguments provided;\n"
            "Write /follow <twitter nickname> to follow someone")
        return  # exit from handler

    # get message sender id
    client_id: int = message.from_user.id
    # strip spaces at the beging/end
    nickname: str = args.strip()
    if is_active(client_id):
        updated: set[str] = await add_entity(client_id, nickname)
        await message.reply("You are now following:\n" + "\n".join(updated))
    else:
        # if it's the first time client sending /follow request
        # -> create new entry in dict
        _ = await add_entity(client_id, nickname)
        await message.reply(f"Now following: {nickname}")


@dp.message_handler(commands=['start'])
async def process_start_command(message: types.Message):
    await message.reply(
        "Hello, I am TwitterForwader Bot 👋!\n"
        "Write /follow <twitter nickname> to start following someone")


if __name__ == '__main__':
    loop.create_task(dp.start_polling())
    loop.run_forever()