import httpx from fastapi import FastAPI, Query from fastapi.responses import PlainTextResponse app = FastAPI( title="Torpedo Proxy Generator", description="Generates IP:PORT lists specifically for Traffic Generators." ) # This is the upstream source that allows country filtering for free BASE_URL = "https://api.proxyscrape.com/v2/" @app.get("/", response_class=PlainTextResponse) def home(): return "Status: Online. Use /get?country=US&protocol=http to get proxies." @app.get("/get", response_class=PlainTextResponse) async def get_proxies( country: str = Query("all", description="2-letter Country Code (e.g., US, DE, FR) or 'all'"), protocol: str = Query("http", description="http, socks4, or socks5"), timeout: int = Query(10000, description="Timeout in ms (lower is faster proxies but fewer results)") ): """ Fetches proxies and returns them in PLAIN TEXT format (IP:PORT) perfect for Torpedo Traffic Generator. """ # Normalize input country = country.lower() protocol = protocol.lower() # Construct the query to the upstream provider api_url = f"{BASE_URL}?request=getproxies&protocol={protocol}&timeout={timeout}&country={country}&ssl=all&anonymity=all" async with httpx.AsyncClient() as client: try: response = await client.get(api_url, timeout=15) if response.status_code == 200: # The upstream API returns text properly, we pass it through data = response.text.strip() # Validation: If provider returns empty or error if not data: return f"No proxies found for country: {country.upper()}" return data else: return "Error fetching from upstream source." except Exception as e: return f"Server Error: {str(e)}" @app.get("/countries", response_class=PlainTextResponse) def list_common_countries(): """Helper to show codes for Torpedo users""" return """ Common Country Codes: US = United States DE = Germany FR = France GB = United Kingdom BR = Brazil IN = India RU = Russia CN = China CA = Canada AU = Australia ALL = All Countries