المملكة العربية السعودية

عناوين IP منزلية حقيقية

بروكسيات المملكة العربية السعودية السكنية

البروكسيات السكنية السعودية هي عناوين IP حقيقية من منازل سعودية، تخصصها شركات الاتصالات الاستهلاكية مثل STC وMobily وZain KSA. وهي تتيح لك تصفح المتاجر السعودية وأسعارها ومحتواها وجمع بياناتها والتحقق منها بصفتك مستخدمًا محليًا حقيقيًا داخل المملكة — في الرياض وجدة والدمام والمدينة المنورة — بدلًا من عنوان IP تابع لمركز بيانات تكشفه وتحجبه فورًا منصات مثل Noon وAmazon Saudi Arabia والبوابات الجغرافية المحلية.

جرِّب مقابل $1.99/GB تسجيل الدخول عبر Google

HTTP / SOCKS5استهداف حسب المدينة ومزوّد ISPحركة المرور لا تنتهي صلاحيتها أبدًاتحقُّق شهري من النظافة

50,924عناوين IP في المجموعة

19+المدن المغطّاة

12شبكات ISP

336 msزمن الاستجابة الوسيط

99.6%نظيفة تمامًا

بروكسيات المملكة العربية السعودية السكنية في لمحة

بروكسيات المملكة العربية السعودية السكنية هي عناوين IP حقيقية من منازل المملكة العربية السعودية، مُخصَّصة من مزوّدي خدمة الإنترنت للمستهلكين. تتيح لك تصفّح مواقع المملكة العربية السعودية واستخراج بياناتها والتحقّق من أسعارها ومحتواها كمستخدم محلي حقيقي داخل البلد — بدلًا من عنوان IP تابع لمركز بيانات تكتشفه المواقع المحلية وتحظره.

نوع المجموعة: سكنية، متعددة مزوّدي ISP
المدن المغطّاة: Riyadh, Jeddah, Dammam
أبرز مزوّدي ISP: STC (Saudi Telecom) · Mobily · Zain KSA
البروتوكولات: HTTP، SOCKS5
التدوير: لكل طلب أو ثابت
الدقة الجغرافية: 99.95%
نظافة المجموعة: 99.6% clean
التسعير: from $2.00/GB
آخر قياس: 2026-06-23
الدعم: 24/7

لماذا تختار الفِرق مجموعتنا السكنية في المملكة العربية السعودية

عناوين IP سكنية حقيقية، واستهداف دقيق، وجودة لا يلزمك تصديقها على عَواهنها.

عناوين IP سكنية حقيقية

اتصالات منزلية أصيلة من مزوّدي خدمة الإنترنت للمستهلكين في المملكة العربية السعودية — وليست نطاقات مراكز بيانات تُحظر فور رصدها.

استهداف حسب المدينة ومزوّد ISP

استهدِف مدنًا ومشغّلين بعينهم حيثما يسمح عمق المجموعة — وصولًا إلى مستوى المزوّد.

متدوِّرة أو ثابتة

تدوير لكل طلب لتوسيع النطاق، أو جلسات ثابتة حتى نحو 72 ساعة للتدفّقات متعددة الخطوات.

HTTP و SOCKS5

تعمل مع أي عميل أو أداة استخراج أو منظومة أتمتة فور الاستخدام ودون إعداد إضافي.

مصدر أخلاقي

عناوين IP من شبكة سكنية قائمة على الاشتراك الطوعي — مستخدمون حقيقيون وافقوا على مشاركة عرض النطاق الترددي.

تحقُّق شهري من النظافة

نعيد قياس المجموعة كل شهر وننشر الأرقام، لتبقى محدَّثة دائمًا.

جودة يمكنك التحقّق منها

كل شهر نُجري الاختبار المرجعي نفسه — عيّنة عشوائية من عناوين IP تُقيَّم مقابل Scamalytics، ويُختبر زمن الاستجابة من VPS محايد، وتُفحص الدقة الجغرافية عبر قواعد بيانات مستقلة. تعكس الأرقام أدناه بيئة الإنتاج، لا تشغيلًا مخبريًا في أفضل الظروف.

النظافة Scamalytics · 100,000 عنوان IP

نظيفة تمامًا99.6%

تلوُّث بمراكز البيانات0%

على أي قائمة سوداء0.2%

مخاطر احتيال منخفضة (0–19)99.4%

الدقة الجغرافية99.95%

زمن الاستجابة والسرعة VPS ← Cloudflare

زمن الاستجابة الوسيط336ms

زمن الاستجابة p90 / p95459 / 510 ms

نجاح الاتصال99.1%

الإنتاجية (الوسيط / p95)0.95 / 3.02 MB/s

إتمام تنزيل الملفات الكبيرة98%

تركيبة الشبكة

نوع الاتصالسكني 100%

شبكات ISP (ASN)12

أوضاع التدويرلكل طلب · ثابت

أبرز المشغّلينSTC (Saudi Telecom) · Mobily · Zain KSA

جرى القياس على 100,000 عنوان من VPS محايد مقابل حافة شبكة Cloudflare، وهي دقيقة حتى آخر اختبار مرجعي. قد يختلف أي تشغيل مكرَّر بهامش بسيط، لكن الأرقام تعكس رتبة المقدار تعكيسًا كاملًا وتُعدّ أساسًا سليمًا للحكم على مدى ملاءمة المجموعة لحالة استخدامك. لا يُنشر معدّل نجاح الاستخراج لكل هدف على حدة — فهو يعتمد على بصمتك وإعداداتك، لا على عنوان IP وحده.

benchmark.py — اختبار مرجعي مفتوح للمجموعة

الاختبار المرجعي المفتوح الكامن خلف الأرقام أعلاه. يأخذ عيّنة عشوائية من عناوين IP في المجموعة ويقيس أربعة أشياء — النظافة (مخاطر الاحتيال وفق Scamalytics، ومراكز البيانات، والقائمة السوداء، والدقة الجغرافية)، وزمن الاستجابة (الذهاب والإياب إلى حافة Cloudflare و ip-api)، والتدوير (عناوين IP الفريدة، و ASN، والشبكات الفرعية /24)، والإنتاجية (سرعة التنزيل والزمن حتى أول بايت) — ثم يُبلِّغ عن كلٍّ منها. وقد نُشر بالكامل ليبقى المنهج الكامن خلف كل رقم مفتوحًا للتدقيق.

# -*- 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.[/]")

استهدِف وصولًا إلى مستوى المدينة

«متوسط المتصلين» هو متوسط عدد الأجهزة المتمايزة المتصلة والمتاحة في اللحظة نفسها — وليس إجماليًا تراكميًا. وهو مُصنَّف إلى مرتفع / متوسط / منخفض.

المناطقالمدنالمزوّدون

المنطقة	زمن الاستجابة	متوسط المتصلين	المجموعة
Riyadh Region	332 ms	~1,713	مرتفع
Mecca Region	323 ms	~1,406	مرتفع
Eastern Province	356 ms	~895	مرتفع
Medina Region	336 ms	~329	متوسط
Al-Qassim Region	328 ms	~107	متوسط
'Asir Region	365 ms	~47	منخفض
Tabuk Region	303 ms	~28	منخفض
Ha'il Region	382 ms	~4	منخفض

المدينة	زمن الاستجابة	متوسط المتصلين	المجموعة
Riyadh	332 ms	~1,714	مرتفع
Jeddah	323 ms	~1,363	مرتفع
Dammam	354 ms	~668	مرتفع
Medina	336 ms	~323	مرتفع
Khobar	433 ms	~109	مرتفع
Buraidah	330 ms	~98	مرتفع
Dhahran	365 ms	~44	مرتفع
Abha	372 ms	~41	متوسط
Al Qaţīf	351 ms	~32	متوسط
Makkah	308 ms	~29	متوسط
Hofuf	349 ms	~29	متوسط
Tabuk	303 ms	~28	متوسط
Ta'if	336 ms	~15	منخفض
Unaizah	306 ms	~9	منخفض
Şafwá	335 ms	~7	منخفض
Al Markaz	417 ms	~6	منخفض
Sulţānah	303 ms	~6	منخفض
An Nimas	312 ms	~4	منخفض
Baq`a' ash Sharqiyah	385 ms	~4	منخفض

المزوّد	زمن الاستجابة	متوسط المتصلين	المجموعة
STC (Saudi Telecom)ثابت	318 ms	~3,521	مرتفع
Mobilyمحمول	433 ms	~933	مرتفع
Zain KSAمحمول	502 ms	~55	متوسط
STCمحمول	399 ms	~15	منخفض
Zajilمحمول	401 ms	~6	منخفض

الدقة الجغرافية — كل عنوان IP يُباع على أنه من المملكة العربية السعودية يُحدَّد جغرافيًا في المملكة العربية السعودية عبر ثلاث قواعد بيانات جغرافية مستقلة (db-ip و MaxMind و ipinfo)؛ والدقة الجغرافية العالية هي الإشارة التي تعتمد عليها بوابات التحقّق الجغرافي المحلية.

بروكسيات المملكة العربية السعودية حسب حالة الاستخدام

مُنظَّمة حسب ما تفعله، لا حسب المنصّة. وتُدرَج الخدمات المحلية حيث يكون لها أهمية.

التجارة الإلكترونية وأبحاث السوق

افحص الواجهات والأسعار والقوائم المحلية من داخل السوق.

مراقبة الواجهات
تتبُّع الأسعار المحلية
أبحاث البائعين

NoonAmazon Saudi ArabiaJarir BookstoreeXtraSACO

التحقّق من الإعلانات والأداء

شاهِد كيف تظهر الإعلانات والتصاميم ونتائج SERP لمستخدم محلي — دون تبديل يدوي.

نتائج SERP المحلية
فحوص التصاميم
صفحات الهبوط

SnapchatTikTokInstagramX (Twitter)Facebook

السفر وذكاء الأسعار

التقِط أسعار الرحلات والفنادق كما يراها العميل المحلي.

أسعار محدَّدة جغرافيًا
مراقبة العروض
مقارنة الأسعار

SaudiaflynasflyadealAlmosaferRehlat

البث والتحقّق الجغرافي

تحقّق من الكتالوجات المقيَّدة إقليميًا والبوابات الجغرافية من الموقع الصحيح.

الفحوص الجغرافية
التحقّق من الكتالوجات
التوافر الإقليمي

Shahid (MBC)STC TVRotanaStarzPlay ArabiaOSN+

الإعلانات المبوَّبة والقوائم المحلية

راقِب السلع المستعملة والسيارات والإيجارات عبر الإعلانات المبوَّبة المحلية.

مراقبة القوائم
تتبُّع الأسعار
التغطية الإقليمية

HarajOpenSooqSyarahMotoryCarSwitch

التطوير والأتمتة

عناوين IP سكنية لخطوط المعالجة والاختبارات وجمع البيانات المسموح به عبر الـ API.

تكامل خطوط المعالجة
إدارة الـ API
الجلسات الثابتة

PythonScrapycURLGitHub

إلى جانب المواقع المحلية، تعمل مجموعة المملكة العربية السعودية نفسها مع كبرى المنصّات العالمية — Google و Meta و Amazon و YouTube و TikTok و Netflix — كما تظهر من النطاق المحلي لـ المملكة العربية السعودية: الأسعار والإعلانات والكتالوجات ونتائج SERP والمحتوى المحلية.

المطوّر API

أنشئ نقاط اتصال محدَّدة النطاق لـ المملكة العربية السعودية من لوحة التحكم، وانسخ المقتطفات المختبَرة، أو أتمِت قوائم البروكسي عبر الـ API العام. تبقى المدينة ومزوّد ISP والجلسات الثابتة والقوائم المحفوظة ضمن سير عمل واحد نظيف.

وثائق الـ 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"])

باقات بلا شروط مخفية

كلما زاد الحجم انخفض سعر الجيجابايت. لا اشتراكات ولا رسوم مخفية.

1 GB$3.70

$3.70 / GB

اختيار

5 GB$18.00

$3.60 / GB

اختيار

10 GB$34.50

$3.45 / GB

اختيار

الأكثر شيوعًا

25 GB$80.00

$3.20 / GB

اختيار

50 GB$147.50

$2.95 / GB

اختيار

100 GB$270.00

$2.70 / GB

اختيار

250 GB$612.50

$2.45 / GB

اختيار

500 GB$1,100.00

$2.20 / GB

اختيار

1 TB$2,000.00

$2.00 / GB

اختيار

الباقة

السعر لكل ⁦GB⁩

الإجمالي

1 GB

$3.70 / GB

$3.70اختيار

5 GB

$3.60 / GB

$18.00اختيار

10 GB

$3.45 / GB

$34.50اختيار

الأكثر شيوعًا

25 GB

$3.20 / GB

$80.00اختيار

50 GB

$2.95 / GB

$147.50اختيار

100 GB

$2.70 / GB

$270.00اختيار

250 GB

$2.45 / GB

$612.50اختيار

500 GB

$2.20 / GB

$1,100.00اختيار

1 TB

$2.00 / GB

$2,000.00اختيار

طرق الدفع المتاحة

أسئلة يطرحها المشترون فعلًا

ما زلت تزن مدى ملاءمة مجموعة المملكة العربية السعودية لحالة استخدامك؟ يجيب فريقنا عن الأسئلة الحقيقية على Telegram، على مدار الساعة — قبل الشراء وبعده.

الدردشة على Telegram

هل ستصلح هذه البروكسيات لجمع البيانات من Noon وAmazon Saudi Arabia؟

هذه هي حالة الاستخدام الأساسية. فالتجمّع سكني حقيقي — وهي السمة التي تبحث عنها أنظمة مكافحة الروبوتات في الأسواق المحلية — لذا تمرّ عناوين IP هذه حيث تُحجب نطاقات مراكز البيانات. وتبقى النتائج رهنًا ببصمتك وترويساتك ووتيرة طلباتك، لذا اقرن البروكسي بإعداد نظيف وتوقيت سعودي مطابق (AST، UTC+3).

هل يمكنني استهداف مدينة محددة مثل الرياض أو جدة؟

نعم، حيثما يسمح عمق التجمّع — اختر المدينة عند توليد قائمة البروكسي. تعرض جداول التغطية لدينا التزامن المباشر للرياض وجدة والدمام والمدينة المنورة والخبر وغيرها، كي تتبيّن المدن صاحبة أعمق تدوير قبل أن تختار.

هل تتحدد المواقع الجغرافية لعناوين IP داخل السعودية؟

نعم — يتحدد موقع التجمّع في السعودية بدقة جغرافية تبلغ 99.95% مقابل Cloudflare وعبر قواعد بيانات جغرافية مستقلة في أحدث اختبار مرجعي لدينا، مع تأكيد عقد خروج في جدة (JED) والرياض (RUH). والدقة الجغرافية العالية هي الإشارة الأساسية التي تعتمد عليها البوابات الجغرافية السعودية.

هل يمكنني الاحتفاظ بعنوان IP سعودي واحد طوال تدفق متعدد الخطوات؟

نعم. استخدم جلسة لاصقة للإبقاء على عنوان IP الخروج السعودي نفسه — يبلغ متوسط مدة الاحتفاظ نحو 4.6 ساعة ويصل إلى 72 ساعة في قياسنا الممتد على 3 أيام. وللتوسّع، بدّل العنوان مع كل طلب: يتغير عنوان IP الخروج في 97% من الطلبات، وعادةً إلى مزوّد مختلف، لا إلى عنوان مجاور فحسب.

ما شركات الاتصالات والمدن السعودية المشمولة بالتغطية؟

يمتد التجمّع عبر 5 مزودي خدمة إنترنت سعوديين تتصدرهم STC وMobily وZain KSA وZajil، عبر 8 مناطق و19 مدينة — من الرياض وجدة إلى الدمام والمدينة المنورة والخبر وبريدة والظهران.