Парсинг телеграм каналов
Задача: необходимо просматривать несколько новостных телеграм каналов, и в случае обнаружения в новости неких стоп-слов, высылать уведомление на электронную почту.
К сожалению воспользоваться для решения этой задачи API для работы с ботами не получится, т.к. такого функционала просто нет. Выходом может служить — воспользоваться одним из многочисленных клиентов Телеграм, реализованых на PHP, Python, JavaScript (NodeJS) и т.д. В моём случае — воспользуюсь python и библиотекой telethon. К ней довольно толковая документация, в том числе и на русском
Итак, для начала нужно зайти на ресурс https://my.telegram.org/apps и получить api_id и api_hash, для того чтобы библиотека смогла создать соединение. Далее создам файл с настройками вида:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
{ "chanels": [ "rus_now_news","-1001237513492" ], "limit": 200, "alert_words": [ "кусь","мейнкун","бабки","лапландия","рабочие","песель-акробат" ], "api_id": "12435245235", "api_hash": "екыпукерпенуркенрке", "notify_email": "екпукеп@укепукеуке.ru", "email_login": "уепукеп-куепукеп@кепукеп.ru", "email_from": "уекпукеп@кепукепук.ru", "from_password": "укацука!укауцка", "smtp_server": "уцкацука-owa.уцкацука.ru", "smtp_port":587 } |
В нём перечисляем каналы которые мониторим и стоп слова, которые ловим. Алгоритм работы скрипта:
- соединяемся с сервером телеграм
- получаем список последних новостей канала
- если ID новости уже смотрели, пропускаем его
- если в тексте новости нашли стоп слово — отправляем уведомление на почту
А вот и сам скрипт:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 |
#!/usr/bin/env python3 # encoding: utf-8 import asyncio import json import sys import re from telethon import TelegramClient import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from telethon.tl.functions.contacts import ResolveUsernameRequest from telethon.tl.functions.channels import GetMessagesRequest from telethon.tl.functions.messages import GetHistoryRequest, ReadHistoryRequest with open('config.json', 'r') as file: config_data = json.load(file) print(config_data) client = TelegramClient("parser_data", config_data["api_id"], config_data["api_hash"]) async def TConnect(): await client.start() async def ListChanels(): async for dialog in client.iter_dialogs(): print(dialog.name, 'has ID', dialog.id) def save_chan_json(chan,chan_data): f = open(chan + '.json', "w+") json.dump(chan_data, f) f.close() def get_chan_json(chan): # узнаём какое последнее сообщение прочитали на канале? chan_data = {} chan_data["las_id"] = 0 try: with open(chan + '.json', 'r') as file: chan_data = json.load(file) print(chan_data) return chan_data except: save_chan_json(chan, chan_data) return chan_data return chan_data; def SendMail(chan,word,message): msg = MIMEMultipart() msg['Subject'] = f"Найдено слово '{word}' в новости на канале {chan} в Телеграм" msg.add_header('Content-Type', 'text/html') message.text=message.text.replace(word,"<strong>"+word+"</strong>") dt_pub=message.date.strftime('%d-%m-%Y %H:%M:%S') msg.set_payload(f"Канал: <a href='https://t.me/{chan}'>https://t.me/{chan}</a>, опубликовано {dt_pub}<hr/>"+message.text) smtpObj = smtplib.SMTP(config_data["smtp_server"], config_data["smtp_port"]) smtpObj.starttls() smtpObj.login(config_data["email_login"], config_data["from_password"]) smtpObj.sendmail(config_data["email_from"], config_data["notify_email"], msg.as_string().encode('utf-8')) smtpObj.quit() async def main(): print("-start") await TConnect() #await ListChanels() for chan in config_data["chanels"]: chan_data=get_chan_json(chan) skeep_after = chan_data["las_id"] if "-" in chan: dp = await client.get_entity(int(chan)) else: dp = await client.get_entity(chan) poz=0 async for message in client.iter_messages(dp,limit=config_data["limit"]): if poz==0: chan_data["las_id"]=message.id save_chan_json(chan, chan_data) if skeep_after==message.id: print("Все новости уже прочитаны...") break print(f"-смотрим message_id:{message.id}") for word in config_data["alert_words"]: if word in message.text: print(f"--нашли слово {word}") SendMail(chan, word, message) print(message) #print(message.id, message.text) poz=poz+1 print("all done.."); if __name__ == '__main__': for param in sys.argv: if param == "--list_chanels": TConnect() ListChanels() with client: client.loop.run_until_complete(main()) |
