Upload 4 files

unzipper/helpers_nexa/__init__.py

@@ -0,0 +1,11 @@
+# ===================================================================== #
+#                      Copyright (c) 2022 Itz-fork                      #
+#                                                                       #
+# This program is distributed in the hope that it will be useful,       #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of        #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.                  #
+# See the GNU General Public License for more details.                  #
+#                                                                       #
+# You should have received a copy of the GNU General Public License     #
+# along with this program. If not, see <http://www.gnu.org/licenses/>   #
+# ===================================================================== #

unzipper/helpers_nexa/buttons.py

@@ -0,0 +1,153 @@
+# ===================================================================== #
+#                      Copyright (c) 2022 Itz-fork                      #
+#                                                                       #
+# This program is distributed in the hope that it will be useful,       #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of        #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.                  #
+# See the GNU General Public License for more details.                  #
+#                                                                       #
+# You should have received a copy of the GNU General Public License     #
+# along with this program. If not, see <http://www.gnu.org/licenses/>   #
+# ===================================================================== #
+
+from os.path import basename
+from pykeyboard import InlineKeyboard
+from unzipper.client.caching import STRINGS
+from unzipper.helpers_nexa.utils import read_json_sync
+from pyrogram.types import InlineKeyboardMarkup, InlineKeyboardButton
+
+
+class Unzipper_Buttons:
+    def __init__(self) -> None:
+        pass
+
+    async def make_button(self, text: str, *args, **kwargs):
+        """
+        Create pyrogram InlineKeyboardMarkup object with 1 button
+        """
+        return InlineKeyboardMarkup([
+            [InlineKeyboardButton(text, *args, **kwargs)]
+        ])
+
+    async def make_files_keyboard(self, files: list, user_id: int, chat_id: int, inlude_files: bool = True):
+        i_kbd = InlineKeyboard(row_width=2)
+        data = [InlineKeyboardButton(STRINGS["buttons"]["upload_all"], f"ext_a|{user_id}|{chat_id}"), InlineKeyboardButton(
+            STRINGS["buttons"]["cancel"], "cancel_dis")]
+        if inlude_files:
+            for num, file in enumerate(files):
+                # Stop iterating if file count is 90
+                if num >= 90:
+                    break
+                data.append(
+                    InlineKeyboardButton(f"{num} - {basename(file)}".encode(
+                        "utf-8").decode("utf-8"), f"ext_f|{user_id}|{chat_id}|{num}")
+                )
+        i_kbd.add(*data)
+        return i_kbd
+
+    START = InlineKeyboardMarkup([[
+        InlineKeyboardButton(
+            STRINGS["buttons"]["help"], callback_data="helpcallback"),
+        InlineKeyboardButton(
+            STRINGS["buttons"]["about"], callback_data="aboutcallback")
+    ]])
+
+    HELP = InlineKeyboardMarkup([
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["help_extract"], callback_data="extracthelp"),
+            InlineKeyboardButton(
+                STRINGS["buttons"]["help_upload"], callback_data="upmodhelp")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["help_thumbnail"], callback_data="thumbhelp"),
+            InlineKeyboardButton(
+                STRINGS["buttons"]["help_backup"], callback_data="backuphelp")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["help_langs"], callback_data="langhelp")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["back"], callback_data="megoinhome")
+        ]
+    ])
+
+    HELP_BACK = InlineKeyboardMarkup(
+        [[InlineKeyboardButton(STRINGS["buttons"]["back_to_help_menu"], callback_data="helpcallback")]])
+
+    EXTRACT_FILE = InlineKeyboardMarkup([
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["extract_file"], callback_data="extract_file|tg_file|no_pass")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["extract_file_pass"], callback_data="extract_file|tg_file|with_pass")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["cancel"], callback_data="cancel_dis")
+        ]
+    ])
+
+    EXTRACT_URL = InlineKeyboardMarkup([
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["extract_url"], callback_data="extract_file|url|no_pass"),
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["extract_url_pass"], callback_data="extract_file|url|with_pass")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["cancel"], callback_data="cancel_dis")
+        ]
+    ])
+
+    CLEAN = InlineKeyboardMarkup([
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["clean"], callback_data="cancel_dis")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["no_cancel"], callback_data="nobully")
+        ]
+    ])
+
+    BACKUP = InlineKeyboardMarkup(
+        [[InlineKeyboardButton("Gofile.io", callback_data="cloudbackup|gofile"), ]])
+
+    SETTINGS_GOFILE = InlineKeyboardMarkup([
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["gofile_set"], callback_data="gf_setting-set"),
+            InlineKeyboardButton(
+                STRINGS["buttons"]["gofile_del"], callback_data="gf_setting-del")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["gofile_get"], callback_data="gf_setting-get")
+        ]
+    ])
+
+    UPLOAD_MODE = InlineKeyboardMarkup([
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["as_doc"], callback_data="set_mode|doc")
+        ],
+        [
+            InlineKeyboardButton(
+                STRINGS["buttons"]["as_vid"], callback_data="set_mode|video")
+        ]
+    ])
+
+    LANGUAGES = InlineKeyboardMarkup([[InlineKeyboardButton(
+        v, f"set_lang|{k}")] for k, v in read_json_sync("unzipper/localization/languages.json", True)])
+
+    BACK = InlineKeyboardMarkup(
+        [[InlineKeyboardButton(STRINGS["buttons"]["back"], callback_data="megoinhome")]])

unzipper/helpers_nexa/checks.py

@@ -0,0 +1,38 @@
+# ===================================================================== #
+#                      Copyright (c) 2022 Itz-fork                      #
+#                                                                       #
+# This program is distributed in the hope that it will be useful,       #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of        #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.                  #
+# See the GNU General Public License for more details.                  #
+#                                                                       #
+# You should have received a copy of the GNU General Public License     #
+# along with this program. If not, see <http://www.gnu.org/licenses/>   #
+# ===================================================================== #
+
+
+import logging
+from pyrogram import enums
+from unzipper import unzip_client
+from config import Config
+
+
+def check_log_channel():
+    try:
+        if Config.LOGS_CHANNEL:
+            c_info = unzip_client.get_chat(chat_id=Config.LOGS_CHANNEL)
+            if c_info.type != enums.ChatType.CHANNEL:
+                logging.warn("Chat is not a channel!")
+                exit()
+            elif c_info.username is not None:
+                logging.warn("Channel is not private!")
+                exit()
+            else:
+                unzip_client.send_message(
+                    chat_id=Config.LOGS_CHANNEL, text="`Unzipper-Bot has Successfully Started!` \n\n**Powered by @NexaBotsUpdates**")
+        else:
+            logging.warn("No Log Channel ID is Given! Imma leaving Now!")
+            exit()
+    except Exception as e:
+        logging.warn("Error Happend while checking Log Channel! Make sure you're not dumb enough to provide a wrong Log Channel ID!")
+        logging.warn(f"Error: \n{e}")

unzipper/helpers_nexa/utils.py

@@ -0,0 +1,152 @@
+# ===================================================================== #
+#                      Copyright (c) 2022 Itz-fork                      #
+#                                                                       #
+# This program is distributed in the hope that it will be useful,       #
+# but WITHOUT ANY WARRANTY; without even the implied warranty of        #
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.                  #
+# See the GNU General Public License for more details.                  #
+#                                                                       #
+# You should have received a copy of the GNU General Public License     #
+# along with this program. If not, see <http://www.gnu.org/licenses/>   #
+# ===================================================================== #
+
+# Credits: SpEcHiDe's AnyDL-Bot for progress_for_pyrogram, humanbytes and TimeFormatter
+
+from re import sub
+from time import time
+from math import floor
+from json import loads
+from os import path, walk
+from functools import partial
+from subprocess import Popen, PIPE
+from asyncio import get_running_loop
+
+
+async def progress_for_pyrogram(current, total, ud_type, message, start):
+    now = time()
+    diff = now - start
+    speed = current / diff
+    if total:
+        if round(diff % 10.00) == 0 or current == total:
+            percentage = current * 100 / total
+            elapsed_time = round(diff) * 1000
+            estimated_total_time = elapsed_time + \
+                round((total - current) / speed) * 1000
+
+            elapsed_time = TimeFormatter(elapsed_time)
+            estimated_total_time = TimeFormatter(estimated_total_time)
+
+            progress = "[{0}{1}] \n**📊 Progress**: {2}%\n".format(
+                # Filled
+                ''.join(["◉" for i in range(floor(percentage / 5))]),
+                # Empty
+                ''.join(["◎" for i in range(20 - floor(percentage / 5))]),
+                round(percentage, 2))
+
+            tmp = progress + "{0} of {1}\n**🏃 Speed:** {2}/s\n**⏰ ETA:** {3}\n".format(
+                humanbytes(current),
+                humanbytes(total),
+                humanbytes(speed),
+                estimated_total_time if estimated_total_time != '' else "0 s"
+            )
+            try:
+                await message.edit("{}\n {} \n\n**Powered by @NexaBotsUpdates**".format(ud_type, tmp))
+            except:
+                pass
+    else:
+        tmp = "**📊 Progress:** {0} of {1}\n**🏃 Speed:** {2}/s\n**⏰ ETA:** {3}\n".format(
+            humanbytes(current),
+            "?",
+            humanbytes(speed),
+            "unknown"
+        )
+        try:
+            await message.edit("{}\n {} \n\n**Powered by @NexaBotsUpdates**".format(ud_type, tmp))
+        except:
+            pass
+
+
+def humanbytes(size: int):
+    if not size:
+        return "N/A"
+    power = 2**10
+    n = 0
+    Dic_powerN = {0: " ", 1: "Ki", 2: "Mi", 3: "Gi", 4: "Ti"}
+    while size > power:
+        size /= power
+        n += 1
+    return f"{round(size, 2)} {Dic_powerN[n]}B"
+
+
+def TimeFormatter(milliseconds: int) -> str:
+    seconds, milliseconds = divmod(int(milliseconds), 1000)
+    minutes, seconds = divmod(seconds, 60)
+    hours, minutes = divmod(minutes, 60)
+    days, hours = divmod(hours, 24)
+    tmp = ((str(days) + "d, ") if days else "") + \
+        ((str(hours) + "h, ") if hours else "") + \
+        ((str(minutes) + "m, ") if minutes else "") + \
+        ((str(seconds) + "s, ") if seconds else "") + \
+        ((str(milliseconds) + "ms, ") if milliseconds else "")
+    return tmp[:-2] if tmp[:-2] else "0ms"
+
+
+def run_shell_cmds(command):
+    """
+    Execute shell commands and returns the output
+    """
+    run = Popen(command, stdout=PIPE,
+                stderr=PIPE, shell=True)
+    shell_ouput = run.stdout.read()[:-1].decode("utf-8")
+    return shell_ouput
+
+
+async def run_cmds_on_cr(func, *args, **kwargs):
+    """
+    Execute blocking functions asynchronously
+    """
+    loop = get_running_loop()
+    return await loop.run_in_executor(
+        None,
+        partial(func, *args, **kwargs)
+    )
+
+
+async def get_files(fpath: str, filter_fn=None):
+    """
+    Returns files in a folder
+
+    Parameters:
+
+        - `fpath` - Path to the folder
+        - `filter_fn` - Function to filter elements in the array
+    """
+    path_list = [val for sublist in [
+        [path.join(i[0], j) for j in i[2]] for i in walk(fpath)] for val in sublist]
+    if filter_fn:
+        path_list = list(filter(filter_fn, path_list))
+    return sorted(path_list)
+
+
+def read_json_sync(name: str, as_items: bool = False) -> dict:
+    """
+    Reads json file and returns a dict
+
+    Parameters:
+
+        - `name` - File path
+        - `as_items` - Pass "True" if you want to return items of the dict
+    """
+    with open(name) as fs:
+        return loads(fs.read()).items() if as_items else loads(fs.read())
+
+
+async def rm_mark_chars(text: str):
+    """
+    Remove basic markdown characters
+
+    Parameters:
+
+        - `text` - Text
+    """
+    return sub("[*`_]", "", text)