# -*- coding: utf-8 -*-
"""来源 IP 提取与访问统计用纯函数。

服务经``域名 → CDN → 公网 EIP → 内网 → 宿主机 nginx → 容器``链路访问，容器内
socket 对端是 Docker 网关而非真实客户端，因此真实 IP 只能来自 nginx 注入的
请求头。安全前提：

* 只有当 socket 对端（``peer``）属于可信代理网段时才采信请求头；
* nginx 用 ``realip`` 模块按 CDN 网段校验后写入 ``X-Real-IP``（见文档）；
* 直接解析到公网 EIP 的请求没有 CDN 头，回退到 ``peer``。

本模块不依赖网络、文件或时钟，便于离线单元测试。
"""

from __future__ import annotations

import ipaddress
import re
from typing import Iterable, Mapping, Sequence

# 默认可信代理网段：回环、RFC1918、链路本地、IPv6 ULA/链路本地。
# Docker 默认桥接网段在 172.16.0.0/12 内，容器内看到的对端即网关地址。
DEFAULT_TRUSTED_PROXIES: tuple[str, ...] = (
    "127.0.0.0/8",
    "::1/128",
    "10.0.0.0/8",
    "172.16.0.0/12",
    "192.168.0.0/16",
    "169.254.0.0/16",
    "fc00::/7",
    "fe80::/10",
)

# 依次尝试的请求头：nginx 归一化后的 X-Real-IP 优先，X-Forwarded-For 兜底。
DEFAULT_CLIENT_IP_HEADERS: tuple[str, ...] = ("X-Real-IP", "X-Forwarded-For")

# 已知前端路由：未知路径统一归为 /other，避免卡片式路径污染统计。
KNOWN_ROUTES: frozenset[str] = frozenset({
    "/home", "/videos", "/ups", "/fans", "/about",
    "/candidates", "/system", "/rank", "/login", "/logout",
})

_VIDEO_ROUTE_RE = re.compile(r"^/video/\d+$")
_UP_ROUTE_RE = re.compile(r"^/up/\d+$")
_MAX_ROUTE_LENGTH = 200

# 常见爬虫 / 监控 UA 关键词；命中时视为机器人（是否计数由调用方决定）。
# 这些标记在 UA 中通常作为子串出现（如 Googlebot / bingbot / Baiduspider），
# 因此不使用词边界，避免漏判。
_BOT_PATTERN = re.compile(
    r"(?i)("
    r"bot|crawler|spider|crawl|slurp|bingpreview|facebookexternalhit|"
    r"python-requests|python-urllib|httpx|aiohttp|curl|wget|libwww|"
    r"monitor|uptime|healthcheck|headlesschrome|phantomjs|"
    r"semrush|ahrefs|dotbot|yandex|baiduspider|sogou|bytespider"
    r")"
)


def normalize_ip(value: object) -> str | None:
    """规范化 IP 字符串；非法或空值返回 ``None``。

    兼容 ``[::1]:443``、``1.2.3.4:5678`` 带端口写法，并把 IPv4-mapped IPv6
    折叠为 IPv4（``::ffff:1.2.3.4`` → ``1.2.3.4``），保证同一客户端去重一致。
    """
    if not isinstance(value, str):
        return None
    text = value.strip()
    if not text:
        return None
    if text.startswith("["):
        end = text.find("]")
        if end == -1:
            return None
        text = text[1:end]
    ip = _parse_ip(text)
    if ip is None:
        # host:port（IPv6 已在上面的直接解析里成功，不会走到这里）
        host, sep, port = text.rpartition(":")
        if not sep or not port.isdigit() or not host:
            return None
        ip = _parse_ip(host)
    if ip is None:
        return None
    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
        ip = ip.ipv4_mapped
    return ip.compressed


def _parse_ip(text: str) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
    try:
        return ipaddress.ip_address(text)
    except ValueError:
        return None


def parse_networks(values: Iterable[str]) -> tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...]:
    """把 CIDR / 裸 IP 字符串解析为网段元组；非法项忽略。"""
    networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
    for raw in values:
        if not isinstance(raw, str) or not raw.strip():
            continue
        text = raw.strip()
        try:
            networks.append(ipaddress.ip_network(text, strict=False))
        except ValueError:
            continue
    return tuple(networks)


def is_trusted_proxy(
    peer_ip: str | None,
    networks: Sequence[ipaddress.IPv4Network | ipaddress.IPv6Network],
) -> bool:
    ip = _parse_ip(peer_ip) if peer_ip else None
    if ip is None:
        return False
    for network in networks:
        if ip.version != network.version:
            continue
        if ip in network:
            return True
    return False


def _header_value(headers: Mapping[str, object], name: str) -> str | None:
    """大小写不敏感地读取请求头；兼容单值与列表两种取值。"""
    plain = headers.get(name)
    if plain is None:
        lowered = name.lower()
        for key, value in headers.items():
            if isinstance(key, str) and key.lower() == lowered:
                plain = value
                break
    if plain is None:
        return None
    if isinstance(plain, (list, tuple)):
        return ",".join(str(item) for item in plain)
    return str(plain)


def _rightmost_valid_forwarded(value: str) -> str | None:
    """X-Forwarded-For 取最右侧合法 IP。

    nginx 的 ``$proxy_add_x_forwarded_for`` 会把真实对端追加到最右侧，客户端
    自带的伪造条目在左侧，因此最右侧最可信。
    """
    for part in reversed(value.split(",")):
        normalized = normalize_ip(part)
        if normalized is not None:
            return normalized
    return None


def extract_client_ip(
    headers: Mapping[str, object],
    peer_ip: str | None,
    *,
    trusted_proxies: Sequence[ipaddress.IPv4Network | ipaddress.IPv6Network],
    header_order: Sequence[str] = DEFAULT_CLIENT_IP_HEADERS,
) -> str:
    """返回用于统计的客户端 IP。

    * 对端不可信（或无法解析）时忽略所有请求头，直接回退到对端；
    * 对端可信时按 ``header_order`` 依次取第一个合法值；
    * 全部缺失时回退到规范化后的对端；对端也无法解析时返回 ``""``。
    """
    peer = normalize_ip(peer_ip)
    if peer is None:
        return ""
    if not is_trusted_proxy(peer, trusted_proxies):
        return peer

    for name in header_order:
        value = _header_value(headers, name)
        if value is None:
            continue
        if name.lower() == "x-forwarded-for":
            candidate = _rightmost_valid_forwarded(value)
        else:
            candidate = normalize_ip(value)
        if candidate is not None:
            return candidate
    return peer


def is_bot(user_agent: object) -> bool:
    """UA 命中已知机器人/监控模式时返回 ``True``；空 UA 不算机器人。"""
    if not isinstance(user_agent, str) or not user_agent.strip():
        return False
    return bool(_BOT_PATTERN.search(user_agent))


def normalize_route(path: object) -> str:
    """把上报路径归一化为有限集合中的路由标签。

    ``/`` → ``/home``；``/video/123`` → ``/video/:aid``；``/up/123`` →
    ``/up/:mid``；已知路由原样保留；其余（含空值、超长、带查询串）→ ``/other``。
    """
    if not isinstance(path, str):
        return "/other"
    text = path.strip()
    if not text:
        return "/other"
    # 丢弃查询串与片段，并折叠末尾斜杠
    text = text.split("?", 1)[0].split("#", 1)[0]
    if not text.startswith("/"):
        return "/other"
    if len(text) > _MAX_ROUTE_LENGTH:
        return "/other"
    if text == "/" or text == "":
        return "/home"
    if text != "/" and text.endswith("/"):
        text = text.rstrip("/") or "/home"
    if _VIDEO_ROUTE_RE.match(text):
        return "/video/:aid"
    if _UP_ROUTE_RE.match(text):
        return "/up/:mid"
    if text in KNOWN_ROUTES:
        return text
    return "/other"