99 lines
3.3 KiB
Python
99 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import datetime
|
|
# import dns # TODO: replace server['ipv4'] with IPv4 address(es)? etc.
|
|
import json
|
|
import re
|
|
import sys
|
|
from urllib.request import urlopen
|
|
##
|
|
import iso3166
|
|
|
|
|
|
servers_json_url = 'https://www.archlinux.org/mirrors/status/json/'
|
|
protos = ('http', 'https')
|
|
|
|
|
|
class MirrorIdx(object):
|
|
def __init__(self,
|
|
country = None,
|
|
proto = None,
|
|
json_url = servers_json_url,
|
|
name_re = None,
|
|
ipv4 = None,
|
|
ipv6 = None,
|
|
isos = None):
|
|
_tmpargs = locals()
|
|
del (_tmpargs['self'])
|
|
for k, v in _tmpargs.items():
|
|
setattr(self, k, v)
|
|
self.validateParams()
|
|
self.servers_json = {}
|
|
self.servers = []
|
|
self.servers_with_scores = []
|
|
self.ranked_servers = []
|
|
self.fetchJSON()
|
|
self.buildServers()
|
|
self.rankServers()
|
|
|
|
def fetchJSON(self):
|
|
with urlopen(self.json_url) as u:
|
|
self.servers_json = json.load(u)
|
|
return()
|
|
|
|
def buildServers(self):
|
|
_limiters = (self.proto, self.ipv4, self.ipv6, self.isos)
|
|
_filters = list(_limiters)
|
|
_filters.extend([self.name_re, self.country])
|
|
_filters = tuple(_filters)
|
|
for s in self.servers_json['urls']:
|
|
if not s['active']:
|
|
continue
|
|
if not any(_filters):
|
|
self.servers.append(s.copy())
|
|
if s['score']:
|
|
self.servers_with_scores.append(s)
|
|
continue
|
|
# These are based on string values.
|
|
if self.name_re:
|
|
if not self.name_re.search(s['url']):
|
|
continue
|
|
if self.country:
|
|
if self.country != s['country_code']:
|
|
continue
|
|
# These are regular True/False switches
|
|
match = False
|
|
# We want to be *very* explicit about the ordering and inclusion/exclusion of these.
|
|
# They MUST match the order of _limiters.
|
|
values = []
|
|
for k in ('protocol', 'ipv4', 'ipv6', 'isos'):
|
|
values.append(s[k])
|
|
valid = all([v for k, v in zip(_limiters, values) if k])
|
|
if valid:
|
|
self.servers.append(s)
|
|
if s['score']:
|
|
self.servers_with_scores.append(s)
|
|
return()
|
|
|
|
def rankServers(self):
|
|
self.ranked_servers = sorted(self.servers_with_scores, key = lambda i: i['score'])
|
|
return()
|
|
|
|
def validateParams(self):
|
|
if self.proto and self.proto.lower() not in protos:
|
|
err = '{0} must be one of: {1}'.format(self.proto, ', '.join([i.upper() for i in protos]))
|
|
raise ValueError(err)
|
|
elif self.proto:
|
|
self.proto = self.proto.upper()
|
|
if self.country and self.country.upper() not in iso3166.countries:
|
|
err = ('{0} must be a valid ISO-3166-1 ALPHA-2 country code. '
|
|
'See https://en.wikipedia.org/wiki/List_of_ISO_3166_country_codes'
|
|
'#Current_ISO_3166_country_codes').format(self.country)
|
|
raise ValueError()
|
|
elif self.country:
|
|
self.country = self.country.upper()
|
|
if self.name_re:
|
|
self.name_re = re.compile(self.name_re)
|
|
return()
|