5f1b9db2dc
简化分流文案与手动检测交互，系统页对齐管线/芯片/ScoreRing 设计语言；README 同步更新。 Co-authored-by: Cursor <cursoragent@cursor.com>
439 lines
13 KiB
Python
439 lines
13 KiB
Python
"""
|
||
静态住宅 IP / 出口画像检测（参考 whoer、ping0.cc 维度）。
|
||
|
||
数据源：
|
||
- ip-api.com：hosting / proxy / mobile、ISP、ASN、地理
|
||
- ping0.cc/geo：中文归属、ASN、运营商（本机默认出口）
|
||
- ipify / ifconfig：公网出口 IP 采样
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
import time
|
||
from typing import Any, Optional
|
||
|
||
import httpx
|
||
|
||
IP_API_FIELDS = (
|
||
"status,message,query,country,countryCode,regionName,city,zip,isp,org,as,asname,"
|
||
"mobile,proxy,hosting,reverse"
|
||
)
|
||
IP_API_SELF = f"http://ip-api.com/json/?fields={IP_API_FIELDS}"
|
||
IP_API_IP = f"http://ip-api.com/json/{{ip}}?fields={IP_API_FIELDS}"
|
||
|
||
PING0_GEO_URL = "https://ping0.cc/geo"
|
||
PING0_IPV4_GEO = "https://ipv4.ping0.cc/geo"
|
||
|
||
IPIFY_URLS = (
|
||
"https://api.ipify.org?format=json",
|
||
"https://api64.ipify.org?format=json",
|
||
)
|
||
IFCONFIG_URL = "https://ifconfig.me/ip"
|
||
|
||
IDC_KEYWORDS = (
|
||
"hosting",
|
||
"datacenter",
|
||
"data center",
|
||
"cloud",
|
||
"colo",
|
||
"vps",
|
||
"server",
|
||
"amazon",
|
||
"google cloud",
|
||
"azure",
|
||
"digitalocean",
|
||
"linode",
|
||
"vultr",
|
||
"ovh",
|
||
"hetzner",
|
||
"leaseweb",
|
||
"psychz",
|
||
"choopa",
|
||
"contabo",
|
||
"spartan",
|
||
"colocrossing",
|
||
)
|
||
|
||
PROXY_KEYWORDS = (
|
||
"vpn",
|
||
"proxy",
|
||
"tor",
|
||
"mullvad",
|
||
"nordvpn",
|
||
"expressvpn",
|
||
"surfshark",
|
||
"datacamp",
|
||
"bright data",
|
||
"luminati",
|
||
)
|
||
|
||
RESIDENTIAL_KEYWORDS = (
|
||
"telecom",
|
||
"broadband",
|
||
"cable",
|
||
"fiber",
|
||
"fibre",
|
||
"communications",
|
||
"chinanet",
|
||
"unicom",
|
||
"cmcc",
|
||
"telecom",
|
||
"comcast",
|
||
"verizon",
|
||
"att",
|
||
"residential",
|
||
"home",
|
||
"宽带",
|
||
"电信",
|
||
"联通",
|
||
"移动",
|
||
)
|
||
|
||
CONNECT_TIMEOUT = 8.0
|
||
READ_TIMEOUT = 12.0
|
||
|
||
UA = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
||
)
|
||
|
||
|
||
def _norm(s: Optional[str]) -> str:
|
||
return (s or "").strip().lower()
|
||
|
||
|
||
def _classify_ip_type(data: dict[str, Any]) -> tuple[str, str]:
|
||
"""返回 (type_id, label_zh)。"""
|
||
if data.get("proxy"):
|
||
return "proxy", "代理 / VPN"
|
||
if data.get("hosting"):
|
||
return "datacenter", "IDC 机房"
|
||
if data.get("mobile"):
|
||
return "mobile", "移动网络"
|
||
|
||
blob = " ".join(
|
||
_norm(data.get(k)) for k in ("isp", "org", "as", "asname", "reverse") if data.get(k)
|
||
)
|
||
if any(k in blob for k in IDC_KEYWORDS):
|
||
return "datacenter", "IDC 机房"
|
||
if any(k in blob for k in PROXY_KEYWORDS):
|
||
return "proxy", "代理 / VPN"
|
||
if any(k in blob for k in RESIDENTIAL_KEYWORDS):
|
||
return "residential", "家庭宽带"
|
||
if data.get("isp"):
|
||
return "residential", "疑似 ISP 宽带"
|
||
return "unknown", "未知类型"
|
||
|
||
|
||
def _quality_score(data: dict[str, Any], ip_type: str) -> int:
|
||
score = 100
|
||
if data.get("hosting") or ip_type == "datacenter":
|
||
score -= 55
|
||
if data.get("proxy") or ip_type == "proxy":
|
||
score -= 40
|
||
if data.get("mobile") or ip_type == "mobile":
|
||
score -= 15
|
||
blob = _norm(data.get("org")) + " " + _norm(data.get("isp"))
|
||
if any(k in blob for k in PROXY_KEYWORDS):
|
||
score -= 25
|
||
if ip_type == "residential":
|
||
score += 5
|
||
return max(0, min(100, score))
|
||
|
||
|
||
def _quality_label(score: int) -> str:
|
||
if score >= 85:
|
||
return "优质（偏住宅）"
|
||
if score >= 70:
|
||
return "较干净"
|
||
if score >= 45:
|
||
return "一般"
|
||
return "高风险 / 机房"
|
||
|
||
|
||
def _guess_native(data: dict[str, Any], ip_type: str) -> tuple[Optional[bool], str]:
|
||
if ip_type in ("datacenter", "proxy"):
|
||
return False, "广播 / 非原生"
|
||
cc = (data.get("countryCode") or "").upper()
|
||
asname = _norm(data.get("asname"))
|
||
org = _norm(data.get("org"))
|
||
if not cc:
|
||
return None, "无法判断"
|
||
if cc == "CN" and any(x in org + asname for x in ("chinanet", "unicom", "cmcc", "telecom", "电信", "联通", "移动")):
|
||
return True, "原生 IP（ISP）"
|
||
if ip_type == "residential":
|
||
return True, "疑似原生"
|
||
if data.get("hosting"):
|
||
return False, "广播 IP"
|
||
return None, "待确认"
|
||
|
||
|
||
def _parse_ping0_geo(text: str) -> dict[str, Any]:
|
||
lines = [ln.strip() for ln in (text or "").splitlines() if ln.strip()]
|
||
if len(lines) < 1:
|
||
return {"error": "ping0 返回为空"}
|
||
out: dict[str, Any] = {"ip": lines[0]}
|
||
if len(lines) > 1:
|
||
out["location"] = lines[1]
|
||
if len(lines) > 2:
|
||
out["asn"] = lines[2]
|
||
if len(lines) > 3:
|
||
out["org"] = lines[3]
|
||
return out
|
||
|
||
|
||
async def _fetch_ping0_geo(client: httpx.AsyncClient, ipv4: bool = True) -> dict[str, Any]:
|
||
url = PING0_IPV4_GEO if ipv4 else PING0_GEO_URL
|
||
try:
|
||
r = await client.get(url, timeout=8.0)
|
||
r.raise_for_status()
|
||
return _parse_ping0_geo(r.text)
|
||
except Exception as e:
|
||
return {"error": str(e)}
|
||
|
||
|
||
async def _ip_api_lookup(client: httpx.AsyncClient, ip: Optional[str] = None) -> dict[str, Any]:
|
||
url = IP_API_IP.format(ip=ip) if ip else IP_API_SELF
|
||
try:
|
||
r = await client.get(url, timeout=8.0)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
except Exception as e:
|
||
return {"error": str(e), "query": ip}
|
||
if data.get("status") != "success":
|
||
return {
|
||
"error": data.get("message") or "lookup failed",
|
||
"query": ip or data.get("query"),
|
||
}
|
||
ip_type, ip_type_label = _classify_ip_type(data)
|
||
score = _quality_score(data, ip_type)
|
||
native, native_label = _guess_native(data, ip_type)
|
||
return {
|
||
"ip": data.get("query"),
|
||
"country": data.get("country"),
|
||
"country_code": data.get("countryCode"),
|
||
"region": data.get("regionName"),
|
||
"city": data.get("city"),
|
||
"zip": data.get("zip"),
|
||
"isp": data.get("isp"),
|
||
"org": data.get("org"),
|
||
"asn": data.get("as"),
|
||
"asname": data.get("asname"),
|
||
"reverse": data.get("reverse"),
|
||
"mobile": bool(data.get("mobile")),
|
||
"proxy": bool(data.get("proxy")),
|
||
"hosting": bool(data.get("hosting")),
|
||
"ip_type": ip_type,
|
||
"ip_type_label": ip_type_label,
|
||
"native": native,
|
||
"native_label": native_label,
|
||
"quality_score": score,
|
||
"quality_label": _quality_label(score),
|
||
"error": None,
|
||
}
|
||
|
||
|
||
async def _fetch_public_ip(client: httpx.AsyncClient) -> tuple[Optional[str], Optional[str]]:
|
||
for url in IPIFY_URLS:
|
||
try:
|
||
r = await client.get(url, timeout=8.0)
|
||
r.raise_for_status()
|
||
data = r.json()
|
||
ip = (data.get("ip") or "").strip()
|
||
if ip:
|
||
return ip, None
|
||
except Exception:
|
||
continue
|
||
try:
|
||
r = await client.get(IFCONFIG_URL, timeout=8.0)
|
||
r.raise_for_status()
|
||
ip = r.text.strip()
|
||
if re.match(r"^\d{1,3}(\.\d{1,3}){3}$", ip) or ":" in ip:
|
||
return ip, None
|
||
except Exception as e:
|
||
return None, str(e)
|
||
return None, "无法获取公网 IP"
|
||
|
||
|
||
def _build_exit_profile(
|
||
label: str,
|
||
ip_api: dict[str, Any],
|
||
ping0: Optional[dict[str, Any]] = None,
|
||
) -> dict[str, Any]:
|
||
out = {"label": label, **ip_api}
|
||
if ping0 and not ping0.get("error"):
|
||
out["ping0"] = ping0
|
||
return out
|
||
|
||
|
||
def _tz_country_hint(tz: str) -> Optional[str]:
|
||
"""粗略：常见时区 → 国家码（仅用于 whoer 式一致性参考）。"""
|
||
tz = (tz or "").strip()
|
||
mapping = {
|
||
"Asia/Shanghai": "CN",
|
||
"Asia/Chongqing": "CN",
|
||
"Asia/Hong_Kong": "HK",
|
||
"Asia/Taipei": "TW",
|
||
"America/New_York": "US",
|
||
"America/Los_Angeles": "US",
|
||
"Europe/London": "GB",
|
||
"Europe/Berlin": "DE",
|
||
"Asia/Tokyo": "JP",
|
||
}
|
||
return mapping.get(tz)
|
||
|
||
|
||
def _build_checks(
|
||
default_exit: dict[str, Any],
|
||
overseas_exit: dict[str, Any],
|
||
*,
|
||
client_tz: str = "",
|
||
client_lang: str = "",
|
||
webrtc_leak: Optional[bool] = None,
|
||
) -> list[dict[str, Any]]:
|
||
checks: list[dict[str, Any]] = []
|
||
|
||
def add(cid: str, label: str, ok: bool, hint: str = "") -> None:
|
||
checks.append({"id": cid, "label": label, "ok": ok, "hint": hint})
|
||
|
||
d = default_exit
|
||
o = overseas_exit
|
||
|
||
add(
|
||
"not_datacenter",
|
||
"默认出口非机房 IP",
|
||
d.get("ip_type") not in ("datacenter", "proxy") and not d.get("hosting"),
|
||
d.get("ip_type_label") or "",
|
||
)
|
||
add(
|
||
"residential_like",
|
||
"默认出口偏住宅 / ISP",
|
||
d.get("ip_type") == "residential",
|
||
"静态住宅推流更稳" if d.get("ip_type") != "residential" else "",
|
||
)
|
||
add("not_proxy", "默认出口非代理标记", not d.get("proxy"), "ip-api 检测到 proxy=true" if d.get("proxy") else "")
|
||
add(
|
||
"native_ip",
|
||
"原生 IP（参考）",
|
||
d.get("native") is True,
|
||
d.get("native_label") or "",
|
||
)
|
||
add(
|
||
"quality_ok",
|
||
"IP 洁净度 ≥ 70",
|
||
(d.get("quality_score") or 0) >= 70,
|
||
f"评分 {d.get('quality_score', '—')} · {d.get('quality_label', '')}",
|
||
)
|
||
|
||
d_ip = d.get("ip")
|
||
o_ip = o.get("ip")
|
||
if d_ip and o_ip:
|
||
split = d_ip != o_ip
|
||
add(
|
||
"split_exit",
|
||
"分流",
|
||
split,
|
||
"两路 IP 不同" if split else "两路 IP 相同",
|
||
)
|
||
|
||
if client_tz and d.get("country_code"):
|
||
hint_cc = _tz_country_hint(client_tz)
|
||
if hint_cc:
|
||
add(
|
||
"tz_match",
|
||
"系统时区与 IP 国家一致（参考）",
|
||
hint_cc == (d.get("country_code") or "").upper(),
|
||
f"时区 {client_tz}",
|
||
)
|
||
|
||
if client_lang and d.get("country_code"):
|
||
lang = client_lang.lower()
|
||
cc = (d.get("country_code") or "").lower()
|
||
if cc == "cn":
|
||
lang_ok = lang.startswith("zh")
|
||
elif cc == "us":
|
||
lang_ok = lang.startswith("en")
|
||
else:
|
||
# 非中美地区不做严格语言判定，避免误报。
|
||
lang_ok = True
|
||
add("lang_soft", "浏览器语言与 IP 地区不冲突", lang_ok, client_lang)
|
||
|
||
if webrtc_leak is not None:
|
||
add(
|
||
"webrtc",
|
||
"WebRTC 未泄露本地网卡 IP",
|
||
not webrtc_leak,
|
||
"检测到额外本地候选地址" if webrtc_leak else "",
|
||
)
|
||
|
||
return checks
|
||
|
||
|
||
async def run_ip_profile(
|
||
*,
|
||
client_tz: str = "",
|
||
client_lang: str = "",
|
||
webrtc_leak: Optional[bool] = None,
|
||
) -> dict[str, Any]:
|
||
timeout = httpx.Timeout(READ_TIMEOUT, connect=CONNECT_TIMEOUT)
|
||
limits = httpx.Limits(max_keepalive_connections=6, max_connections=10)
|
||
|
||
async with httpx.AsyncClient(
|
||
timeout=timeout,
|
||
limits=limits,
|
||
verify=True,
|
||
headers={"User-Agent": UA},
|
||
) as client:
|
||
ping0_task = asyncio.create_task(_fetch_ping0_geo(client))
|
||
default_api_task = asyncio.create_task(_ip_api_lookup(client, None))
|
||
overseas_ip_task = asyncio.create_task(_fetch_public_ip(client))
|
||
|
||
ping0, default_api, (overseas_ip, overseas_ip_err) = await asyncio.gather(
|
||
ping0_task, default_api_task, overseas_ip_task
|
||
)
|
||
|
||
if default_api.get("error") and ping0.get("ip") and not ping0.get("error"):
|
||
default_api = await _ip_api_lookup(client, ping0["ip"])
|
||
|
||
overseas_api: dict[str, Any] = {"error": overseas_ip_err or "unknown"}
|
||
if overseas_ip:
|
||
overseas_api = await _ip_api_lookup(client, overseas_ip)
|
||
|
||
default_exit = _build_exit_profile("当前出口", default_api, ping0)
|
||
overseas_exit = _build_exit_profile("海外线路", overseas_api)
|
||
|
||
checks = _build_checks(
|
||
default_exit,
|
||
overseas_exit,
|
||
client_tz=client_tz,
|
||
client_lang=client_lang,
|
||
webrtc_leak=webrtc_leak,
|
||
)
|
||
|
||
residential_ok = default_exit.get("ip_type") == "residential" and (default_exit.get("quality_score") or 0) >= 70
|
||
|
||
return {
|
||
"default_exit": default_exit,
|
||
"overseas_exit": overseas_exit,
|
||
"checks": checks,
|
||
"summary": {
|
||
"residential_like": residential_ok,
|
||
"headline": (
|
||
"当前出口偏静态住宅"
|
||
if residential_ok
|
||
else (
|
||
"出口疑似机房/代理，不适合伪装住宅"
|
||
if default_exit.get("ip_type") in ("datacenter", "proxy")
|
||
else "出口类型一般，建议结合代理规则确认"
|
||
)
|
||
),
|
||
},
|
||
"meta": {
|
||
"at": time.time(),
|
||
"sources": ["ip-api.com", "ping0.cc/geo", "ipify"],
|
||
"disclaimer": "结果仅供参考；ping0 完整 IDC/原生判定需其付费 API。",
|
||
},
|
||
}
|