From 3c5a1557f1bab72770a7f70aa78230e5734dbfde Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Mon, 19 Jan 2026 20:42:31 +1100 Subject: [PATCH 1/3] Refactor mirror menu files --- archinstall/lib/mirror/__init__.py | 0 archinstall/lib/mirror/mirror_handler.py | 166 ++++++++++++++++ .../lib/{mirrors.py => mirror/mirror_menu.py} | 182 ++---------------- archinstall/lib/models/mirrors.py | 13 +- 4 files changed, 184 insertions(+), 177 deletions(-) create mode 100644 archinstall/lib/mirror/__init__.py create mode 100644 archinstall/lib/mirror/mirror_handler.py rename archinstall/lib/{mirrors.py => mirror/mirror_menu.py} (61%) diff --git a/archinstall/lib/mirror/__init__.py b/archinstall/lib/mirror/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/archinstall/lib/mirror/mirror_handler.py b/archinstall/lib/mirror/mirror_handler.py new file mode 100644 index 0000000000..773346b729 --- /dev/null +++ b/archinstall/lib/mirror/mirror_handler.py @@ -0,0 +1,166 @@ +import time +import urllib.parse +from pathlib import Path + +from ..models.mirrors import ( + MirrorRegion, + MirrorStatusEntryV3, + MirrorStatusListV3, +) +from ..networking import fetch_data_from_url +from ..output import debug, info + + +class MirrorListHandler: + def __init__( + self, + local_mirrorlist: Path = Path('/etc/pacman.d/mirrorlist'), + ) -> None: + self._local_mirrorlist = local_mirrorlist + self._status_mappings: dict[str, list[MirrorStatusEntryV3]] | None = None + self._fetched_remote: bool = False + + 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: + from .args import arch_config_handler + + if arch_config_handler.args.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_locale_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]]: + mirror_status = MirrorStatusListV3.model_validate_json(mirrorlist) + + 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_locale_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 + + +mirror_list_handler = MirrorListHandler() diff --git a/archinstall/lib/mirrors.py b/archinstall/lib/mirror/mirror_menu.py similarity index 61% rename from archinstall/lib/mirrors.py rename to archinstall/lib/mirror/mirror_menu.py index 0abcfd8053..efb8820a07 100644 --- a/archinstall/lib/mirrors.py +++ b/archinstall/lib/mirror/mirror_menu.py @@ -1,6 +1,3 @@ -import time -import urllib.parse -from pathlib import Path from typing import override from archinstall.lib.menu.helpers import Input, Loading, Selection @@ -8,21 +5,19 @@ from archinstall.tui.ui.menu_item import MenuItem, MenuItemGroup from archinstall.tui.ui.result import ResultType -from .menu.abstract_menu import AbstractSubMenu -from .menu.list_manager import ListManager -from .models.mirrors import ( +from ..menu.abstract_menu import AbstractSubMenu +from ..menu.list_manager import ListManager +from ..models.mirrors import ( CustomRepository, CustomServer, MirrorConfiguration, MirrorRegion, - MirrorStatusEntryV3, - MirrorStatusListV3, SignCheck, SignOption, ) -from .models.packages import Repository -from .networking import fetch_data_from_url -from .output import FormattedOutput, debug, info +from ..models.packages import Repository +from ..output import FormattedOutput +from .mirror_handler import mirror_list_handler class CustomMirrorRepositoriesList(ListManager[CustomRepository]): @@ -51,17 +46,17 @@ def handle_action( entry: CustomRepository | None, data: list[CustomRepository], ) -> list[CustomRepository]: - if action == self._actions[0]: # add + if action == self._actions[0]: # add new_repo = self._add_custom_repository() if new_repo is not None: data = [d for d in data if d.name != new_repo.name] data += [new_repo] - elif action == self._actions[1] and entry: # modify repo + elif action == self._actions[1] and entry: # modify repo new_repo = self._add_custom_repository(entry) if new_repo is not None: data = [d for d in data if d.name != entry.name] data += [new_repo] - elif action == self._actions[2] and entry: # delete + elif action == self._actions[2] and entry: # delete data = [d for d in data if d != entry] return data @@ -169,17 +164,17 @@ def handle_action( entry: CustomServer | None, data: list[CustomServer], ) -> list[CustomServer]: - if action == self._actions[0]: # add + if action == self._actions[0]: # add new_server = self._add_custom_server() if new_server is not None: data = [d for d in data if d.url != new_server.url] data += [new_server] - elif action == self._actions[1] and entry: # modify repo + elif action == self._actions[1] and entry: # modify repo new_server = self._add_custom_server(entry) if new_server is not None: data = [d for d in data if d.url != entry.url] data += [new_server] - elif action == self._actions[2] and entry: # delete + elif action == self._actions[2] and entry: # delete data = [d for d in data if d != entry] return data @@ -376,156 +371,3 @@ def select_optional_repositories(preset: list[Repository]) -> list[Repository]: return result.get_values() -class MirrorListHandler: - def __init__( - self, - local_mirrorlist: Path = Path('/etc/pacman.d/mirrorlist'), - ) -> None: - self._local_mirrorlist = local_mirrorlist - self._status_mappings: dict[str, list[MirrorStatusEntryV3]] | None = None - self._fetched_remote: bool = False - - 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: - from .args import arch_config_handler - - if arch_config_handler.args.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_locale_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]]: - mirror_status = MirrorStatusListV3.model_validate_json(mirrorlist) - - 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_locale_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 - - -mirror_list_handler = MirrorListHandler() diff --git a/archinstall/lib/models/mirrors.py b/archinstall/lib/models/mirrors.py index 7768c50b8e..cd74c21437 100644 --- a/archinstall/lib/models/mirrors.py +++ b/archinstall/lib/models/mirrors.py @@ -12,6 +12,7 @@ from ..models.packages import Repository from ..networking import DownloadTimer, ping from ..output import debug +from ..mirror.mirror_handler import mirror_list_handler class MirrorStatusEntryV3(BaseModel): @@ -59,17 +60,17 @@ def speed(self) -> float: assert timer.time is not None self._speed = size / timer.time - debug(f' speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)') + debug(f' speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)') # Do not retry error except urllib.error.URLError as error: - debug(f' speed: ({error}), skip') + debug(f' speed: ({error}), skip') self._speed = 0 # Do retry error except (http.client.IncompleteRead, ConnectionResetError) as error: - debug(f' speed: ({error}), retry') + debug(f' speed: ({error}), retry') # Catch all except Exception as error: - debug(f' speed: ({error}), skip') + debug(f' speed: ({error}), skip') self._speed = 0 retry += 1 @@ -99,7 +100,7 @@ def latency(self) -> float | None: def validate_score(cls, value: float) -> int | None: if value is not None: value = round(value) - debug(f' score: {value}') + debug(f' score: {value}') return value @@ -272,8 +273,6 @@ def custom_servers_config(self) -> str: return config.strip() def regions_config(self, speed_sort: bool = True) -> str: - from ..mirrors import mirror_list_handler - config = '' for mirror_region in self.mirror_regions: From ec7dfa5bf188eb4d4e37a5214b4f09e85b8aaad1 Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Mon, 19 Jan 2026 20:47:33 +1100 Subject: [PATCH 2/3] Update --- archinstall/lib/global_menu.py | 2 +- archinstall/lib/models/mirrors.py | 550 +++++++++++++++--------------- 2 files changed, 276 insertions(+), 276 deletions(-) diff --git a/archinstall/lib/global_menu.py b/archinstall/lib/global_menu.py index 40e114f8a7..b3f65575db 100644 --- a/archinstall/lib/global_menu.py +++ b/archinstall/lib/global_menu.py @@ -25,7 +25,7 @@ from .interactions.system_conf import ask_for_swap, select_kernel from .locale.locale_menu import LocaleMenu from .menu.abstract_menu import CONFIG_KEY, AbstractMenu -from .mirrors import MirrorMenu +from .mirror.mirror_menu import MirrorMenu from .models.bootloader import Bootloader, BootloaderConfiguration from .models.locale import LocaleConfiguration from .models.mirrors import MirrorConfiguration diff --git a/archinstall/lib/models/mirrors.py b/archinstall/lib/models/mirrors.py index cd74c21437..47c2f65ca5 100644 --- a/archinstall/lib/models/mirrors.py +++ b/archinstall/lib/models/mirrors.py @@ -12,320 +12,320 @@ from ..models.packages import Repository from ..networking import DownloadTimer, ping from ..output import debug -from ..mirror.mirror_handler import mirror_list_handler class MirrorStatusEntryV3(BaseModel): - url: str - protocol: str - active: bool - country: str - country_code: str - isos: bool - ipv4: bool - ipv6: bool - details: str - delay: int | None = None - last_sync: datetime.datetime | None = None - duration_avg: float | None = None - duration_stddev: float | None = None - completion_pct: float | None = None - score: float | None = None - _latency: float | None = None - _speed: float | None = None - _hostname: str | None = None - _port: int | None = None - _speedtest_retries: int | None = None - - @property - def server_url(self) -> str: - return f'{self.url}$repo/os/$arch' - - @property - def speed(self) -> float: - if self._speed is None: - if not self._speedtest_retries: - self._speedtest_retries = 3 - elif self._speedtest_retries < 1: - self._speedtest_retries = 1 - - retry = 0 - while retry < self._speedtest_retries and self._speed is None: - debug(f'Checking download speed of {self._hostname}[{self.score}] by fetching: {self.url}core/os/x86_64/core.db') - req = urllib.request.Request(url=f'{self.url}core/os/x86_64/core.db') - - try: - with urllib.request.urlopen(req, None, 5) as handle, DownloadTimer(timeout=5) as timer: - size = len(handle.read()) - - assert timer.time is not None - self._speed = size / timer.time - debug(f' speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)') - # Do not retry error - except urllib.error.URLError as error: - debug(f' speed: ({error}), skip') - self._speed = 0 - # Do retry error - except (http.client.IncompleteRead, ConnectionResetError) as error: - debug(f' speed: ({error}), retry') - # Catch all - except Exception as error: - debug(f' speed: ({error}), skip') - self._speed = 0 - - retry += 1 - - if self._speed is None: - self._speed = 0 - - return self._speed - - @property - def latency(self) -> float | None: - """ - Latency measures the milliseconds between one ICMP request & response. - It only does so once because we check if self._latency is None, and a ICMP timeout result in -1 - We do this because some hosts blocks ICMP so we'll have to rely on .speed() instead which is slower. - """ - if self._latency is None: - debug(f'Checking latency for {self.url}') - assert self._hostname is not None - self._latency = ping(self._hostname, timeout=2) - debug(f' latency: {self._latency}') - - return self._latency - - @classmethod - @field_validator('score', mode='before') - def validate_score(cls, value: float) -> int | None: - if value is not None: - value = round(value) - debug(f' score: {value}') - - return value - - @model_validator(mode='after') - def debug_output(self) -> Self: - from ..args import arch_config_handler - - self._hostname, *port = urllib.parse.urlparse(self.url).netloc.split(':', 1) - self._port = int(port[0]) if port and len(port) >= 1 else None - - if arch_config_handler.args.verbose: - debug(f'Loaded mirror {self._hostname}' + (f' with current score of {self.score}' if self.score else '')) - return self + url: str + protocol: str + active: bool + country: str + country_code: str + isos: bool + ipv4: bool + ipv6: bool + details: str + delay: int | None = None + last_sync: datetime.datetime | None = None + duration_avg: float | None = None + duration_stddev: float | None = None + completion_pct: float | None = None + score: float | None = None + _latency: float | None = None + _speed: float | None = None + _hostname: str | None = None + _port: int | None = None + _speedtest_retries: int | None = None + + @property + def server_url(self) -> str: + return f'{self.url}$repo/os/$arch' + + @property + def speed(self) -> float: + if self._speed is None: + if not self._speedtest_retries: + self._speedtest_retries = 3 + elif self._speedtest_retries < 1: + self._speedtest_retries = 1 + + retry = 0 + while retry < self._speedtest_retries and self._speed is None: + debug(f'Checking download speed of {self._hostname}[{self.score}] by fetching: {self.url}core/os/x86_64/core.db') + req = urllib.request.Request(url=f'{self.url}core/os/x86_64/core.db') + + try: + with urllib.request.urlopen(req, None, 5) as handle, DownloadTimer(timeout=5) as timer: + size = len(handle.read()) + + assert timer.time is not None + self._speed = size / timer.time + debug(f' speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)') + # Do not retry error + except urllib.error.URLError as error: + debug(f' speed: ({error}), skip') + self._speed = 0 + # Do retry error + except (http.client.IncompleteRead, ConnectionResetError) as error: + debug(f' speed: ({error}), retry') + # Catch all + except Exception as error: + debug(f' speed: ({error}), skip') + self._speed = 0 + + retry += 1 + + if self._speed is None: + self._speed = 0 + + return self._speed + + @property + def latency(self) -> float | None: + """ + Latency measures the milliseconds between one ICMP request & response. + It only does so once because we check if self._latency is None, and a ICMP timeout result in -1 + We do this because some hosts blocks ICMP so we'll have to rely on .speed() instead which is slower. + """ + if self._latency is None: + debug(f'Checking latency for {self.url}') + assert self._hostname is not None + self._latency = ping(self._hostname, timeout=2) + debug(f' latency: {self._latency}') + + return self._latency + + @classmethod + @field_validator('score', mode='before') + def validate_score(cls, value: float) -> int | None: + if value is not None: + value = round(value) + debug(f' score: {value}') + + return value + + @model_validator(mode='after') + def debug_output(self) -> Self: + from ..args import arch_config_handler + + self._hostname, *port = urllib.parse.urlparse(self.url).netloc.split(':', 1) + self._port = int(port[0]) if port and len(port) >= 1 else None + + if arch_config_handler.args.verbose: + debug(f'Loaded mirror {self._hostname}' + (f' with current score of {self.score}' if self.score else '')) + return self class MirrorStatusListV3(BaseModel): - cutoff: int - last_check: datetime.datetime - num_checks: int - urls: list[MirrorStatusEntryV3] - version: int + cutoff: int + last_check: datetime.datetime + num_checks: int + urls: list[MirrorStatusEntryV3] + version: int - @model_validator(mode='before') - @classmethod - def check_model( - cls, - data: dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]], - ) -> dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]]: - if data.get('version') == 3: - return data + @model_validator(mode='before') + @classmethod + def check_model( + cls, + data: dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]], + ) -> dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]]: + if data.get('version') == 3: + return data - raise ValueError('MirrorStatusListV3 only accepts version 3 data from https://archlinux.org/mirrors/status/json/') + raise ValueError('MirrorStatusListV3 only accepts version 3 data from https://archlinux.org/mirrors/status/json/') @dataclass class MirrorRegion: - name: str - urls: list[str] + name: str + urls: list[str] - def json(self) -> dict[str, list[str]]: - return {self.name: self.urls} + def json(self) -> dict[str, list[str]]: + return {self.name: self.urls} - @override - def __eq__(self, other: object) -> bool: - if not isinstance(other, MirrorRegion): - return NotImplemented - return self.name == other.name + @override + def __eq__(self, other: object) -> bool: + if not isinstance(other, MirrorRegion): + return NotImplemented + return self.name == other.name class SignCheck(Enum): - Never = 'Never' - Optional = 'Optional' - Required = 'Required' + Never = 'Never' + Optional = 'Optional' + Required = 'Required' class SignOption(Enum): - TrustedOnly = 'TrustedOnly' - TrustAll = 'TrustAll' + TrustedOnly = 'TrustedOnly' + TrustAll = 'TrustAll' class _CustomRepositorySerialization(TypedDict): - name: str - url: str - sign_check: str - sign_option: str + name: str + url: str + sign_check: str + sign_option: str @dataclass class CustomRepository: - name: str - url: str - sign_check: SignCheck - sign_option: SignOption - - def table_data(self) -> dict[str, str]: - return { - 'Name': self.name, - 'Url': self.url, - 'Sign check': self.sign_check.value, - 'Sign options': self.sign_option.value, - } - - def json(self) -> _CustomRepositorySerialization: - return { - 'name': self.name, - 'url': self.url, - 'sign_check': self.sign_check.value, - 'sign_option': self.sign_option.value, - } - - @classmethod - def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: - configs = [] - for arg in args: - configs.append( - cls( - arg['name'], - arg['url'], - SignCheck(arg['sign_check']), - SignOption(arg['sign_option']), - ), - ) - - return configs + name: str + url: str + sign_check: SignCheck + sign_option: SignOption + + def table_data(self) -> dict[str, str]: + return { + 'Name': self.name, + 'Url': self.url, + 'Sign check': self.sign_check.value, + 'Sign options': self.sign_option.value, + } + + def json(self) -> _CustomRepositorySerialization: + return { + 'name': self.name, + 'url': self.url, + 'sign_check': self.sign_check.value, + 'sign_option': self.sign_option.value, + } + + @classmethod + def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: + configs = [] + for arg in args: + configs.append( + cls( + arg['name'], + arg['url'], + SignCheck(arg['sign_check']), + SignOption(arg['sign_option']), + ), + ) + + return configs @dataclass class CustomServer: - url: str + url: str - def table_data(self) -> dict[str, str]: - return {'Url': self.url} + def table_data(self) -> dict[str, str]: + return {'Url': self.url} - def json(self) -> dict[str, str]: - return {'url': self.url} + def json(self) -> dict[str, str]: + return {'url': self.url} - @classmethod - def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: - configs = [] - for arg in args: - configs.append( - cls(arg['url']), - ) + @classmethod + def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: + configs = [] + for arg in args: + configs.append( + cls(arg['url']), + ) - return configs + return configs class _MirrorConfigurationSerialization(TypedDict): - mirror_regions: dict[str, list[str]] - custom_servers: list[CustomServer] - optional_repositories: list[str] - custom_repositories: list[_CustomRepositorySerialization] + mirror_regions: dict[str, list[str]] + custom_servers: list[CustomServer] + optional_repositories: list[str] + custom_repositories: list[_CustomRepositorySerialization] @dataclass class MirrorConfiguration: - mirror_regions: list[MirrorRegion] = field(default_factory=list) - custom_servers: list[CustomServer] = field(default_factory=list) - optional_repositories: list[Repository] = field(default_factory=list) - custom_repositories: list[CustomRepository] = field(default_factory=list) - - @property - def region_names(self) -> str: - return '\n'.join(m.name for m in self.mirror_regions) - - @property - def custom_server_urls(self) -> str: - return '\n'.join(s.url for s in self.custom_servers) - - def json(self) -> _MirrorConfigurationSerialization: - regions = {} - for m in self.mirror_regions: - regions.update(m.json()) - - return { - 'mirror_regions': regions, - 'custom_servers': self.custom_servers, - 'optional_repositories': [r.value for r in self.optional_repositories], - 'custom_repositories': [c.json() for c in self.custom_repositories], - } - - def custom_servers_config(self) -> str: - config = '' - - if self.custom_servers: - config += '## Custom Servers\n' - for server in self.custom_servers: - config += f'Server = {server.url}\n' - - return config.strip() - - def regions_config(self, speed_sort: bool = True) -> str: - config = '' - - for mirror_region in self.mirror_regions: - sorted_stati = mirror_list_handler.get_status_by_region( - mirror_region.name, - speed_sort=speed_sort, - ) - - config += f'\n\n## {mirror_region.name}\n' - - for status in sorted_stati: - config += f'Server = {status.server_url}\n' - - return config - - def repositories_config(self) -> str: - config = '' - - for repo in self.custom_repositories: - config += f'\n\n[{repo.name}]\n' - config += f'SigLevel = {repo.sign_check.value} {repo.sign_option.value}\n' - config += f'Server = {repo.url}\n' - - return config - - @classmethod - def parse_args( - cls, - args: dict[str, Any], - backwards_compatible_repo: list[Repository] = [], - ) -> Self: - config = cls() - - mirror_regions = args.get('mirror_regions', []) - if mirror_regions: - for region, urls in mirror_regions.items(): - config.mirror_regions.append(MirrorRegion(region, urls)) - - if args.get('custom_servers'): - config.custom_servers = CustomServer.parse_args(args['custom_servers']) - - # backwards compatibility with the new custom_repository - if 'custom_mirrors' in args: - config.custom_repositories = CustomRepository.parse_args(args['custom_mirrors']) - if 'custom_repositories' in args: - config.custom_repositories = CustomRepository.parse_args(args['custom_repositories']) - - if 'optional_repositories' in args: - config.optional_repositories = [Repository(r) for r in args['optional_repositories']] - - if backwards_compatible_repo: - for r in backwards_compatible_repo: - if r not in config.optional_repositories: - config.optional_repositories.append(r) - - return config + mirror_regions: list[MirrorRegion] = field(default_factory=list) + custom_servers: list[CustomServer] = field(default_factory=list) + optional_repositories: list[Repository] = field(default_factory=list) + custom_repositories: list[CustomRepository] = field(default_factory=list) + + @property + def region_names(self) -> str: + return '\n'.join(m.name for m in self.mirror_regions) + + @property + def custom_server_urls(self) -> str: + return '\n'.join(s.url for s in self.custom_servers) + + def json(self) -> _MirrorConfigurationSerialization: + regions = {} + for m in self.mirror_regions: + regions.update(m.json()) + + return { + 'mirror_regions': regions, + 'custom_servers': self.custom_servers, + 'optional_repositories': [r.value for r in self.optional_repositories], + 'custom_repositories': [c.json() for c in self.custom_repositories], + } + + def custom_servers_config(self) -> str: + config = '' + + if self.custom_servers: + config += '## Custom Servers\n' + for server in self.custom_servers: + config += f'Server = {server.url}\n' + + return config.strip() + + def regions_config(self, speed_sort: bool = True) -> str: + from ..mirror.mirror_handler import mirror_list_handler + config = '' + + for mirror_region in self.mirror_regions: + sorted_stati = mirror_list_handler.get_status_by_region( + mirror_region.name, + speed_sort=speed_sort, + ) + + config += f'\n\n## {mirror_region.name}\n' + + for status in sorted_stati: + config += f'Server = {status.server_url}\n' + + return config + + def repositories_config(self) -> str: + config = '' + + for repo in self.custom_repositories: + config += f'\n\n[{repo.name}]\n' + config += f'SigLevel = {repo.sign_check.value} {repo.sign_option.value}\n' + config += f'Server = {repo.url}\n' + + return config + + @classmethod + def parse_args( + cls, + args: dict[str, Any], + backwards_compatible_repo: list[Repository] = [], + ) -> Self: + config = cls() + + mirror_regions = args.get('mirror_regions', []) + if mirror_regions: + for region, urls in mirror_regions.items(): + config.mirror_regions.append(MirrorRegion(region, urls)) + + if args.get('custom_servers'): + config.custom_servers = CustomServer.parse_args(args['custom_servers']) + + # backwards compatibility with the new custom_repository + if 'custom_mirrors' in args: + config.custom_repositories = CustomRepository.parse_args(args['custom_mirrors']) + if 'custom_repositories' in args: + config.custom_repositories = CustomRepository.parse_args(args['custom_repositories']) + + if 'optional_repositories' in args: + config.optional_repositories = [Repository(r) for r in args['optional_repositories']] + + if backwards_compatible_repo: + for r in backwards_compatible_repo: + if r not in config.optional_repositories: + config.optional_repositories.append(r) + + return config From 7f291aaea17e4a490c07b32551f1ae368799aef0 Mon Sep 17 00:00:00 2001 From: Daniel Girtler Date: Mon, 19 Jan 2026 20:51:05 +1100 Subject: [PATCH 3/3] Update --- archinstall/lib/mirror/mirror_handler.py | 2 +- archinstall/lib/mirror/mirror_menu.py | 14 +- archinstall/lib/models/mirrors.py | 551 ++++++++++++----------- tests/test_mirrorlist.py | 2 +- 4 files changed, 284 insertions(+), 285 deletions(-) diff --git a/archinstall/lib/mirror/mirror_handler.py b/archinstall/lib/mirror/mirror_handler.py index 773346b729..37d62b0c35 100644 --- a/archinstall/lib/mirror/mirror_handler.py +++ b/archinstall/lib/mirror/mirror_handler.py @@ -39,7 +39,7 @@ def get_mirror_regions(self) -> list[MirrorRegion]: return available_mirrors def load_mirrors(self) -> None: - from .args import arch_config_handler + from archinstall.lib.args import arch_config_handler if arch_config_handler.args.offline: self._fetched_remote = False diff --git a/archinstall/lib/mirror/mirror_menu.py b/archinstall/lib/mirror/mirror_menu.py index efb8820a07..dff9b09e5d 100644 --- a/archinstall/lib/mirror/mirror_menu.py +++ b/archinstall/lib/mirror/mirror_menu.py @@ -46,17 +46,17 @@ def handle_action( entry: CustomRepository | None, data: list[CustomRepository], ) -> list[CustomRepository]: - if action == self._actions[0]: # add + if action == self._actions[0]: # add new_repo = self._add_custom_repository() if new_repo is not None: data = [d for d in data if d.name != new_repo.name] data += [new_repo] - elif action == self._actions[1] and entry: # modify repo + elif action == self._actions[1] and entry: # modify repo new_repo = self._add_custom_repository(entry) if new_repo is not None: data = [d for d in data if d.name != entry.name] data += [new_repo] - elif action == self._actions[2] and entry: # delete + elif action == self._actions[2] and entry: # delete data = [d for d in data if d != entry] return data @@ -164,17 +164,17 @@ def handle_action( entry: CustomServer | None, data: list[CustomServer], ) -> list[CustomServer]: - if action == self._actions[0]: # add + if action == self._actions[0]: # add new_server = self._add_custom_server() if new_server is not None: data = [d for d in data if d.url != new_server.url] data += [new_server] - elif action == self._actions[1] and entry: # modify repo + elif action == self._actions[1] and entry: # modify repo new_server = self._add_custom_server(entry) if new_server is not None: data = [d for d in data if d.url != entry.url] data += [new_server] - elif action == self._actions[2] and entry: # delete + elif action == self._actions[2] and entry: # delete data = [d for d in data if d != entry] return data @@ -369,5 +369,3 @@ def select_optional_repositories(preset: list[Repository]) -> list[Repository]: return [] case ResultType.Selection: return result.get_values() - - diff --git a/archinstall/lib/models/mirrors.py b/archinstall/lib/models/mirrors.py index 47c2f65ca5..bc8110f4d2 100644 --- a/archinstall/lib/models/mirrors.py +++ b/archinstall/lib/models/mirrors.py @@ -15,317 +15,318 @@ class MirrorStatusEntryV3(BaseModel): - url: str - protocol: str - active: bool - country: str - country_code: str - isos: bool - ipv4: bool - ipv6: bool - details: str - delay: int | None = None - last_sync: datetime.datetime | None = None - duration_avg: float | None = None - duration_stddev: float | None = None - completion_pct: float | None = None - score: float | None = None - _latency: float | None = None - _speed: float | None = None - _hostname: str | None = None - _port: int | None = None - _speedtest_retries: int | None = None - - @property - def server_url(self) -> str: - return f'{self.url}$repo/os/$arch' - - @property - def speed(self) -> float: - if self._speed is None: - if not self._speedtest_retries: - self._speedtest_retries = 3 - elif self._speedtest_retries < 1: - self._speedtest_retries = 1 - - retry = 0 - while retry < self._speedtest_retries and self._speed is None: - debug(f'Checking download speed of {self._hostname}[{self.score}] by fetching: {self.url}core/os/x86_64/core.db') - req = urllib.request.Request(url=f'{self.url}core/os/x86_64/core.db') - - try: - with urllib.request.urlopen(req, None, 5) as handle, DownloadTimer(timeout=5) as timer: - size = len(handle.read()) - - assert timer.time is not None - self._speed = size / timer.time - debug(f' speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)') - # Do not retry error - except urllib.error.URLError as error: - debug(f' speed: ({error}), skip') - self._speed = 0 - # Do retry error - except (http.client.IncompleteRead, ConnectionResetError) as error: - debug(f' speed: ({error}), retry') - # Catch all - except Exception as error: - debug(f' speed: ({error}), skip') - self._speed = 0 - - retry += 1 - - if self._speed is None: - self._speed = 0 - - return self._speed - - @property - def latency(self) -> float | None: - """ - Latency measures the milliseconds between one ICMP request & response. - It only does so once because we check if self._latency is None, and a ICMP timeout result in -1 - We do this because some hosts blocks ICMP so we'll have to rely on .speed() instead which is slower. - """ - if self._latency is None: - debug(f'Checking latency for {self.url}') - assert self._hostname is not None - self._latency = ping(self._hostname, timeout=2) - debug(f' latency: {self._latency}') - - return self._latency - - @classmethod - @field_validator('score', mode='before') - def validate_score(cls, value: float) -> int | None: - if value is not None: - value = round(value) - debug(f' score: {value}') - - return value - - @model_validator(mode='after') - def debug_output(self) -> Self: - from ..args import arch_config_handler - - self._hostname, *port = urllib.parse.urlparse(self.url).netloc.split(':', 1) - self._port = int(port[0]) if port and len(port) >= 1 else None - - if arch_config_handler.args.verbose: - debug(f'Loaded mirror {self._hostname}' + (f' with current score of {self.score}' if self.score else '')) - return self + url: str + protocol: str + active: bool + country: str + country_code: str + isos: bool + ipv4: bool + ipv6: bool + details: str + delay: int | None = None + last_sync: datetime.datetime | None = None + duration_avg: float | None = None + duration_stddev: float | None = None + completion_pct: float | None = None + score: float | None = None + _latency: float | None = None + _speed: float | None = None + _hostname: str | None = None + _port: int | None = None + _speedtest_retries: int | None = None + + @property + def server_url(self) -> str: + return f'{self.url}$repo/os/$arch' + + @property + def speed(self) -> float: + if self._speed is None: + if not self._speedtest_retries: + self._speedtest_retries = 3 + elif self._speedtest_retries < 1: + self._speedtest_retries = 1 + + retry = 0 + while retry < self._speedtest_retries and self._speed is None: + debug(f'Checking download speed of {self._hostname}[{self.score}] by fetching: {self.url}core/os/x86_64/core.db') + req = urllib.request.Request(url=f'{self.url}core/os/x86_64/core.db') + + try: + with urllib.request.urlopen(req, None, 5) as handle, DownloadTimer(timeout=5) as timer: + size = len(handle.read()) + + assert timer.time is not None + self._speed = size / timer.time + debug(f' speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)') + # Do not retry error + except urllib.error.URLError as error: + debug(f' speed: ({error}), skip') + self._speed = 0 + # Do retry error + except (http.client.IncompleteRead, ConnectionResetError) as error: + debug(f' speed: ({error}), retry') + # Catch all + except Exception as error: + debug(f' speed: ({error}), skip') + self._speed = 0 + + retry += 1 + + if self._speed is None: + self._speed = 0 + + return self._speed + + @property + def latency(self) -> float | None: + """ + Latency measures the milliseconds between one ICMP request & response. + It only does so once because we check if self._latency is None, and a ICMP timeout result in -1 + We do this because some hosts blocks ICMP so we'll have to rely on .speed() instead which is slower. + """ + if self._latency is None: + debug(f'Checking latency for {self.url}') + assert self._hostname is not None + self._latency = ping(self._hostname, timeout=2) + debug(f' latency: {self._latency}') + + return self._latency + + @classmethod + @field_validator('score', mode='before') + def validate_score(cls, value: float) -> int | None: + if value is not None: + value = round(value) + debug(f' score: {value}') + + return value + + @model_validator(mode='after') + def debug_output(self) -> Self: + from ..args import arch_config_handler + + self._hostname, *port = urllib.parse.urlparse(self.url).netloc.split(':', 1) + self._port = int(port[0]) if port and len(port) >= 1 else None + + if arch_config_handler.args.verbose: + debug(f'Loaded mirror {self._hostname}' + (f' with current score of {self.score}' if self.score else '')) + return self class MirrorStatusListV3(BaseModel): - cutoff: int - last_check: datetime.datetime - num_checks: int - urls: list[MirrorStatusEntryV3] - version: int + cutoff: int + last_check: datetime.datetime + num_checks: int + urls: list[MirrorStatusEntryV3] + version: int - @model_validator(mode='before') - @classmethod - def check_model( - cls, - data: dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]], - ) -> dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]]: - if data.get('version') == 3: - return data + @model_validator(mode='before') + @classmethod + def check_model( + cls, + data: dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]], + ) -> dict[str, int | datetime.datetime | list[MirrorStatusEntryV3]]: + if data.get('version') == 3: + return data - raise ValueError('MirrorStatusListV3 only accepts version 3 data from https://archlinux.org/mirrors/status/json/') + raise ValueError('MirrorStatusListV3 only accepts version 3 data from https://archlinux.org/mirrors/status/json/') @dataclass class MirrorRegion: - name: str - urls: list[str] + name: str + urls: list[str] - def json(self) -> dict[str, list[str]]: - return {self.name: self.urls} + def json(self) -> dict[str, list[str]]: + return {self.name: self.urls} - @override - def __eq__(self, other: object) -> bool: - if not isinstance(other, MirrorRegion): - return NotImplemented - return self.name == other.name + @override + def __eq__(self, other: object) -> bool: + if not isinstance(other, MirrorRegion): + return NotImplemented + return self.name == other.name class SignCheck(Enum): - Never = 'Never' - Optional = 'Optional' - Required = 'Required' + Never = 'Never' + Optional = 'Optional' + Required = 'Required' class SignOption(Enum): - TrustedOnly = 'TrustedOnly' - TrustAll = 'TrustAll' + TrustedOnly = 'TrustedOnly' + TrustAll = 'TrustAll' class _CustomRepositorySerialization(TypedDict): - name: str - url: str - sign_check: str - sign_option: str + name: str + url: str + sign_check: str + sign_option: str @dataclass class CustomRepository: - name: str - url: str - sign_check: SignCheck - sign_option: SignOption - - def table_data(self) -> dict[str, str]: - return { - 'Name': self.name, - 'Url': self.url, - 'Sign check': self.sign_check.value, - 'Sign options': self.sign_option.value, - } - - def json(self) -> _CustomRepositorySerialization: - return { - 'name': self.name, - 'url': self.url, - 'sign_check': self.sign_check.value, - 'sign_option': self.sign_option.value, - } - - @classmethod - def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: - configs = [] - for arg in args: - configs.append( - cls( - arg['name'], - arg['url'], - SignCheck(arg['sign_check']), - SignOption(arg['sign_option']), - ), - ) - - return configs + name: str + url: str + sign_check: SignCheck + sign_option: SignOption + + def table_data(self) -> dict[str, str]: + return { + 'Name': self.name, + 'Url': self.url, + 'Sign check': self.sign_check.value, + 'Sign options': self.sign_option.value, + } + + def json(self) -> _CustomRepositorySerialization: + return { + 'name': self.name, + 'url': self.url, + 'sign_check': self.sign_check.value, + 'sign_option': self.sign_option.value, + } + + @classmethod + def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: + configs = [] + for arg in args: + configs.append( + cls( + arg['name'], + arg['url'], + SignCheck(arg['sign_check']), + SignOption(arg['sign_option']), + ), + ) + + return configs @dataclass class CustomServer: - url: str + url: str - def table_data(self) -> dict[str, str]: - return {'Url': self.url} + def table_data(self) -> dict[str, str]: + return {'Url': self.url} - def json(self) -> dict[str, str]: - return {'url': self.url} + def json(self) -> dict[str, str]: + return {'url': self.url} - @classmethod - def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: - configs = [] - for arg in args: - configs.append( - cls(arg['url']), - ) + @classmethod + def parse_args(cls, args: list[dict[str, str]]) -> list[Self]: + configs = [] + for arg in args: + configs.append( + cls(arg['url']), + ) - return configs + return configs class _MirrorConfigurationSerialization(TypedDict): - mirror_regions: dict[str, list[str]] - custom_servers: list[CustomServer] - optional_repositories: list[str] - custom_repositories: list[_CustomRepositorySerialization] + mirror_regions: dict[str, list[str]] + custom_servers: list[CustomServer] + optional_repositories: list[str] + custom_repositories: list[_CustomRepositorySerialization] @dataclass class MirrorConfiguration: - mirror_regions: list[MirrorRegion] = field(default_factory=list) - custom_servers: list[CustomServer] = field(default_factory=list) - optional_repositories: list[Repository] = field(default_factory=list) - custom_repositories: list[CustomRepository] = field(default_factory=list) - - @property - def region_names(self) -> str: - return '\n'.join(m.name for m in self.mirror_regions) - - @property - def custom_server_urls(self) -> str: - return '\n'.join(s.url for s in self.custom_servers) - - def json(self) -> _MirrorConfigurationSerialization: - regions = {} - for m in self.mirror_regions: - regions.update(m.json()) - - return { - 'mirror_regions': regions, - 'custom_servers': self.custom_servers, - 'optional_repositories': [r.value for r in self.optional_repositories], - 'custom_repositories': [c.json() for c in self.custom_repositories], - } - - def custom_servers_config(self) -> str: - config = '' - - if self.custom_servers: - config += '## Custom Servers\n' - for server in self.custom_servers: - config += f'Server = {server.url}\n' - - return config.strip() - - def regions_config(self, speed_sort: bool = True) -> str: - from ..mirror.mirror_handler import mirror_list_handler - config = '' - - for mirror_region in self.mirror_regions: - sorted_stati = mirror_list_handler.get_status_by_region( - mirror_region.name, - speed_sort=speed_sort, - ) - - config += f'\n\n## {mirror_region.name}\n' - - for status in sorted_stati: - config += f'Server = {status.server_url}\n' - - return config - - def repositories_config(self) -> str: - config = '' - - for repo in self.custom_repositories: - config += f'\n\n[{repo.name}]\n' - config += f'SigLevel = {repo.sign_check.value} {repo.sign_option.value}\n' - config += f'Server = {repo.url}\n' - - return config - - @classmethod - def parse_args( - cls, - args: dict[str, Any], - backwards_compatible_repo: list[Repository] = [], - ) -> Self: - config = cls() - - mirror_regions = args.get('mirror_regions', []) - if mirror_regions: - for region, urls in mirror_regions.items(): - config.mirror_regions.append(MirrorRegion(region, urls)) - - if args.get('custom_servers'): - config.custom_servers = CustomServer.parse_args(args['custom_servers']) - - # backwards compatibility with the new custom_repository - if 'custom_mirrors' in args: - config.custom_repositories = CustomRepository.parse_args(args['custom_mirrors']) - if 'custom_repositories' in args: - config.custom_repositories = CustomRepository.parse_args(args['custom_repositories']) - - if 'optional_repositories' in args: - config.optional_repositories = [Repository(r) for r in args['optional_repositories']] - - if backwards_compatible_repo: - for r in backwards_compatible_repo: - if r not in config.optional_repositories: - config.optional_repositories.append(r) - - return config + mirror_regions: list[MirrorRegion] = field(default_factory=list) + custom_servers: list[CustomServer] = field(default_factory=list) + optional_repositories: list[Repository] = field(default_factory=list) + custom_repositories: list[CustomRepository] = field(default_factory=list) + + @property + def region_names(self) -> str: + return '\n'.join(m.name for m in self.mirror_regions) + + @property + def custom_server_urls(self) -> str: + return '\n'.join(s.url for s in self.custom_servers) + + def json(self) -> _MirrorConfigurationSerialization: + regions = {} + for m in self.mirror_regions: + regions.update(m.json()) + + return { + 'mirror_regions': regions, + 'custom_servers': self.custom_servers, + 'optional_repositories': [r.value for r in self.optional_repositories], + 'custom_repositories': [c.json() for c in self.custom_repositories], + } + + def custom_servers_config(self) -> str: + config = '' + + if self.custom_servers: + config += '## Custom Servers\n' + for server in self.custom_servers: + config += f'Server = {server.url}\n' + + return config.strip() + + def regions_config(self, speed_sort: bool = True) -> str: + from ..mirror.mirror_handler import mirror_list_handler + + config = '' + + for mirror_region in self.mirror_regions: + sorted_stati = mirror_list_handler.get_status_by_region( + mirror_region.name, + speed_sort=speed_sort, + ) + + config += f'\n\n## {mirror_region.name}\n' + + for status in sorted_stati: + config += f'Server = {status.server_url}\n' + + return config + + def repositories_config(self) -> str: + config = '' + + for repo in self.custom_repositories: + config += f'\n\n[{repo.name}]\n' + config += f'SigLevel = {repo.sign_check.value} {repo.sign_option.value}\n' + config += f'Server = {repo.url}\n' + + return config + + @classmethod + def parse_args( + cls, + args: dict[str, Any], + backwards_compatible_repo: list[Repository] = [], + ) -> Self: + config = cls() + + mirror_regions = args.get('mirror_regions', []) + if mirror_regions: + for region, urls in mirror_regions.items(): + config.mirror_regions.append(MirrorRegion(region, urls)) + + if args.get('custom_servers'): + config.custom_servers = CustomServer.parse_args(args['custom_servers']) + + # backwards compatibility with the new custom_repository + if 'custom_mirrors' in args: + config.custom_repositories = CustomRepository.parse_args(args['custom_mirrors']) + if 'custom_repositories' in args: + config.custom_repositories = CustomRepository.parse_args(args['custom_repositories']) + + if 'optional_repositories' in args: + config.optional_repositories = [Repository(r) for r in args['optional_repositories']] + + if backwards_compatible_repo: + for r in backwards_compatible_repo: + if r not in config.optional_repositories: + config.optional_repositories.append(r) + + return config diff --git a/tests/test_mirrorlist.py b/tests/test_mirrorlist.py index aa94e51e0c..9ceeaad238 100644 --- a/tests/test_mirrorlist.py +++ b/tests/test_mirrorlist.py @@ -1,6 +1,6 @@ from pathlib import Path -from archinstall.lib.mirrors import MirrorListHandler +from archinstall.lib.mirror.mirror_handler import MirrorListHandler def test_mirrorlist_no_country(mirrorlist_no_country_fixture: Path) -> None: