Агрегатор

Любой человек может оказаться в неприятной ситуации когда он едет ночью, в лес, в багажнике... Предусмотрительные граждане пытаются избежать подобных инцидентов выбирая сервисы такси известных брендов, которые декларируют безопасность поездки, контроль за водителями и даже вешают в приложении огромную кнопку "БЕЗОПАСНОСТЬ" которую надо жать в случае если что-то пошло не так.

Но помогает ли эта кнопка? Давайте проверим на практике.

Дано: молодая симпатичная девушка и её кнопочный телефон (без навигатора), незнакомый город бывшей союзной республики, зима, метель, мороз.

Требуется: добраться из частного сектора на окраине в центр, путь пролегает мимо промзон, лесопарков.

Решение (казалось бы очевидное): берём смартфон с сервисом ЯндексGO у супруга, заказываем ЯндексТакси, садимся и едем. Счастливый супруг тем временем наблюдает онлайн, как машина везёт пассажира по адресу...

Ой, никто уже никого никуда не везёт.

"Вы приехали" сообщает приложение. "Как всё прошло?"
"Стоимость поездки изменилась, заказ завершён не в указанной ранее точке." - вот тут наступает некоторая, так скажем, паника. Особенно на фоне новостей про онанистов из сервиса такси. Звонок супруге (не дословно):

• "Ты где?! Уже приехала?"

• "Нет, ещё еду."

• "Странно, мне показывает что поездка завершилась досрочно... Что за окном? Уже город?"

• "Нет, поля." (путь вполне может пролегать мимо лесополосы, полей, промзоны)

• "Ок, как приедешь - позвони сразу. Что-то странное твориться, будь осторожна."

• "Хорошо."

Что делать?

По логике - надо узнать где машина, причём быстро. На смартфоне - только маршрут который машина проехала до "отключения" поездки. А далее - перекрёсток, там можно свернуть или в сторону города (лес, поля вокруг) или в лес и поля подальше от города.

Расспрашивать по телефону супругу - водитель, в случае если задумал что-то неправомерное, может "ускориться" в своих действиях. Просить её покинуть машину ввиду непонятности ситуации - см. выше, плюс неопределённость в местоположении с поправкой на погоду, замёрзнуть на обочине в незнакомом месте не очень приятно наверное.

Пытаемся воспользоваться пресловутой кнопкой "Безопасность" которая вроде бы должна помочь в такой ситуации. По быстрому описываем вкратце ситуацию в чате "команде безопасности" и... получаем отфутболивание к службе поддержки которая "ответит в ближайшее время".

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

И вот тут приходит осознание, что если водитель решил вывезти вас в лес, то кнопка "Безопасность" не поможет когда водитель попросту нажал "Завершить поездку".
Только ваш труп в холодном зимнем лесу получит через пару часов от службы поддержки сообщение, что водитель плохо поступил и его отключат от сервиса если он сделает ещё пару раз так же.

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

Ну а водитель, решив позабавиться с пассажиром и поиграть в маньяка, может просто нажать кнопку в приложении и никто не узнает какая точка была финальной в жизненном пути неудачника решившего прокатиться на ЯндексТакси.

Предпосылки и проблематика

Когда я переключился с ВК на телеграм, выбрав последний основным мессенджером, я столкнулся с рядом неудобств: в телеге нет привычной ленты новостей и весь контент надо собирать по каналам. ВК имеет функционал групп (сообщения из которых и собираются в ленту), но меня всегда бесило наличие рекламы явной, рекламных постов и прочего мусора, видеть который я в ленте не хотел.

Одно время, когда ВК начал пихать в ленту свои рекомендации (неотключаемые вообще никак), я пользовался адблоком, альтернативными андроид-клиентами (kate mobile), но все эти полумеры напрягали.

В телеге же, как только количество каналов подросло, очень напрягала необходимость заходить в каждый. И, когда их набралось штук 30, я стал забивать на некоторые, т.к. утомлялся туда-сюда по ним перемещаться.
Плюс, как-то внезапно, много каналов начали спамить крос-сылками с рекламой. И ладно, если это партнерские посты из схожих по тематике каналов. Чаще это были всякие курсы английского, питона, а иногда вообще скам про казино или крипту.

Идея и реализация

Так появилась мысль сделать агрегатор телеграм-каналов с фильтрацией по ключевым словам (которые в рекламных постах достаточно часто попадаются).

Я в курсе про функционал группировки каналов в некоторых клиентах, но по некоторым причинам он мне не подходит:

• группировка работает только на уровне клиента. Т.е. сгруппированные в мобильном приложении каналы не будут сгруппированы в веб-версии или другом клиенте

• нельзя (по крайней мере я не нашел) группировать чаты и каналы

• фильтрация спама и рекламы недоступна

• я собирался экспортировать агрегированный контент в rss (об этом во второй части статьи)

Погуглив существующие инструменты для работы с телегой, я остановился на питоновской библиотеке telethon. Ее ключевой особенностью была возможность работать в режиме клиента, а не бота (об этом ниже, в нюансах).

Название придумал быстро: я люблю соединять слова, так что агрегатор телеграма быстро превратился в Телегрегатор.

Нюансы агрегатора

Процесс установки и использования самой библиотеки я не буду расписывать, в доках все достаточно подробно. Остановлюсь на нескольких технических нюансах, которые всплыли в процессе создания Телегрегатора:

1. Поскольку меня напрягают висячие непрочитанные сообщения (а так же больше 5-7 вкладок в браузере), я хотел, чтобы агрегатор вместо меня подписывался на каналы и пересылал мне сообщения в специально заведенный для этого чат/канал (я это назвал поток, feed).

2. Как следствие из п.1, агрегатор должен был самостоятельно подписываться на каналы (в т.ч. закрытые), вступать в группы и чаты. А это значило, что вариант с ботом не подходил, т.к. он пассивен и его может добавить только участник чата или владелец канала. Здесь и пригодилась возможность telethon'а эмулировать клиент. Т.е. telethon-клиент авторизуется с номером телефона и кодом из смс и эмулирует обычного юзера. Может реагировать на сообщения о наборе текста, заходить в каналы, группы, чаты, пересылать сообщения, и всё это в реалтайме. В общем, полноценный пользователь.

3. Как следствие из п.2, возможности бота (кастомные клавиатуры и команды) оказались мне недоступны (или я не нашел, как клиентом это делать). Если знаете как - напишите в комментах. Так что пришлось велосипедить с обработкой пользовательских команд.

4. Была идея, что пользователю надо общаться с ботом, а тот уже под капотом передает команды клиенту (подписаться на канал), но это показалось слишком сложным для mvp.

Как попробовать?

Сейчас процесс выглядит так:

1. Получаете название канала (либо инвайт-ссылку, для приватных каналов), который хотите добавить в агрегатор.

2. Создаёте новый канал (приватный или публичный), который будет выступать в роли потока (ленты)

3. Находите через поиск и добавляете в поток пользователя Телегрегатор (@telegregator) и даёте права на публикацию сообщений (через меню администраторов).

4. Телегрегатор поздоровается и покажет доступные команды.

5. Отправляете команду /join @имя_канала (из. п.1), после чего Телегрегатор подписывается на нужный канал. Можно указать пачку каналов, через пробел (займет некоторое время, ограничение порядка 50 каналов в минуту).

6. Можно просто переслать в поток сообщение из любого канала, Телегрегатор сам подпишется на канал.

7. Все новые сообщения из подписанных каналов Телегрегатор будет прислать в этот поток.

Потоков может быть несколько. Вы можете добавлять для каждого потока фильтры по стопсловам (это пока тестовая функция, так что если вдруг что-то перестало работать - сбросьте фильтры).

Ограничения и планы

Есть несколько ограничений: альбом из фоток, которые пришли в оригинальный канал в одном сообщении, Телегрегатор пересылает отдельными сообщениями. Или репост в поток иногда происходит не мгновенно, а спустя 5-10 секунд. Иногда Телегрегатор падает (рестарт автоматически), и может пропустить какое-то сообщение. Это редко, но случается.

Видимо, это особенности протокола. Если заметите странное поведение - напишите мне.

В планах добавить следующий функционал:

• показать случайный пост из случайного канала

• топ самых популярных каналов (самых редких, как вариант)

• предлагает случайный канал

Я люблю находить годноту в телеге. Порой, в канале всего человек 200, и о нём вообще не узнать, кроме как по рекомендациям друзей.

Сам пользуюсь Телегрегатором где-то полгода, решил поделиться, вдруг кому-то зайдет. Из своего аккаунта я удалил все подписки, теперь ими рулю через потоки, полет нормальный.

Исходный код пока грязный, так что публиковать не стал.

Часть вторая, про RSS

В процессе работы над Телегрегатором появилась мысль экспортировать это все в RSS-ленту. Я многократно встречал статьи, в т.ч. на Хабре, как автор читает вообще всё через rss, собирая его из разных источников. Кто-то пошел дальше: из rss собирает подборку и формирует ежедневную рассылку на почту. Планирую прийти к такому варианту. Если вы имеете такой опыт - отпишитесь, пожалуйста, в комментах, очень интересна ваша реализация.

Генераторов rss из соцсетей много, пару примеров:

• http://feed.exileed.com/

• https://rsshub.app/telegram/channel/temablog

Но у них есть недостатки:

• не умеют читать приватные каналы

• редко обновляют кеш

• не отображают вложения (фото, видео, музыку)

• медленно или нестабильно работают (иногда падают)

Поэтому, вторым очевидным шагом стало создание моста telegram->rss. По факту, это тот же клиент-Телегрегатор, который по имени канала получает последние сообщения из канала (без подписки) и возвращает в виде rss-ленты. В планах сделать подписку на запрашиваемый канал, тогда можно будет получать ленту и из приватных каналов, но пока так. Отпишитесь, если такой функционал востребован. Проверить можно здесь: http://tg2rss.prosto-tak.ru/.

По вопросам или замечаниям можете писать сюда в личку, в телегу @parotikov или на nik.parotikov@gmail.com.

P.S. В "Я пиарюсь" не хватает кармы. Для телеграма вообще нет хаба, странно.

В течение нескольких часов после публикации статьи о запуске Телегрегатора (http://personeltest.ru/aways/habr.com/ru/post/544414/) приходили сообщения с благодарностями и вопросами по нюансам работы.

Но недолго я радовался: потом пошел поток сообщений о том, что бот перестал добавлять в поток новые каналы либо вообще не работал.

Я это списывал на не слишком подробную инструкцию по запуску, сбои в работе самого бота (о множестве существующих лимитов я узнал из недавней статьи http://personeltest.ru/aways/habr.com/ru/post/543676/), возможные ошибки пользователей.

Путем общения с пользователями, чтением логов и раздумий пришел к выводу, что юзербот (а именно так, в соответствие со статьей выше, принято называть аккаунт телеграма, работающий в режиме бота) упёрся в лимит каналов (500 на аккаунт, спасибо @Slavenin999 за наводку). Я не уточнял, приватных или публичных, да это и не важно. Юзер бот упёрся в них за день.

У меня было в планах выложить код Телегрегатора в открытый доступ, но, перед этим, я хотел его хоть немного привести в подобающий вид. Однако, учитывая события с лимитом, решил не откладывать в долгий ящик, и публикую его на гитхабе (https://github.com/parotikov/telegregator), чтобы каждый желающий смог развернуть его для себя и не ждать решения с обходом лимитов, т.к. это может затянуться..

Сами подробности по установке будут в репе с ботом. В общем-то, на этом всё, ниже приведены несколько ключевых моментов из кода, который местами даже прокомментирован.

inb4: пинать за код не надо, питон только осваиваю. это прототип, который не взлетел.

Инициализация всего

from telethon import TelegramClient, events, sync, functions, types, utilsfrom telethon.tl.custom import Buttonimport asyncioimport osimport loggingfrom TelegramCommand import *from database import *import importlibfrom dotenv import load_dotenvload_dotenv()logger = logging.getLogger("tggt")logger.setLevel(logging.DEBUG)fh = logging.FileHandler("app.log")formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')fh.setFormatter(formatter)logger.addHandler(fh)api_id = os.getenv("API_ID")api_hash = os.getenv("API_HASH")master_account = os.getenv("MASTER_ACCOUNT")bot_token = os.getenv("BOT_TOKEN")client = TelegramClient('telegregator_session', api_id, api_hash)

В начале подключаем все необходимые библиотеки, настраиваю логгер, подключение самого telethon'а. Дальнейшее общение с телегой будет происходить через client и его декораторы.

Все действия Телегрегатора разделены на три группы: реакция на команды пользователя, реакция на новые сообщения в добавленных каналах и реакция на служебные события (бот добавлен в чат, написал новый пользователь и т.д.).

Команды

На все входящие сообщения от пользователя бот реагирует через декоратор @client.on(events.NewMessage):

@client.on(events.NewMessage)    async def my_event_handler(event):        logger.info("\r\n\r\n\r\n-------------Новое сообщение----------------")        ## Определяем кто к нам постучался        if event.message.from_id:            contact = await add_contact(event.message.from_id)        # личка бота        management = 0        handled = 0        if hasattr(event.message.to_id, 'channel_id'):            handled = 1            # logger.info('сообщение пришло в канал (тип channel_id), нужно обрабатывать для пересылки')            try:                chat = await client.get_input_entity(event.message.to_id)            except Exception as e:                logger.error(e)        if hasattr(event.message.to_id, 'user_id'):            if event.message.to_id.user_id == 887053090 or event.message.to_id.user_id == None:                logger.info("сообщение в личку телегрегатора, обрабатывать как управляющую команду")                management = 1                try:                    chat = await client.get_input_entity(event.message.from_id)                    await client.send_read_acknowledge(chat, event.message)                except Exception as e:                    logger.error(e)            else:                try:                    chat = await client.get_input_entity(event.message.to_id)                except Exception as e:                    logger.error(e)        if hasattr(event.message.to_id, 'chat_id'):            logger.info('сообщение в группу (chat_id) %s, нужно обрабатывать как команду' % event.message.to_id.chat_id)            try:                chat = await client.get_input_entity(event.message.to_id)            except Exception as e:                logger.error(e)        if hasattr(event.message.to_id, 'channel_id'):            logger.info('сообщение в канал (channel_id) %s, нужно обрабатывать как команду' % event.message.to_id.channel_id)            try:                chat = await client.get_input_entity(event.message.to_id)            except Exception as e:                logger.error(e)            # для отладки, пересылаем сообщение админу            await event.message.forward_to(master_account)        # mark message as read        await client.send_read_acknowledge(chat, event.message)        #Парсим команду.         command_handler = TelegramCommand()        command, command_arg = command_handler.parseMessage(event.message.message)        # если распознана команда        if command:            # помечаем сообщение прочитанным            await send_read(event.message.from_id, event.message)            chat_command = command            logger.info('Команда: {}, аргумент: {}'.format(chat_command, command_arg))            # на присоединение к каналу            if chat_command == 'help':                await call_command(module='helpmessage', answer_to=chat)            if chat_command == 'new':                await call_command(module='new', answer_to=chat, contact=contact, group_name = command_arg)            if chat_command == 'test':                await call_command(module='test')            if chat_command in ['join', 'add']:                channels_list = command_arg.split(' ')                for channel in channels_list:                    await call_command(module='join', answer_to=chat, channel_name=channel)            # на отписку от канала            if chat_command in ['leave', 'remove', 'delete']:                channels_list = command_arg.split(' ')                for channel in channels_list:                    await call_command(module='leave', answer_to=chat, channel_name=channel)            # на список подписанных каналов            if chat_command == 'list':                await call_command(module='listchannels', answer_to=chat, feed_id=event.message.to_id.chat_id)            ## Вкл/выкл поток            if chat_command in ['stop', 'pause']:                await call_command(module='stop', answer_to=chat)            if chat_command in ['start', 'resume']:                await call_command(module='start', answer_to=chat)            if chat_command in ['deleteall', 'exit']:                await call_command(module='deleteall', answer_to=chat)            ## Фильтры и стоп слова            if chat_command == 'filter':                await call_command(module='filter', answer_to=chat, word = command_arg, contact_id = contact.contact_id)            if chat_command == 'filterremove' or chat_command == 'removefilter' or chat_command == 'unfilter':                await call_command(module='unfilter', answer_to=chat, word = command_arg, contact_id = contact.contact_id)            if chat_command == 'filterlist':                await call_command(module='filterlist', answer_to=chat, contact_id = contact.contact_id)            if chat_command == 'filterclear' or chat_command == 'clearfilter':                await call_command(module='filterclear', answer_to=chat, contact_id = contact.contact_id)            if chat_command == 'filterstop' or chat_command == 'stopfilter':                await call_command(module='filterstop', answer_to=chat)            if chat_command == 'filterstart' or chat_command == 'sartfilter':                await call_command(module='filterstart', answer_to=chat)            if chat_command == 'message':                await call_command(module='message', message=command_arg)

Тут проверяются наличие необходимых полей во входящем сообщении event.message. Определяем, в личку, в канал или в чат написали., И, через кучу ifов, выбирается модуль с соответствующей командой. Они лежат в отдельных файлах, которые импортируются динамически. Это первый кандидат на рефакторинг, надо обертки написать, чтобы поменьше кода дублировалось.

Все команды /join, /filter и прочие обрабатываются здесь.

Как поток подписывается на новые каналы

Тот же самый декоратор @client.on(events.NewMessage)обрабатывает сообщения в группах и каналах (PeerChat и PeerChannel), если переслать в него сообщение из канала для подписки:

if event.message.fwd_from and event.message.fwd_from.channel_id and event.message.to_id.__class__.__name__ == 'PeerChat':            logger.info("Вижу репост канала в группу")            try:                channelInfo = await client.get_entity(event.message.fwd_from.channel_id)            except Exception as e:                logger.error(e)                logger.error("не можем получить channelInfo для канала %s" % event.message.fwd_from.channel_id)                channelInfo = None                await event.reply('не могу подписаться. Если канал закрытый, попробуйте по инвайту')            else:                channelEntity = await client.get_input_entity(event.message.fwd_from.channel_id)                logger.info("Репост канала {} (@{}), id {}".format(channelInfo.title, channelInfo.username, channelInfo.id))                await call_command(module='join', answer_to=chat, channel_name=channelInfo.username)

Пересылка постов из каналов в ленту

Ну и главный функционал, это пересылка всех постов из подписанных каналов в одну ленту (поток):

async def handle_message(event):    try:        logger.info('ищем в БД поток, где channel_id = %s' % event.message.to_id.channel_id)        forwards = Forward.select().where(Forward.channel_id == event.message.to_id.channel_id)    except Forward.DoesNotExist:        logger.info('Канала нет среди пересылаемых')        forwards = None        return    if not len(forwards):        logger.info('нет подписок на этот канал')        return    logger.info("Нашли следующие потоки, подписанные на канал channel_id %s" % event.message.to_id.channel_id)    logger.info(forwards)    chat = await client.get_input_entity(event.message.to_id)    await client.send_read_acknowledge(chat, event.message)    for forward in forwards:        # определяем надо ли пересылать        try:            feed = Feed.get(Feed.feed_id == forward.feed_id)        except Feed.DoesNotExist:            logger.info('Поток не найден')            feed = None            return        if not feed.is_enable:            logger.info("Поток был отключен, не пересылаем")            return        # logger.info('Определяем создателя потока, куда должен транслироваться канал, откуда пришло сообщение')        log_message = 'Сообщение пришло в канал {}'.format(event.message.to_id)        if hasattr(event.message.to_id, 'channel_id'):            feed_id = forward.feed_id            try:                owner = Contact.select().join(Feed, on=(Contact.contact_id == Feed.contact_id)).where(Feed.feed_id == feed_id).get()            except Contact.DoesNotExist:                log_message = log_message + ', владелец канала не определен'                owner = None            else:                log_message = log_message + ', id владельца канала: {}'.format(owner.contact_id)            logger.info(log_message)        else:            logger.error('нет параметра event.message.to_id.chat_id')        log_message = 'пробуем фильтровать.'        filterlist = []        # TODO получать для конкретного юзера. нужно будет поменять логику и порядок, перенести на 438 строку        try:            filters = Filter.select().where(Filter.contact_id == owner.contact_id)        except filterlist.DoesNotExist:            log_message = log_message + ' нет фильтров'            filters = []        else:            log_message = log_message + ' найдены фильтры для юзера'            filterlist = [filter.word.lower() for filter in filters]        blacklistword = filterlist + ['запрещенный', 'выиграй', 'удалю', 'читать продолжение', 'joinchat', 'подписаться', 'подписывайся', 'подпишитесь', 'переходи', 'перейти в канал', 'подписываемся', 'дамы и господа', 'автор канала']        message = event.message        message_text = message.message        if message.entities:            for entity in message.entities:                if entity.__class__.__name__ == 'MessageEntityTextUrl':                    message_text = message_text + entity.url        if message.reply_markup:            for row in message.reply_markup.rows:                for button in row.buttons:                    if hasattr(button, 'url'):                        message_text = message_text + button.url                    if hasattr(button, 'text'):                        message_text = message_text + button.text        if any([word in message_text.lower() for word in blacklistword]): # ищем стоп слова во всех текстах и ссылках            log_message = log_message + "Найдены стоп-слова, не пересылаем"            logger.info(log_message)            return        logger.info(log_message)        logger.info("Фильтры прошли, пересылаем из канала %s в поток %s" % (forward.channel_id, forward.feed_id))        await event.message.forward_to(forward.feed_id)

Здесь по id канала (event.message.to_id.channel_id), куда пришло сообщение определяется, в какие потоки его следует переслать, путем выборки из БД через пивотную таблицу Forward. Forward это many-to-many связь между Feed и Channel (структура будет ниже). Проверяется, активен ли поток (я иногда отключаю потоки, чтобы не флудили, когда нет желания ничего читать).

Сообщения фильтруются по стоп-словам во второй части приведенного кода.

Надо будет вытащить список стоп-слов из кода куда-нибудь в конфиг. Как это лучше сделать?

Ну и вообще, много текстовых переменных в коде, нужно прикрутить какой-нибудь i18n со словарями. Что обычно для такого юзают?

Структура БД для хранения подписок

from peewee import *db = SqliteDatabase('telegregator.db')class BaseModel(Model):    class Meta:        database = dbclass Contact(BaseModel):    contact_id = IntegerField(unique=True)    added_at = DateTimeField(default=datetime.datetime.now)class Channel(BaseModel):    channel_id = IntegerField(unique=True)    channel_name = CharField(default='')    channel_title = CharField(default='')class Feed(BaseModel):    feed_id = IntegerField(unique=True)    contact = ForeignKeyField(Contact, backref='contacts')    is_filter = BooleanField(default=True)    is_enable = BooleanField(default=True)    feed_title = CharField(default='')class Forward(BaseModel):    feed = ForeignKeyField(Feed, backref='feeds')    channel = ForeignKeyField(Channel, backref='channels')class Filter(BaseModel):    word = CharField(unique=True)    contact = ForeignKeyField(Contact, backref='contacts')

На этом, пожалуй, всё. Надеюсь, кому-то будет полезно. Хотелось бы обсудить, как вы решали схожие задачи (призываю @aav в комменты)

Я старался везде оставить отладочные сообщения логгера с комментариями, так что код не должен оказаться слишком запутанным. Еще раз ссылка на репу: https://github.com/parotikov/telegregator

Если будут какие-то вопросы, то можно писать в телегу @parotikov. Ну, и, перед работой с внешними сервисами и API, проверяйте их возможности и лимиты, чтобы не быть как я :)