getstickers 获取整个贴纸包的贴纸

getstickers/main.py

@@ -0,0 +1,90 @@
+import asyncio
+import shutil
+from collections import defaultdict
+import os
+import zipfile
+
+from pyrogram.raw.functions.messages import GetStickerSet
+from pyrogram.raw.types import InputStickerSetShortName
+from pyrogram.raw.types.messages import StickerSet
+from pyrogram.types import Document
+
+from pagermaid import working_dir
+from pagermaid.enums import Message, Client
+from pagermaid.listener import listener
+from pagermaid.single_utils import safe_remove
+
+
+@listener(command="getstickers",
+          description="获取整个贴纸包的贴纸")
+async def get_stickers(bot: Client, message: Message):
+    if not os.path.isdir('data/sticker/'):
+        os.makedirs('data/sticker/')
+    if message.reply_to_message:
+        if not message.reply_to_message.sticker:
+            return await message.edit("请回复一张贴纸。")
+    else:
+        return await message.edit("请回复一张贴纸。")
+    sticker = message.reply_to_message.sticker
+    if not sticker.set_name:
+        return await message.edit("回复的贴纸不属于任何贴纸包。")
+    try:
+        sticker_set: StickerSet = await bot.invoke(
+            GetStickerSet(stickerset=InputStickerSetShortName(short_name=sticker.set_name), hash=0))
+    except Exception:
+        return await message.edit('回复的贴纸不存在于任何贴纸包中。')
+    pack_file = os.path.join('data/sticker/', sticker_set.set.short_name, "pack.txt")
+
+    if os.path.isfile(pack_file):
+        os.remove(pack_file)
+    emojis = defaultdict(str)
+    for pack in sticker_set.packs:
+        for document_id in pack.documents:
+            emojis[document_id] += pack.emoticon
+    file_ext_ns_ion = "webp"
+    if sticker.is_video:
+        file_ext_ns_ion = "mp4"
+    elif sticker.is_animated:
+        file_ext_ns_ion = "tgs"
+
+    async def download(sticker_, emojis_, path, file):
+        sticker_file = Document._parse(bot, sticker_, "sticker.webp")  # noqa
+        await bot.download_media(sticker_file.file_id, file_name=os.path.join(path, file))
+
+        with open(pack_file, "a") as f:
+            f.write(f"{{'image_file': '{file}','emojis':{emojis_[sticker_.id]}}},")
+
+    pending_tasks = [asyncio.ensure_future(
+        download(document, emojis, f"data/sticker/{sticker_set.set.short_name}", f"{i:03d}.{file_ext_ns_ion}")
+    ) for i, document in enumerate(sticker_set.documents)]
+
+    message: Message = await message.edit(f"正在下载 {sticker_set.set.short_name} 中的 {sticker_set.set.count} 张贴纸。。。")
+
+    while 1:
+        done, pending_tasks = await asyncio.wait(pending_tasks, timeout=2.5, return_when=asyncio.FIRST_COMPLETED)
+        if not pending_tasks:
+            break
+
+    await message.edit("下载完毕，打包上传中。")
+    directory_name = sticker_set.set.short_name
+    os.chdir("data/sticker/")
+    zipf = zipfile.ZipFile(f"{directory_name}.zip", "w", zipfile.ZIP_DEFLATED)
+    zipdir(directory_name, zipf)
+    zipf.close()
+    await bot.send_document(
+        message.chat.id,
+        f"{directory_name}.zip",
+        caption=sticker_set.set.short_name,
+        reply_to_message_id=message.reply_to_message_id
+    )
+    safe_remove(f"{directory_name}.zip")
+    shutil.rmtree(directory_name)
+    os.chdir(working_dir)
+    await message.safe_delete()
+
+
+def zipdir(path, zip_):
+    for root, dirs, files in os.walk(path):
+        for file in files:
+            zip_.write(os.path.join(root, file))
+            os.remove(os.path.join(root, file))