Задача: необходимо просматривать несколько новостных телеграм каналов, и в случае обнаружения в новости неких стоп-слов, высылать уведомление на электронную почту.
К сожалению воспользоваться для решения этой задачи API для работы с ботами не получится, т.к. такого функционала просто нет. Выходом может служить — воспользоваться одним из многочисленных клиентов Телеграм, реализованых на PHP, Python, JavaScript (NodeJS) и т.д. В моём случае — воспользуюсь python и библиотекой telethon. К ней довольно толковая документация, в том числе и на русском
Итак, для начала нужно зайти на ресурс https://my.telegram.org/apps и получить api_id и api_hash, для того чтобы библиотека смогла создать соединение. Далее создам файл с настройками вида:
{
"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 новости уже смотрели, пропускаем его
- если в тексте новости нашли стоп слово — отправляем уведомление на почту
А вот и сам скрипт:
#!/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())