Spaces:
Build error
Build error
| # MIT License | |
| # Copyright (c) 2022 Hash Minner | |
| # Permission is hereby granted, free of charge, to any person obtaining a copy | |
| # of this software and associated documentation files (the "Software"), to deal | |
| # in the Software without restriction, including without limitation the rights | |
| # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell | |
| # copies of the Software, and to permit persons to whom the Software is | |
| # furnished to do so, subject to the following conditions: | |
| # The above copyright notice and this permission notice shall be included in all | |
| # copies or substantial portions of the Software. | |
| # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR | |
| # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, | |
| # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE | |
| # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER | |
| # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, | |
| # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE | |
| # SOFTWARE | |
| import math | |
| import time | |
| import logging | |
| logging.basicConfig(level=logging.DEBUG, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s') | |
| logger = logging.getLogger(__name__) | |
| logging.getLogger("pyrogram").setLevel(logging.WARNING) | |
| async def progress_for_pyrogram( | |
| current, | |
| total, | |
| ud_type, | |
| message, | |
| start | |
| ): | |
| now = time.time() | |
| diff = now - start | |
| if round(diff % 10.00) == 0 or current == total: | |
| # if round(current / total * 100, 0) % 5 == 0: | |
| percentage = current * 100 / total | |
| speed = current / diff | |
| elapsed_time = round(diff) * 1000 | |
| time_to_completion = round((total - current) / speed) * 1000 | |
| estimated_total_time = elapsed_time + time_to_completion | |
| elapsed_time = TimeFormatter(milliseconds=elapsed_time) | |
| estimated_total_time = TimeFormatter(milliseconds=estimated_total_time) | |
| progress = "[{0}{1}] \nP: {2}%\n".format( | |
| ''.join(["◾" for _ in range(math.floor(percentage / 5))]), | |
| ''.join(["◽" for _ in range(20 - math.floor(percentage / 5))]), | |
| round(percentage, 2), | |
| ) | |
| tmp = progress + "{0} of {1}\n\nSpeed: {2}/s\n\nETA: {3}\n\n".format( | |
| humanbytes(current), | |
| humanbytes(total), | |
| humanbytes(speed), | |
| # elapsed_time if elapsed_time != '' else "0 s", | |
| estimated_total_time if estimated_total_time != '' else "0 s" | |
| ) | |
| try: | |
| await message.edit(text=f"{ud_type}\n {tmp}") | |
| except Exception as e: | |
| logger.info(f"Error {e}") | |
| return | |
| SIZE_UNITS = ['B', 'KB', 'MB', 'GB', 'TB', 'PB'] | |
| def huanbytes(size_in_bytes) -> str: | |
| if size_in_bytes is None: | |
| return '0B' | |
| index = 0 | |
| while size_in_bytes >= 1024: | |
| size_in_bytes /= 1024 | |
| index += 1 | |
| try: | |
| return f'{round(size_in_bytes, 2)}{SIZE_UNITS[index]}' | |
| except IndexError: | |
| return 'File too large' | |
| def humanbytes(size): | |
| # https://stackoverflow.com/a/49361727/4723940 | |
| # 2**10 = 1024 | |
| if not size: | |
| return "" | |
| power = 2**10 | |
| n = 0 | |
| Dic_powerN = {0: ' ', 1: 'K', 2: 'M', 3: 'G', 4: 'T'} | |
| while size > power: | |
| size /= power | |
| n += 1 | |
| return f"{str(round(size, 2))} {Dic_powerN[n]}B" | |
| def TimeFormatter(milliseconds: int) -> str: | |
| seconds, milliseconds = divmod(milliseconds, 1000) | |
| minutes, seconds = divmod(seconds, 60) | |
| hours, minutes = divmod(minutes, 60) | |
| days, hours = divmod(hours, 24) | |
| tmp = ( | |
| (f"{str(days)}d, " if days else "") | |
| + (f"{str(hours)}h, " if hours else "") | |
| + (f"{str(minutes)}m, " if minutes else "") | |
| + (f"{str(seconds)}s, " if seconds else "") | |
| + (f"{str(milliseconds)}ms, " if milliseconds else "") | |
| ) | |
| return tmp[:-2] | |