TianXuan Developer
55f3b3637a
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 <noreply@anthropic.com>
1261 lines
50 KiB
Python
1261 lines
50 KiB
Python
"""简单分析模块 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'],
|
|
}
|
|
|
|
|
|
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 1: Upload + 预设筛选 + GeoIP 解析
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@csrf_exempt
|
|
@require_POST
|
|
def upload_csv(request):
|
|
"""上传 CSV，列归一化，预设筛选，GeoIP 解析，返回摘要。"""
|
|
files = request.FILES.getlist('files')
|
|
if not files:
|
|
return JsonResponse({'error': '未上传任何文件'}, status=400)
|
|
|
|
session_id = _new_id()
|
|
upload_dir = _TEMP_DIR / session_id
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
saved_paths = []
|
|
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))
|
|
|
|
# 解析预设筛选条件（来自 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'),
|
|
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)
|
|
|
|
# 获取各关键列的 标准内部名（即重命名后的列名）
|
|
# 注意：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
|
|
|
|
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:
|
|
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,
|
|
'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, '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,
|
|
},
|
|
'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
|
|
|
|
|
|
@csrf_exempt
|
|
@require_POST
|
|
def advanced_filter(request):
|
|
"""高级筛选：接受 AND/OR/NOT 组合条件，返回筛选结果。"""
|
|
body = json.loads(request.body)
|
|
session_id = body.get('session_id')
|
|
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 = entry['df'] # 使用原始 df（不含预设筛选），预设筛选已应用
|
|
|
|
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': []})
|
|
|
|
try:
|
|
# 构建条件组
|
|
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: 聚类（GeoIP 驱动 + 性能优化 + 全噪声容错）
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@csrf_exempt
|
|
@require_POST
|
|
def run_clustering(request):
|
|
"""执行聚类分析：IP 子网模式或 GeoIP 地理位置模式。"""
|
|
body = json.loads(request.body)
|
|
session_id = body.get('session_id')
|
|
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.05))
|
|
min_records_per_subnet = int(body.get('min_records_per_subnet', 3))
|
|
max_cluster_size = int(body.get('max_cluster_size', 100))
|
|
|
|
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']
|
|
ips_col = meta.get('ips_col')
|
|
|
|
t_start = time.time()
|
|
clustered = False
|
|
result_data = {}
|
|
|
|
if mode == 'geo':
|
|
clustered, result_data = _cluster_geo(entry, filtered, meta, min_cluster_size, cluster_selection_epsilon)
|
|
else:
|
|
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 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.01):
|
|
"""使用 GeoIP 经纬度进行 HDBSCAN 聚类，含性能优化和全噪声容错。"""
|
|
|
|
# 从 GeoIP 获取经纬度，不再依赖 CSV 列
|
|
lats, lons, valid_mask = _get_geo_coords(entry, df, meta)
|
|
n_valid = int(valid_mask.sum())
|
|
total = len(df)
|
|
|
|
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个有效坐标），全噪声',
|
|
}
|
|
|
|
lat_vals = lats[valid_mask].astype(np.float32)
|
|
lon_vals = lons[valid_mask].astype(np.float32)
|
|
|
|
# ── 性能优化：大数据降采样 ────────────────────────────────────────────
|
|
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
|
|
|
|
# 弧度转换（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
|
|
|
|
df_with_labels = df.with_columns(pl.Series('_cluster_label', labels))
|
|
entry['filtered_df'] = df_with_labels
|
|
|
|
|
|
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': labels.tolist(),
|
|
'data_points': data_points_by_cluster,
|
|
'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 using GeoIP coordinates.
|
|
|
|
Subnets with fewer than *min_records_per_subnet* records are marked noise (-1).
|
|
Large subnets use GeoIP coordinates for automatic geo sub-clustering.
|
|
"""
|
|
ips_col = meta.get('ips_col')
|
|
if not ips_col:
|
|
return False, {'error': '缺少源IP列（:ips），无法执行子网聚类'}
|
|
|
|
try:
|
|
ip_series = df[ips_col].cast(pl.Utf8)
|
|
except Exception as e:
|
|
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:
|
|
if not ip or ip == 'None' or ip == '':
|
|
return None
|
|
parts = ip.split('.')
|
|
if len(parts) < 4:
|
|
return None
|
|
return '.'.join(parts[:3]) + '.0/24'
|
|
|
|
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)
|
|
|
|
labels = []
|
|
cluster_map: dict[int, dict] = {}
|
|
next_label = 0
|
|
subnet_to_label: dict[str, int] = {}
|
|
large_subnet_indices: dict[str, list[int]] = {}
|
|
|
|
for row_idx, s in enumerate(subnets):
|
|
if s is None:
|
|
labels.append(-1)
|
|
else:
|
|
cnt = subnet_count.get(s, 0)
|
|
if cnt < min_records_per_subnet:
|
|
labels.append(-1)
|
|
elif cnt > max_cluster_size:
|
|
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,
|
|
}
|
|
next_label += 1
|
|
large_subnet_indices[s] = []
|
|
lbl = subnet_to_label[s]
|
|
labels.append(lbl)
|
|
cluster_map[lbl]['size'] = cluster_map[lbl].get('size', 0) + 1
|
|
large_subnet_indices[s].append(row_idx)
|
|
else:
|
|
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,
|
|
}
|
|
next_label += 1
|
|
lbl = subnet_to_label[s]
|
|
labels.append(lbl)
|
|
cluster_map[lbl]['size'] = cluster_map[lbl].get('size', 0) + 1
|
|
|
|
# 大子网地理拆分（使用 GeoIP 坐标）
|
|
if large_subnet_indices and geo_valid.any():
|
|
for subnet_name, indices in large_subnet_indices.items():
|
|
old_label = subnet_to_label.get(subnet_name)
|
|
if old_label is None:
|
|
continue
|
|
cluster_map.pop(old_label, None)
|
|
try:
|
|
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=max(3, len(valid_sub) // 10),
|
|
min_samples=2, metric='haversine',
|
|
cluster_selection_epsilon=0.005,
|
|
)
|
|
sub_labels = sub_cluster.fit_predict(sub_coords)
|
|
valid_counter = 0
|
|
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[indices[orig_rank]] = -1
|
|
else:
|
|
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': scn, 'is_noise': False,
|
|
}
|
|
next_label += 1
|
|
nl = subnet_to_label[scn]
|
|
labels[indices[orig_rank]] = nl
|
|
cluster_map[nl]['size'] = cluster_map[nl].get('size', 0) + 1
|
|
else:
|
|
cluster_map[old_label] = {
|
|
'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-split failed %s: %s', subnet_name, e)
|
|
cluster_map[old_label] = {
|
|
'label': old_label, 'size': len(indices),
|
|
'subnet': subnet_name, 'is_noise': False,
|
|
}
|
|
|
|
# 更新簇中心（使用 GeoIP 坐标）
|
|
for lbl in list(cluster_map.keys()):
|
|
mask = np.array(labels) == lbl
|
|
if ips_col:
|
|
try:
|
|
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]['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
|
|
|
|
n_noise = labels.count(-1)
|
|
n_clusters = len(cluster_map)
|
|
|
|
# 数据点
|
|
data_points_by_cluster: dict[int, list] = {}
|
|
for lbl in list(cluster_map.keys()):
|
|
mask = np.array(labels) == lbl
|
|
pts = []
|
|
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
|
|
noise_mask = np.array(labels) == -1
|
|
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
|
|
|
|
df_with_labels = df.with_columns(pl.Series('_cluster_label', labels))
|
|
entry['cluster_labels'] = labels
|
|
entry['cluster_map'] = cluster_map
|
|
entry['cluster_mode'] = 'subnet'
|
|
entry['filtered_df'] = df_with_labels
|
|
entry['n_clusters'] = n_clusters
|
|
entry['n_noise'] = n_noise
|
|
|
|
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 + 边关系
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@require_GET
|
|
def cluster_detail(request, label: int):
|
|
"""返回指定簇的详细画像，含 IP 属性、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)
|
|
|
|
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')
|
|
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:
|
|
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
|
|
|
|
# 使用 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,
|
|
'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_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': {},
|
|
}
|
|
|
|
# 流量统计
|
|
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)
|
|
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:
|
|
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
|
|
|
|
# 时间范围
|
|
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
|
|
|
|
# 地理边界
|
|
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):
|
|
"""Render the simple analysis workbench page."""
|
|
return render(request, 'simple_analysis/simple_analysis.html')
|