File size: 4,092 Bytes
b6cd2aa 7ed539d af22b4b 90475ed 78baf01 75a98c8 af22b4b 1ef33a6 b6cd2aa af22b4b 1ef33a6 b6cd2aa 1ef33a6 dfd834e 1ef33a6 b6cd2aa 85d3c28 1ef33a6 85d3c28 1ef33a6 85d3c28 af22b4b 1ef33a6 af22b4b 75a98c8 af22b4b 75a98c8 af22b4b 75a98c8 1ef33a6 af22b4b 75a98c8 af22b4b b6cd2aa af22b4b 1ef33a6 af22b4b 8e6dd44 af22b4b 85d3c28 af22b4b b6cd2aa 7ed539d af22b4b 85d3c28 |
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 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
import re
import yaml
import aiohttp
import asyncio
import datetime
import sys
import traceback
from aiohttp import web
from urllib.parse import parse_qs
from cachetools import TTLCache
from functools import partial
# 创建一个TTL缓存，最多存储1000个项目，每个项目的有效期为30分钟
cache = TTLCache(maxsize=1000, ttl=1800)
async def fetch_url(url, session):
async with session.get(url) as response:
return await response.text()
async def extract_and_transform_proxies(input_text):
# 使用正则表达式提取代理信息
pattern = r'([a-zA-Z0-9+/=]+)(?:@|:\/\/)([^:]+):(\d+)'
matches = re.findall(pattern, input_text)
proxies = []
for match in matches:
encoded_info, server, port = match
try:
decoded_info = base64.b64decode(encoded_info).decode('utf-8')
method, password = decoded_info.split(':')
proxy = {
'name': f'{server}:{port}',
'type': 'ss',
'server': server,
'port': int(port),
'cipher': method,
'password': password
}
proxies.append(proxy)
except:
continue
# 转换为YAML格式
yaml_data = yaml.dump({'proxies': proxies}, allow_unicode=True)
return yaml_data
async def log_request(request, response):
print(f"{datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - "
f"Request: {request.method} {request.path} - "
f"Response: {response.status}", flush=True)
@web.middleware
async def logging_middleware(request, handler):
response = await handler(request)
await log_request(request, response)
return response
async def handle_request(request):
if request.path == '/':
query_params = parse_qs(request.query_string)
if 'url' in query_params:
url = query_params['url'][0]
force_refresh = 'nocache' in query_params
if not force_refresh and url in cache:
print(f"Cache hit for URL: {url}", flush=True)
return web.Response(text=cache[url], content_type='text/plain')
try:
print(f"Fetching URL: {url}", flush=True)
async with aiohttp.ClientSession() as session:
input_text = await fetch_url(url, session)
print(f"URL content length: {len(input_text)}", flush=True)
result = await extract_and_transform_proxies(input_text)
print(f"Transformed result length: {len(result)}", flush=True)
# 将结果存入缓存
cache[url] = result
return web.Response(text=result, content_type='text/plain')
except Exception as e:
print(f"Error processing request: {str(e)}", flush=True)
traceback.print_exc()
return web.Response(text=f"Error: {str(e)}", status=500)
else:
usage_guide = """
<html>
<body>
<h1>代理配置转换工具</h1>
<p>使用方法：在URL参数中提供包含代理配置的网址。</p>
<p>示例：<code>http://localhost:8080/?url=https://example.com/path-to-proxy-config</code></p>
<p>强制刷新缓存：<code>http://localhost:8080/?url=https://example.com/path-to-proxy-config&nocache</code></p>
</body>
</html>
"""
return web.Response(text=usage_guide, content_type='text/html')
else:
return web.Response(text="Not Found", status=404)
async def init_app():
app = web.Application(middlewares=[logging_middleware])
app.router.add_get('/', handle_request)
return app
if __name__ == "__main__":
print(f"===== Application Startup at {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')} =====")
print("Server running on port 8080")
web.run_app(init_app(), port=8080, print=lambda _: None) # Disable default startup message
|