كازاخستان

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

بروكسيات كازاخستان السكنية

البروكسيات السكنية الكازاخية هي عناوين IP حقيقية من منازل كازاخية، يُسنِدها مزوّدو خدمة الإنترنت الاستهلاكيون مثل Kazakhtelecom وBeeline (Kar-Tel) وTranstelecom وAlma Telecommunications. وهي تتيح لك تصفّح المتاجر والأسعار والمحتوى الكازاخي وكشطها والتحقق منها بصفتك مستخدماً محلياً حقيقياً داخل البلد — عبر Almaty وAstana وShymkent وKaraganda — بدلاً من عنوان IP تابع لمركز بيانات يكشفه ويحجبه على الفور كلٌّ من Kaspi Magazin وHalyk Market والبوابات الجغرافية المحلية.

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

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

148,455عناوين IP في المجموعة

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

45شبكات ISP

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

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

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

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

نوع المجموعة: سكنية، متعددة مزوّدي ISP
المدن المغطّاة: Almaty, Astana, Karaganda
أبرز مزوّدي ISP: Kazakhtelecom · Beeline (Kar-Tel) · Transtelecom
البروتوكولات: HTTP، SOCKS5
التدوير: لكل طلب أو ثابت
الدقة الجغرافية: 99.92%
نظافة المجموعة: 99.8% 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.8%

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

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

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

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

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

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

زمن الاستجابة p90 / p95472 / 533 ms

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

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

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

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

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

شبكات ISP (ASN)45

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

أبرز المشغّلينKazakhtelecom · Beeline (Kar-Tel) · Transtelecom

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

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

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

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

المنطقة	زمن الاستجابة	متوسط المتصلين	المجموعة
Almaty	379 ms	~3,410	مرتفع
Astana	387 ms	~2,854	مرتفع
Karaganda	393 ms	~1,108	مرتفع
Pavlodar Region	284 ms	~788	مرتفع
East Kazakhstan	319 ms	~665	مرتفع
Shymkent	338 ms	~558	مرتفع
Kostanay	422 ms	~558	مرتفع
Aktobe	332 ms	~543	متوسط
Mangystau	458 ms	~490	متوسط
Zhambyl	333 ms	~405	متوسط
North Kazakhstan	404 ms	~356	متوسط
Atyrau Oblisi	272 ms	~337	متوسط
West Kazakhstan	362 ms	~320	منخفض
Aqmola	279 ms	~313	منخفض
Abai Region	277 ms	~194	منخفض
Zhetysu Region	277 ms	~98	منخفض
Kyzylorda	412 ms	~81	منخفض
Ulytau Region	412 ms	~72	منخفض
Almaty Region	467 ms	~58	منخفض

المدينة	زمن الاستجابة	متوسط المتصلين	المجموعة
Almaty	380 ms	~3,474	مرتفع
Astana	387 ms	~2,897	مرتفع
Karaganda	392 ms	~1,107	مرتفع
Pavlodar	285 ms	~765	مرتفع
Shymkent	338 ms	~566	مرتفع
Aktobe	332 ms	~551	مرتفع
Ust-Kamenogorsk	289 ms	~499	مرتفع
Aktau	458 ms	~491	مرتفع
Kostanay	424 ms	~484	مرتفع
Atyrau	271 ms	~337	متوسط
Petropavl	404 ms	~326	متوسط
Kokshetau	279 ms	~309	متوسط
Oral	395 ms	~277	متوسط
Taraz	337 ms	~272	متوسط
Semey	277 ms	~197	متوسط
Glūbokoe	327 ms	~121	متوسط
Taldykorgan	275 ms	~97	منخفض
Kyzylorda	412 ms	~82	منخفض
Rudnyy	331 ms	~78	منخفض
Dzhezkazgan	412 ms	~71	منخفض
Belūsovka	316 ms	~52	منخفض
Sarykemer	383 ms	~50	منخفض
Merke	319 ms	~35	منخفض
Ekibastuz	281 ms	~32	منخفض
Fedorovka	304 ms	~32	منخفض

المزوّد	زمن الاستجابة	متوسط المتصلين	المجموعة
Kazakhtelecomثابت	267 ms	~7,490	مرتفع
Beeline (Kar-Tel)ثابت	411 ms	~2,852	مرتفع
Transtelecomثابت	469 ms	~854	مرتفع
Alma TVثابت	436 ms	~567	مرتفع
NLS Kazakhstanثابت	453 ms	~339	مرتفع
D-Cloudثابت	343 ms	~178	متوسط
XCOMثابت	462 ms	~151	متوسط
Uplinkثابت	458 ms	~140	متوسط
BTCOMثابت	413 ms	~135	متوسط
Optinetثابت	408 ms	~129	منخفض
Asketثابت	456 ms	~113	منخفض
BAINETثابت	406 ms	~99	منخفض
BiKaDaثابت	459 ms	~86	منخفض
Compastelecomثابت	340 ms	~72	منخفض

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

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

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

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

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

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

Kaspi MagazinWildberries KZOzonHalyk MarketTechnodom

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

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

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

InstagramTelegramTikTokVKFacebook

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

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

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

Aviata.kzChocotravelAir AstanaFlyArystanAviasales KZ

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

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

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

Kazakhtelecom TV+Beeline TVMegogoIviOkko

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

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

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

Kolesa.kzKrisha.kzOLX.kzMarket.kzMycar.kz

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

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

هل ستعمل هذه البروكسيات لكشط Kaspi Magazin وHalyk Market؟

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

هل يمكنني استهداف مدينة محددة مثل Almaty أو Astana؟

نعم، حسبما يسمح عمق التجمّع — اختر المدينة عند توليد قائمة البروكسي. تعرض جداول التغطية لدينا التزامن الحيّ لكلٍّ من Almaty وAstana وKaraganda وPavlodar وShymkent والمزيد، فترى أي المدن تتمتع بأعمق تدوير قبل أن تختار.

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

نعم — يُحدَّد موقع التجمّع في كازاخستان بدقة جغرافية بلغت 99.92% مقابل Cloudflare وعبر قواعد بيانات جغرافية مستقلة في آخر اختبار مرجعي لنا، مع عُقد خروج مؤكَّدة في Aktobe (AKX) وAlmaty (ALA). والدقة الجغرافية العالية هي الإشارة الأساسية التي تعتمد عليها البوابات الجغرافية الكازاخية.

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

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

ما مشغّلو الاتصالات والمدن الكازاخية المشمولون؟

يمتد التجمّع عبر 14 مزوّد خدمة إنترنت كازاخي تتصدرها Kazakhtelecom وBeeline (Kar-Tel) وTranstelecom وAlma Telecommunications وNLS Kazakhstan، عبر 19 منطقة و25 مدينة — من Almaty وAstana إلى Karaganda وPavlodar وShymkent وAktobe وUst-Kamenogorsk وAtyrau.