Spaces:

sfun
/

sub

Paused

App Files Files Community

sfun commited on Aug 21, 2024

Commit

af22b4b

verified ·

1 Parent(s): bfdbef9

Update app.py

Browse files

Files changed (1) hide show

app.py +41 -59

app.py CHANGED Viewed

@@ -1,21 +1,24 @@
 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):
     try:
-        # 尝试解析整个输入作为YAML
         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部分
             proxies_match = re.search(r'proxies:\s*\n((?:[-\s]*{.*\n?)*)', input_text, re.MULTILINE)
             if proxies_match:
                 proxies_text = proxies_match.group(1)
@@ -69,61 +72,40 @@ def extract_and_transform_proxies(input_text):
     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 = """
-                <html>
-                <body>
-                <h1>代理配置转换工具</h1>
-                <p>使用方法：在URL参数中提供包含代理配置的网址。</p>
-                <p>示例：<code>http://localhost:8080/?url=https://example.com/path-to-proxy-config</code></p>
-                </body>
-                </html>
-                """
-                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()

 import re
 import yaml
+import aiohttp
+import asyncio
 import datetime
 import sys
+from aiohttp import web
+from urllib.parse import parse_qs
+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):
     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)
     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]
             timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
+            print(f"{timestamp} - {request.remote} - GET /?url={url}")
+            try:
+                async with aiohttp.ClientSession() as session:
+                    input_text = await fetch_url(url, session)
+                result = await extract_and_transform_proxies(input_text)
+                return web.Response(text=result, content_type='text/plain')
+            except Exception as e:
+                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)
+async def init_app():
+    app = web.Application()
+    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)