| import logging |
|
|
| from telethon import TelegramClient, events |
| from telethon.errors import ChannelPrivateError |
|
|
|
|
| logging.basicConfig( |
| format='[%(levelname) 5s/%(asctime)s] %(name)s: %(message)s', |
| level=logging.WARNING |
| ) |
|
|
|
|
| with open('data/tg_api_id.txt') as f: |
| api_id = f.read().strip() |
|
|
| with open('data/tg_app_api_hash.txt') as f: |
| api_hash = f.read().strip() |
|
|
| with open('data/tg_bot_token.txt') as f: |
| bot_token = f.read().strip() |
|
|
| channel_link = 'https://t.me/redakciya_channel' |
|
|
|
|
| client = TelegramClient('data/bot_session.session', api_id, api_hash) |
|
|
| """ для бота это всё не работает! """ |
|
|
| @client.on(events.NewMessage(chats=channel_link)) |
| async def on_new_message(event): |
| if '#ньюсдня' in event.raw_text: |
| print(event.message) |
| else: |
| print('@@@ other message: ') |
| print(event.message) |
|
|
|
|
| async def check_channel(): |
| try: |
| channel = await client.get_entity(channel_link) |
| print(f'Channel open: {channel.title}') |
| async for msg in client.iter_messages(entity=channel, limit=100): |
| if msg.message.lower().find('#ньюсдня') != -1: |
| print(f'Last message:') |
| print(msg.message) |
| break |
| print(f'Listening for new messages...') |
| except ChannelPrivateError: |
| print("The channel is private and you don't seem to have access.") |
| except Exception as e: |
| print(f"An error occurred: {e}") |
|
|
| client.start(bot_token=bot_token) |
| client.loop.run_until_complete(check_channel()) |
| client.run_until_disconnected() |
|
|
|
|
|
|
|
|