How to send messages to telegram from the StreamListener handler (twitter)? tweepy & telethon

Question

I manage to send only 1 message after launching the script, after which it hangs and no longer receives messages from Twitter If I remove the block of code I wrapped in "------------------------------" then I will receive all the tweets, but when I try to send it to Telegram, it stops after the first time

Initially did without separate threads, because I could not achieve the result Wrapped everything in separate threads, but the result is the same

What am I doing wrong?

from telethon import TelegramClient, events, sync
from telethon.tl.types import InputChannel
import tweepy
import yaml
import sys
import coloredlogs, logging
import asyncio
import threading
import concurrent.futures
import time

start_twitter = threading.Event()

forwardinput_channel_entities = []
forwardoutput_channels = {}

class MyStreamListener(tweepy.StreamListener):
    def on_status(self, status):
        user_id = status.user.id

        if user_id in forwardoutput_channels:
            for output_channel in forwardoutput_channels[user_id]:

                message = status.text
                logging.info('-------------')
                logging.info(message)

                # ------------------------------
                try:
                    loop = asyncio.get_event_loop()
                except Exception as e:
                    loop = asyncio.new_event_loop()
                    asyncio.set_event_loop(loop)
                    logging.error(e)
                    pass

                loop.run_until_complete(telegram_client.send_message(
                        output_channel['channel'], message))
                # ------------------------------

def twitter_thred():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    auth = tweepy.OAuthHandler(config['twitter_consumer_api'],
                               config['twitter_consumer_secret'])

    auth.set_access_token(config['twitter_user_api'],
                          config['twitter_user_secret'])

    global twitter_api
    twitter_api = tweepy.API(auth)

    myStreamListener = MyStreamListener()
    myStream = tweepy.Stream(auth=twitter_api.auth, listener=myStreamListener)

    start_twitter.wait()
    myStream.filter(follow=forwardinput_channel_entities,
                    is_async=True)

def telegram_thred():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)

    global telegram_client
    telegram_client = TelegramClient(config['session_name'],
                                     config['api_id'],
                                     config['api_hash'])
    telegram_client.start()

    for forwardto in config['forwardto_list_ids']:

        for twitter_user_id in forwardto['from']:
            forwardinput_channel_entities.append(str(twitter_user_id))

            channels = []

            for channel in telegram_client.iter_dialogs():
                if channel.entity.id in forwardto['to']:
                    channels.append({
                        'channel': InputChannel(
                            channel.entity.id, channel.entity.access_hash),
                    })

            forwardoutput_channels[twitter_user_id] = channels

    start_twitter.set()

    telegram_client.run_until_disconnected()

def start():

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        future = executor.submit(telegram_thred)
        future = executor.submit(twitter_thred)

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print(f'Usage: {sys.argv[0]} {{CONFIG_PATH}}')
        sys.exit(1)
    with open(sys.argv[1], 'rb') as f:
        global config
        config = yaml.safe_load(f)

    coloredlogs.install(
        fmt='%(asctime)s.%(msecs)03d %(message)s',
        datefmt='%H:%M:%S')
        
    start()

An example of a yml config to run the script:

# telegram
api_id: *****************
api_hash: '*****************'
session_name: 'test'

# twitter
twitter_consumer_api: '*****************'
twitter_consumer_secret: '*****************'
twitter_user_api: '*****************'
twitter_user_secret: '*****************'

forwardto_list_ids:
  - from:
      - 0000000000    # account twitter id
    to:
      - 0000000000    # telegram channel id

Seems tweepy is blocking, and Telethon is asyncio. There's an issue for asyncio support in tweepy https://github.com/tweepy/tweepy/issues/732. You may need something like https://stackoverflow.com/a/63862405/4759433 until that issue is resolved. — Lonami, Apr 11 '21 at 14:56
@Lonami, unfortunately this did not help, it seems that telegram itself breaks the event stream or uses the wrong one when sending a message — vellnes, Apr 12 '21 at 20:58

score 0 · Accepted Answer · answered Apr 15 '21 at 02:15

As noted, Tweepy doesn't support asyncio with streaming yet, so it's blocking the event loop when you run the stream. is_async uses a threaded approach.

For now, you should look into using Tweepy's async-streaming branch / https://github.com/tweepy/tweepy/pull/1491.

How to send messages to telegram from the StreamListener handler (twitter)? tweepy & telethon

1 Answers1