tianxuan/analysis/nl_describe.py

"""
Template-based natural language feature descriptions.

Uses value-range → text mappings to generate Chinese natural language
descriptions of cluster features. Consumed by cluster_overview.html and
entity_profile.html for human-readable cluster summaries.

Data sources
------------
- ``analysis.tls_ref.get_tls_ref_dict()`` : ``field_code → {meaning, data_type}``
- ``ClusterFeature`` model rows : feature_name, mean, std, distinguishing_score

Usage::

    from analysis.nl_describe import describe_feature, describe_cluster

    text = describe_feature("8ack", 720.5, std_dev=2.1)
    # "中等数据包 (720 bytes)，比全局均值高2.1个标准差"

    summary = describe_cluster(features, top_n=5)
    # "此簇特征: 低延迟(23ms), 大数据包(1520 bytes), ..."
"""

from __future__ import annotations

from typing import Optional, Sequence

# ---------------------------------------------------------------------------
# Category mapping: field_code → semantic category
# ---------------------------------------------------------------------------

CATEGORY_EXACT: dict[str, str] = {
    # ── latency ──
    "4dur": "latency",
    "2tmo": "latency",
    "8dbd": "latency",
    # ── payload bytes ──
    "8ack": "packet_bytes",
    "8byt": "packet_bytes",
    # ── packet count ──
    "8pak": "packet_count",
    "8ppk": "packet_count",
    # ── port ──
    ":prs": "port",
    ":prd": "port",
    # ── TLS version ──
    "0ver": "tls_version",
    # ── key size ──
    "4ksz": "key_size",
    # ── cipher code ──
    "0cph": "cipher_code",
    # ── curve code ──
    "0crv": "curve_code",
    # ── session / offset ──
    "8seq": "session_offset",
    "8ses": "session_offset",
    "8did": "session_offset",
    # ── IP protocol ──
    "1ipp": "ip_protocol",
    # ── TLS random ──
    "0rnt": "tls_random_time",
    "0rnd": "tls_random",
    # ── database refs ──
    "4dbn": "database_ref",
    "4srs": "device_ref",
    # ── recovery flags ──
    "cnrs": "recovery",
    "isrs": "recovery",
}

# Substring-based matching (checked after exact match fails)
CATEGORY_SUBSTR: list[tuple[str, str]] = [
    ("latd", "latitude"),
    ("lond", "longitude"),
    ("orgn", "organization"),
    ("ispn", "isp"),
    ("anon", "anonymity"),
    ("doma", "domain"),
    ("city", "city"),
    ("_lat", "latitude"),
    ("_lon", "longitude"),
    ("latitude", "latitude"),
    ("longitude", "longitude"),
    ("duration", "latency"),
    ("delay", "latency"),
    ("bytes", "packet_bytes"),
    ("packets", "packet_count"),
    ("count", "count"),
    ("ratio", "ratio"),
    ("tls_ver", "tls_version"),
    ("version", "tls_version"),
]

# ---------------------------------------------------------------------------
# TLS version code → human description
# ---------------------------------------------------------------------------

TLS_VERSION_MAP: dict[int, str] = {
    0x0000: "SSL 2.0",
    0x0300: "SSL 3.0",
    0x0301: "TLS 1.0",
    0x0302: "TLS 1.1",
    0x0303: "TLS 1.2",
    0x0304: "TLS 1.3",
    0x7F00: "TLS 1.3 (draft)",
    0x7F12: "TLS 1.3 (draft 18)",
    0xFEFF: "DTLS 1.0",
}

# ---------------------------------------------------------------------------
# Templates: category → [(predicate, template_string), ...]
#
# Template variables available:
#   {v}      — value (formatted to 1 decimal unless overridden)
#   {v:.0f}  — value with custom format
#   {sv}     — signed value (e.g. "+2.1" for positive zscore)
#   {av}     — absolute value (e.g. "2.1")
#   {z:.1f}  — z-score signed (preserved in deviation suffix)
# ---------------------------------------------------------------------------

TEMPLATES: dict[str, list[tuple]] = {
    "latency": [
        (lambda v: v > 1000, "极高延迟 ({v:.0f}ms)"),
        (lambda v: v > 500, "高延迟 ({v:.0f}ms)"),
        (lambda v: v > 200, "中等延迟 ({v:.0f}ms)"),
        (lambda v: v > 50, "较低延迟 ({v:.0f}ms)"),
        (lambda v: True, "低延迟 ({v:.0f}ms)"),
    ],
    "packet_bytes": [
        (lambda v: v > 1500, "大数据包 ({v:.0f} bytes, 超过MTU)"),
        (lambda v: v > 1000, "较大数据包 ({v:.0f} bytes)"),
        (lambda v: v > 400, "中等数据包 ({v:.0f} bytes)"),
        (lambda v: True, "小数据包 ({v:.0f} bytes)"),
    ],
    "packet_count": [
        (lambda v: v > 200, "大量数据包 ({v:.0f} 个)"),
        (lambda v: v > 50, "中等数据包数 ({v:.0f} 个)"),
        (lambda v: True, "少量数据包 ({v:.0f} 个)"),
    ],
    "port": [
        (lambda v: v <= 1024, "系统端口 ({v:.0f})"),
        (lambda v: v <= 49151, "注册端口 ({v:.0f})"),
        (lambda v: True, "动态/私有端口 ({v:.0f})"),
    ],
    "tls_version": [
        (lambda v: 0x0304 <= v <= 0x0304, "TLS 1.3 (最新)"),
        (lambda v: 0x0303 <= v < 0x0304, "TLS 1.2 (主流)"),
        (lambda v: 0x0302 <= v < 0x0303, "TLS 1.1 (已弃用)"),
        (lambda v: 0x0301 <= v < 0x0302, "TLS 1.0 (过时)"),
        (lambda v: 0x0300 <= v < 0x0301, "SSL 3.0 (不安全)"),
        (lambda v: v < 0x0300, "SSL 2.0 (极不安全)"),
        (lambda v: v > 0x0304, "DTLS/实验性TLS (0x{v:.0X})"),
        (lambda v: True, "未知TLS版本 (0x{v:.0X})"),
    ],
    "key_size": [
        (lambda v: v >= 384, "高强度密钥 ({v:.0f} bits)"),
        (lambda v: v >= 256, "强密钥 ({v:.0f} bits)"),
        (lambda v: v >= 128, "标准密钥 ({v:.0f} bits)"),
        (lambda v: v >= 64, "弱密钥 ({v:.0f} bits)"),
        (lambda v: True, "极弱密钥 ({v:.0f} bits)"),
    ],
    "ip_protocol": [
        (lambda v: abs(v - 6) < 0.5, "TCP 协议 ({v:.0f})"),
        (lambda v: abs(v - 17) < 0.5, "UDP 协议 ({v:.0f})"),
        (lambda v: abs(v - 1) < 0.5, "ICMP 协议 ({v:.0f})"),
        (lambda v: abs(v - 2) < 0.5, "IGMP 协议 ({v:.0f})"),
        (lambda v: True, "IP 协议 #{v:.0f}"),
    ],
    "session_offset": [
        (lambda v: v > 10000, "大型会话 (偏移 {v:.0f})"),
        (lambda v: v > 1000, "中型会话 (偏移 {v:.0f})"),
        (lambda v: True, "小型会话 (偏移 {v:.0f})"),
    ],
    "latitude": [
        (lambda v: v > 60, "高纬度地区 ({v:.1f}°)"),
        (lambda v: v > 23.5, "中纬度地区 — 北温带 ({v:.1f}°)"),
        (lambda v: v >= -23.5, "低纬度地区 — 热带 ({v:.1f}°)"),
        (lambda v: v >= -60, "中纬度地区 — 南温带 ({v:.1f}°)"),
        (lambda v: True, "高纬度地区 — 南极 ({v:.1f}°)"),
    ],
    "longitude": [
        (lambda v: v > 120 or v < -120, "远东地区 ({v:.1f}°)"),
        (lambda v: v > 60 or v < -60, "中亚/东欧地区 ({v:.1f}°)"),
        (lambda v: v > 0 or v < 0, "西欧/非洲地区 ({v:.1f}°)"),
        (lambda v: True, "本初子午线附近 ({v:.1f}°)"),
    ],
    "cipher_code": [
        (lambda v: True, "加密套件代码 0x{v:.0X}"),
    ],
    "curve_code": [
        (lambda v: True, "椭圆曲线代码 0x{v:.0X}"),
    ],
    "tls_random_time": [
        (lambda v: True, "TLS 随机时间戳 ({v:.0f})"),
    ],
    "tls_random": [
        (lambda v: True, "TLS 随机值"),
    ],
    "database_ref": [
        (lambda v: True, "数据库引用 #{v:.0f}"),
    ],
    "device_ref": [
        (lambda v: True, "设备标识 #{v:.0f}"),
    ],
    "recovery": [
        (lambda v: v >= 0.5, "可恢复流"),
        (lambda v: True, "不可恢复流"),
    ],
    "organization": [
        (lambda v: True, "组织指标 ({v:.1f})"),
    ],
    "isp": [
        (lambda v: True, "ISP 指标 ({v:.1f})"),
    ],
    "anonymity": [
        (lambda v: True, "匿名状态指标 ({v:.1f})"),
    ],
    "domain": [
        (lambda v: True, "域名指标 ({v:.1f})"),
    ],
    "city": [
        (lambda v: True, "城市指标 ({v:.1f})"),
    ],
    "count": [
        (lambda v: v > 100, "计数很高 ({v:.0f})"),
        (lambda v: v > 50, "计数较高 ({v:.0f})"),
        (lambda v: True, "计数正常 ({v:.0f})"),
    ],
    "ratio": [
        (lambda v: v > 0.8, "占比极高 ({v:.1%})"),
        (lambda v: v > 0.5, "占比偏高 ({v:.1%})"),
        (lambda v: v >= 0.0, "占比较低 ({v:.1%})"),
        (lambda v: True, "占比 ({v:.1%})"),
    ],
}

# Default template for unrecognised categories / fallback
DEFAULT_TEMPLATES: list[tuple] = [
    (lambda v: True, "{meaning} = {v:.2f}"),
]


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def _resolve_category(feature_name: str) -> str | None:
    """Map a ``feature_name`` (column name) to a template category string.

    Tries exact match against ``CATEGORY_EXACT`` first, then substring
    match against ``CATEGORY_SUBSTR``.  Returns ``None`` if no match.
    """
    # Exact match
    cat = CATEGORY_EXACT.get(feature_name)
    if cat is not None:
        return cat

    # Substring match
    lowered = feature_name.lower()
    for sub, cat_label in CATEGORY_SUBSTR:
        if sub in lowered:
            return cat_label

    return None


def _lookup_meaning(feature_name: str) -> str:
    """Return the Chinese meaning of *feature_name* from the TLS reference DB.

    Falls back to *feature_name* itself when no entry is found.
    """
    try:
        from analysis.tls_ref import get_tls_ref_dict

        ref = get_tls_ref_dict()
        entry = ref.get(feature_name)
        if entry is not None:
            return entry.get("meaning", feature_name)
    except Exception:
        pass
    return feature_name


def describe_feature(
    feature_name: str,
    value: float,
    std_dev: float | None = None,
    cluster_std: float | None = None,
) -> str:
    """Generate a Chinese natural-language description for a single feature.

    Parameters
    ----------
    feature_name : str
        Column name (e.g. ``"8ack"``, ``":ips.latd"``).
    value : float
        The feature's mean value in the current context (cluster mean).
    std_dev : float | None
        Optional Z-score magnitude (``abs(distinguishing_score)``).  When
        provided and >= 1.0, appends a deviation suffix.
    cluster_std : float | None
        Optional within-cluster standard deviation. Describes how spread
        the values are within this cluster (high = scattered, low = tight).

    Returns
    -------
    str
        Natural-language description string.
    """
    meaning = _lookup_meaning(feature_name)
    category = _resolve_category(feature_name)

    # Select templates
    tmpl_list = TEMPLATES.get(category) if category else None
    if tmpl_list is None:
        tmpl_list = DEFAULT_TEMPLATES

    # Evaluate predicates
    text = ""
    for predicate, template in tmpl_list:
        try:
            if predicate(value):
                text = template.format(v=value, sv=_signed(value), av=abs(value),
                                       meaning=meaning)
                break
        except Exception:
            continue

    if not text:
        text = f"{meaning} = {value:.2f}"

    # Append deviation context
    if std_dev is not None and std_dev >= 1.0:
        text += f"，比全局均值高{std_dev:.1f}个标准差"

    # Append within-cluster variance context
    if cluster_std is not None and cluster_std > 0:
        if cluster_std > 2.0:
            text += "（簇内离散度高）"
        elif cluster_std < 0.3:
            text += "（簇内高度集中）"

    return text


def describe_cluster(
    features: Sequence[dict],
    top_n: int = 5,
) -> str:
    """Generate a 2-3 sentence natural-language summary for a cluster.

    Parameters
    ----------
    features : list[dict]
        Each dict must include:

        - ``feature_name`` : str
        - ``mean`` : float (cluster mean)
        - ``distinguishing_score`` : float (Z-score; may be signed)

        Optional keys used when present:

        - ``std`` : float (cluster std dev)

    top_n : int
        Number of most-distinguishing features to include (default 5).

    Returns
    -------
    str
        A short paragraph describing the cluster's distinguishing traits.

        Example:
        ``"低延迟 (23ms)，比全局均值高3.1个标准差; 大数据包 (1520 bytes)，比全局均值高2.3个标准差; 系统端口 (443)"``
    """
    if not features:
        return "无显著区分特征"

    # Sort by |distinguishing_score| descending
    ranked = sorted(
        features,
        key=lambda f: abs(f.get("distinguishing_score", 0.0)),
        reverse=True,
    )[:top_n]

    sentences: list[str] = []
    for f in ranked:
        name = f.get("feature_name", "")
        mean = f.get("mean")
        score = abs(f.get("distinguishing_score", 0.0))
        cstd = f.get("std")

        if mean is None or name == "":
            continue

        desc = describe_feature(name, float(mean), std_dev=score if score >= 1.0 else None,
                                cluster_std=float(cstd) if cstd is not None else None)
        sentences.append(desc)

    if not sentences:
        return "无显著区分特征"

    return "；".join(sentences) + "。"


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _signed(value: float) -> str:
    """Return a signed string representation, e.g. ``"+2.1"``."""
    return f"{value:+.1f}"