Miroirs/slpkg
1
0
Fork 0
mirror of https://gitlab.com/dslackw/slpkg.git synced 2025-01-19 10:27:07 +01:00
Code Issues Projects Releases Packages Wiki Activity

Updated for downloader

Browse source
This commit is contained in:
Dimitris Zlatanidis 2023-05-13 21:53:18 +03:00
parent 368d9cae89
commit 249cab6cc3
1 changed files with 34 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
slpkg/downloader.py
View file

@ -23,6 +23,8 @@ class Downloader(Configs):
self.errors = Errors()
self.utils = Utilities()
self.downloader_command: str = str()
self.option_for_parallel: bool = self.utils.is_option(
['-P', '--parallel'], flags)
@ -32,7 +34,7 @@ class Downloader(Configs):
if self.parallel_downloads or self.option_for_parallel:
for url in self.urls:
p1 = Process(target=self.tool, args=(url,))
p1 = Process(target=self.tools, args=(url,))
process.append(p1)
p1.start()
@ -40,39 +42,49 @@ class Downloader(Configs):
proc.join()
else:
for url in self.urls:
self.tool(url)
self.tools(url)
def tool(self, url: str) -> None:
def tools(self, url: str) -> None:
""" Downloader tools wget, wget2, curl and lftp. """
command: str = ''
path: str = urlparse(url).path
filename: str = unquote(Path(path).name)
downloader_tool: dict = {
'wget': self.wget_downloader,
'wget2': self.wget_downloader,
'curl': self.curl_downloader,
'lftp': self.lftp_downloader
}
if url.startswith('file'):
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)
self.copy_local_binary_file(url)
else:
if self.downloader in ['wget', 'wget2']:
command: str = f'{self.downloader} {self.wget_options} --directory-prefix={self.path} "{url}"'
elif self.downloader == 'curl':
command: str = f'{self.downloader} {self.curl_options} "{url}" --output {self.path}/{filename}'
elif self.downloader == 'lftp':
command: str = f'{self.downloader} {self.lftp_get_options} {url} -o {self.path}'
else:
try:
downloader_tool[self.downloader](url, filename)
except KeyError:
self.errors.raise_error_message(f"Downloader '{self.downloader}' not supported", exit_status=1)
self.utils.process(command)
self.utils.process(self.downloader_command)
self.check_if_downloaded(url, filename)
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 wget_downloader(self, url: str, filename=None) -> None:
self.downloader_command: str = f'{self.downloader} {self.wget_options} --directory-prefix={self.path} "{url}"'
def curl_downloader(self, url: str, filename: str) -> None:
self.downloader_command: str = f'{self.downloader} {self.curl_options} "{url}" --output {self.path}/{filename}'
def lftp_downloader(self, url: str, filename=None) -> None:
self.downloader_command: str = f'{self.downloader} {self.lftp_get_options} {url} -o {self.path}'
def check_if_downloaded(self, url: str, filename: str) -> None:
""" Checks if the file downloaded. """
path_file: Path = Path(self.path, filename)
if not path_file.exists():
self.errors.raise_error_message(f"Download the '{url}' file", exit_status=20)
self.errors.raise_error_message(f"Download the '{url}'", exit_status=20)