import asyncio from collections import defaultdict import os import zipfile from PIL import Image from telethon.tl.functions.messages import GetAllStickersRequest from telethon.tl.functions.messages import GetStickerSetRequest from telethon.errors import MessageNotModifiedError from telethon.errors.rpcerrorlist import StickersetInvalidError from pagermaid import working_dir, version from telethon.tl.types import ( DocumentAttributeSticker, InputStickerSetID, ) from pagermaid.listener import listener from pagermaid.utils import alias_command lottie_import = True try: from lottie.exporters.gif import export_gif from lottie.importers.core import import_tgs except ImportError: lottie_import = False @listener(is_plugin=True, outgoing=True, command=alias_command("getstickers"), description="获取整个贴纸包的贴纸，false 关闭 tgs 转 gif；任意值开启下载所有贴纸包；\n" "转 gif 需要手动安装 pypi 依赖 lottie[gif] \n" "Ubuntu 转换出错请先运行 `-sh apt install libpangocairo-1.0-0 -y`。", parameters="<任意值>") async def getstickers(context): tgs_gif = True if not os.path.isdir('data/sticker/'): os.makedirs('data/sticker/') if len(context.parameter) == 1 or len(context.parameter) == 2: if "false" in context.arguments: tgs_gif = False if "all" in context.arguments: sticker_sets = await context.client(GetAllStickersRequest(0)) for stickerset in sticker_sets.sets: file_ext_ns_ion = "webp" wdnmd = InputStickerSetID(id=stickerset.id, access_hash=stickerset.access_hash) sticker_set = await context.client(GetStickerSetRequest(wdnmd)) pack_file = os.path.join('data/sticker/', sticker_set.set.short_name, "pack.txt") if os.path.isfile(pack_file): os.remove(pack_file) # Sticker emojis emojis = defaultdict(str) for pack in sticker_set.packs: for document_id in pack.documents: emojis[document_id] += pack.emoticon async def download(sticker, emojis, path, file): await context.client.download_media(sticker, file=os.path.join(path, file)) with open(pack_file, "a") as f: f.write(f"{{'image_file': '{file}','emojis':{emojis[sticker.id]}}},") if file_ext_ns_ion == 'tgs' and lottie_import and tgs_gif: animated = import_tgs(os.path.join(path, file)) export_gif(animated, os.path.join(path, file)[:-3] + 'gif') elif file_ext_ns_ion == 'webp': convert_png(os.path.join(path, file)) pending_tasks = [ asyncio.ensure_future( download(document, emojis, 'data/sticker/' + sticker_set.set.short_name, f"{i:03d}.{file_ext_ns_ion}") ) for i, document in enumerate(sticker_set.documents) ] xx = await context.client.send_message(context.chat_id, f"正在下载 {sticker_set.set.short_name} " f"中的 {sticker_set.set.count} 张贴纸。。。") num_tasks = len(pending_tasks) while 1: done, pending_tasks = await asyncio.wait(pending_tasks, timeout=2.5, return_when=asyncio.FIRST_COMPLETED) if file_ext_ns_ion == 'tgs' and lottie_import and tgs_gif: try: await xx.edit( f"正在下载/转换中，进度： {num_tasks - len(pending_tasks)}/{sticker_set.set.count}") except MessageNotModifiedError: pass if not pending_tasks: break await xx.edit("下载完毕，打包上传中。") directory_name = sticker_set.set.short_name os.chdir("data/sticker/") # 修改当前工作目录 zipf = zipfile.ZipFile(directory_name + ".zip", "w", zipfile.ZIP_DEFLATED) zipdir(directory_name, zipf) zipf.close() await context.client.send_file( context.chat_id, directory_name + ".zip", caption=sticker_set.set.short_name, force_document=True, allow_cache=False ) try: os.remove(directory_name + ".zip") os.remove(directory_name) except: pass os.chdir(working_dir) await xx.delete() return reply_message = await context.get_reply_message() if reply_message: if not reply_message.sticker: await context.edit("请回复一张贴纸。") return else: await context.edit("请回复一张贴纸。") return sticker = reply_message.sticker sticker_attrib = find_instance(sticker.attributes, DocumentAttributeSticker) if not sticker_attrib.stickerset: await context.edit("回复的贴纸不属于任何贴纸包。") return is_a_s = is_it_animated_sticker(reply_message) file_ext_ns_ion = "webp" if is_a_s: file_ext_ns_ion = "tgs" if tgs_gif and not lottie_import: await context.reply('`lottie[gif]` 依赖未安装，tgs 无法转换为 gif ，进行标准格式导出。') try: sticker_set = await context.client(GetStickerSetRequest(sticker_attrib.stickerset)) except StickersetInvalidError: await context.edit('回复的贴纸不存在于任何贴纸包中。') return pack_file = os.path.join('data/sticker/', sticker_set.set.short_name, "pack.txt") if os.path.isfile(pack_file): os.remove(pack_file) # Sticker emojis emojis = defaultdict(str) for pack in sticker_set.packs: for document_id in pack.documents: emojis[document_id] += pack.emoticon async def download(sticker, emojis, path, file): await context.client.download_media(sticker, file=os.path.join(path, file)) with open(pack_file, "a") as f: f.write(f"{{'image_file': '{file}','emojis':{emojis[sticker.id]}}},") if file_ext_ns_ion == 'tgs' and lottie_import and tgs_gif: animated = import_tgs(os.path.join(path, file)) export_gif(animated, os.path.join(path, file)[:-3] + 'gif') elif file_ext_ns_ion == 'webp': convert_png(os.path.join(path, file)) pending_tasks = [ asyncio.ensure_future( download(document, emojis, 'data/sticker/' + sticker_set.set.short_name, f"{i:03d}.{file_ext_ns_ion}") ) for i, document in enumerate(sticker_set.documents) ] await context.edit( f"正在下载 {sticker_set.set.short_name} 中的 {sticker_set.set.count} 张贴纸。。。") num_tasks = len(pending_tasks) while 1: done, pending_tasks = await asyncio.wait(pending_tasks, timeout=2.5, return_when=asyncio.FIRST_COMPLETED) if file_ext_ns_ion == 'tgs' and lottie_import and tgs_gif: try: await context.edit( f"正在下载/转换中，进度： {num_tasks - len(pending_tasks)}/{sticker_set.set.count}") except MessageNotModifiedError: pass if not pending_tasks: break await context.edit("下载完毕，打包上传中。") directory_name = sticker_set.set.short_name os.chdir("data/sticker/") # 修改当前工作目录 zipf = zipfile.ZipFile(directory_name + ".zip", "w", zipfile.ZIP_DEFLATED) zipdir(directory_name, zipf) zipf.close() await context.client.send_file( context.chat_id, directory_name + ".zip", caption=sticker_set.set.short_name, force_document=True, allow_cache=False, reply_to=reply_message.id ) try: os.remove(directory_name + ".zip") os.remove(directory_name) except: pass os.chdir(working_dir) await context.delete() def find_instance(items, class_or_tuple): for item in items: if isinstance(item, class_or_tuple): return item return None def is_it_animated_sticker(message): try: if message.media and message.media.document: mime_type = message.media.document.mime_type if "tgsticker" in mime_type: return True else: return False else: return False except: return False def zipdir(path, ziph): for root, dirs, files in os.walk(path): for file in files: ziph.write(os.path.join(root, file)) os.remove(os.path.join(root, file)) def convert_png(path): im = Image.open(path) im = im.convert('RGBA') new_path = path.replace(".webp", ".png") im.save(new_path, 'PNG') return new_path