مصر

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

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

البروكسيات السكنية المصرية هي عناوين IP حقيقية صادرة عن منازل مصرية، يخصصها مزودو خدمة الإنترنت الاستهلاكيون مثل TE Data وVodafone Data وEtisalat Misr وLINKdotNET. تتيح لك تصفّح المتاجر والأسعار والمحتوى المصري وكشطه والتحقق منه بصفتك مستخدمًا حقيقيًا داخل البلد — عبر القاهرة والجيزة والإسكندرية والمنصورة — بدلًا من عنوان IP صادر عن مركز بيانات تكشفه منصات مثل Jumia Egypt وNoon Egypt والبوابات الجغرافية المحلية وتحجبه فور رصده.

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

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

83,389عناوين IP في المجموعة

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

8شبكات ISP

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

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

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

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

نوع المجموعة: سكنية، متعددة مزوّدي ISP
المدن المغطّاة: Cairo, Giza, Alexandria
أبرز مزوّدي ISP: TE Data · Vodafone Data · Etisalat Misr
البروتوكولات: HTTP، SOCKS5
التدوير: لكل طلب أو ثابت
الدقة الجغرافية: 99.87%
نظافة المجموعة: 99.2% 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.2%

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

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

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

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

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

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

زمن الاستجابة p90 / p95541 / 746 ms

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

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

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

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

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

شبكات ISP (ASN)8

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

أبرز المشغّلينTE Data · Vodafone Data · Etisalat Misr

جرى القياس على 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.[/]")

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

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

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

المنطقة	زمن الاستجابة	متوسط المتصلين	المجموعة
Cairo Governorate	429 ms	~3,646	مرتفع
Giza	426 ms	~972	مرتفع
Alexandria	436 ms	~718	مرتفع
Dakahlia	431 ms	~431	مرتفع
Gharbia	432 ms	~242	مرتفع
Sharqia	446 ms	~241	مرتفع
Qalyubia	447 ms	~165	مرتفع
Damietta Governorate	439 ms	~147	مرتفع
Ismailia Governorate	435 ms	~114	متوسط
Beni Suweif	433 ms	~100	متوسط
Port Said	420 ms	~81	متوسط
Red Sea	445 ms	~77	متوسط
Monufia	426 ms	~73	متوسط
Beheira	422 ms	~72	متوسط
Suez	458 ms	~65	متوسط
Asyut	443 ms	~57	منخفض
Minya	399 ms	~49	منخفض
Sohag	406 ms	~42	منخفض
Aswan	421 ms	~42	منخفض
Qena	410 ms	~40	منخفض
Luxor	461 ms	~29	منخفض
Faiyum	417 ms	~10	منخفض
Kafr el-Sheikh	492 ms	~6	منخفض

المدينة	زمن الاستجابة	متوسط المتصلين	المجموعة
Cairo	430 ms	~3,665	مرتفع
Giza	421 ms	~882	مرتفع
Alexandria	437 ms	~761	مرتفع
Al Mansurah	428 ms	~343	مرتفع
Tanta	429 ms	~215	مرتفع
New Cairo	423 ms	~151	مرتفع
Damietta	441 ms	~147	مرتفع
Zagazig	447 ms	~139	مرتفع
Ismailia	435 ms	~125	مرتفع
Port Said	420 ms	~89	متوسط
Banī Suwayf	425 ms	~88	متوسط
Hurghada	445 ms	~85	متوسط
Ash-Shaykh Zāyid	440 ms	~74	متوسط
Maadi	443 ms	~73	متوسط
Banhā	447 ms	~73	متوسط
Suez	458 ms	~71	متوسط
Asyut	443 ms	~63	منخفض
Mīt Ghamr	444 ms	~55	منخفض
Minya	399 ms	~54	منخفض
Al Qāhirah al Jadīdah	371 ms	~48	منخفض
Sohag	406 ms	~46	منخفض
Aswān	421 ms	~46	منخفض
Shibīn al Kawm	426 ms	~43	منخفض
Qina	407 ms	~42	منخفض
Kafr ash Shaykh	450 ms	~40	منخفض

المزوّد	زمن الاستجابة	متوسط المتصلين	المجموعة
TE Dataثابت	430 ms	~5,833	مرتفع
Vodafone Dataمحمول	447 ms	~590	مرتفع
Etisalat Misrمحمول	428 ms	~523	متوسط
LINKdotNETثابت	444 ms	~462	منخفض
Orange Egyptمحمول	470 ms	~10	منخفض

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

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

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

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

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

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

Amazon EgyptNoon EgyptJumia EgyptTemuB.TECH

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

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

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

FacebookInstagramTikTokYouTubeX (Twitter)

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

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

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

EgyptAirAlmatarAlmosaferWegoTravelstart

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

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

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

Watch iTShahid (MBC)OSN+TODSTARZPLAY

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

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

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

Dubizzle Egypt (OLX)Hatla2eeContactCarsWuzzuf

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

عناوين 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": "EG",
6  "city": "Cairo",
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

هل ستعمل هذه البروكسيات لكشط Jumia Egypt وNoon Egypt؟

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

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

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

هل تُحدَّد مواقع عناوين IP داخل مصر جغرافيًا؟

نعم — يتحدد موقع التجمّع في مصر بدقة جغرافية بلغت 99.87% مقابل Cloudflare وعبر قواعد بيانات جغرافية مستقلة في آخر اختبار مرجعي أجريناه، مع تأكيد عُقد الخروج في القاهرة (CAI). والدقة الجغرافية العالية هي الإشارة الأساسية التي تعتمد عليها البوابات الجغرافية المصرية.

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

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

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

يمتد التجمّع عبر 5 مزودي خدمة إنترنت مصريين تتصدرهم TE Data وVodafone Data وEtisalat Misr وLINKdotNET وOrange Egypt، عبر 23 منطقة و25 مدينة — من القاهرة والجيزة والإسكندرية إلى المنصورة وطنطا ودمياط وبورسعيد والغردقة.