app.py

import re
import yaml
import aiohttp
import asyncio
import datetime
import sys
import traceback
from aiohttp import web, ClientTimeout, TCPConnector
from urllib.parse import parse_qs
from cachetools import TTLCache

cache = TTLCache(maxsize=1000, ttl=1800)  # 30 minutes cache

CHROME_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36"

async def fetch_url(url, session, max_retries=3):
    headers = {"User-Agent": CHROME_USER_AGENT}
    for attempt in range(max_retries):
        try:
            async with session.get(url, headers=headers, timeout=ClientTimeout(total=40)) as response:
                response.raise_for_status()
                content = await response.read()
                return content.decode('utf-8', errors='ignore')
        except aiohttp.ClientError as e:
            print(f"Attempt {attempt + 1} failed: {str(e)}", flush=True)
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(1)

async def extract_and_transform_proxies(input_text):
    try:
        data = yaml.safe_load(input_text)
        if isinstance(data, dict) and 'proxies' in data:
            proxies_list = data['proxies']
        elif isinstance(data, list):
            proxies_list = data
        else:
            proxies_match = re.search(r'proxies:\s*\n((?:[-\s]*{.*\n?)*)', input_text, re.MULTILINE)
            if proxies_match:
                proxies_text = proxies_match.group(1)
                proxies_list = yaml.safe_load(proxies_text)
            else:
                return "未找到有效的代理配置"
    except yaml.YAMLError:
        return "YAML解析错误"

    if not proxies_list:
        return "未找到有效的代理配置"

    transformed_proxies = []

    for proxy in proxies_list:
        if proxy.get('type') in ['ss', 'trojan']:
            name = proxy.get('name', '').strip()
            server = proxy.get('server', '').strip()
            port = str(proxy.get('port', '')).strip()
            
            parts = [f"{name} = {proxy['type']}, {server}, {port}"]
            
            if proxy['type'] == 'ss':
                if 'cipher' in proxy:
                    parts.append(f"encrypt-method={proxy['cipher'].strip()}")
            elif proxy['type'] == 'trojan':
                if 'password' in proxy:
                    parts.append(f"password={proxy['password'].strip()}")
                if 'sni' in proxy:
                    parts.append(f"sni={proxy['sni'].strip()}")
                if 'skip-cert-verify' in proxy:
                    parts.append(f"skip-cert-verify={str(proxy['skip-cert-verify']).lower()}")
            
            if 'udp' in proxy:
                parts.append(f"udp-relay={'true' if proxy['udp'] in [True, 'true', 'True'] else 'false'}")

            transformed_proxies.append(", ".join(parts))

    return "\n".join(transformed_proxies) if transformed_proxies else "未找到有效的SS或Trojan代理配置"

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]
            
            if 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(connector=TCPConnector(ssl=False)) 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>
            </body>
            </html>
            """
            return web.Response(text=usage_guide, content_type='text/html')
    else:
        return web.Response(text="Not Found", status=404)

@web.middleware
async def logging_middleware(request, handler):
    start_time = datetime.datetime.now()
    try:
        response = await handler(request)
        end_time = datetime.datetime.now()
        print(f"{end_time.strftime('%Y-%m-%d %H:%M:%S')} - {request.remote} - \"{request.method} {request.path}\" {response.status} {response.content_length}", flush=True)
        print(f"Request processing time: {end_time - start_time}", flush=True)
        return response
    except Exception as e:
        end_time = datetime.datetime.now()
        print(f"Error occurred: {str(e)}", flush=True)
        print(f"Request processing time: {end_time - start_time}", flush=True)
        print("Traceback:", flush=True)
        traceback.print_exc()
        return web.Response(text=f"Internal Server Error: {str(e)}", status=500)

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)