Update app.py

app.py

@@ -1,64 +1,49 @@
-from flask import Flask, request, jsonify
-import requests
-import threading
-import time
-import logging
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
+import httpx
+import asyncio
+
+app = FastAPI()app = Flask(__name__)
+
+# Enable CORS
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Allows all origins
+    allow_credentials=True,
+    allow_methods=["*"],  # Allows all methods
+    allow_headers=["*"],  # Allows all headers
+)
+
+class ProxyTest(BaseModel):
+    proxy: str
+
+@app.post("/test-proxy")
+async def test_proxy(proxy_test: ProxyTest):
+    proxy = proxy_test.proxy
+    test_url = "http://httpbin.org/ip"
+    timeout = 10
 
-app = Flask(__name__)
-
-# Set up logging
-logging.basicConfig(level=logging.DEBUG)
-
-def test_proxy(proxy):
     try:
-        start_time = time.time()
-        response = requests.get('http://httpbin.org/ip', proxies={'http': proxy, 'https': proxy}, timeout=5)
-        end_time = time.time()
-        return {
-            'proxy': proxy,
-            'working': response.status_code == 200,
-            'time': round((end_time - start_time) * 1000, 2)
-        }
-    except:
-        return {
-            'proxy': proxy,
-            'working': False,
-            'time': None
-        }
-
-def test_proxies_threaded(proxies):
-    results = []
-    threads = []
-
-    def worker(proxy):
-        result = test_proxy(proxy)
-        results.append(result)
-
-    for proxy in proxies:
-        thread = threading.Thread(target=worker, args=(proxy,))
-        threads.append(thread)
-        thread.start()
-
-    for thread in threads:
-        thread.join()
-
-    return results
-
-@app.route('/test-proxies', methods=['POST'])
-def test_proxies():
-    data = request.json
-    logging.debug(f"Received data: {data}")
-
-    proxies = data.get('proxies', [])
-
-    if not isinstance(proxies, list):
-        return jsonify({'error': 'Proxies must be a list'}), 400
-
-    if not proxies:
-        return jsonify({'error': 'No proxies provided'}), 400
-
-    results = test_proxies_threaded(proxies)
-    return jsonify(results)
-
-if __name__ == '__main__':
-    app.run(host='0.0.0.0', port=7860, debug=True)
+        async with httpx.AsyncClient(proxies={"http://": f"http://{proxy}", "https://": f"http://{proxy}"}) as client:
+            start_time = asyncio.get_event_loop().time()
+            response = await client.get(test_url, timeout=timeout)
+            end_time = asyncio.get_event_loop().time()
+
+        if response.status_code == 200:
+            response_time = round((end_time - start_time) * 1000, 2)  # Convert to milliseconds
+            return {
+                "status": "success",
+                "message": f"Proxy {proxy} is working",
+                "response_time": f"{response_time} ms"
+            }
+        else:
+            raise HTTPException(status_code=400, detail=f"Proxy test failed with status code: {response.status_code}")
+    except httpx.TimeoutException:
+        raise HTTPException(status_code=408, detail=f"Proxy {proxy} timed out after {timeout} seconds")
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Proxy test failed: {str(e)}")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)    uvicorn.run(app, host="0.0.0.0", port=8000)