| index : reflector32 | |
| Archlinux32 fork of reflector | gitolite user |
| summaryrefslogtreecommitdiff |
| -rw-r--r-- | Reflector.py | 964 |
diff --git a/Reflector.py b/Reflector.py index b81f910..bcbb82f 100644 --- a/Reflector.py +++ b/Reflector.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -# Copyright (C) 2012, 2013 Xyne +# Copyright (C) 2012-2019 Xyne # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License @@ -22,6 +22,7 @@ import datetime import errno import getpass import http.client +import itertools import json import logging import os @@ -37,10 +38,287 @@ import time import urllib.error import urllib.request +################################## Constants ################################### +NAME = 'Reflector' + +URL = 'https://www.archlinux.org/mirrors/status/json/' + +DISPLAY_TIME_FORMAT = '%Y-%m-%d %H:%M:%S UTC' +PARSE_TIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ' +PARSE_TIME_FORMAT_WITH_USEC = '%Y-%m-%dT%H:%M:%S.%fZ' + +DB_SUBPATH = 'core/os/x86_64/core.db' + +MIRROR_URL_FORMAT = '{0}{1}/os/{2}' +MIRRORLIST_ENTRY_FORMAT = "Server = " + MIRROR_URL_FORMAT + "\n" + +DEFAULT_CONNECTION_TIMEOUT = 5 +DEFAULT_CACHE_TIMEOUT = 300 +DEFAULT_N_THREADS = 5 + +SORT_TYPES = { + 'age' : 'last server synchronization', + 'rate' : 'download rate', + 'country': 'server\'s location', + 'score' : 'MirrorStatus score', + 'delay' : 'MirrorStatus delay', +} + + +################################# IO Functions ################################# + +def get_cache_file(): + ''' + Get a nearly XDG-compliant cache directory. PyXDG is not used to avoid the + external dependency. It is not fully compliant because it omits the + application name, but the mirror status file can be reused by other + applications and this stores no other files. + ''' + base_name = 'mirrorstatus.json' + cache_dir = os.getenv('XDG_CACHE_HOME', default=os.path.expanduser('~/.cache')) + try: + os.makedirs(cache_dir, exist_ok=True) + # Raised by makedirs if permissions do not match umask + except FileExistsError: + pass + return os.path.join(cache_dir, base_name) + + + +def get_mirrorstatus( + connection_timeout=DEFAULT_CONNECTION_TIMEOUT, + cache_timeout=DEFAULT_CACHE_TIMEOUT +): + ''' + Retrieve the mirror status JSON object. The downloaded data will be cached + locally and re-used within the cache timeout period. Returns the object and + the local cache's modification time. + ''' + cache_path = get_cache_file() + try: + mtime = os.path.getmtime(cache_path) + invalid = (time.time() - mtime) > cache_timeout + except FileNotFoundError: + mtime = None + invalid = True + + try: + if invalid: + with urllib.request.urlopen(URL, None, connection_timeout) as h: + obj = json.loads(h.read().decode()) + with open(cache_path, 'w') as h: + json.dump(obj, h, sort_keys=True, indent=2) + mtime = time.time() + else: + with open(cache_path, 'r') as h: + obj = json.load(h) + + return obj, mtime + except (IOError, urllib.error.URLError, socket.timeout) as e: + raise MirrorStatusError(str(e)) + + + +################################ Miscellaneous ################################# + +def get_logger(): + ''' + Get the logger used by this module. Use this to be sure that the right logger + is used. + ''' + return logging.getLogger(NAME) + + + +def format_last_sync(mirrors): + ''' + Parse and format the "last_sync" field. + ''' + for m in mirrors: + last_sync = calendar.timegm(time.strptime(m['last_sync'], PARSE_TIME_FORMAT)) + m.update(last_sync=last_sync) + yield m + + + +def count_countries(mirrors): + ''' + Count the mirrors in each country. + ''' + countries = dict() + for m in mirrors: + k = (m['country'], m['country_code']) + if not any(k): + continue + try: + countries[k] += 1 + except KeyError: + countries[k] = 1 + return countries + + + +################################### Sorting #################################### + +def sort(mirrors, by=None, n_threads=DEFAULT_N_THREADS): + ''' + Sort mirrors by different criteria. + ''' + # Ensure that "mirrors" is a list that can be sorted. + if not isinstance(mirrors, list): + mirrors = list(mirrors) + + if by == 'age': + mirrors.sort(key=lambda m: m['last_sync'], reverse=True) + + elif by == 'rate': + rates = rate(mirrors, n_threads=n_threads) + mirrors = sorted(mirrors, key=lambda m: rates[m['url']], reverse=True) + + else: + try: + mirrors.sort(key=lambda m: m[by]) + except KeyError: + raise MirrorStatusError('attempted to sort mirrors by unrecognized criterion: "{}"'.format(by)) + + return mirrors + + +#################################### Rating #################################### + +def rate_rsync(db_url, connection_timeout=DEFAULT_CONNECTION_TIMEOUT): + ''' + Download a database via rsync and return the time and rate of the download. + ''' + rsync_cmd = [ + 'rsync', + '-avL', '--no-h', '--no-motd', + '--contimeout={}'.format(connection_timeout), + db_url + ] + try: + with tempfile.TemporaryDirectory() as tmpdir: + t0 = time.time() + subprocess.check_call( + rsync_cmd + [tmpdir], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL + ) + dt = time.time() - t0 + size = os.path.getsize( + os.path.join(tmpdir, os.path.basename(DB_SUBPATH)) + ) + r = size / dt + return dt, r + except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): + return None, 0 + + + +def rate_http(db_url, connection_timeout=DEFAULT_CONNECTION_TIMEOUT): + ''' + Download a database via any protocol supported by urlopen and return the time + and rate of the download. + ''' + req = urllib.request.Request(url=db_url) + try: + with urllib.request.urlopen(req, None, connection_timeout) as f: + t0 = time.time() + size = len(f.read()) + dt = time.time() - t0 + r = size / (dt) + return dt, r + except (OSError, urllib.error.HTTPError, http.client.HTTPException): + return None, 0 + + + +def rate(mirrors, n_threads=DEFAULT_N_THREADS, connection_timeout=DEFAULT_CONNECTION_TIMEOUT): + ''' + Rate mirrors by timing the download the core repo's database for each one. + ''' + # Ensure that mirrors is not a generator so that its length can be determined. + if not isinstance(mirrors, tuple): + mirrors = tuple(mirrors) + + if not mirrors: + return None + + # At least 1 thread and not more than the number of mirrors. + n_threads = max(1, min(n_threads, len(mirrors))) + + # URL input queue. + q_in = queue.Queue() + # URL, elapsed time and rate output queue. + q_out = queue.Queue() + + + def worker(): + while True: + # To stop a thread, an integer will be inserted in the input queue. Each + # thread will increment it and re-insert it until it equals the + # threadcount. After encountering the integer, the thread exits the loop. + url = q_in.get() + + if isinstance(url, int): + if url < n_threads: + q_in.put(url + 1) + + else: + db_url = url + DB_SUBPATH + scheme = urllib.parse.urlparse(url).scheme + + if scheme == 'rsync': + dt, r = rate_rsync(db_url, connection_timeout) + else: + dt, r = rate_http(db_url, connection_timeout) + + q_out.put((url, dt, r)) + + q_in.task_done() + + + workers = tuple(threading.Thread(target=worker) for _ in range(n_threads)) + for w in workers: + w.daemon = True + w.start() + + url_len = max(len(m['url']) for m in mirrors) + logger = get_logger() + for m in mirrors: + url = m['url'] + logger.info("rating {}".format(url)) + q_in.put(url) + + # To exit the threads. + q_in.put(0) + q_in.join() + + header_fmt = '{{:{:d}s}} {{:>14s}} {{:>9s}}'.format(url_len) + logger.info(header_fmt.format('Server', 'Rate', 'Time')) + fmt = '{{:{:d}s}} {{:8.2f}} KiB/s {{:7.2f}} s'.format(url_len) + + # Loop over the mirrors just to ensure that we get the rate for each mirror. + # The value in the loop does not (necessarily) correspond to the mirror. + rates = dict() + for _ in mirrors: + url, dt, r = q_out.get() + kibps = r / 1024.0 + logger.info(fmt.format(url, kibps, dt)) + rates[url] = r + q_out.task_done() + + return rates + + + +############################## MirrorStatusError ############################### -# Generic MirrorStatus Exception class MirrorStatusError(Exception): + ''' + Common base exception raised by this module. + ''' def __init__(self, msg): self.msg = msg def __str__(self): @@ -48,44 +326,168 @@ class MirrorStatusError(Exception): -def get_cache_file(): - bname = 'mirrorstatus.json' - path = os.getenv('XDG_CACHE_HOME') - if path: - try: - os.makedirs(path, exist_ok=True) - # Raised if permissions do not match umask - except FileExistsError: - pass - return os.path.join(path, bname) + +############################## MirrorStatusFilter ############################## + +class MirrorStatusFilter(): + + def __init__( + self, + min_completion_pct=1.0, + countries=None, + protocols=None, + include=None, + exclude=None, + age=None, + isos=False, + ipv4=False, + ipv6=False + ): + self.min_completion_pct = min_completion_pct + self.countries = tuple(c.upper() for c in countries) if countries else None + self.protocols = protocols + self.include = tuple(re.compile(r) for r in include) if include else None + self.exclude = tuple(re.compile(r) for r in exclude) if exclude else None + self.age = age + self.isos = isos + self.ipv4 = ipv4 + self.ipv6 = ipv6 + + + def filter_mirrors(self, mirrors): + # Filter unsynced mirrors. + mirrors = (m for m in mirrors if m['last_sync']) + + # Parse the last sync time. + mirrors = format_last_sync(mirrors) + + # Filter by completion "percent" [0-1]. + mirrors = (m for m in mirrors if m['completion_pct'] >= self.min_completion_pct) + + # Filter by countries. + if self.countries: + mirrors = ( + m for m in mirrors + if m['country'].upper() in self.countries + or m['country_code'].upper() in self.countries + ) + + # Filter by protocols. + if self.protocols: + mirrors = (m for m in mirrors if m['protocol'] in self.protocols) + + # Filter by include expressions. + if self.include: + mirrors = (m for m in mirrors if any(r.search(m['url']) for r in self.include)) + + # Filter by exclude expressions. + if self.exclude: + mirrors = (m for m in mirrors if not any(r.search(m['url']) for r in self.exclude)) + + # Filter by age. The age is given in hours and converted to seconds. Servers + # with a last refresh older than the age are omitted. + if self.age and self.age > 0: + t = time.time() + a = self.age * 60**2 + mirrors = (m for m in mirrors if (m['last_sync'] + a) >= t) + + # The following does not work. Only the final iteration affects "mirrors". + # TODO: Understand exactly why the code above works but the loop doesn't. + # for field in ('isos', 'ipv4', 'ipv6'): + # if getattr(self, field): + # mirrors = (m for m in mirrors if m[field]) + + # Filter by ISO hosing. + if self.isos: + mirrors = (m for m in mirrors if m['isos']) + + # Filter by IPv4 support. + if self.ipv4: + mirrors = (m for m in mirrors if m['ipv4']) + + # Filter by IPv6 support. + if self.ipv6: + mirrors = (m for m in mirrors if m['ipv6']) + + yield from mirrors + + + +################################## Formatting ################################## + +def format_mirrorlist(mirror_status, mtime, include_country=False, command=None): + if command is None: + command = '?' else: - return '/tmp/.{}.{}'.format(getpass.getuser(), bname) - - - -class MirrorStatus(object): - # JSON URI - URL = 'https://www.archlinux.org/mirrors/status/json/' - # Mirror URL format. Accepts server base URL, repository, and architecture. - MIRROR_URL_FORMAT = '{0}{1}/os/{2}' - MIRRORLIST_ENTRY_FORMAT = "Server = " + MIRROR_URL_FORMAT + "\n" - DISPLAY_TIME_FORMAT = '%Y-%m-%d %H:%M:%S UTC' - PARSE_TIME_FORMAT = '%Y-%m-%dT%H:%M:%SZ' - # Required for the last_check field, which oddly includes microseconds. - PARSE_TIME_FORMAT_WITH_USEC = '%Y-%m-%dT%H:%M:%S.%fZ' - # Recognized list sort types and their descriptions. - SORT_TYPES = { - 'age' : 'last server synchronization', - 'rate' : 'download rate', - 'country': 'server\'s location', - 'score' : 'MirrorStatus score', - 'delay' : 'MirrorStatus delay', - } - # Known repositories, i.e. those that should be on each mirror. - # Used to replace the "$repo" variable. - # TODO - # Avoid using a hard-coded list. - # See https://bugs.archlinux.org/task/32895 + command = 'reflector ' + ' '.join(pipes.quote(x) for x in command) + + last_check = mirror_status['last_check'] + # For some reason the "last_check" field included microseconds. + try: + parsed_last_check = datetime.datetime.strptime( + last_check, + PARSE_TIME_FORMAT_WITH_USEC, + ).timetuple() + except ValueError: + parsed_last_check = datetime.datetime.strptime( + last_check, + PARSE_TIME_FORMAT, + ).timetuple() + + width = 80 + colw = 11 + header = '# Arch Linux mirrorlist generated by Reflector #'.center(width, '#') + border = '#' * len(header) + mirrorlist = '' + mirrorlist = '{}\n{}\n{}\n'.format(border, header, border) + \ + '\n' + \ + '\n'.join( + '# {{:<{:d}s}} {{}}'.format(colw).format(k, v) for k, v in ( + ('With:', command), + ('When:', time.strftime(DISPLAY_TIME_FORMAT, time.gmtime())), + ('From:', URL), + ('Retrieved:', time.strftime(DISPLAY_TIME_FORMAT, time.gmtime(mtime))), + ('Last Check:', time.strftime(DISPLAY_TIME_FORMAT, parsed_last_check)) + ) + ) + \ + '\n\n' + + country = None + + mirrors = mirror_status['urls'] + for mirror in mirrors: + # Include country tags. This is intended for lists that are sorted by + # country. + if include_country: + c = '{} [{}]'.format(mirror['country'], mirror['country_code']) + if c != country: + if country: + mirrorlist += '\n' + mirrorlist += '# {}\n'.format(c) + country = c + mirrorlist += MIRRORLIST_ENTRY_FORMAT.format(mirror['url'], '$repo', '$arch') + + if mirrors: + return mirrorlist + else: + return None + + + + + +############################ MirrorStatus Retriever ############################ + +class MirrorStatus(): + ''' + This is a legacy class that will likely be removed in the future. It + previously held most of this module's functionality until it was refactored + into more modular functions. Seemingly pointless code is still used by + importers of this module. + ''' + + # TODO: move these to another module or remove them completely + # Related: https://bugs.archlinux.org/task/32895 REPOSITORIES = ( 'community', 'community-staging', @@ -99,325 +501,85 @@ class MirrorStatus(object): 'staging', 'testing' ) - - # Known system architectures, as used to replace the "$arch" variable. + # Officially supported system architectures. ARCHITECTURES = ['x86_64'] - # Initialize - # refresh_interval: - # The cached list will be replaced after this many seconds have passed. - # 0 effectively disables caching. - # Caching is only useful if the object persists, e.g. if it were embedded - # in a server. + MIRROR_URL_FORMAT = MIRROR_URL_FORMAT + MIRRORLIST_ENTRY_FORMAT = MIRRORLIST_ENTRY_FORMAT + + def __init__( self, - refresh_interval=0, - verbose=False, - connection_timeout=5, -# download_timeout=None, - cache_timeout=300, - min_completion_pct=1., - threads=5 + connection_timeout=DEFAULT_CONNECTION_TIMEOUT, + cache_timeout=DEFAULT_CACHE_TIMEOUT, + min_completion_pct=1.0, + threads=DEFAULT_N_THREADS ): - self.refresh_interval = refresh_interval - - # Last modification time of the json object. - self.json_mtime = 0 - # The parsed JSON object. - self.json_obj = {} - # Display extra information. - self.verbose = verbose - # Connection timeout self.connection_timeout = connection_timeout - # Download timeout -# self.download_timeout = download_timeout - # Cache timeout self.cache_timeout = cache_timeout - # Minimum completion percent, for filtering mirrors. self.min_completion_pct = min_completion_pct - # Threads self.threads = threads + self.mirror_status = None + self.ms_mtime = 0 def retrieve(self): - """Retrieve the current mirror status JSON data.""" - self.json_obj = None - json_str = None - save_json = False - - cache_file = get_cache_file() - if self.cache_timeout > 0: - save_json = True - try: - mtime = os.path.getmtime(cache_file) - if time.time() - mtime < self.cache_timeout: - try: - with open(cache_file) as f: - self.json_obj = json.load(f) - self.json_mtime = mtime - save_json = False - except IOError as e: - raise MirrorStatusError('failed to load cached JSON data ({})'.format(e)) - except OSError as e: - if e.errno != errno.ENOENT: - raise MirrorStatusError('failed to get cache file mtime ({})'.format(e)) - - if not self.json_obj: - try: - with urllib.request.urlopen(MirrorStatus.URL, None, self.connection_timeout) as f: - json_str = f.read() - self.json_obj = json.loads(json_str.decode()) - self.json_mtime = time.time() - except (urllib.error.URLError, socket.timeout) as e: - raise MirrorStatusError('failed to retrieve mirror data: ({})'.format(e)) - except ValueError as e: - raise MirrorStatusError('failed to parse retrieved mirror data: ({})'.format(e)) - - try: - # Remove servers that have not synced, and parse the "last_sync" times for - # comparison later. - mirrors = self.json_obj['urls'] - # Filter incomplete mirrors and mirrors that haven't synced. - mirrors = list( - m for m in mirrors - if m['last_sync'] - and m['completion_pct'] >= self.min_completion_pct - ) - # Parse 'last_sync' times for future comparison. - for mirror in mirrors: - mirror['last_sync'] = calendar.timegm( - time.strptime(mirror['last_sync'], - MirrorStatus.PARSE_TIME_FORMAT) - ) - self.json_obj['urls'] = mirrors - except KeyError: - raise MirrorStatusError('failed to parse retrieved mirror data (the format may have changed or there may be a transient error)') - - if save_json and json_str: - try: - with open(cache_file, 'wb') as f: - f.write(json_str) - except IOError as e: - raise MirrorStatusError('failed to cache JSON data ({})'.format(e)) - - - + self.mirror_status, self.ms_mtime = get_mirrorstatus( + connection_timeout=self.connection_timeout, + cache_timeout=self.cache_timeout + ) def get_obj(self): - """Return the JSON object, retrieving new data if necessary.""" - if not self.json_obj \ - or time.time() > (self.json_mtime + self.refresh_interval): + ''' + Get the JSON mirror status. + ''' + t = time.time() + if (t - self.ms_mtime) > self.cache_timeout: self.retrieve() - - return self.json_obj + return self.mirror_status def get_mirrors(self): - """Get the mirrors.""" - return self.get_obj()['urls'] - + ''' + Get the mirror from the mirror status. + ''' + obj = self.get_obj() + try: + return obj['urls'] + except KeyError: + raise MirrorStatusError('no mirrors detected in mirror status output') - def filter( - self, - mirrors=None, - countries=None, - regexes=None, # TODO: remove - include=None, - exclude=None, - age=None, - protocols=None - ): - """Filter using different parameters.""" - # TODO: remove - if regexes: -# raise MirrorStatusError('The "regexes" keyword has been deprecated and replaced by "include" and "exclude".') - if not include: - include = regexes - sys.stderr.write('''WARNING: The "regexes" keyword has been deprecated and replaced by "include" and "exclude". - Support will be soon removed without further warning.''') - if mirrors is None: - mirrors = self.get_mirrors() - t = time.time() - n = 0 - - # Make country arguments case-insensitive. - uc_countries = tuple(c.upper() for c in countries) if countries else None - for mirror in mirrors: - # Filter by country. - if countries \ - and not ( \ - mirror['country'].upper() in uc_countries or \ - mirror['country_code'].upper() in uc_countries \ - ): - continue - # Filter by protocol. - if protocols and not mirror['protocol'] in protocols: - continue - # Filter by regex. - # TODO: Find a better way to do this. - if include: - for regex in include: - if re.search(regex, mirror['url']): - break - else: - continue - if exclude: - discard = False - for regex in exclude: - if re.search(regex, mirror['url']): - discard = True - break - if discard: - continue - # Filter by hours since last sync. - if age and t > (age * 60**2 + mirror['last_sync']): - continue - - # Yield if we're still here. - yield mirror - - - - def sort(self, mirrors=None, by=None): - """Sort using different parameters.""" + def filter(self, mirrors=None, **kwargs): + ''' + Filter mirrors by various criteria. + ''' if mirrors is None: mirrors = self.get_mirrors() - # Ensure that "mirrors" is a list that can be sorted. - if not isinstance(mirrors, list): - mirrors = list(mirrors) + msf = MirrorStatusFilter(min_completion_pct=self.min_completion_pct, **kwargs) + yield from msf.filter_mirrors(mirrors) - if by == 'age': - mirrors.sort(key=lambda m: m['last_sync'], reverse=True) - elif by == 'rate': - mirrors = self.rate(mirrors) - elif by in ('country', 'country_code', 'delay', 'score'): - mirrors.sort(key=lambda m: m[by]) - return mirrors - - # Sort mirrors by download speed. Download speed will be calculated from the - # download time of the [core] database from each server. - # TODO: Consider ways to improve this. - # TODO: Consider the effects of threading (do the threads affect the results - # by competing for bandwidth?) - def rate(self, mirrors=None, threads=5): + def sort(self, mirrors=None, **kwargs): + ''' + Sort mirrors by various criteria. + ''' if mirrors is None: mirrors = self.get_mirrors() - if not threads: - threads = self.threads - # Ensure that "mirrors" is a list and not a generator. - if not isinstance(mirrors, list): - mirrors = list(mirrors) - - if not mirrors: - logging.warning('no mirrors selected for rating') - return mirrors - - # Ensure a sane number of threads. - if threads < 1: - threads = 1 - else: - threads = min(threads, len(mirrors)) - - rates = {} - - # URL input queue.Queue - q_in = queue.Queue() - # URL and rate output queue.Queue - q_out = queue.Queue() - def worker(): - while True: - url = q_in.get() - db_subpath = 'core/os/x86_64/core.db' - db_url = url + db_subpath - scheme = urllib.parse.urlparse(url).scheme - # Leave the rate as 0 if the connection fails. - # TODO: Consider more graceful error handling. - rate = 0 - dt = float('NaN') - - # urllib cannot handle rsync protocol - if scheme == 'rsync': - rsync_cmd = [ - 'rsync', - '-avL', '--no-h', '--no-motd', - '--contimeout={}'.format(self.connection_timeout), - db_url - ] - try: - with tempfile.TemporaryDirectory() as tmpdir: - t0 = time.time() - subprocess.check_call( - rsync_cmd + [tmpdir], - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL - ) - dt = time.time() - t0 - size = os.path.getsize(os.path.join( - tmpdir, - os.path.basename(db_subpath) - )) - rate = size / dt - except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError): - pass - else: - req = urllib.request.Request(url=db_url) - try: - t0 = time.time() - with urllib.request.urlopen(req, None, self.connection_timeout) as f: - size = len(f.read()) - dt = time.time() - t0 - rate = size / (dt) - except (OSError, urllib.error.HTTPError, http.client.HTTPException): - pass - q_out.put((url, rate, dt)) - q_in.task_done() - - # Launch threads - for i in range(threads): - t = threading.Thread(target=worker) - t.daemon = True - t.start() - - - # Load the input queue.Queue - url_len = max(len(m['url']) for m in mirrors) - for mirror in mirrors: - logging.info("rating {}".format(mirror['url'])) - q_in.put(mirror['url']) - - q_in.join() - - - # Get the results - # The "in mirrors" loop is just used to ensure that the right number of - # items is retrieved. + yield from sort(mirrors, n_threads=self.threads, **kwargs) - # Display some extra data. - header_fmt = '{{:{:d}s}} {{:>14s}} {{:>9s}}'.format(url_len) - logging.info(header_fmt.format('Server', 'Rate', 'Time')) - fmt = '{{:{:d}s}} {{:8.2f}} KiB/s {{:7.2f}} s'.format(url_len) - # Loop over the mirrors just to ensure that we get the rate for each mirror. - # The value in the loop does not (necessarily) correspond to the mirror. - for _ in mirrors: - url, rate, dt = q_out.get() - kibps = rate / 1024.0 - logging.info(fmt.format(url, kibps, dt)) - rates[url] = rate - q_out.task_done() - - - # Sort by rate. - rated_mirrors = [m for m in mirrors if rates[m['url']] > 0] - rated_mirrors.sort(key=lambda m: rates[m['url']], reverse=True) - - return rated_mirrors + [m for m in mirrors if rates[m['url']] == 0] + def rate(self, mirrors=None, **kwargs): + ''' + Sort mirrors by download speed. + ''' + if mirrors is None: + mirrors = self.get_mirrors() + yield from sort(mirrors, n_threads=self.threads, by='rate', **kwargs) @@ -427,85 +589,29 @@ class MirrorStatus(object): - - # Return a Pacman-formatted mirrorlist - # TODO: Reconsider the assumption that self.json_obj has been retrieved. def get_mirrorlist(self, mirrors=None, include_country=False, cmd=None): - if mirrors is None: - mirrors = self.get_mirrors() - if cmd is None: - cmd = '?' - else: - cmd = 'reflector ' + ' '.join(pipes.quote(x) for x in cmd) - - last_check = self.json_obj['last_check'] - # For some reason the "last_check" field included microseconds. - try: - parsed_last_check = datetime.datetime.strptime( - last_check, - self.PARSE_TIME_FORMAT_WITH_USEC, - ).timetuple() - except ValueError: - parsed_last_check = datetime.datetime.strptime( - last_check, - self.PARSE_TIME_FORMAT, - ).timetuple() - - width = 80 - colw = 11 - header = '# Arch Linux mirrorlist generated by Reflector #'.center(width, '#') - border = '#' * len(header) - mirrorlist = '{}\n{}\n{}\n'.format(border, header, border) + \ - '\n' + \ - '\n'.join( - '# {{:<{:d}s}} {{}}'.format(colw).format(k, v) for k, v in ( - ('With:', cmd), - ('When:', self.display_time(time.gmtime())), - ('From:', MirrorStatus.URL), - ('Retrieved:', self.display_time(time.gmtime(self.json_mtime))), - ('Last Check:', self.display_time(parsed_last_check)), - ) - ) + \ - '\n\n' - - country = None - - # mirrors may be a generator so "if mirrors" will not work - no_mirrors = True - for mirror in mirrors: - no_mirrors = False - # Include country tags. This is intended for lists that are sorted by - # country. - if include_country: - c = '{} [{}]'.format(mirror['country'], mirror['country_code']) - if c != country: - if country: - mirrorlist += '\n' - mirrorlist += '# {}\n'.format(c) - country = c - mirrorlist += MirrorStatus.MIRRORLIST_ENTRY_FORMAT.format(mirror['url'], '$repo', '$arch') - - if no_mirrors: - return None - else: - return mirrorlist + ''' + Get a Pacman-formatted mirrorlist. + ''' + obj = self.get_obj().copy() + if mirrors is not None: + if not isinstance(mirrors, list): + mirrors = list(mirrors) + obj['urls'] = mirrors + return format_mirrorlist(obj, self.ms_mtime, include_country=include_country, command=cmd) def list_countries(self): - countries = dict() - for m in self.get_mirrors(): - k = (m['country'], m['country_code']) - try: - countries[k] += 1 - except KeyError: - countries[k] = 1 - return countries - - + ''' + List countries along with a server count for each one. + ''' + mirrors = self.get_mirrors() + return count_countries(mirrors) +############################### argparse Actions ############################### class ListCountries(argparse.Action): ''' @@ -523,11 +629,12 @@ class ListCountries(argparse.Action): -def print_mirror_info(mirrors, time_fmt=MirrorStatus.DISPLAY_TIME_FORMAT): +def print_mirror_info(mirrors, time_fmt=DISPLAY_TIME_FORMAT): ''' Print information about each mirror to STDOUT. ''' if mirrors: + # mirrors = format_last_sync(mirrors) if not isinstance(mirrors, list): mirrors = list(mirrors) ks = sorted(k for k in mirrors[0].keys() if k != 'url') @@ -551,8 +658,8 @@ def add_arguments(parser): parser = argparse.ArgumentParser(description='retrieve and filter a list of the latest Arch Linux mirrors') parser.add_argument( - '--connection-timeout', type=int, metavar='n', default=5, - help='The number of seconds to wait before a connection times out.' + '--connection-timeout', type=int, metavar='n', default=DEFAULT_CONNECTION_TIMEOUT, + help='The number of seconds to wait before a connection times out. Default: %(default)s' ) # parser.add_argument( @@ -566,8 +673,8 @@ def add_arguments(parser): ) parser.add_argument( - '--cache-timeout', type=int, metavar='n', default=300, - help='The cache timeout in seconds for the data retrieved from the Arch Linux Mirror Status API. The default is 300 (5 minutes).' + '--cache-timeout', type=int, metavar='n', default=DEFAULT_CACHE_TIMEOUT, + help='The cache timeout in seconds for the data retrieved from the Arch Linux Mirror Status API. The default is %(default)s.' ) parser.add_argument( @@ -575,15 +682,15 @@ def add_arguments(parser): help='Save the mirrorlist to the given path.' ) - sort_help = '; '.join('"{}": {}'.format(k, v) for k, v in MirrorStatus.SORT_TYPES.items()) + sort_help = '; '.join('"{}": {}'.format(k, v) for k, v in SORT_TYPES.items()) parser.add_argument( - '--sort', choices=MirrorStatus.SORT_TYPES, + '--sort', choices=SORT_TYPES, help='Sort the mirrorlist. {}.'.format(sort_help) ) parser.add_argument( - '--threads', type=int, metavar='n', - help='The number of threads to use when rating mirrors.' + '--threads', type=int, metavar='n', default=DEFAULT_N_THREADS, + help='The maximum number of threads to use when rating mirrors. Default: %(default)s' ) parser.add_argument( @@ -653,6 +760,21 @@ def add_arguments(parser): help='Set the minimum completion percent for the returned mirrors. Check the mirrorstatus webpage for the meaning of this parameter. Default value: %(default)s.' ) + filters.add_argument( + '--isos', action='store_true', + help='Only return mirrors that host ISOs.' + ) + + filters.add_argument( + '--ipv4', action='store_true', + help='Only return mirrors that support IPv4.' + ) + + filters.add_argument( + '--ipv6', action='store_true', + help='Only return mirrors that support IPv6.' + ) + return parser @@ -680,7 +802,6 @@ def process_options(options, ms=None, mirrors=None): ''' if not ms: ms = MirrorStatus( - verbose=options.verbose, connection_timeout=options.connection_timeout, # download_timeout=options.download_timeout, cache_timeout=options.cache_timeout, @@ -698,20 +819,23 @@ def process_options(options, ms=None, mirrors=None): include=options.include, exclude=options.exclude, age=options.age, - protocols=options.protocols + protocols=options.protocols, + isos=options.isos, + ipv4=options.ipv4, + ipv6=options.ipv6 ) if options.latest and options.latest > 0: mirrors = ms.sort(mirrors, by='age') - mirrors = mirrors[:options.latest] + mirrors = itertools.islice(mirrors, options.latest) if options.score and options.score > 0: mirrors = ms.sort(mirrors, by='score') - mirrors = mirrors[:options.score] + mirrors = itertools.islice(mirrors, options.score) if options.fastest and options.fastest > 0: mirrors = ms.sort(mirrors, by='rate') - mirrors = mirrors[:options.fastest] + mirrors = itertools.islice(mirrors, options.fastest) if options.sort and not (options.sort == 'rate' and options.fastest): mirrors = ms.sort(mirrors, by=options.sort) @@ -723,8 +847,6 @@ def process_options(options, ms=None, mirrors=None): - - def main(args=None, configure_logging=False): if args: cmd = tuple(args) @@ -733,17 +855,25 @@ def main(args=None, configure_logging=False): options = parse_args(args) + # Configure logging. + logger = get_logger() + if configure_logging: if options.verbose: level = logging.INFO else: level = logging.WARNING - logging.basicConfig( - format='[{asctime:s}] {levelname:s}: {message:s}', + + logger.setLevel(level) + ch = logging.StreamHandler() + formatter = logging.Formatter( + fmt='[{asctime:s}] {levelname:s}: {message:s}', style='{', - datefmt='%Y-%m-%d %H:%M:%S', - level=level + datefmt='%Y-%m-%d %H:%M:%S' ) + ch.setFormatter(formatter) + logger.addHandler(ch) + try: ms, mirrors = process_options(options) |