Release 0.1.4.1

alphacoders_downloader/__init__.py

@@ -1,4 +1,4 @@
 __title__ = "AlphacodersDownloader"
-__version__ = "0.1.4.1"
+__version__ = "0.1.4.2"
 __author__ = "Asthowen"
 __license__ = "GNU v3.0"

alphacoders_downloader/alphacoders_downloader.py

@@ -23,8 +23,8 @@ def __init__(self, url: str, path: str, client_session: aiohttp.ClientSession):
         self.path = path if path[-1] == os.sep else path + os.sep
         self.client_session = client_session
 
-        self.temp_path = self.path + 'temp' + os.sep
-        self.page_char = ''
+        self.temp_path = self.path + "temp" + os.sep
+        self.page_char = ""
         self.progress_bar: Union[ProgressBar, None] = None
         self.spinner: Spinner = Spinner()
         self.temp_images_list = []
@@ -35,50 +35,59 @@ def __init__(self, url: str, path: str, client_session: aiohttp.ClientSession):
         self.total_size_downloaded = 0
 
     async def fetch_images(self, image_url: str):
-        images_link_list_split = image_url.split('/')
+        images_link_list_split = image_url.split("/")
         images_name_file_thumb = images_link_list_split[len(images_link_list_split) - 1]
-        images_name_file = images_name_file_thumb.split('-')[len(images_name_file_thumb.split('-')) - 1]
+        images_name_file = images_name_file_thumb.split("-")[
+            len(images_name_file_thumb.split("-")) - 1
+        ]
         if os.path.isfile(os.path.join(self.path, images_name_file)) is False:
-            image_url = image_url.replace(images_name_file_thumb, '') + images_name_file
+            image_url = image_url.replace(images_name_file_thumb, "") + images_name_file
             async with self.client_session.head(image_url) as request:
-                file_size = int(request.headers['Content-Length'])
+                file_size = int(request.headers["Content-Length"])
                 self.total_size_to_download += file_size
 
             self.images_list.append([image_url, images_name_file, file_size])
 
     async def parse_url(self, url: str):
-        async with self.client_session.get(url, cookies={'AlphaCodersView': 'paged'}) as request:
-            page = BeautifulSoup(await request.text(), 'html.parser')
+        async with self.client_session.get(
+            url, cookies={"AlphaCodersView": "paged"}
+        ) as request:
+            page = BeautifulSoup(await request.text(), "html.parser")
 
-        find_images_urls = page.find('div', {'id': 'page_container'}).find_all('div', 'thumb-container-big')
+        find_images_urls = page.find("div", {"id": "page_container"}).find_all(
+            "div", "thumb-container-big"
+        )
 
         if find_images_urls is None:
             raise WallpapersNotFounds(url)
 
         for a_element in find_images_urls:
-            href = str(a_element.find('img').get('src'))
+            href = str(a_element.find("img").get("src"))
 
-            if (href.startswith('https://images') or href.startswith('https://mfiles')) and href not in \
-                    self.temp_images_list:
+            if (
+                href.startswith("https://images") or href.startswith("https://mfiles")
+            ) and href not in self.temp_images_list:
                 self.temp_images_list.append(href)
 
         self.links_got += 1
 
-        a_elements = page.find('div', {'class': 'pagination-simple center'}).find_all('a')
+        a_elements = page.find("div", {"class": "pagination-simple center"}).find_all(
+            "a"
+        )
         changed_element = None
         if a_elements is not None and a_elements:
             for a_element in a_elements:
-                if a_element.text.strip() == 'Next >>':
+                if a_element.text.strip() == "Next >>":
                     changed_element = a_element
                     break
         if changed_element is not None:
-            url = changed_element.get('href')
+            url = changed_element.get("href")
             try:
-                url_spliced = url.split('&page=')[1]
+                url_spliced = url.split("&page=")[1]
             except IndexError:
-                url_spliced = url.split('?page=')[1]
+                url_spliced = url.split("?page=")[1]
 
-            await self.parse_url(f'{self.url}{self.page_char}page=' + url_spliced)
+            await self.parse_url(f"{self.url}{self.page_char}page=" + url_spliced)
 
     async def download(self, element: list):
         path = os.path.join(self.path, element[1])
@@ -88,12 +97,12 @@ async def download(self, element: list):
             temp_file_exist = os.path.isfile(temp_path)
             if temp_file_exist:
                 file_size = os.stat(temp_path).st_size
-                headers['Range'] = f'bytes={file_size}-'
+                headers["Range"] = f"bytes={file_size}-"
             file_downloaded = 0
 
             async with self.client_session.get(element[0], headers=headers) as request:
                 try:
-                    write_mode = 'ab' if temp_file_exist else 'wb'
+                    write_mode = "ab" if temp_file_exist else "wb"
                     async with aiofiles.open(temp_path, write_mode) as file:
                         try:
                             async for data in request.content.iter_chunked(1024):
@@ -102,93 +111,119 @@ async def download(self, element: list):
                                 file_downloaded += len(data)
                                 self.progress_bar.progress(len(data))
                         except asyncio.TimeoutError:
-                            self.progress_bar.progress(element[2] - file_downloaded, True)
-                            return self.progress_bar.print_error(f"Download of file: {element[0]} has been timeout.")
+                            self.progress_bar.progress(
+                                element[2] - file_downloaded, True
+                            )
+                            return self.progress_bar.print_error(
+                                f"Download of file: {element[0]} has been timeout."
+                            )
                 except aiohttp.ClientPayloadError:
                     self.progress_bar.progress(element[2] - file_downloaded, True)
-                    return self.progress_bar.print_error(f"Download of file: {element[0]} raise ClientPayloadError.")
+                    return self.progress_bar.print_error(
+                        f"Download of file: {element[0]} raise ClientPayloadError."
+                    )
 
                 if os.path.isfile(temp_path):
                     shutil.move(temp_path, path)
         else:
             self.progress_bar.progress(element[2], True)
 
     async def start(self):
-        self.spinner.set_text('Recovery of the URLS of all the pages...')
+        self.spinner.set_text("Recovery of the URLS of all the pages...")
         await self.spinner.start()
 
         create_folder_recursively(self.path)
         create_folder_recursively(self.temp_path)
 
-        self.spinner.set_text('Recovery of the URLS of all the wallpapers...')
-        self.page_char = '&' if 'https://mobile.alphacoders.com/' not in self.url else '?'
-        await self.parse_url(f'{self.url}{self.page_char}page=1')
-        self.spinner.set_text('Recovery of the informations about wallpapers...')
-        await limit_tasks(10, *[self.fetch_images(element) for element in self.temp_images_list])
+        self.spinner.set_text("Recovery of the URLS of all the wallpapers...")
+        self.page_char = (
+            "&" if "https://mobile.alphacoders.com/" not in self.url else "?"
+        )
+        await self.parse_url(f"{self.url}{self.page_char}page=1")
+        self.spinner.set_text("Recovery of the informations about wallpapers...")
+        await limit_tasks(
+            10, *[self.fetch_images(element) for element in self.temp_images_list]
+        )
         self.temp_images_list.clear()
         self.spinner.stop()
 
-        self.progress_bar = ProgressBar(self.total_size_to_download, 'Downloading wallpapers', speed=True)
+        self.progress_bar = ProgressBar(
+            self.total_size_to_download, "Downloading wallpapers", speed=True
+        )
         await limit_tasks(10, *[self.download(element) for element in self.images_list])
 
         shutil.rmtree(self.temp_path)
-        print('\033[1mCompleted!\033[0m')
+        print("\033[1mCompleted!\033[0m")
 
 
 class CommandsHandler:
     @staticmethod
     async def download(command_return: dict):
-        wallpapers_url = command_return['args'][command_return['args'].index('-S') + 1]
-        if 'https://' not in wallpapers_url and 'alphacoders.com' not in wallpapers_url:
+        wallpapers_url = command_return["args"][command_return["args"].index("-S") + 1]
+        if "https://" not in wallpapers_url and "alphacoders.com" not in wallpapers_url:
             print_error("This URL isn't correct.")
             sys.exit()
-        path_to_download = command_return['args'][command_return['args'].index('-P') + 1]
+        path_to_download = command_return["args"][
+            command_return["args"].index("-P") + 1
+        ]
         if os.access(os.path.dirname(path_to_download), os.W_OK) is False:
             print_error("This path isn't correct.")
             sys.exit()
         async with aiohttp.ClientSession() as client_session:
-            await AlphacodersDownloader(wallpapers_url, path_to_download, client_session).start()
+            await AlphacodersDownloader(
+                wallpapers_url, path_to_download, client_session
+            ).start()
 
     @staticmethod
     def get_version(_):
-        version_text = f'\033[1mAlphacodersDownloader {__version__}\033[0m\n'
-        version_text += 'Created by \033[1mAsthowen\033[0m - \033[[email protected]\033[0m\n'
-        version_text += f'License: \033[1m{__license__}\033[0m'
+        version_text = f"\033[1mAlphacodersDownloader {__version__}\033[0m\n"
+        version_text += (
+            "Created by \033[1mAsthowen\033[0m - \033[[email protected]\033[0m\n"
+        )
+        version_text += f"License: \033[1m{__license__}\033[0m"
 
         print(version_text)
 
 
 async def main():
-    setproctitle.setproctitle('AlphacodersDownloader')
+    setproctitle.setproctitle("AlphacodersDownloader")
 
     if len(sys.argv) <= 1:
-        url = ''
-        while 'https://' not in url and 'alphacoders.com' not in url:
+        url = ""
+        while "https://" not in url and "alphacoders.com" not in url:
             url = input(
-                'Please enter the download url (e.g. '
-                'https://wall.alphacoders.com/search.php?search=sword+art+online). > '
-            ).replace(' ', '')
+                "Please enter the download url (e.g. "
+                "https://wall.alphacoders.com/search.php?search=sword+art+online). > "
+            ).replace(" ", "")
             clear_line()
 
-        path = ''
+        path = ""
         while os.access(os.path.dirname(path), os.W_OK) is False:
-            path = input('Please enter the folder where the images are saved (e.g. ~/downloads/wallpapers/). > ')
+            path = input(
+                "Please enter the folder where the images are saved (e.g. ~/downloads/wallpapers/). > "
+            )
             clear_line()
 
         with HiddenCursor() as _:
             async with aiohttp.ClientSession() as client_session:
                 await AlphacodersDownloader(url, path, client_session).start()
     else:
         parser = ArgumentsBuilder(
-            'A script for download wallpapers on https://alphacoders.com/.', 'alphacoders-downloader'
+            "A script for download wallpapers on https://alphacoders.com/.",
+            "alphacoders-downloader",
         )
 
         parser.add_argument(
-            '-S', action=CommandsHandler().download, description='Download wallpapers.',
-            command_usage='-S wallpapers_url -P path'
+            "-S",
+            action=CommandsHandler().download,
+            description="Download wallpapers.",
+            command_usage="-S wallpapers_url -P path",
         )
         parser.add_argument(
-            '-V', action=CommandsHandler.get_version, description='Get version infos.', command_usage='-V'
+            "-V",
+            action=CommandsHandler.get_version,
+            description="Get version infos.",
+            command_usage="-V",
         )
         with HiddenCursor() as _:
             await parser.build()
@@ -200,12 +235,14 @@ def start():
         os.get_terminal_size(0)
         asyncio.get_event_loop().run_until_complete(main())
     except OSError:
-        print_error('Your terminal does not support all the features needed for AlphacodersDownloader, please use '
-                    'another one.')
+        print_error(
+            "Your terminal does not support all the features needed for AlphacodersDownloader, please use "
+            "another one."
+        )
         show()
     except KeyboardInterrupt:
         clear_line()
-        print('Stop the script...')
+        print("Stop the script...")
         show()
     except Exception as exception:
         print_error(str(exception))

alphacoders_downloader/util/__init__.py

@@ -1,4 +1,4 @@
 __title__ = "AlphacodersDownloader"
-__version__ = "0.1.4.1"
+__version__ = "0.1.4.2"
 __author__ = "Asthowen"
 __license__ = "GNU v3.0"

alphacoders_downloader/util/arguments_builder.py

@@ -16,23 +16,33 @@ def __init__(self, description: str, command_base: str, args: list = None):
         self.__help_content = None
 
     def add_argument(
-            self, argument_name: str, action=None, description: str = None, command_usage: str = None
+        self,
+        argument_name: str,
+        action=None,
+        description: str = None,
+        command_usage: str = None,
     ):
         self.arguments[argument_name.lower()] = {
-            'action': action, 'description': description, 'command_usage': command_usage
+            "action": action,
+            "description": description,
+            "command_usage": command_usage,
         }
 
     def build_help(self):
         if self.__help_content is None:
-            self.__help_content = f"\n\033[1m{self.description}\033[0m\n\nCommand list:\n"
+            self.__help_content = (
+                f"\n\033[1m{self.description}\033[0m\n\nCommand list:\n"
+            )
             for _, command_json in self.arguments.items():
-                self.__help_content += f'・\033[1m{self.command_base} {command_json["command_usage"]}\033[' \
-                                       f'0m | {command_json["description"]}\n'
+                self.__help_content += (
+                    f'・\033[1m{self.command_base} {command_json["command_usage"]}\033['
+                    f'0m | {command_json["description"]}\n'
+                )
 
         print(self.__help_content)
 
     async def build(self):
-        if len(self.args) == 1 or '--help' in self.args or '-h' in self.args:
+        if len(self.args) == 1 or "--help" in self.args or "-h" in self.args:
             return self.build_help()
 
         has_been_found = False
@@ -41,10 +51,12 @@ async def build(self):
             argument = argument.lower()
             if argument in self.arguments:
                 has_been_found = True
-                self.arguments[argument]['args'] = self.args
-                if asyncio.iscoroutinefunction(self.arguments[argument]['action']):
-                    await self.arguments[argument]['action'](self.arguments[argument])
+                self.arguments[argument]["args"] = self.args
+                if asyncio.iscoroutinefunction(self.arguments[argument]["action"]):
+                    await self.arguments[argument]["action"](self.arguments[argument])
                 else:
-                    self.arguments[argument]['action'](self.arguments[argument])
+                    self.arguments[argument]["action"](self.arguments[argument])
         if has_been_found is False:
-            print_error(f"\033[1mThis command doesn't exist. Please check the command: {self.command_base} -H.\033[0m")
+            print_error(
+                f"\033[1mThis command doesn't exist. Please check the command: {self.command_base} -H.\033[0m"
+            )

alphacoders_downloader/util/cursor.py

@@ -7,37 +7,36 @@
 import sys
 import os
 
-if os.name == 'nt':
+if os.name == "nt":
     import ctypes
 
-
     class _CursorInfo(ctypes.Structure):
-        _fields_ = [('size', ctypes.c_int), ('visible', ctypes.c_byte)]
+        _fields_ = [("size", ctypes.c_int), ("visible", ctypes.c_byte)]
 
 
 # pylint: disable=W0201
 def hide(stream=sys.stdout):
-    if os.name == 'nt':
+    if os.name == "nt":
         cursor_info = _CursorInfo()
         handle = ctypes.windll.kernel32.GetStdHandle(-11)
         ctypes.windll.kernel32.GetConsoleCursorInfo(handle, ctypes.byref(cursor_info))
         cursor_info.visible = False
         ctypes.windll.kernel32.SetConsoleCursorInfo(handle, ctypes.byref(cursor_info))
-    elif os.name == 'posix':
-        stream.write('\033[?25l')
+    elif os.name == "posix":
+        stream.write("\033[?25l")
         stream.flush()
 
 
 # pylint: disable=W0201
 def show(stream=sys.stdout):
-    if os.name == 'nt':
+    if os.name == "nt":
         cursor_info = _CursorInfo()
         handle = ctypes.windll.kernel32.GetStdHandle(-11)
         ctypes.windll.kernel32.GetConsoleCursorInfo(handle, ctypes.byref(cursor_info))
         cursor_info.visible = True
         ctypes.windll.kernel32.SetConsoleCursorInfo(handle, ctypes.byref(cursor_info))
-    elif os.name == 'posix':
-        stream.write('\033[?25h')
+    elif os.name == "posix":
+        stream.write("\033[?25h")
         stream.flush()