#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
proxycheck —— 代理可用性与稳定性检测工具

只用 Python 标准库,无需安装依赖。Python 3.8+。

用法:
  # 基础连通性 + 出口信息
  proxycheck test --proxy http://user:pass@host:1080

  # 稳定性抽样:连续请求,统计成功率与延迟
  proxycheck load --proxy ... --count 50 --concurrency 5

  # 会话保持检测:间隔取样,看出口地址是否真的固定
  proxycheck session --proxy ... --count 10 --interval 60

  # 公平比较两个出口:同一目标、同一参数、同样指标
  proxycheck compare --a ... --b ... --count 30

  # 不做任何代理,直接检测本机出口(用于对照)
  proxycheck test
"""
import argparse
import json
import statistics
import sys
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

DEFAULT_TARGET = "https://api.ipify.org?format=json"
DEFAULT_IPINFO = "https://ipinfo.io/{ip}/json"
TIMEOUT = 20


# ---------- 代理配置 ----------

def build_opener(proxy):
    """根据代理地址构造 opener。支持 http/https/socks5 前缀。"""
    handlers = []
    if proxy:
        if proxy.startswith("socks"):
            try:
                import socks  # type: ignore
                from sockshandler import SocksiPyHandler  # type: ignore
                handlers.append(SocksiPyHandler(socks.SOCKS5, *parse_socks(proxy)))
            except ImportError:
                die("SOCKS 代理需要额外安装: pip install PySocks")
        else:
            handlers.append(urllib.request.ProxyHandler({"http": proxy, "https": proxy}))
    return urllib.request.build_opener(*handlers)


def parse_socks(proxy: str):
    body = proxy.split("://", 1)[1]
    auth, hostport = (body.rsplit("@", 1) if "@" in body else (None, body))
    host, port = hostport.rsplit(":", 1)
    if auth:
        user, pw = auth.split(":", 1)
        return host, int(port), True, user, pw
    return host, int(port)


def die(msg: str, code: int = 1):
    print(f"错误: {msg}", file=sys.stderr)
    sys.exit(code)


# ---------- 单次请求 ----------

def fetch(url: str, opener, timeout: int = TIMEOUT):
    """返回 (成功?, 内容, 耗时毫秒, 错误信息)"""
    t0 = time.perf_counter()
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "proxycheck/1.0"})
        with opener.open(req, timeout=timeout) as r:
            body = r.read().decode("utf-8", "replace")
        return True, body, (time.perf_counter() - t0) * 1000, None
    except urllib.error.HTTPError as e:
        return False, "", (time.perf_counter() - t0) * 1000, f"HTTP {e.code}"
    except Exception as e:
        return False, "", (time.perf_counter() - t0) * 1000, type(e).__name__


def exit_ip(opener, timeout: int = TIMEOUT):
    ok, body, ms, err = fetch(DEFAULT_TARGET, opener, timeout)
    if not ok:
        return None, ms, err
    try:
        return json.loads(body).get("ip"), ms, None
    except Exception:
        return body.strip(), ms, None


def ip_info(ip: str, timeout: int = TIMEOUT):
    """查询 IP 归属。失败返回 None,不视为错误。"""
    try:
        op = urllib.request.build_opener()  # 直连查询,不走代理
        with op.open(DEFAULT_IPINFO.format(ip=ip), timeout=timeout) as r:
            d = json.loads(r.read().decode("utf-8", "replace"))
        return {k: d.get(k) for k in ("country", "region", "city", "org", "timezone")}
    except Exception:
        return None


# ---------- 子命令 ----------

def cmd_test(args):
    opener = build_opener(args.proxy)
    print(f"代理: {args.proxy or '(未使用,检测本机出口)'}")
    print(f"目标: {DEFAULT_TARGET}")
    print()

    ip, ms, err = exit_ip(opener)
    if err:
        print(f"❌ 失败: {err}  (耗时 {ms:.0f}ms)")
        print()
        print("排查顺序建议:")
        print("  1. 去掉 --proxy 再跑一次,确认本机网络本身是通的")
        print("  2. 核对地址格式:http://用户:密码@主机:端口")
        print("  3. 换一个目标站点试,区分「代理不通」和「目标站点拒绝」")
        return 1

    print(f"✅ 连通  出口 IP: {ip}   耗时 {ms:.0f}ms")
    info = ip_info(ip)
    if info:
        loc = " / ".join(x for x in (info.get("country"), info.get("region"), info.get("city")) if x)
        print(f"   归属: {loc or '未知'}")
        if info.get("org"):
            print(f"   运营方: {info['org']}")
        if info.get("timezone"):
            print(f"   时区: {info['timezone']}")
    else:
        print("   归属: 查询失败(不影响连通性判断)")
    return 0


def _worker(opener, url, timeout):
    return fetch(url, opener, timeout)


def cmd_load(args):
    opener = build_opener(args.proxy)
    url = args.url or DEFAULT_TARGET
    print(f"代理: {args.proxy or '(本机)'}")
    print(f"目标: {url}")
    print(f"请求: {args.count} 次,并发 {args.concurrency}")
    print()

    results = []
    t0 = time.perf_counter()
    with ThreadPoolExecutor(max_workers=args.concurrency) as ex:
        futs = [ex.submit(_worker, opener, url, args.timeout) for _ in range(args.count)]
        for i, f in enumerate(as_completed(futs), 1):
            results.append(f.result())
            done = int(i / args.count * 20)
            print(f"\r  [{'█'*done}{'·'*(20-done)}] {i}/{args.count}", end="", flush=True)
    total = time.perf_counter() - t0
    print("\n")

    ok = [r for r in results if r[0]]
    fail = [r for r in results if not r[0]]
    lat = sorted(r[2] for r in ok)

    print(f"  成功率:   {len(ok)}/{len(results)}  ({len(ok)/len(results):.1%})")
    print(f"  总耗时:   {total:.1f}s")
    if lat:
        print(f"  延迟(ms): 最小 {lat[0]:.0f} / 中位 {statistics.median(lat):.0f} / "
              f"P95 {lat[int(len(lat)*0.95)-1]:.0f} / 最大 {lat[-1]:.0f}")
    if fail:
        from collections import Counter
        print(f"  失败原因: {dict(Counter(r[3] for r in fail))}")
    print()
    if len(ok) / len(results) < 0.9:
        print("  ⚠️ 成功率偏低。建议先加对照请求(不用代理跑同一目标),确认问题在代理侧还是目标站点侧。")
    return 0


def cmd_session(args):
    """间隔取样退出地址,判断「静态」是否真的静态。"""
    opener = build_opener(args.proxy)
    print(f"会话保持检测:取样 {args.count} 次,间隔 {args.interval}s")
    print(f"预计耗时约 {(args.count-1)*args.interval/60:.1f} 分钟")
    print()

    seen = []
    for i in range(args.count):
        ip, ms, err = exit_ip(opener)
        ts = time.strftime("%H:%M:%S")
        if err:
            print(f"  [{i+1:>2}] {ts}  ❌ {err}")
            seen.append(None)
        else:
            changed = "" if not seen or seen[-1] in (None, ip) else "   ← 地址已变化"
            print(f"  [{i+1:>2}] {ts}  {ip}  ({ms:.0f}ms){changed}")
            seen.append(ip)
        if i < args.count - 1:
            time.sleep(args.interval)

    valid = [s for s in seen if s]
    uniq = sorted(set(valid))
    print()
    print(f"  取样成功: {len(valid)}/{args.count}")
    print(f"  不同出口: {len(uniq)} 个")
    if len(uniq) == 1:
        print(f"  ✅ 出口保持稳定: {uniq[0]}")
    elif len(uniq) == 0:
        print("  ❌ 全部失败,无法判断")
    else:
        print(f"  ⚠️ 出口发生了变化: {', '.join(uniq[:6])}{' …' if len(uniq) > 6 else ''}")
        print("     若你购买的是「静态」方案,请用这份记录与商家核对保持期限的具体约定。")
    return 0


def cmd_compare(args):
    """同一目标、同一参数、同样指标 —— 否则比较没有意义。"""
    url = args.url or DEFAULT_TARGET
    print("公平比较:同一目标、同一参数、同一时间段")
    print(f"目标: {url}   请求数: {args.count}/侧   并发: {args.concurrency}")
    print()

    rows = []
    for label, proxy in (("A", args.a), ("B", args.b)):
        opener = build_opener(proxy)
        t0 = time.perf_counter()
        with ThreadPoolExecutor(max_workers=args.concurrency) as ex:
            futs = [ex.submit(_worker, opener, url, args.timeout) for _ in range(args.count)]
            res = [f.result() for f in as_completed(futs)]
        elapsed = time.perf_counter() - t0
        ok = [r for r in res if r[0]]
        lat = sorted(r[2] for r in ok)
        ip, _, _ = exit_ip(opener)
        rows.append({
            "侧": label,
            "代理": proxy or "(本机)",
            "出口": ip or "-",
            "成功率": f"{len(ok)/len(res):.1%}",
            "中位延迟": f"{statistics.median(lat):.0f}ms" if lat else "-",
            "P95": f"{lat[int(len(lat)*0.95)-1]:.0f}ms" if lat else "-",
            "总耗时": f"{elapsed:.1f}s",
        })
        print(f"  {label} 完成: 成功率 {rows[-1]['成功率']}  出口 {rows[-1]['出口']}")

    print()
    w = [max(len(str(r[k])) for r in rows + [{"侧":"侧","代理":"代理","出口":"出口","成功率":"成功率","中位延迟":"中位延迟","P95":"P95","总耗时":"总耗时"}]) for k in rows[0]]
    keys = list(rows[0].keys())
    print("  " + "  ".join(k.ljust(w[i]) for i, k in enumerate(keys)))
    print("  " + "  ".join("-"*w[i] for i in range(len(keys))))
    for r in rows:
        print("  " + "  ".join(str(r[k]).ljust(w[i]) for i, k in enumerate(keys)))
    print()
    print("  提示:成功率接近时,再看延迟与成本结构;单看单价容易得出错误结论。")
    return 0


# ---------- 入口 ----------

def main():
    p = argparse.ArgumentParser(
        prog="proxycheck",
        description="代理可用性与稳定性检测工具(仅用标准库)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__)
    sub = p.add_subparsers(dest="cmd", required=True)

    def common(sp, need_proxy=True):
        # --proxy 全部可选:不带即为直连,用作对照基线
        sp.add_argument("--proxy", required=False, default=None,
                        help="代理地址,如 http://user:pass@host:1080;不填表示直连(对照用)")
        sp.add_argument("--timeout", type=int, default=TIMEOUT, help="单次请求超时秒数")
        sp.add_argument("--url", default=None, help="自定义检测目标(默认查询出口 IP)")

    t = sub.add_parser("test", help="连通性检测 + 出口归属")
    common(t, need_proxy=False)
    t.set_defaults(func=cmd_test)

    l = sub.add_parser("load", help="并发压力抽样,统计成功率与延迟")
    common(l)
    l.add_argument("--count", type=int, default=50)
    l.add_argument("--concurrency", type=int, default=5)
    l.set_defaults(func=cmd_load)

    s = sub.add_parser("session", help="会话保持检测:出口地址是否真的固定")
    common(s)
    s.add_argument("--count", type=int, default=10)
    s.add_argument("--interval", type=int, default=60, help="取样间隔秒数")
    s.set_defaults(func=cmd_session)

    c = sub.add_parser("compare", help="公平比较两个出口")
    c.add_argument("--a", default=None, help="A 侧代理(留空表示直连)")
    c.add_argument("--b", required=True, help="B 侧代理")
    c.add_argument("--count", type=int, default=30)
    c.add_argument("--concurrency", type=int, default=3)
    c.add_argument("--timeout", type=int, default=TIMEOUT)
    c.add_argument("--url", default=None)
    c.set_defaults(func=cmd_compare)

    args = p.parse_args()
    sys.exit(args.func(args))


if __name__ == "__main__":
    main()