Как исправить бота с несколькими потоками?

Question

[Elvis]

Привет!
Пишу бота для телеги и мне нужно реализовать следующий функционал: бот раз в 10 минут(к примеру) парсит определенный URL и если с предыдущего захода были изменения - писал в наш чат.
Так как бот занимается еще и другими делами, то решил зациклить выполнение парсинга в функции с sleep в конце. Если есть изменения - пытаюсь отправить в чат сообщение, но тут и случается проблема.
Так как удачное стечение обстоятельств не возникает от ивента в чате - я не могу из event вытащить entity для функции send_message. поэтому приходится доставать посредством функции get_entity и ссылки на чат как параметр, но почему то из другого потока это не получается. ниже приведу код упрощенный:

import threading
import asyncio
from telethon.sync import TelegramClient, events
import config as cfg

bot = TelegramClient('Bot', cfg.api_id, cfg.api_hash)


@bot.on(events.NewMessage(pattern=r'^(?i)(idchat){1}$'))
async def echoidchat(event):
    # просто команда чтобы видеть что бот в принципе работает
    # я знаю что можно вытащить entity из event, но это работает если из чата пришло событие,
    # но не работает если бот просто должен отправить в чат сообщение сам
    # поэтому вытаскиваю entity из ссылки на чат
    # тут всё проходит отлично
    channel = await bot.get_entity('https://t.me/elvistest')
    await bot.send_message(channel, 'ответ')


async def parseurls():
    # а вот тут уже не работает.
    # на get_entity выдает RuntimeWarning: coroutine 'UserMethods.get_input_entity' was never awaited
    # и не возвращает entity
    while True:
        channel = await bot.get_entity('https://t.me/elvistest')
        await bot.send_message(channel, 'ответ из другого потока')
        asyncio.sleep(5)

if __name__ == '__main__':
    bot.start(bot_token=cfg.bot_token)
    my_thread_flights = threading.Thread(target=asyncio.run, args=(parseurls(),))
    my_thread_flights.start()
    bot.run_until_disconnected()

Собственно как мне в таком варианте отправить сообщение в чат?

Answer 1

[Elvis]

Решил я свой вопрос. вот этот код работает ровно так как мне нужно.
parseurls - работает отдельно и время от времени дергает sendmsg для отправки сообщения в чат
Сам бот работает параллельно и отвечает на команду вне зависимости от "сна" parseurls.

import random
import asyncio
from telethon import TelegramClient, events
import config as cfg

bot = TelegramClient('Bot', cfg.api_id, cfg.api_hash)

@bot.on(events.NewMessage(pattern=r'^(?i)(idchat){1}$'))
async def echoidchat(event):
    await bot.send_message(event.chat, 'ответ')


async def parseurls():
    while True:
        ts = abs(int(random.random()*10))
        print(f'parseurls({ts})')
        await sendmsg(ts)
        await asyncio.sleep(ts)


async def sendmsg(msg):
    print(f'sendmsg({msg}) - start')
    channel = await bot.get_entity('https://t.me/elvistest')
    await bot.send_message(channel, f'ответ из другого потока {msg}')
    print(f'sendmsg({msg}) - done')


def main():
    bot.start(bot_token=cfg.bot_token)
    loop = asyncio.get_event_loop()
    tasks = [
        loop.create_task(parseurls()),
        loop.create_task(bot.run_until_disconnected()),
    ]
    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

if __name__ == '__main__':
    main()

Answer 2

[Vindicar]

Я бы посоветовал использовать asyncio.Queue, но там есть тонкости при работе с несколькими потоками.
Пусть поток-парсер вообще не занимается вопросами работы с ботом, а просто периодически производит парсинг, и если есть новости - кладёт их описание в выходную очередь. Он должен делать это с помощью call_soon_threadsafe(), как описано тут, так как класс asyncio.Queue не является потокобезопасным.
Зато бот в главном потоке может просто в цикле await'ить метод get() очереди, и отправлять сообщения по получению очередного объекта.

Альтернатива - использовать классический потокобезопасный queue.
Так будет проще помещать в него результаты парсинга, но сложнее их извлекать - придётся периодически (типа раз в 1-5 секунд) вызывать метод get_nowait(), чтобы узнать, есть ли что-то в очереди. Блокирующий get() подвесит бота.

Comments

[Elvis]

Начал разбираться в этом, но что-то всё никак. Получилось создать очередь, занести туда данные одной функцией parseurls (это будет потом парсинг, который в очередь будет складывать url страницы где произошли изменения), другую функцию сделал чтобы парсить очередь в цикле pre_sendmsg и третьей функцией вынес отправку сообщения в чат. Последние 2 можно объединить, но не суть(просто потом хочу еще использовать код отправки, чтобы был един).
Однако всё останавливается на первом же вызове bot.get_entity ((( ошибку уже не пишет, просто работает дальше. я так понимаю очередь не уменьшается, но сам бот работает функция echoidchat в это время отрабатывает.

import threading, queue
from time import sleep
import asyncio
from telethon.sync import TelegramClient, events
import config as cfg

bot = TelegramClient('Bot', cfg.api_id, cfg.api_hash)

@bot.on(events.NewMessage(pattern=r'^(?i)(idchat){1}$'))
async def echoidchat(event):
    channelaa = await bot.get_entity('https://t.me/elvistest')
    await bot.send_message(channelaa, 'ответ')


def parseurls():
    for x in range(10):
        q.put(x)
    pass
    

async def pre_sendmsg():
    while True:
        if not q.empty():
            msg = q.get_nowait()
            await sendmsg(msg)
            q.task_done()

async def sendmsg(msg):
    channel = await bot.get_entity('https://t.me/elvistest')
    await bot.send_message(channel, f'ответ из другого потока {msg}')

if __name__ == '__main__':
    q = queue.Queue()
    parseurls()
    bot.start(bot_token=cfg.bot_token)
    threading.Thread(target=asyncio.run, daemon=True, args=(pre_sendmsg(),)).start()
    bot.run_until_disconnected()

[Vindicar]

Нее, погоди, что-то тут не так. Вообще причём тут get_entity()? Зачем send прямо в функции-потоке? Вся возня с queue была для того, чтобы его оттуда вытащить, и отвязать поток от бота полностью! А если тебе нужно позарез
использовать методы бота, то тогда зачем тебе вообще второй поток?

Ну и касательно queue, я бы сделал так:

if not q.empty():
    msg = q.get_nowait()

Нет нужды, просто

try:
    msg = q.get_nowait()
except Empty: #очередь пуста, спим дальше.
    await asyncio.sleep(1.0)
else:
    await sendmsg(msg)
    q.task_done()

[Elvis]

Vindicar, да, мне нужно использовать методы бота для отправки сообщения в чат. я понял идею того что нужно парсить в одном потоке, если что-то есть, то положить в очередь данные и парсить дальше, а в потоке бота проверять очередь и если там что-то появилось - отправить это что-то в сообщении. но я видимо не до конца понимаю как это делается.