mirror of
https://github.com/spalencsar/bearhub.git
synced 2026-07-07 00:04:15 +02:00
109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
from threading import Thread
|
|
from typing import Set, List, Tuple
|
|
|
|
from bauh.gems.arch import pacman
|
|
from bauh.gems.arch.aur import AURClient
|
|
|
|
|
|
class DependenciesAnalyser:
|
|
|
|
def __init__(self, aur_client: AURClient):
|
|
self.aur_client = aur_client
|
|
|
|
def _fill_mirror(self, name: str, output: List[Tuple[str, str]]):
|
|
|
|
mirror = pacman.read_repository_from_info(name)
|
|
|
|
if mirror:
|
|
output.append((name, mirror))
|
|
return
|
|
|
|
guess = pacman.guess_repository(name)
|
|
|
|
if guess:
|
|
output.append(guess)
|
|
return
|
|
|
|
aur_info = self.aur_client.get_src_info(name)
|
|
|
|
if aur_info:
|
|
output.append((name, 'aur'))
|
|
return
|
|
|
|
output.append((name, ''))
|
|
|
|
def get_missing_packages(self, names: Set[str], mirror: str = None, in_analysis: Set[str] = None) -> List[Tuple[str, str]]:
|
|
"""
|
|
:param names:
|
|
:param mirror:
|
|
:param in_analysis: global set storing all names in analysis to avoid repeated recursion
|
|
:return:
|
|
"""
|
|
global_in_analysis = in_analysis if in_analysis else set()
|
|
|
|
missing_names = pacman.check_missing({n for n in names if n not in in_analysis})
|
|
|
|
if missing_names:
|
|
missing_root = []
|
|
threads = []
|
|
|
|
if not mirror:
|
|
for name in missing_names:
|
|
t = Thread(target=self._fill_mirror, args=(name, missing_root))
|
|
t.start()
|
|
threads.append(t)
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
threads.clear()
|
|
|
|
# checking if there is any unknown dependency:
|
|
for rdep in missing_root:
|
|
if not rdep[1]:
|
|
return missing_root
|
|
else:
|
|
global_in_analysis.add(rdep[0])
|
|
else:
|
|
for missing in missing_names:
|
|
missing_root.append((missing, mirror))
|
|
global_in_analysis.add(missing)
|
|
|
|
missing_sub = []
|
|
for rdep in missing_root:
|
|
subdeps = self.aur_client.get_all_dependencies(rdep[0]) if rdep[1] == 'aur' else pacman.read_dependencies(rdep[0])
|
|
subdeps_not_analysis = {sd for sd in subdeps if sd not in global_in_analysis}
|
|
|
|
if subdeps_not_analysis:
|
|
missing_subdeps = self.get_missing_packages(subdeps_not_analysis, in_analysis=global_in_analysis)
|
|
|
|
# checking if there is any unknown:
|
|
if missing_subdeps:
|
|
for subdep in missing_subdeps:
|
|
if not subdep[0]:
|
|
return [*missing_subdeps, *missing_root]
|
|
|
|
if subdep[0] not in missing_names:
|
|
missing_sub.append(subdep)
|
|
return [*missing_sub, *missing_root]
|
|
|
|
def get_missing_subdeps_of(self, names: Set[str], mirror: str) -> List[Tuple[str, str]]:
|
|
missing = []
|
|
already_added = {*names}
|
|
in_analyses = {*names}
|
|
|
|
for name in names:
|
|
subdeps = self.aur_client.get_all_dependencies(name) if mirror == 'aur' else pacman.read_dependencies(name)
|
|
|
|
if subdeps:
|
|
missing_subdeps = self.get_missing_packages(subdeps, in_analysis=in_analyses)
|
|
|
|
if missing_subdeps:
|
|
for subdep in missing_subdeps: # checking if there is any unknown:
|
|
if subdep[0] not in already_added:
|
|
missing.append(subdep)
|
|
|
|
if not subdep[0]:
|
|
return missing
|
|
return missing
|