ipyou.py

#!/usr/bin/env python3
"""ipyou.py — ipyou.net IP 查询工具 / 库（零依赖，仅用标准库）

命令行用法:
    python3 ipyou.py                    # 本机 IP 详情(纯文本)
    python3 ipyou.py -q                 # 只输出 IP
    python3 ipyou.py 1.1.1.1            # 查指定 IP
    python3 ipyou.py -j 1.1.1.1         # JSON
    python3 ipyou.py -b ips.txt         # 批量质检,输出 CSV
    python3 ipyou.py -f type 1.1.1.1    # 只取某字段

作为库使用:
    from ipyou import myip, details, batch
    print(myip())                       # '1.2.3.4'
    d = details('1.1.1.1')
    print(d['usage_inferred']['label'], d['ip_score']['score'])
    for row in batch(['8.8.8.8', '1.1.1.1']):
        print(row['ip'], row['type'], row['score'])

无需注册、无需 API Key。
"""
from __future__ import annotations

import csv
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request

BASE = os.environ.get("IPYOU_BASE", "https://ipyou.net")
TIMEOUT = float(os.environ.get("IPYOU_TIMEOUT", "20"))
UA = "ipyou.py (python-urllib)"


def _get(path: str, as_json: bool = False):
    """请求站点。as_json=True 时带 Accept 头拿 JSON —— 站点只有这两个接口:
    `/`(查自己) 与 `/<IP>`(查指定 IP),格式由 Accept 头决定。"""
    headers = {"User-Agent": UA}
    if as_json:
        headers["Accept"] = "application/json"
    req = urllib.request.Request(f"{BASE}/{path}", headers=headers)
    with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
        body = r.read().decode("utf-8")
    return json.loads(body) if as_json else body


# ---------- 公开 API ----------

def myip() -> str:
    """返回本机出口 IP。"""
    return str(_get("", as_json=True).get("ip", "")).strip()


def myip_details() -> dict:
    """返回本机 IP 的完整信息(dict)。"""
    return _get("", as_json=True)


def details(ip: str) -> dict:
    """返回指定 IP 的完整信息(dict)。"""
    return _get(ip, as_json=True)


def text(ip: str = "") -> str:
    """返回纯文本信息;ip 为空则查本机。"""
    return _get(ip)


def batch(ips: list[str], sleep: float = 1.0) -> list[dict]:
    """查一批 IP。站点不提供批量接口,这里就是循环调用单个接口 ——
    自动去重并在每次之间留间隔(默认 1s),避免触发限流。"""
    seen, uniq = set(), []
    for i in ips:
        i = i.strip()
        if i and i not in seen:
            seen.add(i)
            uniq.append(i)
    out: list[dict] = []
    for n, ip in enumerate(uniq):
        if n:
            time.sleep(sleep)
        try:
            out.append(details(ip))
        except urllib.error.HTTPError as e:
            print(f"  ! {ip} 查询失败: HTTP {e.code}", file=sys.stderr)
    return out


def is_residential(ip: str) -> bool:
    """是否住宅类 IP(家庭宽带 / 移动网络)。"""
    label = (details(ip).get("usage_inferred") or {}).get("label", "")
    return label in ("家庭宽带", "移动网络")


# ---------- 命令行 ----------

IP_RE = re.compile(r"(?:\d{1,3}\.){3}\d{1,3}")

_FIELDS = {
    "ip":          lambda d: d.get("ip", ""),
    "country":     lambda d: (d.get("geo_consensus") or {}).get("country_name") or (d.get("geo") or {}).get("country", ""),
    "city":        lambda d: (d.get("geo_consensus") or {}).get("city") or (d.get("geo") or {}).get("city", ""),
    "isp":         lambda d: (d.get("geo") or {}).get("isp") or (d.get("geo_consensus") or {}).get("isp", ""),
    "asn":         lambda d: f"AS{d['asn']['number']} {d['asn'].get('org','')}".strip() if d.get("asn", {}).get("number") else "",
    "type":        lambda d: (d.get("usage_inferred") or {}).get("label", ""),
    "score":       lambda d: (d.get("ip_score") or {}).get("score", ""),
    "level":       lambda d: (d.get("ip_score") or {}).get("level", ""),
    "blocklisted": lambda d: d.get("blocklisted", False),
}


def _batch_csv(path: str) -> None:
    raw = sys.stdin.read() if path == "-" else open(path, encoding="utf-8").read()
    ips = IP_RE.findall(re.sub(r"#.*", "", raw))
    if not ips:
        sys.exit("没有找到可检测的 IP")
    print(f"共 {len(set(ips))} 个 IP,查询中…", file=sys.stderr)

    w = csv.writer(sys.stdout)
    w.writerow(["ip", "country", "city", "isp", "asn", "type", "score",
                "level", "blocklisted", "tiktok", "ecom", "social", "ai"])
    for r in batch(ips):
        s = {x.get("name"): x.get("stars") for x in (r.get("scenarios") or [])}
        geo, gc = r.get("geo") or {}, r.get("geo_consensus") or {}
        asn, usage, score = r.get("asn") or {}, r.get("usage_inferred") or {}, r.get("ip_score") or {}
        w.writerow([
            r.get("ip", ""),
            gc.get("country_name") or geo.get("country", ""),
            gc.get("city") or geo.get("city", ""),
            geo.get("isp") or gc.get("isp", ""),
            f"AS{asn['number']}" if asn.get("number") else "",
            usage.get("label", ""), score.get("score", ""), score.get("level", ""),
            r.get("blocklisted", ""),
            s.get("TikTok", ""), s.get("跨境电商", ""), s.get("社媒运营", ""), s.get("AI 应用", ""),
        ])


def main(argv: list[str]) -> int:
    args = argv[1:]
    if any(a in ("-h", "--help") for a in args):
        print(__doc__)
        return 0
    try:
        if "-b" in args:
            _batch_csv(args[args.index("-b") + 1] if len(args) > args.index("-b") + 1 else "-")
            return 0
        if "-f" in args:
            i = args.index("-f")
            field, targets = args[i + 1], args[i + 2:]
            if field not in _FIELDS:
                sys.exit(f"未知字段: {field}(可选: {', '.join(_FIELDS)})")
            for ip in targets or [myip()]:
                print(_FIELDS[field](details(ip)))
            return 0
        if "-q" in args:
            print(myip())
            return 0

        as_json = "-j" in args
        targets = [a for a in args if not a.startswith("-")]
        if not targets:
            print(json.dumps(myip_details(), ensure_ascii=False, indent=2) if as_json else text(), end="" if not as_json else "\n")
            return 0
        for ip in targets:
            print(json.dumps(details(ip), ensure_ascii=False, indent=2) if as_json else text(ip), end="" if not as_json else "\n")
        return 0
    except urllib.error.HTTPError as e:
        sys.exit(f"HTTP {e.code}: {e.reason}" + (" —— 触发限流,请稍后重试" if e.code == 429 else ""))
    except urllib.error.URLError as e:
        sys.exit(f"网络错误: {e.reason}")


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))