如何在 class 中将 Telegram 群消息监听器实现为后台线程？

Question

我已经开始工作了，但它正在订阅来自所有频道的所有消息。

如何让它只订阅来自特定频道 ID 的消息？

import asyncio
import threading

from telethon import TelegramClient, events

from time import sleep

api_id = '...'
api_hash = '...'

channelId = ...

class TelegramListener:
    async def actual_work(self):
        loop = asyncio.get_event_loop()
        client = TelegramClient('my_listener', self.api_id, self.api_hash, loop=loop)
        await client.connect()
        client.add_event_handler(self.callback)
        await client.run_until_disconnected()

    def go(self):
        asyncio.run(self.actual_work())

    def __init__(self, api_id, api_hash, callback):
        self.api_id = api_id
        self.api_hash = api_hash
        self.callback = callback
        threading.Thread(target=self.go).start()

async def on_message_callback(m):
    print(m)

t = TelegramListener(api_id, api_hash, on_message_callback)

# placeholder for some other runloop on main thread
while True:
    print('tick')
    sleep(1)

这会产生输出：

tick
tick
UpdateNewChannelMessage(message=Message(id=262, peer_id=PeerChannel(channel_id=...), date=datetime.datetime(2021, 3, 18, 1, 10, 41, tzinfo=datetime.timezone.utc), message='foo', out=False, mentioned=False, media_unread=False, silent=False, post=False, from_scheduled=False, legacy=False, edit_hide=False, pinned=False, from_id=PeerUser(user_id=...), fwd_from=None, via_bot_id=None, reply_to=None, media=None, reply_markup=None, entities=[], views=None, forwards=None, replies=MessageReplies(replies=0, replies_pts=263, comments=False, recent_repliers=[], channel_id=None, max_id=None, read_max_id=None), edit_date=None, post_author=None, grouped_id=None, restriction_reason=[], ttl_period=None), pts=263, pts_count=1)
tick
tick

PS 如果您发现代码有改进之处，请发表评论！

Answer 1

根据 docs you can pass event builder class or instance to add_event_handler, in your case events.NewMessage. Besides as to NewMessage class 定义，只需尝试将 chats 设置为您标记的 channelId:

channelId = ...

class TelegramListener:
    async def actual_work(self):
        loop = asyncio.get_event_loop()
        client = TelegramClient('my_listener', self.api_id, self.api_hash, loop=loop)
        await client.connect()
        client.add_event_handler(self.callback, events.NewMessage(chats=channelId))
        await client.run_until_disconnected()
    ...

如何在 class 中将 Telegram 群消息监听器实现为后台线程？

How to implement Telegram group message listener as background thread in class?

multithreading

telegram

telegram-bot

background-process

telethon