Автоматизируем постинг мемов в Telegram без расходов на аренду сервера

Предпосылки, или «куда делось всё место на телефоне»

Одним прекрасным днем, обнаружив, что на новом свежем купленном айфоне из 256гб осталось примерно половина места после переноса данных со старого, я задался вопросом — а куда делось место?
Потребление по категориям расставило точки над i — 70гб было занято приложением Photos. Приложение Photos обрадовало наличием 15 000 объектов в галерее.

Быстро проскроллив часть галереи, я пришел к трем фактам:
— Большинство объектов — сохраненные откуда‑то мемы;
— У меня нет желания их удалять;
— У меня нет желания платить за облачное хранилище.

Спустя недолгое время раздумий, было найдено решение, позволяющее сохранить бесценные богатства и повеселить друзей — постить мемы в телеграм.

Поиск технического решения

Обозначим вводные:

1. 15 000 файлов для ручной модерации;

2. Файлы находятся на телефоне;

3. Постить необходимо 1 мем раз в N часов, а не 100 постов за минуту, иначе никто не будет смотреть на этот спам;

4. Мемы должны поститься, даже если я нахожусь в самолете без интернета.

1 и 2 пункты решаются тем, что постепенно листая галерею, я отправляю понравившиеся мне мемы в специально созданный закрытый чат в телеграме с помощью кнопки «поделиться». Работает быстро, работает отлично, ошибок быть не может.

3 пункт можно решить с помощью бота для телеграма.
Алгоритм простой — раз в N времени бот заходит в закрытый чат, забирает файл из сообщения, постит его в канал, удаляет оригинальное сообщение.

4 пункт можно решить с помощью запуска бота на VPS.
У меня уже есть два VPS, заходить на них и ставить патчи безопасности — то еще удовольствие, а тут на горизонте маячит третий. К тому же, тратить деньги на это хобби у меня не было в планах.
«Придумаем что‑то после», подумал я, и приступил к чтению документации.

Реализация бота

Первой остановкой была попытка написать бота, ведь не зря на хабре каждую неделю появляются туториалы вида «Пишем телеграм бота за пять минут»?

Зарегистрировав бота, я приступил к экспериментам. Эксперименты, к сожалению, продлились не долго, ведь у телеграм ботов есть существенные ограничения для моей цели:

• Не может получить сообщения, которые были отправлены в то время, когда бот был офлайн;

• Не может отправлять отложенные сообщения.

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

Уже почти что опустив руки и потеряв свою мечту, я внезапно вспомнил, что телеграм предлагает апи не только для ботов, но еще и для клиентов. И раз можно создать свой собственный клиент, значит, можно и воспользоваться отложенными сообщениями — как раз то, что нужно для zero‑cost решения.

Реализация клиента

Проще сказать, чем сделать (на самом деле сделать тоже довольно просто), но уровень чуть‑чуть повыше, чем телеграм боты с курсов, которыми завален хабр.

Для реализации идеи возьмем первую попавшуюся популярную библиотеку для телеграма на питоне — https://github.com/LonamiWebs/Telethon

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

Получаем картинки из секретного чятика

import asyncio from telethon import TelegramClient from telethon.tl.types import InputMessagesFilterPhotos  api_id = 12345678 api_hash = 'some_md5_hash' MEMES_DIRECTORY = "/some/directory/to/save/pictures" _CHANNEL_SOURCE_NAME = "secret channel for memes"  async def download_planned_messages_images(client):   messages = await client.get_messages(_CHANNEL_SOURCE_NAME, 0, filter=InputMessagesFilterPhotos)   for message in messages:     try:       filename = f"{MEMES_DIRECTORY}/meme_posting_{message.id}.jpg"       await message.download_media(file=filename)       print(f"downloaded image from message={message.id}")       try:         await client.delete_messages(_CHANNEL_SOURCE_NAME, message_ids=[message.id])         print(f"removed message={message.id}")       except Exception as e:         print(f"cannot remove post with downloaded media for postid={message.id}; exception={e}")     except Exception as e:       print(f"cannot download media for postid={message.id}; exception={e}")  async def main():   # setup   client = TelegramClient('session_meme', api_id, api_hash)   await client.start()   await client.get_dialogs()  # load all dialogs, otherwise GET_MESSAGES won't work      await download_planned_messages_images(client)  if __name__ == "__main__":   asyncio.run(main())

Пробуем запустить:

1. Клиент запустился;

2. Спросил авторизацию (опционально, первый запуск на устройстве);

3. Получил список чатов/каналов для аккаунта;

4. Зашел в чат/канал под названием secret channel for memes;

5. Получил список сообщений, в которых содержится изображение;

6. Для каждого из сообщений:

   1. Сохранили картинку по пути /some/directory/to/save/pictures/meme_posting_N.jpg (N всегда увеличивается инкрементом на 1 на стороне телеграма);

   2. Удалили оригинальное сообщение.

Добавляем постинг (отложенных) сообщений

... from telethon import TelegramClient, functions  _CHANNEL_NAME = "memes_from_22_century" ...  def get_all_files() -> [str]:   return [f"{MEMES_DIRECTORY}/{file}" for file in os.listdir(MEMES_DIRECTORY) if not file.startswith(".")]  async def post_message(client, file_path: str, date: datetime.datetime) -> bool:   try:     await client.send_message(_CHANNEL_NAME, silent=True, file=file_path, schedule=date, link_preview=False)     print("posted message")     return True   except Exception as e:     print(f"failed posting: {e}; file={file_path}; date={date}")     return False  async def main():   ...   # post     for file in get_all_files():     success = await post_message(client, file, datetime.datetime.now())      if success:         try:           os.remove(file)         except Exception as e:           print(f"failed to remove file={file}, exception={e}")

Запускаем скрипт, радуемся, что теперь сообщения постятся, а исходные картинки удаляются с диска — никаких дубликатов!

Вносим разнообразие и постим действительно отложенные сообщения

Следующим этапом чуть‑чуть присыпем сверху сахаром наши отправляемые сообщения — добавим рандомный текст, перемешаем мемы, сделаем правильное время.

import asyncio import datetime import os import random  import pytz  from telethon import TelegramClient, functions from telethon.tl.types import InputMessagesFilterPhotos  _CHANNEL = "-12345" _CHANNEL_NAME = "memes_from_22_century" api_id = 12345678 api_hash = 'some_md5_hash' MEMES_DIRECTORY = "/some/directory/to/save/pictures" _CHANNEL_SOURCE_NAME = "secret channel for memes" emojis = ["?", "?", "?", "?"]   class GetPostingHour:     POSTING_TIMES = [6, 9, 12, 15]  # utc, hour; minutes always 30      def __init__(self, last_date: datetime.datetime):         self._start_date: datetime.datetime = last_date         self._processed_start_date = False         self._current_date: datetime.datetime = self._start_date      def next_date(self) -> datetime.datetime:         year = self._current_date.year         month = self._current_date.month         day = self._current_date.day          # hour >= max available hour in the day         if self._current_date.hour >= max(self.POSTING_TIMES):             _next_day_datetime = self._current_date + datetime.timedelta(days=1)             year = _next_day_datetime.year             month = _next_day_datetime.month             day = _next_day_datetime.day             hour = min(self.POSTING_TIMES)             self._processed_start_date = True         # hour < max available hour         elif self._processed_start_date:             hour = self.POSTING_TIMES[self.POSTING_TIMES.index(self._current_date.hour) + 1]         # hour < max available hour AND is being processed first time         else:             # take closest available value             _closest_hour = min(self.POSTING_TIMES, key=lambda x: abs(x - self._current_date.hour))             # check if its less than current, if so - replace with next value             if _closest_hour <= self._current_date.hour:                 hour = self.POSTING_TIMES[self.POSTING_TIMES.index(_closest_hour) + 1]             else:                 hour = _closest_hour              self._processed_start_date = True          next_date = datetime.datetime(year=year, month=month, day=day, hour=hour, minute=30, tzinfo=pytz.UTC)          self._current_date = next_date         print(next_date)         return next_date   def get_all_files() -> [str]:     return [f"{MEMES_DIRECTORY}/{file}" for file in os.listdir(MEMES_DIRECTORY) if not file.startswith(".")]   async def download_all_messages_images(client, initial_images_count: int):     photos = await client.get_messages(_CHANNEL_SOURCE_NAME, 0, filter=InputMessagesFilterPhotos)     limit = min(photos.total, 100 - initial_images_count)  # download no more than 100 images per run; no more than 100 in folder      for message in await client.get_messages(_CHANNEL_SOURCE_NAME, limit, filter=InputMessagesFilterPhotos):         try:             filename = f"{MEMES_DIRECTORY}/meme_posting_{message.id}.jpg"             await message.download_media(file=filename)             print(f"downloaded image from message={message.id}")             try:                 await client.delete_messages(_CHANNEL_SOURCE_NAME, message_ids=[message.id])                 print(f"removed message={message.id}")             except Exception as e:                 print(f"cannot remove post with downloaded media for postid={message.id}; exception={e}")         except Exception as e:             print(f"cannot download media for postid={message.id}; exception={e}")   async def get_all_scheduled_messages(client):     result = await client(functions.messages.GetScheduledHistoryRequest(peer=int(_CHANNEL), hash=int(_CHANNEL)))     return result.messages   async def get_last_scheduled_message_datetime(messages: list):     return messages[0].date   async def post_message(client, file_path: str, date: datetime.datetime) -> bool:     try:         message = f"[Я пощу мемы {random.choice(emojis)}](https://t.me/memes_from_22_century)"         await client.send_message(_CHANNEL_NAME, message=message, silent=True, file=file_path, schedule=date, link_preview=False)         print("posted message")         return True     except Exception as e:         print(f"failed posting: {e}; file={file_path}; date={date}")         return False   async def main():     # setup     client = TelegramClient('session_meme', api_id, api_hash)     await client.start()     await client.get_dialogs()  # load all dialogs, otherwise GET_MESSAGES won't work      # check images     scheduled_messages = await get_all_scheduled_messages(client)     initial_images_count = len(get_all_files()) + len(scheduled_messages)     await download_all_messages_images(client, initial_images_count)     all_files = get_all_files()     random.shuffle(all_files)     if not all_files:         print(f"no files to upload, check directory={MEMES_DIRECTORY}")         return      # set time     last_date = await get_last_scheduled_message_datetime(scheduled_messages)     hours_counter = GetPostingHour(last_date)      # post     success = False     hour = hours_counter.next_date()      for file in all_files:         if success:             hour = hours_counter.next_date()          success = await post_message(client, file, hour)          if success:             try:                 os.remove(file)             except Exception as e:                 print(f"failed to remove file={file}, exception={e}")   if __name__ == "__main__":     asyncio.run(main())

Теперь наш скрипт при запуске умеет:

• Самостоятельно определять время для следующего поста на основе последнего scheduled message;

• Постит до 100 scheduled messages, сначала используя файлы с диска, потом используя файлы из чата‑прослойки. Ограничение в 100 сообщений есть со стороны телеграма;

• Разбавляет нескучные мемы не менее нескучными эмодзи.

Получившееся решение позволяет с одного запуска запланировать постинг мемов на 25 дней вперед (4 картинки в день), будучи полностью офлайн.
Вполне уместная самореклама — https://t.me/memes_from_22_century

Что можно улучшить

Точки роста:

• crontab — чтобы никогда не запускать скрипт, а только кидать мемы в специальный закрытый чат;

• Использование ссылок на файлы вместо скачивания файлов — телеграм позволяет использовать медиа из другого сообщения, просто указав ссылку на него из сообщения;

• Множество изображений в одном сообщении — нынешняя реализация предлагает только 1 сообщение = 1 изображение;

• Репост текста из сообщений — иногда мемы бывают сложные;

• Поддержка видеофайлов и анимаций — лично не любитель видео, но вроде как людям нравятся.