TianXuan Developer
c49475d207
后端:
- traffic_timeline_view: 返回edges[{first_ts,last_ts}]替代windows
- datetime.strptime + .timestamp()保留微秒精度
前端:
- 完全移除窗口概念, edge._fs/_ls定义生命周期
- 时间桶索引(O(1)查找可见边)
- 生命周期<5s自动扩展±30s
- 5%软边缘淡入淡出
- 每帧仅检查当前时间桶±1的边(性能优化)
- 最大渲染500条(按count排序)
验证: 500条边, ~127K像素, 206活跃节点, 边持久可见
Co-Authored-By: Claude <noreply@anthropic.com>
2471 lines
97 KiB
Python
2471 lines
97 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'],
|
||
'host': ['host', 'hostname', 'host_name', 'remote_host', 'peer_name'],
|
||
}
|
||
|
||
|
||
def _norm(name: str) -> str:
|
||
"""归一化列名：去首尾空格，特殊字符→下划线，转小写。"""
|
||
n = name.strip().lower()
|
||
n = re.sub(r'[-:.\s]+', '_', n)
|
||
return n
|
||
|
||
|
||
def normalize_columns(df: pl.DataFrame) -> tuple[pl.DataFrame, dict[str, str]]:
|
||
"""对 DataFrame 列名进行统一规范化，返回 (新df, 规范化映射 {原始名→标准内部名})。
|
||
|
||
规范化规则：
|
||
1. 去首尾空格
|
||
2. 连续特殊字符（- : . 空格）→ 下划线
|
||
3. 全部小写
|
||
然后根据 _COLUMN_ALIASES 模糊匹配到标准内部名称。
|
||
"""
|
||
col_map = {} # 原始名 → 标准内部名
|
||
reverse_map = {} # norm(原始名) → 原始名
|
||
|
||
for c in df.columns:
|
||
nc = _norm(c)
|
||
reverse_map[nc] = c
|
||
|
||
# 遍历别名表，尝试匹配
|
||
used_originals = set()
|
||
for std_name, aliases in _COLUMN_ALIASES.items():
|
||
for alias in aliases:
|
||
n_alias = _norm(alias)
|
||
# 精确匹配
|
||
if n_alias in reverse_map and reverse_map[n_alias] not in used_originals:
|
||
col_map[reverse_map[n_alias]] = std_name
|
||
used_originals.add(reverse_map[n_alias])
|
||
break
|
||
# 后缀匹配（如列名 _ips 匹配 :ips）
|
||
for orig, n_orig in reverse_map.items():
|
||
if orig in used_originals:
|
||
continue
|
||
if n_orig == n_alias or n_orig.endswith('_' + n_alias) or n_orig.startswith(n_alias + '_'):
|
||
col_map[orig] = std_name
|
||
used_originals.add(orig)
|
||
break
|
||
if std_name in col_map.values():
|
||
break
|
||
|
||
# 未匹配的保留原名
|
||
for c in df.columns:
|
||
if c not in col_map:
|
||
col_map[c] = c
|
||
|
||
# 重命名
|
||
rename_dict = {orig: std for orig, std in col_map.items() if orig != std}
|
||
if rename_dict:
|
||
df = df.rename(rename_dict)
|
||
|
||
return df, col_map
|
||
|
||
|
||
def find_col(norm_col_map: dict[str, str], *std_names: str) -> str | None:
|
||
"""在规范化映射中查找第一个匹配的标准列名，返回原始列名。"""
|
||
for std in std_names:
|
||
for orig, mapped in norm_col_map.items():
|
||
if mapped == std:
|
||
return orig
|
||
return None
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# GeoIP 辅助（复用天璇离线库）
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def resolve_ip_geo(ip_str: str | None) -> dict:
|
||
"""用离线 GeoIP 库解析单个 IP，返回 {lat, lon, city, country} 或空字典。"""
|
||
if not ip_str or ip_str in ('None', '', '+', '-'):
|
||
return {}
|
||
try:
|
||
result = geoip_lookup(ip_str)
|
||
if result:
|
||
return result
|
||
except Exception:
|
||
pass
|
||
return {}
|
||
|
||
|
||
def batch_geoip_resolve(ip_list: list[str]) -> dict[str, dict]:
|
||
"""批量解析 IP 地理信息，返回 {ip: {lat, lon, city, country}}。"""
|
||
cache: dict[str, dict] = {}
|
||
for ip in set(ip_list):
|
||
if ip and ip not in ('None', '', '+', '-'):
|
||
result = geoip_lookup(ip)
|
||
if result:
|
||
cache[ip] = result
|
||
return cache
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# Step 0: Quick column scan (header-only, files kept for lazy value scan)
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def quick_scan(request):
|
||
"""仅读取 CSV 表头 + 保留文件供后续单列值扫描。
|
||
|
||
文件保存在 scan_dir，scan_id 存入 _cache。
|
||
后续 column_values 可用 scan_id 做单列 streaming Top-N 查询。
|
||
调用 upload_csv 时若提供 scan_id，可复用已保存的文件。
|
||
"""
|
||
files = request.FILES.getlist('files')
|
||
if not files:
|
||
return JsonResponse({'error': '未上传任何文件'}, status=400)
|
||
|
||
scan_id = _new_id()
|
||
scan_dir = _TEMP_DIR / scan_id
|
||
scan_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
saved_paths = []
|
||
for f in files:
|
||
dest = scan_dir / f.name
|
||
with open(dest, 'wb') as out:
|
||
for chunk in f.chunks():
|
||
out.write(chunk)
|
||
saved_paths.append(str(dest))
|
||
|
||
try:
|
||
# Read header only (1 row is enough for column names)
|
||
lf = pl.scan_csv(saved_paths[0], infer_schema_length=0)
|
||
if len(saved_paths) > 1:
|
||
frames = [pl.scan_csv(p, infer_schema_length=0) for p in saved_paths]
|
||
lf = pl.concat(frames, how='diagonal_relaxed')
|
||
df = lf.head(1).collect()
|
||
except Exception as e:
|
||
shutil.rmtree(scan_dir, ignore_errors=True)
|
||
return JsonResponse({'error': f'CSV 表头读取失败: {e}'}, status=400)
|
||
|
||
# Normalize column names
|
||
df_norm, col_map = normalize_columns(df)
|
||
|
||
# Keep files for lazy value scanning — store paths in cache
|
||
_store(scan_id, {
|
||
'scan_paths': saved_paths,
|
||
'col_map': col_map,
|
||
'columns': list(df_norm.columns),
|
||
})
|
||
|
||
return JsonResponse({
|
||
'scan_id': scan_id,
|
||
'columns': list(df_norm.columns),
|
||
'columns_std': list(df_norm.columns),
|
||
'col_map': {orig: std for orig, std in col_map.items() if orig != std},
|
||
})
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# Step 1: Upload + 预设筛选 + GeoIP 解析
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def upload_csv(request):
|
||
"""上传 CSV，列归一化，预设筛选（支持树状高级筛选），GeoIP 解析，返回摘要。
|
||
|
||
若提供 scan_id，复用 quick_scan 阶段已保存的文件，避免重复写入磁盘。
|
||
"""
|
||
scan_id = request.POST.get('scan_id', request.GET.get('scan_id', ''))
|
||
|
||
session_id = _new_id()
|
||
upload_dir = _TEMP_DIR / session_id
|
||
|
||
saved_paths = []
|
||
|
||
# 复用 scan 阶段的文件（避免重复存储）
|
||
if scan_id:
|
||
scan_entry = _cached(scan_id)
|
||
if scan_entry and scan_entry.get('scan_paths'):
|
||
scan_dir = Path(scan_entry['scan_paths'][0]).parent
|
||
# Move the scan directory to the session directory
|
||
shutil.move(str(scan_dir), str(upload_dir))
|
||
# Rebuild saved_paths from new location
|
||
for p in upload_dir.iterdir():
|
||
if p.is_file():
|
||
saved_paths.append(str(p))
|
||
# Clean scan cache
|
||
_cache.pop(scan_id, None)
|
||
else:
|
||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Fallback: normal file upload
|
||
if not saved_paths:
|
||
files = request.FILES.getlist('files')
|
||
if not files:
|
||
return JsonResponse({'error': '未上传任何文件'}, status=400)
|
||
upload_dir.mkdir(parents=True, exist_ok=True)
|
||
for f in files:
|
||
dest = upload_dir / f.name
|
||
with open(dest, 'wb') as out:
|
||
for chunk in f.chunks():
|
||
out.write(chunk)
|
||
saved_paths.append(str(dest))
|
||
|
||
# 解析预设筛选条件（支持树状结构 + 旧格式兼容）
|
||
preset_root = None
|
||
try:
|
||
pf_raw = request.POST.get('preset_filters', request.GET.get('preset_filters', '{}'))
|
||
if pf_raw:
|
||
pf = json.loads(pf_raw)
|
||
# 新格式：树状 root
|
||
if 'root' in pf:
|
||
preset_root = pf['root']
|
||
# 旧格式兼容：{cnrs: "+", ...}
|
||
elif any(k in pf for k in ('cnrs', 'isrs', 'cnam', 'snam')):
|
||
# Convert legacy to tree
|
||
items = []
|
||
for key in ('cnrs', 'isrs', 'cnam', 'snam'):
|
||
if key in pf and pf[key]:
|
||
if key in ('cnrs', 'isrs'):
|
||
items.append({'type': 'condition', 'col': key, 'op': 'eq', 'val': pf[key]})
|
||
else:
|
||
items.append({'type': 'condition', 'col': key, 'op': 'contains', 'val': pf[key]})
|
||
if items:
|
||
preset_root = {'type': 'group', 'logic': 'AND', 'items': items}
|
||
except (json.JSONDecodeError, Exception):
|
||
pass
|
||
|
||
# Load & merge CSV
|
||
try:
|
||
lf = pl.scan_csv(
|
||
saved_paths[0] if len(saved_paths) == 1 else str(upload_dir / '*.csv'),
|
||
infer_schema_length=10000,
|
||
try_parse_dates=True,
|
||
)
|
||
if len(saved_paths) > 1:
|
||
frames = [pl.scan_csv(p, infer_schema_length=10000) for p in saved_paths]
|
||
lf = pl.concat(frames, how='diagonal_relaxed')
|
||
df = lf.collect(streaming=True)
|
||
except Exception as e:
|
||
shutil.rmtree(upload_dir, ignore_errors=True)
|
||
return JsonResponse({'error': f'CSV 解析失败: {e}'}, status=400)
|
||
|
||
# 列名规范化
|
||
df, col_map = normalize_columns(df)
|
||
total_rows = len(df)
|
||
|
||
# 获取各关键列的 标准内部名（即重命名后的列名）
|
||
def _get_std(*names):
|
||
"""从 col_map 中查找第一个匹配的标准名。"""
|
||
for n in names:
|
||
for orig, std in col_map.items():
|
||
if std == n:
|
||
return std
|
||
return None
|
||
|
||
ips_col = _get_std('ips')
|
||
ipd_col = _get_std('ipd')
|
||
cnam_col = _get_std('cnam')
|
||
cnrs_col = _get_std('cnrs')
|
||
isrs_col = _get_std('isrs')
|
||
snam_col = _get_std('snam')
|
||
dport_col = _get_std('dport')
|
||
proto_col = _get_std('proto')
|
||
tls_ver_col = _get_std('tls_ver')
|
||
tls_cph_col = _get_std('tls_cph')
|
||
bytes_col = _get_std('bytes')
|
||
packets_col = _get_std('packets')
|
||
dur_col = _get_std('duration')
|
||
time_col = _get_std('time')
|
||
ispn_col = _get_std('ispn')
|
||
orgn_col = _get_std('orgn')
|
||
city_col = _get_std('city')
|
||
scnt_col = _get_std('scnt')
|
||
text_col = _get_std('text')
|
||
host_col = _get_std('host')
|
||
|
||
# ── 预设筛选（树状高级筛选） ──────────────────────────────────────────
|
||
preset_filtered = total_rows
|
||
if preset_root is not None:
|
||
try:
|
||
cond = _build_tree_condition(preset_root, df)
|
||
if cond is not None:
|
||
df = df.filter(cond)
|
||
preset_filtered = len(df)
|
||
except Exception as e:
|
||
logger.warning('Preset filter error: %s', e)
|
||
|
||
# ── GeoIP 批量解析（使用标准化列名） ────────────────────────────────────
|
||
geo_cache: dict[str, dict] = {}
|
||
geo_skipped = 0
|
||
if ips_col and ips_col in df.columns:
|
||
all_ips = df[ips_col].cast(pl.Utf8).unique().to_list()
|
||
geo_cache = batch_geoip_resolve([ip for ip in all_ips if ip not in (None, '', 'None', '+', '-')])
|
||
logger.info('GeoIP resolved %d IPs', len(geo_cache))
|
||
|
||
# ── CNAM 频率 ────────────────────────────────────────────────────────────
|
||
cnam_freq = []
|
||
top_cnam = None
|
||
if cnam_col:
|
||
try:
|
||
cnam_series = df[cnam_col].cast(pl.Utf8).fill_null('')
|
||
counts = Counter(cnam_series.to_list())
|
||
total_cnam = sum(counts.values())
|
||
for val, cnt in counts.most_common():
|
||
cnam_freq.append({
|
||
'value': val if val else '(空)',
|
||
'count': cnt,
|
||
'pct': round(cnt / total_cnam * 100, 1),
|
||
})
|
||
if cnam_freq:
|
||
top_cnam = cnam_freq[0]
|
||
except Exception as e:
|
||
logger.warning('cnam frequency failed: %s', e)
|
||
|
||
# 存储元数据
|
||
meta = {
|
||
'session_id': session_id,
|
||
'total_rows': total_rows,
|
||
'preset_filtered': preset_filtered,
|
||
'col_names': list(df.columns),
|
||
'col_map': col_map, # {原始名→标准内部名}
|
||
'geo_cache': geo_cache, # {ip: {lat, lon, city, country}}
|
||
'geo_skipped': geo_skipped,
|
||
'ips_col': ips_col,
|
||
'ipd_col': ipd_col,
|
||
'cnam_col': cnam_col,
|
||
'cnrs_col': cnrs_col,
|
||
'isrs_col': isrs_col,
|
||
'snam_col': snam_col,
|
||
'dport_col': dport_col,
|
||
'proto_col': proto_col,
|
||
'tls_ver_col': tls_ver_col,
|
||
'tls_cph_col': tls_cph_col,
|
||
'bytes_col': bytes_col,
|
||
'packets_col': packets_col,
|
||
'dur_col': dur_col,
|
||
'time_col': time_col,
|
||
'ispn_col': ispn_col,
|
||
'orgn_col': orgn_col,
|
||
'city_col': city_col,
|
||
'scnt_col': scnt_col,
|
||
'text_col': text_col,
|
||
'host_col': host_col,
|
||
'snam_col': snam_col,
|
||
}
|
||
_store(session_id, {'meta': meta, 'df': df, 'geo_cache': geo_cache})
|
||
|
||
return JsonResponse({
|
||
'session_id': session_id,
|
||
'total_rows': total_rows,
|
||
'preset_filtered': preset_filtered,
|
||
'columns': list(df.columns)[:50],
|
||
'columns_std': list(df.columns), # 标准化后列名，供前端筛选器使用
|
||
'col_map': {orig: std for orig, std in col_map.items() if orig != std},
|
||
'detected_columns': {
|
||
'cnam': cnam_col,
|
||
'snam': snam_col,
|
||
'cnrs': cnrs_col,
|
||
'isrs': isrs_col,
|
||
'ips': ips_col,
|
||
'ipd': ipd_col,
|
||
'ispn': ispn_col,
|
||
'orgn': orgn_col,
|
||
'city': city_col,
|
||
'host': host_col,
|
||
'text': text_col,
|
||
},
|
||
'cnam_frequency': cnam_freq[:30],
|
||
'top_cnam': top_cnam,
|
||
'geo_stats': {
|
||
'ips_resolved': len(geo_cache),
|
||
},
|
||
})
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# Step 2: 高级筛选构建器（AND/OR/NOT 组合条件）
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
def _build_condition(col_name: str, operator: str, value: str, logic: str = 'AND'):
|
||
"""构建单个筛选条件表达式。"""
|
||
col = pl.col(col_name)
|
||
v = value.strip()
|
||
try:
|
||
if operator == 'eq':
|
||
cond = col.cast(pl.Utf8) == v
|
||
elif operator == 'neq':
|
||
cond = col.cast(pl.Utf8) != v
|
||
elif operator == 'contains':
|
||
cond = col.cast(pl.Utf8).str.contains(v)
|
||
elif operator == 'not_contains':
|
||
cond = ~col.cast(pl.Utf8).str.contains(v)
|
||
elif operator == 'gt':
|
||
cond = col.cast(pl.Float64) > float(v)
|
||
elif operator == 'lt':
|
||
cond = col.cast(pl.Float64) < float(v)
|
||
elif operator == 'gte':
|
||
cond = col.cast(pl.Float64) >= float(v)
|
||
elif operator == 'lte':
|
||
cond = col.cast(pl.Float64) <= float(v)
|
||
elif operator == 'is_empty':
|
||
cond = col.is_null() | (col.cast(pl.Utf8) == '')
|
||
elif operator == 'not_empty':
|
||
cond = col.is_not_null() & (col.cast(pl.Utf8) != '')
|
||
else:
|
||
cond = col.cast(pl.Utf8) == v
|
||
except Exception:
|
||
cond = col.cast(pl.Utf8) == v
|
||
return cond
|
||
|
||
|
||
def _build_tree_condition(node, df) -> pl.Expr | None:
|
||
"""递归构建条件树，支持嵌套 AND/OR/NOT 组。
|
||
|
||
JSON 节点格式:
|
||
{"type": "condition", "col": "...", "op": "...", "val": "..."}
|
||
{"type": "group", "logic": "AND|OR", "items": [...]}
|
||
{"type": "group", "logic": "NOT", "items": [...]} # 否定整个组
|
||
"""
|
||
if node is None:
|
||
return None
|
||
t = node.get('type', 'condition')
|
||
|
||
if t == 'condition':
|
||
col = node.get('col', '')
|
||
if not col or col not in df.columns:
|
||
return None
|
||
c = _build_condition(col, node.get('op', 'eq'), node.get('val', ''))
|
||
if node.get('logic') == 'NOT':
|
||
c = ~c
|
||
return c
|
||
|
||
if t == 'group':
|
||
items = node.get('items', [])
|
||
if not items:
|
||
return None
|
||
logic = node.get('logic', 'AND')
|
||
conds = []
|
||
for item in items:
|
||
c = _build_tree_condition(item, df)
|
||
if c is not None:
|
||
conds.append(c)
|
||
if not conds:
|
||
return None
|
||
if logic == 'NOT':
|
||
combined = conds[0]
|
||
for c in conds[1:]:
|
||
combined = combined & c # NOT 组内默认 AND 组合
|
||
return ~combined
|
||
combined = conds[0]
|
||
for c in conds[1:]:
|
||
if logic == 'OR':
|
||
combined = combined | c
|
||
else:
|
||
combined = combined & c
|
||
return combined
|
||
|
||
return None
|
||
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def advanced_filter(request):
|
||
"""高级筛选：接受树状 AND/OR/NOT 分组条件，支持嵌套括号语义。"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
root = body.get('root') # 新格式：树状结构
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期，请重新上传'}, status=400)
|
||
|
||
df = entry['df']
|
||
|
||
try:
|
||
if root:
|
||
# 新格式：递归构建条件树
|
||
combined = _build_tree_condition(root, df)
|
||
if combined is None:
|
||
entry['filtered_df'] = df
|
||
entry['meta']['filtered_rows'] = len(df)
|
||
return JsonResponse({'filtered_rows': len(df), 'total_before_filter': len(df), 'preview_columns': df.columns[:30], 'preview_rows': []})
|
||
else:
|
||
# 旧格式（向后兼容）
|
||
conditions = body.get('conditions', [])
|
||
global_logic = body.get('global_logic', 'AND')
|
||
if not conditions:
|
||
entry['filtered_df'] = df
|
||
entry['meta']['filtered_rows'] = len(df)
|
||
return JsonResponse({'filtered_rows': len(df), 'total_before_filter': len(df), 'preview_columns': df.columns[:30], 'preview_rows': []})
|
||
|
||
group_conds = []
|
||
for cond_item in conditions:
|
||
col = cond_item.get('col', '')
|
||
op = cond_item.get('op', 'eq')
|
||
val = cond_item.get('val', '')
|
||
if not col or col not in df.columns:
|
||
continue
|
||
c = _build_condition(col, op, val)
|
||
if cond_item.get('logic') == 'NOT':
|
||
c = ~c
|
||
group_conds.append(c)
|
||
|
||
if not group_conds:
|
||
return JsonResponse({'error': '无可用的筛选条件'}, status=400)
|
||
|
||
if global_logic == 'OR':
|
||
combined = group_conds[0]
|
||
for c in group_conds[1:]:
|
||
combined = combined | c
|
||
else:
|
||
combined = group_conds[0]
|
||
for c in group_conds[1:]:
|
||
combined = combined & c
|
||
|
||
filtered = df.filter(combined)
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'筛选构建失败: {e}'}, status=400)
|
||
|
||
filtered_rows = len(filtered)
|
||
entry['meta']['filtered_rows'] = filtered_rows
|
||
entry['filtered_df'] = filtered
|
||
|
||
preview_cols = [c for c in filtered.columns if not c.startswith('_')][:30]
|
||
preview_rows = []
|
||
for row in filtered.head(20).select(preview_cols).iter_rows(named=True):
|
||
clean = {}
|
||
for k, v in row.items():
|
||
if v is None or (isinstance(v, float) and np.isnan(v)):
|
||
clean[k] = None
|
||
else:
|
||
clean[k] = str(v)[:120]
|
||
preview_rows.append(clean)
|
||
|
||
return JsonResponse({
|
||
'filtered_rows': filtered_rows,
|
||
'preview_columns': preview_cols,
|
||
'preview_rows': preview_rows,
|
||
'total_before_filter': len(df),
|
||
})
|
||
|
||
|
||
# ── 兼容原 filter 接口 ──────────────────────────────────────────────────────
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def filter_data(request):
|
||
"""兼容旧版简单筛选接口，内部调用高级筛选。"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
selected_cnam = body.get('selected_cnam', '')
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期，请重新上传'}, status=400)
|
||
|
||
df = entry['df']
|
||
meta = entry['meta']
|
||
cnam_col = meta.get('cnam_col')
|
||
cnrs_col = meta.get('cnrs_col')
|
||
isrs_col = meta.get('isrs_col')
|
||
|
||
if not cnam_col:
|
||
return JsonResponse({'error': '缺少 cnam/snam 列', 'missing_columns': ['cnam']}, status=400)
|
||
|
||
try:
|
||
conditions = [{'col': cnam_col, 'op': 'eq', 'val': selected_cnam}]
|
||
if cnrs_col:
|
||
conditions.append({'col': cnrs_col, 'op': 'contains', 'val': '+'})
|
||
if isrs_col:
|
||
conditions.append({'col': isrs_col, 'op': 'contains', 'val': '+'})
|
||
|
||
combined = None
|
||
for c in conditions:
|
||
cond = _build_condition(c['col'], c['op'], c['val'])
|
||
combined = cond if combined is None else (combined & cond)
|
||
|
||
filtered = df.filter(combined) if combined else df
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'筛选失败: {e}'}, status=400)
|
||
|
||
filtered_rows = len(filtered)
|
||
meta['filtered_rows'] = filtered_rows
|
||
entry['filtered_df'] = filtered
|
||
|
||
preview_cols = [c for c in filtered.columns if not c.startswith('_')][:30]
|
||
preview_rows = []
|
||
for row in filtered.head(20).select(preview_cols).iter_rows(named=True):
|
||
clean = {}
|
||
for k, v in row.items():
|
||
if v is None or (isinstance(v, float) and np.isnan(v)):
|
||
clean[k] = None
|
||
else:
|
||
clean[k] = str(v)[:120]
|
||
preview_rows.append(clean)
|
||
|
||
return JsonResponse({
|
||
'filtered_rows': filtered_rows,
|
||
'preview_columns': preview_cols,
|
||
'preview_rows': preview_rows,
|
||
'total_before_filter': len(df),
|
||
})
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# Step 3: 聚类 — IP 通信网络社区发现
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def run_clustering(request):
|
||
"""IP 通信网络社区发现聚类。
|
||
|
||
构建 IP 通信图（节点=唯一IP，边=聚合通信记录），
|
||
按 /24 子网前缀粗分组，子网内用 HDBSCAN (haversine) 二次聚类。
|
||
"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
min_cluster_size = int(body.get('min_cluster_size', 5))
|
||
cluster_selection_epsilon = float(body.get('cluster_selection_epsilon', 0.05))
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期，请重新上传'}, status=400)
|
||
|
||
filtered = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
|
||
if filtered is None:
|
||
return JsonResponse({'error': '请先执行筛选'}, status=400)
|
||
|
||
meta = entry['meta']
|
||
t_start = time.time()
|
||
|
||
result_data = _cluster_network(entry, filtered, meta, min_cluster_size, cluster_selection_epsilon)
|
||
|
||
elapsed = round(time.time() - t_start, 2)
|
||
result_data['elapsed_seconds'] = elapsed
|
||
_store(session_id, entry)
|
||
return JsonResponse(result_data)
|
||
|
||
|
||
def _cluster_network(entry: dict, df: pl.DataFrame, meta: dict,
|
||
min_cluster_size: int, cluster_selection_epsilon: float) -> dict:
|
||
"""构建 IP 通信网络并进行社区发现聚类。
|
||
|
||
流程：
|
||
1. 从数据中提取所有唯一 IP（源+目的），去重
|
||
2. 通过 GeoIP 获取每个 IP 的经纬度和属性
|
||
3. 按 /24 子网前缀将 IP 分组
|
||
4. 每个子网内，若节点数 > min_cluster_size，用 HDBSCAN (haversine) 二次聚类
|
||
5. 不同子网的节点不会被合并
|
||
6. 构建边列表（IP 对之间的聚合通信统计）
|
||
"""
|
||
ips_col = meta.get('ips_col')
|
||
ipd_col = meta.get('ipd_col')
|
||
time_col = meta.get('time_col')
|
||
bytes_col = meta.get('bytes_col')
|
||
host_col = meta.get('host_col')
|
||
snam_col = meta.get('snam_col')
|
||
text_col = meta.get('text_col')
|
||
orgn_col = meta.get('orgn_col')
|
||
ispn_col = meta.get('ispn_col')
|
||
city_col = meta.get('city_col')
|
||
dport_col = meta.get('dport_col')
|
||
dur_col = meta.get('dur_col')
|
||
cnrs_col = meta.get('cnrs_col')
|
||
tls_ver_col = meta.get('tls_ver_col')
|
||
tls_cph_col = meta.get('tls_cph_col')
|
||
cnam_col = meta.get('cnam_col')
|
||
packets_col = meta.get('packets_col')
|
||
geo_cache = entry.get('geo_cache', {})
|
||
|
||
if not ips_col or not ipd_col:
|
||
return {'error': '缺少源/目的IP列，无法构建通信网络'}
|
||
if ips_col not in df.columns or ipd_col not in df.columns:
|
||
return {'error': 'IP列不在数据中'}
|
||
|
||
# ── 1. 提取唯一 IP 列表 ──────────────────────────────────────────────
|
||
src_ips = df[ips_col].cast(pl.Utf8).to_list()
|
||
dst_ips = df[ipd_col].cast(pl.Utf8).to_list()
|
||
|
||
# Collect all unique IPs
|
||
unique_ips_set = set()
|
||
for ip in src_ips:
|
||
if ip and ip not in ('None', '', '+', '-'):
|
||
unique_ips_set.add(ip)
|
||
for ip in dst_ips:
|
||
if ip and ip not in ('None', '', '+', '-'):
|
||
unique_ips_set.add(ip)
|
||
|
||
unique_ips = sorted(unique_ips_set)
|
||
n_nodes = len(unique_ips)
|
||
ip_to_idx = {ip: i for i, ip in enumerate(unique_ips)}
|
||
|
||
logger.info('Network clustering: %d unique IPs from %d records', n_nodes, len(df))
|
||
|
||
# ── 2. GeoIP 解析所有 IP ───────────────────────────────────────────
|
||
# MMDB优先（高精度），upload时的geo_cache兜底（低精度文本回退）
|
||
node_geo = {} # ip → {lat, lon, city, country, isp, org}
|
||
for ip in unique_ips:
|
||
info = geoip_lookup(ip) # 优先用当前加载的MMDB数据库
|
||
if not info:
|
||
info = geo_cache.get(ip) # 回退到上传时的缓存
|
||
if info:
|
||
node_geo[ip] = info
|
||
|
||
geo_resolved = len(node_geo)
|
||
logger.info('GeoIP resolved %d / %d IPs', geo_resolved, n_nodes)
|
||
|
||
# ── 3. 按 /24 子网分组 ─────────────────────────────────────────────
|
||
def _subnet24(ip: str) -> str | None:
|
||
parts = ip.split('.')
|
||
if len(parts) != 4:
|
||
return None
|
||
try:
|
||
return '.'.join(parts[:3]) + '.0/24'
|
||
except Exception:
|
||
return None
|
||
|
||
subnet_nodes: dict[str, list[str]] = {}
|
||
nodes_without_geo = []
|
||
for ip in unique_ips:
|
||
sn = _subnet24(ip)
|
||
if sn is None:
|
||
nodes_without_geo.append(ip)
|
||
continue
|
||
if sn not in subnet_nodes:
|
||
subnet_nodes[sn] = []
|
||
subnet_nodes[sn].append(ip)
|
||
|
||
# ── 4. 每个子网内 HDBSCAN 聚类 ───────────────────────────────────
|
||
cluster_labels: dict[str, int] = {} # ip → cluster_id
|
||
cluster_id_counter = 0
|
||
cluster_info: dict[int, dict] = {} # cluster_id → {size, center_lat, center_lon, subnet, nodes}
|
||
|
||
for sn, ips in subnet_nodes.items():
|
||
# 收集该子网内有 GeoIP 坐标的节点
|
||
geo_ips = []
|
||
coords_list = []
|
||
for ip in ips:
|
||
info = node_geo.get(ip)
|
||
if info and info.get('lat') is not None and info.get('lon') is not None:
|
||
geo_ips.append(ip)
|
||
coords_list.append([np.radians(float(info['lat'])), np.radians(float(info['lon']))])
|
||
|
||
n_in_subnet = len(ips)
|
||
n_with_geo = len(geo_ips)
|
||
|
||
# 小规模子网 → 直接作为一个簇
|
||
if n_in_subnet <= min_cluster_size or n_with_geo < 3:
|
||
cid = cluster_id_counter
|
||
cluster_id_counter += 1
|
||
for ip in ips:
|
||
cluster_labels[ip] = cid
|
||
|
||
# 计算中心坐标
|
||
center_lat, center_lon = None, None
|
||
if coords_list:
|
||
coords_arr = np.array(coords_list, dtype=np.float32)
|
||
center_lat = round(float(np.degrees(np.median(coords_arr[:, 0]))), 4)
|
||
center_lon = round(float(np.degrees(np.median(coords_arr[:, 1]))), 4)
|
||
|
||
cluster_info[cid] = {
|
||
'label': cid, 'size': n_in_subnet, 'n_geo': n_with_geo,
|
||
'subnet': sn, 'center_lat': center_lat, 'center_lon': center_lon,
|
||
}
|
||
continue
|
||
|
||
# 大规模子网 → HDBSCAN 二次聚类
|
||
coords_arr = np.array(coords_list, dtype=np.float32)
|
||
n_coords = len(coords_arr)
|
||
|
||
# 去重坐标加微小抖动
|
||
if n_coords > 10:
|
||
_, unique_idx = np.unique(coords_arr, axis=0, return_index=True)
|
||
if len(unique_idx) < n_coords * 0.8:
|
||
jitter = np.random.RandomState(42).normal(0, 1e-6, coords_arr.shape).astype(np.float32)
|
||
coords_arr = coords_arr + jitter
|
||
|
||
try:
|
||
sub_clusterer = HDBSCAN(
|
||
min_cluster_size=max(2, min(n_coords // 20, min_cluster_size)),
|
||
min_samples=2,
|
||
metric='haversine',
|
||
cluster_selection_epsilon=cluster_selection_epsilon,
|
||
)
|
||
sub_labels = sub_clusterer.fit_predict(coords_arr)
|
||
except Exception as e:
|
||
logger.warning('Subnet %s HDBSCAN failed: %s — treating as single cluster', sn, e)
|
||
cid = cluster_id_counter
|
||
cluster_id_counter += 1
|
||
for ip in ips:
|
||
cluster_labels[ip] = cid
|
||
cluster_info[cid] = {
|
||
'label': cid, 'size': n_in_subnet, 'n_geo': n_with_geo,
|
||
'subnet': sn, 'center_lat': None, 'center_lon': None,
|
||
}
|
||
continue
|
||
|
||
# 分配子标签（每个噪声节点独立成簇）
|
||
unique_sl = set(sub_labels)
|
||
for sl in unique_sl:
|
||
mask = sub_labels == sl
|
||
subset_ips = [geo_ips[i] for i in range(n_coords) if mask[i]]
|
||
|
||
if sl == -1:
|
||
# 无法归类的节点：每个单独成簇
|
||
for ip in subset_ips:
|
||
cid = cluster_id_counter
|
||
cluster_id_counter += 1
|
||
cluster_labels[ip] = cid
|
||
cluster_info[cid] = {
|
||
'label': cid, 'size': 1, 'n_geo': 1,
|
||
'subnet': sn,
|
||
'center_lat': round(float(node_geo[ip].get('lat', 0)), 4) if ip in node_geo else None,
|
||
'center_lon': round(float(node_geo[ip].get('lon', 0)), 4) if ip in node_geo else None,
|
||
}
|
||
else:
|
||
cid = cluster_id_counter
|
||
cluster_id_counter += 1
|
||
for ip in subset_ips:
|
||
cluster_labels[ip] = cid
|
||
|
||
sub_coords = coords_arr[mask]
|
||
center_lat = round(float(np.degrees(np.median(sub_coords[:, 0]))), 4)
|
||
center_lon = round(float(np.degrees(np.median(sub_coords[:, 1]))), 4)
|
||
|
||
cluster_info[cid] = {
|
||
'label': cid, 'size': len(subset_ips), 'n_geo': len(subset_ips),
|
||
'subnet': sn, 'center_lat': center_lat, 'center_lon': center_lon,
|
||
}
|
||
|
||
# 子网内无 GeoIP 的节点 → 分配到子网的主簇，否则各自成簇
|
||
non_geo_ips = [ip for ip in ips if ip not in geo_ips]
|
||
if non_geo_ips:
|
||
subnet_cids = [cid for cid, info in cluster_info.items()
|
||
if info.get('subnet') == sn]
|
||
if subnet_cids:
|
||
main_cid = min(subnet_cids, key=lambda c: cluster_info[c]['label'])
|
||
for ip in non_geo_ips:
|
||
cluster_labels[ip] = main_cid
|
||
cluster_info[main_cid]['size'] += len(non_geo_ips)
|
||
else:
|
||
for ip in non_geo_ips:
|
||
cid = cluster_id_counter
|
||
cluster_id_counter += 1
|
||
cluster_labels[ip] = cid
|
||
cluster_info[cid] = {'label': cid, 'size': 1, 'n_geo': 0, 'subnet': sn,
|
||
'center_lat': None, 'center_lon': None}
|
||
|
||
# 无子网的节点 → 各自成簇
|
||
for ip in nodes_without_geo:
|
||
cid = cluster_id_counter
|
||
cluster_id_counter += 1
|
||
cluster_labels[ip] = cid
|
||
cluster_info[cid] = {'label': cid, 'size': 1, 'n_geo': 0, 'subnet': None,
|
||
'center_lat': None, 'center_lon': None}
|
||
|
||
# ── 5. 构建节点列表 ──────────────────────────────────────────────
|
||
# 检测服务器 IP（在 dst_ip 中出现过的 IP）
|
||
server_ips = set()
|
||
for ip in dst_ips:
|
||
if ip and ip not in ('None', '', '+', '-'):
|
||
server_ips.add(ip)
|
||
|
||
# ── 收集 per-IP 元数据 (host, text, org, peers, TLS, traffic) ────
|
||
ip_hosts: dict[str, str] = {} # ip → 最常见的 host
|
||
ip_texts: dict[str, list] = {} # ip → 前几条 payload 样本
|
||
ip_orgs: dict[str, str] = {} # ip → 最常见的 org
|
||
ip_peers: dict[str, list] = {} # ip → 对端 IP 列表
|
||
ip_snis: dict[str, list] = {} # ip → SNI/域名列表
|
||
ip_tls_vers: dict[str, list] = {} # ip → TLS 版本列表
|
||
ip_ciphers: dict[str, list] = {} # ip → 加密套件列表
|
||
ip_dports: dict[str, list] = {} # ip → 目的端口列表
|
||
ip_categories: dict[str, list] = {} # ip → 分类标签列表
|
||
ip_total_bytes: dict[str, float] = {} # ip → 总字节数
|
||
ip_total_pkts: dict[str, int] = {} # ip → 总包数
|
||
ip_durations: dict[str, list] = {} # ip → 持续时间列表
|
||
ip_conn_ok: dict[str, int] = {} # ip → 成功连接次数
|
||
ip_conn_total: dict[str, int] = {} # ip → 总连接次数
|
||
ip_first_ts: dict[str, str] = {} # ip → 首次出现时间
|
||
ip_last_ts: dict[str, str] = {} # ip → 最后出现时间
|
||
|
||
scan_limit = min(len(df), 50000)
|
||
# Pre-extract column lists for faster access
|
||
_host_col = host_col or snam_col
|
||
_has_host = _host_col and _host_col in df.columns
|
||
_has_text = text_col and text_col in df.columns
|
||
_has_orgn = orgn_col and orgn_col in df.columns
|
||
_has_snam = snam_col and snam_col in df.columns
|
||
_has_tls_ver = tls_ver_col and tls_ver_col in df.columns
|
||
_has_tls_cph = tls_cph_col and tls_cph_col in df.columns
|
||
_has_dport = dport_col and dport_col in df.columns
|
||
_has_cnam = cnam_col and cnam_col in df.columns
|
||
_has_bytes = bytes_col and bytes_col in df.columns
|
||
_has_pkts = packets_col and packets_col in df.columns if packets_col else False
|
||
_has_dur = dur_col and dur_col in df.columns
|
||
_has_cnrs = cnrs_col and cnrs_col in df.columns
|
||
_has_time = time_col and time_col in df.columns
|
||
|
||
for i in range(scan_limit):
|
||
sip = str(src_ips[i]) if i < len(src_ips) and src_ips[i] not in (None, '', 'None') else None
|
||
if not sip or sip not in unique_ips_set:
|
||
continue
|
||
try:
|
||
if _has_host:
|
||
hval = str(df[_host_col][i]) if df[_host_col][i] is not None else None
|
||
if hval and hval not in ('None', '', 'nan'):
|
||
ip_hosts.setdefault(sip, []).append(hval)
|
||
if _has_text:
|
||
tval = str(df[text_col][i]) if df[text_col][i] is not None else None
|
||
if tval and tval not in ('None', '', 'nan'):
|
||
lst = ip_texts.setdefault(sip, [])
|
||
if len(lst) < 3:
|
||
lst.append(tval[:200])
|
||
if _has_orgn:
|
||
oval = str(df[orgn_col][i]) if df[orgn_col][i] is not None else None
|
||
if oval and oval not in ('None', '', 'nan'):
|
||
ip_orgs.setdefault(sip, []).append(oval)
|
||
if _has_snam:
|
||
sval = str(df[snam_col][i]) if df[snam_col][i] is not None else None
|
||
if sval and sval not in ('None', '', 'nan', '+', '-'):
|
||
ip_snis.setdefault(sip, []).append(sval)
|
||
if _has_tls_ver:
|
||
tv = str(df[tls_ver_col][i]) if df[tls_ver_col][i] is not None else None
|
||
if tv and tv not in ('None', '', 'nan'):
|
||
ip_tls_vers.setdefault(sip, []).append(tv)
|
||
if _has_tls_cph:
|
||
tc = str(df[tls_cph_col][i]) if df[tls_cph_col][i] is not None else None
|
||
if tc and tc not in ('None', '', 'nan'):
|
||
ip_ciphers.setdefault(sip, []).append(tc[:50])
|
||
if _has_dport:
|
||
dp = str(df[dport_col][i]) if df[dport_col][i] is not None else None
|
||
if dp and dp not in ('None', '', 'nan'):
|
||
ip_dports.setdefault(sip, []).append(dp)
|
||
if _has_cnam:
|
||
cv = str(df[cnam_col][i]) if df[cnam_col][i] is not None else None
|
||
if cv and cv not in ('None', '', 'nan', '+', '-'):
|
||
ip_categories.setdefault(sip, []).append(cv)
|
||
if _has_bytes:
|
||
try:
|
||
bv = float(df[bytes_col][i] or 0)
|
||
ip_total_bytes[sip] = ip_total_bytes.get(sip, 0.0) + bv
|
||
except Exception:
|
||
pass
|
||
if _has_pkts:
|
||
try:
|
||
pv = int(df[packets_col][i] or 0)
|
||
ip_total_pkts[sip] = ip_total_pkts.get(sip, 0) + pv
|
||
except Exception:
|
||
pass
|
||
if _has_dur:
|
||
try:
|
||
dv = float(df[dur_col][i] or 0)
|
||
if dv > 0:
|
||
ip_durations.setdefault(sip, []).append(dv)
|
||
except Exception:
|
||
pass
|
||
if _has_cnrs:
|
||
cv = str(df[cnrs_col][i]) if df[cnrs_col][i] is not None else ''
|
||
ip_conn_total[sip] = ip_conn_total.get(sip, 0) + 1
|
||
if cv in ('+', '1', 'true', 'True', 'success'):
|
||
ip_conn_ok[sip] = ip_conn_ok.get(sip, 0) + 1
|
||
if _has_time:
|
||
tv = str(df[time_col][i]) if df[time_col][i] is not None else ''
|
||
if tv and tv not in ('None', '', 'nan'):
|
||
if sip not in ip_first_ts or tv < ip_first_ts[sip]:
|
||
ip_first_ts[sip] = tv
|
||
if sip not in ip_last_ts or tv > ip_last_ts[sip]:
|
||
ip_last_ts[sip] = tv
|
||
except Exception:
|
||
pass
|
||
|
||
# 取每 IP 最常见的值
|
||
def _most_common(lst: list, top_n: int = 8) -> list:
|
||
if not lst: return []
|
||
cnt = Counter(lst)
|
||
return [{'value': k, 'count': v} for k, v in cnt.most_common(top_n)]
|
||
|
||
def _top1(lst: list) -> str:
|
||
if not lst: return ''
|
||
return Counter(lst).most_common(1)[0][0]
|
||
|
||
for ip, hosts in ip_hosts.items():
|
||
ip_hosts[ip] = _top1(hosts) if isinstance(hosts, list) else hosts
|
||
for ip, orgs in ip_orgs.items():
|
||
ip_orgs[ip] = _top1(orgs) if isinstance(orgs, list) else orgs
|
||
|
||
# ── 5. 构建节点列表 ──────────────────────────────────────────────
|
||
# Build peer IP lists from raw src/dst pairs
|
||
for i in range(min(len(df), 50000)):
|
||
sip = str(src_ips[i]) if i < len(src_ips) and src_ips[i] not in (None, '', 'None') else None
|
||
dip = str(dst_ips[i]) if i < len(dst_ips) and dst_ips[i] not in (None, '', 'None') else None
|
||
if sip and dip:
|
||
ip_peers.setdefault(sip, set()).add(dip)
|
||
ip_peers.setdefault(dip, set()).add(sip)
|
||
|
||
nodes = []
|
||
for ip in unique_ips:
|
||
info = node_geo.get(ip, {})
|
||
cid = cluster_labels.get(ip, -1)
|
||
peer_set = ip_peers.get(ip, set())
|
||
dur_list = ip_durations.get(ip, [])
|
||
conn_total = ip_conn_total.get(ip, 0)
|
||
conn_ok = ip_conn_ok.get(ip, 0)
|
||
nodes.append({
|
||
'ip': ip,
|
||
'cluster_id': int(cid),
|
||
'lat': info.get('lat'),
|
||
'lon': info.get('lon'),
|
||
'city': info.get('city', ''),
|
||
'country': info.get('country', ''),
|
||
'region': info.get('region', ''),
|
||
'isp': info.get('isp', ''),
|
||
'asn': info.get('asn', ''),
|
||
'is_server': ip in server_ips,
|
||
'host': ip_hosts.get(ip, info.get('host', '')),
|
||
'org': ip_orgs.get(ip, info.get('org', '')),
|
||
'texts': ip_texts.get(ip, [])[:3],
|
||
'peer_ips': sorted(peer_set)[:20],
|
||
'peer_count': len(peer_set),
|
||
# Enriched attributes
|
||
'top_snis': _most_common(ip_snis.get(ip, []), 5),
|
||
'top_tls_vers': _most_common(ip_tls_vers.get(ip, []), 5),
|
||
'top_ciphers': _most_common(ip_ciphers.get(ip, []), 5),
|
||
'top_ports': _most_common(ip_dports.get(ip, []), 5),
|
||
'top_categories': _most_common(ip_categories.get(ip, []), 5),
|
||
'total_bytes': round(ip_total_bytes.get(ip, 0.0), 2),
|
||
'total_packets': ip_total_pkts.get(ip, 0),
|
||
'avg_duration': round(sum(dur_list) / len(dur_list), 2) if dur_list else None,
|
||
'conn_success_rate': round(conn_ok / conn_total * 100, 1) if conn_total > 0 else None,
|
||
'conn_count': conn_total,
|
||
'first_seen': ip_first_ts.get(ip, ''),
|
||
'last_seen': ip_last_ts.get(ip, ''),
|
||
})
|
||
|
||
# ── 6. 构建边列表（IP 对聚合通信统计） ──────────────────────────
|
||
pair_data: dict[tuple[str, str], dict] = {}
|
||
|
||
# Pre-extract column lists
|
||
bytes_list = None
|
||
if bytes_col and bytes_col in df.columns:
|
||
try:
|
||
bytes_list = df[bytes_col].cast(pl.Float64).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
times_list = None
|
||
if time_col and time_col in df.columns:
|
||
try:
|
||
times_list = df[time_col].cast(pl.Utf8).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
pkts_list = None
|
||
if packets_col and packets_col in df.columns:
|
||
try:
|
||
pkts_list = df[packets_col].cast(pl.Float64).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
dur_list = None
|
||
if dur_col and dur_col in df.columns:
|
||
try:
|
||
dur_list = df[dur_col].cast(pl.Float64).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
dport_list = None
|
||
if dport_col and dport_col in df.columns:
|
||
try:
|
||
dport_list = df[dport_col].cast(pl.Utf8).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
tls_ver_list = None
|
||
if tls_ver_col and tls_ver_col in df.columns:
|
||
try:
|
||
tls_ver_list = df[tls_ver_col].cast(pl.Utf8).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
cph_list = None
|
||
if tls_cph_col and tls_cph_col in df.columns:
|
||
try:
|
||
cph_list = df[tls_cph_col].cast(pl.Utf8).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
snam_list = None
|
||
if snam_col and snam_col in df.columns:
|
||
try:
|
||
snam_list = df[snam_col].cast(pl.Utf8).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
cnam_list = None
|
||
if cnam_col and cnam_col in df.columns:
|
||
try:
|
||
cnam_list = df[cnam_col].cast(pl.Utf8).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
for i in range(len(df)):
|
||
s = str(src_ips[i]) if src_ips[i] not in (None, '', 'None') else None
|
||
d = str(dst_ips[i]) if dst_ips[i] not in (None, '', 'None') else None
|
||
if s is None or d is None:
|
||
continue
|
||
key = (s, d)
|
||
if key not in pair_data:
|
||
pair_data[key] = {
|
||
'count': 0, 'total_bytes': 0.0, 'total_packets': 0.0,
|
||
'times': [], 'durations': [], 'ports': [],
|
||
'tls_vers': [], 'ciphers': [], 'snis': [], 'categories': [],
|
||
}
|
||
pair_data[key]['count'] += 1
|
||
if bytes_list and i < len(bytes_list):
|
||
try:
|
||
pair_data[key]['total_bytes'] += float(bytes_list[i] or 0)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if times_list and i < len(times_list):
|
||
t = times_list[i]
|
||
if t and t not in ('None', ''):
|
||
pair_data[key]['times'].append(t)
|
||
if pkts_list and i < len(pkts_list):
|
||
try:
|
||
pair_data[key]['total_packets'] += float(pkts_list[i] or 0)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if dur_list and i < len(dur_list):
|
||
try:
|
||
dv = float(dur_list[i] or 0)
|
||
if dv > 0:
|
||
pair_data[key]['durations'].append(dv)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if dport_list and i < len(dport_list):
|
||
pv = dport_list[i]
|
||
if pv and pv not in ('None', '', 'nan'):
|
||
pair_data[key]['ports'].append(pv)
|
||
if tls_ver_list and i < len(tls_ver_list):
|
||
tv = tls_ver_list[i]
|
||
if tv and tv not in ('None', '', 'nan'):
|
||
pair_data[key]['tls_vers'].append(tv)
|
||
if cph_list and i < len(cph_list):
|
||
cv = cph_list[i]
|
||
if cv and cv not in ('None', '', 'nan'):
|
||
pair_data[key]['ciphers'].append(cv[:50])
|
||
if snam_list and i < len(snam_list):
|
||
sv = snam_list[i]
|
||
if sv and sv not in ('None', '', 'nan', '+', '-'):
|
||
pair_data[key]['snis'].append(sv)
|
||
if cnam_list and i < len(cnam_list):
|
||
catv = cnam_list[i]
|
||
if catv and catv not in ('None', '', 'nan', '+', '-'):
|
||
pair_data[key]['categories'].append(catv)
|
||
|
||
# Build IP metadata lookup for edges
|
||
ip_meta = {}
|
||
for n in nodes:
|
||
ip_meta[n['ip']] = n
|
||
|
||
edges = []
|
||
for (src, dst), data in pair_data.items():
|
||
sm = ip_meta.get(src, {})
|
||
dm = ip_meta.get(dst, {})
|
||
durs = data.get('durations', [])
|
||
edge = {
|
||
'source': src,
|
||
'target': dst,
|
||
'comm_count': data['count'],
|
||
'total_bytes': round(data['total_bytes'], 2),
|
||
'total_packets': round(data.get('total_packets', 0), 2),
|
||
'avg_duration': round(sum(durs) / len(durs), 2) if durs else None,
|
||
'top_ports': _most_common(data.get('ports', []), 5),
|
||
'top_tls_vers': _most_common(data.get('tls_vers', []), 5),
|
||
'top_ciphers': _most_common(data.get('ciphers', []), 5),
|
||
'top_snis': _most_common(data.get('snis', []), 5),
|
||
'top_categories': _most_common(data.get('categories', []), 5),
|
||
'src_host': sm.get('host', ''),
|
||
'dst_host': dm.get('host', ''),
|
||
'src_org': sm.get('org', ''),
|
||
'dst_org': dm.get('org', ''),
|
||
'src_city': sm.get('city', ''),
|
||
'dst_city': dm.get('city', ''),
|
||
'src_country': sm.get('country', ''),
|
||
'dst_country': dm.get('country', ''),
|
||
'src_cluster': sm.get('cluster_id', -1),
|
||
'dst_cluster': dm.get('cluster_id', -1),
|
||
}
|
||
# 时间统计
|
||
times = data.get('times', [])
|
||
if times:
|
||
times_sorted = sorted(times)
|
||
edge['first_seen'] = times_sorted[0][:25]
|
||
edge['last_seen'] = times_sorted[-1][:25]
|
||
# 时间规律：按小时统计
|
||
hour_counts = [0] * 24
|
||
for t_str in times:
|
||
try:
|
||
# Try ISO format or simple format
|
||
t_clean = t_str.strip()
|
||
if 'T' in t_clean or ' ' in t_clean:
|
||
# Extract hour
|
||
sep = 'T' if 'T' in t_clean else ' '
|
||
time_part = t_clean.split(sep)[-1]
|
||
hour = int(time_part.split(':')[0])
|
||
if 0 <= hour < 24:
|
||
hour_counts[hour] += 1
|
||
except Exception:
|
||
pass
|
||
peak_hours = sorted(range(24), key=lambda h: hour_counts[h], reverse=True)[:3]
|
||
edge['peak_hours'] = ','.join(f'{h:02d}' for h in peak_hours)
|
||
edge['time_count'] = len(times)
|
||
else:
|
||
edge['first_seen'] = ''
|
||
edge['last_seen'] = ''
|
||
edge['peak_hours'] = ''
|
||
edge['time_count'] = 0
|
||
edges.append(edge)
|
||
|
||
# 按通信次数排序，限制返回前 500 条边
|
||
edges.sort(key=lambda e: e['comm_count'], reverse=True)
|
||
edges = edges[:500]
|
||
|
||
# ── 7. 存储结果 ──────────────────────────────────────────────────
|
||
n_clusters = len(cluster_info)
|
||
|
||
entry['cluster_labels'] = cluster_labels # ip → cluster_id
|
||
entry['cluster_map'] = cluster_info
|
||
entry['cluster_mode'] = 'network'
|
||
entry['n_clusters'] = n_clusters
|
||
entry['nodes'] = nodes
|
||
entry['edges_data'] = edges
|
||
entry['cluster_info'] = {'note': ''}
|
||
|
||
return {
|
||
'n_clusters': n_clusters,
|
||
'n_nodes': n_nodes,
|
||
'n_edges': len(edges),
|
||
'geo_resolved': geo_resolved,
|
||
'nodes': nodes,
|
||
'edges': edges,
|
||
'clusters': {str(k): v for k, v in cluster_info.items()},
|
||
'note': f'{n_nodes} 个IP节点, {len(edges)} 条边, {n_clusters} 个社区',
|
||
}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# Step 4: Cluster detail + Node detail + 边关系
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
@require_GET
|
||
def cluster_detail(request, label: int):
|
||
"""返回指定簇的详细画像（节点列表、GeoIP 属性、流量特征等）。"""
|
||
session_id = request.GET.get('session_id')
|
||
if not session_id:
|
||
return JsonResponse({'error': '缺少 session_id'}, status=400)
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
nodes = entry.get('nodes', [])
|
||
edges_data = entry.get('edges_data', [])
|
||
cluster_info = entry.get('cluster_map', {}).get(label, {})
|
||
|
||
if not nodes:
|
||
# Old format fallback
|
||
return _cluster_detail_legacy(entry, label)
|
||
|
||
# Filter nodes by cluster
|
||
cluster_nodes = [n for n in nodes if n.get('cluster_id') == label]
|
||
if not cluster_nodes:
|
||
return JsonResponse({'error': f'簇 {label} 无数据'}, status=404)
|
||
|
||
# Collect IPs in this cluster
|
||
cluster_ips = set(n['ip'] for n in cluster_nodes)
|
||
|
||
# Find edges where both ends are in this cluster (internal)
|
||
internal_edges = []
|
||
related_edges = []
|
||
for e in edges_data:
|
||
src_in = e['source'] in cluster_ips
|
||
dst_in = e['target'] in cluster_ips
|
||
if src_in and dst_in:
|
||
internal_edges.append(e)
|
||
elif src_in or dst_in:
|
||
related_edges.append(e)
|
||
|
||
# Geo bounds
|
||
lats = [n['lat'] for n in cluster_nodes if n.get('lat') is not None]
|
||
lons = [n['lon'] for n in cluster_nodes if n.get('lon') is not None]
|
||
geo_bounds = None
|
||
if lats and lons:
|
||
geo_bounds = {
|
||
'lat_min': round(min(lats), 4), 'lat_max': round(max(lats), 4),
|
||
'lon_min': round(min(lons), 4), 'lon_max': round(max(lons), 4),
|
||
}
|
||
|
||
# ISP / city / country distribution
|
||
def _dist(items, key):
|
||
counts = Counter(i.get(key, '') or '(未知)' for i in items)
|
||
total = sum(counts.values()) or 1
|
||
return [{'value': k, 'count': v, 'pct': round(v / total * 100, 1)}
|
||
for k, v in counts.most_common(10)]
|
||
|
||
# Total internal traffic
|
||
total_internal_bytes = sum(e.get('total_bytes', 0) or 0 for e in internal_edges)
|
||
total_related_bytes = sum(e.get('total_bytes', 0) or 0 for e in related_edges)
|
||
|
||
profile = {
|
||
'label': label,
|
||
'size': len(cluster_nodes),
|
||
'cluster_info': cluster_info,
|
||
'unique_ips': sorted(cluster_ips)[:100],
|
||
'unique_ips_count': len(cluster_ips),
|
||
'ip_attributes': [{
|
||
'ip': n['ip'], 'lat': n.get('lat'), 'lon': n.get('lon'),
|
||
'city': n.get('city', ''), 'country': n.get('country', ''),
|
||
'isp': n.get('isp', ''), 'is_server': n.get('is_server', False),
|
||
} for n in cluster_nodes[:50]],
|
||
'top_cities': _dist(cluster_nodes, 'city'),
|
||
'top_countries': _dist(cluster_nodes, 'country'),
|
||
'top_isps': _dist(cluster_nodes, 'isp'),
|
||
'internal_edges': internal_edges[:100],
|
||
'related_edges': related_edges[:100],
|
||
'n_internal_edges': len(internal_edges),
|
||
'n_related_edges': len(related_edges),
|
||
'total_internal_bytes': round(total_internal_bytes, 2),
|
||
'total_related_bytes': round(total_related_bytes, 2),
|
||
'geo_bounds': geo_bounds,
|
||
}
|
||
|
||
return JsonResponse(profile)
|
||
|
||
|
||
def _cluster_detail_legacy(entry: dict, label: int) -> JsonResponse:
|
||
"""Fallback cluster detail for old record-based clustering format."""
|
||
df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
|
||
meta = entry.get('meta', {})
|
||
cluster_map = entry.get('cluster_map', {})
|
||
geo_cache = entry.get('geo_cache', {})
|
||
|
||
if df is None:
|
||
return JsonResponse({'error': '数据未就绪'}, status=400)
|
||
if '_cluster_label' not in df.columns:
|
||
return JsonResponse({'error': '未执行聚类'}, status=400)
|
||
|
||
cluster_df = df.filter(pl.col('_cluster_label') == label)
|
||
rows = len(cluster_df)
|
||
if rows == 0:
|
||
return JsonResponse({'error': f'簇 {label} 无数据'}, status=404)
|
||
|
||
cluster_info = cluster_map.get(label, {})
|
||
|
||
def _top_values(col_name: str, top_n: int = 5) -> list:
|
||
if not col_name or col_name not in df.columns:
|
||
return []
|
||
try:
|
||
vals = (
|
||
cluster_df[col_name].cast(pl.Utf8).fill_null('(空)')
|
||
.value_counts().sort('count', descending=True).head(top_n)
|
||
)
|
||
total = vals['count'].sum()
|
||
result = []
|
||
for row in vals.iter_rows():
|
||
result.append({
|
||
'value': str(row[0])[:80],
|
||
'count': int(row[1]),
|
||
'pct': round(int(row[1]) / total * 100, 1) if total else 0,
|
||
})
|
||
return result
|
||
except Exception:
|
||
return []
|
||
|
||
ips_col = meta.get('ips_col')
|
||
ipd_col = meta.get('ipd_col')
|
||
dport_col = meta.get('dport_col')
|
||
tls_ver_col = meta.get('tls_ver_col')
|
||
snam_col = meta.get('snam_col')
|
||
ispn_col = meta.get('ispn')
|
||
orgn_col = meta.get('orgn')
|
||
city_col = meta.get('city')
|
||
time_col = meta.get('time_col')
|
||
bytes_col = meta.get('bytes_col')
|
||
|
||
unique_ips = []
|
||
if ips_col and ips_col in cluster_df.columns:
|
||
try:
|
||
ip_vals = cluster_df[ips_col].cast(pl.Utf8).unique().to_list()
|
||
unique_ips = sorted(str(v) for v in ip_vals if v not in (None, '', 'None'))
|
||
except Exception:
|
||
pass
|
||
|
||
ip_attrs = []
|
||
for ip in unique_ips[:50]:
|
||
info = geo_cache.get(ip) or geoip_lookup(ip) or {}
|
||
ip_attrs.append({
|
||
'ip': ip,
|
||
'lat': info.get('lat'), 'lon': info.get('lon'),
|
||
'city': info.get('city', ''), 'country': info.get('country', ''),
|
||
})
|
||
|
||
profile = {
|
||
'label': label, 'size': rows,
|
||
'cluster_info': cluster_info,
|
||
'unique_ips': unique_ips[:100],
|
||
'unique_ips_count': len(unique_ips),
|
||
'ip_attributes': ip_attrs,
|
||
'top_dst_ips': _top_values(ipd_col) if ipd_col else [],
|
||
'top_dst_ports': _top_values(dport_col) if dport_col else [],
|
||
'top_tls_versions': _top_values(tls_ver_col) if tls_ver_col else [],
|
||
'top_sni': _top_values(snam_col) if snam_col else [],
|
||
'top_isp': _top_values(ispn_col) if ispn_col else [],
|
||
'top_org': _top_values(orgn_col) if orgn_col else [],
|
||
'top_cities': _top_values(city_col) if city_col else [],
|
||
'traffic': {},
|
||
}
|
||
|
||
if bytes_col and bytes_col in cluster_df.columns:
|
||
try:
|
||
profile['traffic']['total_bytes'] = round(float(cluster_df[bytes_col].cast(pl.Float64).sum()), 2)
|
||
except Exception:
|
||
pass
|
||
|
||
if time_col and time_col in cluster_df.columns:
|
||
try:
|
||
times = cluster_df[time_col].cast(pl.Utf8).drop_nulls()
|
||
if len(times) > 0:
|
||
profile['time_range'] = {'first': str(times.min())[:25], 'last': str(times.max())[:25]}
|
||
except Exception:
|
||
pass
|
||
|
||
return JsonResponse(profile)
|
||
|
||
|
||
@require_GET
|
||
def node_detail(request):
|
||
"""返回单个 IP 节点的详细属性及其关联边。"""
|
||
session_id = request.GET.get('session_id')
|
||
ip = request.GET.get('ip', '')
|
||
if not session_id or not ip:
|
||
return JsonResponse({'error': '缺少 session_id 或 ip'}, status=400)
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
nodes = entry.get('nodes', [])
|
||
edges_data = entry.get('edges_data', [])
|
||
|
||
# Find the node
|
||
node = next((n for n in nodes if n.get('ip') == ip), None)
|
||
if not node:
|
||
# Fallback: try GeoIP direct lookup
|
||
info = geoip_lookup(ip)
|
||
if info:
|
||
node = {
|
||
'ip': ip,
|
||
'cluster_id': -1,
|
||
'lat': info.get('lat'), 'lon': info.get('lon'),
|
||
'city': info.get('city', ''), 'country': info.get('country', ''),
|
||
'isp': info.get('isp', ''), 'is_server': False,
|
||
}
|
||
else:
|
||
return JsonResponse({'error': f'IP {ip} 不在网络中'}, status=404)
|
||
|
||
# Find related edges
|
||
related = [e for e in edges_data if e.get('source') == ip or e.get('target') == ip]
|
||
related.sort(key=lambda e: e.get('comm_count', 0), reverse=True)
|
||
|
||
# Peer IPs with metadata
|
||
peers = set()
|
||
for e in related:
|
||
if e.get('source') == ip:
|
||
peers.add(e.get('target'))
|
||
else:
|
||
peers.add(e.get('source'))
|
||
|
||
# Enrich related edges with peer metadata
|
||
enriched_edges = []
|
||
for e in related[:100]:
|
||
ee = dict(e)
|
||
if e.get('source') == ip:
|
||
ee['peer_ip'] = e.get('target', '')
|
||
ee['peer_host'] = e.get('dst_host', '')
|
||
ee['peer_org'] = e.get('dst_org', '')
|
||
ee['peer_city'] = e.get('dst_city', '')
|
||
else:
|
||
ee['peer_ip'] = e.get('source', '')
|
||
ee['peer_host'] = e.get('src_host', '')
|
||
ee['peer_org'] = e.get('src_org', '')
|
||
ee['peer_city'] = e.get('src_city', '')
|
||
enriched_edges.append(ee)
|
||
|
||
total_bytes = sum(e.get('total_bytes', 0) or 0 for e in related)
|
||
|
||
return JsonResponse({
|
||
'node': node,
|
||
'node_texts': node.get('texts', []),
|
||
'related_edges': enriched_edges,
|
||
'n_edges': len(related),
|
||
'peer_ips': sorted(peers)[:100],
|
||
'n_peers': len(peers),
|
||
'total_flow_bytes': round(total_bytes, 2),
|
||
})
|
||
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def edge_relationships(request):
|
||
"""返回边关系数据（使用预计算的边列表）。"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
min_freq = int(body.get('min_freq', 1))
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
edges_data = entry.get('edges_data', [])
|
||
|
||
if not edges_data:
|
||
# Fallback: compute on-the-fly from old format
|
||
return _edge_relationships_legacy(entry, body)
|
||
|
||
# Filter by min frequency
|
||
filtered = [e for e in edges_data if e.get('comm_count', 0) >= min_freq]
|
||
filtered.sort(key=lambda e: e.get('comm_count', 0), reverse=True)
|
||
|
||
return JsonResponse({
|
||
'edges': filtered[:500],
|
||
'total_edges': len(filtered),
|
||
'min_freq': min_freq,
|
||
})
|
||
|
||
|
||
def _edge_relationships_legacy(entry: dict, body: dict) -> JsonResponse:
|
||
"""Fallback edge computation for old record-based format."""
|
||
min_freq = int(body.get('min_freq', 10))
|
||
df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
|
||
meta = entry.get('meta', {})
|
||
if df is None:
|
||
return JsonResponse({'error': '数据未就绪'}, status=400)
|
||
|
||
ips_col = meta.get('ips_col')
|
||
ipd_col = meta.get('ipd_col')
|
||
time_col = meta.get('time_col')
|
||
bytes_col = meta.get('bytes_col')
|
||
|
||
if not ips_col or not ipd_col:
|
||
return JsonResponse({'error': '缺少源/目的IP列'}, status=400)
|
||
if ips_col not in df.columns or ipd_col not in df.columns:
|
||
return JsonResponse({'error': 'IP列不在数据中'}, status=400)
|
||
|
||
try:
|
||
src_ips = df[ips_col].cast(pl.Utf8).to_list()
|
||
dst_ips = df[ipd_col].cast(pl.Utf8).to_list()
|
||
bytes_list = df[bytes_col].cast(pl.Float64).to_list() if bytes_col and bytes_col in df.columns else None
|
||
times_list = df[time_col].cast(pl.Utf8).to_list() if time_col and time_col in df.columns else None
|
||
|
||
pair_counts: dict[tuple[str, str], int] = {}
|
||
pair_bytes: dict[tuple[str, str], float] = {}
|
||
pair_times: dict[tuple[str, str], list[str]] = {}
|
||
|
||
for i in range(len(df)):
|
||
s = str(src_ips[i]) if src_ips[i] not in (None, '', 'None') else '?'
|
||
d = str(dst_ips[i]) if dst_ips[i] not in (None, '', 'None') else '?'
|
||
key = (s, d)
|
||
pair_counts[key] = pair_counts.get(key, 0) + 1
|
||
if bytes_list and i < len(bytes_list):
|
||
try:
|
||
pair_bytes[key] = pair_bytes.get(key, 0.0) + float(bytes_list[i] or 0)
|
||
except (ValueError, TypeError):
|
||
pass
|
||
if times_list and i < len(times_list):
|
||
v = times_list[i]
|
||
if v and v not in ('None', ''):
|
||
pair_times.setdefault(key, []).append(v)
|
||
|
||
edges = []
|
||
for (src, dst), count in pair_counts.items():
|
||
if count < min_freq:
|
||
continue
|
||
edge = {
|
||
'source': src, 'target': dst,
|
||
'comm_count': count,
|
||
'total_bytes': round(pair_bytes.get((src, dst), 0), 2),
|
||
}
|
||
times = pair_times.get((src, dst), [])
|
||
if times:
|
||
times_sorted = sorted(times)
|
||
edge['first_seen'] = times_sorted[0][:25]
|
||
edge['last_seen'] = times_sorted[-1][:25]
|
||
edge['time_count'] = len(times)
|
||
edges.append(edge)
|
||
|
||
edges.sort(key=lambda e: e['comm_count'], reverse=True)
|
||
edges = edges[:200]
|
||
|
||
return JsonResponse({
|
||
'edges': edges,
|
||
'total_pairs': len(pair_counts),
|
||
'total_edges': len(edges),
|
||
'min_freq': min_freq,
|
||
})
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'边关系计算失败: {e}'}, status=400)
|
||
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def edge_time_series(request):
|
||
"""返回指定边的时间序列数据（按时间桶聚合），用于绘制流量-时间图。
|
||
|
||
POST body: session_id, source, target, bucket (可选: 'auto'|'hour'|'minute'|'day')
|
||
返回: {time_series: [{t: "yyyy-mm-dd HH:MM (北京)", bytes: N, count: N}, ...], span_hours: N}
|
||
"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
source = body.get('source', '')
|
||
target = body.get('target', '')
|
||
bucket = body.get('bucket', 'auto')
|
||
|
||
if not session_id or not source or not target:
|
||
return JsonResponse({'error': '缺少 session_id / source / target'}, status=400)
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
|
||
if df is None:
|
||
return JsonResponse({'error': '数据未就绪'}, status=400)
|
||
|
||
meta = entry.get('meta', {})
|
||
ips_col = meta.get('ips_col')
|
||
ipd_col = meta.get('ipd_col')
|
||
time_col = meta.get('time_col')
|
||
bytes_col = meta.get('bytes_col')
|
||
|
||
if not ips_col or not ipd_col:
|
||
return JsonResponse({'error': '缺少源/目的IP列'}, status=400)
|
||
if not time_col or time_col not in df.columns:
|
||
return JsonResponse({'error': '缺少时间列，无法绘制时间序列'}, status=400)
|
||
|
||
# ── 1. 筛选该边的所有行 ──
|
||
try:
|
||
mask = (
|
||
(df[ips_col].cast(pl.Utf8) == source) &
|
||
(df[ipd_col].cast(pl.Utf8) == target)
|
||
)
|
||
edge_rows = df.filter(mask)
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'数据筛选失败: {e}'}, status=400)
|
||
|
||
n_rows = len(edge_rows)
|
||
if n_rows == 0:
|
||
return JsonResponse({'error': '未找到该边的原始记录'}, status=400)
|
||
|
||
# ── 2. 提取时间列并转为 datetime ──
|
||
try:
|
||
times_series = edge_rows[time_col].cast(pl.Utf8)
|
||
except Exception:
|
||
return JsonResponse({'error': '时间列读取失败'}, status=400)
|
||
|
||
# 提取 bytes（如果有的话）
|
||
has_bytes = bytes_col and bytes_col in df.columns
|
||
bytes_series = None
|
||
if has_bytes:
|
||
try:
|
||
bytes_series = edge_rows[bytes_col].cast(pl.Float64)
|
||
except Exception:
|
||
has_bytes = False
|
||
|
||
# ── 3. 解析时间戳为 datetime ──
|
||
raw_times = times_series.to_list()
|
||
raw_bytes = bytes_series.to_list() if has_bytes else []
|
||
parsed_times = []
|
||
for row_idx, t_str in enumerate(raw_times):
|
||
try:
|
||
t_clean = str(t_str).strip() if t_str else ''
|
||
if not t_clean or t_clean in ('None', '', 'nan'):
|
||
continue
|
||
# 尝试多种格式
|
||
parsed = None
|
||
for fmt in [
|
||
'%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S',
|
||
'%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S.%f',
|
||
'%Y/%m/%d %H:%M:%S', '%Y/%m/%dT%H:%M:%S',
|
||
'%d/%m/%Y %H:%M:%S', '%m/%d/%Y %H:%M:%S',
|
||
]:
|
||
try:
|
||
parsed = time.strptime(t_clean[:26], fmt)
|
||
break
|
||
except ValueError:
|
||
continue
|
||
if parsed is None:
|
||
# Fallback: try to extract Y-M-D H:M:S via regex
|
||
m = re.search(r'(\d{4})-(\d{2})-(\d{2})[T ](\d{2}):(\d{2}):(\d{2})', t_clean)
|
||
if m:
|
||
parsed = time.strptime(f"{m.group(1)}-{m.group(2)}-{m.group(3)} "
|
||
f"{m.group(4)}:{m.group(5)}:{m.group(6)}",
|
||
'%Y-%m-%d %H:%M:%S')
|
||
if parsed is None:
|
||
continue
|
||
# 转为北京时间 (UTC+8)
|
||
import calendar
|
||
ts_utc = calendar.timegm(parsed)
|
||
ts_bj = ts_utc + 8 * 3600 # 北京时间 = UTC+8
|
||
parsed_times.append((ts_bj, row_idx))
|
||
except Exception:
|
||
continue
|
||
|
||
if not parsed_times:
|
||
return JsonResponse({'error': '无法解析时间戳'}, status=400)
|
||
|
||
# ── 4. 确定时间桶粒度 ──
|
||
min_ts = min(t[0] for t in parsed_times)
|
||
max_ts = max(t[0] for t in parsed_times)
|
||
span_seconds = max_ts - min_ts
|
||
span_hours = span_seconds / 3600.0
|
||
|
||
if bucket == 'auto':
|
||
if span_hours <= 2:
|
||
bucket_seconds = 60 # 1 分钟桶
|
||
elif span_hours <= 24:
|
||
bucket_seconds = 300 # 5 分钟桶
|
||
elif span_hours <= 72:
|
||
bucket_seconds = 900 # 15 分钟桶
|
||
elif span_hours <= 168:
|
||
bucket_seconds = 3600 # 1 小时桶
|
||
else:
|
||
bucket_seconds = 14400 # 4 小时桶
|
||
elif bucket == 'minute':
|
||
bucket_seconds = 60
|
||
elif bucket == 'hour':
|
||
bucket_seconds = 3600
|
||
elif bucket == 'day':
|
||
bucket_seconds = 86400
|
||
else:
|
||
bucket_seconds = 300
|
||
|
||
# ── 5. 按时间桶聚合 ──
|
||
bucket_data: dict[int, dict] = {} # bucket_start_ts → {count, bytes}
|
||
for ts_bj, idx in parsed_times:
|
||
bucket_key = int(ts_bj // bucket_seconds) * bucket_seconds
|
||
if bucket_key not in bucket_data:
|
||
bucket_data[bucket_key] = {'count': 0, 'bytes': 0.0}
|
||
bucket_data[bucket_key]['count'] += 1
|
||
if has_bytes and idx < len(raw_bytes):
|
||
try:
|
||
bval = raw_bytes[idx]
|
||
if bval is not None:
|
||
bucket_data[bucket_key]['bytes'] += float(bval)
|
||
except Exception:
|
||
pass
|
||
|
||
# ── 6. 构建输出 ──
|
||
def fmt_bj(ts: int, bucket_sec: int) -> str:
|
||
"""格式化北京时间字符串。"""
|
||
tm = time.gmtime(ts)
|
||
if bucket_sec >= 86400:
|
||
return time.strftime('%m-%d', tm)
|
||
elif bucket_sec >= 3600:
|
||
return time.strftime('%m-%d %H:00', tm)
|
||
elif bucket_sec >= 60:
|
||
return time.strftime('%m-%d %H:%M', tm)
|
||
else:
|
||
return time.strftime('%H:%M:%S', tm)
|
||
|
||
time_series = []
|
||
for ts_key in sorted(bucket_data.keys()):
|
||
d = bucket_data[ts_key]
|
||
time_series.append({
|
||
't': fmt_bj(ts_key, bucket_seconds),
|
||
'ts': ts_key,
|
||
'count': d['count'],
|
||
'bytes': round(d['bytes'], 2),
|
||
})
|
||
|
||
return JsonResponse({
|
||
'time_series': time_series,
|
||
'span_hours': round(span_hours, 2),
|
||
'bucket_minutes': bucket_seconds // 60,
|
||
'n_raw': len(parsed_times),
|
||
'has_bytes': has_bytes,
|
||
})
|
||
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def cluster_nodes_edges(request):
|
||
"""返回指定簇的所有节点、内部边、关联外部边及邻接节点。
|
||
|
||
POST body: {session_id, cluster_label}
|
||
返回: {nodes, internal_edges, external_edges, adjacent_nodes, geo_bounds, cluster_info}
|
||
"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
cluster_label = body.get('cluster_label')
|
||
|
||
if not session_id or cluster_label is None:
|
||
return JsonResponse({'error': '缺少 session_id / cluster_label'}, status=400)
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
nodes = entry.get('nodes', [])
|
||
edges_data = entry.get('edges_data', [])
|
||
cluster_info = entry.get('cluster_map', {}).get(str(cluster_label), {})
|
||
|
||
if not nodes:
|
||
return JsonResponse({'error': '无节点数据'}, status=400)
|
||
|
||
# ── 1. 筛选簇内节点 ──
|
||
cluster_nodes = [n for n in nodes if n.get('cluster_id') == cluster_label]
|
||
if not cluster_nodes:
|
||
return JsonResponse({'error': f'簇 {cluster_label} 无数据'}, status=404)
|
||
|
||
cluster_ips = set(n['ip'] for n in cluster_nodes)
|
||
|
||
# ── 2. 查找所有关联边（至少一端在簇内） ──
|
||
internal_edges = []
|
||
external_edges = []
|
||
adjacent_ip_set = set()
|
||
|
||
for e in edges_data:
|
||
src_in = e['source'] in cluster_ips
|
||
dst_in = e['target'] in cluster_ips
|
||
if src_in and dst_in:
|
||
internal_edges.append(e)
|
||
elif src_in:
|
||
external_edges.append(e)
|
||
adjacent_ip_set.add(e['target'])
|
||
elif dst_in:
|
||
external_edges.append(e)
|
||
adjacent_ip_set.add(e['source'])
|
||
|
||
# ── 3. 查找邻接节点 ──
|
||
adjacent_nodes = [n for n in nodes if n['ip'] in adjacent_ip_set]
|
||
|
||
# ── 4. 计算 Geo bounds ──
|
||
lats = [n.get('lat') for n in cluster_nodes if n.get('lat') is not None]
|
||
lons = [n.get('lon') for n in cluster_nodes if n.get('lon') is not None]
|
||
lats += [n.get('lat') for n in adjacent_nodes if n.get('lat') is not None]
|
||
lons += [n.get('lon') for n in adjacent_nodes if n.get('lon') is not None]
|
||
|
||
geo_bounds = None
|
||
if lats and lons:
|
||
geo_bounds = {
|
||
'lat_min': round(min(lats), 4), 'lat_max': round(max(lats), 4),
|
||
'lon_min': round(min(lons), 4), 'lon_max': round(max(lons), 4),
|
||
}
|
||
|
||
# ── 5. 簇级别统计 ──
|
||
def _dist(items, key):
|
||
counts = Counter(i.get(key, '') or '(未知)' for i in items)
|
||
total = sum(counts.values()) or 1
|
||
return [{'value': k, 'count': v, 'pct': round(v / total * 100, 1)}
|
||
for k, v in counts.most_common(10)]
|
||
|
||
total_internal_bytes = sum(e.get('total_bytes', 0) or 0 for e in internal_edges)
|
||
total_external_bytes = sum(e.get('total_bytes', 0) or 0 for e in external_edges)
|
||
|
||
return JsonResponse({
|
||
'cluster_label': cluster_label,
|
||
'cluster_info': cluster_info,
|
||
'nodes': cluster_nodes,
|
||
'n_nodes': len(cluster_nodes),
|
||
'internal_edges': internal_edges,
|
||
'external_edges': external_edges,
|
||
'n_internal_edges': len(internal_edges),
|
||
'n_external_edges': len(external_edges),
|
||
'adjacent_nodes': adjacent_nodes,
|
||
'n_adjacent_nodes': len(adjacent_nodes),
|
||
'geo_bounds': geo_bounds,
|
||
'top_isps': _dist(cluster_nodes, 'isp'),
|
||
'top_cities': _dist(cluster_nodes, 'city'),
|
||
'top_countries': _dist(cluster_nodes, 'country'),
|
||
'total_internal_bytes': round(total_internal_bytes, 2),
|
||
'total_external_bytes': round(total_external_bytes, 2),
|
||
})
|
||
|
||
|
||
@require_GET
|
||
def column_values(request):
|
||
"""返回指定列的唯一值 Top-50（供筛选器下拉建议）。
|
||
|
||
支持两种模式：
|
||
- session_id：从已上传的数据中查（全量，精确 Top-K）
|
||
- scan_id：从已保存的文件中 streaming 单列扫描（低内存，仅读一列）
|
||
"""
|
||
session_id = request.GET.get('session_id', '')
|
||
scan_id = request.GET.get('scan_id', '')
|
||
col_name = request.GET.get('col', '')
|
||
|
||
if not col_name:
|
||
return JsonResponse({'error': '缺少 col'}, status=400)
|
||
|
||
# ── Mode 1: session_id (full data in memory) ──────────────────────
|
||
if session_id:
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
df = entry.get('df')
|
||
if df is None or col_name not in df.columns:
|
||
return JsonResponse({'error': '列不存在'}, status=400)
|
||
try:
|
||
values = (
|
||
df[col_name]
|
||
.cast(pl.Utf8)
|
||
.fill_null('(空)')
|
||
.value_counts(sort=True)
|
||
.head(50)
|
||
)
|
||
result = [
|
||
{'value': str(row[0])[:80], 'count': int(row[1])}
|
||
for row in values.iter_rows()
|
||
]
|
||
return JsonResponse({'col': col_name, 'values': result, 'total': len(result)})
|
||
except Exception as e:
|
||
return JsonResponse({'error': str(e)}, status=400)
|
||
|
||
# ── Mode 2: scan_id (streaming single-column scan) ────────────────
|
||
if scan_id:
|
||
entry = _cached(scan_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '扫描会话已过期，请重新选择文件'}, status=400)
|
||
|
||
scan_paths = entry.get('scan_paths', [])
|
||
col_map = entry.get('col_map', {})
|
||
|
||
if not scan_paths:
|
||
return JsonResponse({'error': '扫描文件已清理，请重新选择文件'}, status=400)
|
||
|
||
# Resolve the original column name from col_map
|
||
# col_map is {original → std}, we need original for the file
|
||
original_col = col_name
|
||
for orig, std in col_map.items():
|
||
if std == col_name:
|
||
original_col = orig
|
||
break
|
||
|
||
try:
|
||
# Streaming scan — only reads ONE column, O(1 col) memory
|
||
lf = pl.scan_csv(scan_paths[0], infer_schema_length=0)
|
||
if len(scan_paths) > 1:
|
||
frames = [pl.scan_csv(p, infer_schema_length=0) for p in scan_paths]
|
||
lf = pl.concat(frames, how='diagonal_relaxed')
|
||
|
||
# Select only the target column (by original name)
|
||
if original_col in lf.columns:
|
||
col_expr = pl.col(original_col)
|
||
elif col_name in lf.columns:
|
||
col_expr = pl.col(col_name)
|
||
else:
|
||
return JsonResponse({'error': f'列 "{col_name}" 不在文件中'}, status=400)
|
||
|
||
# Streaming Top-50 value counts
|
||
values = (
|
||
lf.select(col_expr.cast(pl.Utf8).fill_null('(空)').alias('_val'))
|
||
.collect(streaming=True)
|
||
.to_series()
|
||
.value_counts(sort=True)
|
||
.head(50)
|
||
)
|
||
|
||
result = [
|
||
{'value': str(row[0])[:80], 'count': int(row[1])}
|
||
for row in values.iter_rows()
|
||
]
|
||
return JsonResponse({'col': col_name, 'values': result, 'total': len(result)})
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'列扫描失败: {e}'}, status=400)
|
||
|
||
return JsonResponse({'error': '缺少 session_id 或 scan_id'}, status=400)
|
||
|
||
|
||
@require_GET
|
||
def geoip_status(request):
|
||
"""返回当前 GeoIP 数据库状态。"""
|
||
from analysis import geoip
|
||
return JsonResponse({
|
||
'available': geoip.is_available(),
|
||
'source': geoip.source_name(),
|
||
'using_mmdb': 'mmdb' in geoip.source_name().lower(),
|
||
})
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
# Traffic timeline — dynamic flow animation
|
||
# ═══════════════════════════════════════════════════════════════════════════════
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def traffic_timeline_view(request):
|
||
"""Return time-windowed traffic data for global flow animation.
|
||
|
||
POST body: {session_id, window_seconds (10|30|60|300|600), max_edges (default 500)}
|
||
Returns per-window edge activity and node activity.
|
||
"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
window_sec = int(body.get('window_seconds', 60))
|
||
max_edges = int(body.get('max_edges', 500))
|
||
|
||
if not session_id:
|
||
return JsonResponse({'error': '缺少 session_id'}, status=400)
|
||
if window_sec not in (10, 30, 60, 300, 600):
|
||
window_sec = 60
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
|
||
if df is None:
|
||
return JsonResponse({'error': '数据未就绪'}, status=400)
|
||
meta = entry.get('meta', {})
|
||
ips_col = meta.get('ips_col')
|
||
ipd_col = meta.get('ipd_col')
|
||
time_col = meta.get('time_col')
|
||
bytes_col = meta.get('bytes_col')
|
||
|
||
if not ips_col or not ipd_col or not time_col:
|
||
return JsonResponse({'error': '缺少IP或时间列'}, status=400)
|
||
|
||
try:
|
||
# Extract columns
|
||
src_list = df[ips_col].cast(pl.Utf8).to_list()
|
||
dst_list = df[ipd_col].cast(pl.Utf8).to_list()
|
||
time_list = df[time_col].cast(pl.Utf8).to_list()
|
||
bytes_raw = None
|
||
if bytes_col and bytes_col in df.columns:
|
||
try:
|
||
bytes_raw = df[bytes_col].cast(pl.Float64).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
# Parse timestamps → Beijing time Unix seconds
|
||
import calendar as cal_mod
|
||
bj_timestamps = []
|
||
for t_str in time_list:
|
||
try:
|
||
t_clean = str(t_str).strip() if t_str else ''
|
||
if not t_clean or t_clean in ('None', '', 'nan'):
|
||
continue
|
||
# Try ISO format or 'YYYY-MM-DD HH:MM:SS' format
|
||
for fmt in ['%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S',
|
||
'%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S.%f']:
|
||
try:
|
||
parsed = time.strptime(t_clean[:26], fmt)
|
||
bj_ts = int(cal_mod.timegm(parsed)) + 8 * 3600
|
||
bj_timestamps.append(bj_ts)
|
||
break
|
||
except ValueError:
|
||
continue
|
||
except Exception:
|
||
continue
|
||
|
||
if not bj_timestamps:
|
||
return JsonResponse({'error': '无法解析时间戳'}, status=400)
|
||
|
||
min_ts = float(min(bj_timestamps))
|
||
max_ts = float(max(bj_timestamps))
|
||
|
||
# Build time windows
|
||
windows = []
|
||
t = min_ts
|
||
while t <= max_ts:
|
||
windows.append(t)
|
||
t += window_sec
|
||
|
||
# Aggregate per window
|
||
window_data = []
|
||
for wi, w_start in enumerate(windows):
|
||
w_end = w_start + window_sec
|
||
# Find rows in this window
|
||
edge_map = {} # (src,dst) → {count, bytes}
|
||
active_nodes = set()
|
||
row_count = 0
|
||
for idx, ts in enumerate(bj_timestamps):
|
||
if wi < len(windows) - 1 and ts >= w_end:
|
||
continue
|
||
if wi > 0 and ts < w_start:
|
||
continue
|
||
|
||
# In-window row
|
||
s = str(src_list[idx]) if idx < len(src_list) and src_list[idx] not in (None, '', 'None') else '?'
|
||
d = str(dst_list[idx]) if idx < len(dst_list) and dst_list[idx] not in (None, '', 'None') else '?'
|
||
key = (s, d)
|
||
if key not in edge_map:
|
||
edge_map[key] = {'count': 0, 'bytes': 0.0}
|
||
edge_map[key]['count'] += 1
|
||
if bytes_raw and idx < len(bytes_raw):
|
||
try:
|
||
edge_map[key]['bytes'] += float(bytes_raw[idx] or 0)
|
||
except Exception:
|
||
pass
|
||
active_nodes.add(s)
|
||
active_nodes.add(d)
|
||
row_count += 1
|
||
|
||
# Sort edges by activity
|
||
edges = []
|
||
for (src, dst), v in sorted(edge_map.items(), key=lambda x: -x[1]['count'])[:max_edges]:
|
||
edges.append({
|
||
'source': src,
|
||
'target': dst,
|
||
'count': v['count'],
|
||
'bytes': round(v['bytes'], 2),
|
||
})
|
||
|
||
# Node activity
|
||
node_activity = {}
|
||
nodes_data = entry.get('nodes', [])
|
||
for n in nodes_data:
|
||
ip = n.get('ip', '')
|
||
if ip in active_nodes:
|
||
node_activity[ip] = {'active': True}
|
||
|
||
if edges or active_nodes:
|
||
window_data.append({
|
||
'ts': w_start,
|
||
'ts_end': w_end,
|
||
't_label': time.strftime('%H:%M:%S', time.gmtime(w_start)),
|
||
'edges': edges,
|
||
'active_nodes': list(active_nodes)[:1000],
|
||
'row_count': row_count,
|
||
})
|
||
|
||
# Node coordinates for map
|
||
nodes_coords = {}
|
||
nodes_data = entry.get('nodes', [])
|
||
for n in nodes_data:
|
||
if n.get('lat') is not None and n.get('lon') is not None:
|
||
nodes_coords[n['ip']] = [n['lat'], n['lon']]
|
||
|
||
# ── Per-edge lifecycle: first_ts, last_ts (continuous rendering) ──
|
||
# Parse timestamps with microsecond precision using datetime
|
||
from datetime import datetime, timezone, timedelta
|
||
BJ_TZ = timezone(timedelta(hours=8))
|
||
bj_floats = []
|
||
for t_str in time_list:
|
||
try:
|
||
t_clean = str(t_str).strip() if t_str else ''
|
||
if not t_clean or t_clean in ('None', '', 'nan'):
|
||
bj_floats.append(None); continue
|
||
# Try common formats
|
||
dt = None
|
||
for fmt in ['%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S.%f',
|
||
'%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S']:
|
||
try:
|
||
dt = datetime.strptime(t_clean[:26], fmt)
|
||
break
|
||
except ValueError: continue
|
||
if dt is None: bj_floats.append(None); continue
|
||
# Convert to Beijing time Unix timestamp
|
||
bj_floats.append(dt.replace(tzinfo=BJ_TZ).timestamp())
|
||
except Exception:
|
||
bj_floats.append(None)
|
||
|
||
edge_lifecycle = {} # (src,dst) → {count,bytes,first_ts,last_ts}
|
||
for idx in range(len(src_list)):
|
||
ts = bj_floats[idx] if idx < len(bj_floats) else None
|
||
if ts is None: continue
|
||
s = str(src_list[idx]) if idx < len(src_list) and src_list[idx] not in (None, '', 'None') else '?'
|
||
d = str(dst_list[idx]) if idx < len(dst_list) and dst_list[idx] not in (None, '', 'None') else '?'
|
||
key = (s, d)
|
||
if key not in edge_lifecycle:
|
||
edge_lifecycle[key] = {'count': 0, 'bytes': 0.0, 'first_ts': ts, 'last_ts': ts}
|
||
e = edge_lifecycle[key]
|
||
e['count'] += 1
|
||
if bytes_raw and idx < len(bytes_raw):
|
||
try: e['bytes'] += float(bytes_raw[idx] or 0)
|
||
except Exception: pass
|
||
if ts < e['first_ts']: e['first_ts'] = ts
|
||
if ts > e['last_ts']: e['last_ts'] = ts
|
||
|
||
edges_flat = []
|
||
for (src, dst), e in sorted(edge_lifecycle.items(), key=lambda x: -x[1]['count'])[:max_edges]:
|
||
edges_flat.append({
|
||
'source': src, 'target': dst,
|
||
'count': e['count'], 'bytes': round(e['bytes'], 2),
|
||
'first_ts': e['first_ts'], 'last_ts': e['last_ts'],
|
||
})
|
||
|
||
return JsonResponse({
|
||
'edges': edges_flat,
|
||
'window_seconds': window_sec,
|
||
'min_ts': min_ts,
|
||
'max_ts': max_ts,
|
||
'nodes_coords': nodes_coords,
|
||
})
|
||
except Exception as e:
|
||
return JsonResponse({'error': f'时间线构建失败: {e}'}, status=400)
|
||
|
||
|
||
@csrf_exempt
|
||
@require_POST
|
||
def entity_traffic_timeline_view(request):
|
||
"""Return per-entity traffic timeline for local flow animation.
|
||
|
||
POST body: {session_id, entity_ip, window_seconds}
|
||
Returns timeline focused on one IP's related edges.
|
||
"""
|
||
body = json.loads(request.body)
|
||
session_id = body.get('session_id')
|
||
entity_ip = body.get('entity_ip', '')
|
||
window_sec = int(body.get('window_seconds', 60))
|
||
|
||
if not session_id or not entity_ip:
|
||
return JsonResponse({'error': '缺少 session_id / entity_ip'}, status=400)
|
||
if window_sec not in (10, 30, 60, 300, 600):
|
||
window_sec = 60
|
||
|
||
entry = _cached(session_id)
|
||
if entry is None:
|
||
return JsonResponse({'error': '会话已过期'}, status=400)
|
||
|
||
df = entry.get('filtered_df') if entry.get('filtered_df') is not None else entry.get('df')
|
||
if df is None:
|
||
return JsonResponse({'error': '数据未就绪'}, status=400)
|
||
meta = entry.get('meta', {})
|
||
ips_col = meta.get('ips_col')
|
||
ipd_col = meta.get('ipd_col')
|
||
time_col = meta.get('time_col')
|
||
bytes_col = meta.get('bytes_col')
|
||
|
||
if not ips_col or not ipd_col or not time_col:
|
||
return JsonResponse({'error': '缺少IP或时间列'}, status=400)
|
||
|
||
try:
|
||
# Filter rows where entity_ip is src OR dst
|
||
src_list = df[ips_col].cast(pl.Utf8).to_list()
|
||
dst_list = df[ipd_col].cast(pl.Utf8).to_list()
|
||
time_list = df[time_col].cast(pl.Utf8).to_list()
|
||
bytes_raw = None
|
||
if bytes_col and bytes_col in df.columns:
|
||
try:
|
||
bytes_raw = df[bytes_col].cast(pl.Float64).to_list()
|
||
except Exception:
|
||
pass
|
||
|
||
# Parse timestamps for filtered rows
|
||
import calendar as cal_mod
|
||
filtered_data = [] # [(ts_bj, src, dst, bytes_val)]
|
||
for idx in range(len(df)):
|
||
s = str(src_list[idx]) if src_list[idx] not in (None, '', 'None') else ''
|
||
d = str(dst_list[idx]) if dst_list[idx] not in (None, '', 'None') else ''
|
||
if entity_ip not in (s, d):
|
||
continue
|
||
t_str = str(time_list[idx]) if time_list[idx] else ''
|
||
if not t_str or t_str in ('None', '', 'nan'):
|
||
continue
|
||
bj_ts = None
|
||
for fmt in ['%Y-%m-%dT%H:%M:%S', '%Y-%m-%d %H:%M:%S',
|
||
'%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S.%f']:
|
||
try:
|
||
parsed = time.strptime(t_str[:26], fmt)
|
||
bj_ts = int(cal_mod.timegm(parsed)) + 8 * 3600
|
||
break
|
||
except ValueError:
|
||
continue
|
||
if bj_ts is None:
|
||
continue
|
||
bv = 0.0
|
||
if bytes_raw and idx < len(bytes_raw):
|
||
try:
|
||
bv = float(bytes_raw[idx] or 0)
|
||
except Exception:
|
||
pass
|
||
filtered_data.append((bj_ts, s, d, bv))
|
||
|
||
if not filtered_data:
|
||
return JsonResponse({'error': f'IP {entity_ip} 无时序数据'}, status=400)
|
||
|
||
min_ts = min(d[0] for d in filtered_data)
|
||
max_ts = max(d[0] for d in filtered_data)
|
||
|
||
# Build windows
|
||
windows = []
|
||
t = min_ts
|
||
while t <= max_ts:
|
||
windows.append(t)
|
||
t += window_sec
|
||
|
||
window_data = []
|
||
peer_set = set()
|
||
total_bytes = 0.0
|
||
total_count = 0
|
||
|
||
for wi, w_start in enumerate(windows):
|
||
w_end = w_start + window_sec
|
||
edge_map = {}
|
||
out_bytes = 0.0
|
||
in_bytes = 0.0
|
||
for (ts, s, d, bv) in filtered_data:
|
||
if wi < len(windows) - 1 and ts >= w_end:
|
||
continue
|
||
if wi > 0 and ts < w_start:
|
||
continue
|
||
key = (s, d)
|
||
if key not in edge_map:
|
||
edge_map[key] = {'count': 0, 'bytes': 0.0}
|
||
edge_map[key]['count'] += 1
|
||
edge_map[key]['bytes'] += bv
|
||
if s == entity_ip:
|
||
out_bytes += bv
|
||
if d == entity_ip:
|
||
in_bytes += bv
|
||
peer = d if s == entity_ip else s
|
||
peer_set.add(peer)
|
||
total_bytes += bv
|
||
total_count += 1
|
||
|
||
edges = []
|
||
for (src, dst), v in sorted(edge_map.items(), key=lambda x: -x[1]['count'])[:50]:
|
||
edges.append({
|
||
'source': src, 'target': dst,
|
||
'count': v['count'], 'bytes': round(v['bytes'], 2),
|
||
})
|
||
|
||
if edges:
|
||
window_data.append({
|
||
'ts': w_start,
|
||
't_label': time.strftime('%H:%M:%S', time.gmtime(w_start)),
|
||
'edges': edges,
|
||
'out_bytes': round(out_bytes, 2),
|
||
'in_bytes': round(in_bytes, 2),
|
||
})
|
||
|
||
# Aggregate stats
|
||
hour_counts = [0] * 24
|
||
for (ts, s, d, bv) in filtered_data:
|
||
tm = time.gmtime(ts)
|
||
hour_counts[tm.tm_hour] += 1
|
||
|
||
return JsonResponse({
|
||
'entity_ip': entity_ip,
|
||
'windows': window_data,
|
||
'window_seconds': window_sec,
|
||
'min_ts': min_ts,
|
||
'max_ts': max_ts,
|
||
'n_windows': len(window_data),
|
||
'n_peers': len(peer_set),
|
||
'total_bytes': round(total_bytes, 2),
|
||
'total_count': total_count,
|
||
'peers': sorted(peer_set)[:50],
|
||
'hourly_distribution': [{'hour': h, 'count': c} for h, c in enumerate(hour_counts)],
|
||
})
|
||
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')
|