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

Одним прекрасным днем, обнаружив, что на новом свежем купленном айфоне из 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 изображение;

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

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