Ả Rập Xê-út

IP nhà dân thật

Proxy dân cư Ả Rập Xê-út

Proxy dân cư Ả Rập Xê Út là những địa chỉ IP thật từ các hộ gia đình Ả Rập Xê Út, do các nhà mạng dân dụng như STC, Mobily và Zain KSA cấp phát. Chúng cho phép bạn duyệt web, thu thập dữ liệu và kiểm tra các gian hàng, giá cả cùng nội dung tại Ả Rập Xê Út với tư cách một người dùng thực thụ ngay trong nước — trải khắp Riyadh, Jeddah, Dammam và Medina — thay vì từ một IP trung tâm dữ liệu mà Noon, Amazon Saudi Arabia và các cổng chặn theo vùng trong nước phát hiện và chặn ngay lập tức.

Dùng thử với $1.99/GB Đăng nhập bằng Google

HTTP / SOCKS5Nhắm theo thành phố & ISPLưu lượng không hết hạnKiểm định sạch hàng tháng

50.924IP trong pool

19+Thành phố phủ sóng

12Mạng ISP

336 msĐộ trễ trung vị

99.6%Sạch tuyệt đối

Tổng quan nhanh về proxy dân cư Ả Rập Xê-út

Proxy dân cư Ả Rập Xê-út là địa chỉ IP thật từ các hộ gia đình Ả Rập Xê-út, do các ISP tiêu dùng cấp phát. Chúng cho phép bạn duyệt web, thu thập dữ liệu và kiểm tra website, mức giá cùng nội dung Ả Rập Xê-út với tư cách một người dùng nội địa thực thụ — thay vì từ một IP trung tâm dữ liệu bị các trang nội địa gắn cờ và chặn.

Loại pool: Dân cư, đa ISP
Thành phố phủ sóng: Riyadh, Jeddah, Dammam
ISP hàng đầu: STC (Saudi Telecom) · Mobily · Zain KSA
Giao thức: HTTP, SOCKS5
Xoay IP: Theo từng yêu cầu hoặc cố định
Độ chính xác địa lý: 99.95%
Độ sạch của pool: 99.6% clean
Giá: from $2.00/GB
Đo lần cuối: 2026-06-23
Hỗ trợ: 24/7

Vì sao các đội ngũ chọn pool Ả Rập Xê-út của chúng tôi

IP dân cư thật, nhắm mục tiêu chính xác và chất lượng bạn không phải tin suông.

IP dân cư thật

Kết nối nhà dân đích thực từ các ISP tiêu dùng Ả Rập Xê-út — không phải dải IP trung tâm dữ liệu bị chặn ngay tức khắc.

Nhắm theo thành phố & ISP

Nhắm đến từng thành phố và nhà mạng cụ thể khi độ sâu của pool cho phép — chi tiết đến từng nhà cung cấp.

Xoay IP hoặc cố định

Xoay IP theo từng yêu cầu để mở rộng quy mô, hoặc phiên cố định lên đến ~72h cho các quy trình nhiều bước.

HTTP & SOCKS5

Hoạt động ngay với mọi ứng dụng, công cụ scraping hay nền tảng tự động hóa.

Nguồn gốc minh bạch

IP từ mạng dân cư theo cơ chế đăng ký tự nguyện — người dùng thật đã đồng ý chia sẻ băng thông.

Kiểm định sạch, hàng tháng

Chúng tôi đo lại pool mỗi tháng và công bố số liệu, nhờ vậy chúng luôn được cập nhật.

Chất lượng bạn có thể tự kiểm chứng

Mỗi tháng, chúng tôi chạy cùng một bộ benchmark — một mẫu IP ngẫu nhiên được chấm điểm trên Scamalytics, kiểm tra độ trễ từ một VPS trung lập và đối chiếu độ chính xác địa lý trên nhiều cơ sở dữ liệu độc lập. Các con số dưới đây phản ánh môi trường thực tế, không phải kết quả phòng thí nghiệm trong điều kiện lý tưởng.

Độ sạch Scamalytics · 100.000 IP

Sạch tuyệt đối99.6%

Lẫn IP trung tâm dữ liệu0%

Có trong danh sách đen bất kỳ0.2%

Rủi ro gian lận thấp (0–19)99.4%

Độ chính xác địa lý99.95%

Độ trễ & tốc độ VPS → Cloudflare

Độ trễ trung vị336ms

Độ trễ p90 / p95459 / 510 ms

Tỷ lệ kết nối thành công99.1%

Thông lượng (trung vị / p95)0.95 / 3.02 MB/s

Hoàn tất tải tệp lớn98%

Cấu thành mạng lưới

Loại kết nối100% dân cư

Mạng ISP (ASN)12

Chế độ xoay IPtheo từng yêu cầu · cố định

Nhà mạng hàng đầuSTC (Saudi Telecom) · Mobily · Zain KSA

Đo trên 100.000 địa chỉ từ một VPS trung lập, đối chiếu với hạ tầng edge của Cloudflare, chính xác tại thời điểm benchmark gần nhất. Một lần chạy lại có thể chênh lệch đôi chút, nhưng các con số vẫn phản ánh đúng độ lớn thực tế và là cơ sở vững chắc để đánh giá liệu pool có phù hợp với nhu cầu của bạn hay không. Tỷ lệ scraping thành công trên từng mục tiêu không được công bố — điều đó phụ thuộc vào fingerprint và cấu hình của bạn, chứ không chỉ riêng IP.

benchmark.py — benchmark pool nguồn mở

Bộ benchmark mở đứng sau các con số phía trên. Nó lấy một mẫu IP ngẫu nhiên từ pool và đo bốn yếu tố — độ sạch (rủi ro gian lận Scamalytics, trung tâm dữ liệu, danh sách đen và độ chính xác địa lý), độ trễ (khứ hồi đến Cloudflare edge và ip-api), xoay IP (số IP, ASN và subnet /24 duy nhất) cùng thông lượng (tốc độ tải xuống và thời gian đến byte đầu tiên) — rồi báo cáo từng yếu tố. Mã nguồn được công bố đầy đủ để phương pháp luận đằng sau mỗi con số luôn rộng mở cho việc kiểm chứng.

# -*- coding: utf-8 -*-

"""
JETPROXY — POOL BENCHMARK

End-to-end audit of a residential proxy pool. A single run measures cleanliness,
latency, rotation and throughput, then prints every metric as one list.

  Cleanliness   fraud-risk, datacenter, blacklist and geo-accuracy (Scamalytics)
  Latency       Cloudflare edge (responsiveness) and ip-api origin (round-trip)
  Rotation      unique IPs / ASNs / subnets, success rate, country match
  Throughput    median & p95 download speed, time-to-first-byte, drop rate

Every stage size is a parameter, so the same script runs a quick check or a deep
audit:

    python benchmark.py --country ID
    python benchmark.py --quality 200 --downloads 50 --latency 1000 --rotation 5000 --country ID
    python benchmark.py --proxies proxies.txt --country ID --json result.json

Proxies   paste at the prompt (one per line) or pass --proxies FILE.
          host:port:user:pass  ·  host:port  ·  user:pass@host:port
The Scamalytics key for the cleanliness stage is read from the environment or
../.env (SCAMALYTICS_USER + SCAMALYTICS_KEY); without it that stage is skipped.
"""
from __future__ import annotations

import argparse
import asyncio
import csv
import ipaddress
import json
import os
import re
import statistics
import sys
import time
from collections import Counter
from datetime import datetime

try:
    sys.stdout.reconfigure(encoding="utf-8")
    sys.stderr.reconfigure(encoding="utf-8")
except Exception:
    pass

try:
    import aiohttp
    from rich.console import Console
    from rich.table import Table
    from rich import box
except ImportError:
    print("\n[!] Need: pip install aiohttp rich\n")
    sys.exit(1)

console = Console()

# ── endpoints ────────────────────────────────────────────────────────────────

IP_ENDPOINT = ("http://ip-api.com/json/?fields="
               "status,message,country,countryCode,regionName,city,isp,as,org,query")
CF_TRACE = "https://speed.cloudflare.com/cdn-cgi/trace"   # tiny edge response (~350 B)

FILE_MB = 10
DOWN_URL = f"https://speed.cloudflare.com/__down?bytes={FILE_MB * 1_000_000}"
SCAM_HOST = "https://api11.scamalytics.com/v3"            # US node (api12 = EU)


SCORE_CONCURRENCY = 10
DOWN_CONCURRENCY = 3   # downloads share the local uplink — keep low so the exit node is the bottleneck



# ── proxy parsing ──────────────────────────────────────────────────────────────

def parse_proxy(raw: str):
    """host:port | host:port:user:pass | user:pass@host:port | scheme://..."""
    raw = raw.strip()
    if not raw or raw.startswith("#"):
        return None
    scheme = "http"
    if "://" in raw:
        scheme, raw = raw.split("://", 1)
    user = pwd = host = port = None
    if "," in raw:
        parts = [p.strip() for p in raw.split(",")]
        if len(parts) == 4:
            host, port, user, pwd = parts
        elif len(parts) == 2:
            host, port = parts
        else:
            return None
    elif "@" in raw:
        cred, hostport = raw.rsplit("@", 1)
        if ":" in cred:
            user, pwd = cred.split(":", 1)
        host, port = hostport.rsplit(":", 1)
    else:
        parts = raw.split(":")
        if len(parts) == 2:
            host, port = parts
        elif len(parts) == 4:
            host, port, user, pwd = parts
        else:
            return None
    try:
        port = int(port)
    except (ValueError, TypeError):
        return None
    return {"scheme": scheme, "host": host, "port": port, "user": user, "pwd": pwd,
            "url": f"{scheme}://{host}:{port}", "label": f"{host}:{port}"}


def subnet24(ip: str) -> str:
    try:
        return str(ipaddress.ip_network(ip + "/24", strict=False))
    except ValueError:
        return ""


def pct(values, p):
    if not values:
        return 0.0
    s = sorted(values)
    k = min(len(s) - 1, max(0, int(round(p / 100 * (len(s) - 1)))))
    return s[k]


# ── ip-api request (identity + geo + round-trip) ───────────────────────────────

def _fail(idx, proxy, err, latency):
    return {"idx": idx, "ok": False, "proxy": proxy["label"], "error": err, "latency": latency}


async def _one_attempt(session, proxy, idx, timeout):
    t0 = time.perf_counter()
    auth = aiohttp.BasicAuth(proxy["user"], proxy["pwd"]) if proxy["user"] else None
    try:
        async with session.get(IP_ENDPOINT, proxy=proxy["url"], proxy_auth=auth,
                               timeout=aiohttp.ClientTimeout(total=timeout)) as r:
            latency = (time.perf_counter() - t0) * 1000
            if r.status != 200:
                return _fail(idx, proxy, f"HTTP {r.status}", latency)
            data = await r.json(content_type=None)
            if data.get("status") != "success":
                return _fail(idx, proxy, data.get("message", "api error"), latency)
            ip = data.get("query", "")
            asn = data.get("as", "").split(" ")[0] if data.get("as") else ""
            return {"idx": idx, "ok": True, "proxy": proxy["label"], "ip": ip,
                    "country": data.get("country", ""), "cc": data.get("countryCode", ""),
                    "region": data.get("regionName", ""), "city": data.get("city", ""),
                    "isp": data.get("isp") or data.get("org", ""), "asn": asn,
                    "org": data.get("org") or data.get("as", ""),
                    "subnet": subnet24(ip), "latency": latency, "error": ""}
    except asyncio.TimeoutError:
        return _fail(idx, proxy, "timeout", (time.perf_counter() - t0) * 1000)
    except Exception as e:
        return _fail(idx, proxy, type(e).__name__, (time.perf_counter() - t0) * 1000)


async def fetch(session, proxy, sem, idx, timeout, max_retries=2):
    async with sem:
        res = None
        for attempt in range(1, max_retries + 2):
            res = await _one_attempt(session, proxy, idx, timeout)
            if res["ok"]:
                res["attempts"] = attempt
                res["first_try_ok"] = (attempt == 1)
                return res
        res["attempts"] = max_retries + 1
        res["first_try_ok"] = False
        return res


# ── Cloudflare edge probe (responsiveness + nearest datacenter) ────────────────

async def cf_probe(proxy, sem, timeout):
    """One warm-up request opens the tunnel + TLS, then the next request on the same
    kept-alive connection is timed — so the figure is the edge round-trip, not the
    one-off connection setup."""
    async with sem:
        auth = aiohttp.BasicAuth(proxy["user"], proxy["pwd"]) if proxy["user"] else None
        try:
            async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=2, ssl=False)) as sess:
                async with sess.get(CF_TRACE, proxy=proxy["url"], proxy_auth=auth,
                                    timeout=aiohttp.ClientTimeout(total=15)) as w:
                    await w.read()
                t0 = time.perf_counter()
                async with sess.get(CF_TRACE, proxy=proxy["url"], proxy_auth=auth,
                                    timeout=aiohttp.ClientTimeout(total=timeout)) as r:
                    txt = await r.text()
                    rtt = (time.perf_counter() - t0) * 1000
                    if r.status != 200:
                        return {"ok": False, "error": f"HTTP {r.status}"}
                    d = {}
                    for ln in txt.splitlines():
                        if "=" in ln:
                            k, v = ln.split("=", 1)
                            d[k] = v
                    return {"ok": True, "rtt": rtt, "ip": d.get("ip", ""),
                            "loc": d.get("loc", ""), "colo": d.get("colo", "")}
        except asyncio.TimeoutError:
            return {"ok": False, "error": "timeout"}
        except Exception as e:
            return {"ok": False, "error": type(e).__name__}


# ── download (throughput + TTFB + mid-transfer drops) ──────────────────────────

async def download(proxy, sem, idx, timeout, retries):
    async with sem:
        auth = aiohttp.BasicAuth(proxy["user"], proxy["pwd"]) if proxy["user"] else None
        last_err = ""
        dropped_at = []
        for attempt in range(1, retries + 2):
            try:
                async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(limit=4, ssl=False)) as sess:
                    country = "?"
                    try:
                        async with sess.get(CF_TRACE, proxy=proxy["url"], proxy_auth=auth,
                                            timeout=aiohttp.ClientTimeout(total=15)) as tr:
                            for ln in (await tr.text()).splitlines():
                                if ln.startswith("loc="):
                                    country = ln[4:].strip() or "?"
                    except Exception:
                        pass
                    t0 = time.perf_counter()
                    async with sess.get(DOWN_URL, proxy=proxy["url"], proxy_auth=auth,
                                        timeout=aiohttp.ClientTimeout(total=timeout)) as r:
                        if r.status != 200:
                            last_err = f"HTTP {r.status}"
                            continue
                        first = None
                        nbytes = 0
                        try:
                            async for chunk in r.content.iter_chunked(65536):
                                if first is None:
                                    first = time.perf_counter()
                                nbytes += len(chunk)
                        except Exception as stream_err:
                            if nbytes > 0:
                                dropped_at.append(nbytes)
                                last_err = f"dropped@{nbytes // 1_000_000}MB"
                            else:
                                last_err = type(stream_err).__name__
                            continue
                        end = time.perf_counter()
                        if nbytes == 0:
                            last_err = "empty"
                            continue
                        ttfb = (first - t0) * 1000 if first else 0
                        dl_time = (end - first) if first and end > first else (end - t0)
                        mbps = (nbytes / 1_000_000) / dl_time if dl_time > 0 else 0
                        return {"idx": idx, "ok": True, "country": country, "bytes": nbytes,
                                "ttfb_ms": ttfb, "mbps": mbps, "dropped": bool(dropped_at), "error": ""}
            except asyncio.TimeoutError:
                last_err = "timeout"
            except Exception as e:
                last_err = type(e).__name__
        return {"idx": idx, "ok": False, "country": "?", "bytes": 0, "ttfb_ms": 0,
                "mbps": 0, "dropped": bool(dropped_at), "error": last_err}


# ── Scamalytics scoring ────────────────────────────────────────────────────────

def load_creds():
    user = os.environ.get("SCAMALYTICS_USER")
    key = os.environ.get("SCAMALYTICS_KEY")
    if user and key:
        return user.strip(), key.strip()
    found = {}
    for path in ("../.env", "../.env.local", ".env"):
        p = os.path.join(os.path.dirname(__file__), path)
        if os.path.exists(p):
            with open(p, encoding="utf-8") as f:
                for line in f:
                    m = re.match(r'\s*(SCAMALYTICS_USER|SCAMALYTICS_KEY)\s*=\s*"?([^"\s]+)"?', line)
                    if m:
                        found[m.group(1)] = m.group(2)
    return found.get("SCAMALYTICS_USER"), found.get("SCAMALYTICS_KEY")


def parse_scam(ip, data):
    s = data.get("scamalytics", {})
    if s.get("status") != "ok":
        return {"ip": ip, "ok": False, "error": s.get("status") or "bad_response"}
    ext = data.get("external_datasources", {})
    score = s.get("scamalytics_score")
    risk = s.get("scamalytics_risk", "?")
    px = s.get("scamalytics_proxy", {})
    is_dc = bool(px.get("is_datacenter"))
    is_vpn = bool(px.get("is_vpn"))
    fh = bool(ext.get("firehol", {}).get("is_proxy"))
    ips_bl = bool(ext.get("ipsum", {}).get("ip_blacklisted"))
    sh = bool(ext.get("spamhaus_drop", {}).get("ip_blacklisted"))
    x4 = bool(ext.get("x4bnet", {}).get("is_blacklisted_spambot"))
    on_bl = bool(s.get("is_blacklisted_external")) or fh or ips_bl or sh or x4
    cc = (ext.get("dbip", {}).get("ip_country_code")
          or ext.get("ipinfo", {}).get("ip_country_code")
          or ext.get("maxmind_geolite2", {}).get("ip_country_code")
          or ext.get("ip2proxy_lite", {}).get("ip_country_code") or "?")
    isp = (ext.get("dbip", {}).get("isp_name")
           or ext.get("ip2proxy_lite", {}).get("isp_name")
           or s.get("scamalytics_isp") or "?")
    asn = ext.get("ipinfo", {}).get("asn") or ext.get("ip2proxy_lite", {}).get("asn") or "?"
    return {"ip": ip, "ok": True, "error": "", "score": score if isinstance(score, int) else 0,
            "risk": risk, "is_dc": is_dc, "is_vpn": is_vpn, "on_bl": on_bl,
            "country": cc, "isp": isp, "asn": asn,
            "credits": s.get("credits", {}).get("remaining"),
            "strict_clean": (risk == "low" and not is_dc and not on_bl)}


async def score_one(session, user, key, ip, sem):
    async with sem:
        try:
            async with session.get(f"{SCAM_HOST}/{user}/", params={"key": key, "ip": ip},
                                   timeout=aiohttp.ClientTimeout(total=20)) as r:
                return parse_scam(ip, await r.json(content_type=None))
        except Exception as e:
            return {"ip": ip, "ok": False, "error": type(e).__name__}


# ── stage runners ──────────────────────────────────────────────────────────────

def progress(tag, done, total):
    step = max(1, total // 10)
    if done % step == 0 or done == total:
        console.print(f"   [{tag}] {done}/{total}", highlight=False)


async def run_ipapi(proxies, n, concurrency, timeout, retries, tag):
    sem = asyncio.Semaphore(concurrency)
    conn = aiohttp.TCPConnector(limit=concurrency * 3, ssl=False)
    out = []
    async with aiohttp.ClientSession(connector=conn) as http:
        tasks = [asyncio.create_task(fetch(http, proxies[i % len(proxies)], sem, i, timeout, retries))
                 for i in range(n)]
        for done, coro in enumerate(asyncio.as_completed(tasks), 1):
            out.append(await coro)
            progress(tag, done, n)
    return out


async def run_cf(proxies, n, concurrency, timeout, tag):
    sem = asyncio.Semaphore(concurrency)
    tasks = [asyncio.create_task(cf_probe(proxies[i % len(proxies)], sem, timeout))
             for i in range(n)]
    out = []
    for done, coro in enumerate(asyncio.as_completed(tasks), 1):
        out.append(await coro)
        progress(tag, done, n)
    return out


async def run_download(proxies, n, timeout, retries, tag):
    sem = asyncio.Semaphore(DOWN_CONCURRENCY)
    tasks = [asyncio.create_task(download(proxies[i % len(proxies)], sem, i, timeout, retries))
             for i in range(n)]
    out = []
    for done, coro in enumerate(asyncio.as_completed(tasks), 1):
        out.append(await coro)
        progress(tag, done, n)
    return out


async def score_ips(ips, user, key, tag):
    sem = asyncio.Semaphore(SCORE_CONCURRENCY)
    out = []
    async with aiohttp.ClientSession() as session:
        tasks = [asyncio.create_task(score_one(session, user, key, ip, sem)) for ip in ips]
        for done, coro in enumerate(asyncio.as_completed(tasks), 1):
            out.append(await coro)
            progress(tag, done, len(ips))
    return out


# ── input ──────────────────────────────────────────────────────────────────────

def read_proxies(path):
    if path:
        with open(path, encoding="utf-8") as f:
            lines = f.readlines()
    else:
        console.print("Paste proxies (one per line), then an empty line:")
        lines = []
        while True:
            try:
                ln = input()
            except EOFError:
                break
            if ln.strip() == "":
                if lines:
                    break
                continue
            lines.append(ln)
    return [p for p in (parse_proxy(l) for l in lines) if p]


def fmt(v, suffix=""):
    return f"{v}{suffix}"


# ── main ─────────────────────────────────────────────────────────────────────

def build_report(M):
    t = Table(title="POOL BENCHMARK — RESULTS", box=box.DOUBLE_EDGE, header_style="bold cyan",
              title_style="bold green")
    t.add_column("Metric", style="cyan")
    t.add_column("Value", style="bold white", justify="right")
    for section, rows in M["sections"]:
        t.add_row(f"[bold yellow]{section}[/]", "")
        for k, v in rows:
            t.add_row(f"  {k}", str(v))
    console.print()
    console.print(t)


async def run(args, proxies):
    M = {"label": args.label, "country": args.country or None,
         "timestamp": datetime.now().isoformat(timespec="seconds"),
         "proxies_input": len(proxies), "stages": {}}
    user = key = None

    # 1 — cleanliness (Scamalytics)

    clean_rows = []
    if not args.no_quality and args.quality > 0:
        user, key = load_creds()
        if not user or not key:
            console.print("[yellow]Scamalytics key not found — skipping cleanliness stage.[/]")
        else:
            console.rule(f"[bold]Cleanliness — {args.quality} IPs[/]")
            collected = await run_ipapi(proxies, int(args.quality * 1.6) + 10,
                                        args.concurrency, 15, 1, "collect")
            uniq = {}
            for r in collected:
                if r.get("ok") and r.get("ip") and r["ip"] not in uniq:
                    uniq[r["ip"]] = r
            ips = list(uniq.keys())[:args.quality]
            scored = await score_ips(ips, user, key, "score")
            ok = [r for r in scored if r.get("ok")]
            n = len(ok)
            if n:
                strict = sum(r["strict_clean"] for r in ok) / n * 100
                low = sum(1 for r in ok if r["risk"] == "low") / n * 100
                dc = sum(r["is_dc"] for r in ok) / n * 100
                bl = sum(r["on_bl"] for r in ok) / n * 100
                vpn = sum(r["is_vpn"] for r in ok)
                geo = None
                if args.country:
                    geo = sum(1 for r in ok if (r["country"] or "").upper() == args.country.upper()) / n * 100
                M["stages"]["cleanliness"] = {
                    "n": n, "strict_clean_pct": round(strict, 1), "low_risk_pct": round(low, 1),
                    "datacenter_pct": round(dc, 1), "blacklist_pct": round(bl, 1),
                    "geo_accuracy_pct": round(geo, 1) if geo is not None else None, "vpn": vpn}
                clean_rows = [("Strictly clean", f"{strict:.1f}%"),
                              ("Low fraud-risk (0–19)", f"{low:.1f}%"),
                              ("Datacenter", f"{dc:.1f}%"),
                              ("Blacklisted", f"{bl:.1f}%")]
                if geo is not None:
                    clean_rows.append((f"Geo-accuracy ({args.country.upper()})", f"{geo:.1f}%"))
                # per-IP CSV

                ts = datetime.now().strftime("%Y%m%d_%H%M%S")
                with open(f"benchmark_ips_{ts}.csv", "w", newline="", encoding="utf-8") as f:
                    w = csv.writer(f)
                    w.writerow(["ip", "score", "risk", "is_datacenter", "is_vpn", "on_blacklist",
                                "country", "isp", "asn", "strict_clean"])
                    for r in scored:
                        if r.get("ok"):
                            w.writerow([r["ip"], r["score"], r["risk"], r["is_dc"], r["is_vpn"],
                                        r["on_bl"], r["country"], r["isp"], r["asn"], r["strict_clean"]])

    # 2 — throughput

    console.rule(f"[bold]Throughput — {args.downloads} downloads[/]")
    dl = await run_download(proxies, args.downloads, 60, 1, "download")
    dok = [r for r in dl if r["ok"]]
    dl_rows = []
    if dok:
        mbps = [r["mbps"] for r in dok]
        ttfb = [r["ttfb_ms"] for r in dok]
        nodrop = sum(1 for r in dl if r["ok"] and not r["dropped"]) / len(dl) * 100
        M["stages"]["throughput"] = {
            "n": len(dl), "ok": len(dok),
            "median_mbps": round(statistics.median(mbps), 2), "p95_mbps": round(pct(mbps, 95), 2),
            "median_ttfb_ms": round(statistics.median(ttfb)), "completed_no_drop_pct": round(nodrop, 1)}
        dl_rows = [("Median download", f"{statistics.median(mbps):.1f} MB/s ({statistics.median(mbps)*8:.0f} Mbps)"),
                   ("p95 download", f"{pct(mbps,95):.1f} MB/s ({pct(mbps,95)*8:.0f} Mbps)"),
                   ("Median TTFB", f"{statistics.median(ttfb):.0f} ms"),
                   ("Completed without drop", f"{nodrop:.0f}%")]

    # 3 — latency (Cloudflare edge + ip-api origin)

    console.rule(f"[bold]Latency — {args.latency} probes[/]")
    cf = await run_cf(proxies, args.latency, args.concurrency, 15, "cf")
    cfok = [r for r in cf if r["ok"]]
    lat_rows = []
    if cfok:
        rtts = [r["rtt"] for r in cfok]
        colo = Counter(r["colo"] for r in cfok if r["colo"]).most_common(1)
        M["stages"].setdefault("latency", {})
        M["stages"]["latency"]["cloudflare"] = {
            "n": len(cfok), "p50_ms": round(statistics.median(rtts)),
            "p90_ms": round(pct(rtts, 90)), "p95_ms": round(pct(rtts, 95)),
            "top_colo": colo[0][0] if colo else None}
        lat_rows += [("Cloudflare edge — p50", f"{statistics.median(rtts):.0f} ms"),
                     ("Cloudflare edge — p90", f"{pct(rtts,90):.0f} ms"),
                     ("Cloudflare edge — p95", f"{pct(rtts,95):.0f} ms")]
        if colo:
            lat_rows.append(("Nearest datacenter", colo[0][0]))

    # 4 — rotation (also yields ip-api round-trip latency)

    console.rule(f"[bold]Rotation — {args.rotation} requests[/]")
    rot = await run_ipapi(proxies, args.rotation, args.concurrency, args.timeout, 2, "rotation")
    rok = [r for r in rot if r["ok"]]
    rot_rows = []
    if rok:
        uips = {r["ip"] for r in rok if r.get("ip")}
        uasn = {r["asn"] for r in rok if r.get("asn")}
        usub = {r["subnet"] for r in rok if r.get("subnet")}
        success = len(rok) / len(rot) * 100
        first = sum(1 for r in rok if r.get("first_try_ok")) / len(rot) * 100
        cmatch = None
        if args.country:
            cmatch = sum(1 for r in rok if (r["cc"] or "").upper() == args.country.upper()) / len(rok) * 100
        ipapi_rtts = [r["latency"] for r in rok]
        M["stages"]["rotation"] = {
            "n": len(rot), "success_pct": round(success, 1), "first_try_pct": round(first, 1),
            "unique_ips": len(uips), "unique_asns": len(uasn), "unique_subnets": len(usub),
            "country_match_pct": round(cmatch, 2) if cmatch is not None else None}
        M["stages"].setdefault("latency", {})
        M["stages"]["latency"]["ip_api"] = {
            "n": len(ipapi_rtts), "p50_ms": round(statistics.median(ipapi_rtts)),
            "p90_ms": round(pct(ipapi_rtts, 90))}
        lat_rows += [("ip-api origin — p50", f"{statistics.median(ipapi_rtts):.0f} ms"),
                     ("ip-api origin — p90", f"{pct(ipapi_rtts,90):.0f} ms")]
        rot_rows = [("Success rate", f"{success:.1f}%"),
                    ("First-try success", f"{first:.1f}%"),
                    ("Unique exit IPs", f"{len(uips)}"),
                    ("Unique networks (ASN)", f"{len(uasn)}"),
                    ("Unique /24 subnets", f"{len(usub)}")]
        if cmatch is not None:
            rot_rows.append((f"Country match ({args.country.upper()})", f"{cmatch:.2f}%"))

    M["sections"] = [s for s in [
        ("CLEANLINESS", clean_rows), ("LATENCY", lat_rows),
        ("ROTATION", rot_rows), ("THROUGHPUT", dl_rows)] if s[1]]
    build_report(M)

    out_path = args.json or f"benchmark_{args.label}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
    M.pop("sections", None)
    with open(out_path, "w", encoding="utf-8") as f:
        json.dump(M, f, indent=2, ensure_ascii=False)
    console.print(f"\n[green]✓[/] Saved: [bold]{out_path}[/]")


def main():
    ap = argparse.ArgumentParser(description="Jetproxy pool benchmark")
    ap.add_argument("--quality", type=int, default=100, help="IPs scored for cleanliness")
    ap.add_argument("--downloads", type=int, default=50, help="download samples")
    ap.add_argument("--latency", type=int, default=1000, help="latency probes")
    ap.add_argument("--rotation", type=int, default=2000, help="rotation requests")
    ap.add_argument("--concurrency", type=int, default=20)
    ap.add_argument("--timeout", type=float, default=15.0)
    ap.add_argument("--country", default="", help="expected ISO code for geo-accuracy")
    ap.add_argument("--proxies", default="", help="proxy list file (else paste)")
    ap.add_argument("--label", default="pool")
    ap.add_argument("--json", default="")
    ap.add_argument("--no-quality", action="store_true", help="skip the cleanliness stage")
    args = ap.parse_args()

    console.print("\n[bold cyan]═══  JETPROXY — POOL BENCHMARK  ═══[/]\n")
    proxies = read_proxies(args.proxies)
    if not proxies:
        console.print("[red]No valid proxies.[/]")
        return
    console.print(f"[green]✓[/] Proxies loaded: [bold]{len(proxies)}[/]")
    asyncio.run(run(args, proxies))


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        console.print("\n[yellow]Interrupted.[/]")

Nhắm mục tiêu chi tiết đến từng thành phố

«Trung bình trực tuyến» là số lượng thiết bị riêng biệt trung bình đang kết nối và sẵn sàng tại cùng một thời điểm — không phải tổng tích lũy. Được phân nhóm thành Cao / Trung bình / Thấp.

VùngThành phốNhà cung cấp

Vùng	Độ trễ	Trung bình trực tuyến	Pool
Riyadh Region	332 ms	~1.713	Cao
Mecca Region	323 ms	~1.406	Cao
Eastern Province	356 ms	~895	Cao
Medina Region	336 ms	~329	Trung bình
Al-Qassim Region	328 ms	~107	Trung bình
'Asir Region	365 ms	~47	Thấp
Tabuk Region	303 ms	~28	Thấp
Ha'il Region	382 ms	~4	Thấp

Thành phố	Độ trễ	Trung bình trực tuyến	Pool
Riyadh	332 ms	~1.714	Cao
Jeddah	323 ms	~1.363	Cao
Dammam	354 ms	~668	Cao
Medina	336 ms	~323	Cao
Khobar	433 ms	~109	Cao
Buraidah	330 ms	~98	Cao
Dhahran	365 ms	~44	Cao
Abha	372 ms	~41	Trung bình
Al Qaţīf	351 ms	~32	Trung bình
Makkah	308 ms	~29	Trung bình
Hofuf	349 ms	~29	Trung bình
Tabuk	303 ms	~28	Trung bình
Ta'if	336 ms	~15	Thấp
Unaizah	306 ms	~9	Thấp
Şafwá	335 ms	~7	Thấp
Al Markaz	417 ms	~6	Thấp
Sulţānah	303 ms	~6	Thấp
An Nimas	312 ms	~4	Thấp
Baq`a' ash Sharqiyah	385 ms	~4	Thấp

Nhà cung cấp	Độ trễ	Trung bình trực tuyến	Pool
STC (Saudi Telecom)cố định	318 ms	~3.521	Cao
Mobilydi động	433 ms	~933	Cao
Zain KSAdi động	502 ms	~55	Trung bình
STCdi động	399 ms	~15	Thấp
Zajildi động	401 ms	~6	Thấp

Độ chính xác địa lý — mọi IP bán dưới nhãn Ả Rập Xê-út đều phân giải về Ả Rập Xê-út trên ba cơ sở dữ liệu địa lý độc lập (db-ip, MaxMind, ipinfo); độ chính xác địa lý cao chính là tín hiệu mà các cổng kiểm tra địa lý nội địa dựa vào.

Proxy Ả Rập Xê-út theo trường hợp sử dụng

Được sắp xếp theo công việc bạn đang làm, không phải theo nền tảng. Các dịch vụ nội địa được liệt kê ở đúng chỗ chúng phát huy tác dụng.

Thương mại điện tử & nghiên cứu thị trường

Kiểm tra gian hàng, mức giá và tin đăng nội địa từ ngay bên trong thị trường.

Giám sát gian hàng
Theo dõi giá nội địa
Nghiên cứu người bán

NoonAmazon Saudi ArabiaJarir BookstoreeXtraSACO

Xác minh & hiệu suất quảng cáo

Xem cách quảng cáo, mẫu sáng tạo và SERP hiển thị với người dùng nội địa — không cần chuyển đổi thủ công.

SERP nội địa
Kiểm tra mẫu sáng tạo
Trang đích

SnapchatTikTokInstagramX (Twitter)Facebook

Du lịch & phân tích giá

Thu thập giá vé máy bay và khách sạn đúng như cách một khách hàng nội địa nhìn thấy.

Giá theo khu vực
Giám sát ưu đãi
So sánh mức giá

SaudiaflynasflyadealAlmosaferRehlat

Streaming & xác minh địa lý

Xác minh danh mục bị khóa theo khu vực và các cổng kiểm tra địa lý từ đúng vị trí.

Kiểm tra địa lý
Xác minh danh mục
Khả dụng theo khu vực

Shahid (MBC)STC TVRotanaStarzPlay ArabiaOSN+

Rao vặt & tin đăng nội địa

Giám sát hàng cũ, ô tô và bất động sản cho thuê trên các trang rao vặt nội địa.

Giám sát tin đăng
Theo dõi giá
Phủ sóng theo khu vực

HarajOpenSooqSyarahMotoryCarSwitch

Phát triển & tự động hóa

IP dân cư cho các pipeline, kiểm thử và thu thập dữ liệu hợp lệ qua API.

Tích hợp pipeline
Quản lý API
Phiên cố định

PythonScrapycURLGitHub

Ngoài các trang nội địa, cùng pool Ả Rập Xê-út đó cũng hoạt động với các nền tảng toàn cầu lớn — Google, Meta, Amazon, YouTube, TikTok và Netflix — nhìn từ ngôn ngữ và khu vực Ả Rập Xê-út: giá nội địa, quảng cáo, danh mục, SERP và nội dung.

Nhà phát triển API

Tạo các endpoint khoanh vùng theo Ả Rập Xê-út trong bảng điều khiển, sao chép các đoạn mã đã kiểm thử, hoặc tự động hóa danh sách proxy qua API công khai. Thành phố, ISP, phiên cố định và các danh sách đã lưu đều nằm gọn trong một quy trình làm việc duy nhất.

Tài liệu API

1import requests
2 
3headers = {"x-api-key": "gpx_live_YOUR_KEY"}
4payload = {
5  "country_code": "SA",
6  "city": "Riyadh",
7  "session_type": "sticky",
8  "quantity": 5,
9  "protocol": "socks5"
10}
11 
12response = requests.post(
13  "https://jetproxy.io/api/v1/generate",
14  headers=headers,
15  json=payload,
16  timeout=30
17)
18print(response.json()["proxies"])

Các gói minh bạch, không điều khoản ẩn

Mua càng nhiều, giá mỗi gigabyte càng rẻ. Không thuê bao, không phí ẩn.

1 GB$3.70

$3.70 / GB

Chọn

5 GB$18.00

$3.60 / GB

Chọn

10 GB$34.50

$3.45 / GB

Chọn

Phổ biến

25 GB$80.00

$3.20 / GB

Chọn

50 GB$147.50

$2.95 / GB

Chọn

100 GB$270.00

$2.70 / GB

Chọn

250 GB$612.50

$2.45 / GB

Chọn

500 GB$1,100.00

$2.20 / GB

Chọn

1 TB$2,000.00

$2.00 / GB

Chọn

Gói

Giá mỗi GB

Tổng

1 GB

$3.70 / GB

$3.70Chọn

5 GB

$3.60 / GB

$18.00Chọn

10 GB

$3.45 / GB

$34.50Chọn

Phổ biến

25 GB

$3.20 / GB

$80.00Chọn

50 GB

$2.95 / GB

$147.50Chọn

100 GB

$2.70 / GB

$270.00Chọn

250 GB

$2.45 / GB

$612.50Chọn

500 GB

$2.20 / GB

$1,100.00Chọn

1 TB

$2.00 / GB

$2,000.00Chọn

Phương thức thanh toán khả dụng

Những câu hỏi mà người mua thực sự quan tâm

Vẫn đang cân nhắc liệu pool Ả Rập Xê-út có phù hợp với nhu cầu của bạn? Đội ngũ của chúng tôi giải đáp những câu hỏi thực tế trên Telegram, suốt ngày đêm — cả trước và sau khi bạn mua.

Trò chuyện trên Telegram

Liệu chúng có dùng được để thu thập dữ liệu trên Noon và Amazon Saudi Arabia không?

Đó chính là trường hợp sử dụng chính. Nguồn IP là dân cư thật — đúng đặc điểm mà các hệ thống chống bot của sàn thương mại địa phương tìm kiếm — nên những IP này lọt qua ở nơi mà các dải IP trung tâm dữ liệu bị chặn. Kết quả vẫn phụ thuộc vào fingerprint, header và nhịp độ gửi yêu cầu của bạn, vì vậy hãy kết hợp proxy với một cấu hình sạch và đúng múi giờ Ả Rập Xê Út (AST, UTC+3).

Tôi có thể nhắm mục tiêu một thành phố cụ thể như Riyadh hoặc Jeddah không?

Có, trong phạm vi quy mô nguồn IP cho phép — hãy chọn thành phố khi bạn tạo danh sách proxy. Các bảng phủ sóng của chúng tôi hiển thị mức truy cập đồng thời theo thời gian thực cho Riyadh, Jeddah, Dammam, Medina, Khobar và nhiều nơi khác, để bạn thấy được thành phố nào có khả năng xoay vòng IP dồi dào nhất trước khi quyết định.

Các IP có được định vị địa lý bên trong Ả Rập Xê Út không?

Có — nguồn IP phân giải về Ả Rập Xê Út với độ chính xác địa lý 99,95% khi đối chiếu với Cloudflare và trên nhiều cơ sở dữ liệu địa lý độc lập trong lần benchmark gần nhất của chúng tôi, với các nút thoát được xác nhận tại Jeddah (JED) và Riyadh (RUH). Độ chính xác địa lý cao chính là tín hiệu cốt lõi mà các cổng chặn theo vùng của Ả Rập Xê Út dựa vào.

Tôi có thể giữ một IP Ả Rập Xê Út cho một quy trình nhiều bước không?

Có. Hãy dùng phiên sticky để giữ cùng một IP thoát Ả Rập Xê Út — thời gian giữ trung bình khoảng 4,6 giờ và lên tới 72 giờ trong phép đo 3 ngày của chúng tôi. Để mở rộng quy mô, hãy xoay vòng theo từng yêu cầu: IP thoát thay đổi trong 97% số yêu cầu, thường là chuyển hẳn sang một nhà cung cấp khác chứ không chỉ là một địa chỉ lân cận.

Những nhà mạng và thành phố nào của Ả Rập Xê Út được phủ sóng?

Nguồn IP trải khắp 5 ISP của Ả Rập Xê Út, dẫn đầu là STC, Mobily, Zain KSA và Zajil, qua 8 vùng và 19 thành phố — từ Riyadh và Jeddah đến Dammam, Medina, Khobar, Buraidah và Dhahran.