webshare/fetch_proxies.py

755 lines
23 KiB
Python
Executable File
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
Webshare SOCKS代理获取脚本
使用简单的下载接口直接获取代理列表
"""
import os
import requests
import yaml
import json
import base64
from typing import Optional, List, Dict
def load_api_key() -> Optional[str]:
"""
从.env文件或环境变量中加载API密钥
Returns:
API密钥或None
"""
# 首先尝试从.env文件读取
env_file = ".env"
if os.path.exists(env_file):
try:
with open(env_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
if '=' in line:
# 处理 TOKEN=xxx 格式
key, value = line.split('=', 1)
if key.strip() == 'TOKEN':
return value.strip()
else:
# 如果第一行不包含=则认为是直接的token
return line
except Exception as e:
print(f"读取.env文件时出错: {e}")
# 然后尝试从环境变量读取
api_key = os.getenv('TOKEN') or os.getenv('WEBSHARE_API_KEY')
if api_key:
return api_key.strip()
return None
def get_download_token(api_key: str) -> Optional[str]:
"""
获取下载token
Args:
api_key: API密钥
Returns:
下载token或None
"""
url = "https://proxy.webshare.io/api/v2/download_token/proxy_list/"
headers = {
"Authorization": f"Token {api_key}",
"Content-Type": "application/json"
}
try:
print("🔑 正在获取下载token...")
response = requests.post(url, headers=headers)
if response.status_code == 200:
data = response.json()
download_token = data.get("key") # API返回的字段是'key'而不是'download_token'
if download_token:
print("✅ 成功获取下载token")
return download_token
else:
print("❌ 响应中未找到下载token")
print(f"🔍 可用的键: {list(data.keys())}")
return None
else:
print(f"❌ 获取下载token失败 {response.status_code}: {response.text}")
return None
except requests.exceptions.RequestException as e:
print(f"获取下载token时出错: {e}")
return None
def download_proxies(download_token: str, country: str = "-", auth_method: str = "username",
endpoint_mode: str = "direct", location: str = ""):
"""
使用下载token获取代理列表
Args:
download_token: 下载token (不是API key)
country: 国家代码 (- 表示任意国家)
auth_method: 认证方式 (username 或 ipauth)
endpoint_mode: 端点模式 (direct 或 backbone)
location: 位置过滤 (可选)
Returns:
代理列表文本
"""
# 构建下载URL - 正确的格式
if location:
# URL编码位置参数
location_encoded = requests.utils.quote(location)
url = f"https://proxy.webshare.io/api/v2/proxy/list/download/{download_token}/{country}/any/{auth_method}/{endpoint_mode}/{location_encoded}/"
else:
# 没有位置过滤时,不包含最后的路径参数
url = f"https://proxy.webshare.io/api/v2/proxy/list/download/{download_token}/{country}/any/{auth_method}/{endpoint_mode}/"
try:
print(f"📡 正在从下载接口获取代理: {url}")
response = requests.get(url)
# 检查响应状态
if response.status_code == 404:
print("❌ 404错误 - 可能是URL格式不正确或token无效")
return None
elif response.status_code == 401:
print("❌ 401错误 - 下载token无效")
return None
elif response.status_code != 200:
print(f"❌ HTTP错误 {response.status_code}: {response.text}")
return None
# 检查是否返回了错误信息
if "invalid" in response.text.lower() or "error" in response.text.lower():
print(f"❌ API返回错误: {response.text}")
return None
return response.text
except requests.exceptions.RequestException as e:
print(f"下载代理列表时出错: {e}")
return None
def format_proxy_list(proxy_text: str, output_format: str = "socks5") -> list:
"""
格式化代理列表
Args:
proxy_text: 原始代理文本
output_format: 输出格式 (socks5, http, raw)
Returns:
格式化的代理列表
"""
if not proxy_text:
return []
lines = proxy_text.strip().split('\n')
formatted_proxies = []
for line in lines:
line = line.strip()
if not line:
continue
# 假设格式为: ip:port:username:password
parts = line.split(':')
if len(parts) >= 4:
ip, port, username, password = parts[0], parts[1], parts[2], parts[3]
if output_format == "socks5":
formatted_proxy = f"socks5://{username}:{password}@{ip}:{port}"
elif output_format == "http":
formatted_proxy = f"http://{username}:{password}@{ip}:{port}"
else: # raw
formatted_proxy = line
formatted_proxies.append(formatted_proxy)
elif len(parts) >= 2:
# 无认证格式: ip:port
ip, port = parts[0], parts[1]
if output_format == "socks5":
formatted_proxy = f"socks5://{ip}:{port}"
elif output_format == "http":
formatted_proxy = f"http://{ip}:{port}"
else: # raw
formatted_proxy = line
formatted_proxies.append(formatted_proxy)
return formatted_proxies
def save_proxies_to_file(proxies: list, filename: str = "socks_proxies.txt"):
"""
保存代理列表到文件
Args:
proxies: 代理列表
filename: 文件名
"""
try:
with open(filename, 'w', encoding='utf-8') as f:
for proxy in proxies:
f.write(f"{proxy}\n")
print(f"✅ 代理列表已保存到 {filename}")
except Exception as e:
print(f"❌ 保存文件时出错: {e}")
def parse_proxies(proxy_text: str) -> list:
"""
解析代理文本,提取代理信息
Args:
proxy_text: 代理文本
Returns:
代理信息列表
"""
proxies = []
lines = proxy_text.strip().split('\n')
for line in lines:
line = line.strip()
if not line:
continue
# 解析格式: host:port:username:password
parts = line.split(':')
if len(parts) >= 4:
proxy = {
'host': parts[0],
'port': parts[1],
'username': parts[2],
'password': ':'.join(parts[3:]) # 密码可能包含冒号
}
proxies.append(proxy)
return proxies
def save_proxies(proxies: list, filename: str, format_type: str = "raw"):
"""
保存代理列表到文件
Args:
proxies: 代理列表
filename: 文件名
format_type: 格式类型 ("raw", "http", "socks5")
"""
with open(filename, 'w', encoding='utf-8') as f:
for proxy in proxies:
if format_type == "raw":
# 原始格式: host:port:username:password
line = f"{proxy['host']}:{proxy['port']}:{proxy['username']}:{proxy['password']}"
elif format_type == "http":
# HTTP格式: http://username:password@host:port
line = f"http://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}"
elif format_type == "socks5":
# SOCKS5格式: socks5://username:password@host:port
line = f"socks5://{proxy['username']}:{proxy['password']}@{proxy['host']}:{proxy['port']}"
else:
line = f"{proxy['host']}:{proxy['port']}:{proxy['username']}:{proxy['password']}"
f.write(line + '\n')
print(f"📄 已保存 {len(proxies)} 个代理到 {filename} ({format_type}格式)")
def generate_clash_config(proxies: List[Dict[str, str]]) -> Dict:
"""
生成Clash配置文件
Args:
proxies: 代理列表
Returns:
Clash配置字典
"""
clash_proxies = []
proxy_names = []
for i, proxy in enumerate(proxies):
proxy_name = f"proxy-{i+1}"
proxy_names.append(proxy_name)
# 创建HTTP代理配置
clash_proxy = {
"name": proxy_name,
"type": "http",
"server": proxy['host'],
"port": int(proxy['port']),
"username": proxy['username'],
"password": proxy['password']
}
clash_proxies.append(clash_proxy)
# 创建完整的Clash配置
clash_config = {
"port": 7890,
"socks-port": 7891,
"allow-lan": False,
"mode": "rule",
"log-level": "info",
"external-controller": "127.0.0.1:9090",
"proxies": clash_proxies,
"proxy-groups": [
{
"name": "🚀 节点选择",
"type": "select",
"proxies": ["♻️ 自动选择", "🎯 故障转移"] + proxy_names
},
{
"name": "♻️ 自动选择",
"type": "url-test",
"proxies": proxy_names,
"url": "http://www.gstatic.com/generate_204",
"interval": 300
},
{
"name": "🎯 故障转移",
"type": "fallback",
"proxies": proxy_names,
"url": "http://www.gstatic.com/generate_204",
"interval": 300
},
{
"name": "🌍 国外媒体",
"type": "select",
"proxies": ["🚀 节点选择", "♻️ 自动选择", "🎯 故障转移"] + proxy_names
},
{
"name": "🍃 应用净化",
"type": "select",
"proxies": ["REJECT", "DIRECT"]
}
],
"rules": [
"DOMAIN-SUFFIX,google.com,🚀 节点选择",
"DOMAIN-SUFFIX,youtube.com,🌍 国外媒体",
"DOMAIN-SUFFIX,facebook.com,🌍 国外媒体",
"DOMAIN-SUFFIX,twitter.com,🌍 国外媒体",
"DOMAIN-SUFFIX,instagram.com,🌍 国外媒体",
"DOMAIN-SUFFIX,netflix.com,🌍 国外媒体",
"DOMAIN-KEYWORD,google,🚀 节点选择",
"DOMAIN-KEYWORD,youtube,🌍 国外媒体",
"DOMAIN-KEYWORD,facebook,🌍 国外媒体",
"DOMAIN-KEYWORD,github,🚀 节点选择",
"GEOIP,CN,DIRECT",
"MATCH,🚀 节点选择"
]
}
return clash_config
def save_clash_config(proxies: List[Dict[str, str]], filename: str = "clash_config.yaml"):
"""
保存Clash配置文件
Args:
proxies: 代理列表
filename: 文件名
"""
clash_config = generate_clash_config(proxies)
try:
with open(filename, 'w', encoding='utf-8') as f:
yaml.dump(clash_config, f, default_flow_style=False, allow_unicode=True, indent=2)
print(f"📄 已保存Clash配置到 {filename}")
except Exception as e:
print(f"❌ 保存Clash配置时出错: {e}")
def generate_singbox_config(proxies: List[Dict[str, str]]) -> Dict:
"""
生成SingBox配置
Args:
proxies: 代理列表
Returns:
SingBox配置字典
"""
outbounds = []
# 添加代理节点
for i, proxy in enumerate(proxies):
outbound = {
"type": "http",
"tag": f"proxy-{i+1}",
"server": proxy['host'],
"server_port": int(proxy['port']),
"username": proxy['username'],
"password": proxy['password']
}
outbounds.append(outbound)
# 添加选择器
proxy_tags = [outbound["tag"] for outbound in outbounds]
selector = {
"type": "selector",
"tag": "proxy",
"outbounds": proxy_tags + ["auto", "direct"]
}
# 添加自动选择
auto_select = {
"type": "urltest",
"tag": "auto",
"outbounds": proxy_tags,
"url": "https://www.gstatic.com/generate_204",
"interval": "1m",
"tolerance": 50
}
# 添加直连
direct = {
"type": "direct",
"tag": "direct"
}
# 添加阻断
block = {
"type": "block",
"tag": "block"
}
# 完整配置
config = {
"log": {
"level": "info"
},
"dns": {
"servers": [
{
"tag": "google",
"address": "tls://8.8.8.8",
"strategy": "prefer_ipv4"
},
{
"tag": "local",
"address": "223.5.5.5",
"strategy": "prefer_ipv4"
}
],
"rules": [
{
"outbound": "any",
"server": "local"
}
]
},
"inbounds": [
{
"type": "mixed",
"tag": "mixed-in",
"listen": "127.0.0.1",
"listen_port": 2080
}
],
"outbounds": [selector, auto_select] + outbounds + [direct, block],
"route": {
"rules": [
{
"protocol": "dns",
"outbound": "dns-out"
},
{
"network": "udp",
"port": 443,
"outbound": "block"
},
{
"geoip": "cn",
"outbound": "direct"
},
{
"domain_suffix": [
".cn",
".com.cn",
".net.cn",
".org.cn",
".gov.cn",
".edu.cn"
],
"outbound": "direct"
},
{
"domain_keyword": [
"baidu",
"taobao",
"qq",
"weixin",
"alipay",
"douyin",
"bilibili"
],
"outbound": "direct"
}
],
"auto_detect_interface": True
}
}
return config
def save_singbox_config(proxies: List[Dict[str, str]], filename: str = "singbox_config.json"):
"""
保存SingBox配置文件
Args:
proxies: 代理列表
filename: 文件名
"""
singbox_config = generate_singbox_config(proxies)
try:
with open(filename, 'w', encoding='utf-8') as f:
json.dump(singbox_config, f, indent=2, ensure_ascii=False)
print(f"📄 已保存SingBox配置到 {filename}")
except Exception as e:
print(f"❌ 保存SingBox配置时出错: {e}")
def generate_v2ray_subscription(proxies: List[Dict[str, str]]) -> str:
"""
生成V2Ray订阅链接
Args:
proxies: 代理列表
Returns:
Base64编码的订阅内容
"""
vmess_links = []
for i, proxy in enumerate(proxies):
# 生成VMess配置
vmess_config = {
"v": "2",
"ps": f"WebShare-{i+1}",
"add": proxy['host'],
"port": proxy['port'],
"id": "00000000-0000-0000-0000-000000000000", # 默认UUID实际使用时应该生成随机UUID
"aid": "0",
"scy": "auto",
"net": "tcp",
"type": "none",
"host": "",
"path": "",
"tls": "",
"sni": "",
"alpn": ""
}
# 将配置转换为JSON并编码
vmess_json = json.dumps(vmess_config, separators=(',', ':'))
vmess_b64 = base64.b64encode(vmess_json.encode('utf-8')).decode('utf-8')
vmess_link = f"vmess://{vmess_b64}"
vmess_links.append(vmess_link)
# 将所有链接合并并编码
subscription_content = '\n'.join(vmess_links)
subscription_b64 = base64.b64encode(subscription_content.encode('utf-8')).decode('utf-8')
return subscription_b64
def generate_v2ray_config(proxies: List[Dict[str, str]]) -> Dict:
"""
生成V2Ray客户端配置
Args:
proxies: 代理列表
Returns:
V2Ray配置字典
"""
outbounds = []
# 添加代理节点
for i, proxy in enumerate(proxies):
outbound = {
"tag": f"proxy-{i+1}",
"protocol": "http",
"settings": {
"servers": [
{
"address": proxy['host'],
"port": int(proxy['port']),
"users": [
{
"user": proxy['username'],
"pass": proxy['password']
}
]
}
]
}
}
outbounds.append(outbound)
# 添加直连
direct_outbound = {
"tag": "direct",
"protocol": "freedom",
"settings": {}
}
# 添加阻断
block_outbound = {
"tag": "blocked",
"protocol": "blackhole",
"settings": {}
}
# 完整配置
config = {
"log": {
"loglevel": "warning"
},
"inbounds": [
{
"tag": "socks",
"port": 1080,
"listen": "127.0.0.1",
"protocol": "socks",
"sniffing": {
"enabled": True,
"destOverride": ["http", "tls"]
},
"settings": {
"auth": "noauth",
"udp": True
}
},
{
"tag": "http",
"port": 1087,
"listen": "127.0.0.1",
"protocol": "http",
"sniffing": {
"enabled": True,
"destOverride": ["http", "tls"]
}
}
],
"outbounds": outbounds + [direct_outbound, block_outbound],
"routing": {
"rules": [
{
"type": "field",
"outboundTag": "blocked",
"protocol": ["bittorrent"]
},
{
"type": "field",
"outboundTag": "direct",
"domain": ["geosite:cn"]
},
{
"type": "field",
"outboundTag": "direct",
"ip": ["geoip:cn", "geoip:private"]
},
{
"type": "field",
"outboundTag": "proxy-1",
"network": "tcp,udp"
}
]
}
}
return config
def save_v2ray_subscription(proxies: List[Dict[str, str]], filename: str = "v2ray_subscription.txt"):
"""
保存V2Ray订阅文件
Args:
proxies: 代理列表
filename: 文件名
"""
try:
subscription = generate_v2ray_subscription(proxies)
with open(filename, 'w', encoding='utf-8') as f:
f.write(subscription)
print(f"📄 已保存V2Ray订阅到 {filename}")
except Exception as e:
print(f"❌ 保存V2Ray订阅时出错: {e}")
def save_v2ray_config(proxies: List[Dict[str, str]], filename: str = "v2ray_config.json"):
"""
保存V2Ray配置文件
Args:
proxies: 代理列表
filename: 文件名
"""
try:
config = generate_v2ray_config(proxies)
with open(filename, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=2, ensure_ascii=False)
print(f"📄 已保存V2Ray配置到 {filename}")
except Exception as e:
print(f"❌ 保存V2Ray配置时出错: {e}")
def main():
"""主函数"""
print("🚀 开始获取Webshare代理列表...")
# 加载API密钥
api_key = load_api_key()
if not api_key:
print("❌ 无法加载API密钥")
return
# 获取下载token
download_token = get_download_token(api_key)
if not download_token:
print("❌ 获取下载token失败")
return
# 下载代理列表
proxy_text = download_proxies(download_token)
if not proxy_text:
print("❌ 获取代理列表失败")
return
# 解析代理
proxies = parse_proxies(proxy_text)
if not proxies:
print("❌ 未找到有效的代理")
return
print(f"✅ 成功获取 {len(proxies)} 个代理")
# 保存原始代理列表
save_proxies(proxies, "proxies_raw.txt", format_type="raw")
# 保存HTTP格式代理列表
save_proxies(proxies, "proxies_http.txt", format_type="http")
# 保存SOCKS5格式代理列表
save_proxies(proxies, "proxies_socks5.txt", format_type="socks5")
# 生成Clash配置文件
save_clash_config(proxies, "clash_config.yaml")
# 生成SingBox配置文件
save_singbox_config(proxies, "singbox_config.json")
# 生成V2Ray订阅和配置文件
save_v2ray_subscription(proxies, "v2ray_subscription.txt")
save_v2ray_config(proxies, "v2ray_config.json")
print("🎉 代理列表已保存完成!")
print("📁 文件说明:")
print(" - proxies_raw.txt: 原始格式 (host:port:username:password)")
print(" - proxies_http.txt: HTTP格式 (http://username:password@host:port)")
print(" - proxies_socks5.txt: SOCKS5格式 (socks5://username:password@host:port)")
print(" - clash_config.yaml: Clash配置文件")
print(" - singbox_config.json: SingBox配置文件")
print(" - v2ray_subscription.txt: V2Ray订阅链接 (Base64编码)")
print(" - v2ray_config.json: V2Ray客户端配置文件")
if __name__ == "__main__":
main()