Miroirs/slpkg
1
0
Fork 0
mirror of https://gitlab.com/dslackw/slpkg.git synced 2024-11-16 07:47:35 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Improved code quality

Browse source
This commit is contained in:
Dimitris Zlatanidis 2024-04-27 19:44:27 +03:00
parent 9253e019f2
commit abbf44b741
1 changed files with 41 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

59
slpkg/downloader.py
View file

@ -3,7 +3,6 @@
import os
import shutil
from pathlib import Path
from multiprocessing import Process, Semaphore
from urllib.parse import unquote, urlparse
@ -15,7 +14,11 @@ from slpkg.multi_process import MultiProcess
from slpkg.views.views import View
class Downloader(Configs):
class Downloader(Configs): # pylint: disable=[R0902]
"""
Downloads the sources using external tools.
"""
def __init__(self, flags: list):
super(Configs, self).__init__()
@ -66,40 +69,60 @@ class Downloader(Configs):
self.tools(url, path)
def tools(self, url: str, path: Path) -> None:
""" Run the tool to downloading.
Args:
url (str): The URL link.
path (Path): Path to save.
"""
self.semaphore.acquire()
url_parse: str = urlparse(url).path
self.filename: str = unquote(Path(url_parse).name)
if url.startswith('file'):
self.copy_local_binary_file(url)
else:
try:
self.downloader_tools[self.downloader](url, path)
except KeyError:
self.errors.raise_error_message(f"Downloader '{self.downloader}' not supported", exit_status=1)
try:
self.downloader_tools[self.downloader](url, path)
except KeyError:
self.errors.raise_error_message(f"Downloader '{self.downloader}' not supported", exit_status=1)
self.multi_process.process(self.downloader_command)
self.check_if_downloaded(url, path)
self.multi_process.process(self.downloader_command)
self.check_if_downloaded(url, path)
self.semaphore.release()
def copy_local_binary_file(self, url: str) -> None:
try:
shutil.copy2(Path(url.replace('file:', '')), self.tmp_slpkg)
print(f"{self.byellow}Copying{self.endc}: {Path(url.replace('file:', ''))} -> {self.tmp_slpkg}")
except FileNotFoundError as error:
self.errors.raise_error_message(f'{error}', 1)
def set_wget_downloader(self, url: str, path: Path) -> None:
""" Set for wget tool.
Args:
url (str): URL link.
path (Path): Path to save.
"""
self.downloader_command: str = f'{self.downloader} {self.wget_options} --directory-prefix={path} "{url}"'
def set_curl_downloader(self, url: str, path: Path) -> None:
""" Set for curl tool.
Args:
url (str): URL link.
path (Path): Path to save.
"""
self.downloader_command: str = (f'{self.downloader} {self.curl_options} "{url}" '
f'--output {path}/{self.filename}')
def set_lftp_downloader(self, url: str, path: Path) -> None:
""" Set for lftp tool.
Args:
url (str): URL link.
path (Path): Path to save.
"""
self.downloader_command: str = f'{self.downloader} {self.lftp_get_options} {url} -o {path}'
def check_if_downloaded(self, url: str, path: Path) -> None:
""" Checking if file downloaded.
Args:
url (str): URL link.
path (Path): Path to check the file.
"""
path_file: Path = Path(path, self.filename)
if not path_file.exists():
parsed_url = urlparse(url)