#!/usr/bin/env python3
"""
Proxy Qualifier — gate every proxy before it enters an account-creation flow.
Per candidate: alive+latency -> egress IP -> ip-api cleanliness -> Spamhaus DNSBL
-> ACTIVE Instagram probe (i.instagram.com) -> Google probe (accounts.google.com).
Only proxies passing ALL get 'ig_ok' and are eligible for the factory.
"""
import concurrent.futures as cf
import json
import re
import socket
import subprocess
import sys
import time
import urllib.request

DB = "/opt/android-fleet/fleet.db"
PROBE_IG = "https://i.instagram.com/api/v1/web/search/topsearch/?query=instagram"
PROBE_GOOGLE = "https://accounts.google.com/"
UA = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"}

def http_proxy(p):
    proto, host, port = p["proto"], p["host"], p["port"]
    return {proto: f"{proto}://{host}:{port}"} if proto == "http" else {"https": f"socks5h://{host}:{port}"} if proto == "socks5" else {"https": f"socks4://{host}:{port}"}

def get(url, proxy, timeout=15):
    req = urllib.request.Request(url, headers=UA)
    op = urllib.request.build_opener(urllib.request.ProxyHandler(proxy))
    with op.open(req, timeout=timeout) as r:
        return r.status, r.read(400).decode(errors="ignore")

def egress(proxy):
    try:
        st, body = get("https://api.ipify.org?format=json", proxy)
        return json.loads(body).get("ip") if st == 200 else None
    except Exception:
        try:
            st, body = get("http://ip-api.com/json/?fields=query", proxy, 10)
            return json.loads(body).get("query") if st == 200 else None
        except Exception:
            return None

def ipapi(ip):
    try:
        st, body = get(f"http://ip-api.com/json/{ip}?fields=status,country,isp,hosting,proxy,mobile", {}, 10)
        return json.loads(body) if st == 200 else {}
    except Exception:
        return {}

def spamhaus(ip):
    try:
        rev = ".".join(reversed(ip.split(".")))
        socket.gethostbyname(f"{rev}.zen.spamhaus.org")
        return True  # listed
    except socket.gaierror:
        return False
    except Exception:
        return None

def probe_ig(proxy):
    """Instagram verdict: any IG response = reachable; 429 = flagged; timeout = dead."""
    try:
        st, body = get(PROBE_IG, proxy)
        if st == 429:
            return "rate_limited", st
        # 200/401/404/405/etc all mean Instagram's servers ANSWERED this IP
        return "reachable", st
    except Exception as e:
        return "unreachable", str(e)[:60]

def probe_google(proxy):
    try:
        st, _ = get(PROBE_GOOGLE, proxy)
        return ("pass" if st in (200, 301, 302) else f"http{st}"), st
    except Exception as e:
        return "unreachable", str(e)[:60]

def qualify(p):
    res = {"name": p["name"], "host": p["host"], "port": p["port"], "proto": p["proto"]}
    px = http_proxy(p)
    t0 = time.time()
    ip = egress(px)
    res["latency_ms"] = int((time.time() - t0) * 1000)
    if not ip:
        res["verdict"] = "dead"
        return res
    res["egress"] = ip
    info = ipapi(ip)
    res["hosting"] = bool(info.get("hosting"))
    res["proxy_flag"] = bool(info.get("proxy"))
    res["mobile"] = bool(info.get("mobile"))
    res["country"] = info.get("country")
    res["spamhaus"] = spamhaus(ip)
    ig, igcode = probe_ig(px)
    res["ig"] = ig
    res["ig_code"] = igcode
    g, gcode = probe_google(px)
    res["google"] = g
    res["google_code"] = gcode
    clean = (not res["hosting"]) and (not res["proxy_flag"]) and (not res["spamhaus"])
    res["clean"] = clean
    # transparent proxies leak the LOCAL egress — they don't actually tunnel.
    local = res["egress"] in ("76.146.9.44", "10.30.20.85")
    res["transparent_leak"] = local
    ig_ok = (ig == "reachable" or ig == "rate_limited") and not local
    res["verdict"] = "ig_ok" if (ig_ok and g == "pass" and clean) else (
        "ig_only" if ig_ok else "fail")
    return res

def bank_proxies():
    import sqlite3
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    rows = con.execute("SELECT * FROM proxies WHERE health='healthy' ORDER BY clean DESC, latency_ms ASC LIMIT 40").fetchall()
    con.close()
    return [dict(r) for r in rows]

if __name__ == "__main__":
    mode = sys.argv[1] if len(sys.argv) > 1 else "bank"
    if mode == "bank":
        cands = bank_proxies()
    else:  # arbitrary list: "host:port:socks5" ...
        cands = []
        for spec in sys.argv[1:]:
            h, port, proto = spec.split(":")
            cands.append({"name": f"manual-{h}", "host": h, "port": int(port), "proto": proto})
    print(f"qualifying {len(cands)} candidates...", flush=True)
    results = []
    with cf.ThreadPoolExecutor(max_workers=8) as ex:
        for r in ex.map(qualify, cands):
            results.append(r)
    for r in results:
        print(json.dumps(r), flush=True)
    ok = [r for r in results if r["verdict"] in ("ig_ok", "ig_only")]
    print(f"\n=== {len(ok)}/{len(results)} proxies eligible for account flows ===", flush=True)
    for r in ok:
        print(f"  {r['name']} {r['host']}:{r['port']} egress={r['egress']} country={r['country']} "
              f"clean={r['clean']} ig={r['ig']} google={r['google']} lat={r['latency_ms']}ms", flush=True)