"""简单分析模块 v2 — 离线 GeoIP + 高级筛选 + 高性能聚类 + 丰富可视化。

改进：
- 使用离线 GeoIP 数据库（analysis.geoip）解析 IP 经纬度，抛弃 CSV 自带列
- 列名统一规范化，内置 50+ 列映射字典
- 上传预设筛选 + 高级 AND/OR/NOT 筛选
- HDBSCAN 性能优化（float32、降采样、DBSCAN 回退）
- 全噪声时可视化、点属性、边关系
"""

import json
import logging
import os
import re
import shutil
import time
import uuid
from pathlib import Path
from collections import Counter

import numpy as np
import polars as pl
from django.conf import settings
from django.http import JsonResponse
from django.shortcuts import render
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST, require_GET
from sklearn.cluster import HDBSCAN, DBSCAN
from sklearn.preprocessing import StandardScaler

# 复用天璇原有离线 GeoIP 模块（data/geoip_data.txt 覆盖 23 城市 ISP 段）
from analysis.geoip import lookup as geoip_lookup

logger = logging.getLogger(__name__)

# ── In-memory cache ──────────────────────────────────────────────────────────
_cache: dict[str, dict] = {}
_TEMP_DIR = Path(settings.FILE_UPLOAD_TEMP_DIR) / 'simple_analysis'
_TEMP_DIR.mkdir(parents=True, exist_ok=True)


def _new_id() -> str:
    return uuid.uuid4().hex[:12]


def _cached(key: str) -> dict | None:
    return _cache.get(key)


def _store(key: str, data: dict) -> None:
    _cache[key] = data


# ═══════════════════════════════════════════════════════════════════════════════
#  列名规范化 — 完整映射字典
# ═══════════════════════════════════════════════════════════════════════════════

# 标准内部名称 → 可能的 CSV 原始列名（按优先级）
_COLUMN_ALIASES: dict[str, list[str]] = {
    # IP 地址
    'ips':      [':ips', 'src_ip', 'source_ip', 'ip_src', 'client-ip', 'client_ip'],
    'ipd':      [':ipd', 'dst_ip', 'dest_ip', 'destination_ip', 'ip_dst', 'server-ip', 'server_ip'],
    # 布尔/状态
    'cnrs':     ['cnrs', 'conn_success', 'result', '4srs'],
    'isrs':     ['isrs', 'is_resolved', 'resolved', '8ses'],
    # 类别/名称
    'cnam':     ['cnam', 'snam', 'class_name', 'category', 'target', 'name', 'tabl'],
    'snam':     ['snam', 'server_name', 'tls_sni', 'sni'],
    # 经纬度（仍保留检测，但聚类不再依赖）
    'src_lat':  [':ips.latd', 'src_latitude', 'src_lat', 'source_lat'],
    'src_lon':  [':ips.lond', 'src_longitude', 'src_lon', 'source_lon'],
    'dst_lat':  [':ipd.latd', 'dst_latitude', 'dst_lat', 'dest_lat'],
    'dst_lon':  [':ipd.lond', 'dst_longitude', 'dst_lon', 'dest_lon'],
    # 端口/协议
    'dport':    ['1ipp', 'dst_port', 'dest_port', 'port', 'dport'],
    'proto':    ['proto', 'protocol', 'l4_proto'],
    # TLS
    'tls_ver':  ['0ver', 'tls_version', 'version', 'tlsver'],
    'tls_cph':  ['0cph', 'cipher_suite', 'cipher', 'tls_cipher'],
    'tls_rnd':  ['0rnd', 'tls_random', 'random'],
    'tls_rnt':  ['0rnt', 'tls_rnt'],
    # 流量
    'bytes':    ['8byt', 'bytes_sent', 'bytes', 'src_bytes', '8ppk'],
    'duration': ['4dur', 'duration', 'dur', 'flow_duration'],
    'packets':  ['8pak', 'packets', 'pkt_count'],
    # 时序
    'time':     ['time', 'timestamp', '4dbn'],
    'seq':      ['8seq', 'seq', 'sequence'],
    'ack':      ['8ack', 'ack', 'acknowledgement'],
    # 地理信息
    'ispn':     [':ips.ispn', 'isp_name', 'isp'],
    'orgn':     [':ips.orgn', 'organization', 'org'],
    'city':     [':ips.city', 'city', 'src_city'],
    'ipd_ispn': [':ipd.ispn', 'dst_ispn', 'dst_isp'],
    'ipd_orgn': [':ipd.orgn', 'dst_organization', 'dst_org'],
    'ipd_city': [':ipd.city', 'dst_city'],
    # 其他
    'scnt':     ['scnt', 'country', 'country_code'],
    'dcnt':     ['dcnt', 'dst_country'],
    'source_node': ['source-node', 'source_node'],
    'prd':      [':prd', 'prd', 'product'],
    'prs':      [':prs', 'prs', 'process'],
    'ipd_anon': [':ipd.anon', 'dst_anon', 'anon'],
    '8did':     ['8did', 'did', 'device_id'],
    '8dbd':     ['8dbd', 'dbd'],
    '2tmo':     ['2tmo', 'tmo', 'timeout'],
    'text':     ['text', 'payload', 'content'],
    'host':     ['host', 'hostname', 'host_name', 'remote_host', 'peer_name'],
}


def _norm(name: str) -> str:
    """归一化列名：去首尾空格，特殊字符→下划线，转小写。"""
    n = name.strip().lower()
    n = re.sub(r'[-:.\s]+', '_', n)
    return n


def normalize_columns(df: pl.DataFrame) -> tuple[pl.DataFrame, dict[str, str]]:
    """对 DataFrame 列名进行统一规范化，返回 (新df, 规范化映射 {原始名→标准内部名})。

    规范化规则：
      1. 去首尾空格
      2. 连续特殊字符（- : . 空格）→ 下划线
      3. 全部小写
    然后根据 _COLUMN_ALIASES 模糊匹配到标准内部名称。
    """
    col_map = {}      # 原始名 → 标准内部名
    reverse_map = {}  # norm(原始名) → 原始名

    for c in df.columns:
        nc = _norm(c)
        reverse_map[nc] = c

    # 遍历别名表，尝试匹配
    used_originals = set()
    for std_name, aliases in _COLUMN_ALIASES.items():
        for alias in aliases:
            n_alias = _norm(alias)
            # 精确匹配
            if n_alias in reverse_map and reverse_map[n_alias] not in used_originals:
                col_map[reverse_map[n_alias]] = std_name
                used_originals.add(reverse_map[n_alias])
                break
            # 后缀匹配（如列名 _ips 匹配 :ips）
            for orig, n_orig in reverse_map.items():
                if orig in used_originals:
                    continue
                if n_orig == n_alias or n_orig.endswith('_' + n_alias) or n_orig.startswith(n_alias + '_'):
                    col_map[orig] = std_name
                    used_originals.add(orig)
                    break
            if std_name in col_map.values():
                break

    # 未匹配的保留原名
    for c in df.columns:
        if c not in col_map:
            col_map[c] = c

    # 重命名
    rename_dict = {orig: std for orig, std in col_map.items() if orig != std}
    if rename_dict:
        df = df.rename(rename_dict)

    return df, col_map


def find_col(norm_col_map: dict[str, str], *std_names: str) -> str | None:
    """在规范化映射中查找第一个匹配的标准列名，返回原始列名。"""
    for std in std_names:
        for orig, mapped in norm_col_map.items():
            if mapped == std:
                return orig
    return None


# ═══════════════════════════════════════════════════════════════════════════════
#  GeoIP 辅助（复用天璇离线库）
# ═══════════════════════════════════════════════════════════════════════════════

def resolve_ip_geo(ip_str: str | None) -> dict:
    """用离线 GeoIP 库解析单个 IP，返回 {lat, lon, city, country} 或空字典。"""
    if not ip_str or ip_str in ('None', '', '+', '-'):
        return {}
    try:
        result = geoip_lookup(ip_str)
        if result:
            return result
    except Exception:
        pass
    return {}


def batch_geoip_resolve(ip_list: list[str]) -> dict[str, dict]:
    """批量解析 IP 地理信息，返回 {ip: {lat, lon, city, country}}。"""
    cache: dict[str, dict] = {}
    for ip in set(ip_list):
        if ip and ip not in ('None', '', '+', '-'):
            result = geoip_lookup(ip)
            if result:
                cache[ip] = result
    return cache


# ═══════════════════════════════════════════════════════════════════════════════
#  Step 0: Quick column scan (header-only, files kept for lazy value scan)
# ═══════════════════════════════════════════════════════════════════════════════

@csrf_exempt
@require_POST
def quick_scan(request):
    """仅读取 CSV 表头 + 保留文件供后续单列值扫描。

    文件保存在 scan_dir，scan_id 存入 _cache。
    后续 column_values 可用 scan_id 做单列 streaming Top-N 查询。
    调用 upload_csv 时若提供 scan_id，可复用已保存的文件。
    """
    files = request.FILES.getlist('files')
    if not files:
        return JsonResponse({'error': '未上传任何文件'}, status=400)

    scan_id = _new_id()
    scan_dir = _TEMP_DIR / scan_id
    scan_dir.mkdir(parents=True, exist_ok=True)

    saved_paths = []
    for f in files:
        dest = scan_dir / f.name
        with open(dest, 'wb') as out:
            for chunk in f.chunks():
                out.write(chunk)
        saved_paths.append(str(dest))

    try:
        # Read header only (1 row is enough for column names)
        lf = pl.scan_csv(saved_paths[0], infer_schema_length=0)
        if len(saved_paths) > 1:
            frames = [pl.scan_csv(p, infer_schema_length=0) for p in saved_paths]
            lf = pl.concat(frames, how='diagonal_relaxed')
        df = lf.head(1).collect()
    except Exception as e:
        shutil.rmtree(scan_dir, ignore_errors=True)
        return JsonResponse({'error': f'CSV 表头读取失败: {e}'}, status=400)

    # Normalize column names
    df_norm, col_map = normalize_columns(df)

    # Keep files for lazy value scanning — store paths in cache
    _store(scan_id, {
        'scan_paths': saved_paths,
        'col_map': col_map,
        'columns': list(df_norm.columns),
    })

    return JsonResponse({
        'scan_id': scan_id,
        'columns': list(df_norm.columns),
        'columns_std': list(df_norm.columns),
        'col_map': {orig: std for orig, std in col_map.items() if orig != std},
    })


# ═══════════════════════════════════════════════════════════════════════════════
#  Step 1: Upload + 预设筛选 + GeoIP 解析
# ═══════════════════════════════════════════════════════════════════════════════

@csrf_exempt
@require_POST
def upload_csv(request):
    """上传 CSV，列归一化，预设筛选（支持树状高级筛选），GeoIP 解析，返回摘要。

    若提供 scan_id，复用 quick_scan 阶段已保存的文件，避免重复写入磁盘。
    """
    scan_id = request.POST.get('scan_id', request.GET.get('scan_id', ''))

    session_id = _new_id()
    upload_dir = _TEMP_DIR / session_id

    saved_paths = []

    # 复用 scan 阶段的文件（避免重复存储）
    if scan_id:
        scan_entry = _cached(scan_id)
        if scan_entry and scan_entry.get('scan_paths'):
            scan_dir = Path(scan_entry['scan_paths'][0]).parent
            # Move the scan directory to the session directory
            shutil.move(str(scan_dir), str(upload_dir))
            # Rebuild saved_paths from new location
            for p in upload_dir.iterdir():
                if p.is_file():
                    saved_paths.append(str(p))
            # Clean scan cache
            _cache.pop(scan_id, None)
        else:
            upload_dir.mkdir(parents=True, exist_ok=True)

    # Fallback: normal file upload
    if not saved_paths:
        files = request.FILES.getlist('files')
        if not files:
            return JsonResponse({'error': '未上传任何文件'}, status=400)
        upload_dir.mkdir(parents=True, exist_ok=True)
        for f in files:
            dest = upload_dir / f.name
            with open(dest, 'wb') as out:
                for chunk in f.chunks():
                    out.write(chunk)
            saved_paths.append(str(dest))

    # 解析预设筛选条件（支持树状结构 + 旧格式兼容）
    preset_root = None
    try:
        pf_raw = request.POST.get('preset_filters', request.GET.get('preset_filters', '{}'))
        if pf_raw:
            pf = json.loads(pf_raw)
            # 新格式：树状 root
            if 'root' in pf:
                preset_root = pf['root']
            # 旧格式兼容：{cnrs: "+", ...}
            elif any(k in pf for k in ('cnrs', 'isrs', 'cnam', 'snam')):
                # Convert legacy to tree
                items = []
                for key in ('cnrs', 'isrs', 'cnam', 'snam'):
                    if key in pf and pf[key]:
                        if key in ('cnrs', 'isrs'):
                            items.append({'type': 'condition', 'col': key, 'op': 'eq', 'val': pf[key]})
                        else:
                            items.append({'type': 'condition', 'col': key, 'op': 'contains', 'val': pf[key]})
                if items:
                    preset_root = {'type': 'group', 'logic': 'AND', 'items': items}
    except (json.JSONDecodeError, Exception):
        pass

    # Load & merge CSV
    try:
        lf = pl.scan_csv(
            saved_paths[0] if len(saved_paths) == 1 else str(upload_dir / '*.csv'),
            infer_schema_length=10000,
            try_parse_dates=True,
        )
        if len(saved_paths) > 1:
            frames = [pl.scan_csv(p, infer_schema_length=10000) for p in saved_paths]
            lf = pl.concat(frames, how='diagonal_relaxed')
        df = lf.collect(streaming=True)
    except Exception as e:
        shutil.rmtree(upload_dir, ignore_errors=True)
        return JsonResponse({'error': f'CSV 解析失败: {e}'}, status=400)

    # 列名规范化
    df, col_map = normalize_columns(df)
    total_rows = len(df)

    # 获取各关键列的 标准内部名（即重命名后的列名）
    def _get_std(*names):
        """从 col_map 中查找第一个匹配的标准名。"""
        for n in names:
            for orig, std in col_map.items():
                if std == n:
                    return std
        return None

    ips_col = _get_std('ips')
    ipd_col = _get_std('ipd')
    cnam_col = _get_std('cnam')
    cnrs_col = _get_std('cnrs')
    isrs_col = _get_std('isrs')
    snam_col = _get_std('snam')
    dport_col = _get_std('dport')
    proto_col = _get_std('proto')
    tls_ver_col = _get_std('tls_ver')
    tls_cph_col = _get_std('tls_cph')
    bytes_col = _get_std('bytes')
    packets_col = _get_std('packets')
    dur_col = _get_std('duration')
    time_col = _get_std('time')
    ispn_col = _get_std('ispn')
    orgn_col = _get_std('orgn')
    city_col = _get_std('city')
    scnt_col = _get_std('scnt')
    text_col = _get_std('text')
    host_col = _get_std('host')

    # ── 预设筛选（树状高级筛选） ──────────────────────────────────────────
    preset_filtered = total_rows
    if preset_root is not None:
        try:
            cond = _build_tree_condition(preset_root, df)
            if cond is not None:
                df = df.filter(cond)
                preset_filtered = len(df)
        except Exception as e:
            logger.warning('Preset filter error: %s', e)

    # ── GeoIP 批量解析（使用标准化列名） ────────────────────────────────────
    geo_cache: dict[str, dict] = {}
    geo_skipped = 0
    if ips_col and ips_col in df.columns:
        all_ips = df[ips_col].cast(pl.Utf8).unique().to_list()
        geo_cache = batch_geoip_resolve([ip for ip in all_ips if ip not in (None, '', 'None', '+', '-')])
        logger.info('GeoIP resolved %d IPs', len(geo_cache))

    # ── CNAM 频率 ────────────────────────────────────────────────────────────
    cnam_freq = []
    top_cnam = None
    if cnam_col:
        try:
            cnam_series = df[cnam_col].cast(pl.Utf8).fill_null('')
            counts = Counter(cnam_series.to_list())
            total_cnam = sum(counts.values())
            for val, cnt in counts.most_common():
                cnam_freq.append({
                    'value': val if val else '(空)',
                    'count': cnt,
                    'pct': round(cnt / total_cnam * 100, 1),
                })
            if cnam_freq:
                top_cnam = cnam_freq[0]
        except Exception as e:
            logger.warning('cnam frequency failed: %s', e)

    # 存储元数据
    meta = {
        'session_id': session_id,
        'total_rows': total_rows,
        'preset_filtered': preset_filtered,
        'col_names': list(df.columns),
        'col_map': col_map,               # {原始名→标准内部名}
        'geo_cache': geo_cache,            # {ip: {lat, lon, city, country}}
        'geo_skipped': geo_skipped,
        'ips_col': ips_col,
        'ipd_col': ipd_col,
        'cnam_col': cnam_col,
        'cnrs_col': cnrs_col,
        'isrs_col': isrs_col,
        'snam_col': snam_col,
        'dport_col': dport_col,
        'proto_col': proto_col,
        'tls_ver_col': tls_ver_col,
        'tls_cph_col': tls_cph_col,
        'bytes_col': bytes_col,
        'packets_col': packets_col,
        'dur_col': dur_col,
        'time_col': time_col,
        'ispn_col': ispn_col,
        'orgn_col': orgn_col,
        'city_col': city_col,
        'scnt_col': scnt_col,
        'text_col': text_col,
        'host_col': host_col,
        'snam_col': snam_col,
    }
    _store(session_id, {'meta': meta, 'df': df, 'geo_cache': geo_cache})

    return JsonResponse({
        'session_id': session_id,
        'total_rows': total_rows,
        'preset_filtered': preset_filtered,
        'columns': list(df.columns)[:50],
        'columns_std': list(df.columns),  # 标准化后列名，供前端筛选器使用
        'col_map': {orig: std for orig, std in col_map.items() if orig != std},
        'detected_columns': {
            'cnam': cnam_col,
            'snam': snam_col,
            'cnrs': cnrs_col,
            'isrs': isrs_col,
            'ips': ips_col,
            'ipd': ipd_col,
            'ispn': ispn_col,
            'orgn': orgn_col,
            'city': city_col,
            'host': host_col,
            'text': text_col,
        },
        'cnam_frequency': cnam_freq[:30],
        'top_cnam': top_cnam,
        'geo_stats': {
            'ips_resolved': len(geo_cache),
        },
    })


# ═══════════════════════════════════════════════════════════════════════════════
#  Step 2: 高级筛选构建器（AND/OR/NOT 组合条件）
# ═══════════════════════════════════════════════════════════════════════════════

def _build_condition(col_name: str, operator: str, value: str, logic: str = 'AND'):
    """构建单个筛选条件表达式。"""
    col = pl.col(col_name)
    v = value.strip()
    try:
        if operator == 'eq':
            cond = col.cast(pl.Utf8) == v
        elif operator == 'neq':
            cond = col.cast(pl.Utf8) != v
        elif operator == 'contains':
            cond = col.cast(pl.Utf8).str.contains(v)
        elif operator == 'not_contains':
            cond = ~col.cast(pl.Utf8).str.contains(v)
        elif operator == 'gt':
            cond = col.cast(pl.Float64) > float(v)
        elif operator == 'lt':
            cond = col.cast(pl.Float64) < float(v)
        elif operator == 'gte':
            cond = col.cast(pl.Float64) >= float(v)
        elif operator == 'lte':
            cond = col.cast(pl.Float64) <= float(v)
        elif operator == 'is_empty':
            cond = col.is_null() | (col.cast(pl.Utf8) == '')
        elif operator == 'not_empty':
            cond = col.is_not_null() & (col.cast(pl.Utf8) != '')
        else:
            cond = col.cast(pl.Utf8) == v
    except Exception:
        cond = col.cast(pl.Utf8) == v
    return cond


def _build_tree_condition(node, df) -> pl.Expr | None:
    """递归构建条件树，支持嵌套 AND/OR/NOT 组。

    JSON 节点格式:
      {"type": "condition", "col": "...", "op": "...", "val": "..."}
      {"type": "group", "logic": "AND|OR", "items": [...]}
      {"type": "group", "logic": "NOT", "items": [...]}   # 否定整个组
    """
    if node is None:
        return None
    t = node.get('type', 'condition')

    if t == 'condition':
        col = node.get('col', '')
        if not col or col not in df.columns:
            return None
        c = _build_condition(col, node.get('op', 'eq'), node.get('val', ''))
        if node.get('logic') == 'NOT':
            c = ~c
        return c

    if t == 'group':
        items = node.get('items', [])
        if not items:
            return None
        logic = node.get('logic', 'AND')
        conds = []
        for item in items:
            c = _build_tree_condition(item, df)
            if c is not None:
                conds.append(c)
        if not conds:
            return None
        if logic == 'NOT':
            combined = conds[0]
            for c in conds[1:]:
                combined = combined & c  # NOT 组内默认 AND 组合
            return ~combined
        combined = conds[0]
        for c in conds[1:]:
            if logic == 'OR':
                combined = combined | c
            else:
                combined = combined & c
        return combined

    return None


@csrf_exempt
@require_POST
def advanced_filter(request):
    """高级筛选：接受树状 AND/OR/NOT 分组条件，支持嵌套括号语义。"""
    body = json.loads(request.body)
    session_id = body.get('session_id')
    root = body.get('root')  # 新格式：树状结构

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期，请重新上传'}, status=400)

    df = entry['df']

    try:
        if root:
            # 新格式：递归构建条件树
            combined = _build_tree_condition(root, df)
            if combined is None:
                entry['filtered_df'] = df
                entry['meta']['filtered_rows'] = len(df)
                return JsonResponse({'filtered_rows': len(df), 'total_before_filter': len(df), 'preview_columns': df.columns[:30], 'preview_rows': []})
        else:
            # 旧格式（向后兼容）
            conditions = body.get('conditions', [])
            global_logic = body.get('global_logic', 'AND')
            if not conditions:
                entry['filtered_df'] = df
                entry['meta']['filtered_rows'] = len(df)
                return JsonResponse({'filtered_rows': len(df), 'total_before_filter': len(df), 'preview_columns': df.columns[:30], 'preview_rows': []})

            group_conds = []
            for cond_item in conditions:
                col = cond_item.get('col', '')
                op = cond_item.get('op', 'eq')
                val = cond_item.get('val', '')
                if not col or col not in df.columns:
                    continue
                c = _build_condition(col, op, val)
                if cond_item.get('logic') == 'NOT':
                    c = ~c
                group_conds.append(c)

            if not group_conds:
                return JsonResponse({'error': '无可用的筛选条件'}, status=400)

            if global_logic == 'OR':
                combined = group_conds[0]
                for c in group_conds[1:]:
                    combined = combined | c
            else:
                combined = group_conds[0]
                for c in group_conds[1:]:
                    combined = combined & c

        filtered = df.filter(combined)
    except Exception as e:
        return JsonResponse({'error': f'筛选构建失败: {e}'}, status=400)

    filtered_rows = len(filtered)
    entry['meta']['filtered_rows'] = filtered_rows
    entry['filtered_df'] = filtered

    preview_cols = [c for c in filtered.columns if not c.startswith('_')][:30]
    preview_rows = []
    for row in filtered.head(20).select(preview_cols).iter_rows(named=True):
        clean = {}
        for k, v in row.items():
            if v is None or (isinstance(v, float) and np.isnan(v)):
                clean[k] = None
            else:
                clean[k] = str(v)[:120]
        preview_rows.append(clean)

    return JsonResponse({
        'filtered_rows': filtered_rows,
        'preview_columns': preview_cols,
        'preview_rows': preview_rows,
        'total_before_filter': len(df),
    })


# ── 兼容原 filter 接口 ──────────────────────────────────────────────────────

@csrf_exempt
@require_POST
def filter_data(request):
    """兼容旧版简单筛选接口，内部调用高级筛选。"""
    body = json.loads(request.body)
    session_id = body.get('session_id')
    selected_cnam = body.get('selected_cnam', '')

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期，请重新上传'}, status=400)

    df = entry['df']
    meta = entry['meta']
    cnam_col = meta.get('cnam_col')
    cnrs_col = meta.get('cnrs_col')
    isrs_col = meta.get('isrs_col')

    if not cnam_col:
        return JsonResponse({'error': '缺少 cnam/snam 列', 'missing_columns': ['cnam']}, status=400)

    try:
        conditions = [{'col': cnam_col, 'op': 'eq', 'val': selected_cnam}]
        if cnrs_col:
            conditions.append({'col': cnrs_col, 'op': 'contains', 'val': '+'})
        if isrs_col:
            conditions.append({'col': isrs_col, 'op': 'contains', 'val': '+'})

        combined = None
        for c in conditions:
            cond = _build_condition(c['col'], c['op'], c['val'])
            combined = cond if combined is None else (combined & cond)

        filtered = df.filter(combined) if combined else df
    except Exception as e:
        return JsonResponse({'error': f'筛选失败: {e}'}, status=400)

    filtered_rows = len(filtered)
    meta['filtered_rows'] = filtered_rows
    entry['filtered_df'] = filtered

    preview_cols = [c for c in filtered.columns if not c.startswith('_')][:30]
    preview_rows = []
    for row in filtered.head(20).select(preview_cols).iter_rows(named=True):
        clean = {}
        for k, v in row.items():
            if v is None or (isinstance(v, float) and np.isnan(v)):
                clean[k] = None
            else:
                clean[k] = str(v)[:120]
        preview_rows.append(clean)

    return JsonResponse({
        'filtered_rows': filtered_rows,
        'preview_columns': preview_cols,
        'preview_rows': preview_rows,
        'total_before_filter': len(df),
    })


# ═══════════════════════════════════════════════════════════════════════════════
#  Step 3: 聚类 — IP 通信网络社区发现
# ═══════════════════════════════════════════════════════════════════════════════

@csrf_exempt
@require_POST
def run_clustering(request):
    """IP 通信网络社区发现聚类。

    构建 IP 通信图（节点=唯一IP，边=聚合通信记录），
    按 /24 子网前缀粗分组，子网内用 HDBSCAN (haversine) 二次聚类。
    """
    body = json.loads(request.body)
    session_id = body.get('session_id')
    min_cluster_size = int(body.get('min_cluster_size', 5))
    cluster_selection_epsilon = float(body.get('cluster_selection_epsilon', 0.05))

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期，请重新上传'}, status=400)

    filtered = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
    if filtered is None:
        return JsonResponse({'error': '请先执行筛选'}, status=400)

    meta = entry['meta']
    t_start = time.time()

    result_data = _cluster_network(entry, filtered, meta, min_cluster_size, cluster_selection_epsilon)

    elapsed = round(time.time() - t_start, 2)
    result_data['elapsed_seconds'] = elapsed
    _store(session_id, entry)
    return JsonResponse(result_data)


def _cluster_network(entry: dict, df: pl.DataFrame, meta: dict,
                      min_cluster_size: int, cluster_selection_epsilon: float) -> dict:
    """构建 IP 通信网络并进行社区发现聚类。

    流程：
      1. 从数据中提取所有唯一 IP（源+目的），去重
      2. 通过 GeoIP 获取每个 IP 的经纬度和属性
      3. 按 /24 子网前缀将 IP 分组
      4. 每个子网内，若节点数 > min_cluster_size，用 HDBSCAN (haversine) 二次聚类
      5. 不同子网的节点不会被合并
      6. 构建边列表（IP 对之间的聚合通信统计）
    """
    ips_col = meta.get('ips_col')
    ipd_col = meta.get('ipd_col')
    time_col = meta.get('time_col')
    bytes_col = meta.get('bytes_col')
    host_col = meta.get('host_col')
    snam_col = meta.get('snam_col')
    text_col = meta.get('text_col')
    orgn_col = meta.get('orgn_col')
    ispn_col = meta.get('ispn_col')
    city_col = meta.get('city_col')
    dport_col = meta.get('dport_col')
    dur_col = meta.get('dur_col')
    cnrs_col = meta.get('cnrs_col')
    tls_ver_col = meta.get('tls_ver_col')
    tls_cph_col = meta.get('tls_cph_col')
    cnam_col = meta.get('cnam_col')
    packets_col = meta.get('packets_col')
    geo_cache = entry.get('geo_cache', {})

    if not ips_col or not ipd_col:
        return {'error': '缺少源/目的IP列，无法构建通信网络'}
    if ips_col not in df.columns or ipd_col not in df.columns:
        return {'error': 'IP列不在数据中'}

    # ── 1. 提取唯一 IP 列表 ──────────────────────────────────────────────
    src_ips = df[ips_col].cast(pl.Utf8).to_list()
    dst_ips = df[ipd_col].cast(pl.Utf8).to_list()

    # Collect all unique IPs
    unique_ips_set = set()
    for ip in src_ips:
        if ip and ip not in ('None', '', '+', '-'):
            unique_ips_set.add(ip)
    for ip in dst_ips:
        if ip and ip not in ('None', '', '+', '-'):
            unique_ips_set.add(ip)

    unique_ips = sorted(unique_ips_set)
    n_nodes = len(unique_ips)
    ip_to_idx = {ip: i for i, ip in enumerate(unique_ips)}

    logger.info('Network clustering: %d unique IPs from %d records', n_nodes, len(df))

    # ── 2. GeoIP 解析所有 IP ───────────────────────────────────────────
    # MMDB优先（高精度），upload时的geo_cache兜底（低精度文本回退）
    node_geo = {}  # ip → {lat, lon, city, country, isp, org}
    for ip in unique_ips:
        info = geoip_lookup(ip)  # 优先用当前加载的MMDB数据库
        if not info:
            info = geo_cache.get(ip)  # 回退到上传时的缓存
        if info:
            node_geo[ip] = info

    geo_resolved = len(node_geo)
    logger.info('GeoIP resolved %d / %d IPs', geo_resolved, n_nodes)

    # ── 3. 按 /24 子网分组 ─────────────────────────────────────────────
    def _subnet24(ip: str) -> str | None:
        parts = ip.split('.')
        if len(parts) != 4:
            return None
        try:
            return '.'.join(parts[:3]) + '.0/24'
        except Exception:
            return None

    subnet_nodes: dict[str, list[str]] = {}
    nodes_without_geo = []
    for ip in unique_ips:
        sn = _subnet24(ip)
        if sn is None:
            nodes_without_geo.append(ip)
            continue
        if sn not in subnet_nodes:
            subnet_nodes[sn] = []
        subnet_nodes[sn].append(ip)

    # ── 4. 每个子网内 HDBSCAN 聚类 ───────────────────────────────────
    cluster_labels: dict[str, int] = {}  # ip → cluster_id
    cluster_id_counter = 0
    cluster_info: dict[int, dict] = {}   # cluster_id → {size, center_lat, center_lon, subnet, nodes}

    for sn, ips in subnet_nodes.items():
        # 收集该子网内有 GeoIP 坐标的节点
        geo_ips = []
        coords_list = []
        for ip in ips:
            info = node_geo.get(ip)
            if info and info.get('lat') is not None and info.get('lon') is not None:
                geo_ips.append(ip)
                coords_list.append([np.radians(float(info['lat'])), np.radians(float(info['lon']))])

        n_in_subnet = len(ips)
        n_with_geo = len(geo_ips)

        # 小规模子网 → 直接作为一个簇
        if n_in_subnet <= min_cluster_size or n_with_geo < 3:
            cid = cluster_id_counter
            cluster_id_counter += 1
            for ip in ips:
                cluster_labels[ip] = cid

            # 计算中心坐标
            center_lat, center_lon = None, None
            if coords_list:
                coords_arr = np.array(coords_list, dtype=np.float32)
                center_lat = round(float(np.degrees(np.median(coords_arr[:, 0]))), 4)
                center_lon = round(float(np.degrees(np.median(coords_arr[:, 1]))), 4)

            cluster_info[cid] = {
                'label': cid, 'size': n_in_subnet, 'n_geo': n_with_geo,
                'subnet': sn, 'center_lat': center_lat, 'center_lon': center_lon,
            }
            continue

        # 大规模子网 → HDBSCAN 二次聚类
        coords_arr = np.array(coords_list, dtype=np.float32)
        n_coords = len(coords_arr)

        # 去重坐标加微小抖动
        if n_coords > 10:
            _, unique_idx = np.unique(coords_arr, axis=0, return_index=True)
            if len(unique_idx) < n_coords * 0.8:
                jitter = np.random.RandomState(42).normal(0, 1e-6, coords_arr.shape).astype(np.float32)
                coords_arr = coords_arr + jitter

        try:
            sub_clusterer = HDBSCAN(
                min_cluster_size=max(2, min(n_coords // 20, min_cluster_size)),
                min_samples=2,
                metric='haversine',
                cluster_selection_epsilon=cluster_selection_epsilon,
            )
            sub_labels = sub_clusterer.fit_predict(coords_arr)
        except Exception as e:
            logger.warning('Subnet %s HDBSCAN failed: %s — treating as single cluster', sn, e)
            cid = cluster_id_counter
            cluster_id_counter += 1
            for ip in ips:
                cluster_labels[ip] = cid
            cluster_info[cid] = {
                'label': cid, 'size': n_in_subnet, 'n_geo': n_with_geo,
                'subnet': sn, 'center_lat': None, 'center_lon': None,
            }
            continue

        # 分配子标签（每个噪声节点独立成簇）
        unique_sl = set(sub_labels)
        for sl in unique_sl:
            mask = sub_labels == sl
            subset_ips = [geo_ips[i] for i in range(n_coords) if mask[i]]

            if sl == -1:
                # 无法归类的节点：每个单独成簇
                for ip in subset_ips:
                    cid = cluster_id_counter
                    cluster_id_counter += 1
                    cluster_labels[ip] = cid
                    cluster_info[cid] = {
                        'label': cid, 'size': 1, 'n_geo': 1,
                        'subnet': sn,
                        'center_lat': round(float(node_geo[ip].get('lat', 0)), 4) if ip in node_geo else None,
                        'center_lon': round(float(node_geo[ip].get('lon', 0)), 4) if ip in node_geo else None,
                    }
            else:
                cid = cluster_id_counter
                cluster_id_counter += 1
                for ip in subset_ips:
                    cluster_labels[ip] = cid

                sub_coords = coords_arr[mask]
                center_lat = round(float(np.degrees(np.median(sub_coords[:, 0]))), 4)
                center_lon = round(float(np.degrees(np.median(sub_coords[:, 1]))), 4)

                cluster_info[cid] = {
                    'label': cid, 'size': len(subset_ips), 'n_geo': len(subset_ips),
                    'subnet': sn, 'center_lat': center_lat, 'center_lon': center_lon,
                }

        # 子网内无 GeoIP 的节点 → 分配到子网的主簇，否则各自成簇
        non_geo_ips = [ip for ip in ips if ip not in geo_ips]
        if non_geo_ips:
            subnet_cids = [cid for cid, info in cluster_info.items()
                          if info.get('subnet') == sn]
            if subnet_cids:
                main_cid = min(subnet_cids, key=lambda c: cluster_info[c]['label'])
                for ip in non_geo_ips:
                    cluster_labels[ip] = main_cid
                cluster_info[main_cid]['size'] += len(non_geo_ips)
            else:
                for ip in non_geo_ips:
                    cid = cluster_id_counter
                    cluster_id_counter += 1
                    cluster_labels[ip] = cid
                    cluster_info[cid] = {'label': cid, 'size': 1, 'n_geo': 0, 'subnet': sn,
                                          'center_lat': None, 'center_lon': None}

    # 无子网的节点 → 各自成簇
    for ip in nodes_without_geo:
        cid = cluster_id_counter
        cluster_id_counter += 1
        cluster_labels[ip] = cid
        cluster_info[cid] = {'label': cid, 'size': 1, 'n_geo': 0, 'subnet': None,
                              'center_lat': None, 'center_lon': None}

    # ── 5. 构建节点列表 ──────────────────────────────────────────────
    # 检测服务器 IP（在 dst_ip 中出现过的 IP）
    server_ips = set()
    for ip in dst_ips:
        if ip and ip not in ('None', '', '+', '-'):
            server_ips.add(ip)

    # ── 收集 per-IP 元数据 (host, text, org, peers, TLS, traffic) ────
    ip_hosts: dict[str, str] = {}       # ip → 最常见的 host
    ip_texts: dict[str, list] = {}      # ip → 前几条 payload 样本
    ip_orgs: dict[str, str] = {}        # ip → 最常见的 org
    ip_peers: dict[str, list] = {}      # ip → 对端 IP 列表
    ip_snis: dict[str, list] = {}       # ip → SNI/域名列表
    ip_tls_vers: dict[str, list] = {}   # ip → TLS 版本列表
    ip_ciphers: dict[str, list] = {}    # ip → 加密套件列表
    ip_dports: dict[str, list] = {}     # ip → 目的端口列表
    ip_categories: dict[str, list] = {} # ip → 分类标签列表
    ip_total_bytes: dict[str, float] = {}    # ip → 总字节数
    ip_total_pkts: dict[str, int] = {}       # ip → 总包数
    ip_durations: dict[str, list] = {}       # ip → 持续时间列表
    ip_conn_ok: dict[str, int] = {}          # ip → 成功连接次数
    ip_conn_total: dict[str, int] = {}       # ip → 总连接次数
    ip_first_ts: dict[str, str] = {}         # ip → 首次出现时间
    ip_last_ts: dict[str, str] = {}          # ip → 最后出现时间

    scan_limit = min(len(df), 50000)
    # Pre-extract column lists for faster access
    _host_col = host_col or snam_col
    _has_host = _host_col and _host_col in df.columns
    _has_text = text_col and text_col in df.columns
    _has_orgn = orgn_col and orgn_col in df.columns
    _has_snam = snam_col and snam_col in df.columns
    _has_tls_ver = tls_ver_col and tls_ver_col in df.columns
    _has_tls_cph = tls_cph_col and tls_cph_col in df.columns
    _has_dport = dport_col and dport_col in df.columns
    _has_cnam = cnam_col and cnam_col in df.columns
    _has_bytes = bytes_col and bytes_col in df.columns
    _has_pkts = packets_col and packets_col in df.columns if packets_col else False
    _has_dur = dur_col and dur_col in df.columns
    _has_cnrs = cnrs_col and cnrs_col in df.columns
    _has_time = time_col and time_col in df.columns

    for i in range(scan_limit):
        sip = str(src_ips[i]) if i < len(src_ips) and src_ips[i] not in (None, '', 'None') else None
        if not sip or sip not in unique_ips_set:
            continue
        try:
            if _has_host:
                hval = str(df[_host_col][i]) if df[_host_col][i] is not None else None
                if hval and hval not in ('None', '', 'nan'):
                    ip_hosts.setdefault(sip, []).append(hval)
            if _has_text:
                tval = str(df[text_col][i]) if df[text_col][i] is not None else None
                if tval and tval not in ('None', '', 'nan'):
                    lst = ip_texts.setdefault(sip, [])
                    if len(lst) < 3:
                        lst.append(tval[:200])
            if _has_orgn:
                oval = str(df[orgn_col][i]) if df[orgn_col][i] is not None else None
                if oval and oval not in ('None', '', 'nan'):
                    ip_orgs.setdefault(sip, []).append(oval)
            if _has_snam:
                sval = str(df[snam_col][i]) if df[snam_col][i] is not None else None
                if sval and sval not in ('None', '', 'nan', '+', '-'):
                    ip_snis.setdefault(sip, []).append(sval)
            if _has_tls_ver:
                tv = str(df[tls_ver_col][i]) if df[tls_ver_col][i] is not None else None
                if tv and tv not in ('None', '', 'nan'):
                    ip_tls_vers.setdefault(sip, []).append(tv)
            if _has_tls_cph:
                tc = str(df[tls_cph_col][i]) if df[tls_cph_col][i] is not None else None
                if tc and tc not in ('None', '', 'nan'):
                    ip_ciphers.setdefault(sip, []).append(tc[:50])
            if _has_dport:
                dp = str(df[dport_col][i]) if df[dport_col][i] is not None else None
                if dp and dp not in ('None', '', 'nan'):
                    ip_dports.setdefault(sip, []).append(dp)
            if _has_cnam:
                cv = str(df[cnam_col][i]) if df[cnam_col][i] is not None else None
                if cv and cv not in ('None', '', 'nan', '+', '-'):
                    ip_categories.setdefault(sip, []).append(cv)
            if _has_bytes:
                try:
                    bv = float(df[bytes_col][i] or 0)
                    ip_total_bytes[sip] = ip_total_bytes.get(sip, 0.0) + bv
                except Exception:
                    pass
            if _has_pkts:
                try:
                    pv = int(df[packets_col][i] or 0)
                    ip_total_pkts[sip] = ip_total_pkts.get(sip, 0) + pv
                except Exception:
                    pass
            if _has_dur:
                try:
                    dv = float(df[dur_col][i] or 0)
                    if dv > 0:
                        ip_durations.setdefault(sip, []).append(dv)
                except Exception:
                    pass
            if _has_cnrs:
                cv = str(df[cnrs_col][i]) if df[cnrs_col][i] is not None else ''
                ip_conn_total[sip] = ip_conn_total.get(sip, 0) + 1
                if cv in ('+', '1', 'true', 'True', 'success'):
                    ip_conn_ok[sip] = ip_conn_ok.get(sip, 0) + 1
            if _has_time:
                tv = str(df[time_col][i]) if df[time_col][i] is not None else ''
                if tv and tv not in ('None', '', 'nan'):
                    if sip not in ip_first_ts or tv < ip_first_ts[sip]:
                        ip_first_ts[sip] = tv
                    if sip not in ip_last_ts or tv > ip_last_ts[sip]:
                        ip_last_ts[sip] = tv
        except Exception:
            pass

    # 取每 IP 最常见的值
    def _most_common(lst: list, top_n: int = 8) -> list:
        if not lst: return []
        cnt = Counter(lst)
        return [{'value': k, 'count': v} for k, v in cnt.most_common(top_n)]

    def _top1(lst: list) -> str:
        if not lst: return ''
        return Counter(lst).most_common(1)[0][0]

    for ip, hosts in ip_hosts.items():
        ip_hosts[ip] = _top1(hosts) if isinstance(hosts, list) else hosts
    for ip, orgs in ip_orgs.items():
        ip_orgs[ip] = _top1(orgs) if isinstance(orgs, list) else orgs

    # ── 5. 构建节点列表 ──────────────────────────────────────────────
    # Build peer IP lists from raw src/dst pairs
    for i in range(min(len(df), 50000)):
        sip = str(src_ips[i]) if i < len(src_ips) and src_ips[i] not in (None, '', 'None') else None
        dip = str(dst_ips[i]) if i < len(dst_ips) and dst_ips[i] not in (None, '', 'None') else None
        if sip and dip:
            ip_peers.setdefault(sip, set()).add(dip)
            ip_peers.setdefault(dip, set()).add(sip)

    nodes = []
    for ip in unique_ips:
        info = node_geo.get(ip, {})
        cid = cluster_labels.get(ip, -1)
        peer_set = ip_peers.get(ip, set())
        dur_list = ip_durations.get(ip, [])
        conn_total = ip_conn_total.get(ip, 0)
        conn_ok = ip_conn_ok.get(ip, 0)
        nodes.append({
            'ip': ip,
            'cluster_id': int(cid),
            'lat': info.get('lat'),
            'lon': info.get('lon'),
            'city': info.get('city', ''),
            'country': info.get('country', ''),
            'region': info.get('region', ''),
            'isp': info.get('isp', ''),
            'asn': info.get('asn', ''),
            'is_server': ip in server_ips,
            'host': ip_hosts.get(ip, info.get('host', '')),
            'org': ip_orgs.get(ip, info.get('org', '')),
            'texts': ip_texts.get(ip, [])[:3],
            'peer_ips': sorted(peer_set)[:20],
            'peer_count': len(peer_set),
            # Enriched attributes
            'top_snis': _most_common(ip_snis.get(ip, []), 5),
            'top_tls_vers': _most_common(ip_tls_vers.get(ip, []), 5),
            'top_ciphers': _most_common(ip_ciphers.get(ip, []), 5),
            'top_ports': _most_common(ip_dports.get(ip, []), 5),
            'top_categories': _most_common(ip_categories.get(ip, []), 5),
            'total_bytes': round(ip_total_bytes.get(ip, 0.0), 2),
            'total_packets': ip_total_pkts.get(ip, 0),
            'avg_duration': round(sum(dur_list) / len(dur_list), 2) if dur_list else None,
            'conn_success_rate': round(conn_ok / conn_total * 100, 1) if conn_total > 0 else None,
            'conn_count': conn_total,
            'first_seen': ip_first_ts.get(ip, ''),
            'last_seen': ip_last_ts.get(ip, ''),
        })

    # ── 6. 构建边列表（IP 对聚合通信统计） ──────────────────────────
    pair_data: dict[tuple[str, str], dict] = {}

    # Pre-extract column lists
    bytes_list = None
    if bytes_col and bytes_col in df.columns:
        try:
            bytes_list = df[bytes_col].cast(pl.Float64).to_list()
        except Exception:
            pass

    times_list = None
    if time_col and time_col in df.columns:
        try:
            times_list = df[time_col].cast(pl.Utf8).to_list()
        except Exception:
            pass

    pkts_list = None
    if packets_col and packets_col in df.columns:
        try:
            pkts_list = df[packets_col].cast(pl.Float64).to_list()
        except Exception:
            pass

    dur_list = None
    if dur_col and dur_col in df.columns:
        try:
            dur_list = df[dur_col].cast(pl.Float64).to_list()
        except Exception:
            pass

    dport_list = None
    if dport_col and dport_col in df.columns:
        try:
            dport_list = df[dport_col].cast(pl.Utf8).to_list()
        except Exception:
            pass

    tls_ver_list = None
    if tls_ver_col and tls_ver_col in df.columns:
        try:
            tls_ver_list = df[tls_ver_col].cast(pl.Utf8).to_list()
        except Exception:
            pass

    cph_list = None
    if tls_cph_col and tls_cph_col in df.columns:
        try:
            cph_list = df[tls_cph_col].cast(pl.Utf8).to_list()
        except Exception:
            pass

    snam_list = None
    if snam_col and snam_col in df.columns:
        try:
            snam_list = df[snam_col].cast(pl.Utf8).to_list()
        except Exception:
            pass

    cnam_list = None
    if cnam_col and cnam_col in df.columns:
        try:
            cnam_list = df[cnam_col].cast(pl.Utf8).to_list()
        except Exception:
            pass

    for i in range(len(df)):
        s = str(src_ips[i]) if src_ips[i] not in (None, '', 'None') else None
        d = str(dst_ips[i]) if dst_ips[i] not in (None, '', 'None') else None
        if s is None or d is None:
            continue
        key = (s, d)
        if key not in pair_data:
            pair_data[key] = {
                'count': 0, 'total_bytes': 0.0, 'total_packets': 0.0,
                'times': [], 'durations': [], 'ports': [],
                'tls_vers': [], 'ciphers': [], 'snis': [], 'categories': [],
            }
        pair_data[key]['count'] += 1
        if bytes_list and i < len(bytes_list):
            try:
                pair_data[key]['total_bytes'] += float(bytes_list[i] or 0)
            except (ValueError, TypeError):
                pass
        if times_list and i < len(times_list):
            t = times_list[i]
            if t and t not in ('None', ''):
                pair_data[key]['times'].append(t)
        if pkts_list and i < len(pkts_list):
            try:
                pair_data[key]['total_packets'] += float(pkts_list[i] or 0)
            except (ValueError, TypeError):
                pass
        if dur_list and i < len(dur_list):
            try:
                dv = float(dur_list[i] or 0)
                if dv > 0:
                    pair_data[key]['durations'].append(dv)
            except (ValueError, TypeError):
                pass
        if dport_list and i < len(dport_list):
            pv = dport_list[i]
            if pv and pv not in ('None', '', 'nan'):
                pair_data[key]['ports'].append(pv)
        if tls_ver_list and i < len(tls_ver_list):
            tv = tls_ver_list[i]
            if tv and tv not in ('None', '', 'nan'):
                pair_data[key]['tls_vers'].append(tv)
        if cph_list and i < len(cph_list):
            cv = cph_list[i]
            if cv and cv not in ('None', '', 'nan'):
                pair_data[key]['ciphers'].append(cv[:50])
        if snam_list and i < len(snam_list):
            sv = snam_list[i]
            if sv and sv not in ('None', '', 'nan', '+', '-'):
                pair_data[key]['snis'].append(sv)
        if cnam_list and i < len(cnam_list):
            catv = cnam_list[i]
            if catv and catv not in ('None', '', 'nan', '+', '-'):
                pair_data[key]['categories'].append(catv)

    # Build IP metadata lookup for edges
    ip_meta = {}
    for n in nodes:
        ip_meta[n['ip']] = n

    edges = []
    for (src, dst), data in pair_data.items():
        sm = ip_meta.get(src, {})
        dm = ip_meta.get(dst, {})
        durs = data.get('durations', [])
        edge = {
            'source': src,
            'target': dst,
            'comm_count': data['count'],
            'total_bytes': round(data['total_bytes'], 2),
            'total_packets': round(data.get('total_packets', 0), 2),
            'avg_duration': round(sum(durs) / len(durs), 2) if durs else None,
            'top_ports': _most_common(data.get('ports', []), 5),
            'top_tls_vers': _most_common(data.get('tls_vers', []), 5),
            'top_ciphers': _most_common(data.get('ciphers', []), 5),
            'top_snis': _most_common(data.get('snis', []), 5),
            'top_categories': _most_common(data.get('categories', []), 5),
            'src_host': sm.get('host', ''),
            'dst_host': dm.get('host', ''),
            'src_org': sm.get('org', ''),
            'dst_org': dm.get('org', ''),
            'src_city': sm.get('city', ''),
            'dst_city': dm.get('city', ''),
            'src_country': sm.get('country', ''),
            'dst_country': dm.get('country', ''),
            'src_cluster': sm.get('cluster_id', -1),
            'dst_cluster': dm.get('cluster_id', -1),
        }
        # 时间统计
        times = data.get('times', [])
        if times:
            times_sorted = sorted(times)
            edge['first_seen'] = times_sorted[0][:25]
            edge['last_seen'] = times_sorted[-1][:25]
            # 时间规律：按小时统计
            hour_counts = [0] * 24
            for t_str in times:
                try:
                    # Try ISO format or simple format
                    t_clean = t_str.strip()
                    if 'T' in t_clean or ' ' in t_clean:
                        # Extract hour
                        sep = 'T' if 'T' in t_clean else ' '
                        time_part = t_clean.split(sep)[-1]
                        hour = int(time_part.split(':')[0])
                        if 0 <= hour < 24:
                            hour_counts[hour] += 1
                except Exception:
                    pass
            peak_hours = sorted(range(24), key=lambda h: hour_counts[h], reverse=True)[:3]
            edge['peak_hours'] = ','.join(f'{h:02d}' for h in peak_hours)
            edge['time_count'] = len(times)
        else:
            edge['first_seen'] = ''
            edge['last_seen'] = ''
            edge['peak_hours'] = ''
            edge['time_count'] = 0
        edges.append(edge)

    # 按通信次数排序，限制返回前 500 条边
    edges.sort(key=lambda e: e['comm_count'], reverse=True)
    edges = edges[:500]

    # ── 7. 存储结果 ──────────────────────────────────────────────────
    n_clusters = len(cluster_info)

    entry['cluster_labels'] = cluster_labels  # ip → cluster_id
    entry['cluster_map'] = cluster_info
    entry['cluster_mode'] = 'network'
    entry['n_clusters'] = n_clusters
    entry['nodes'] = nodes
    entry['edges_data'] = edges
    entry['cluster_info'] = {'note': ''}

    return {
        'n_clusters': n_clusters,
        'n_nodes': n_nodes,
        'n_edges': len(edges),
        'geo_resolved': geo_resolved,
        'nodes': nodes,
        'edges': edges,
        'clusters': {str(k): v for k, v in cluster_info.items()},
        'note': f'{n_nodes} 个IP节点, {len(edges)} 条边, {n_clusters} 个社区',
    }


# ═══════════════════════════════════════════════════════════════════════════════
#  Step 4: Cluster detail + Node detail + 边关系
# ═══════════════════════════════════════════════════════════════════════════════

@require_GET
def cluster_detail(request, label: int):
    """返回指定簇的详细画像（节点列表、GeoIP 属性、流量特征等）。"""
    session_id = request.GET.get('session_id')
    if not session_id:
        return JsonResponse({'error': '缺少 session_id'}, status=400)

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期'}, status=400)

    nodes = entry.get('nodes', [])
    edges_data = entry.get('edges_data', [])
    cluster_info = entry.get('cluster_map', {}).get(label, {})

    if not nodes:
        # Old format fallback
        return _cluster_detail_legacy(entry, label)

    # Filter nodes by cluster
    cluster_nodes = [n for n in nodes if n.get('cluster_id') == label]
    if not cluster_nodes:
        return JsonResponse({'error': f'簇 {label} 无数据'}, status=404)

    # Collect IPs in this cluster
    cluster_ips = set(n['ip'] for n in cluster_nodes)

    # Find edges where both ends are in this cluster (internal)
    internal_edges = []
    related_edges = []
    for e in edges_data:
        src_in = e['source'] in cluster_ips
        dst_in = e['target'] in cluster_ips
        if src_in and dst_in:
            internal_edges.append(e)
        elif src_in or dst_in:
            related_edges.append(e)

    # Geo bounds
    lats = [n['lat'] for n in cluster_nodes if n.get('lat') is not None]
    lons = [n['lon'] for n in cluster_nodes if n.get('lon') is not None]
    geo_bounds = None
    if lats and lons:
        geo_bounds = {
            'lat_min': round(min(lats), 4), 'lat_max': round(max(lats), 4),
            'lon_min': round(min(lons), 4), 'lon_max': round(max(lons), 4),
        }

    # ISP / city / country distribution
    def _dist(items, key):
        counts = Counter(i.get(key, '') or '(未知)' for i in items)
        total = sum(counts.values()) or 1
        return [{'value': k, 'count': v, 'pct': round(v / total * 100, 1)}
                for k, v in counts.most_common(10)]

    # Total internal traffic
    total_internal_bytes = sum(e.get('total_bytes', 0) or 0 for e in internal_edges)
    total_related_bytes = sum(e.get('total_bytes', 0) or 0 for e in related_edges)

    profile = {
        'label': label,
        'size': len(cluster_nodes),
        'cluster_info': cluster_info,
        'unique_ips': sorted(cluster_ips)[:100],
        'unique_ips_count': len(cluster_ips),
        'ip_attributes': [{
            'ip': n['ip'], 'lat': n.get('lat'), 'lon': n.get('lon'),
            'city': n.get('city', ''), 'country': n.get('country', ''),
            'isp': n.get('isp', ''), 'is_server': n.get('is_server', False),
        } for n in cluster_nodes[:50]],
        'top_cities': _dist(cluster_nodes, 'city'),
        'top_countries': _dist(cluster_nodes, 'country'),
        'top_isps': _dist(cluster_nodes, 'isp'),
        'internal_edges': internal_edges[:100],
        'related_edges': related_edges[:100],
        'n_internal_edges': len(internal_edges),
        'n_related_edges': len(related_edges),
        'total_internal_bytes': round(total_internal_bytes, 2),
        'total_related_bytes': round(total_related_bytes, 2),
        'geo_bounds': geo_bounds,
    }

    return JsonResponse(profile)


def _cluster_detail_legacy(entry: dict, label: int) -> JsonResponse:
    """Fallback cluster detail for old record-based clustering format."""
    df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
    meta = entry.get('meta', {})
    cluster_map = entry.get('cluster_map', {})
    geo_cache = entry.get('geo_cache', {})

    if df is None:
        return JsonResponse({'error': '数据未就绪'}, status=400)
    if '_cluster_label' not in df.columns:
        return JsonResponse({'error': '未执行聚类'}, status=400)

    cluster_df = df.filter(pl.col('_cluster_label') == label)
    rows = len(cluster_df)
    if rows == 0:
        return JsonResponse({'error': f'簇 {label} 无数据'}, status=404)

    cluster_info = cluster_map.get(label, {})

    def _top_values(col_name: str, top_n: int = 5) -> list:
        if not col_name or col_name not in df.columns:
            return []
        try:
            vals = (
                cluster_df[col_name].cast(pl.Utf8).fill_null('(空)')
                .value_counts().sort('count', descending=True).head(top_n)
            )
            total = vals['count'].sum()
            result = []
            for row in vals.iter_rows():
                result.append({
                    'value': str(row[0])[:80],
                    'count': int(row[1]),
                    'pct': round(int(row[1]) / total * 100, 1) if total else 0,
                })
            return result
        except Exception:
            return []

    ips_col = meta.get('ips_col')
    ipd_col = meta.get('ipd_col')
    dport_col = meta.get('dport_col')
    tls_ver_col = meta.get('tls_ver_col')
    snam_col = meta.get('snam_col')
    ispn_col = meta.get('ispn')
    orgn_col = meta.get('orgn')
    city_col = meta.get('city')
    time_col = meta.get('time_col')
    bytes_col = meta.get('bytes_col')

    unique_ips = []
    if ips_col and ips_col in cluster_df.columns:
        try:
            ip_vals = cluster_df[ips_col].cast(pl.Utf8).unique().to_list()
            unique_ips = sorted(str(v) for v in ip_vals if v not in (None, '', 'None'))
        except Exception:
            pass

    ip_attrs = []
    for ip in unique_ips[:50]:
        info = geo_cache.get(ip) or geoip_lookup(ip) or {}
        ip_attrs.append({
            'ip': ip,
            'lat': info.get('lat'), 'lon': info.get('lon'),
            'city': info.get('city', ''), 'country': info.get('country', ''),
        })

    profile = {
        'label': label, 'size': rows,
        'cluster_info': cluster_info,
        'unique_ips': unique_ips[:100],
        'unique_ips_count': len(unique_ips),
        'ip_attributes': ip_attrs,
        'top_dst_ips': _top_values(ipd_col) if ipd_col else [],
        'top_dst_ports': _top_values(dport_col) if dport_col else [],
        'top_tls_versions': _top_values(tls_ver_col) if tls_ver_col else [],
        'top_sni': _top_values(snam_col) if snam_col else [],
        'top_isp': _top_values(ispn_col) if ispn_col else [],
        'top_org': _top_values(orgn_col) if orgn_col else [],
        'top_cities': _top_values(city_col) if city_col else [],
        'traffic': {},
    }

    if bytes_col and bytes_col in cluster_df.columns:
        try:
            profile['traffic']['total_bytes'] = round(float(cluster_df[bytes_col].cast(pl.Float64).sum()), 2)
        except Exception:
            pass

    if time_col and time_col in cluster_df.columns:
        try:
            times = cluster_df[time_col].cast(pl.Utf8).drop_nulls()
            if len(times) > 0:
                profile['time_range'] = {'first': str(times.min())[:25], 'last': str(times.max())[:25]}
        except Exception:
            pass

    return JsonResponse(profile)


@require_GET
def node_detail(request):
    """返回单个 IP 节点的详细属性及其关联边。"""
    session_id = request.GET.get('session_id')
    ip = request.GET.get('ip', '')
    if not session_id or not ip:
        return JsonResponse({'error': '缺少 session_id 或 ip'}, status=400)

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期'}, status=400)

    nodes = entry.get('nodes', [])
    edges_data = entry.get('edges_data', [])

    # Find the node
    node = next((n for n in nodes if n.get('ip') == ip), None)
    if not node:
        # Fallback: try GeoIP direct lookup
        info = geoip_lookup(ip)
        if info:
            node = {
                'ip': ip,
                'cluster_id': -1,
                'lat': info.get('lat'), 'lon': info.get('lon'),
                'city': info.get('city', ''), 'country': info.get('country', ''),
                'isp': info.get('isp', ''), 'is_server': False,
            }
        else:
            return JsonResponse({'error': f'IP {ip} 不在网络中'}, status=404)

    # Find related edges
    related = [e for e in edges_data if e.get('source') == ip or e.get('target') == ip]
    related.sort(key=lambda e: e.get('comm_count', 0), reverse=True)

    # Peer IPs with metadata
    peers = set()
    for e in related:
        if e.get('source') == ip:
            peers.add(e.get('target'))
        else:
            peers.add(e.get('source'))

    # Enrich related edges with peer metadata
    enriched_edges = []
    for e in related[:100]:
        ee = dict(e)
        if e.get('source') == ip:
            ee['peer_ip'] = e.get('target', '')
            ee['peer_host'] = e.get('dst_host', '')
            ee['peer_org'] = e.get('dst_org', '')
            ee['peer_city'] = e.get('dst_city', '')
        else:
            ee['peer_ip'] = e.get('source', '')
            ee['peer_host'] = e.get('src_host', '')
            ee['peer_org'] = e.get('src_org', '')
            ee['peer_city'] = e.get('src_city', '')
        enriched_edges.append(ee)

    total_bytes = sum(e.get('total_bytes', 0) or 0 for e in related)

    return JsonResponse({
        'node': node,
        'node_texts': node.get('texts', []),
        'related_edges': enriched_edges,
        'n_edges': len(related),
        'peer_ips': sorted(peers)[:100],
        'n_peers': len(peers),
        'total_flow_bytes': round(total_bytes, 2),
    })


@csrf_exempt
@require_POST
def edge_relationships(request):
    """返回边关系数据（使用预计算的边列表）。"""
    body = json.loads(request.body)
    session_id = body.get('session_id')
    min_freq = int(body.get('min_freq', 1))

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期'}, status=400)

    edges_data = entry.get('edges_data', [])

    if not edges_data:
        # Fallback: compute on-the-fly from old format
        return _edge_relationships_legacy(entry, body)

    # Filter by min frequency
    filtered = [e for e in edges_data if e.get('comm_count', 0) >= min_freq]
    filtered.sort(key=lambda e: e.get('comm_count', 0), reverse=True)

    return JsonResponse({
        'edges': filtered[:500],
        'total_edges': len(filtered),
        'min_freq': min_freq,
    })


def _edge_relationships_legacy(entry: dict, body: dict) -> JsonResponse:
    """Fallback edge computation for old record-based format."""
    min_freq = int(body.get('min_freq', 10))
    df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
    meta = entry.get('meta', {})
    if df is None:
        return JsonResponse({'error': '数据未就绪'}, status=400)

    ips_col = meta.get('ips_col')
    ipd_col = meta.get('ipd_col')
    time_col = meta.get('time_col')
    bytes_col = meta.get('bytes_col')

    if not ips_col or not ipd_col:
        return JsonResponse({'error': '缺少源/目的IP列'}, status=400)
    if ips_col not in df.columns or ipd_col not in df.columns:
        return JsonResponse({'error': 'IP列不在数据中'}, status=400)

    try:
        src_ips = df[ips_col].cast(pl.Utf8).to_list()
        dst_ips = df[ipd_col].cast(pl.Utf8).to_list()
        bytes_list = df[bytes_col].cast(pl.Float64).to_list() if bytes_col and bytes_col in df.columns else None
        times_list = df[time_col].cast(pl.Utf8).to_list() if time_col and time_col in df.columns else None

        pair_counts: dict[tuple[str, str], int] = {}
        pair_bytes: dict[tuple[str, str], float] = {}
        pair_times: dict[tuple[str, str], list[str]] = {}

        for i in range(len(df)):
            s = str(src_ips[i]) if src_ips[i] not in (None, '', 'None') else '?'
            d = str(dst_ips[i]) if dst_ips[i] not in (None, '', 'None') else '?'
            key = (s, d)
            pair_counts[key] = pair_counts.get(key, 0) + 1
            if bytes_list and i < len(bytes_list):
                try:
                    pair_bytes[key] = pair_bytes.get(key, 0.0) + float(bytes_list[i] or 0)
                except (ValueError, TypeError):
                    pass
            if times_list and i < len(times_list):
                v = times_list[i]
                if v and v not in ('None', ''):
                    pair_times.setdefault(key, []).append(v)

        edges = []
        for (src, dst), count in pair_counts.items():
            if count < min_freq:
                continue
            edge = {
                'source': src, 'target': dst,
                'comm_count': count,
                'total_bytes': round(pair_bytes.get((src, dst), 0), 2),
            }
            times = pair_times.get((src, dst), [])
            if times:
                times_sorted = sorted(times)
                edge['first_seen'] = times_sorted[0][:25]
                edge['last_seen'] = times_sorted[-1][:25]
                edge['time_count'] = len(times)
            edges.append(edge)

        edges.sort(key=lambda e: e['comm_count'], reverse=True)
        edges = edges[:200]

        return JsonResponse({
            'edges': edges,
            'total_pairs': len(pair_counts),
            'total_edges': len(edges),
            'min_freq': min_freq,
        })
    except Exception as e:
        return JsonResponse({'error': f'边关系计算失败: {e}'}, status=400)


@csrf_exempt
@require_POST
def edge_time_series(request):
    """返回指定边的时间序列数据（按时间桶聚合），用于绘制流量-时间图。

    POST body: session_id, source, target, bucket (可选: 'auto'|'hour'|'minute'|'day')
    返回: {time_series: [{t: "yyyy-mm-dd HH:MM (北京)", bytes: N, count: N}, ...], span_hours: N}
    """
    body = json.loads(request.body)
    session_id = body.get('session_id')
    source = body.get('source', '')
    target = body.get('target', '')
    bucket = body.get('bucket', 'auto')

    if not session_id or not source or not target:
        return JsonResponse({'error': '缺少 session_id / source / target'}, status=400)

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期'}, status=400)

    df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
    if df is None:
        return JsonResponse({'error': '数据未就绪'}, status=400)

    meta = entry.get('meta', {})
    ips_col = meta.get('ips_col')
    ipd_col = meta.get('ipd_col')
    time_col = meta.get('time_col')
    bytes_col = meta.get('bytes_col')

    if not ips_col or not ipd_col:
        return JsonResponse({'error': '缺少源/目的IP列'}, status=400)
    if not time_col or time_col not in df.columns:
        return JsonResponse({'error': '缺少时间列，无法绘制时间序列'}, status=400)

    # ── 1. 筛选该边的所有行 ──
    try:
        mask = (
            (df[ips_col].cast(pl.Utf8) == source) &
            (df[ipd_col].cast(pl.Utf8) == target)
        )
        edge_rows = df.filter(mask)
    except Exception as e:
        return JsonResponse({'error': f'数据筛选失败: {e}'}, status=400)

    n_rows = len(edge_rows)
    if n_rows == 0:
        return JsonResponse({'error': '未找到该边的原始记录'}, status=400)

    # ── 2. 提取时间列并转为 datetime ──
    try:
        times_series = edge_rows[time_col].cast(pl.Utf8)
    except Exception:
        return JsonResponse({'error': '时间列读取失败'}, status=400)

    # 提取 bytes（如果有的话）
    has_bytes = bytes_col and bytes_col in df.columns
    bytes_series = None
    if has_bytes:
        try:
            bytes_series = edge_rows[bytes_col].cast(pl.Float64)
        except Exception:
            has_bytes = False

    # ── 3. 解析时间戳为 datetime ──
    raw_times = times_series.to_list()
    raw_bytes = bytes_series.to_list() if has_bytes else []
    parsed_times = []
    for row_idx, t_str in enumerate(raw_times):
        try:
            t_clean = str(t_str).strip() if t_str else ''
            if not t_clean or t_clean in ('None', '', 'nan'):
                continue
            # 尝试多种格式
            parsed = None
            for fmt in [
                '%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S',
                '%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S.%f',
                '%Y/%m/%d %H:%M:%S', '%Y/%m/%dT%H:%M:%S',
                '%d/%m/%Y %H:%M:%S', '%m/%d/%Y %H:%M:%S',
            ]:
                try:
                    parsed = time.strptime(t_clean[:26], fmt)
                    break
                except ValueError:
                    continue
            if parsed is None:
                # Fallback: try to extract Y-M-D H:M:S via regex
                m = re.search(r'(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2}):(\d{2})', t_clean)
                if m:
                    parsed = time.strptime(f"{m.group(1)}-{m.group(2)}-{m.group(3)} "
                                           f"{m.group(4)}:{m.group(5)}:{m.group(6)}",
                                           '%Y-%m-%d %H:%M:%S')
            if parsed is None:
                continue
            # 转为北京时间 (UTC+8)
            import calendar
            ts_utc = calendar.timegm(parsed)
            ts_bj = ts_utc + 8 * 3600  # 北京时间 = UTC+8
            parsed_times.append((ts_bj, row_idx))
        except Exception:
            continue

    if not parsed_times:
        return JsonResponse({'error': '无法解析时间戳'}, status=400)

    # ── 4. 确定时间桶粒度 ──
    min_ts = min(t[0] for t in parsed_times)
    max_ts = max(t[0] for t in parsed_times)
    span_seconds = max_ts - min_ts
    span_hours = span_seconds / 3600.0

    if bucket == 'auto':
        if span_hours <= 2:
            bucket_seconds = 60       # 1 分钟桶
        elif span_hours <= 24:
            bucket_seconds = 300      # 5 分钟桶
        elif span_hours <= 72:
            bucket_seconds = 900      # 15 分钟桶
        elif span_hours <= 168:
            bucket_seconds = 3600     # 1 小时桶
        else:
            bucket_seconds = 14400    # 4 小时桶
    elif bucket == 'minute':
        bucket_seconds = 60
    elif bucket == 'hour':
        bucket_seconds = 3600
    elif bucket == 'day':
        bucket_seconds = 86400
    else:
        bucket_seconds = 300

    # ── 5. 按时间桶聚合 ──
    bucket_data: dict[int, dict] = {}  # bucket_start_ts → {count, bytes}
    for ts_bj, idx in parsed_times:
        bucket_key = int(ts_bj // bucket_seconds) * bucket_seconds
        if bucket_key not in bucket_data:
            bucket_data[bucket_key] = {'count': 0, 'bytes': 0.0}
        bucket_data[bucket_key]['count'] += 1
        if has_bytes and idx < len(raw_bytes):
            try:
                bval = raw_bytes[idx]
                if bval is not None:
                    bucket_data[bucket_key]['bytes'] += float(bval)
            except Exception:
                pass

    # ── 6. 构建输出 ──
    def fmt_bj(ts: int, bucket_sec: int) -> str:
        """格式化北京时间字符串。"""
        tm = time.gmtime(ts)
        if bucket_sec >= 86400:
            return time.strftime('%m-%d', tm)
        elif bucket_sec >= 3600:
            return time.strftime('%m-%d %H:00', tm)
        elif bucket_sec >= 60:
            return time.strftime('%m-%d %H:%M', tm)
        else:
            return time.strftime('%H:%M:%S', tm)

    time_series = []
    for ts_key in sorted(bucket_data.keys()):
        d = bucket_data[ts_key]
        time_series.append({
            't': fmt_bj(ts_key, bucket_seconds),
            'ts': ts_key,
            'count': d['count'],
            'bytes': round(d['bytes'], 2),
        })

    return JsonResponse({
        'time_series': time_series,
        'span_hours': round(span_hours, 2),
        'bucket_minutes': bucket_seconds // 60,
        'n_raw': len(parsed_times),
        'has_bytes': has_bytes,
    })


@csrf_exempt
@require_POST
def cluster_nodes_edges(request):
    """返回指定簇的所有节点、内部边、关联外部边及邻接节点。

    POST body: {session_id, cluster_label}
    返回: {nodes, internal_edges, external_edges, adjacent_nodes, geo_bounds, cluster_info}
    """
    body = json.loads(request.body)
    session_id = body.get('session_id')
    cluster_label = body.get('cluster_label')

    if not session_id or cluster_label is None:
        return JsonResponse({'error': '缺少 session_id / cluster_label'}, status=400)

    entry = _cached(session_id)
    if entry is None:
        return JsonResponse({'error': '会话已过期'}, status=400)

    nodes = entry.get('nodes', [])
    edges_data = entry.get('edges_data', [])
    cluster_info = entry.get('cluster_map', {}).get(str(cluster_label), {})

    if not nodes:
        return JsonResponse({'error': '无节点数据'}, status=400)

    # ── 1. 筛选簇内节点 ──
    cluster_nodes = [n for n in nodes if n.get('cluster_id') == cluster_label]
    if not cluster_nodes:
        return JsonResponse({'error': f'簇 {cluster_label} 无数据'}, status=404)

    cluster_ips = set(n['ip'] for n in cluster_nodes)

    # ── 2. 查找所有关联边（至少一端在簇内） ──
    internal_edges = []
    external_edges = []
    adjacent_ip_set = set()

    for e in edges_data:
        src_in = e['source'] in cluster_ips
        dst_in = e['target'] in cluster_ips
        if src_in and dst_in:
            internal_edges.append(e)
        elif src_in:
            external_edges.append(e)
            adjacent_ip_set.add(e['target'])
        elif dst_in:
            external_edges.append(e)
            adjacent_ip_set.add(e['source'])

    # ── 3. 查找邻接节点 ──
    adjacent_nodes = [n for n in nodes if n['ip'] in adjacent_ip_set]

    # ── 4. 计算 Geo bounds ──
    lats = [n.get('lat') for n in cluster_nodes if n.get('lat') is not None]
    lons = [n.get('lon') for n in cluster_nodes if n.get('lon') is not None]
    lats += [n.get('lat') for n in adjacent_nodes if n.get('lat') is not None]
    lons += [n.get('lon') for n in adjacent_nodes if n.get('lon') is not None]

    geo_bounds = None
    if lats and lons:
        geo_bounds = {
            'lat_min': round(min(lats), 4), 'lat_max': round(max(lats), 4),
            'lon_min': round(min(lons), 4), 'lon_max': round(max(lons), 4),
        }

    # ── 5. 簇级别统计 ──
    def _dist(items, key):
        counts = Counter(i.get(key, '') or '(未知)' for i in items)
        total = sum(counts.values()) or 1
        return [{'value': k, 'count': v, 'pct': round(v / total * 100, 1)}
                for k, v in counts.most_common(10)]

    total_internal_bytes = sum(e.get('total_bytes', 0) or 0 for e in internal_edges)
    total_external_bytes = sum(e.get('total_bytes', 0) or 0 for e in external_edges)

    return JsonResponse({
        'cluster_label': cluster_label,
        'cluster_info': cluster_info,
        'nodes': cluster_nodes,
        'n_nodes': len(cluster_nodes),
        'internal_edges': internal_edges,
        'external_edges': external_edges,
        'n_internal_edges': len(internal_edges),
        'n_external_edges': len(external_edges),
        'adjacent_nodes': adjacent_nodes,
        'n_adjacent_nodes': len(adjacent_nodes),
        'geo_bounds': geo_bounds,
        'top_isps': _dist(cluster_nodes, 'isp'),
        'top_cities': _dist(cluster_nodes, 'city'),
        'top_countries': _dist(cluster_nodes, 'country'),
        'total_internal_bytes': round(total_internal_bytes, 2),
        'total_external_bytes': round(total_external_bytes, 2),
    })


@require_GET
def column_values(request):
    """返回指定列的唯一值 Top-50（供筛选器下拉建议）。

    支持两种模式：
      - session_id：从已上传的数据中查（全量，精确 Top-K）
      - scan_id：从已保存的文件中 streaming 单列扫描（低内存，仅读一列）
    """
    session_id = request.GET.get('session_id', '')
    scan_id = request.GET.get('scan_id', '')
    col_name = request.GET.get('col', '')

    if not col_name:
        return JsonResponse({'error': '缺少 col'}, status=400)

    # ── Mode 1: session_id (full data in memory) ──────────────────────
    if session_id:
        entry = _cached(session_id)
        if entry is None:
            return JsonResponse({'error': '会话已过期'}, status=400)
        df = entry.get('df')
        if df is None or col_name not in df.columns:
            return JsonResponse({'error': '列不存在'}, status=400)
        try:
            values = (
                df[col_name]
                .cast(pl.Utf8)
                .fill_null('(空)')
                .value_counts(sort=True)
                .head(50)
            )
            result = [
                {'value': str(row[0])[:80], 'count': int(row[1])}
                for row in values.iter_rows()
            ]
            return JsonResponse({'col': col_name, 'values': result, 'total': len(result)})
        except Exception as e:
            return JsonResponse({'error': str(e)}, status=400)

    # ── Mode 2: scan_id (streaming single-column scan) ────────────────
    if scan_id:
        entry = _cached(scan_id)
        if entry is None:
            return JsonResponse({'error': '扫描会话已过期，请重新选择文件'}, status=400)

        scan_paths = entry.get('scan_paths', [])
        col_map = entry.get('col_map', {})

        if not scan_paths:
            return JsonResponse({'error': '扫描文件已清理，请重新选择文件'}, status=400)

        # Resolve the original column name from col_map
        # col_map is {original → std}, we need original for the file
        original_col = col_name
        for orig, std in col_map.items():
            if std == col_name:
                original_col = orig
                break

        try:
            # Streaming scan — only reads ONE column, O(1 col) memory
            lf = pl.scan_csv(scan_paths[0], infer_schema_length=0)
            if len(scan_paths) > 1:
                frames = [pl.scan_csv(p, infer_schema_length=0) for p in scan_paths]
                lf = pl.concat(frames, how='diagonal_relaxed')

            # Select only the target column (by original name)
            if original_col in lf.columns:
                col_expr = pl.col(original_col)
            elif col_name in lf.columns:
                col_expr = pl.col(col_name)
            else:
                return JsonResponse({'error': f'列 "{col_name}" 不在文件中'}, status=400)

            # Streaming Top-50 value counts
            values = (
                lf.select(col_expr.cast(pl.Utf8).fill_null('(空)').alias('_val'))
                .collect(streaming=True)
                .to_series()
                .value_counts(sort=True)
                .head(50)
            )

            result = [
                {'value': str(row[0])[:80], 'count': int(row[1])}
                for row in values.iter_rows()
            ]
            return JsonResponse({'col': col_name, 'values': result, 'total': len(result)})
        except Exception as e:
            return JsonResponse({'error': f'列扫描失败: {e}'}, status=400)

    return JsonResponse({'error': '缺少 session_id 或 scan_id'}, status=400)


@require_GET
def geoip_status(request):
    """返回当前 GeoIP 数据库状态。"""
    from analysis import geoip
    return JsonResponse({
        'available': geoip.is_available(),
        'source': geoip.source_name(),
        'using_mmdb': 'mmdb' in geoip.source_name().lower(),
    })


# ── Main page ────────────────────────────────────────────────────────────────

def index(request):
    """Render the simple analysis workbench page."""
    return render(request, 'simple_analysis/simple_analysis.html')