Miroirs/slpkg
1
0
Fork 0
mirror of https://gitlab.com/dslackw/slpkg.git synced 2024-12-28 09:58:21 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Updated for check updates

Browse source
This commit is contained in:
Dimitris Zlatanidis 2023-05-08 20:41:04 +03:00
parent cb0bd7aa65
commit 1eea85cbec
1 changed files with 44 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

84
slpkg/check_updates.py
View file

@ -7,6 +7,11 @@ from pathlib import Path
from multiprocessing import Process
from urllib3 import PoolManager, ProxyManager, make_headers
try: # Try to import PySocks if it's installed.
from urllib3.contrib.socks import SOCKSProxyManager
except (ModuleNotFoundError, ImportError):
pass
from slpkg.configs import Configs
from slpkg.utilities import Utilities
from slpkg.progress_bar import ProgressBar
@ -27,6 +32,9 @@ class CheckUpdates(Configs):
self.repos = Repositories()
self.compare: dict = {}
self.http = PoolManager()
self.proxy_default_headers = make_headers(
proxy_basic_auth=f'{self.proxy_username}:{self.proxy_password}')
self.is_binary: bool = self.utils.is_binary_repo(repository)
@ -40,29 +48,34 @@ class CheckUpdates(Configs):
def check_the_repositories(self) -> dict:
if self.option_for_repository:
sbo_repository: dict = {
self.repos.sbo_repo_name: self.sbo_repository,
self.repos.ponce_repo_name: self.ponce_repository
}
if self.is_binary:
self.binary_repository(self.repository)
else:
sbo_repository[self.repository]()
self.check_for_repository()
else:
if self.repos.sbo_repo:
self.sbo_repository()
if self.repos.ponce_repo:
self.ponce_repository()
for repo in list(self.repos.repositories.keys())[2:]:
if self.repos.repositories[repo][0]:
self.binary_repository(repo)
self.check_for_repositories()
return self.compare
def check_for_repository(self):
sbo_repository: dict = {
self.repos.sbo_repo_name: self.sbo_repository,
self.repos.ponce_repo_name: self.ponce_repository
}
if self.is_binary:
self.binary_repository(self.repository)
else:
sbo_repository[self.repository]()
def check_for_repositories(self):
if self.repos.sbo_repo:
self.sbo_repository()
if self.repos.ponce_repo:
self.ponce_repository()
for repo in list(self.repos.repositories.keys())[2:]:
if self.repos.repositories[repo][0]:
self.binary_repository(repo)
def binary_repository(self, repo: str) -> None:
local_chg_txt: Path = Path(self.repos.repositories[repo][1], self.repos.repositories[repo][5])
repo_chg_txt: str = f'{self.repos.repositories[repo][2][0]}{self.repos.repositories[repo][5]}'
@ -81,29 +94,13 @@ class CheckUpdates(Configs):
def compare_the_changelogs(self, local_chg_txt: Path, repo_chg_txt: str) -> bool:
local_size: int = 0
try:
http = PoolManager()
proxy_default_headers = make_headers(proxy_basic_auth=f'{self.proxy_username}:{self.proxy_password}')
if self.proxy_address.startswith('http'):
self.set_http_proxy_server()
if self.proxy_address.startswith('http'):
http = ProxyManager(f'{self.proxy_address}', headers=proxy_default_headers)
elif self.proxy_address.startswith('socks'):
# https://urllib3.readthedocs.io/en/stable/advanced-usage.html#socks-proxies
try: # Try to import PySocks if it's installed.
from urllib3.contrib.socks import SOCKSProxyManager
except (ModuleNotFoundError, ImportError):
logger = logging.getLogger(LoggingConfig.date)
logger.exception(f'{self.__class__.__name__}: '
f'{self.__class__.compare_the_changelogs.__name__}')
raise SystemExit()
http = SOCKSProxyManager(f'{self.proxy_address}', headers=proxy_default_headers)
repo = http.request('GET', repo_chg_txt)
except KeyboardInterrupt:
raise SystemExit(1)
if self.proxy_address.startswith('socks'):
self.set_socks_proxy_server()
repo = self.http.request('GET', repo_chg_txt)
if local_chg_txt.is_file():
local_size: int = int(os.stat(local_chg_txt).st_size)
@ -118,6 +115,13 @@ class CheckUpdates(Configs):
return local_size != repo_size
def set_http_proxy_server(self):
self.http = ProxyManager(f'{self.proxy_address}', headers=self.proxy_default_headers)
def set_socks_proxy_server(self):
# https://urllib3.readthedocs.io/en/stable/advanced-usage.html#socks-proxies
self.http = SOCKSProxyManager(f'{self.proxy_address}', headers=self.proxy_default_headers)
def check_for_updates(self) -> None:
self.check_the_repositories()
self.print_the_messages()