DESKTOP-77LRNCU\dsx2016
ae9531f2c2
新增 /ip_profile、/ip_quality API 与网络 UI 组件，默认隐藏摄像头/拉流导航；同步更新 README。 Co-authored-by: Cursor <cursoragent@cursor.com>
415 lines
16 KiB
Python
415 lines
16 KiB
Python
"""
|
||
IP 质量检测（集成 xykt/IPQuality 思路与数据源）。
|
||
|
||
脚本来源：third_party/IPQuality/ip.sh (v2026-03-29)
|
||
可用数据源：ipinfo.check.place、ipinfo.io、api.ipapi.is、流媒体解锁探测
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
import time
|
||
from typing import Any, Optional
|
||
|
||
import httpx
|
||
|
||
SCRIPT_VERSION = "v2026-03-29"
|
||
UA = (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
|
||
)
|
||
CONNECT_TIMEOUT = 8.0
|
||
READ_TIMEOUT = 14.0
|
||
|
||
TYPE_LABELS = {
|
||
"business": "商业",
|
||
"isp": "运营商",
|
||
"hosting": "机房",
|
||
"education": "教育",
|
||
"government": "政府",
|
||
"banking": "金融",
|
||
"other": "其他",
|
||
}
|
||
|
||
RISK_LABELS = {
|
||
"verylow": "极低",
|
||
"low": "低",
|
||
"elevated": "偏高",
|
||
"high": "高",
|
||
"veryhigh": "极高",
|
||
"suspicious": "可疑",
|
||
"risky": "风险",
|
||
"highrisk": "高危",
|
||
}
|
||
|
||
|
||
def _type_label(raw: Optional[str]) -> str:
|
||
if not raw:
|
||
return "未知"
|
||
return TYPE_LABELS.get(str(raw).lower(), str(raw))
|
||
|
||
|
||
def _parse_abuser_score(text: Optional[str]) -> tuple[Optional[float], Optional[str]]:
|
||
if not text:
|
||
return None, None
|
||
m = re.match(r"([\d.]+)\s*\(([^)]+)\)", str(text).strip())
|
||
if not m:
|
||
return None, None
|
||
try:
|
||
num = float(m.group(1))
|
||
except ValueError:
|
||
num = None
|
||
risk = m.group(2).strip().lower().replace(" ", "")
|
||
risk_key = {
|
||
"verylow": "verylow",
|
||
"low": "low",
|
||
"elevated": "elevated",
|
||
"high": "high",
|
||
"veryhigh": "veryhigh",
|
||
}.get(risk, risk)
|
||
return num, risk_key
|
||
|
||
|
||
def _ipqs_risk(score: Optional[int]) -> Optional[str]:
|
||
if score is None:
|
||
return None
|
||
if score < 75:
|
||
return "low"
|
||
if score < 85:
|
||
return "suspicious"
|
||
if score < 90:
|
||
return "risky"
|
||
return "highrisk"
|
||
|
||
|
||
async def _fetch_json(
|
||
client: httpx.AsyncClient, url: str, *, label: str = ""
|
||
) -> tuple[Optional[dict[str, Any]], Optional[str]]:
|
||
try:
|
||
r = await client.get(url)
|
||
if r.status_code != 200:
|
||
return None, f"{label or url}: HTTP {r.status_code}"
|
||
data = r.json()
|
||
if not isinstance(data, dict):
|
||
return None, f"{label or url}: 非 JSON 对象"
|
||
return data, None
|
||
except Exception as e:
|
||
return None, f"{label or url}: {e}"
|
||
|
||
|
||
async def _fetch_text(client: httpx.AsyncClient, url: str, **kwargs: Any) -> tuple[str, Optional[str]]:
|
||
try:
|
||
r = await client.get(url, **kwargs)
|
||
if r.status_code != 200:
|
||
return "", f"HTTP {r.status_code}"
|
||
return r.text, None
|
||
except Exception as e:
|
||
return "", str(e)
|
||
|
||
|
||
async def _resolve_public_ip(client: httpx.AsyncClient) -> tuple[Optional[str], Optional[str]]:
|
||
for url in ("https://api.ipify.org?format=json", "https://api64.ipify.org?format=json"):
|
||
data, err = await _fetch_json(client, url, label="ipify")
|
||
if data and data.get("ip"):
|
||
return str(data["ip"]), None
|
||
if err:
|
||
last = err
|
||
return None, last if "last" in dir() else "无法获取公网 IP"
|
||
|
||
|
||
async def _fetch_maxmind(client: httpx.AsyncClient, ip: str) -> dict[str, Any]:
|
||
data, err = await _fetch_json(client, f"https://ipinfo.check.place/{ip}?lang=cn", label="maxmind")
|
||
if not data:
|
||
return {"error": err, "lite": True}
|
||
return {
|
||
"asn": data.get("ASN", {}).get("AutonomousSystemNumber"),
|
||
"org": data.get("ASN", {}).get("AutonomousSystemOrganization"),
|
||
"city": data.get("City", {}).get("Name"),
|
||
"postal": data.get("City", {}).get("PostalCode"),
|
||
"lat": data.get("City", {}).get("Latitude"),
|
||
"lon": data.get("City", {}).get("Longitude"),
|
||
"timezone": (data.get("City", {}).get("Location") or {}).get("TimeZone"),
|
||
"country_code": data.get("Country", {}).get("IsoCode"),
|
||
"country": data.get("Country", {}).get("Name"),
|
||
"reg_country_code": (data.get("Country", {}).get("RegisteredCountry") or {}).get("IsoCode"),
|
||
"reg_country": (data.get("Country", {}).get("RegisteredCountry") or {}).get("Name"),
|
||
"continent": (data.get("City", {}).get("Continent") or {}).get("Name"),
|
||
"lite": False,
|
||
}
|
||
|
||
|
||
async def _fetch_ipinfo_widget(client: httpx.AsyncClient, ip: str) -> dict[str, Any]:
|
||
data, err = await _fetch_json(client, f"https://ipinfo.io/widget/demo/{ip}", label="ipinfo")
|
||
if not data:
|
||
return {"error": err}
|
||
inner = data.get("data") if isinstance(data.get("data"), dict) else {}
|
||
privacy = inner.get("privacy") or {}
|
||
asn = inner.get("asn") or {}
|
||
loc = str(inner.get("loc") or "")
|
||
lat, lon = (loc.split(",", 1) + [None, None])[:2]
|
||
return {
|
||
"usage_type": asn.get("type"),
|
||
"company_type": (inner.get("company") or {}).get("type"),
|
||
"country_code": inner.get("country"),
|
||
"city": inner.get("city"),
|
||
"timezone": inner.get("timezone"),
|
||
"asn": str(asn.get("asn", "")).replace("AS", ""),
|
||
"org": asn.get("name"),
|
||
"lat": lat,
|
||
"lon": lon,
|
||
"proxy": privacy.get("proxy"),
|
||
"tor": privacy.get("tor"),
|
||
"vpn": privacy.get("vpn"),
|
||
"hosting": privacy.get("hosting"),
|
||
}
|
||
|
||
|
||
async def _fetch_ipapi(client: httpx.AsyncClient, ip: str) -> dict[str, Any]:
|
||
data, err = await _fetch_json(client, f"https://api.ipapi.is/?q={ip}", label="ipapi")
|
||
if not data:
|
||
return {"error": err}
|
||
score_num, risk_key = _parse_abuser_score((data.get("company") or {}).get("abuser_score"))
|
||
loc = data.get("location") or {}
|
||
return {
|
||
"usage_type": (data.get("asn") or {}).get("type"),
|
||
"company_type": (data.get("company") or {}).get("type"),
|
||
"country_code": loc.get("country_code"),
|
||
"city": loc.get("city"),
|
||
"country": loc.get("country"),
|
||
"timezone": loc.get("timezone"),
|
||
"asn": (data.get("asn") or {}).get("asn"),
|
||
"org": (data.get("asn") or {}).get("org"),
|
||
"lat": loc.get("latitude"),
|
||
"lon": loc.get("longitude"),
|
||
"proxy": data.get("is_proxy"),
|
||
"tor": data.get("is_tor"),
|
||
"vpn": data.get("is_vpn"),
|
||
"hosting": data.get("is_datacenter"),
|
||
"abuser": data.get("is_abuser"),
|
||
"crawler": data.get("is_crawler"),
|
||
"score_num": score_num,
|
||
"score_pct": round(score_num * 100, 2) if score_num is not None else None,
|
||
"risk_key": risk_key,
|
||
}
|
||
|
||
|
||
async def _fetch_check_place_db(client: httpx.AsyncClient, ip: str, db: str) -> dict[str, Any]:
|
||
data, err = await _fetch_json(client, f"https://ipinfo.check.place/{ip}?db={db}", label=db)
|
||
if not data:
|
||
return {"error": err}
|
||
return data
|
||
|
||
|
||
async def _test_youtube(client: httpx.AsyncClient) -> dict[str, Any]:
|
||
cookies = (
|
||
"YSC=BiCUU3-5Gdk; CONSENT=YES+cb.20220301-11-p0.en+FX+700; "
|
||
"GPS=1; VISITOR_INFO1_LIVE=4VwPMkB7W5A; PREF=tz=Asia.Shanghai"
|
||
)
|
||
text, err = await _fetch_text(
|
||
client,
|
||
"https://www.youtube.com/premium",
|
||
headers={"Accept-Language": "en", "Cookie": cookies},
|
||
)
|
||
if err:
|
||
return {"status": "error", "status_label": "检测失败", "region": None, "error": err}
|
||
if "www.google.cn" in text:
|
||
return {"status": "cn", "status_label": "中国版", "region": "CN"}
|
||
if "Premium is not available in your country" in text:
|
||
return {"status": "no_premium", "status_label": "无 Premium", "region": None}
|
||
region_m = re.search(r'"contentRegion":"([^"]+)"', text)
|
||
region = region_m.group(1) if region_m else None
|
||
if "ad-free" in text:
|
||
return {"status": "yes", "status_label": "Premium 可用", "region": region}
|
||
return {"status": "bad", "status_label": "状态未知", "region": region}
|
||
|
||
|
||
async def _test_tiktok(client: httpx.AsyncClient) -> dict[str, Any]:
|
||
text, err = await _fetch_text(client, "https://www.tiktok.com/", headers={"Accept-Language": "en"})
|
||
if err:
|
||
return {"status": "error", "status_label": "检测失败", "region": None, "error": err}
|
||
if "Please wait" in text:
|
||
return {"status": "blocked", "status_label": "需验证/封锁", "region": None}
|
||
region_m = re.search(r'"region":"([A-Z]{2})"', text) or re.search(r'"countryCode":"([A-Z]{2})"', text)
|
||
region = region_m.group(1) if region_m else None
|
||
if region:
|
||
return {"status": "yes", "status_label": "可访问", "region": region}
|
||
return {"status": "maybe", "status_label": "页面可达", "region": region}
|
||
|
||
|
||
def _bool_factor(value: Any) -> Optional[bool]:
|
||
if value is True or value == "true":
|
||
return True
|
||
if value is False or value == "false":
|
||
return False
|
||
return None
|
||
|
||
|
||
def _build_info(ip: str, maxmind: dict[str, Any], ipinfo: dict[str, Any], ipapi: dict[str, Any]) -> dict[str, Any]:
|
||
if not maxmind.get("lite") and not maxmind.get("error"):
|
||
src = maxmind
|
||
ip_type = "原生 IP" if src.get("country_code") == src.get("reg_country_code") else "广播 IP"
|
||
else:
|
||
src = ipinfo if not ipinfo.get("error") else ipapi
|
||
cc = src.get("country_code")
|
||
ip_type = "原生 IP" if cc else "未知"
|
||
return {
|
||
"ip": ip,
|
||
"asn": maxmind.get("asn") or ipinfo.get("asn") or ipapi.get("asn"),
|
||
"organization": maxmind.get("org") or ipinfo.get("org") or ipapi.get("org"),
|
||
"city": maxmind.get("city") or ipinfo.get("city") or ipapi.get("city"),
|
||
"country": maxmind.get("country") or ipapi.get("country"),
|
||
"country_code": maxmind.get("country_code") or ipinfo.get("country_code") or ipapi.get("country_code"),
|
||
"timezone": maxmind.get("timezone") or ipinfo.get("timezone") or ipapi.get("timezone"),
|
||
"latitude": maxmind.get("lat") or ipinfo.get("lat") or ipapi.get("lat"),
|
||
"longitude": maxmind.get("lon") or ipinfo.get("lon") or ipapi.get("lon"),
|
||
"ip_type": ip_type,
|
||
"mode_lite": bool(maxmind.get("lite")),
|
||
}
|
||
|
||
|
||
def _build_type_usage(ipinfo: dict[str, Any], ipapi: dict[str, Any]) -> dict[str, str]:
|
||
out: dict[str, str] = {}
|
||
if not ipinfo.get("error"):
|
||
out["IPinfo"] = _type_label(ipinfo.get("usage_type"))
|
||
out["IPinfo_company"] = _type_label(ipinfo.get("company_type"))
|
||
if not ipapi.get("error"):
|
||
out["ipapi"] = _type_label(ipapi.get("usage_type"))
|
||
out["ipapi_company"] = _type_label(ipapi.get("company_type"))
|
||
return out
|
||
|
||
|
||
def _build_scores(ipapi: dict[str, Any], ipqs: dict[str, Any], scam: dict[str, Any]) -> list[dict[str, Any]]:
|
||
scores: list[dict[str, Any]] = []
|
||
if ipapi.get("score_pct") is not None:
|
||
scores.append(
|
||
{
|
||
"source": "ipapi",
|
||
"value": ipapi["score_pct"],
|
||
"unit": "%",
|
||
"risk": ipapi.get("risk_key"),
|
||
"risk_label": RISK_LABELS.get(str(ipapi.get("risk_key") or ""), ipapi.get("risk_key")),
|
||
}
|
||
)
|
||
if ipqs.get("fraud_score") is not None:
|
||
fs = int(ipqs["fraud_score"])
|
||
rk = _ipqs_risk(fs)
|
||
scores.append(
|
||
{
|
||
"source": "IPQS",
|
||
"value": fs,
|
||
"unit": "分",
|
||
"risk": rk,
|
||
"risk_label": RISK_LABELS.get(rk or "", rk),
|
||
}
|
||
)
|
||
scam_score = scam.get("score") or scam.get("scamalytics_score")
|
||
if scam_score is not None:
|
||
try:
|
||
sv = float(scam_score)
|
||
scores.append({"source": "SCAMALYTICS", "value": sv, "unit": "分", "risk": None, "risk_label": ""})
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return scores
|
||
|
||
|
||
def _build_factors(
|
||
ipinfo: dict[str, Any], ipapi: dict[str, Any], ipqs: dict[str, Any]
|
||
) -> dict[str, dict[str, Optional[bool]]]:
|
||
factors: dict[str, dict[str, Optional[bool]]] = {
|
||
"Proxy": {},
|
||
"VPN": {},
|
||
"Tor": {},
|
||
"Hosting": {},
|
||
}
|
||
if not ipinfo.get("error"):
|
||
factors["Proxy"]["IPinfo"] = _bool_factor(ipinfo.get("proxy"))
|
||
factors["VPN"]["IPinfo"] = _bool_factor(ipinfo.get("vpn"))
|
||
factors["Tor"]["IPinfo"] = _bool_factor(ipinfo.get("tor"))
|
||
factors["Hosting"]["IPinfo"] = _bool_factor(ipinfo.get("hosting"))
|
||
if not ipapi.get("error"):
|
||
factors["Proxy"]["ipapi"] = _bool_factor(ipapi.get("proxy"))
|
||
factors["VPN"]["ipapi"] = _bool_factor(ipapi.get("vpn"))
|
||
factors["Tor"]["ipapi"] = _bool_factor(ipapi.get("tor"))
|
||
factors["Hosting"]["ipapi"] = _bool_factor(ipapi.get("hosting"))
|
||
if not ipqs.get("error"):
|
||
factors["Proxy"]["IPQS"] = _bool_factor(ipqs.get("proxy"))
|
||
factors["VPN"]["IPQS"] = _bool_factor(ipqs.get("vpn"))
|
||
factors["Tor"]["IPQS"] = _bool_factor(ipqs.get("tor"))
|
||
return factors
|
||
|
||
|
||
async def run_ip_quality() -> dict[str, Any]:
|
||
timeout = httpx.Timeout(READ_TIMEOUT, connect=CONNECT_TIMEOUT)
|
||
limits = httpx.Limits(max_keepalive_connections=8, max_connections=12)
|
||
errors: list[str] = []
|
||
|
||
async with httpx.AsyncClient(
|
||
timeout=timeout,
|
||
limits=limits,
|
||
verify=True,
|
||
headers={"User-Agent": UA},
|
||
follow_redirects=True,
|
||
) as client:
|
||
ip, ip_err = await _resolve_public_ip(client)
|
||
if not ip:
|
||
return {"error": ip_err or "无法获取公网 IP", "meta": {"at": time.time()}}
|
||
|
||
maxmind_t = asyncio.create_task(_fetch_maxmind(client, ip))
|
||
ipinfo_t = asyncio.create_task(_fetch_ipinfo_widget(client, ip))
|
||
ipapi_t = asyncio.create_task(_fetch_ipapi(client, ip))
|
||
ipqs_t = asyncio.create_task(_fetch_check_place_db(client, ip, "ipqualityscore"))
|
||
scam_t = asyncio.create_task(_fetch_check_place_db(client, ip, "scamalytics"))
|
||
yt_t = asyncio.create_task(_test_youtube(client))
|
||
tt_t = asyncio.create_task(_test_tiktok(client))
|
||
|
||
maxmind, ipinfo, ipapi, ipqs_raw, scam_raw, youtube, tiktok = await asyncio.gather(
|
||
maxmind_t, ipinfo_t, ipapi_t, ipqs_t, scam_t, yt_t, tt_t
|
||
)
|
||
|
||
ipqs = ipqs_raw if not ipqs_raw.get("error") else {}
|
||
scam = scam_raw if not scam_raw.get("error") else {}
|
||
for label, blob in (("maxmind", maxmind), ("ipinfo", ipinfo), ("ipapi", ipapi), ("ipqs", ipqs_raw), ("scam", scam_raw)):
|
||
if blob.get("error"):
|
||
errors.append(f"{label}: {blob['error']}")
|
||
|
||
info = _build_info(ip, maxmind, ipinfo, ipapi)
|
||
usage = _build_type_usage(ipinfo, ipapi)
|
||
scores = _build_scores(ipapi, ipqs, scam)
|
||
factors = _build_factors(ipinfo, ipapi, ipqs)
|
||
|
||
media = {
|
||
"Youtube": youtube,
|
||
"TikTok": tiktok,
|
||
}
|
||
|
||
proxy_flags = [v for row in factors.values() for v in row.values() if v is True]
|
||
headline = "出口较干净，类型偏住宅/运营商" if not proxy_flags else "检测到代理/VPN/机房特征，录制前请确认"
|
||
|
||
return {
|
||
"head": {
|
||
"ip": ip,
|
||
"version": SCRIPT_VERSION,
|
||
"project": "IPQuality",
|
||
"github": "https://github.com/xykt/IPQuality",
|
||
},
|
||
"info": info,
|
||
"type": {"usage": usage},
|
||
"scores": scores,
|
||
"factors": factors,
|
||
"media": media,
|
||
"summary": {"headline": headline, "proxy_like": bool(proxy_flags)},
|
||
"meta": {
|
||
"at": time.time(),
|
||
"mode": "lite" if info.get("mode_lite") else "full",
|
||
"sources": [
|
||
"ipinfo.check.place",
|
||
"ipinfo.io",
|
||
"api.ipapi.is",
|
||
"xykt/IPQuality",
|
||
],
|
||
"errors": errors,
|
||
"disclaimer": "部分数据库（IPQS/Scamalytics）可能因网络环境不可用；结果仅供参考。",
|
||
},
|
||
}
|