TianXuan Developer
46013afa74
- Add independent 'simple_analysis' Django app with full workflow: 1. Upload/merge CSV files with auto column detection 2. Filter by cnam + cnrs + isrs (optional) 3. Clustering via HDBSCAN+haversine (geo) or /24 subnet (IP) 4. Interactive Leaflet global map with cluster-colored data points - Integrate into nav bar, settings, and URL routing without modifying existing analysis logic - Cluster detail panel with IPs, top dst IPs, TLS/cipher profiles - Chart.js bar/pie charts for cluster size distribution - E2E Playwright tests for all workflow steps - Support multiple CSV upload with diagonal_relaxed merge Co-Authored-By: Claude <noreply@anthropic.com>
797 lines
31 KiB
Python
797 lines
31 KiB
Python
"""简单分析模块 — 独立的上传→筛选→聚类→地图可视化工作流。
|
||
|
||
不与天璇原有分析模块共享数据库或 session store，所有中间数据
|
||
存储在模块级内存缓存（SimpleAnalysisCache）或临时目录中。
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import shutil
|
||
import tempfile
|
||
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
|
||
from sklearn.preprocessing import StandardScaler
|
||
from sklearn.cluster import DBSCAN
|
||
|
||
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)
|
||
|
||
|
||
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
|
||
|
||
|
||
# ── Column-name helpers ──────────────────────────────────────────────────────
|
||
|
||
def _norm(name: str) -> str:
|
||
"""Normalise column name: lower-case, strip whitespace, replace hyphens."""
|
||
return name.strip().lower().replace('-', '_').replace(' ', '_')
|
||
|
||
|
||
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 + '_'):
|
||
return orig
|
||
return None
|
||
|
||
|
||
# ── Step 1: Upload ───────────────────────────────────────────────────────────
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def upload_csv(request):
|
||
"""Upload one or more CSV files, merge, return column info + cnam frequency."""
|
||
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))
|
||
|
||
# Load & merge CSV files using Polars
|
||
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)
|
||
|
||
col_names = df.columns
|
||
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')
|
||
|
||
# CNAM frequency
|
||
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)
|
||
|
||
# Store metadata in cache
|
||
meta = {
|
||
'session_id': session_id,
|
||
'upload_dir': str(upload_dir),
|
||
'total_rows': total_rows,
|
||
'col_names': col_names,
|
||
'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,
|
||
}
|
||
_store(session_id, {'meta': meta, 'df': df})
|
||
|
||
return JsonResponse({
|
||
'session_id': session_id,
|
||
'total_rows': total_rows,
|
||
'columns': col_names[:50],
|
||
'detected_columns': {
|
||
'cnam': cnam_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,
|
||
},
|
||
'cnam_frequency': cnam_freq[:30],
|
||
'top_cnam': top_cnam,
|
||
})
|
||
|
||
|
||
# ── Step 2: Filter ───────────────────────────────────────────────────────────
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def filter_data(request):
|
||
"""Apply cnam + cnrs + isrs filters; return preview rows."""
|
||
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: pl.DataFrame = entry['df']
|
||
meta = entry['meta']
|
||
cnam_col = meta['cnam_col']
|
||
cnrs_col = meta['cnrs_col']
|
||
isrs_col = meta['isrs_col']
|
||
|
||
# Validate required columns
|
||
missing = []
|
||
if not cnam_col:
|
||
missing.append('cnam/snam (目标类别)')
|
||
|
||
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)
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'筛选失败: {e}'}, status=400)
|
||
|
||
filtered_rows = len(filtered)
|
||
meta['filtered_cnam'] = selected_cnam
|
||
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):
|
||
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: Cluster ──────────────────────────────────────────────────────────
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def run_clustering(request):
|
||
"""Run HDBSCAN geo-clustering or IP-subnet soft-clustering."""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
mode = body.get('mode', 'geo') # 'geo' or 'subnet'
|
||
min_cluster_size = int(body.get('min_cluster_size', 5))
|
||
cluster_selection_epsilon = float(body.get('cluster_selection_epsilon', 0.005))
|
||
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 filtered is None:
|
||
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']
|
||
|
||
if mode == 'geo':
|
||
result = _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
|
||
_store(session_id, entry)
|
||
return result
|
||
|
||
|
||
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']
|
||
|
||
if not src_lat_col or not src_lon_col:
|
||
return JsonResponse({'error': '缺少源经纬度列（:ips.latd / :ips.lond），无法执行地理聚类'}, status=400)
|
||
|
||
# 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),
|
||
}
|
||
|
||
# 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
|
||
|
||
# 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])
|
||
|
||
entry['cluster_labels'] = label_series.tolist()
|
||
entry['cluster_map'] = cluster_map
|
||
entry['cluster_mode'] = 'geo'
|
||
entry['n_clusters'] = n_clusters
|
||
entry['n_noise'] = n_noise
|
||
|
||
# Update filtered_df with cluster label column
|
||
df_with_labels = df.with_columns(pl.Series('_cluster_label', label_series))
|
||
entry['filtered_df'] = df_with_labels
|
||
|
||
return JsonResponse({
|
||
'n_clusters': n_clusters,
|
||
'n_noise': n_noise,
|
||
'n_valid_points': len(clean),
|
||
'clusters': cluster_map,
|
||
'labels': label_series.tolist(),
|
||
'data_points': data_points_by_cluster,
|
||
'fallback_suggested': fallback_suggested,
|
||
})
|
||
|
||
|
||
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.
|
||
|
||
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.
|
||
"""
|
||
ips_col = meta['ips_col']
|
||
src_lat_col = meta['src_lat_col']
|
||
src_lon_col = meta['src_lon_col']
|
||
if not ips_col:
|
||
return JsonResponse({'error': '缺少源IP列（:ips），无法执行子网聚类'}, status=400)
|
||
|
||
try:
|
||
ip_series = df[ips_col].cast(pl.Utf8)
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'IP列转换失败: {e}'}, status=400)
|
||
|
||
# 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)
|
||
|
||
# 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):
|
||
if s is None:
|
||
labels.append(-1)
|
||
else:
|
||
cnt = subnet_count.get(s, 0)
|
||
if cnt < min_records_per_subnet:
|
||
labels.append(-1) # noise (too few records)
|
||
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,
|
||
}
|
||
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
|
||
|
||
# ── Geo sub-clustering for large subnets ────────────────────────────────
|
||
if large_subnet_indices and src_lat_col and src_lon_col:
|
||
for subnet_name, indices in large_subnet_indices.items():
|
||
old_label = subnet_to_label[subnet_name]
|
||
# Remove the temporary large-subnet cluster entry
|
||
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_cluster = HDBSCAN(
|
||
min_cluster_size=min(3, len(sub_lats) // 5),
|
||
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)
|
||
valid_counter = 0
|
||
for orig_rank, orig_idx in enumerate(indices):
|
||
if not valid[orig_rank]:
|
||
labels[orig_idx] = -1 # no lat/lon → noise
|
||
else:
|
||
sl = int(sub_labels[valid_counter])
|
||
valid_counter += 1
|
||
if sl == -1:
|
||
labels[orig_idx] = -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
|
||
cluster_map[next_label] = {
|
||
'label': next_label,
|
||
'size': 0,
|
||
'subnet': sub_cluster_name,
|
||
'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
|
||
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,
|
||
}
|
||
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
|
||
cluster_map[old_label] = {
|
||
'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
|
||
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:
|
||
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)
|
||
except Exception:
|
||
cluster_map[lbl]['center_lat'] = None
|
||
cluster_map[lbl]['center_lon'] = None
|
||
else:
|
||
cluster_map[lbl]['center_lat'] = None
|
||
cluster_map[lbl]['center_lon'] = None
|
||
|
||
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
|
||
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
|
||
|
||
# Annotate DataFrame
|
||
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 JsonResponse({
|
||
'n_clusters': n_clusters,
|
||
'n_noise': n_noise,
|
||
'n_valid_points': len(df),
|
||
'clusters': cluster_map,
|
||
'labels': labels,
|
||
'data_points': data_points_by_cluster,
|
||
})
|
||
|
||
|
||
# ── Step 4: Cluster detail ───────────────────────────────────────────────────
|
||
|
||
@require_GET
|
||
def cluster_detail(request, label: int):
|
||
"""Return detailed profile for a specific cluster."""
|
||
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')
|
||
meta = entry.get('meta', {})
|
||
cluster_map = entry.get('cluster_map', {})
|
||
|
||
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, {})
|
||
|
||
# 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:
|
||
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()
|
||
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()
|
||
]
|
||
except Exception:
|
||
return []
|
||
|
||
# Source IPs
|
||
ips_col = meta.get('ips_col')
|
||
unique_ips = []
|
||
if ips_col and ips_col in cluster_df.columns:
|
||
try:
|
||
ip_vals = cluster_df[ips_col].cast(pl.Utf8).unique().to_list()
|
||
unique_ips = sorted(str(v) for v in ip_vals if v not in (None, '', 'None'))
|
||
except Exception:
|
||
pass
|
||
|
||
# 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')
|
||
|
||
profile = {
|
||
'label': label,
|
||
'size': rows,
|
||
'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 [],
|
||
'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
|
||
'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
|
||
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
|
||
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:
|
||
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),
|
||
}
|
||
except Exception:
|
||
pass
|
||
|
||
return JsonResponse(profile)
|
||
|
||
|
||
# ── Main page ────────────────────────────────────────────────────────────────
|
||
|
||
def index(request):
|
||
"""Render the simple analysis workbench page."""
|
||
return render(request, 'simple_analysis/simple_analysis.html')
|