#!/usr/bin/env python3
"""
ProxyFly integration: harvest free proxies -> live-test -> cleanliness-grade
(hosting/proxy flags + Spamhaus DNSBL) -> feed the fleet proxy bank.
Every machine gets a verified egress IP, 100% backend, zero-touch.
"""
import json
import os
import socket
import sqlite3
import threading
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor

DB_PATH = "/opt/android-fleet/fleet.db"
SOURCES = [
    ("proxifly", "https://cdn.jsdelivr.net/gh/proxifly/free-proxy-list@main/proxies/all/data.json"),
    ("geonode", "https://proxylist.geonode.com/api/proxy-list?limit=500&page=1&sort_by=lastChecked&sort_type=desc"),
    ("proxyscrape", "https://api.proxyscrape.com/v4/free-proxy-list/get?request=display_proxies&proxy_format=protocolipport&format=json"),
]
MAX_FETCH = 200          # candidates per cycle
MAX_TEST = 120           # live-test at most this many per cycle
MAX_POOL = 60            # cap the proxies table (perf)
TEST_TIMEOUT = 7
HARVEST_INTERVAL = 480   # 8 min — free proxies churn fast, harvest often
MAX_LATENCY_MS = 4000    # drop anything slower than this (routeable only)

def log(msg):
    print(f"[harvest] {msg}", flush=True)

def dbcon():
    con = sqlite3.connect(DB_PATH, timeout=30)
    con.row_factory = sqlite3.Row
    return con

def fetch_candidates():
    cands = {"socks5": [], "socks4": [], "http": []}
    for src_name, src in SOURCES:
        try:
            req = urllib.request.Request(src, headers={
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/120 Safari/537.36",
                "Accept": "application/json"})
            with urllib.request.urlopen(req, timeout=45) as r:
                data = json.loads(r.read())
            if src_name == "proxifly":
                for p in data:
                    proto = p.get("protocol", "")
                    if proto not in ("socks5", "https", "http", "socks4"):
                        continue
                    # transparent = leaks real IP only on plain-HTTP; our traffic is
                    # HTTPS, so include them (mark them in the name later).
                    cands.setdefault(proto, []).append({
                        "host": p["ip"], "port": p["port"], "proto": proto,
                        "country": (p.get("geolocation") or {}).get("country", "?"),
                        "anon": p.get("anonymity", ""),
                    })
            elif src_name == "geonode":
                for p in data.get("data", []):
                    protos = [x for x in (p.get("protocols") or []) if x in ("socks5", "socks4", "http", "https")]
                    if not protos:
                        continue
                    for proto in protos[:1]:
                        cands.setdefault(proto, []).append({
                            "host": p["ip"], "port": int(p["port"]), "proto": proto,
                            "country": p.get("country", "?"),
                            "anon": p.get("anonymityLevel", ""),
                        })
            elif src_name == "proxyscrape":
                for p in data.get("proxies", []):
                    proto = p.get("protocol", "")
                    if proto not in ("socks5", "socks4", "http"):
                        continue
                    cands.setdefault(proto, []).append({
                        "host": p["ip"], "port": int(p["port"]), "proto": proto,
                        "country": p.get("country", "?"),
                        "anon": "elite",
                    })
        except Exception as e:
            log(f"source error {src_name}: {e}")
    # STRATIFY + SHUFFLE: feeds are protocol-grouped; sampling the head would
    # test only http proxies. Take an even mix across protocols.
    import random as _random
    out = []
    for pool in cands.values():
        _random.shuffle(pool)
    while len(out) < MAX_FETCH and any(cands.values()):
        for proto in ("socks5", "socks4", "http"):
            if cands[proto]:
                out.append(cands[proto].pop())
    # dedupe by host:port
    seen = set()
    final = []
    for c in out:
        k = f"{c['host']}:{c['port']}"
        if k not in seen:
            seen.add(k)
            final.append(c)
    return final[:MAX_FETCH]

def test_proxy(c):
    """Returns (c, egress_ip, latency_ms) or (c, None, None). HTTPS first, HTTP fallback."""
    proto_map = {"socks5": "socks5h://", "https": "https://", "http": "http://",
                 "socks4": "socks4://"}
    url = proto_map[c["proto"]] + f"{c['host']}:{c['port']}"
    import urllib.request as ur
    opener = ur.build_opener(ur.ProxyHandler({"http": url, "https": url}))
    opener.addheaders = [("User-Agent", "Mozilla/5.0 (Linux; Android 9) AppleWebKit/537.36")]
    for target in ("https://api.ipify.org", "http://api.ipify.org"):
        t0 = time.time()
        try:
            with opener.open(target, timeout=TEST_TIMEOUT) as r:
                ip = r.read().decode().strip()
            lat = int((time.time() - t0) * 1000)
            return (c, ip, lat)
        except Exception:
            continue
    return (c, None, None)

def check_clean(ips):
    """ip-api batch: flags for each ip. Returns {ip: dict}."""
    out = {}
    ips = [i for i in ips if i]
    for i in range(0, len(ips), 15):
        batch = ips[i:i + 15]
        try:
            req = urllib.request.Request(
                "http://ip-api.com/batch?fields=status,country,isp,as,hosting,proxy,mobile,query",
                data=json.dumps(batch).encode(),
                headers={"Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=20) as r:
                res = json.loads(r.read())
            for e in res:
                if e.get("status") == "success":
                    out[e["query"]] = e
        except Exception as e:
            log(f"ip-api batch error: {e}")
        time.sleep(4)  # 45 req/min limit — throttle hard
    return out

def dnsbl_listed(ip):
    """Spamhaus ZEN: True if listed."""
    try:
        rev = ".".join(reversed(ip.split("."))) + ".zen.spamhaus.org"
        socket.gethostbyname(rev)
        return True
    except socket.gaierror:
        return False
    except Exception:
        return False

def prune_pool():
    con = dbcon()
    # drop dead proxifly entries every cycle (they die fast — keep the list honest)
    con.execute("DELETE FROM proxies WHERE source='proxifly' AND health='dead' AND assigned_mac IS NULL")
    rows = con.execute("SELECT id, egress_ip, source FROM proxies ORDER BY id").fetchall()
    # remove duplicates by egress ip (keep lowest id)
    seen = set()
    for r in rows:
        if r["egress_ip"] and r["egress_ip"] in seen:
            con.execute("DELETE FROM proxies WHERE id=?", (r["id"],))
        elif r["egress_ip"]:
            seen.add(r["egress_ip"])
    # cap pool size (drop oldest proxifly entries beyond cap)
    over = [r for r in rows if r["source"] == "proxifly"]
    if len(over) > MAX_POOL:
        for r in over[:-MAX_POOL]:
            con.execute("DELETE FROM proxies WHERE id=?", (r["id"],))
    con.commit()
    con.close()

def harvest():
    log(f"cycle start — fetching {len(SOURCES)} source(s)")
    cands = fetch_candidates()
    log(f"{len(cands)} candidates, testing up to {MAX_TEST}")
    tested = []
    with ThreadPoolExecutor(max_workers=20) as ex:
        for c, ip, lat in ex.map(test_proxy, cands[:MAX_TEST]):
            if ip:
                c["egress_ip"] = ip
                c["latency_ms"] = lat
                tested.append(c)
    log(f"{len(tested)} proxies alive; grading cleanliness")
    flags = check_clean([c["egress_ip"] for c in tested])
    dns_flags = {}
    with ThreadPoolExecutor(max_workers=20) as ex:
        for c in tested:
            dns_flags[c["egress_ip"]] = ex.submit(dnsbl_listed, c["egress_ip"])
    for k, f in dns_flags.items():
        dns_flags[k] = f.result()
    con = dbcon()
    added = 0
    for c in tested:
        f = flags.get(c["egress_ip"], {})
        dns = dns_flags.get(c["egress_ip"], False)
        clean = 1 if (not f.get("hosting") and not f.get("proxy") and not dns) else 0
        mobile = 1 if f.get("mobile") else 0
        if c["latency_ms"] and c["latency_ms"] > MAX_LATENCY_MS:
            continue  # too slow to route through
        # BANK EVERYTHING ALIVE: grade it, route clean-first. The list stays populated.
        suffix = "-T" if c.get("anon") == "transparent" else ""
        name = f"proxifly-{c['country']}-{c['egress_ip'].split('.')[-2]}{suffix}"
        con.execute("""INSERT INTO proxies(name, host, port, proto, health, latency_ms,
                       assigned_mac, last_check, clean, country, egress_ip, source)
                       VALUES(?,?,?,?,'healthy',?,NULL,?,?,?,?,?)
                       ON CONFLICT(egress_ip) DO UPDATE SET health='healthy', latency_ms=?,
                       last_check=?, clean=?""",
                    (name, c["host"], c["port"], c["proto"], c["latency_ms"], time.time(),
                     clean, c["country"], c["egress_ip"], "proxifly",
                     c["latency_ms"], time.time(), clean))
        added += 1
    con.commit()
    con.close()
    prune_pool()
    log(f"cycle done: +{added} clean proxies into the bank")
    assign_best()
    log("auto-assignment sweep complete")

def assign_best():
    """Give every online device without a proxy the best clean one available."""
    con = dbcon()
    con.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_prox_egress ON proxies(egress_ip)")
    devs = con.execute("SELECT * FROM devices WHERE adb_state='online' AND ip IS NOT NULL").fetchall()
    for d in devs:
        have = con.execute("SELECT * FROM proxies WHERE assigned_mac=?", (d["mac"],)).fetchone()
        if have:
            continue  # already assigned — keep stable
        # assignment preference: verified-clean first, then any live proxy
        best = con.execute("""SELECT * FROM proxies
                              WHERE clean=1 AND health='healthy' AND assigned_mac IS NULL
                              ORDER BY latency_ms ASC LIMIT 1""").fetchone()
        if not best:
            best = con.execute("""SELECT * FROM proxies
                                  WHERE health='healthy' AND assigned_mac IS NULL
                                  ORDER BY clean DESC, latency_ms ASC LIMIT 1""").fetchone()
        if not best:
            continue
        con.execute("UPDATE proxies SET assigned_mac=? WHERE id=?", (d["mac"], best["id"]))
        con.execute("UPDATE devices SET proxy_name=?, public_ip=? WHERE mac=?",
                    (best["name"], best["egress_ip"], d["mac"]))
        log(f"assigned {best['name']} (egress {best['egress_ip']}, clean={best['clean']}) -> {d['mac'][:14]}")
    con.commit()
    con.close()

def loop():
    while True:
        time.sleep(HARVEST_INTERVAL)
        try:
            harvest()
        except Exception as e:
            log(f"cycle error: {e}")

if __name__ == "__main__":
    log("proxyfly harvest engine starting")
    try:
        harvest()
    except Exception as e:
        log(f"startup harvest error: {e}")
    loop()