"""
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）可能因网络环境不可用；结果仅供参考。",
            },
        }