(country: str, force_refresh: bool = False)
| 634 | |
| 635 | |
| 636 | def fetch_proxy_pool(country: str, force_refresh: bool = False) -> List[str]: |
| 637 | cache = load_json(PROXY_CACHE_FILE, {}) |
| 638 | key = country.upper() |
| 639 | now = int(time.time()) |
| 640 | cached = cache.get(key) |
| 641 | gathered: List[str] = [] |
| 642 | |
| 643 | if (not force_refresh) and cached and now - int(cached.get("ts", 0)) < PROXY_CACHE_TTL: |
| 644 | gathered = cached.get("items", []) |
| 645 | else: |
| 646 | seen = set() |
| 647 | for loader in (fetch_proxies_from_proxyscrape_v4, fetch_proxies_from_geonode, fetch_proxies_from_proxyscrape_legacy): |
| 648 | for proxy in loader(key): |
| 649 | if proxy not in seen: |
| 650 | seen.add(proxy) |
| 651 | gathered.append(proxy) |
| 652 | cache[key] = {"ts": now, "items": gathered} |
| 653 | save_json(PROXY_CACHE_FILE, cache) |
| 654 | |
| 655 | bad = get_bad_proxy_map(key) |
| 656 | usable = [proxy for proxy in gathered if proxy not in bad] |
| 657 | |
| 658 | if len(usable) < MIN_PROXY_POOL and not force_refresh: |
| 659 | return fetch_proxy_pool(key, force_refresh=True) |
| 660 | return usable |
| 661 | |
| 662 | |
| 663 | def direct_fetch(sess: requests.Session, method: str, url: str, *, data=None, files=None, stream=False): |
no test coverage detected