用下载方法查找CF最快下载速度的IP

cf_speed.py

import requests
import threadpool
from datetime import datetime

g_cnt = 0

def requests_test(ip):
    global g_cnt
    g_cnt = g_cnt + 1
    g_cnt_str = '[' + str(g_cnt) + ']'
    try:
        dl_speed = list()
        for i in range(5):
            headers = {
                'Host': 'vpsro.imghost.tech'
            }
            response = requests.get("http://" + ip + "/LookingGlass/1GB.test", stream=True, headers=headers,
                                    timeout=1)
            dl = 0
            dt = datetime.now()
            # 每块16KB,越大分块可能越快.
            for data in response.iter_content(chunk_size=16384):
                dl += len(data)
                # 大于10秒直接出结果,10秒内下载完你大概是火箭.
                if (datetime.now() - dt).seconds > 10:
                    dl_speed.append(round((dl / 10) / 1024 / 1024, 2))
                    break
        print(g_cnt_str + ip + '速度:' + str(dl_speed[0]) + ' MB/s | ' + str(dl_speed[1]) + ' MB/s | ' + str(
            dl_speed[2]) + ' MB/s | ' + str(dl_speed[3]) + ' MB/s | ' + str(dl_speed[4]) + ' MB/s')
    except requests.exceptions.ConnectionError:
        print(g_cnt_str + '无法使用(失效):' + ip)
    except requests.exceptions.ReadTimeout:
        print(g_cnt_str + '无法使用(低速):' + ip)
    except requests.exceptions.ChunkedEncodingError:
        print(g_cnt_str + '无法使用(错误):' + ip)
    except KeyboardInterrupt:
        exit(0)
    except:
        print(g_cnt_str + '无法使用(未知):' + ip)


with open('cf_ips.txt') as f:
    ips = f.read().split('\n')

    # 要注意总带宽容量,每线程至少需要1Mbps才能消化.
    pool = threadpool.ThreadPool(5)  # 线程池设置,最多同时跑N个线程
    tasks = threadpool.makeRequests(requests_test, ips)
    [pool.putRequest(task) for task in tasks]
    pool.wait()

你好，根据手动优选Cloudflare最快IP – 得到最快回源IP里面的方法，获取到优选CF IP之后该如何使用才能加速YouTube视频加载速度？

1GB.test 文件是否已经失效