agenticworkflowsspace/proxyvault

1"""2FastAPI backend for ProxyUtility WebApp.3Streams proxy results in real-time via WebSocket.4Filters by country (location substring match) and ip_type.5Sends progress updates after every proxy test.6"""7import os, sys, asyncio, json, time8 9from fastapi import FastAPI, WebSocket, WebSocketDisconnect10from fastapi.middleware.cors import CORSMiddleware11from fastapi.staticfiles import StaticFiles12 13sys.path.insert(0, os.path.dirname(__file__))14 15from proxy_scraper import scrape_proxies16from proxy_tester import get_whoer_info, TEST_URL, TIMEOUT_SECONDS17 18import aiohttp19 20app = FastAPI()21 22app.add_middleware(23    CORSMiddleware,24    allow_origins=["*"],25    allow_credentials=True,26    allow_methods=["*"],27    allow_headers=["*"],28)29 30 31def parse_ip(proxy: str):32    if "@" in proxy:33        return proxy.split("@")[-1].split(":")[0]34    return proxy.split(":")[0]35 36 37def proxy_url_and_auth(proxy: str):38    if "@" in proxy:39        return f"http://{proxy}", None40    elif proxy.count(":") == 3:41        parts = proxy.split(":")42        auth = aiohttp.BasicAuth(parts[2], parts[3])43        return f"http://{parts[0]}:{parts[1]}", auth44    return f"http://{proxy}", None45 46 47async def test_and_enrich(48    session: aiohttp.ClientSession,49    proxy: str,50    semaphore: asyncio.Semaphore,51    country_filter: str,52    type_filter: str,53    counters: dict,54    ws: WebSocket,55):56    """Test one proxy, enrich with Whoer, stream progress + result."""57    async with semaphore:58        ip = parse_ip(proxy)59        purl, auth = proxy_url_and_auth(proxy)60        start = time.time()61        try:62            timeout = aiohttp.ClientTimeout(total=TIMEOUT_SECONDS)63            async with session.get(TEST_URL, proxy=purl, proxy_auth=auth, timeout=timeout) as resp:64                if resp.status != 200:65                    async with counters["lock"]:66                        counters["tested"] += 167                        counters["unmatched"] += 168                    await _send_progress(ws, counters)69                    return70                speed_ms = round((time.time() - start) * 1000, 1)71        except Exception:72            async with counters["lock"]:73                counters["tested"] += 174                counters["unmatched"] += 175            await _send_progress(ws, counters)76            return77 78        # Enrich with Whoer79        whoer = await get_whoer_info(session, ip)80 81        result = {82            "proxy": proxy,83            "speed_ms": speed_ms,84            "location": whoer.get("location", "N/A"),85            "ip_type": whoer.get("ip_type", "N/A"),86            "fraud_score": whoer.get("fraud_score", "N/A"),87        }88 89        # Apply filters90        loc_lower = result["location"].lower()91        type_lower = result["ip_type"].lower()92        country_ok = (not country_filter) or (country_filter.lower() in loc_lower)93        type_ok = (not type_filter) or (type_filter.lower() in type_lower)94 95        async with counters["lock"]:96            counters["tested"] += 197            if country_ok and type_ok:98                counters["matched"] += 199            else:100                counters["unmatched"] += 1101 102        await _send_progress(ws, counters)103 104        if country_ok and type_ok:105            try:106                await ws.send_text(json.dumps({"type": "proxy", "data": result}))107            except Exception:108                pass109 110 111async def _send_progress(ws: WebSocket, counters: dict):112    try:113        await ws.send_text(json.dumps({114            "type": "progress",115            "tested": counters["tested"],116            "total": counters["total"],117            "matched": counters["matched"],118            "unmatched": counters["unmatched"],119        }))120    except Exception:121        pass122 123 124@app.websocket("/ws/scan")125async def scan_proxies(ws: WebSocket):126    await ws.accept()127 128    try:129        data = await ws.receive_text()130        params = json.loads(data)131        country_filter = params.get("country", "").strip()132        type_filter = params.get("ip_type", "").strip()133 134        await ws.send_text(json.dumps({"type": "status", "message": "Scraping proxy sources..."}))135 136        proxies = scrape_proxies()137 138        counters = {139            "tested": 0,140            "total": len(proxies),141            "matched": 0,142            "unmatched": 0,143            "lock": asyncio.Lock(),144        }145 146        await ws.send_text(json.dumps({147            "type": "status",148            "message": f"Testing {len(proxies)} proxies...",149        }))150        await ws.send_text(json.dumps({151            "type": "progress",152            "tested": 0,153            "total": len(proxies),154            "matched": 0,155            "unmatched": 0,156        }))157 158        MAX_CONCURRENT = 80159        semaphore = asyncio.Semaphore(MAX_CONCURRENT)160        connector = aiohttp.TCPConnector(ssl=False, limit=MAX_CONCURRENT)161 162        async with aiohttp.ClientSession(connector=connector) as session:163            tasks = [164                asyncio.create_task(165                    test_and_enrich(session, proxy, semaphore, country_filter, type_filter, counters, ws)166                )167                for proxy in proxies168            ]169            await asyncio.gather(*tasks, return_exceptions=True)170 171        await ws.send_text(json.dumps({172            "type": "done",173            "message": f"Done. {counters['matched']} matched out of {counters['total']} proxies tested.",174        }))175 176    except WebSocketDisconnect:177        pass178    except Exception as e:179        try:180            await ws.send_text(json.dumps({"type": "error", "message": str(e)}))181        except Exception:182            pass183 184 185# Serve the Next.js static export — must come LAST (after all API routes)186_static_dir = os.path.join(os.path.dirname(__file__), "webapp", "out")187if os.path.isdir(_static_dir):188    app.mount("/", StaticFiles(directory=_static_dir, html=True), name="frontend")189