Add Python proxy waterfall examples

README.md

@@ -57,6 +57,46 @@ python python/run_tests.py requests httpx
 ### Other Python scripts
 
 * [requests-random-proxy.py](python/requests-random-proxy.py) - Random proxy rotation
+* [requests-waterfall-proxy.py](python/requests-waterfall-proxy.py) - Ordered proxy waterfall (cheap tiers first)
+* [requests-waterfall-cache-proxy.py](python/requests-waterfall-cache-proxy.py) - Waterfall with TTL decision cache per host
+* [curl-cffi-waterfall-proxy.py](python/curl-cffi-waterfall-proxy.py) - Waterfall with free TLS-fingerprint tier (`curl_cffi`)
+
+These waterfall examples follow the [proxy waterfall](https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04) pattern: try direct (and optionally TLS impersonation) before datacenter / residential / unlocker proxies, escalate only when content validation fails, and optionally remember the winning tier. Shared helpers live in [waterfall_common.py](python/waterfall_common.py). They are not part of `run_tests.py`.
+
+```bash
+# At least one proxy tier required
+export PROXY_URL='http://user:pass@proxy.example.com:8080'
+# Optional more expensive tiers
+export PROXY_URL_RESIDENTIAL='http://user:pass@residential.example:8080'
+export PROXY_URL_UNLOCKER='http://user:pass@unlocker.example:8080'
+
+# Optional content checks (defaults work with api.ipify.org)
+export EXPECT_MUST_CONTAIN='ip'
+export EXPECT_MIN_BYTES=10
+
+python python/requests-waterfall-proxy.py
+
+# Decision cache (optional file persistence)
+export WATERFALL_CACHE_PATH=/tmp/waterfall-cache.json
+export WATERFALL_CACHE_TTL=86400
+python python/requests-waterfall-cache-proxy.py
+
+# TLS fingerprint tier (install optional dep first)
+pip install 'curl_cffi>=0.6.0'
+python python/curl-cffi-waterfall-proxy.py
+```
+
+| Variable | Purpose |
+|----------|---------|
+| `PROXY_URL` / `PROXY_URL_DATACENTER` | Datacenter tier (also `HTTPS_PROXY`) |
+| `PROXY_URL_RESIDENTIAL` | Residential tier |
+| `PROXY_URL_UNLOCKER` | Managed anti-bot / unlocker tier |
+| `EXPECT_STATUS` | Required HTTP status (default `200`) |
+| `EXPECT_MIN_BYTES` | Minimum body length (default `0`) |
+| `EXPECT_MUST_CONTAIN` | Required body substring |
+| `EXPECT_BLOCK_MARKERS` | Extra soft-block markers (comma-separated) |
+| `WATERFALL_CACHE_TTL` | Cache TTL seconds (default `86400`) |
+| `WATERFALL_CACHE_PATH` | Optional JSON cache file |
 
 > **Note:** Like the Ruby, JavaScript, and PHP examples here, these scripts use each library's normal proxy options only. Most of them do not send custom headers on the HTTPS `CONNECT` tunnel or surface proxy `CONNECT` response headers. For that, see [python-proxy-headers](https://github.com/proxymesh/python-proxy-headers) or [scrapy-proxy-headers](https://github.com/proxymesh/scrapy-proxy-headers).
 

python/curl-cffi-waterfall-proxy.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""
+Proxy waterfall with a free TLS-fingerprint tier (article Tier 0.5).
+
+Same ladder and content validation as requests-waterfall-proxy.py, but inserts
+a direct request that impersonates Chrome via curl_cffi before spending on
+proxy IPs. Requires: pip install curl_cffi
+
+Tiers (skip missing optional proxies):
+  1. direct              — curl_cffi, no impersonation, no proxy
+  2. tls                 — curl_cffi impersonate=chrome, no proxy
+  3. datacenter          — PROXY_URL / PROXY_URL_DATACENTER / HTTPS_PROXY
+  4. residential         — PROXY_URL_RESIDENTIAL
+  5. unlocker            — PROXY_URL_UNLOCKER
+
+Example:
+    pip install 'curl_cffi>=0.6.0'
+    export PROXY_URL='http://user:pass@us-ca.proxymesh.com:31280'
+    python python/curl-cffi-waterfall-proxy.py
+
+Article:
+https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04
+"""
+import sys
+
+from waterfall_common import (
+    build_tiers,
+    get_test_url,
+    load_expect,
+    print_config,
+    print_success,
+    require_proxy_tier,
+    run_waterfall,
+)
+
+try:
+    from curl_cffi import requests as curl_requests
+except ImportError:
+    print(
+        'Error: curl_cffi is required for this example\n'
+        "  pip install 'curl_cffi>=0.6.0'",
+        file=sys.stderr,
+    )
+    sys.exit(1)
+
+
+def fetch(tier, url):
+    kwargs = {'timeout': 30}
+    if tier.proxy_url:
+        kwargs['proxies'] = {'http': tier.proxy_url, 'https': tier.proxy_url}
+    if tier.impersonate:
+        kwargs['impersonate'] = tier.impersonate
+    response = curl_requests.get(url, **kwargs)
+    return response.status_code, response.text
+
+
+def main() -> int:
+    tiers = build_tiers(include_tls=True)
+    require_proxy_tier(tiers)
+    test_url = get_test_url()
+    expect = load_expect()
+    print_config(tiers, test_url, expect)
+
+    result = run_waterfall(tiers, test_url, expect, fetch)
+    if result.ok:
+        print_success(result)
+        return 0
+
+    print('\nAll tiers failed.')
+    return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

python/requests-waterfall-cache-proxy.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""
+Proxy waterfall with a TTL decision cache.
+
+Same ladder and content validation as requests-waterfall-proxy.py, plus
+remembering which tier worked for a host so later requests start there
+instead of re-walking cheaper tiers every time.
+
+Cache:
+    WATERFALL_CACHE_TTL   Seconds to remember a winning tier (default: 86400)
+    WATERFALL_CACHE_PATH  Optional JSON file path for persistence across runs
+
+On a cache hit, start at the cached tier. If that tier fails, invalidate the
+entry and continue down the remaining ladder. On success, refresh the cache.
+
+Example:
+    export PROXY_URL='http://user:pass@us-ca.proxymesh.com:31280'
+    export PROXY_URL_RESIDENTIAL='http://user:pass@residential.example:8080'
+    export WATERFALL_CACHE_PATH=/tmp/waterfall-cache.json
+    python python/requests-waterfall-cache-proxy.py
+
+Article:
+https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import requests
+
+from waterfall_common import (
+    build_tiers,
+    get_test_url,
+    load_expect,
+    print_config,
+    print_success,
+    require_proxy_tier,
+    run_waterfall,
+    tier_index_by_name,
+    url_pattern,
+)
+
+
+def fetch(tier, url):
+    proxies = None
+    if tier.proxy_url:
+        proxies = {'http': tier.proxy_url, 'https': tier.proxy_url}
+    response = requests.get(url, proxies=proxies, timeout=30)
+    return response.status_code, response.text
+
+
+def _cache_ttl() -> int:
+    return int(os.environ.get('WATERFALL_CACHE_TTL', '86400'))
+
+
+def _cache_path() -> Optional[Path]:
+    raw = os.environ.get('WATERFALL_CACHE_PATH')
+    return Path(raw) if raw else None
+
+
+def load_cache() -> Dict[str, Any]:
+    path = _cache_path()
+    if not path or not path.is_file():
+        return {}
+    try:
+        with path.open() as f:
+            data = json.load(f)
+        return data if isinstance(data, dict) else {}
+    except (OSError, json.JSONDecodeError) as exc:
+        print(f'Warning: could not read cache {path}: {exc}', file=sys.stderr)
+        return {}
+
+
+def save_cache(cache: Dict[str, Any]) -> None:
+    path = _cache_path()
+    if not path:
+        return
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        with path.open('w') as f:
+            json.dump(cache, f, indent=2, sort_keys=True)
+            f.write('\n')
+    except OSError as exc:
+        print(f'Warning: could not write cache {path}: {exc}', file=sys.stderr)
+
+
+def get_cached_tier(cache: Dict[str, Any], key: str) -> Optional[str]:
+    entry = cache.get(key)
+    if not isinstance(entry, dict):
+        return None
+    expires_at = entry.get('expires_at')
+    tier_name = entry.get('tier')
+    if not isinstance(expires_at, (int, float)) or not isinstance(tier_name, str):
+        return None
+    if time.time() >= expires_at:
+        return None
+    return tier_name
+
+
+def set_cached_tier(cache: Dict[str, Any], key: str, tier_name: str) -> None:
+    cache[key] = {
+        'tier': tier_name,
+        'expires_at': time.time() + _cache_ttl(),
+    }
+
+
+def invalidate(cache: Dict[str, Any], key: str) -> None:
+    cache.pop(key, None)
+
+
+def main() -> int:
+    tiers = build_tiers(include_tls=False)
+    require_proxy_tier(tiers)
+    test_url = get_test_url()
+    expect = load_expect()
+    print_config(tiers, test_url, expect)
+
+    key = url_pattern(test_url)
+    cache = load_cache()
+    cached_name = get_cached_tier(cache, key)
+    start_index = 0
+
+    if cached_name:
+        idx = tier_index_by_name(tiers, cached_name)
+        if idx is not None:
+            start_index = idx
+            print(f'Cache HIT for {key!r}: start at tier {cached_name!r} (index {idx})')
+        else:
+            print(
+                f'Cache HIT for {key!r}: tier {cached_name!r} not in current ladder; '
+                'starting from cheapest'
+            )
+            invalidate(cache, key)
+    else:
+        print(f'Cache MISS for {key!r}: start at cheapest tier')
+    print()
+
+    result = run_waterfall(
+        tiers, test_url, expect, fetch, start_index=start_index
+    )
+
+    if result.ok:
+        set_cached_tier(cache, key, result.tier.name)
+        save_cache(cache)
+        print(f'Cached winner {result.tier.name!r} for {key!r} (ttl={_cache_ttl()}s)')
+        print_success(result)
+        return 0
+
+    # Cached start failed and remaining ladder failed — drop stale entry.
+    if cached_name:
+        invalidate(cache, key)
+        save_cache(cache)
+        print(f'Invalidated cache for {key!r}')
+
+    print('\nAll tiers failed.')
+    return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

python/requests-waterfall-proxy.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""
+Basic proxy waterfall: try cheap tiers first, escalate only on failure.
+
+Uses content-level validation (status, min body size, soft-block markers,
+optional must_contain) so a soft block (HTTP 200 captcha page) escalates.
+
+Tiers (skip missing optional proxies):
+  1. direct              — no proxy
+  2. datacenter          — PROXY_URL / PROXY_URL_DATACENTER / HTTPS_PROXY
+  3. residential         — PROXY_URL_RESIDENTIAL
+  4. unlocker            — PROXY_URL_UNLOCKER
+
+Configuration via environment variables (see waterfall_common.py).
+
+Example:
+    export PROXY_URL='http://user:pass@us-ca.proxymesh.com:31280'
+    export PROXY_URL_RESIDENTIAL='http://user:pass@residential.example:8080'
+    python python/requests-waterfall-proxy.py
+
+Article:
+https://dev.to/votiakov/anti-bot-without-melting-your-budget-the-proxy-waterfall-4a04
+"""
+import sys
+
+import requests
+
+from waterfall_common import (
+    build_tiers,
+    get_test_url,
+    load_expect,
+    print_config,
+    print_success,
+    require_proxy_tier,
+    run_waterfall,
+)
+
+
+def fetch(tier, url):
+    proxies = None
+    if tier.proxy_url:
+        proxies = {'http': tier.proxy_url, 'https': tier.proxy_url}
+    response = requests.get(url, proxies=proxies, timeout=30)
+    return response.status_code, response.text
+
+
+def main() -> int:
+    tiers = build_tiers(include_tls=False)
+    require_proxy_tier(tiers)
+    test_url = get_test_url()
+    expect = load_expect()
+    print_config(tiers, test_url, expect)
+
+    result = run_waterfall(tiers, test_url, expect, fetch)
+    if result.ok:
+        print_success(result)
+        return 0
+
+    print('\nAll tiers failed.')
+    return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

python/requirements.txt

@@ -11,3 +11,5 @@ autoscraper>=1.1.0
 beautifulsoup4>=4.12.0,<4.13
 lxml>=5.0.0
 scrapy>=2.11.0
+# Optional: only needed for curl-cffi-waterfall-proxy.py (Tier 0.5 TLS fingerprint)
+# curl_cffi>=0.6.0