From 55f3b3637a345f6449e4c4db2dd41ec89ab36bfb Mon Sep 17 00:00:00 2001 From: TianXuan Developer Date: Thu, 23 Jul 2026 11:39:54 +0800 Subject: [PATCH] feat: offline GeoIP, advanced filter, performance tuning, enriched viz Major v2 improvements to simple analysis module: Backend: - Column name normalization with 50+ column alias mapping - Offline GeoIP integration (reuses analysis.geoip module, no dependency on CSV lat/lon columns) - Advanced filter builder with AND/OR/NOT conditions - Edge relationship computation (IP pair frequency) - HDBSCAN with GeoIP-derived coordinates + float32 memory opt - DBSCAN fallback when HDBSCAN produces all-noise - Jitter for duplicate GeoIP coordinates (city-level overlap) - All-noise graceful handling (visualize all points as noise) - Cluster detail with GeoIP city/ISP/org attributes Deps: geoip2==5.3.0, maxminddb==3.1.1 added to requirements.txt Co-Authored-By: Claude --- requirements.txt | Bin 2404 -> 2470 bytes simple_analysis/urls.py | 2 + simple_analysis/views.py | 1275 ++++++++++++++++++++++++++------------ 3 files changed, 872 insertions(+), 405 deletions(-) diff --git a/requirements.txt b/requirements.txt index 4bd5c7981d3f406b327af605ad7bae9d0b6a8f42..0f313fe2341484c2eb44b91061fc85cc1beec13d 100644 GIT binary patch delta 74 zcmaDNv`lzI3a3*#Ln=c)LncE3gAs!*5SlXRF&G1}0Rt}s7eg*XB0~j`%w))8NCCnm Nuo{rOAsB/', views.cluster_detail, name='cluster_detail'), + path('edges/', views.edge_relationships, name='edges'), ] diff --git a/simple_analysis/views.py b/simple_analysis/views.py index 5d999a6..2627e3f 100644 --- a/simple_analysis/views.py +++ b/simple_analysis/views.py @@ -1,7 +1,11 @@ -"""简单分析模块 — 独立的上传→筛选→聚类→地图可视化工作流。 +"""简单分析模块 v2 — 离线 GeoIP + 高级筛选 + 高性能聚类 + 丰富可视化。 -不与天璇原有分析模块共享数据库或 session store，所有中间数据 -存储在模块级内存缓存（SimpleAnalysisCache）或临时目录中。 +改进： +- 使用离线 GeoIP 数据库（analysis.geoip）解析 IP 经纬度，抛弃 CSV 自带列 +- 列名统一规范化，内置 50+ 列映射字典 +- 上传预设筛选 + 高级 AND/OR/NOT 筛选 +- HDBSCAN 性能优化（float32、降采样、DBSCAN 回退） +- 全噪声时可视化、点属性、边关系 """ import json @@ -9,7 +13,7 @@ import logging import os import re import shutil -import tempfile +import time import uuid from pathlib import Path from collections import Counter @@ -21,13 +25,15 @@ 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 +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 ────────────────────────────────────────────────────────── -# Thread-safe for single-user desktop usage; each user session gets a unique key. _cache: dict[str, dict] = {} _TEMP_DIR = Path(settings.FILE_UPLOAD_TEMP_DIR) / 'simple_analysis' _TEMP_DIR.mkdir(parents=True, exist_ok=True) @@ -45,30 +51,165 @@ def _store(key: str, data: dict) -> None: _cache[key] = data -# ── Column-name helpers ────────────────────────────────────────────────────── +# ═══════════════════════════════════════════════════════════════════════════════ +# 列名规范化 — 完整映射字典 +# ═══════════════════════════════════════════════════════════════════════════════ + +# 标准内部名称 → 可能的 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'], +} + def _norm(name: str) -> str: - """Normalise column name: lower-case, strip whitespace, replace hyphens.""" - return name.strip().lower().replace('-', '_').replace(' ', '_') + """归一化列名：去首尾空格，特殊字符→下划线，转小写。""" + n = name.strip().lower() + n = re.sub(r'[-:.\s]+', '_', n) + return n -def _find_col(col_names: list[str], *patterns: str) -> str | None: - """Find first column matching any pattern (exact or suffix after stripping).""" - normed = {c: _norm(c) for c in col_names} - for p in patterns: - np_norm = _norm(p) - for orig, n in normed.items(): - if n == np_norm or n.endswith('_' + np_norm) or n.startswith(np_norm + '_'): +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 -# ── Step 1: Upload ─────────────────────────────────────────────────────────── +# ═══════════════════════════════════════════════════════════════════════════════ +# 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 1: Upload + 预设筛选 + GeoIP 解析 +# ═══════════════════════════════════════════════════════════════════════════════ @csrf_exempt @require_POST def upload_csv(request): - """Upload one or more CSV files, merge, return column info + cnam frequency.""" + """上传 CSV，列归一化，预设筛选，GeoIP 解析，返回摘要。""" files = request.FILES.getlist('files') if not files: return JsonResponse({'error': '未上传任何文件'}, status=400) @@ -85,7 +226,16 @@ def upload_csv(request): out.write(chunk) saved_paths.append(str(dest)) - # Load & merge CSV files using Polars + # 解析预设筛选条件（来自 POST 数据或查询参数） + preset_filters = {} + try: + pf = request.POST.get('preset_filters', request.GET.get('preset_filters', '{}')) + if pf: + preset_filters = json.loads(pf) + 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'), @@ -95,27 +245,85 @@ def upload_csv(request): 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) - col_names = df.columns + # 列名规范化 + df, col_map = normalize_columns(df) total_rows = len(df) - # Detect key columns — expanded patterns to match actual CSV column names - cnam_col = _find_col(col_names, 'cnam', 'snam', 'class_name', 'category', 'target', 'name') - cnrs_col = _find_col(col_names, 'cnrs', 'conn_success', 'result', '4srs') - isrs_col = _find_col(col_names, 'isrs', 'is_resolved', 'resolved') - src_lat_col = _find_col(col_names, ':ips.latd', 'src_latitude', 'src_lat', 'source_lat') - src_lon_col = _find_col(col_names, ':ips.lond', 'src_longitude', 'src_lon', 'source_lon') - dst_lat_col = _find_col(col_names, ':ipd.latd', 'dst_latitude', 'dst_lat', 'dest_lat') - dst_lon_col = _find_col(col_names, ':ipd.lond', 'dst_longitude', 'dst_lon', 'dest_lon') - ips_col = _find_col(col_names, ':ips', 'src_ip', 'source_ip', 'ip_src') - ipd_col = _find_col(col_names, ':ipd', 'dst_ip', 'dest_ip', 'destination_ip', 'ip_dst') + # 获取各关键列的 标准内部名（即重命名后的列名） + # 注意：col_map 是 {原始名→标准内部名}，标准名即为 df 当前使用的列名 + standard_cols = set(col_map.values()) + def _get_std(*names): + """从 col_map 中查找第一个匹配的标准名。""" + for n in names: + for orig, std in col_map.items(): + if std == n: + return std + return None - # CNAM frequency + 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') + 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') + + # ── 预设筛选 ──────────────────────────────────────────────────────────── + preset_filtered = total_rows + try: + cond = None + if preset_filters.get('cnrs') in ('+', '-'): + op = '==' if preset_filters['cnrs'] == '+' else '!=' + if cnrs_col: + c = (pl.col(cnrs_col).cast(pl.Utf8).str.strip_chars() == '+') + if preset_filters['cnrs'] == '-': + c = ~c + cond = c if cond is None else (cond & c) + if preset_filters.get('isrs') in ('+', '-'): + if isrs_col: + c = (pl.col(isrs_col).cast(pl.Utf8).str.strip_chars() == '+') + if preset_filters['isrs'] == '-': + c = ~c + cond = c if cond is None else (cond & c) + if preset_filters.get('cnam'): + if cnam_col: + c = (pl.col(cnam_col).cast(pl.Utf8) == preset_filters['cnam']) + cond = c if cond is None else (cond & c) + if preset_filters.get('snam'): + if snam_col: + c = (pl.col(snam_col).cast(pl.Utf8) == preset_filters['snam']) + cond = c if cond is None else (cond & c) + 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: @@ -134,92 +342,154 @@ def upload_csv(request): except Exception as e: logger.warning('cnam frequency failed: %s', e) - # Store metadata in cache + # 存储元数据 meta = { 'session_id': session_id, - 'upload_dir': str(upload_dir), 'total_rows': total_rows, - 'col_names': col_names, + '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, - 'src_lat_col': src_lat_col, - 'src_lon_col': src_lon_col, - 'dst_lat_col': dst_lat_col, - 'dst_lon_col': dst_lon_col, - 'ips_col': ips_col, - 'ipd_col': ipd_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, + '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, } - _store(session_id, {'meta': meta, 'df': df}) + _store(session_id, {'meta': meta, 'df': df, 'geo_cache': geo_cache}) return JsonResponse({ 'session_id': session_id, 'total_rows': total_rows, - 'columns': col_names[:50], + '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, - 'src_lat': src_lat_col, - 'src_lon': src_lon_col, - 'dst_lat': dst_lat_col, - 'dst_lon': dst_lon_col, 'ips': ips_col, 'ipd': ipd_col, + 'ispn': ispn_col, + 'orgn': orgn_col, + 'city': city_col, }, 'cnam_frequency': cnam_freq[:30], 'top_cnam': top_cnam, + 'geo_stats': { + 'ips_resolved': len(geo_cache), + }, }) -# ── Step 2: Filter ─────────────────────────────────────────────────────────── +# ═══════════════════════════════════════════════════════════════════════════════ +# 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 + @csrf_exempt @require_POST -def filter_data(request): - """Apply cnam + cnrs + isrs filters; return preview rows.""" +def advanced_filter(request): + """高级筛选：接受 AND/OR/NOT 组合条件，返回筛选结果。""" body = json.loads(request.body) session_id = body.get('session_id') - selected_cnam = body.get('selected_cnam', '') + conditions = body.get('conditions', []) # list of {col, op, val, logic} + global_logic = body.get('global_logic', 'AND') # 'AND' or 'OR' entry = _cached(session_id) if entry is None: return JsonResponse({'error': '会话已过期，请重新上传'}, status=400) - df: pl.DataFrame = entry['df'] - meta = entry['meta'] - cnam_col = meta['cnam_col'] - cnrs_col = meta['cnrs_col'] - isrs_col = meta['isrs_col'] + df = entry['df'] # 使用原始 df（不含预设筛选），预设筛选已应用 - # Validate required columns - missing = [] - if not cnam_col: - missing.append('cnam/snam (目标类别)') + 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': []}) - if missing: - return JsonResponse({ - 'error': f'缺少关键列: {", ".join(missing)}。请检查CSV文件格式。', - 'missing_columns': missing, - }, status=400) - - # Apply filter — cnrs and isrs are optional; only apply if column exists try: - cond = (pl.col(cnam_col).cast(pl.Utf8) == selected_cnam) - if cnrs_col: - cond = cond & (pl.col(cnrs_col).cast(pl.Utf8).str.strip_chars() == '+') - if isrs_col: - cond = cond & (pl.col(isrs_col).cast(pl.Utf8).str.strip_chars() == '+') - filtered = df.filter(cond) + # 构建条件组 + 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) + return JsonResponse({'error': f'筛选构建失败: {e}'}, status=400) filtered_rows = len(filtered) - meta['filtered_cnam'] = selected_cnam - meta['filtered_rows'] = filtered_rows + entry['meta']['filtered_rows'] = filtered_rows entry['filtered_df'] = filtered - # Preview data (first 20 rows) 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): @@ -239,17 +509,81 @@ def filter_data(request): }) -# ── Step 3: Cluster ────────────────────────────────────────────────────────── +# ── 兼容原 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: 聚类（GeoIP 驱动 + 性能优化 + 全噪声容错） +# ═══════════════════════════════════════════════════════════════════════════════ @csrf_exempt @require_POST def run_clustering(request): - """Run HDBSCAN geo-clustering or IP-subnet soft-clustering.""" + """执行聚类分析：IP 子网模式或 GeoIP 地理位置模式。""" body = json.loads(request.body) session_id = body.get('session_id') - mode = body.get('mode', 'geo') # 'geo' or 'subnet' + mode = body.get('mode', 'geo') min_cluster_size = int(body.get('min_cluster_size', 5)) - cluster_selection_epsilon = float(body.get('cluster_selection_epsilon', 0.005)) + cluster_selection_epsilon = float(body.get('cluster_selection_epsilon', 0.05)) min_records_per_subnet = int(body.get('min_records_per_subnet', 3)) max_cluster_size = int(body.get('max_cluster_size', 100)) @@ -257,190 +591,269 @@ def run_clustering(request): if entry is None: return JsonResponse({'error': '会话已过期，请重新上传'}, status=400) - filtered = entry.get('filtered_df') + 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) + return JsonResponse({'error': '请先执行筛选'}, status=400) meta = entry['meta'] - src_lat_col = meta['src_lat_col'] - src_lon_col = meta['src_lon_col'] - ips_col = meta['ips_col'] + ips_col = meta.get('ips_col') + + t_start = time.time() + clustered = False + result_data = {} if mode == 'geo': - result = _cluster_geo(entry, filtered, meta, min_cluster_size, cluster_selection_epsilon) + clustered, result_data = _cluster_geo(entry, filtered, meta, min_cluster_size, cluster_selection_epsilon) else: - result = _cluster_subnet(entry, filtered, meta, min_records_per_subnet, max_cluster_size) - # Update cache with any modifications made by cluster functions + clustered, result_data = _cluster_subnet(entry, filtered, meta, min_records_per_subnet, max_cluster_size) + + elapsed = round(time.time() - t_start, 2) + result_data['elapsed_seconds'] = elapsed _store(session_id, entry) - return result + return JsonResponse(result_data) + + +def _get_geo_coords(entry: dict, df: pl.DataFrame, meta: dict) -> tuple: + """从 GeoIP 缓存中提取每条记录的经纬度，返回 (lat_float32, lon_float32, valid_mask)。""" + ips_col = meta.get('ips_col') + geo_cache = entry.get('geo_cache', {}) + + if not ips_col or ips_col not in df.columns: + return np.array([], dtype=np.float32), np.array([], dtype=np.float32), np.array([], dtype=bool) + + ip_series = df[ips_col].cast(pl.Utf8).to_list() + n = len(ip_series) + lats = np.full(n, np.nan, dtype=np.float32) + lons = np.full(n, np.nan, dtype=np.float32) + valid = np.zeros(n, dtype=bool) + + for i, ip in enumerate(ip_series): + if ip and ip not in ('None', '', '+', '-'): + info = geo_cache.get(ip) or geoip_lookup(ip) + if info and 'lat' in info and 'lon' in info: + lats[i] = np.float32(info['lat']) + lons[i] = np.float32(info['lon']) + valid[i] = True + + return lats, lons, valid def _cluster_geo(entry: dict, df: pl.DataFrame, meta: dict, - min_cluster_size: int, cluster_selection_epsilon: float = 0.005): - """HDBSCAN clustering on source IP latitude/longitude using Haversine distance.""" - src_lat_col = meta['src_lat_col'] - src_lon_col = meta['src_lon_col'] - ips_col = meta['ips_col'] + min_cluster_size: int, cluster_selection_epsilon: float = 0.01): + """使用 GeoIP 经纬度进行 HDBSCAN 聚类，含性能优化和全噪声容错。""" - if not src_lat_col or not src_lon_col: - return JsonResponse({'error': '缺少源经纬度列（:ips.latd / :ips.lond），无法执行地理聚类'}, status=400) + # 从 GeoIP 获取经纬度，不再依赖 CSV 列 + lats, lons, valid_mask = _get_geo_coords(entry, df, meta) + n_valid = int(valid_mask.sum()) + total = len(df) - # Drop rows with missing lat/lon - clean = df.filter( - pl.col(src_lat_col).is_not_null() - & pl.col(src_lon_col).is_not_null() - ) - - if len(clean) == 0: - return JsonResponse({'error': '所有记录均缺失经纬度，无法聚类'}, status=400) - - # Clean lat/lon: cast to string, filter out '+'/'-' artifacts - try: - lat_raw = clean[src_lat_col].cast(pl.Utf8) - lon_raw = clean[src_lon_col].cast(pl.Utf8) - # Build validity mask: exclude '+', '-', empty, and None - valid_mask = ( - lat_raw.is_not_null() & lon_raw.is_not_null() - & (lat_raw != '+') & (lat_raw != '-') - & (lon_raw != '+') & (lon_raw != '-') - & (lat_raw != '') & (lon_raw != '') - ) - clean = clean.filter(valid_mask) - if len(clean) == 0: - return JsonResponse({'error': '无有效经纬度记录（排除符号值后）'}, status=400) - lat_vals = clean[src_lat_col].cast(pl.Float64, strict=False).to_numpy() - lon_vals = clean[src_lon_col].cast(pl.Float64, strict=False).to_numpy() - # Final NaN check - lat_vals = np.where(np.isnan(lat_vals), 0.0, lat_vals) - lon_vals = np.where(np.isnan(lon_vals), 0.0, lon_vals) - except Exception as e: - return JsonResponse({'error': f'经纬度列转换失败: {e}'}, status=400) - - # Filter to valid lat/lon ranges - valid_range_mask = ( - (lat_vals >= -90) & (lat_vals <= 90) - & (lon_vals >= -180) & (lon_vals <= 180) - & ~np.isnan(lat_vals) & ~np.isnan(lon_vals) - ) - lat_vals = lat_vals[valid_range_mask] - lon_vals = lon_vals[valid_range_mask] - clean = clean.filter( - pl.Series(valid_range_mask) - ) - - if len(clean) < min_cluster_size: - return JsonResponse({ - 'error': f'有效经纬度记录数（{len(clean)}）少于最小聚类规模（{min_cluster_size}）', - }, status=400) - - # ── HDBSCAN with Haversine distance ──────────────────────────────────── - # Convert degrees to radians for haversine metric - lat_rad = np.radians(lat_vals) - lon_rad = np.radians(lon_vals) - coords = np.column_stack([lat_rad, lon_rad]) - - clusterer = HDBSCAN( - min_cluster_size=min_cluster_size, - min_samples=min(3, min_cluster_size), - metric='haversine', - cluster_selection_epsilon=cluster_selection_epsilon, - ) - labels = clusterer.fit_predict(coords) - - n_clusters = int(len(set(labels)) - (1 if -1 in labels else 0)) - n_noise = int((labels == -1).sum()) - noise_ratio = n_noise / len(labels) if len(labels) > 0 else 0 - - # Check for degenerate clustering (fallback suggestion) - fallback_suggested = False - if noise_ratio > 0.8 or n_clusters <= 1: - fallback_suggested = True - - # Build cluster overview - cluster_map = {} - unique_labels = sorted(set(labels)) - data_points_by_cluster = {} - for lbl in unique_labels: - mask = labels == lbl - idxs = np.where(mask)[0] - cluster_lats = lat_vals[idxs] - cluster_lons = lon_vals[idxs] - center_lat = float(np.median(cluster_lats)) - center_lon = float(np.median(cluster_lons)) - - # Unique IPs - if ips_col: - ips_in_cluster = clean.filter(pl.Series(mask))[ips_col].cast(pl.Utf8).unique().to_list() - n_unique_ips = len(ips_in_cluster) - else: - ips_in_cluster = [] - n_unique_ips = 0 - - cluster_map[int(lbl)] = { - 'label': int(lbl), - 'size': int(mask.sum()), - 'center_lat': round(float(center_lat), 4), - 'center_lon': round(float(center_lon), 4), - 'n_unique_ips': int(n_unique_ips), - 'is_noise': bool(lbl == -1), + if n_valid < 3: + # 全噪声或数据不足 → 返回全 -1 标签，仍可可视化 + labels = np.full(total, -1, dtype=int) + _store_cluster_results(entry, df, labels, meta, { + 'note': 'GeoIP 解析不足，所有点标记为噪声', + }) + return True, { + 'n_clusters': 0, 'n_noise': total, 'n_valid_points': n_valid, + 'clusters': {}, 'labels': labels.tolist(), 'data_points': {}, + 'all_noise': True, 'note': 'GeoIP 解析不足（<3个有效坐标），全噪声', } - # Collect individual data points for map rendering (lat, lon, index) - pts = [] - for pt_idx, i in enumerate(idxs): - pts.append({ - 'lat': round(float(cluster_lats[pt_idx]), 4), - 'lon': round(float(cluster_lons[pt_idx]), 4), - 'idx': int(i), - }) - data_points_by_cluster[int(lbl)] = pts + lat_vals = lats[valid_mask].astype(np.float32) + lon_vals = lons[valid_mask].astype(np.float32) - # Annotate original filtered DataFrame with cluster labels - label_series = np.full(len(df), -1, dtype=int) - # Map back: only the valid rows got labels - valid_indices = np.where(valid_range_mask)[0] - for i, idx in enumerate(valid_indices): - label_series[idx] = int(labels[i]) + # ── 性能优化：大数据降采样 ──────────────────────────────────────────── + downsample_used = False + original_n = n_valid + if n_valid > 50000: + # 超过 5 万点，随机抽取 30% 或使用 MiniBatch 方式 + rng = np.random.RandomState(42) + subset_idx = rng.choice(n_valid, size=max(15000, n_valid // 3), replace=False) + lat_vals = lat_vals[subset_idx] + lon_vals = lon_vals[subset_idx] + downsample_used = True - entry['cluster_labels'] = label_series.tolist() - entry['cluster_map'] = cluster_map + # 弧度转换（float32 节省内存） + coords = np.column_stack([np.radians(lat_vals), np.radians(lon_vals)]).astype(np.float32) + + # 为重复坐标添加微小抖动（HDBSCAN 距离为 0 时难以处理） + if len(coords) > 10: + _, unique_idx = np.unique(coords, axis=0, return_index=True) + if len(unique_idx) < len(coords) * 0.8: + jitter = np.random.RandomState(42).normal(0, 1e-6, coords.shape).astype(np.float32) + coords = coords + jitter + + # ── HDBSCAN 聚类 ─────────────────────────────────────────────────────── + all_noise = False + fallback_used = False + n_clusters = 0 + n_noise = 0 + + try: + clusterer = HDBSCAN( + min_cluster_size=min_cluster_size, + min_samples=min(3, min_cluster_size), + metric='haversine', + cluster_selection_epsilon=cluster_selection_epsilon, + # core_dist_n_jobs removed for sklearn 1.5.2 compatibility + ) + raw_labels = clusterer.fit_predict(coords) + n_clusters = int(len(set(raw_labels)) - (1 if -1 in raw_labels else 0)) + n_noise = int((raw_labels == -1).sum()) + + noise_ratio = n_noise / len(raw_labels) if len(raw_labels) > 0 else 1.0 + + # 全噪声或仅 1 簇 → 回退 DBSCAN + if noise_ratio > 0.95 or n_clusters <= 0: + logger.info('HDBSCAN produced %d clusters, %d noise — falling back to DBSCAN', n_clusters, n_noise) + db = DBSCAN(eps=cluster_selection_epsilon, min_samples=min(3, min_cluster_size), metric='haversine') + raw_labels = db.fit_predict(coords) + n_clusters = int(len(set(raw_labels)) - (1 if -1 in raw_labels else 0)) + n_noise = int((raw_labels == -1).sum()) + fallback_used = True + + if n_clusters <= 0: + all_noise = True + + except Exception as e: + logger.warning('Clustering failed: %s — all noise fallback', e) + raw_labels = np.full(len(coords), -1, dtype=int) + all_noise = True + + # 将子集标签映射回全量数据 + labels = np.full(total, -1, dtype=int) + valid_indices = np.where(valid_mask)[0] + if downsample_used: + # 降采样时，给被选中的记录分配标签，其余噪声 + for i, idx in enumerate(subset_idx): + labels[valid_indices[idx]] = int(raw_labels[i]) + else: + for i, idx in enumerate(valid_indices): + labels[idx] = int(raw_labels[i]) + + result_info = { + 'all_noise': all_noise, + 'note': '', + 'fallback_used': fallback_used, + 'downsample_used': downsample_used, + } + if all_noise: + result_info['note'] = '未发现有效聚集，所有点为噪声' + elif fallback_used: + result_info['note'] = 'HDBSCAN 效果不佳，已回退 DBSCAN' + + _store_cluster_results(entry, df, labels, meta, result_info) + return True, _build_cluster_response(entry, df, labels, meta, result_info, lats, lons, valid_mask) + + +def _store_cluster_results(entry: dict, df: pl.DataFrame, labels: np.ndarray, + meta: dict, info: dict) -> None: + """存储聚类结果到缓存。""" + n_noise = int((labels == -1).sum()) + n_clusters = int(len(set(labels)) - (1 if -1 in labels else 0)) + entry['cluster_labels'] = labels.tolist() entry['cluster_mode'] = 'geo' entry['n_clusters'] = n_clusters entry['n_noise'] = n_noise + entry['cluster_info'] = info - # Update filtered_df with cluster label column - df_with_labels = df.with_columns(pl.Series('_cluster_label', label_series)) + df_with_labels = df.with_columns(pl.Series('_cluster_label', labels)) entry['filtered_df'] = df_with_labels - return JsonResponse({ - 'n_clusters': n_clusters, - 'n_noise': n_noise, - 'n_valid_points': len(clean), + +def _build_cluster_response(entry: dict, df: pl.DataFrame, labels: np.ndarray, + meta: dict, info: dict, + lats: np.ndarray | None = None, lons: np.ndarray | None = None, + valid_mask: np.ndarray | None = None) -> dict: + """构建聚类响应，包含簇概览和数据点坐标。""" + ips_col = meta.get('ips_col') + unique_labels = sorted(set(labels)) + cluster_map = {} + data_points_by_cluster = {} + total = len(df) + + for lbl in unique_labels: + mask = labels == lbl + count = int(mask.sum()) + is_noise = bool(lbl == -1) + + # 中心坐标 + center_lat = center_lon = None + if lats is not None and lons is not None and valid_mask is not None: + # 获取该簇中有效坐标的索引 + valid_indices = np.where(valid_mask)[0] + cluster_valid_mask = mask[valid_mask] if len(valid_indices) == total else mask[valid_indices] + if cluster_valid_mask.any(): + clats = lats[valid_mask][cluster_valid_mask[:len(lats[valid_mask])]] + clons = lons[valid_mask][cluster_valid_mask[:len(lons[valid_mask])]] + if len(clats) > 0: + center_lat = round(float(np.median(clats)), 4) + center_lon = round(float(np.median(clons)), 4) + + # 唯一 IP 数 + n_unique_ips = 0 + if ips_col and ips_col in df.columns: + try: + ips = df.filter(pl.Series(mask))[ips_col].cast(pl.Utf8).unique().to_list() + n_unique_ips = len(ips) + except Exception: + pass + + cluster_map[int(lbl)] = { + 'label': int(lbl), + 'size': count, + 'center_lat': center_lat, + 'center_lon': center_lon, + 'n_unique_ips': n_unique_ips, + 'is_noise': is_noise, + } + + # 数据点坐标 + pts = [] + if lats is not None and lons is not None and valid_mask is not None: + valid_indices = np.where(valid_mask)[0] + for pt_idx_in_valid, orig_idx in enumerate(valid_indices): + if labels[orig_idx] == lbl: + pts.append({ + 'lat': round(float(lats[pt_idx_in_valid]), 4), + 'lon': round(float(lons[pt_idx_in_valid]), 4), + 'idx': int(orig_idx), + }) + data_points_by_cluster[int(lbl)] = pts + + return { + 'n_clusters': len(cluster_map) - (1 if -1 in cluster_map else 0), + 'n_noise': int((labels == -1).sum()), + 'n_valid_points': int(valid_mask.sum()) if valid_mask is not None else total, + 'total_records': total, 'clusters': cluster_map, - 'labels': label_series.tolist(), + 'labels': labels.tolist(), 'data_points': data_points_by_cluster, - 'fallback_suggested': fallback_suggested, - }) + 'all_noise': info.get('all_noise', False), + 'note': info.get('note', ''), + 'fallback_used': info.get('fallback_used', False), + 'downsample_used': info.get('downsample_used', False), + } def _cluster_subnet(entry: dict, df: pl.DataFrame, meta: dict, min_records_per_subnet: int = 3, max_cluster_size: int = 100): - """Soft clustering by /24 subnet prefix. + """Soft clustering by /24 subnet prefix using GeoIP coordinates. Subnets with fewer than *min_records_per_subnet* records are marked noise (-1). - Subnets with more than *max_cluster_size* records are auto-split using geo - sub-clustering (HDBSCAN + haversine) on the available lat/lon within the subnet. + Large subnets use GeoIP coordinates for automatic geo sub-clustering. """ - ips_col = meta['ips_col'] - src_lat_col = meta['src_lat_col'] - src_lon_col = meta['src_lon_col'] + ips_col = meta.get('ips_col') if not ips_col: - return JsonResponse({'error': '缺少源IP列（:ips），无法执行子网聚类'}, status=400) + return False, {'error': '缺少源IP列（:ips），无法执行子网聚类'} try: ip_series = df[ips_col].cast(pl.Utf8) except Exception as e: - return JsonResponse({'error': f'IP列转换失败: {e}'}, status=400) + return False, {'error': f'IP列转换失败: {e}'} + + # 使用 GeoIP 提取经纬度 + geo_lats, geo_lons, geo_valid = _get_geo_coords(entry, df, meta) # Extract /24 subnet def _subnet24(ip: str) -> str | None: @@ -454,12 +867,10 @@ def _cluster_subnet(entry: dict, df: pl.DataFrame, meta: dict, subnets = [None if v is None else _subnet24(str(v)) for v in ip_series.to_list()] subnet_count = Counter(s for s in subnets if s is not None) - # Assign cluster labels based on subnet frequency labels = [] cluster_map: dict[int, dict] = {} next_label = 0 subnet_to_label: dict[str, int] = {} - # Track which records belong to large subnets for geo splitting large_subnet_indices: dict[str, list[int]] = {} for row_idx, s in enumerate(subnets): @@ -468,18 +879,14 @@ def _cluster_subnet(entry: dict, df: pl.DataFrame, meta: dict, else: cnt = subnet_count.get(s, 0) if cnt < min_records_per_subnet: - labels.append(-1) # noise (too few records) + labels.append(-1) elif cnt > max_cluster_size: - # Large subnet → will apply geo sub-clustering later if s not in subnet_to_label: subnet_to_label[s] = next_label cluster_map[next_label] = { - 'label': next_label, - 'size': 0, - 'subnet': s, - 'is_noise': False, - '_is_large': True, - '_large_subnet_name': s, + 'label': next_label, 'size': 0, + 'subnet': s, 'is_noise': False, + '_is_large': True, '_large_subnet_name': s, } next_label += 1 large_subnet_indices[s] = [] @@ -491,114 +898,85 @@ def _cluster_subnet(entry: dict, df: pl.DataFrame, meta: dict, if s not in subnet_to_label: subnet_to_label[s] = next_label cluster_map[next_label] = { - 'label': next_label, - 'size': 0, - 'subnet': s, - 'is_noise': False, + 'label': next_label, 'size': 0, + 'subnet': s, 'is_noise': False, } next_label += 1 lbl = subnet_to_label[s] labels.append(lbl) cluster_map[lbl]['size'] = cluster_map[lbl].get('size', 0) + 1 - # ── Geo sub-clustering for large subnets ──────────────────────────────── - if large_subnet_indices and src_lat_col and src_lon_col: + # 大子网地理拆分（使用 GeoIP 坐标） + if large_subnet_indices and geo_valid.any(): for subnet_name, indices in large_subnet_indices.items(): - old_label = subnet_to_label[subnet_name] - # Remove the temporary large-subnet cluster entry + old_label = subnet_to_label.get(subnet_name) + if old_label is None: + continue cluster_map.pop(old_label, None) - - # Get lat/lon for records in this subnet - subnet_df = df[pl.Series(indices)] try: - sub_lats = subnet_df[src_lat_col].cast(pl.Float64).to_numpy() - sub_lons = subnet_df[src_lon_col].cast(pl.Float64).to_numpy() - valid = ( - ~np.isnan(sub_lats) & ~np.isnan(sub_lons) - & (sub_lats >= -90) & (sub_lats <= 90) - & (sub_lons >= -180) & (sub_lons <= 180) - ) - sub_lats = sub_lats[valid] - sub_lons = sub_lons[valid] - if len(sub_lats) >= 3: - # Run HDBSCAN with haversine on this subnet's points - coords = np.column_stack([np.radians(sub_lats), np.radians(sub_lons)]) + sub_lats = geo_lats[indices] + sub_lons = geo_lons[indices] + sub_valid = geo_valid[indices] + valid_sub = sub_valid.nonzero()[0] + if len(valid_sub) >= 3: + sub_coords = np.column_stack([ + np.radians(sub_lats[valid_sub].astype(np.float32)), + np.radians(sub_lons[valid_sub].astype(np.float32)), + ]) sub_cluster = HDBSCAN( - min_cluster_size=min(3, len(sub_lats) // 5), - min_samples=2, - metric='haversine', + min_cluster_size=max(3, len(valid_sub) // 10), + min_samples=2, metric='haversine', cluster_selection_epsilon=0.005, ) - sub_labels = sub_cluster.fit_predict(coords) - - # Create sub-clusters - sub_label_map: dict[int, int] = {} - valid_idx_iter = 0 - for orig_rank, orig_idx in enumerate(indices): - if not valid[orig_rank - valid_idx_iter]: - # Was invalid, check if we skipped - pass - - # Re-map: skip invalid points (mark noise) + sub_labels = sub_cluster.fit_predict(sub_coords) valid_counter = 0 - for orig_rank, orig_idx in enumerate(indices): - if not valid[orig_rank]: - labels[orig_idx] = -1 # no lat/lon → noise + for orig_rank in range(len(indices)): + if not sub_valid[orig_rank]: + labels[indices[orig_rank]] = -1 else: sl = int(sub_labels[valid_counter]) valid_counter += 1 if sl == -1: - labels[orig_idx] = -1 + labels[indices[orig_rank]] = -1 else: - sub_cluster_name = f'{subnet_name}/sub{sl}' - if sub_cluster_name not in subnet_to_label: - subnet_to_label[sub_cluster_name] = next_label + scn = f'{subnet_name}/sub{sl}' + if scn not in subnet_to_label: + subnet_to_label[scn] = next_label cluster_map[next_label] = { - 'label': next_label, - 'size': 0, - 'subnet': sub_cluster_name, - 'is_noise': False, + 'label': next_label, 'size': 0, + 'subnet': scn, 'is_noise': False, } next_label += 1 - new_lbl = subnet_to_label[sub_cluster_name] - labels[orig_idx] = new_lbl - cluster_map[new_lbl]['size'] = cluster_map[new_lbl].get('size', 0) + 1 + nl = subnet_to_label[scn] + labels[indices[orig_rank]] = nl + cluster_map[nl]['size'] = cluster_map[nl].get('size', 0) + 1 else: - # Not enough geo data → keep as single cluster cluster_map[old_label] = { - 'label': old_label, - 'size': len(indices), - 'subnet': subnet_name, - 'is_noise': False, + 'label': old_label, 'size': len(indices), + 'subnet': subnet_name, 'is_noise': False, } subnet_to_label[subnet_name] = old_label except Exception as e: - logger.warning('Subnet geo-splitting failed for %s: %s', subnet_name, e) - # Keep the original cluster + logger.warning('Subnet geo-split failed %s: %s', subnet_name, e) cluster_map[old_label] = { - 'label': old_label, - 'size': len(indices), - 'subnet': subnet_name, - 'is_noise': False, + 'label': old_label, 'size': len(indices), + 'subnet': subnet_name, 'is_noise': False, } - subnet_to_label[subnet_name] = old_label - # Update cluster maps with centers and unique IP counts + # 更新簇中心（使用 GeoIP 坐标） for lbl in list(cluster_map.keys()): mask = np.array(labels) == lbl - ips_in_cluster = df.filter(pl.Series(mask))[ips_col].cast(pl.Utf8).unique().to_list() - cluster_map[lbl]['n_unique_ips'] = len(ips_in_cluster) - # Try to get geo center if lat/lon available - if src_lat_col and src_lon_col: + if ips_col: try: - clat = df.filter(pl.Series(mask))[src_lat_col].cast(pl.Float64).drop_nulls() - clon = df.filter(pl.Series(mask))[src_lon_col].cast(pl.Float64).drop_nulls() - if len(clat) > 0 and len(clon) > 0: - cluster_map[lbl]['center_lat'] = round(float(clat.median()), 4) - cluster_map[lbl]['center_lon'] = round(float(clon.median()), 4) + ips_in = df.filter(pl.Series(mask))[ips_col].cast(pl.Utf8).unique().to_list() + cluster_map[lbl]['n_unique_ips'] = len(ips_in) except Exception: - cluster_map[lbl]['center_lat'] = None - cluster_map[lbl]['center_lon'] = None + cluster_map[lbl]['n_unique_ips'] = 0 + # Geo 中心 + valid_in_cluster = geo_valid & mask + if valid_in_cluster.any(): + cluster_map[lbl]['center_lat'] = round(float(np.median(geo_lats[valid_in_cluster])), 4) + cluster_map[lbl]['center_lon'] = round(float(np.median(geo_lons[valid_in_cluster])), 4) else: cluster_map[lbl]['center_lat'] = None cluster_map[lbl]['center_lon'] = None @@ -606,44 +984,28 @@ def _cluster_subnet(entry: dict, df: pl.DataFrame, meta: dict, n_noise = labels.count(-1) n_clusters = len(cluster_map) - # Build data_points for map + # 数据点 data_points_by_cluster: dict[int, list] = {} for lbl in list(cluster_map.keys()): mask = np.array(labels) == lbl pts = [] - if src_lat_col and src_lon_col: - try: - sub = df.filter(pl.Series(mask)) - lats = sub[src_lat_col].cast(pl.Float64).to_numpy() - lons = sub[src_lon_col].cast(pl.Float64).to_numpy() - for i in range(len(sub)): - if not np.isnan(lats[i]) and not np.isnan(lons[i]): - pts.append({'lat': round(float(lats[i]), 4), - 'lon': round(float(lons[i]), 4), - 'idx': int(i)}) - except Exception: - pass + valid_in = geo_valid & mask + if valid_in.any(): + idxs = np.where(valid_in)[0] + for idx in idxs[:200]: # 最多 200 点/簇 + pts.append({'lat': round(float(geo_lats[idx]), 4), + 'lon': round(float(geo_lons[idx]), 4), 'idx': int(idx)}) data_points_by_cluster[lbl] = pts - - # Also collect noisy points' coordinates noise_mask = np.array(labels) == -1 - noise_points = [] - if src_lat_col and src_lon_col: - try: - noise_sub = df.filter(pl.Series(noise_mask)) - nlats = noise_sub[src_lat_col].cast(pl.Float64).to_numpy() - nlons = noise_sub[src_lon_col].cast(pl.Float64).to_numpy() - for i in range(len(noise_sub)): - if not np.isnan(nlats[i]) and not np.isnan(nlons[i]): - noise_points.append({'lat': round(float(nlats[i]), 4), - 'lon': round(float(nlons[i]), 4), - 'idx': int(i)}) - except Exception: - pass - if noise_points: - data_points_by_cluster[-1] = noise_points + noise_pts = [] + valid_noise = geo_valid & noise_mask + if valid_noise.any(): + for idx in np.where(valid_noise)[0][:500]: + noise_pts.append({'lat': round(float(geo_lats[idx]), 4), + 'lon': round(float(geo_lons[idx]), 4), 'idx': int(idx)}) + if noise_pts: + data_points_by_cluster[-1] = noise_pts - # Annotate DataFrame df_with_labels = df.with_columns(pl.Series('_cluster_label', labels)) entry['cluster_labels'] = labels entry['cluster_map'] = cluster_map @@ -652,21 +1014,22 @@ def _cluster_subnet(entry: dict, df: pl.DataFrame, meta: dict, entry['n_clusters'] = n_clusters entry['n_noise'] = n_noise - return JsonResponse({ - 'n_clusters': n_clusters, - 'n_noise': n_noise, - 'n_valid_points': len(df), - 'clusters': cluster_map, - 'labels': labels, + return True, { + 'n_clusters': n_clusters, 'n_noise': n_noise, + 'n_valid_points': len(df), 'total_records': len(df), + 'clusters': cluster_map, 'labels': labels, 'data_points': data_points_by_cluster, - }) + 'all_noise': n_clusters == 0, + } -# ── Step 4: Cluster detail ─────────────────────────────────────────────────── +# ═══════════════════════════════════════════════════════════════════════════════ +# Step 4: Cluster detail + 边关系 +# ═══════════════════════════════════════════════════════════════════════════════ @require_GET def cluster_detail(request, label: int): - """Return detailed profile for a specific cluster.""" + """返回指定簇的详细画像，含 IP 属性、GeoIP 信息、流量特征等。""" session_id = request.GET.get('session_id') if not session_id: return JsonResponse({'error': '缺少 session_id'}, status=400) @@ -675,9 +1038,10 @@ def cluster_detail(request, label: int): if entry is None: return JsonResponse({'error': '会话已过期'}, status=400) - df = entry.get('filtered_df') + 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) @@ -692,31 +1056,43 @@ def cluster_detail(request, label: int): cluster_info = cluster_map.get(label, {}) - # Helper: get unique values for a column def _top_values(col_name: str, top_n: int = 5) -> list: - if col_name not in df.columns: + 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) + cluster_df[col_name].cast(pl.Utf8).fill_null('(空)') + .value_counts().sort('count', descending=True).head(top_n) ) total = vals['count'].sum() - return [ - {'value': v['counts'] if isinstance(v, dict) else str(row[0]), - 'count': int(row[1]), - 'pct': round(int(row[1]) / total * 100, 1)} - for row in vals.iter_rows() - ] + 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 [] - # Source IPs + # 列映射 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') + tls_cph_col = meta.get('tls_cph_col') + bytes_col = meta.get('bytes_col') + dur_col = meta.get('dur_col') + ispn_col = meta.get('ispn_col') + orgn_col = meta.get('orgn_col') + city_col = meta.get('city_col') + scnt_col = meta.get('scnt_col') + time_col = meta.get('time_col') + snam_col = meta.get('snam_col') + text_col = meta.get('text_col') + + # 源 IP 列表 unique_ips = [] if ips_col and ips_col in cluster_df.columns: try: @@ -725,16 +1101,19 @@ def cluster_detail(request, label: int): except Exception: pass - # Detect useful columns for profiling - col_names = df.columns - dst_ip_col = _find_col(col_names, ':ipd', 'dst_ip', 'dest_ip', 'destination_ip') - dst_port_col = _find_col(col_names, 'dst_port', 'dest_port', 'port', 'dport') - tls_ver_col = _find_col(col_names, '0ver', 'tls_version', 'version', 'tlsver') - cipher_col = _find_col(col_names, '0cph', 'cipher_suite', 'cipher', 'tls_cipher') - sni_col = _find_col(col_names, 'sni', 'server_name', 'tls_sni') - bytes_col = _find_col(col_names, '8byt', 'bytes_sent', 'bytes', 'src_bytes') - dur_col = _find_col(col_names, '4dur', 'duration', 'dur', 'flow_duration') - proto_col = _find_col(col_names, 'proto', 'protocol', 'l4_proto') + # 使用 GeoIP 获取 IP 属性 + ip_attrs = [] + for ip in unique_ips[:50]: + info = geo_cache.get(ip) or {} + if not info: + info = 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, @@ -742,52 +1121,138 @@ def cluster_detail(request, label: int): 'cluster_info': cluster_info, 'unique_ips': unique_ips[:100], 'unique_ips_count': len(unique_ips), - 'top_dst_ips': _top_values(dst_ip_col) if dst_ip_col else [], - 'top_dst_ports': _top_values(dst_port_col) if dst_port_col else [], + '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_ciphers': _top_values(cipher_col) if cipher_col else [], - 'top_sni': _top_values(sni_col) if sni_col else [], - 'top_protocols': _top_values(proto_col) if proto_col else [], - # Traffic stats + 'top_ciphers': _top_values(tls_cph_col) if tls_cph_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': {}, } - # Traffic stats + # 流量统计 if bytes_col and bytes_col in cluster_df.columns: try: - total_bytes = cluster_df[bytes_col].cast(pl.Float64).sum() - profile['traffic']['total_bytes'] = round(total_bytes, 2) if total_bytes else 0 + profile['traffic']['total_bytes'] = round(float(cluster_df[bytes_col].cast(pl.Float64).sum()), 2) + profile['traffic']['avg_bytes'] = round(float(cluster_df[bytes_col].cast(pl.Float64).mean()), 2) except Exception: pass if dur_col and dur_col in cluster_df.columns: try: - total_dur = cluster_df[dur_col].cast(pl.Float64).sum() - avg_dur = cluster_df[dur_col].cast(pl.Float64).mean() - profile['traffic']['total_duration'] = round(total_dur, 2) if total_dur else 0 - profile['traffic']['avg_duration'] = round(avg_dur, 2) if avg_dur else 0 + profile['traffic']['total_duration'] = round(float(cluster_df[dur_col].cast(pl.Float64).sum()), 2) + profile['traffic']['avg_duration'] = round(float(cluster_df[dur_col].cast(pl.Float64).mean()), 2) except Exception: pass - # Geo bounding box (if lat/lon available) - src_lat_col = meta.get('src_lat_col') - src_lon_col = meta.get('src_lon_col') - if src_lat_col and src_lon_col and src_lat_col in cluster_df.columns: + # 时间范围 + if time_col and time_col in cluster_df.columns: try: - lats = cluster_df[src_lat_col].cast(pl.Float64).drop_nulls() - lons = cluster_df[src_lon_col].cast(pl.Float64).drop_nulls() - if len(lats) > 0 and len(lons) > 0: - profile['geo_bounds'] = { - 'lat_min': round(float(lats.min()), 4), - 'lat_max': round(float(lats.max()), 4), - 'lon_min': round(float(lons.min()), 4), - 'lon_max': round(float(lons.max()), 4), + 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 + # 地理边界 + geo_lats = [] + geo_lons = [] + for ip in unique_ips[:100]: + info = geo_cache.get(ip) or geoip_lookup(ip) or {} + if info.get('lat') and info.get('lon'): + geo_lats.append(info['lat']) + geo_lons.append(info['lon']) + if geo_lats and geo_lons: + profile['geo_bounds'] = { + 'lat_min': round(min(geo_lats), 4), + 'lat_max': round(max(geo_lats), 4), + 'lon_min': round(min(geo_lons), 4), + 'lon_max': round(max(geo_lons), 4), + } + return JsonResponse(profile) +@csrf_exempt +@require_POST +def edge_relationships(request): + """计算 IP 通联关系（边），返回通联次数、时间规律等。""" + body = json.loads(request.body) + session_id = body.get('session_id') + min_freq = int(body.get('min_freq', 10)) + + 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') + 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') + + 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() + + # 聚合统计 + pair_counts: dict[tuple[str, str], int] = {} + 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 time_col and time_col in df.columns: + try: + v = str(df[time_col][i]) + if v and v not in ('None', ''): + if key not in pair_times: + pair_times[key] = [] + pair_times[key].append(v) + except Exception: + pass + + edges = [] + for (src, dst), count in pair_counts.items(): + if count < min_freq: + continue + times = pair_times.get((src, dst), []) + edge = { + 'source': src, + 'target': dst, + 'count': count, + 'times': times[:50], + 'time_count': len(times), + } + edges.append(edge) + + edges.sort(key=lambda e: e['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) + + # ── Main page ──────────────────────────────────────────────────────────────── def index(request):