Предпосылки, или «куда делось всё место на телефоне»
Одним прекрасным днем, обнаружив, что на новом свежем купленном айфоне из 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 изображение;
-
Репост текста из сообщений — иногда мемы бывают сложные;
-
Поддержка видеофайлов и анимаций — лично не любитель видео, но вроде как людям нравятся.
ссылка на оригинал статьи https://habr.com/ru/articles/849158/
Добавить комментарий