File size: 3,508 Bytes
b6cd2aa 7ed539d b6cd2aa 7ed539d 87402ad b6cd2aa 87402ad b6cd2aa 87402ad b6cd2aa 8e6dd44 7ed539d b6cd2aa 7ed539d 8e6dd44 87402ad 8e6dd44 87402ad b6cd2aa 7ed539d b6cd2aa 8e6dd44 b6cd2aa 7ed539d b6cd2aa 7ed539d 87402ad 8e6dd44 7ed539d b6cd2aa 7ed539d b6cd2aa 8e6dd44 7ed539d b6cd2aa 7ed539d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 |
import re
import yaml
import requests
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
def extract_and_transform_proxies(input_text):
print("原始输入数据:")
print(input_text)
print("------------------------")
# 尝试直接解析整个输入作为YAML
try:
data = yaml.safe_load(input_text)
if isinstance(data, dict) and 'proxies' in data:
proxies_list = data['proxies']
else:
proxies_list = data
except yaml.YAMLError:
# 如果整个输入不是有效的YAML，尝试提取proxies部分
proxies_match = re.search(r'proxies:\s*\n((?:[-\s]*{.*\n?)*)', input_text, re.MULTILINE)
if proxies_match:
proxies_text = proxies_match.group(1)
try:
proxies_list = yaml.safe_load(proxies_text)
except yaml.YAMLError:
# 如果YAML解析失败，尝试使用正则表达式解析
proxies_list = re.findall(r'{(.*?)}', proxies_text, re.DOTALL)
proxies_list = [dict(item.strip().split(': ') for item in proxy.split(',')) for proxy in proxies_list]
else:
return "未找到proxies部分"
if not proxies_list:
return "未找到有效的代理配置"
transformed_proxies = []
for proxy in proxies_list:
if isinstance(proxy, str):
# 如果proxy是字符串，尝试解析为字典
try:
proxy = dict(item.strip().split(': ') for item in proxy.split(','))
except ValueError:
continue
if proxy.get('type') == 'ss':
name = proxy.get('name', '').strip()
server = proxy.get('server', '').strip()
port = proxy.get('port', '').strip()
cipher = proxy.get('cipher', '').strip()
password = proxy.get('password', '').strip()
udp = 'true' if proxy.get('udp') in [True, 'true', 'True'] else 'false'
transformed = f"{name} = ss, {server}, {port}, encrypt-method={cipher}, password={password}, udp-relay={udp}"
transformed_proxies.append(transformed)
return "\n".join(transformed_proxies) if transformed_proxies else "未找到有效的SS代理配置"
class RequestHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed_path = urlparse(self.path)
query_params = parse_qs(parsed_path.query)
if 'url' in query_params:
url = query_params['url'][0]
try:
response = requests.get(url)
response.raise_for_status()
input_text = response.text
result = extract_and_transform_proxies(input_text)
self.send_response(200)
self.send_header('Content-type', 'text/plain; charset=utf-8')
self.end_headers()
self.wfile.write(result.encode('utf-8'))
except requests.RequestException as e:
self.send_error(500, f"Error fetching data: {str(e)}")
else:
self.send_error(400, "Missing 'url' parameter")
def log_message(self, format, *args):
# 重写此方法以禁用默认的服务器日志
pass
def run_server(port=8080):
server_address = ('', port)
httpd = HTTPServer(server_address, RequestHandler)
print(f"Server running on port {port}")
httpd.serve_forever()
if __name__ == "__main__":
run_server()
|