Ultimate Proxy Checker

Public

.PY

xx-HITMAN-xx Dec 27, 2025 43 .PY

Save New

import sys,os,asyncio,re,time,random,signal,subprocess
def _p(x):
    try:__import__(x)
    except:subprocess.check_call([sys.executable,"-m","pip","install",x])
for m in ("aiohttp","aiohttp_socks","rich"):_p(m)

import aiohttp
from aiohttp_socks import ProxyConnector
from rich.console import Console
from rich.live import Live
from rich.table import Table
from rich.panel import Panel
from rich.layout import Layout
from rich import box

try:
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except:pass

C=Console()
RX=re.compile(r"(?:\d{1,3}\.){3}\d{1,3}:\d{2,5}")
SRC=[
 "https://api.proxyscrape.com/v2/?request=getproxies&protocol=http",
 "https://api.proxyscrape.com/v2/?request=getproxies&protocol=socks4",
 "https://api.proxyscrape.com/v2/?request=getproxies&protocol=socks5",
 "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt",
 "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks4.txt",
 "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks5.txt",
]

MW,TO,URL=500,5,"http://httpbin.org/ip"
UA=["Mozilla/5.0","Mozilla/5.0 (X11)","Mozilla/5.0 (Macintosh)"]

class _S:
    def __init__(s):
        s.t=s.c=s.h=s.s4=s.s5=s.d=0
        s.st=time.time()
S=_S()

class _X:
    def __init__(x):
        x.q=asyncio.Queue()
        x.e=asyncio.Event()
        x.p=set()

async def _f(x,s,u):
        try:
            async with s.get(u,timeout=10) as r:
                if r.status==200:x.p|=set(RX.findall(await r.text()))
        except:pass

async def _load(x):
        async with aiohttp.ClientSession() as s:
            await asyncio.gather(*[x._f(s,u) for u in SRC])
        S.t=len(x.p)
        for i in x.p:x.q.put_nowait(i)

async def _chk(x,p,o):
        try:
            c=ProxyConnector.from_url(f"{o}://{p}")
            async with aiohttp.ClientSession(
                connector=c,
                timeout=aiohttp.ClientTimeout(total=TO)
            ) as s:
                async with s.get(URL,headers={"User-Agent":random.choice(UA)}) as r:
                    return r.status==200
        except:return False

async def _w(x):
        while not x.q.empty() and not x.e.is_set():
            p=await x.q.get()
            if await x._chk(p,"http"):S.h+=1
            elif await x._chk(p,"socks4"):S.s4+=1
            elif await x._chk(p,"socks5"):S.s5+=1
            else:S.d+=1
            S.c+=1
            x.q.task_done()

async def run(x):
        await x._load()
        L=Layout()
        L.split_column(Layout(name="h",size=3),Layout(name="b"))
        L["h"].update(
            Panel(
                f"? [bold magenta]ULTIMATE PROXY CHECKER[/bold magenta] | Workers: {MW} | Timeout: {TO}s",
                style="bold white"
            )
        )
        W=[asyncio.create_task(x._w()) for _ in range(min(MW,S.t))]
        with Live(L,refresh_per_second=4,console=C):
            while not x.q.empty():
                pc=(S.c/S.t)*100 if S.t else 0
                el=time.time()-S.st
                cpm=int((S.c/el)*60) if el>0 else 0
                T=Table(box=box.ROUNDED,expand=True,show_header=True,header_style="bold magenta")
                T.add_column("METRIC",style="cyan")
                T.add_column("VALUE",justify="right",style="bold white")
                T.add_row("Progress",f"{pc:.1f}%")
                T.add_row("Checked",f"{S.c}/{S.t}")
                T.add_row("Speed",f"{cpm} CPM")
                T.add_section()
                T.add_row("HTTP Valid",f"● {S.h}")
                T.add_row("SOCKS4 Valid",f"● {S.s4}")
                T.add_row("SOCKS5 Valid",f"● {S.s5}")
                T.add_row("Dead",f"● {S.d}")
                L["b"].update(Panel(T,title="[b]LIVE MONITOR[/b]",border_style="blue"))
                await asyncio.sleep(0.5)
        await x.q.join()
        for w in W:w.cancel()

X=_X()
signal.signal(signal.SIGINT,lambda a,b:X.e.set())
asyncio.run(X.run())

Ultimate Proxy Checker

Scan to View