Proxys d'arbitrage : votre tunnel vu depuis la géo cible

#!/usr/bin/env python3
"""Landing and redirect checker for Advanced.name proxies.

Opens each URL through a proxy from the target country, follows the redirect
chain by hand and reports the final address, HTTP status and the whole hop path.
Use it to see where a funnel really lands from a given geo before you buy traffic.
Standard library only. Run: python3 landings.py
"""

# ================================ CONFIG ====================================
CSV_URL = "https://get.advanced.name/YOUR_LIST_ID.csv"  # CSV export link from your dashboard
COUNTRY = "DE"          # open the funnel as a visitor from this country
MIN_SPEED = 3
LINKS_FILE = "links.txt"                 # one URL per line
OUT_FILE = "landings.csv"                # url;final_url;status;hops;proxy is written here
MAX_HOPS = 10
THREADS = 30
TIMEOUT = 20
# ============================================================================

import csv
import io
import sys
import threading
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from urllib.parse import urljoin

COL_IP, COL_PORT, COL_COUNTRY, COL_SPEED = 0, 1, 2, 3
UA = "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1 Mobile Safari/604.1"


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args):
        return None            # we follow redirects ourselves to record every hop


def load_proxies():
    raw = urllib.request.urlopen(CSV_URL, timeout=30).read().decode("utf-8", "replace")
    pool = []
    for row in csv.reader(io.StringIO(raw), delimiter=";"):
        if len(row) <= COL_SPEED or not row[COL_PORT].strip().isdigit():
            continue
        if COUNTRY and row[COL_COUNTRY].strip().upper() != COUNTRY.upper():
            continue
        if int(row[COL_SPEED] or 0) < MIN_SPEED:
            continue
        pool.append(row[COL_IP].strip() + ":" + row[COL_PORT].strip())
    return pool


def opener(proxy):
    return urllib.request.build_opener(
        NoRedirect, urllib.request.ProxyHandler({"http": "http://" + proxy, "https": "http://" + proxy}))


def trace(url, proxy):
    hops = 0
    op = opener(proxy)
    while hops < MAX_HOPS:
        req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept-Language": COUNTRY.lower()})
        try:
            resp = op.open(req, timeout=TIMEOUT)
        except urllib.error.HTTPError as err:
            resp = err
        status = resp.getcode()
        location = resp.headers.get("Location")
        if location and 300 <= status < 400:
            url = urljoin(url, location)
            hops += 1
            continue
        return url, status, hops
    return url, 599, hops


def main():
    proxies = load_proxies()
    if not proxies:
        sys.exit("No proxies for COUNTRY=%r." % COUNTRY)
    with open(LINKS_FILE, encoding="utf-8") as fh:
        links = [line.strip() for line in fh if line.strip() and not line.startswith("#")]
    print("proxies: %d (%s), links: %d" % (len(proxies), COUNTRY, len(links)))

    lock = threading.Lock()
    counter = {"i": 0}
    rows = []

    def check(url):
        with lock:
            proxy = proxies[counter["i"] % len(proxies)]
            counter["i"] += 1
        try:
            final, status, hops = trace(url, proxy)
        except Exception as exc:
            final, status, hops = "err: %s" % exc.__class__.__name__, 0, 0
        return url, final, status, hops, proxy

    with ThreadPoolExecutor(max_workers=THREADS) as pool:
        for fut in as_completed([pool.submit(check, u) for u in links]):
            row = fut.result()
            rows.append(row)
            print("%-3s hops=%d  %s -> %s" % (row[2], row[3], row[0][:40], str(row[1])[:50]))

    with open(OUT_FILE, "w", encoding="utf-8", newline="") as out:
        writer = csv.writer(out, delimiter=";")
        writer.writerow(["url", "final_url", "status", "hops", "proxy"])
        writer.writerows(rows)
    print("written %s" % OUT_FILE)


if __name__ == "__main__":
    main()