wah.su/stickerbridge
1
0
Fork 0
mirror of https://github.com/wah-su/stickerbridge.git synced 2025-04-05 15:54:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Multiprocessed convertion

Browse source
This commit is contained in:
Dawid Rejowski 2022-09-14 00:33:37 +02:00
parent 5d8e5c4854
commit 77eb829a69
1 changed files with 19 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
stickerbridge/telegram_exporter.py
View file

@ -1,3 +1,4 @@
from multiprocessing import Pool
from typing import List from typing import List
from lottie.importers import importers from lottie.importers import importers
@ -47,6 +48,15 @@ def _convert_animation(data: bytes, width=256, height=0):
return out.getvalue() return out.getvalue()
def _process_sticker(document) -> Sticker:
alt = document.attributes[1].alt
if document.mime_type == 'image/webp':
data, width, height = _convert_image(document.downloaded_data_)
if document.mime_type == 'application/x-tgsticker':
data = _convert_animation(document.downloaded_data_)
return Sticker(data, alt)
class TelegramExporter: class TelegramExporter:
def __init__(self, api_id: int, api_hash: str, bot_token: str, secrets_filename: str): def __init__(self, api_id: int, api_hash: str, bot_token: str, secrets_filename: str):
self.api_id = api_id self.api_id = api_id
@ -65,14 +75,14 @@ class TelegramExporter:
try: try:
sticker_set = await self.client(GetStickerSetRequest(InputStickerSetShortName(short_name=pack_name), hash=0)) sticker_set = await self.client(GetStickerSetRequest(InputStickerSetShortName(short_name=pack_name), hash=0))
except StickersetInvalidError: except StickersetInvalidError:
return result return result # return empty on fail
for sticker_document in sticker_set.documents:
alt = sticker_document.attributes[1].alt downloaded_documents = []
raw_data = await self.client.download_media(sticker_document, file=bytes) for document_data in sticker_set.documents:
if sticker_document.mime_type == 'image/webp': document_data.downloaded_data_ = await self.client.download_media(document_data, file=bytes)
data, width, height = _convert_image(raw_data) downloaded_documents.append(document_data)
if sticker_document.mime_type == 'application/x-tgsticker':
data = _convert_animation(raw_data) pool = Pool()
result.append(Sticker(data, alt)) result = pool.map(_process_sticker, downloaded_documents)
return result return result