Предпосылки, или «куда делось всё место на телефоне»
Одним прекрасным днем, обнаружив, что на новом свежем купленном айфоне из 256гб осталось примерно половина места после переноса данных со старого, я задался вопросом — а куда делось место?
Потребление по категориям расставило точки над i — 70гб было занято приложением Photos. Приложение Photos обрадовало наличием 15 000 объектов в галерее.
Быстро проскроллив часть галереи, я пришел к трем фактам:
— Большинство объектов — сохраненные откуда‑то мемы;
— У меня нет желания их удалять;
— У меня нет желания платить за облачное хранилище.
Спустя недолгое время раздумий, было найдено решение, позволяющее сохранить бесценные богатства и повеселить друзей — постить мемы в телеграм.
Поиск технического решения
Обозначим вводные:
15 000 файлов для ручной модерации;
Файлы находятся на телефоне;
Постить необходимо 1 мем раз в N часов, а не 100 постов за минуту, иначе никто не будет смотреть на этот спам;
Мемы должны поститься, даже если я нахожусь в самолете без интернета.
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())Пробуем запустить:
Клиент запустился;
Спросил авторизацию (опционально, первый запуск на устройстве);
Получил список чатов/каналов для аккаунта;
Зашел в чат/канал под названием
secret channel for memes;Получил список сообщений, в которых содержится изображение;
Для каждого из сообщений:
Сохранили картинку по пути
/some/directory/to/save/pictures/meme_posting_N.jpg(N всегда увеличивается инкрементом на 1 на стороне телеграма);Удалили оригинальное сообщение.
Добавляем постинг (отложенных) сообщений
...
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 изображение;
Репост текста из сообщений — иногда мемы бывают сложные;
Поддержка видеофайлов и анимаций — лично не любитель видео, но вроде как людям нравятся.
