import re
import yaml
import requests
import datetime
import sys
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
def extract_and_transform_proxies(input_text):
proxies_match = re.search(r'proxies:\s*\n((?:[-\s]*{.*\n?)*)', input_text, re.MULTILINE)
if not proxies_match:
return "未找到proxies部分"
proxies_text = proxies_match.group(1)
try:
proxies_list = yaml.safe_load(proxies_text)
except yaml.YAMLError:
proxies_list = []
for proxy_str in re.findall(r'{(.*?)}', proxies_text, re.DOTALL):
proxy_dict = {}
for item in proxy_str.split(','):
key, value = item.split(':', 1)
proxy_dict[key.strip()] = value.strip()
proxies_list.append(proxy_dict)
if not proxies_list:
return "未找到有效的代理配置"
transformed_proxies = []
for proxy in proxies_list:
if proxy.get('type') == 'ss':
name = proxy.get('name', '').strip()
server = proxy.get('server', '').strip()
port = str(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)
elif proxy.get('type') == 'trojan':
name = proxy.get('name', '').strip()
server = proxy.get('server', '').strip()
port = str(proxy.get('port', '')).strip()
password = proxy.get('password', '').strip()
sni = proxy.get('sni', '').strip()
skip_cert_verify = str(proxy.get('skip-cert-verify', '')).lower()
transformed = f"{name} = trojan, {server}, {port}, password={password}, sni={sni}, skip-cert-verify={skip_cert_verify}"
transformed_proxies.append(transformed)
return "\n".join(transformed_proxies) if transformed_proxies else "未找到有效的SS或Trojan代理配置"
class RequestHandler(BaseHTTPRequestHandler):
def log_request(self, code='-', size='-'):
if 'url=' in self.path:
timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
sys.stdout.write(f"{timestamp} - {self.client_address[0]} - \"{self.requestline}\" {code} {size}\n")
sys.stdout.flush()
def do_GET(self):
parsed_path = urlparse(self.path)
query_params = parse_qs(parsed_path.query)
if parsed_path.path == '/':
if 'url' in query_params:
url = query_params['url'][0]
sys.stdout.write(f"处理URL: {url}\n")
sys.stdout.flush()
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_response(200)
self.send_header('Content-type', 'text/html; charset=utf-8')
self.end_headers()
usage_guide = """
代理配置转换工具
使用方法：在URL参数中提供包含代理配置的网址。
示例：http://localhost:8080/?url=https://example.com/path-to-proxy-config
"""
self.wfile.write(usage_guide.encode('utf-8'))
else:
self.send_error(404, "Not Found")
def log_message(self, format, *args):
# 重写此方法以禁用默认的服务器日志
pass
def run_server(port=8080):
server_address = ('0.0.0.0', port)
httpd = HTTPServer(server_address, RequestHandler)
sys.stdout.write(f"===== Application Startup at {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====\n")
sys.stdout.write(f"Server running on port {port}\n")
sys.stdout.flush()
httpd.serve_forever()
if __name__ == "__main__":
run_server()