逛L站的时候发现工益节点其实大把多，各种订阅资源并不难找到，但问题就是节点太多了，一个工益订阅里甚至可以有数百个节点，于是如何利用这些节点就成了问题。
或者更进一步说，如何批量处理这些量很大，并且存活期可能并不长的节点才是问题。
而处理这些节点的目的当然是筛选出其中较为优质的部分，比如可能偶尔会有一两个家宽IP，或者不是家宽但是风险也很低的，又或者速度特别快的。虽然基本上等于沙堆里淘金 ，但金子确实贵呀 。
而我搜索了半天只找到了一个fulltclash项目可能比较适合处理这些…节点数据。但是我对它并不满意，因为它主要的用途其实还是测速和检测流媒体解锁情况，而我恰恰不是很需要这两个功能。毕竟我对直接访问迪士尼、openai、claude这些没有太大兴趣，毕竟我要么不用要么用第三方站
点，测速这东西也不好说，有的鸡场你敢测速他就敢封你号，而且要用的时候扔进clash里需要的时候实时测试就够了。
而且fulltclash部署起来还好麻烦。
哦，本站确实有个项目是检测clash订阅文件节点质量的，但是……限制太大了，而且用起来太麻烦。
2 Likes
fork一下改改检测脚本不就行了， 几个request请求而已
直接丢给Bot会不会不太好？
有道理，明天有空看看
插眼
- StairSpeedtest 本地运行直接 crash 了
- SSRSpeedN 安装运行起来比较复杂
- nodesCatch 闭源工具不敢用
- starudream/clash-speedtest 勉强能用，但不支持 Proxy Provider
- FullTclash 有一些机场在用，个人用起来比较麻烦
- GitHub - faceair/clash-speedtest: clash speedtest
以上参考来源：基于 Clash 核心的测速工具，帮我发现了我正在使用的机场是个垃圾 - V2EX
很多这种，但是你如果会程序建议自己写一个，比如我自己写的,供参考，其实很简单，就是用到了clash的api 批量检测一下重新生成可用的yaml
import itertools
import json
import string
import subprocess
import time
import urllib
import urllib.parse
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
import requests
import yaml
clash_path = "/Users/clashmeta" # 请替换为 Clash 可执行文件的实际路径
config_url = "https://sub.store/download/test?target=Clash" # clash配置url
config_path = "/Users/config.yaml" # 下载配置文件的保存路径
# 下载 Clash 配置文件
def download_config(url, dest_path):
try:
response = requests.get(url,verify=False)
response.raise_for_status()
with open(dest_path, 'wb') as file:
file.write(response.content)
print(f"Config downloaded to {dest_path}")
except Exception as e:
print(f"Failed to download config: {e}")
return False
return True
# 启动 Clash 进程
def start_clash(config_path):
process = subprocess.Popen([clash_path, "-f", config_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
time.sleep(5) # 等待 Clash 完全启动
print("Starting clash")
return process
# 停止 Clash 进程
def stop_clash(process):
process.terminate()
print("closing clash")
process.wait()
# 模拟 http_get 方法
def http_get(url, timeout, retry=3):
while retry > 0:
try:
response = requests.get(url, timeout=timeout)
if response.status_code == 200:
return response.text
except requests.exceptions.RequestException as e:
print(f"HTTP GET 请求失败: {e}")
retry -= 1
return None
def check_proxy(proxy, api_url, timeout=5000, test_url="http://www.gstatic.com/generate_204", delay=2500):
try:
proxy_name = urllib.parse.quote(proxy.get("name", ""))
url = f"http://{api_url}/proxies/{proxy_name}/delay?timeout={timeout}&url={urllib.parse.quote(test_url)}"
# Use http_get instead of requests.get
response_content = http_get(url=url, timeout=timeout / 1000, retry=2)
if not response_content:
return None
data = json.loads(response_content)
if data.get("delay", -1) > 0 and data.get("delay", -1) <= delay:
print(f"Proxy {proxy.get('name')} check passed")
return proxy
except Exception as e:
print(f"Proxy {proxy.get('name')} check failed: {e}")
return None
# 更新配置文件，添加必要的设置
# 更新配置文件，添加必要的设置并去重
def update_config(config_path):
try:
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)
proxies = config.get("proxies", [])
unique_proxies = []
unique_names = set()
for proxy in proxies:
if proxy.get("name") not in unique_names:
unique_proxies.append(proxy)
unique_names.add(proxy.get("name"))
# 更新配置
config.update({
'external-controller': '127.0.0.1:9090',
'log-level': 'info',
'mixed-port': 7890,
'mode': 'Rule',
'proxies': unique_proxies # 去重后的代理列表
})
with open(config_path, 'w', encoding='utf-8') as f:
yaml.dump(config, f, allow_unicode=True)
print(f'Config updated and saved to {config_path}')
except Exception as e:
print(f'Failed to update config: {e}')
# 生成新的配置文件
def generate_config(config_path, proxies):
# 过滤代理
external_config = filter_proxies(proxies)
# 初始配置模板
config = {
"mixed-port": 7890,
"external-controller": "127.0.0.1:9090",
"mode": "Rule",
"log-level": "silent",
"proxies": external_config["proxies"]
}
# DNS 设置
dns_settings = {
'enable': True,
'enhanced-mode': 'fake-ip',
'fake-ip-range': '198.18.0.1/16',
'default-nameserver': [
'114.114.114.114',
'223.5.5.5',
'8.8.8.8'
],
'nameserver': [
'https://doh.pub/dns-query'
]
}
config['dns'] = dns_settings
config['listeners'] = []
# 为每个代理生成监听器
for i, proxy in enumerate(external_config['proxies']):
listener = {
'name': f'mixed{i}',
'type': 'mixed',
'port': 42000 + i,
'proxy': proxy['name']
}
config['listeners'].append(listener)
# 保存配置文件
with open(config_path, "w+", encoding="utf8") as f:
yaml.dump(config, f, allow_unicode=True)
return config.get("proxies", [])
# 过滤代理
def filter_proxies(proxies):
config = {
"proxies": [],
"proxy-groups": [
{"name": "节点选择", "type": "select", "proxies": []},
],
"rules": ["MATCH,节点选择"],
}
# 按名字排序方便在节点相同时优先保留名字靠前的
proxies.sort(key=lambda p: str(p.get("name", "")))
unique_proxies, hosts = [], defaultdict(list)
for item in proxies:
if not proxies_exists(item, hosts):
unique_proxies.append(item)
key = f"{item.get('server')}:{item.get('port')}"
hosts[key].append(item)
# 防止多个代理节点名字相同导致clash配置错误
groups, unique_names = {}, set()
for key, group in itertools.groupby(unique_proxies, key=lambda p: p.get("name", "")):
items = groups.get(key, [])
items.extend(list(group))
groups[key] = items
# 优先保留不重复的节点的名字
unique_proxies = sorted(groups.values(), key=lambda x: len(x))
proxies.clear()
for items in unique_proxies:
size = len(items)
if size <= 1:
proxies.extend(items)
unique_names.add(items[0].get("name"))
continue
for i in range(size):
item = items[i]
mode = i % 26
factor = i // 26 + 1
letter = string.ascii_uppercase[mode]
name = "{}-{}{}".format(item.get("name"), factor, letter)
while name in unique_names:
mode += 1
factor = factor + mode // 26
mode = mode % 26
letter = string.ascii_uppercase[mode]
name = "{}-{}{}".format(item.get("name"), factor, letter)
item["name"] = name
proxies.append(item)
unique_names.add(name)
config["proxies"] += proxies
config["proxy-groups"][0]["proxies"] += list(unique_names)
return config
# 判断代理是否已经存在
def proxies_exists(proxy, hosts):
if not proxy:
return True
if not hosts:
return False
key = f"{proxy.get('server')}:{proxy.get('port')}"
proxies = hosts.get(key, [])
if not proxies:
return False
protocol = proxy.get("type", "")
if protocol == "http" or protocol == "socks5":
return True
elif protocol == "ss" or protocol == "trojan":
return any(p.get("password", "") == proxy.get("password", "") for p in proxies)
elif protocol == "ssr":
return any(
str(p.get("protocol-param", "")).lower() == str(proxy.get("protocol-param", "")).lower() for p in proxies
)
elif protocol == "vmess" or protocol == "vless":
return any(p.get("uuid", "") == proxy.get("uuid", "") for p in proxies)
elif protocol == "snell":
return any(p.get("psk", "") == proxy.get("psk", "") for p in proxies)
elif protocol == "tuic":
if proxy.get("token", ""):
return any(p.get("token", "") == proxy.get("token", "") for p in proxies)
return any(p.get("uuid", "") == proxy.get("uuid", "") for p in proxies)
elif protocol == "hysteria2":
return any(p.get("password", "") == proxy.get("password", "") for p in proxies)
elif protocol == "hysteria":
key = "auth-str" if "auth-str" in proxy else "auth_str"
value = proxy.get(key, "")
for p in proxies:
if "auth-str" in p and p.get("auth-str", "") == value:
return True
if "auth_str" in p and p.get("auth_str", "") == value:
return True
return False
def main():
# 下载配置文件
if not download_config(config_url, config_path):
return
update_config(config_path)
# 启动 Clash
clash_process = start_clash(config_path)
if not clash_process:
print("Failed to start Clash")
return
# 读取下载的配置文件
with open(config_path, 'r', encoding='utf8') as file:
config = yaml.safe_load(file)
proxies = config.get("proxies", [])
api_url = "127.0.0.1:9090" # Clash 的 API 地址
# 并发检测代理的可用性
available_proxies = []
with ThreadPoolExecutor(max_workers=20) as executor:
futures = {executor.submit(check_proxy, proxy, api_url): proxy for proxy in proxies}
for future in as_completed(futures):
result = future.result()
if result:
available_proxies.append(result)
# 停止 Clash
stop_clash(clash_process)
# 生成新的可用配置文件
generate_config(config_path, available_proxies)
print(f"Available proxies saved to f{config_path}")
if __name__ == "__main__":
main()
7 Likes
佬啊 
有空请看看这里
我试着跑了好几遍你的代码，发现每次都报错，大抵都是目标服务器积极拒绝。我只是个Python初学者，所以把报错日志扔给gpt分析后他回复我说可能是clash没有正确启动的原因，然后我就把你启动clash的函数扔给他改了几回，结果clash内核是能正常启动了，进行HTTP请求的时候还是报错，甚至是跟之前一样的错误……所以gpt还是一样的回答和解决方案。
这就给我整挺麻的，所以您看有时间这里交流一下吗？您跑脚本的时候正常吗？
脚本没问题，应该是环境问题，你可以手动进行http请求看看，或者改一下测试延迟的地址，有很多可以测试延迟地址的，你可以了解下clash dashboard，在浏览器里面发开面板看看是怎么测试延迟的，我的脚本就是模拟了这个过程而已
我用這個xream/lite-test，自動去重
插眼插眼插眼插眼
正好也有这个需求，cy