archlinux_mirror_list_updater.py

#!/bin/env python3


# run doing `./archlinux_mirror_list_updater > /etc/pacman.d/mirrorlist`


import urllib.request
import re
import time
import sys
from datetime import datetime
import signal

country = 'US'
mirror_list = f'https://archlinux.org/mirrorlist/?country={country}&protocol=http&protocol=https&ip_version=4&ip_version=6&use_mirror_status=on'

def eprint(*args, **kwargs):
    print(*args, file=sys.stderr, **kwargs)

def signal_handler(signum, frame):
    eprint('Sending timeout exception')
    raise TimeoutError('signal timeout')

signal.signal(signal.SIGALRM, signal_handler)

def time_url(url):
    try:
        signal.alarm(8)
        start = time.time()
        with urllib.request.urlopen(url, timeout=6) as request:
            response = request.read()
        stop = time.time()
        signal.alarm(0)
    except urllib.request.HTTPError:
        eprint(f'failed {url}')
        return None
    except urllib.request.URLError as err:
        if 'timed out' in str(err):
            eprint(f'timed out {url}')
            return None
        raise
    except TimeoutError:
        eprint(f'timed out {url}')
        return None
    rate = len(response) / (stop - start)
    eprint(f'Timing {url} got {rate/1000:.2f} kB/sec')
    return rate

def get_mirrors():
    with urllib.request.urlopen(mirror_list) as request:
        response = request.read()
    for line in response.decode('ascii').splitlines():
        if m := re.match('.*Server = (.*)$', line):
            yield m.group(1)

def get_test_url(url):
    # should be around a 5-10 Mb file
    return url.replace('$repo', 'extra').replace('$arch', 'x86_64') + '/extra.db'

def main():
    mirrors = list(get_mirrors())

    mirrors = [m for m in mirrors if not ('http:' in m and m.replace('http:', 'https:') in mirrors)]  # do not include mirrors with different protocals

    eprint(f'Got {len(mirrors)} mirrors to check')

    timings = [(time_url(get_test_url(u)), u) for u in mirrors]
    signal.alarm(0)
    timings = [(a,b) for a,b in timings if a is not None]
    timings = list(reversed(sorted(timings)))

    if len(timings) < .3*len(mirrors):
        eprint('Not enough mirrors successfully connected, error')
        sys.exit(1)

    print('# Pacman Mirror List')
    print(f'# Generated at {datetime.now()}')
    print()

    for rate, server in timings:
        print(f'# Measured at {rate/1000:.2f} kB/sec')
        print(f'Server = {server}')
    print()

if __name__ == '__main__':
    main()