INNER CODE UNIT · Python
classify
ermaozi/get_subscribe · check_nodes.py:86
def classify(directory='subscribe'):
directory = Path(directory)
config = yaml.safe_load((directory / 'clash.yml').read_text(encoding='utf-8'))
uris = (directory / 'v2ray.txt').read_text(encoding='utf-8').splitlines()
proxies = config['proxies']
proxy_addresses = [endpoint(proxy) for proxy in proxies]
uri_addresses = [endpoint(uri) for uri in uris]
addresses = list(dict.fromkeys(a for a in proxy_addresses + uri_addresses if a is not None))
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
results = dict(zip(addresses, pool.map(tcp_status, addresses)))
summary = {'checked_at': datetime.now(timezone.utc).isoformat(), 'method': 'TCP connect',
'timeout_seconds': TIMEOUT, 'unique_endpoints': len(addresses), 'counts': {}}
all_names = {proxy['name'] for proxy in proxies}
for status in STATUSES:
selected = [p for p, address in zip(proxies, proxy_addresses) if results.get(address, 'untested') == status]
selected_uris = [uri for uri, address in zip(uris, uri_addresses) if results.get(address, 'untested') == status]
removed = all_names - {proxy['name'] for proxy in selected}
filtered = {**config, 'proxies': selected}