Refactor mirrors

archinstall/lib/global_menu.py

@@ -12,7 +12,8 @@
 from archinstall.lib.interactions.system_conf import select_kernel, select_swap
 from archinstall.lib.locale.locale_menu import LocaleMenu
 from archinstall.lib.menu.abstract_menu import CONFIG_KEY, AbstractMenu
-from archinstall.lib.mirrors import MirrorListHandler, MirrorMenu
+from archinstall.lib.mirror.mirror_handler import MirrorListHandler
+from archinstall.lib.mirror.mirror_menu import MirrorMenu
 from archinstall.lib.models.application import ApplicationConfiguration, ZramConfiguration
 from archinstall.lib.models.authentication import AuthenticationConfiguration
 from archinstall.lib.models.bootloader import Bootloader, BootloaderConfiguration

archinstall/lib/installer.py

@@ -30,7 +30,7 @@
 from archinstall.lib.hardware import SysInfo
 from archinstall.lib.locale.utils import verify_keyboard_layout, verify_x11_keyboard_layout
 from archinstall.lib.luks import Luks2, unlock_luks2_dev
-from archinstall.lib.mirrors import MirrorListHandler
+from archinstall.lib.mirror.mirror_handler import MirrorListHandler
 from archinstall.lib.models.application import ZramAlgorithm
 from archinstall.lib.models.bootloader import Bootloader
 from archinstall.lib.models.device import (

archinstall/lib/mirror/mirror_handler.py

@@ -0,0 +1,163 @@
+import time
+import urllib
+from pathlib import Path
+
+from archinstall.lib.models import MirrorRegion
+from archinstall.lib.models.mirrors import MirrorStatusEntryV3, MirrorStatusListV3
+from archinstall.lib.networking import fetch_data_from_url
+from archinstall.lib.output import debug, info
+
+
+class MirrorListHandler:
+	def __init__(
+		self,
+		local_mirrorlist: Path = Path('/etc/pacman.d/mirrorlist'),
+		offline: bool = False,
+		verbose: bool = False,
+	) -> None:
+		self._local_mirrorlist = local_mirrorlist
+		self._status_mappings: dict[str, list[MirrorStatusEntryV3]] | None = None
+		self._fetched_remote: bool = False
+		self.offline = offline
+		self.verbose = verbose
+
+	def _mappings(self) -> dict[str, list[MirrorStatusEntryV3]]:
+		if self._status_mappings is None:
+			self.load_mirrors()
+
+		assert self._status_mappings is not None
+		return self._status_mappings
+
+	def get_mirror_regions(self) -> list[MirrorRegion]:
+		available_mirrors = []
+		mappings = self._mappings()
+
+		for region_name, status_entry in mappings.items():
+			urls = [entry.server_url for entry in status_entry]
+			region = MirrorRegion(region_name, urls)
+			available_mirrors.append(region)
+
+		return available_mirrors
+
+	def load_mirrors(self) -> None:
+		if self.offline:
+			self._fetched_remote = False
+			self.load_local_mirrors()
+		else:
+			self._fetched_remote = self.load_remote_mirrors()
+			debug(f'load mirrors: {self._fetched_remote}')
+			if not self._fetched_remote:
+				self.load_local_mirrors()
+
+	def load_remote_mirrors(self) -> bool:
+		url = 'https://archlinux.org/mirrors/status/json/'
+		attempts = 3
+
+		for attempt_nr in range(attempts):
+			try:
+				mirrorlist = fetch_data_from_url(url)
+				self._status_mappings = self._parse_remote_mirror_list(mirrorlist)
+				return True
+			except Exception as e:
+				debug(f'Error while fetching mirror list: {e}')
+				time.sleep(attempt_nr + 1)
+
+		debug('Unable to fetch mirror list remotely, falling back to local mirror list')
+		return False
+
+	def load_local_mirrors(self) -> None:
+		with self._local_mirrorlist.open('r') as fp:
+			mirrorlist = fp.read()
+			self._status_mappings = self._parse_local_mirrors(mirrorlist)
+
+	def get_status_by_region(self, region: str, speed_sort: bool) -> list[MirrorStatusEntryV3]:
+		mappings = self._mappings()
+		region_list = mappings[region]
+
+		# Only sort if we have remote mirror data with score/speed info
+		# Local mirrors lack this data and can be modified manually before-hand
+		# Or reflector potentially ran already
+		if self._fetched_remote and speed_sort:
+			info('Sorting your selected mirror list based on the speed between you and the individual mirrors (this might take a while)')
+			# Sort by speed descending (higher is better in bitrate form core.db download)
+			return sorted(region_list, key=lambda mirror: -mirror.speed)
+		# just return as-is without sorting?
+		return region_list
+
+	def _parse_remote_mirror_list(self, mirrorlist: str) -> dict[str, list[MirrorStatusEntryV3]]:
+		context = {'verbose': self.verbose}
+		mirror_status = MirrorStatusListV3.model_validate_json(mirrorlist, context=context)
+
+		sorting_placeholder: dict[str, list[MirrorStatusEntryV3]] = {}
+
+		for mirror in mirror_status.urls:
+			# We filter out mirrors that have bad criteria values
+			if any(
+				[
+					mirror.active is False,  # Disabled by mirror-list admins
+					mirror.last_sync is None,  # Has not synced recently
+					# mirror.score (error rate) over time reported from backend:
+					# https://github.com/archlinux/archweb/blob/31333d3516c91db9a2f2d12260bd61656c011fd1/mirrors/utils.py#L111C22-L111C66
+					(mirror.score is None or mirror.score >= 100),
+				]
+			):
+				continue
+
+			if mirror.country == '':
+				# TODO: This should be removed once RFC!29 is merged and completed
+				# Until then, there are mirrors which lacks data in the backend
+				# and there is no way of knowing where they're located.
+				# So we have to assume world-wide
+				mirror.country = 'Worldwide'
+
+			if mirror.url.startswith('http'):
+				sorting_placeholder.setdefault(mirror.country, []).append(mirror)
+
+		sorted_by_regions: dict[str, list[MirrorStatusEntryV3]] = dict(
+			{region: unsorted_mirrors for region, unsorted_mirrors in sorted(sorting_placeholder.items(), key=lambda item: item[0])}
+		)
+
+		return sorted_by_regions
+
+	def _parse_local_mirrors(self, mirrorlist: str) -> dict[str, list[MirrorStatusEntryV3]]:
+		lines = mirrorlist.splitlines()
+
+		# remove empty lines
+		# lines = [line for line in lines if line]
+
+		mirror_list: dict[str, list[MirrorStatusEntryV3]] = {}
+
+		current_region = ''
+
+		for line in lines:
+			line = line.strip()
+
+			if line.startswith('## '):
+				current_region = line.replace('## ', '').strip()
+				mirror_list.setdefault(current_region, [])
+
+			if line.startswith('Server = '):
+				if not current_region:
+					current_region = 'Local'
+					mirror_list.setdefault(current_region, [])
+
+				url = line.removeprefix('Server = ')
+
+				mirror_entry = MirrorStatusEntryV3(
+					url=url.removesuffix('$repo/os/$arch'),
+					protocol=urllib.parse.urlparse(url).scheme,
+					active=True,
+					country=current_region or 'Worldwide',
+					# The following values are normally populated by
+					# archlinux.org mirror-list endpoint, and can't be known
+					# from just the local mirror-list file.
+					country_code='WW',
+					isos=True,
+					ipv4=True,
+					ipv6=True,
+					details='Locally defined mirror',
+				)
+
+				mirror_list[current_region].append(mirror_entry)
+
+		return mirror_list

archinstall/lib/mirrors.py → archinstall/lib/mirror/mirror_menu.py

@@ -1,24 +1,19 @@
-import time
-import urllib.parse
-from pathlib import Path
 from typing import override
 
 from archinstall.lib.menu.abstract_menu import AbstractSubMenu
 from archinstall.lib.menu.helpers import Input, Loading, Selection
 from archinstall.lib.menu.list_manager import ListManager
+from archinstall.lib.mirror.mirror_handler import MirrorListHandler
 from archinstall.lib.models.mirrors import (
 	CustomRepository,
 	CustomServer,
 	MirrorConfiguration,
 	MirrorRegion,
-	MirrorStatusEntryV3,
-	MirrorStatusListV3,
 	SignCheck,
 	SignOption,
 )
 from archinstall.lib.models.packages import Repository
-from archinstall.lib.networking import fetch_data_from_url
-from archinstall.lib.output import FormattedOutput, debug, info
+from archinstall.lib.output import FormattedOutput
 from archinstall.lib.translationhandler import tr
 from archinstall.tui.ui.menu_item import MenuItem, MenuItemGroup
 from archinstall.tui.ui.result import ResultType
@@ -206,161 +201,6 @@ def _add_custom_server(self, preset: CustomServer | None = None) -> CustomServer
 				return None
 
 
-class MirrorListHandler:
-	def __init__(
-		self,
-		local_mirrorlist: Path = Path('/etc/pacman.d/mirrorlist'),
-		offline: bool = False,
-		verbose: bool = False,
-	) -> None:
-		self._local_mirrorlist = local_mirrorlist
-		self._status_mappings: dict[str, list[MirrorStatusEntryV3]] | None = None
-		self._fetched_remote: bool = False
-		self.offline = offline
-		self.verbose = verbose
-
-	def _mappings(self) -> dict[str, list[MirrorStatusEntryV3]]:
-		if self._status_mappings is None:
-			self.load_mirrors()
-
-		assert self._status_mappings is not None
-		return self._status_mappings
-
-	def get_mirror_regions(self) -> list[MirrorRegion]:
-		available_mirrors = []
-		mappings = self._mappings()
-
-		for region_name, status_entry in mappings.items():
-			urls = [entry.server_url for entry in status_entry]
-			region = MirrorRegion(region_name, urls)
-			available_mirrors.append(region)
-
-		return available_mirrors
-
-	def load_mirrors(self) -> None:
-		if self.offline:
-			self._fetched_remote = False
-			self.load_local_mirrors()
-		else:
-			self._fetched_remote = self.load_remote_mirrors()
-			debug(f'load mirrors: {self._fetched_remote}')
-			if not self._fetched_remote:
-				self.load_local_mirrors()
-
-	def load_remote_mirrors(self) -> bool:
-		url = 'https://archlinux.org/mirrors/status/json/'
-		attempts = 3
-
-		for attempt_nr in range(attempts):
-			try:
-				mirrorlist = fetch_data_from_url(url)
-				self._status_mappings = self._parse_remote_mirror_list(mirrorlist)
-				return True
-			except Exception as e:
-				debug(f'Error while fetching mirror list: {e}')
-				time.sleep(attempt_nr + 1)
-
-		debug('Unable to fetch mirror list remotely, falling back to local mirror list')
-		return False
-
-	def load_local_mirrors(self) -> None:
-		with self._local_mirrorlist.open('r') as fp:
-			mirrorlist = fp.read()
-			self._status_mappings = self._parse_local_mirrors(mirrorlist)
-
-	def get_status_by_region(self, region: str, speed_sort: bool) -> list[MirrorStatusEntryV3]:
-		mappings = self._mappings()
-		region_list = mappings[region]
-
-		# Only sort if we have remote mirror data with score/speed info
-		# Local mirrors lack this data and can be modified manually before-hand
-		# Or reflector potentially ran already
-		if self._fetched_remote and speed_sort:
-			info('Sorting your selected mirror list based on the speed between you and the individual mirrors (this might take a while)')
-			# Sort by speed descending (higher is better in bitrate form core.db download)
-			return sorted(region_list, key=lambda mirror: -mirror.speed)
-		# just return as-is without sorting?
-		return region_list
-
-	def _parse_remote_mirror_list(self, mirrorlist: str) -> dict[str, list[MirrorStatusEntryV3]]:
-		context = {'verbose': self.verbose}
-		mirror_status = MirrorStatusListV3.model_validate_json(mirrorlist, context=context)
-
-		sorting_placeholder: dict[str, list[MirrorStatusEntryV3]] = {}
-
-		for mirror in mirror_status.urls:
-			# We filter out mirrors that have bad criteria values
-			if any(
-				[
-					mirror.active is False,  # Disabled by mirror-list admins
-					mirror.last_sync is None,  # Has not synced recently
-					# mirror.score (error rate) over time reported from backend:
-					# https://github.com/archlinux/archweb/blob/31333d3516c91db9a2f2d12260bd61656c011fd1/mirrors/utils.py#L111C22-L111C66
-					(mirror.score is None or mirror.score >= 100),
-				]
-			):
-				continue
-
-			if mirror.country == '':
-				# TODO: This should be removed once RFC!29 is merged and completed
-				# Until then, there are mirrors which lacks data in the backend
-				# and there is no way of knowing where they're located.
-				# So we have to assume world-wide
-				mirror.country = 'Worldwide'
-
-			if mirror.url.startswith('http'):
-				sorting_placeholder.setdefault(mirror.country, []).append(mirror)
-
-		sorted_by_regions: dict[str, list[MirrorStatusEntryV3]] = dict(
-			{region: unsorted_mirrors for region, unsorted_mirrors in sorted(sorting_placeholder.items(), key=lambda item: item[0])}
-		)
-
-		return sorted_by_regions
-
-	def _parse_local_mirrors(self, mirrorlist: str) -> dict[str, list[MirrorStatusEntryV3]]:
-		lines = mirrorlist.splitlines()
-
-		# remove empty lines
-		# lines = [line for line in lines if line]
-
-		mirror_list: dict[str, list[MirrorStatusEntryV3]] = {}
-
-		current_region = ''
-
-		for line in lines:
-			line = line.strip()
-
-			if line.startswith('## '):
-				current_region = line.replace('## ', '').strip()
-				mirror_list.setdefault(current_region, [])
-
-			if line.startswith('Server = '):
-				if not current_region:
-					current_region = 'Local'
-					mirror_list.setdefault(current_region, [])
-
-				url = line.removeprefix('Server = ')
-
-				mirror_entry = MirrorStatusEntryV3(
-					url=url.removesuffix('$repo/os/$arch'),
-					protocol=urllib.parse.urlparse(url).scheme,
-					active=True,
-					country=current_region or 'Worldwide',
-					# The following values are normally populated by
-					# archlinux.org mirror-list endpoint, and can't be known
-					# from just the local mirror-list file.
-					country_code='WW',
-					isos=True,
-					ipv4=True,
-					ipv6=True,
-					details='Locally defined mirror',
-				)
-
-				mirror_list[current_region].append(mirror_entry)
-
-		return mirror_list
-
-
 class MirrorMenu(AbstractSubMenu[MirrorConfiguration]):
 	def __init__(
 		self,