cleanup: consolidate TLS_HEX_MAP, extract constants, remove dead code, add traceback logging

3.1 Unified TLS_HEX_MAP: defined once in type_classifier.py, imported in globe.py and clustering.py
3.2 Created analysis/constants.py: centralized RANDOM_SEED, MAX_SAMPLE_ROWS, UMAP constants, GLOBE_MAX_ROWS, etc.
3.3 Removed 14 pointless try/except:raise dead code blocks across tools/ and views/
3.4 Added logger.error(traceback) to all bare except Exception: blocks in tools/ and views/
3.5 Moved inline import traceback/asyncio to top-level imports in features.py, clustering.py, pipeline.py
3.6 Removed auto_profile_module side-effect imports from 6 modules
This commit is contained in:
PM-pinou
2026-07-23 22:44:42 +08:00
parent 885da89b6f
commit ef60e3e489
26 changed files with 606 additions and 493 deletions
+96
View File
@@ -0,0 +1,96 @@
"""Centralized constants for the TianXuan analysis module.
All magic numbers from analysis/tools/ and analysis/views/ are defined here
so they can be updated in one place. Import with::
from analysis.constants import RANDOM_SEED, MAX_SAMPLE_ROWS, ...
"""
import polars as pl
# ---------------------------------------------------------------------------
# Clustering / UMAP
# ---------------------------------------------------------------------------
MAX_SAMPLE_ROWS = 50000
"""Maximum rows for clustering sample (downsampling threshold)."""
MAX_DISTRIBUTION_SAMPLE = 10000
"""Default sample size for distribution exploration and scoring."""
UMAP_TRAIN_SAMPLE = 10000
"""Max rows used to train a UMAP reducer in extract_features."""
UMAP_BATCH_SIZE = 1000
"""Batch size for UMAP transform when dataset exceeds UMAP_TRAIN_SAMPLE."""
LOW_MEMORY_THRESHOLD_GB = 2
"""If available memory < this (in GiB), enable low-memory clustering path."""
LOW_MEMORY_THRESHOLD_ROWS = 100000
"""Row count above which low-memory downsampling is triggered."""
RANDOM_SEED = 42
"""Default random seed for all sklearn/numpy operations."""
# ---------------------------------------------------------------------------
# Globe
# ---------------------------------------------------------------------------
GLOBE_MAX_ROWS = 300
"""Max rows to load per run for 3D globe visualization."""
# ---------------------------------------------------------------------------
# Filter
# ---------------------------------------------------------------------------
FILTER_MAX_ROWS = 5
"""Maximum visible filter rows in the manual analysis UI (deprecated / reserved)."""
MAX_COLUMNS_PROFILE = 200
"""Max columns to include in a dataset profile summary."""
# ---------------------------------------------------------------------------
# File upload limits
# ---------------------------------------------------------------------------
FILE_UPLOAD_MAX_SIZE = 500 * 1024 * 1024 # 500 MB
"""Maximum size for a single uploaded file."""
TOTAL_UPLOAD_MAX_SIZE = 50 * 1024 * 1024 * 1024 # 50 GB
"""Maximum total upload size across all files."""
# ---------------------------------------------------------------------------
# Tool limits (numeric type tuples used across tools/)
# ---------------------------------------------------------------------------
NUMERIC_DTYPES: tuple[type[pl.DataType], ...] = (
pl.Int8, pl.Int16, pl.Int32, pl.Int64,
pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64,
pl.Float32, pl.Float64,
)
"""Polars numeric data types — used for column selection in clustering,
entity aggregation, anomaly detection, and feature extraction."""
# ---------------------------------------------------------------------------
# Diagnostics / Validation
# ---------------------------------------------------------------------------
MAX_VALIDATE_SAMPLE = 1000
"""Max rows sampled during validate_data."""
MAX_DIAGNOSE_SAMPLE = 5000
"""Max rows sampled for clustering diagnosis (SVD)."""
MAX_CLUSTER_FEATURES = 50
"""Feature count threshold above which TruncatedSVD is applied."""
CLUSTER_MAX_FEATURES_TOP = 10
"""Max feature columns selected by auto-detection in clustering tools."""
# ---------------------------------------------------------------------------
# Globe / Flow extraction
# ---------------------------------------------------------------------------
GLOBE_FLOW_MAX_ROWS = 300
"""Alias for GLOBE_MAX_ROWS — max rows loaded for flow extraction."""
-5
View File
@@ -771,8 +771,3 @@ def load_csv_directory(
total_row_count = min(total_row_count, head)
return merged_lf, merged_schema, total_row_count, len(paths), memory_mb
# Auto-profile all public functions in this module
from analysis.profile_util import auto_profile_module # noqa: E402
auto_profile_module(__name__)
-5
View File
@@ -331,8 +331,3 @@ def profile_dataset(
}
return truncate_profile(result, max_kb=max_kb)
# Auto-profile all public functions in this module
from analysis.profile_util import auto_profile_module # noqa: E402
auto_profile_module(__name__)
-5
View File
@@ -400,8 +400,3 @@ def describe_cluster(
def _signed(value: float) -> str:
"""Return a signed string representation, e.g. ``"+2.1"``."""
return f"{value:+.1f}"
# Auto-profile all public functions in this module
from analysis.profile_util import auto_profile_module # noqa: E402
auto_profile_module(__name__)
-5
View File
@@ -481,8 +481,3 @@ class SessionStore:
types[t] = types.get(t, 0) + 1
parts = [f"{k}={v}" for k, v in sorted(types.items())]
return f"<SessionStore {'; '.join(parts)}>"
# Auto-profile all public functions in this module
from analysis.profile_util import auto_profile_module # noqa: E402
auto_profile_module(__name__)
-4
View File
@@ -60,7 +60,3 @@ from .data_mgmt import (
_handle_clone_dataset,
)
from .distance_matrix import _handle_compute_distance_matrix
# Auto-profile all public functions in this module
from analysis.profile_util import auto_profile_module # noqa: E402
auto_profile_module(__name__)
+4
View File
@@ -9,11 +9,14 @@ Provides:
- _get_numeric_columns: extract numeric column names from schema
"""
import json
import logging
import traceback
from typing import Any
import polars as pl
from ..session_store import SessionStore
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════════════
@@ -116,6 +119,7 @@ def _count_rows(lf: pl.LazyFrame) -> int:
try:
return lf.select(pl.len()).collect(streaming=True).item()
except Exception:
logger.error("_count_rows failed: {}".format(traceback.format_exc()))
return 0
+23 -14
View File
@@ -1,9 +1,13 @@
"""Read-only analysis handlers — patterns, temporal, FFT, TLS health, geo, entity detail."""
import traceback
from analysis.constants import LOW_MEMORY_THRESHOLD_ROWS
import numpy as np
import polars as pl
from ..session_store import SessionStore
from ._helpers import _resolve_dataset, _count_rows, NUMERIC_TYPE_NAMES
import logging
logger = logging.getLogger(__name__)
async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
@@ -14,9 +18,9 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
lf = entry['lazyframe']
schema = entry.get('schema', {})
n = _count_rows(lf)
df = lf.head(100000).collect(streaming=True)
df = lf.head(LOW_MEMORY_THRESHOLD_ROWS).collect(streaming=True)
result = {'total_rows': n, 'sampled': min(n, 100000)}
result = {'total_rows': n, 'sampled': min(n, LOW_MEMORY_THRESHOLD_ROWS)}
for col_name, label in [(':ips', 'src_ips'), (':ipd', 'dst_ips'),
('snam', 'snis'), ('cnam', 'cert_names')]:
@@ -29,6 +33,7 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
'unique': int(df[col_name].n_unique()),
}
except Exception:
logger.error("_handle_analyze_patterns failed: {}".format(traceback.format_exc()))
pass
for port_col in [':prs', ':prd']:
@@ -41,6 +46,7 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
'unique': int(df[port_col].n_unique()),
}
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
for tls_col in ['0ver']:
@@ -50,6 +56,7 @@ async def _handle_analyze_patterns(dataset_id: str, top_n: int = 10) -> dict:
result['tls_versions'] = [{'version': str(r[tls_col]), 'count': int(r['count'])}
for r in vc.iter_rows(named=True)]
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
return result
@@ -116,6 +123,7 @@ async def _handle_analyze_fft(
ts_parsed = ts_series.str.strptime(pl.Datetime, format=None, strict=False)
ts_numeric = ts_parsed.cast(pl.Int64) / 1_000_000_000.0
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
try:
ts_numeric = ts_series.cast(pl.Float64)
except Exception as e:
@@ -202,9 +210,9 @@ async def _handle_analyze_tls_health(dataset_id: str) -> dict:
lf = entry['lazyframe']
schema = entry.get('schema', {})
n = _count_rows(lf)
df = lf.head(100000).collect(streaming=True)
df = lf.head(LOW_MEMORY_THRESHOLD_ROWS).collect(streaming=True)
result = {'total_rows': n, 'sampled': min(n, 100000)}
result = {'total_rows': n, 'sampled': min(n, LOW_MEMORY_THRESHOLD_ROWS)}
if '0ver' in df.columns:
vc = df['0ver'].value_counts(sort=True)
@@ -236,6 +244,7 @@ async def _handle_analyze_tls_health(dataset_id: str) -> dict:
try:
weak_count += df['cipher-suite'].str.to_lowercase().str.contains(w).sum()
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
result['weak_cipher_pct'] = round(float(weak_count) / max(total_ciphers, 1) * 100, 1) if total_ciphers else 0
@@ -254,9 +263,9 @@ async def _handle_analyze_geo_distribution(dataset_id: str) -> dict:
lf = entry['lazyframe']
schema = entry.get('schema', {})
n = _count_rows(lf)
df = lf.head(100000).collect(streaming=True)
df = lf.head(LOW_MEMORY_THRESHOLD_ROWS).collect(streaming=True)
result = {'total_rows': n, 'sampled': min(n, 100000)}
result = {'total_rows': n, 'sampled': min(n, LOW_MEMORY_THRESHOLD_ROWS)}
for geo_col, label in [('scnt', 'src_countries'), ('dcnt', 'dst_countries')]:
if geo_col in df.columns:
@@ -266,6 +275,7 @@ async def _handle_analyze_geo_distribution(dataset_id: str) -> dict:
for r in vc.iter_rows(named=True)]
result[f'{label}_unique'] = int(df[geo_col].n_unique())
except Exception:
logger.error("_handle_analyze_geo_distribution failed: {}".format(traceback.format_exc()))
pass
if 'scnt' in df.columns and 'dcnt' in df.columns:
@@ -274,6 +284,7 @@ async def _handle_analyze_geo_distribution(dataset_id: str) -> dict:
result['domestic_pct'] = round(float(same) / max(len(df), 1) * 100, 1)
result['international_pct'] = round(100.0 - result['domestic_pct'], 1)
except Exception:
logger.error("_handle_analyze_geo_distribution failed: {}".format(traceback.format_exc()))
pass
return result
@@ -296,14 +307,11 @@ async def _handle_analyze_entity_detail(dataset_id: str, entity_value: str) -> d
if not entity_col:
entity_col = list(schema.keys())[0]
try:
filtered = lf.filter(pl.col(entity_col).cast(pl.Utf8).str.contains(entity_value, literal=True))
n = _count_rows(filtered)
if n == 0:
return {'error': f"Entity '{entity_value}' not found in column '{entity_col}'", 'truncated': False}
df = filtered.collect(streaming=True)
except Exception as exc:
raise
filtered = lf.filter(pl.col(entity_col).cast(pl.Utf8).str.contains(entity_value, literal=True))
n = _count_rows(filtered)
if n == 0:
return {'error': f"Entity '{entity_value}' not found in column '{entity_col}'", 'truncated': False}
df = filtered.collect(streaming=True)
result = {
'entity_value': entity_value,
@@ -321,6 +329,7 @@ async def _handle_analyze_entity_detail(dataset_id: str, entity_value: str) -> d
'max': float(df[c].max()) if df[c].max() is not None else None,
}
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
return result
+9 -4
View File
@@ -1,14 +1,18 @@
"""Anomaly detection + visualization handlers."""
import numpy as np
from analysis.constants import MAX_DISTRIBUTION_SAMPLE, RANDOM_SEED
import polars as pl
from ..session_store import SessionStore
from ._helpers import _resolve_dataset, NUMERIC_TYPE_NAMES
import logging
logger = logging.getLogger(__name__)
async def _handle_detect_anomalies(dataset_id: str, contamination: float = 0.05) -> dict:
"""Run Isolation Forest on scored entity data."""
from sklearn.ensemble import IsolationForest
import traceback
entry, err = _resolve_dataset(dataset_id)
if err:
return err
@@ -24,12 +28,12 @@ async def _handle_detect_anomalies(dataset_id: str, contamination: float = 0.05)
mat = np.nan_to_num(mat, nan=0.0)
n = len(mat)
chunk_size = min(50000, n)
chunk_size = min(MAX_SAMPLE_ROWS, n)
labels = np.zeros(n, dtype=int)
for start in range(0, n, chunk_size):
end = min(start + chunk_size, n)
iso = IsolationForest(n_estimators=100, max_samples=min(256, chunk_size),
contamination=contamination, random_state=42, n_jobs=-1)
contamination=contamination, random_state=RANDOM_SEED, n_jobs=-1)
labels[start:end] = iso.fit_predict(mat[start:end])
anomaly_count = int((labels == -1).sum())
@@ -59,12 +63,13 @@ async def _handle_visualize_anomalies(dataset_id: str) -> dict:
try:
n_total = lf.select(pl.len()).collect(streaming=True).item()
except Exception:
logger.error("_handle_visualize_anomalies failed: {}".format(traceback.format_exc()))
n_total = 0
sample_lf = lf if n_total <= 10000 else lf.collect(streaming=True).sample(n=10000, seed=42).lazy()
sample_lf = lf if n_total <= MAX_DISTRIBUTION_SAMPLE else lf.collect(streaming=True).sample(n=MAX_DISTRIBUTION_SAMPLE, seed=RANDOM_SEED).lazy()
df = sample_lf.select(num_cols).collect(streaming=True)
mat = StandardScaler().fit_transform(df.to_numpy())
mat = np.nan_to_num(mat, nan=0.0)
reducer = umap.UMAP(n_components=2, random_state=42,
reducer = umap.UMAP(n_components=2, random_state=RANDOM_SEED,
n_neighbors=15, min_dist=0.1, metric='euclidean')
coords = reducer.fit_transform(mat)
+157 -161
View File
@@ -6,6 +6,7 @@ Contains:
- _cluster_core: shared clustering pipeline extracted from both handlers
"""
import logging
from analysis.constants import LOW_MEMORY_THRESHOLD_GB, LOW_MEMORY_THRESHOLD_ROWS, MAX_CLUSTER_FEATURES, MAX_SAMPLE_ROWS, RANDOM_SEED
from typing import Optional
import numpy as np
@@ -13,6 +14,7 @@ import polars as pl
from ..session_store import SessionStore, _generate_id
from ._helpers import (
logger = logging.getLogger(__name__)
NUMERIC_DTYPES,
_build_filter_expr,
_resolve_dataset,
@@ -27,7 +29,7 @@ def _cluster_core(
feature_cols: list[str],
algorithm: str,
params: dict,
random_state: int = 42,
random_state: int = RANDOM_SEED,
):
"""Shared clustering pipeline: scale, filter, reduce, fit, score.
@@ -48,20 +50,20 @@ def _cluster_core(
try:
import psutil
mem = psutil.virtual_memory()
low_memory = mem.available < 2 * 1024 ** 3 # 2 GB
low_memory = mem.available < LOW_MEMORY_THRESHOLD_GB * 1024 ** 3 # 2 GB
except ImportError:
pass
if low_memory and len(data) > 100000:
rng = np.random.default_rng(42)
idx = rng.choice(len(data), 50000, replace=False)
if low_memory and len(data) > LOW_MEMORY_THRESHOLD_ROWS:
rng = np.random.default_rng(RANDOM_SEED)
idx = rng.choice(len(data), MAX_SAMPLE_ROWS, replace=False)
data = data[idx]
# ── Downsampling to 50K max ────────────────────────────────────────
_ds_idx = None
if len(data) > 50000:
rng = np.random.default_rng(42)
_ds_idx = rng.choice(len(data), 50000, replace=False)
if len(data) > MAX_SAMPLE_ROWS:
rng = np.random.default_rng(RANDOM_SEED)
_ds_idx = rng.choice(len(data), MAX_SAMPLE_ROWS, replace=False)
data = data[_ds_idx]
# ── Variance filtering ─────────────────────────────────────────────
@@ -98,15 +100,15 @@ def _cluster_core(
# ── SVD dimensionality reduction (if >50 features) ─────────────────
n_features = data_scaled.shape[1]
if n_features > 50:
if n_features > MAX_CLUSTER_FEATURES:
from sklearn.decomposition import TruncatedSVD
svd = TruncatedSVD(n_components=50, random_state=random_state)
svd = TruncatedSVD(n_components=MAX_CLUSTER_FEATURES, random_state=random_state)
data_scaled = svd.fit_transform(data_scaled)
# ── Hyperparameter auto-tuning ─────────────────────────────────────
n = len(data_scaled)
if algorithm == 'hdbscan':
min_cluster = p.get('min_cluster_size', max(3, min(50, n // 20)))
min_cluster = p.get('min_cluster_size', max(3, min(MAX_CLUSTER_FEATURES, n // 20)))
p['min_cluster_size'] = min_cluster
p['min_samples'] = p.get('min_samples', min_cluster)
p['cluster_selection_epsilon'] = p.get('cluster_selection_epsilon', 0.0)
@@ -159,18 +161,22 @@ def _cluster_core(
score = silhouette_score(data_scaled, labels_array, random_state=random_state)
quality['silhouette_score'] = round(float(score), 4)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
quality['silhouette_score'] = None
try:
from sklearn.metrics import davies_bouldin_score
dbs = davies_bouldin_score(data_scaled, labels_array)
quality['davies_bouldin_score'] = round(float(dbs), 4)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
quality['davies_bouldin_score'] = None
try:
from sklearn.metrics import calinski_harabasz_score
import traceback
chs = calinski_harabasz_score(data_scaled, labels_array)
quality['calinski_harabasz_score'] = round(float(chs), 4)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
quality['calinski_harabasz_score'] = None
# Cluster sizes
@@ -191,7 +197,7 @@ async def _handle_run_clustering(
cluster_columns: list[str],
algorithm: str = 'agglomerative',
params: Optional[dict] = None,
random_state: int = 42,
random_state: int = RANDOM_SEED,
) -> dict:
"""Run HDBSCAN, KMeans, or Agglomerative clustering on selected columns."""
entry, err = _resolve_dataset(dataset_id)
@@ -201,91 +207,89 @@ async def _handle_run_clustering(
lf: pl.LazyFrame = entry['lazyframe']
p = params or {}
try:
# Collect cluster features - numeric only
available = lf.collect_schema()
available_names = available.names()
available_dtypes = available.dtypes()
# Collect cluster features - numeric only
available = lf.collect_schema()
available_names = available.names()
available_dtypes = available.dtypes()
feature_cols = [
c for c in cluster_columns
if c in available_names
and available_dtypes[available_names.index(c)] in NUMERIC_DTYPES
]
if not feature_cols:
# Fall back: auto-select all numeric columns (exclude internal ones)
feature_cols = [
c for c in cluster_columns
if c in available_names
and available_dtypes[available_names.index(c)] in NUMERIC_DTYPES
]
available_names[i] for i, dt in enumerate(available_dtypes)
if dt in NUMERIC_DTYPES and not available_names[i].startswith('_')
][:10]
if not feature_cols:
# Fallback: try coercing Utf8 columns to Float64
utf8_cols = [
available_names[i] for i, dt in enumerate(available_dtypes)
if dt == pl.Utf8 and not available_names[i].startswith('_')
][:10]
if utf8_cols:
for col in utf8_cols:
try:
casted = lf.select(
pl.col(col).cast(pl.Float64, strict=False).alias(col)
).collect(streaming=True)
null_ratio = casted[col].is_null().sum() / max(len(casted), 1)
if null_ratio < 0.9:
lf = lf.with_columns(
pl.col(col).cast(pl.Float64, strict=False)
)
feature_cols.append(col)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
if not feature_cols:
# Fall back: auto-select all numeric columns (exclude internal ones)
feature_cols = [
available_names[i] for i, dt in enumerate(available_dtypes)
if dt in NUMERIC_DTYPES and not available_names[i].startswith('_')
][:10]
if not feature_cols:
# Fallback: try coercing Utf8 columns to Float64
utf8_cols = [
available_names[i] for i, dt in enumerate(available_dtypes)
if dt == pl.Utf8 and not available_names[i].startswith('_')
][:10]
if utf8_cols:
for col in utf8_cols:
try:
casted = lf.select(
pl.col(col).cast(pl.Float64, strict=False).alias(col)
).collect(streaming=True)
null_ratio = casted[col].is_null().sum() / max(len(casted), 1)
if null_ratio < 0.9:
lf = lf.with_columns(
pl.col(col).cast(pl.Float64, strict=False)
)
feature_cols.append(col)
except Exception:
pass
if not feature_cols:
all_numeric = [f"{available_names[i]}" for i, dt in enumerate(available_dtypes)
if dt in NUMERIC_DTYPES]
return {
'error': f'No numeric cluster columns found in dataset. Available: {all_numeric}',
'truncated': False
}
# Downsample before collect
row_count = _count_rows(lf)
if row_count > 100000:
lf = lf.collect(streaming=True).sample(n=100000, seed=42).lazy()
df_features = lf.select(feature_cols).collect(streaming=True)
# Drop all-NaN columns
for col in df_features.columns:
if df_features[col].is_null().all():
df_features = df_features.drop(col)
feature_cols = [c for c in feature_cols if c != col]
if df_features.width == 0:
return {'error': 'No valid feature columns after dropping all-NaN columns',
'truncated': False}
data = df_features.to_numpy()
# Fill remaining NaN with column mean
if data.dtype.kind == 'f':
col_mean = np.nanmean(data, axis=0)
col_mean = np.nan_to_num(col_mean, nan=0.0)
data = np.where(np.isnan(data), col_mean, data)
# Zero-sample guard
if len(data) < 3:
all_numeric = [f"{available_names[i]}" for i, dt in enumerate(available_dtypes)
if dt in NUMERIC_DTYPES]
return {
'cluster_result_id': _generate_id('clust'),
'n_clusters': 0,
'n_noise': 0,
'quality_metrics': {'note': 'too few samples'},
'truncated': False,
'error': f'No numeric cluster columns found in dataset. Available: {all_numeric}',
'truncated': False
}
labels_array, n_found, noise_count, quality, _ds_idx = _cluster_core(
data, feature_cols, algorithm, p, random_state,
)
# Downsample before collect
row_count = _count_rows(lf)
if row_count > LOW_MEMORY_THRESHOLD_ROWS:
lf = lf.collect(streaming=True).sample(n=LOW_MEMORY_THRESHOLD_ROWS, seed=RANDOM_SEED).lazy()
df_features = lf.select(feature_cols).collect(streaming=True)
# Drop all-NaN columns
for col in df_features.columns:
if df_features[col].is_null().all():
df_features = df_features.drop(col)
feature_cols = [c for c in feature_cols if c != col]
if df_features.width == 0:
return {'error': 'No valid feature columns after dropping all-NaN columns',
'truncated': False}
data = df_features.to_numpy()
# Fill remaining NaN with column mean
if data.dtype.kind == 'f':
col_mean = np.nanmean(data, axis=0)
col_mean = np.nan_to_num(col_mean, nan=0.0)
data = np.where(np.isnan(data), col_mean, data)
# Zero-sample guard
if len(data) < 3:
return {
'cluster_result_id': _generate_id('clust'),
'n_clusters': 0,
'n_noise': 0,
'quality_metrics': {'note': 'too few samples'},
'truncated': False,
}
labels_array, n_found, noise_count, quality, _ds_idx = _cluster_core(
data, feature_cols, algorithm, p, random_state,
)
except Exception:
raise
result_id = _generate_id('clust')
store = SessionStore()
@@ -332,50 +336,45 @@ async def _handle_filter_and_cluster(
full_schema = entry.get('schema', {})
# ── Step 1: Apply filters ─────────────────────────────────────────────
try:
exprs = [_build_filter_expr(
f['column'], f.get('op', 'eq'), f.get('value'),
) for f in filters]
exprs = [_build_filter_expr(
f['column'], f.get('op', 'eq'), f.get('value'),
) for f in filters]
if exprs:
if logic == 'and':
combined = exprs[0]
for e in exprs[1:]:
combined = combined & e
else:
combined = exprs[0]
for e in exprs[1:]:
combined = combined | e
filtered_lf = lf.filter(combined)
if exprs:
if logic == 'and':
combined = exprs[0]
for e in exprs[1:]:
combined = combined & e
else:
filtered_lf = lf
try:
row_count = _count_rows(filtered_lf)
except Exception:
row_count = None
combined = exprs[0]
for e in exprs[1:]:
combined = combined | e
filtered_lf = lf.filter(combined)
else:
filtered_lf = lf
try:
row_count = _count_rows(filtered_lf)
except Exception:
raise
logger.error("unknown failed: {}".format(traceback.format_exc()))
row_count = None
# ── Step 2: Auto-detect numeric feature columns ───────────────────────
try:
available = filtered_lf.collect_schema()
available_names = available.names()
available_dtypes = available.dtypes()
feature_cols = [
available_names[i] for i, dt in enumerate(available_dtypes)
if dt in NUMERIC_DTYPES and not available_names[i].startswith('_')
][:10]
available = filtered_lf.collect_schema()
available_names = available.names()
available_dtypes = available.dtypes()
feature_cols = [
available_names[i] for i, dt in enumerate(available_dtypes)
if dt in NUMERIC_DTYPES and not available_names[i].startswith('_')
][:10]
if not feature_cols:
return {
'error': 'No numeric columns found for clustering after filtering',
'filtered_dataset_id': None,
'truncated': False,
}
except Exception:
raise
if not feature_cols:
return {
'error': 'No numeric columns found for clustering after filtering',
'filtered_dataset_id': None,
'truncated': False,
}
# ── Step 3: Store filtered dataset ────────────────────────────────────
filtered_dataset_id = _generate_id('flt')
@@ -395,46 +394,43 @@ async def _handle_filter_and_cluster(
# ── Step 4: Run clustering ────────────────────────────────────────────
p = params or {}
try:
total_count = _count_rows(filtered_lf)
clamp_lf = filtered_lf
if total_count > 100000:
clamp_lf = filtered_lf.collect(streaming=True).sample(n=100000, seed=42).lazy()
total_count = _count_rows(filtered_lf)
clamp_lf = filtered_lf
if total_count > LOW_MEMORY_THRESHOLD_ROWS:
clamp_lf = filtered_lf.collect(streaming=True).sample(n=LOW_MEMORY_THRESHOLD_ROWS, seed=RANDOM_SEED).lazy()
df_features = clamp_lf.select(feature_cols).collect(streaming=True)
df_features = clamp_lf.select(feature_cols).collect(streaming=True)
for col in df_features.columns:
if df_features[col].is_null().all():
df_features = df_features.drop(col)
feature_cols = [c for c in feature_cols if c != col]
for col in df_features.columns:
if df_features[col].is_null().all():
df_features = df_features.drop(col)
feature_cols = [c for c in feature_cols if c != col]
if df_features.width == 0:
return {'error': 'No valid feature columns after dropping all-NaN columns',
'filtered_dataset_id': filtered_dataset_id, 'truncated': False}
if df_features.width == 0:
return {'error': 'No valid feature columns after dropping all-NaN columns',
'filtered_dataset_id': filtered_dataset_id, 'truncated': False}
data = df_features.to_numpy()
data = df_features.to_numpy()
if data.dtype.kind == 'f':
col_mean = np.nanmean(data, axis=0)
col_mean = np.nan_to_num(col_mean, nan=0.0)
data = np.where(np.isnan(data), col_mean, data)
if data.dtype.kind == 'f':
col_mean = np.nanmean(data, axis=0)
col_mean = np.nan_to_num(col_mean, nan=0.0)
data = np.where(np.isnan(data), col_mean, data)
if len(data) < 3:
return {
'cluster_result_id': _generate_id('clust'),
'n_clusters': 0,
'n_noise': 0,
'quality_metrics': {'note': 'too few samples'},
'filtered_dataset_id': filtered_dataset_id,
'truncated': False,
}
if len(data) < 3:
return {
'cluster_result_id': _generate_id('clust'),
'n_clusters': 0,
'n_noise': 0,
'quality_metrics': {'note': 'too few samples'},
'filtered_dataset_id': filtered_dataset_id,
'truncated': False,
}
labels_array, n_found, noise_count, quality, _ds_idx = _cluster_core(
data, feature_cols, algorithm, p, 42,
)
labels_array, n_found, noise_count, quality, _ds_idx = _cluster_core(
data, feature_cols, algorithm, p, RANDOM_SEED,
)
except Exception:
raise
result_id = _generate_id('clust')
store.store_cluster_result(
@@ -444,7 +440,7 @@ async def _handle_filter_and_cluster(
params={
'algorithm': algorithm,
'params': p,
'random_state': 42,
'random_state': RANDOM_SEED,
'feature_columns': feature_cols,
'_ds_idx': _ds_idx.tolist() if _ds_idx is not None else None,
},
+40 -40
View File
@@ -1,8 +1,12 @@
"""Diagnostic tool handlers — validate, explore, diagnose, compare, repair."""
import traceback
from analysis.constants import LOW_MEMORY_THRESHOLD_ROWS, RANDOM_SEED
import polars as pl
from ..session_store import SessionStore
from ._helpers import _resolve_dataset, _count_rows, NUMERIC_TYPE_NAMES
import logging
logger = logging.getLogger(__name__)
async def _handle_validate_data(dataset_id: str, fix_issues: bool = False) -> dict:
@@ -13,20 +17,18 @@ async def _handle_validate_data(dataset_id: str, fix_issues: bool = False) -> di
lf = entry['lazyframe']
schema = entry.get('schema', {})
issues = []
try:
df_sample = lf.head(1000).collect(streaming=True)
for col in schema:
null_count = df_sample[col].is_null().sum()
if null_count > 500:
issues.append(f'{col}: {null_count/10:.0f}% null')
try:
unique = df_sample[col].n_unique()
if unique <= 1 and null_count == 0:
issues.append(f'{col}: constant value (no variation)')
except Exception:
pass
except Exception as exc:
raise
df_sample = lf.head(1000).collect(streaming=True)
for col in schema:
null_count = df_sample[col].is_null().sum()
if null_count > 500:
issues.append(f'{col}: {null_count/10:.0f}% null')
try:
unique = df_sample[col].n_unique()
if unique <= 1 and null_count == 0:
issues.append(f'{col}: constant value (no variation)')
except Exception:
logger.error("_handle_validate_data failed: {}".format(traceback.format_exc()))
pass
total_rows = _count_rows(lf)
return {
'status': 'ok',
@@ -55,7 +57,7 @@ async def _handle_explore_distributions(dataset_id: str, columns: list[str] = No
try:
n = _count_rows(lf)
if n > max_sample:
sample_lf = lf.select(columns).sample(n=max_sample, seed=42)
sample_lf = lf.select(columns).sample(n=max_sample, seed=RANDOM_SEED)
else:
sample_lf = lf.select(columns)
df = sample_lf.collect(streaming=True)
@@ -78,6 +80,7 @@ async def _handle_explore_distributions(dataset_id: str, columns: list[str] = No
stats[col]['mean'] = round(float(s.mean()), 4) if s.mean() is not None else None
stats[col]['std'] = round(float(s.std()), 4) if s.std() is not None else None
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
stats[col] = {'error': 'failed to compute'}
return {'columns': stats, 'total_rows': n, 'sampled': min(n, max_sample)}
@@ -92,12 +95,9 @@ async def _handle_find_outliers(dataset_id: str, columns: list[str] = None,
schema = entry.get('schema', {})
if not columns:
columns = [c for c in schema if schema[c].split('(')[0].strip() in NUMERIC_TYPE_NAMES]
try:
n = _count_rows(lf)
sample_lf = lf if n <= 100000 else lf.collect(streaming=True).sample(n=100000, seed=42).lazy()
df = sample_lf.select(columns).collect(streaming=True)
except Exception as exc:
raise
n = _count_rows(lf)
sample_lf = lf if n <= 100000 else lf.collect(streaming=True).sample(n=LOW_MEMORY_THRESHOLD_ROWS, seed=RANDOM_SEED).lazy()
df = sample_lf.select(columns).collect(streaming=True)
outlier_counts = {}
for col in columns:
@@ -111,6 +111,7 @@ async def _handle_find_outliers(dataset_id: str, columns: list[str] = None,
if n_out > 0:
outlier_counts[col] = int(n_out)
except Exception:
logger.error("_handle_find_outliers failed: {}".format(traceback.format_exc()))
pass
return {'outlier_counts': outlier_counts, 'total_sampled': len(df), 'threshold_iqr': threshold}
@@ -152,11 +153,12 @@ async def _handle_diagnose_clustering(dataset_id: str, cluster_result_id: str) -
df_sample = lf.select(num_cols).head(5000).collect(streaming=True)
mat = df_sample.to_numpy()
n_comp = min(mat.shape)
svd = TruncatedSVD(n_components=n_comp, random_state=42).fit(mat)
svd = TruncatedSVD(n_components=n_comp, random_state=RANDOM_SEED).fit(mat)
var_explained = svd.explained_variance_ratio_
if var_explained[0] > 0.95:
diagnosis.append(f'First SVD component explains {var_explained[0]:.0%} variance — data nearly 1D')
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
return {
@@ -187,6 +189,7 @@ async def _handle_compare_datasets(dataset_id_a: str, dataset_id_b: str) -> dict
try:
n = _count_rows(lf)
except Exception:
logger.error("_get_info failed: {}".format(traceback.format_exc()))
n = -1
return {'schema': set(schema.keys()), 'row_count': n, 'columns': len(schema)}
@@ -210,24 +213,21 @@ async def _handle_export_debug_sample(dataset_id: str, max_rows: int = 50) -> di
entry, err = _resolve_dataset(dataset_id)
if err:
return err
try:
df = entry['lazyframe'].head(max_rows).collect(streaming=True)
rows = df.to_dict(as_series=False)
result = []
for i in range(len(df)):
row = {}
for col in df.columns:
val = rows[col][i]
if isinstance(val, (float, int, str, bool)):
row[col] = val
elif val is None:
row[col] = None
else:
row[col] = str(val)
result.append(row)
return {'rows': result, 'count': len(result), 'columns': df.columns}
except Exception as exc:
raise
df = entry['lazyframe'].head(max_rows).collect(streaming=True)
rows = df.to_dict(as_series=False)
result = []
for i in range(len(df)):
row = {}
for col in df.columns:
val = rows[col][i]
if isinstance(val, (float, int, str, bool)):
row[col] = val
elif val is None:
row[col] = None
else:
row[col] = str(val)
result.append(row)
return {'rows': result, 'count': len(result), 'columns': df.columns}
async def _handle_repair_schema(dataset_id: str) -> dict:
+6 -1
View File
@@ -1,4 +1,6 @@
"""compute_distance_matrix handler — LLM-driven per-row evaluation."""
import traceback
from analysis.constants import RANDOM_SEED
from typing import Optional
import numpy as np
@@ -6,6 +8,8 @@ import polars as pl
from ..session_store import SessionStore
from ._helpers import _resolve_dataset, NUMERIC_DTYPES
import logging
logger = logging.getLogger(__name__)
def _indent_function_body(body: str, indent: str = ' ') -> str:
@@ -43,7 +47,7 @@ async def _handle_compute_distance_matrix(
n_total = len(df)
if n_total > max_sample:
df_sampled = df.sample(n=max_sample, seed=42)
df_sampled = df.sample(n=max_sample, seed=RANDOM_SEED)
n_evaluated = max_sample
else:
df_sampled = df
@@ -103,6 +107,7 @@ async def _handle_compute_distance_matrix(
scores.append(None)
errors += 1
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
scores.append(None)
errors += 1
+9 -8
View File
@@ -1,4 +1,6 @@
"""Entity profile building + adaptive scoring handlers."""
import traceback
from analysis.constants import MAX_DISTRIBUTION_SAMPLE, RANDOM_SEED
import logging
from typing import Optional
@@ -8,6 +10,7 @@ import polars as pl
from ..session_store import SessionStore, _generate_id
from ..profile_util import profile
from ._helpers import _resolve_dataset, _count_rows, NUMERIC_DTYPES
logger = logging.getLogger(__name__)
# ═══════════════════════════════════════════════════════════════════════
@@ -113,11 +116,8 @@ async def _handle_build_entity_profiles(
agg_exprs.append(pl.col(c).n_unique().alias(alias_name))
# ── Execute grouping ───────────────────────────────────────────────
try:
entity_lf = lf.group_by(group_cols).agg(agg_exprs)
n_entities = _count_rows(entity_lf)
except Exception:
raise
entity_lf = lf.group_by(group_cols).agg(agg_exprs)
n_entities = _count_rows(entity_lf)
# ── Store result ───────────────────────────────────────────────────
new_schema = {name: str(dtype) for name, dtype in
@@ -160,7 +160,7 @@ def _add_adaptive_scores(lf: pl.LazyFrame, feature_names: list[str]) -> pl.LazyF
try:
n_rows = _count_rows(lf)
sample_lf = lf if n_rows <= 10000 else lf.sample(n=10000, seed=42)
sample_lf = lf if n_rows <= MAX_DISTRIBUTION_SAMPLE else lf.sample(n=MAX_DISTRIBUTION_SAMPLE, seed=RANDOM_SEED)
df_sample = sample_lf.select(avail).collect(streaming=True)
mat = df_sample.to_numpy()
mat = np.nan_to_num(mat, nan=0.0)
@@ -169,7 +169,7 @@ def _add_adaptive_scores(lf: pl.LazyFrame, feature_names: list[str]) -> pl.LazyF
n_comp = min(1, min(mat.shape) - 1)
if n_comp < 1 or mat.shape[1] < 2:
return lf
svd = TruncatedSVD(n_components=1, random_state=42).fit(mat)
svd = TruncatedSVD(n_components=1, random_state=RANDOM_SEED).fit(mat)
raw_weights = np.abs(svd.components_[0])
weights = dict(zip(avail, raw_weights / raw_weights.sum()))
@@ -222,7 +222,7 @@ def _add_adaptive_scores(lf: pl.LazyFrame, feature_names: list[str]) -> pl.LazyF
'[SCORE] adaptive weights from SVD: %s | thresholds: p50=%.3f upper=%.3f extreme=%.3f | sample=%d/%d',
{c: f'{w:.3f}' for c, w in sorted(weights.items(), key=lambda x: -x[1])},
float(q50), upper_fence, extreme_fence,
min(n_rows, 10000), n_rows,
min(n_rows, MAX_DISTRIBUTION_SAMPLE), n_rows,
)
except Exception as exc:
@@ -248,6 +248,7 @@ async def _handle_compute_scores(dataset_id: str) -> dict:
new_schema = {name: str(dtype) for name, dtype in
zip(updated_schema.names(), updated_schema.dtypes())}
except Exception:
logger.error("_handle_compute_scores failed: {}".format(traceback.format_exc()))
new_schema = entry.get('schema', {})
store = SessionStore()
store.store_dataset(dataset_id, lf, schema=new_schema,
+37 -39
View File
@@ -1,5 +1,6 @@
"""evaluate_clustering handler — compute clustering quality metrics."""
from typing import Optional
from analysis.constants import RANDOM_SEED
import numpy as np
import polars as pl
@@ -30,49 +31,46 @@ async def _handle_evaluate_clustering(
scores: dict = {'n_clusters': cluster_entry['n_clusters']}
try:
lf: pl.LazyFrame = dataset_entry['lazyframe']
params = cluster_entry.get('params', {})
feature_cols = params.get('feature_columns', [])
if not feature_cols and 'feature_columns' in params:
pass
elif feature_cols:
df_features = lf.select(feature_cols).collect(streaming=True)
data = df_features.to_numpy()
lf: pl.LazyFrame = dataset_entry['lazyframe']
params = cluster_entry.get('params', {})
feature_cols = params.get('feature_columns', [])
if not feature_cols and 'feature_columns' in params:
pass
elif feature_cols:
df_features = lf.select(feature_cols).collect(streaming=True)
data = df_features.to_numpy()
# Align labels with data (handle missing rows)
valid_mask = ~np.isnan(data).any(axis=1) if data.dtype.kind == 'f' else np.ones(len(data), dtype=bool)
data_clean = data[valid_mask]
labels_clean = np.array(labels)[:len(data_clean)]
# Align labels with data (handle missing rows)
valid_mask = ~np.isnan(data).any(axis=1) if data.dtype.kind == 'f' else np.ones(len(data), dtype=bool)
data_clean = data[valid_mask]
labels_clean = np.array(labels)[:len(data_clean)]
unique_labels = set(labels_clean)
n_clusters_actual = len(unique_labels - {-1})
unique_labels = set(labels_clean)
n_clusters_actual = len(unique_labels - {-1})
for m in requested:
if m == 'silhouette' and n_clusters_actual >= 2:
from sklearn.metrics import silhouette_score
scores['silhouette'] = round(float(
silhouette_score(data_clean, labels_clean, random_state=42)), 4)
elif m == 'davies_bouldin' and n_clusters_actual >= 2:
from sklearn.metrics import davies_bouldin_score
scores['davies_bouldin'] = round(float(
davies_bouldin_score(data_clean, labels_clean)), 4)
elif m == 'calinski_harabasz' and n_clusters_actual >= 2:
from sklearn.metrics import calinski_harabasz_score
scores['calinski_harabasz'] = round(float(
calinski_harabasz_score(data_clean, labels_clean)), 4)
elif m == 'noise_ratio':
noise = int((labels_clean == -1).sum())
scores['noise_ratio'] = round(noise / len(labels_clean), 4)
elif m == 'cluster_sizes':
sizes = {}
for label in unique_labels:
sizes[int(label)] = int((labels_clean == label).sum())
scores['cluster_sizes'] = sizes
for m in requested:
if m == 'silhouette' and n_clusters_actual >= 2:
from sklearn.metrics import silhouette_score
scores['silhouette'] = round(float(
silhouette_score(data_clean, labels_clean, random_state=RANDOM_SEED)), 4)
elif m == 'davies_bouldin' and n_clusters_actual >= 2:
from sklearn.metrics import davies_bouldin_score
scores['davies_bouldin'] = round(float(
davies_bouldin_score(data_clean, labels_clean)), 4)
elif m == 'calinski_harabasz' and n_clusters_actual >= 2:
from sklearn.metrics import calinski_harabasz_score
scores['calinski_harabasz'] = round(float(
calinski_harabasz_score(data_clean, labels_clean)), 4)
elif m == 'noise_ratio':
noise = int((labels_clean == -1).sum())
scores['noise_ratio'] = round(noise / len(labels_clean), 4)
elif m == 'cluster_sizes':
sizes = {}
for label in unique_labels:
sizes[int(label)] = int((labels_clean == label).sum())
scores['cluster_sizes'] = sizes
scores['n_clusters'] = cluster_entry['n_clusters']
scores['n_clusters'] = cluster_entry['n_clusters']
except Exception:
raise
return {'scores': scores}
+17 -20
View File
@@ -23,30 +23,27 @@ async def _handle_export_results(
dataset_entry = store.get_dataset(result_id)
if dataset_entry is not None:
lf: pl.LazyFrame = dataset_entry['lazyframe']
try:
df = lf.collect(streaming=True)
if out.suffix == '':
out = out / f"export_{result_id}.{format}"
df = lf.collect(streaming=True)
if out.suffix == '':
out = out / f"export_{result_id}.{format}"
if out.exists() and not overwrite:
return {'error': f'Output exists and overwrite=False: {out}',
'truncated': False}
if out.exists() and not overwrite:
return {'error': f'Output exists and overwrite=False: {out}',
'truncated': False}
out.parent.mkdir(parents=True, exist_ok=True)
out.parent.mkdir(parents=True, exist_ok=True)
if format == 'csv':
df.write_csv(out)
elif format == 'parquet':
df.write_parquet(out)
elif format == 'json':
df.write_json(out)
else:
return {'error': f'Unsupported format: {format}',
'truncated': False}
if format == 'csv':
df.write_csv(out)
elif format == 'parquet':
df.write_parquet(out)
elif format == 'json':
df.write_json(out)
else:
return {'error': f'Unsupported format: {format}',
'truncated': False}
file_paths.append(str(out.resolve()))
except Exception:
raise
file_paths.append(str(out.resolve()))
return {'file_paths': file_paths, 'format': format,
'row_count': len(df)}
+167 -162
View File
@@ -1,9 +1,15 @@
"""extract_features handler — per-cluster feature ranking + ORM persistence."""
import sys
import traceback
import numpy as np
import polars as pl
from ..session_store import SessionStore, _generate_id
from ..profile_util import profile
from analysis.constants import RANDOM_SEED, UMAP_BATCH_SIZE, UMAP_TRAIN_SAMPLE
import logging
logger = logging.getLogger(__name__)
async def _handle_extract_features(
@@ -25,180 +31,177 @@ async def _handle_extract_features(
return {'error': f'Cluster result not found: {cluster_result_id}',
'truncated': False}
try:
lf: pl.LazyFrame = dataset_entry['lazyframe']
labels = cluster_entry['labels']
lf: pl.LazyFrame = dataset_entry['lazyframe']
labels = cluster_entry['labels']
# D5: If clustering downsampled the data, filter matching rows
_ds_idx = cluster_entry.get('params', {}).get('_ds_idx')
if _ds_idx is not None:
df = (lf.with_row_index('__ds_idx')
.filter(pl.col('__ds_idx').is_in(_ds_idx))
.drop('__ds_idx')
.collect(streaming=True))
else:
df = lf.collect(streaming=True)
# D5: If clustering downsampled the data, filter matching rows
_ds_idx = cluster_entry.get('params', {}).get('_ds_idx')
if _ds_idx is not None:
df = (lf.with_row_index('__ds_idx')
.filter(pl.col('__ds_idx').is_in(_ds_idx))
.drop('__ds_idx')
.collect(streaming=True))
else:
df = lf.collect(streaming=True)
# Truncate labels if df was filtered but labels weren't
if len(labels) != len(df):
labels = labels[:len(df)]
# Truncate labels if df was filtered but labels weren't
if len(labels) != len(df):
labels = labels[:len(df)]
# Only use numeric columns for feature extraction
numeric_cols = [c for c in df.columns
if df[c].dtype in (pl.Float32, pl.Float64,
pl.Int32, pl.Int64,
pl.UInt32, pl.UInt64)]
if len(labels) != len(df):
labels = labels[:len(df)]
# Only use numeric columns for feature extraction
numeric_cols = [c for c in df.columns
if df[c].dtype in (pl.Float32, pl.Float64,
pl.Int32, pl.Int64,
pl.UInt32, pl.UInt64)]
if len(labels) != len(df):
labels = labels[:len(df)]
# Global statistics per column
global_stats = {}
for col in numeric_cols:
s = df[col].drop_nulls()
if len(s) > 0:
global_stats[col] = {
'mean': float(s.mean()),
'std': float(s.std()) if s.std() is not None else 1.0,
}
unique_labels = sorted(set(labels))
all_features = []
for label in unique_labels:
mask = np.array(labels) == label
if mask.sum() < 2:
continue
cluster_df = df.filter(pl.Series('__mask__', mask))
cluster_features = []
# Global statistics per column
global_stats = {}
for col in numeric_cols:
s = df[col].drop_nulls()
if len(s) > 0:
global_stats[col] = {
'mean': float(s.mean()),
'std': float(s.std()) if s.std() is not None else 1.0,
}
unique_labels = sorted(set(labels))
all_features = []
for label in unique_labels:
mask = np.array(labels) == label
if mask.sum() < 2:
c_series = cluster_df[col].drop_nulls()
gs = global_stats.get(col)
if gs is None or gs['std'] == 0 or len(c_series) == 0:
continue
cluster_df = df.filter(pl.Series('__mask__', mask))
cluster_features = []
c_mean = float(c_series.mean())
c_std = float(c_series.std()) if c_series.std() is not None else 0.0
for col in numeric_cols:
c_series = cluster_df[col].drop_nulls()
gs = global_stats.get(col)
if gs is None or gs['std'] == 0 or len(c_series) == 0:
continue
if method == 'zscore':
score = (c_mean - gs['mean']) / max(gs['std'], 1e-10)
else:
score = abs(c_mean - gs['mean']) / max(c_std + gs['std'], 1e-10)
c_mean = float(c_series.mean())
c_std = float(c_series.std()) if c_series.std() is not None else 0.0
cluster_features.append({
'feature_name': col,
'cluster_label': int(label),
'mean': round(c_mean, 4),
'std': round(c_std, 4),
'global_mean': round(gs['mean'], 4),
'global_std': round(gs['std'], 4),
'distinguishing_score': round(float(score), 4),
'distinguishing_method': method,
})
if method == 'zscore':
score = (c_mean - gs['mean']) / max(gs['std'], 1e-10)
cluster_features.sort(key=lambda x: abs(x['distinguishing_score']), reverse=True)
all_features.extend(cluster_features[:top_k])
# ── Compute UMAP-2D embedding ──────────────────────────────────
if len(numeric_cols) >= 2 and len(df) > 2:
try:
from sklearn.preprocessing import StandardScaler
import umap
MAX_UMAP_TRAIN = UMAP_TRAIN_SAMPLE
BATCH_SIZE = UMAP_BATCH_SIZE
data_matrix = df.select(numeric_cols).to_numpy()
if len(df) > MAX_UMAP_TRAIN:
idx_sample = np.random.RandomState(RANDOM_SEED).choice(
len(df), size=MAX_UMAP_TRAIN, replace=False)
mat_sample = data_matrix[idx_sample]
scaler = StandardScaler().fit(mat_sample)
mat_sample_scaled = np.nan_to_num(scaler.transform(mat_sample),
nan=0.0, posinf=0.0, neginf=0.0)
reducer = umap.UMAP(n_components=2, random_state=RANDOM_SEED,
n_neighbors=15, min_dist=0.1,
metric='euclidean')
reducer.fit(mat_sample_scaled)
coords_list = []
for start in range(0, len(df), BATCH_SIZE):
end = min(start + BATCH_SIZE, len(df))
mat_batch = scaler.transform(data_matrix[start:end])
mat_batch = np.nan_to_num(mat_batch, nan=0.0, posinf=0.0, neginf=0.0)
coords_list.append(reducer.transform(mat_batch))
coords = np.vstack(coords_list)
else:
data_scaled = StandardScaler().fit_transform(data_matrix)
data_scaled = np.nan_to_num(data_scaled, nan=0.0, posinf=0.0, neginf=0.0)
reducer = umap.UMAP(n_components=2, random_state=RANDOM_SEED,
n_neighbors=15, min_dist=0.1,
metric='euclidean')
coords = reducer.fit_transform(data_scaled)
non_numeric = [c for c in df.columns if c not in numeric_cols and not c.startswith('_')]
entity_col_umap = non_numeric[0] if non_numeric else df.columns[0]
labels_list = cluster_entry.get('labels', [])
from analysis.models import EntityProfile as EP
from analysis.models import ClusterResult as CR
from analysis.models import AnalysisRun as AR
from asgiref.sync import sync_to_async
csv_glob = dataset_entry.get('metadata', {}).get('csv_glob', 'manual')
run_obj = await sync_to_async(
AR.objects.filter(csv_glob=csv_glob).order_by('-id').first,
thread_sensitive=True
)()
if run_obj:
run_id_for_ep = run_obj.id
profiles = []
cr_cache = {}
for i, row in enumerate(df.iter_rows(named=True)):
if i >= len(coords): break
ev = str(row.get(entity_col_umap, ''))
if not ev: continue
lbl = labels_list[i] if i < len(labels_list) else -1
cache_key = f'{run_id_for_ep}_{lbl}'
if cache_key not in cr_cache:
cr_cache[cache_key] = CR.objects.filter(run_id=run_id_for_ep, cluster_label=lbl).first()
profiles.append(EP(
entity_value=ev, run_id=run_id_for_ep, cluster_label=lbl,
cluster=cr_cache[cache_key],
embedding_x=float(coords[i, 0]),
embedding_y=float(coords[i, 1]),
))
if profiles:
EP.objects.bulk_create(profiles, ignore_conflicts=True, batch_size=1000)
except Exception:
print(f'[UMAP_ERR] {traceback.format_exc()}', file=sys.stderr)
# Persist to Django ORM
db_saved = False
if save_to_db:
try:
from asgiref.sync import sync_to_async
from analysis.models import AnalysisRun as AnalysisRunModel
ce = cluster_entry
async def _do_save(ce_inner):
if run_id is not None:
run = await sync_to_async(AnalysisRunModel.objects.get)(id=run_id)
else:
score = abs(c_mean - gs['mean']) / max(c_std + gs['std'], 1e-10)
csv_glob_fallback = dataset_entry.get('metadata', {}).get('csv_glob', 'manual')
run = await sync_to_async(
AnalysisRunModel.objects.filter(csv_glob=csv_glob_fallback).order_by('-id').first,
thread_sensitive=True
)()
if run is None:
run = await sync_to_async(AnalysisRunModel.objects.create)(
csv_glob=csv_glob_fallback, status='running', total_flows=0,
)
run.status = 'completed'
run.cluster_count = ce_inner['n_clusters']
await sync_to_async(run.save)()
return await sync_to_async(_save_features_to_db)(cluster_result_id, all_features, run_id=run.id)
db_saved = await _do_save(ce)
except Exception as e:
print(f'[DB_SAVE_ERROR] {e}\n{traceback.format_exc()}', file=sys.stderr)
db_saved = False
cluster_features.append({
'feature_name': col,
'cluster_label': int(label),
'mean': round(c_mean, 4),
'std': round(c_std, 4),
'global_mean': round(gs['mean'], 4),
'global_std': round(gs['std'], 4),
'distinguishing_score': round(float(score), 4),
'distinguishing_method': method,
})
cluster_features.sort(key=lambda x: abs(x['distinguishing_score']), reverse=True)
all_features.extend(cluster_features[:top_k])
# ── Compute UMAP-2D embedding ──────────────────────────────────
if len(numeric_cols) >= 2 and len(df) > 2:
try:
from sklearn.preprocessing import StandardScaler
import umap
MAX_UMAP_TRAIN = 10000
BATCH_SIZE = 1000
data_matrix = df.select(numeric_cols).to_numpy()
if len(df) > MAX_UMAP_TRAIN:
idx_sample = np.random.RandomState(42).choice(
len(df), size=MAX_UMAP_TRAIN, replace=False)
mat_sample = data_matrix[idx_sample]
scaler = StandardScaler().fit(mat_sample)
mat_sample_scaled = np.nan_to_num(scaler.transform(mat_sample),
nan=0.0, posinf=0.0, neginf=0.0)
reducer = umap.UMAP(n_components=2, random_state=42,
n_neighbors=15, min_dist=0.1,
metric='euclidean')
reducer.fit(mat_sample_scaled)
coords_list = []
for start in range(0, len(df), BATCH_SIZE):
end = min(start + BATCH_SIZE, len(df))
mat_batch = scaler.transform(data_matrix[start:end])
mat_batch = np.nan_to_num(mat_batch, nan=0.0, posinf=0.0, neginf=0.0)
coords_list.append(reducer.transform(mat_batch))
coords = np.vstack(coords_list)
else:
data_scaled = StandardScaler().fit_transform(data_matrix)
data_scaled = np.nan_to_num(data_scaled, nan=0.0, posinf=0.0, neginf=0.0)
reducer = umap.UMAP(n_components=2, random_state=42,
n_neighbors=15, min_dist=0.1,
metric='euclidean')
coords = reducer.fit_transform(data_scaled)
non_numeric = [c for c in df.columns if c not in numeric_cols and not c.startswith('_')]
entity_col_umap = non_numeric[0] if non_numeric else df.columns[0]
labels_list = cluster_entry.get('labels', [])
from analysis.models import EntityProfile as EP
from analysis.models import ClusterResult as CR
from analysis.models import AnalysisRun as AR
from asgiref.sync import sync_to_async
csv_glob = dataset_entry.get('metadata', {}).get('csv_glob', 'manual')
run_obj = await sync_to_async(
AR.objects.filter(csv_glob=csv_glob).order_by('-id').first,
thread_sensitive=True
)()
if run_obj:
run_id_for_ep = run_obj.id
profiles = []
cr_cache = {}
for i, row in enumerate(df.iter_rows(named=True)):
if i >= len(coords): break
ev = str(row.get(entity_col_umap, ''))
if not ev: continue
lbl = labels_list[i] if i < len(labels_list) else -1
cache_key = f'{run_id_for_ep}_{lbl}'
if cache_key not in cr_cache:
cr_cache[cache_key] = CR.objects.filter(run_id=run_id_for_ep, cluster_label=lbl).first()
profiles.append(EP(
entity_value=ev, run_id=run_id_for_ep, cluster_label=lbl,
cluster=cr_cache[cache_key],
embedding_x=float(coords[i, 0]),
embedding_y=float(coords[i, 1]),
))
if profiles:
EP.objects.bulk_create(profiles, ignore_conflicts=True, batch_size=1000)
except Exception:
import traceback, sys; print(f'[UMAP_ERR] {traceback.format_exc()}', file=sys.stderr)
# Persist to Django ORM
db_saved = False
if save_to_db:
try:
from asgiref.sync import sync_to_async
from analysis.models import AnalysisRun as AnalysisRunModel
ce = cluster_entry
async def _do_save(ce_inner):
if run_id is not None:
run = await sync_to_async(AnalysisRunModel.objects.get)(id=run_id)
else:
csv_glob_fallback = dataset_entry.get('metadata', {}).get('csv_glob', 'manual')
run = await sync_to_async(
AnalysisRunModel.objects.filter(csv_glob=csv_glob_fallback).order_by('-id').first,
thread_sensitive=True
)()
if run is None:
run = await sync_to_async(AnalysisRunModel.objects.create)(
csv_glob=csv_glob_fallback, status='running', total_flows=0,
)
run.status = 'completed'
run.cluster_count = ce_inner['n_clusters']
await sync_to_async(run.save)()
return await sync_to_async(_save_features_to_db)(cluster_result_id, all_features, run_id=run.id)
db_saved = await _do_save(ce)
except Exception as e:
import traceback, sys; print(f'[DB_SAVE_ERROR] {e}\n{traceback.format_exc()}', file=sys.stderr)
db_saved = False
except Exception:
raise
result_id = _generate_id('feat')
store.store_feature_result(
@@ -251,6 +254,7 @@ def _save_features_to_db(
cluster_count=n_clusters_global,
)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
return False
from analysis.models import ClusterResult as ClusterResultModel
@@ -302,6 +306,7 @@ def _save_features_to_db(
}
)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
continue
run.cluster_count = n_clusters
+4
View File
@@ -1,8 +1,11 @@
"""filter_data handler — apply column-level filters to a dataset."""
import traceback
import polars as pl
from ..session_store import SessionStore, _generate_id
from ._helpers import _build_filter_expr, _resolve_dataset
import logging
logger = logging.getLogger(__name__)
async def _handle_filter_data(
@@ -40,6 +43,7 @@ async def _handle_filter_data(
try:
row_count = filtered_lf.select(pl.len()).collect(streaming=True).item()
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
row_count = None
new_id = _generate_id('flt')
+2 -1
View File
@@ -7,6 +7,7 @@ import polars as pl
from ..data_loader import load_csv_directory
from ..session_store import SessionStore, _generate_id
from analysis.constants import MAX_DISTRIBUTION_SAMPLE
async def _handle_load_data(
@@ -24,7 +25,7 @@ async def _handle_load_data(
encoding=encoding,
delimiter=delimiter,
config_path=config_path,
sample_rows=10000,
sample_rows=MAX_DISTRIBUTION_SAMPLE,
schema_strict=schema_strict,
recursive=recursive,
)
-4
View File
@@ -11,7 +11,3 @@ from .globe import globe_view, _extract_flows_from_df
from .config import config_view, llm_test
from .log_viewer import log_viewer
from .tools import tool_lab, tool_lab_run, tool_plan, apply_filter, reload_run_data, retry_run
# Auto-profile all sub-modules
from analysis.profile_util import auto_profile_module
auto_profile_module(__name__)
+2
View File
@@ -130,11 +130,13 @@ def run_llm_analysis_view(request):
try:
lf = load_from_db(run.sqlite_table)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
if lf is None and run.csv_glob:
try:
lf, schema, _, _, _ = load_csv_directory(run.csv_glob)
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
if lf is None:
return JsonResponse({'error': '数据集不存在,请先上传并等待预处理完成'})
+12 -8
View File
@@ -1,4 +1,5 @@
"""Cluster views: overview, detail, globe flows, and the clustering pipeline."""
import asyncio
import json
import traceback
import logging
@@ -8,6 +9,7 @@ from django.http import Http404
from analysis.models import AnalysisRun, ClusterResult, EntityProfile
from analysis import nl_describe
from analysis.constants import RANDOM_SEED, UMAP_BATCH_SIZE, UMAP_TRAIN_SAMPLE
from analysis.profile_util import profile
from .helpers import _extract_lat, _extract_lon
@@ -127,9 +129,9 @@ def _get_globe_flows(run, max_rows=500):
if lf is None:
return []
df = lf.head(max_rows).collect()
from analysis.type_classifier import TLS_HEX_MAP
from analysis.geoip import lookup as geo_lookup
flows = []
TLS_HEX_MAP = {'0303': 'TLSv1.2', '0304': 'TLSv1.3', '0302': 'TLSv1.1', '0301': 'TLSv1.0'}
src_ip_col = next((c for c in df.columns if 'src_ip' in c.lower() or ':ips' in c.lower()), None)
dst_ip_col = next((c for c in df.columns if 'dst_ip' in c.lower() or ':ipd' in c.lower()), None)
tls_col = next((c for c in df.columns if c.lower() in ('0ver', 'tls_version', '_tlsver', 'version')), None)
@@ -153,6 +155,7 @@ def _get_globe_flows(run, max_rows=500):
})
return flows
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
return []
@@ -209,7 +212,6 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
"""
import warnings
warnings.filterwarnings('ignore', category=RuntimeWarning, module='sklearn')
import asyncio
try:
from analysis.tool_registry import _handle_run_clustering, _handle_extract_features
@@ -230,6 +232,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
try:
row_count = lf.select(pl.len()).collect(streaming=True).item()
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
row_count = 0
if row_count > head:
@@ -280,7 +283,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
cluster_columns=feature_cols,
algorithm=algorithm,
params={'min_cluster_size': min_cluster_size},
random_state=42,
random_state=RANDOM_SEED,
))
if 'error' in result:
@@ -357,11 +360,12 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
try:
lf = entry['lazyframe']
# UMAP: sample max 10K for training, batch-transform rest in 1K batches
MAX_UMAP_TRAIN = 10000
BATCH_SIZE = 1000
MAX_UMAP_TRAIN = UMAP_TRAIN_SAMPLE
BATCH_SIZE = UMAP_BATCH_SIZE
try:
n_total = lf.select(pl.len()).collect(streaming=True).item()
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
n_total = 0
from sklearn.preprocessing import StandardScaler
import umap
@@ -377,7 +381,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
if n_total > MAX_UMAP_TRAIN:
run.progress_msg = f'UMAP 采样 {MAX_UMAP_TRAIN}/{n_total} 实体训练...'
run.save(update_fields=['progress_msg'])
df_sample = lf.sample(n=MAX_UMAP_TRAIN, seed=42).collect(streaming=True)
df_sample = lf.sample(n=MAX_UMAP_TRAIN, seed=RANDOM_SEED).collect(streaming=True)
df_umap = lf.collect(streaming=True)
else:
df_umap = lf.collect(streaming=True)
@@ -396,7 +400,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
mat_sample_scaled = np.nan_to_num(
scaler.transform(mat_sample), nan=0.0)
reducer = umap.UMAP(n_components=attempt_n,
random_state=42,
random_state=RANDOM_SEED,
n_neighbors=15, min_dist=0.1,
metric='euclidean')
reducer.fit(mat_sample_scaled)
@@ -413,7 +417,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith
mat = np.nan_to_num(StandardScaler().fit_transform(
df_umap.select(num_cols).to_numpy()), nan=0.0)
reducer = umap.UMAP(n_components=attempt_n,
random_state=42,
random_state=RANDOM_SEED,
n_neighbors=15, min_dist=0.1,
metric='euclidean')
coords = reducer.fit_transform(mat)
+2
View File
@@ -1,4 +1,5 @@
"""Configuration views: config editor and LLM connectivity test."""
import traceback
import json
import time
import urllib.request
@@ -112,6 +113,7 @@ def llm_test(request):
detail = json.loads(e.read().decode('utf-8', errors='replace'))
msg = detail.get('error', {}).get('message', str(e))
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
msg = str(e)
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
return JsonResponse({'success': False, 'message': f'HTTP {e.code}: {msg}', 'latency_ms': elapsed})
+5 -3
View File
@@ -1,9 +1,11 @@
"""3D Globe view and flow extraction."""
import json
import logging
import traceback
from django.shortcuts import render
from analysis.constants import GLOBE_MAX_ROWS
from analysis.models import AnalysisRun
from .helpers import _extract_lat, _extract_lon
@@ -13,8 +15,7 @@ logger = logging.getLogger(__name__)
def _extract_flows_from_df(df, MAX_ROWS):
"""Extract flow data (lat/lon/TLS/bytes) from a Polars DataFrame."""
from analysis.geoip import lookup as geo_lookup
TLS_HEX_MAP = {'0303': 'TLSv1.2', '0304': 'TLSv1.3', '0302': 'TLSv1.1', '0301': 'TLSv1.0'}
# NOTE: Duplicated in type_classifier.py — keep both in sync.
from analysis.type_classifier import TLS_HEX_MAP
src_lat_col = next((c for c in df.columns if c.lower() in (
':ips.latd', ':ips_latd', ':ips_lat', 'src_latitude', 'src_lat', 'source_latitude', 'source_lat')), None)
@@ -117,7 +118,7 @@ def globe_view(request):
from analysis.models import AnalysisRun
from analysis.session_store import SessionStore
MAX_ROWS_PER_RUN = 300 # Limit per run to prevent browser overload
MAX_ROWS_PER_RUN = GLOBE_MAX_ROWS # Limit per run to prevent browser overload
# Check for ?data=dataset_id parameter (direct data loading from session store)
data_param = request.GET.get('data')
@@ -197,6 +198,7 @@ def globe_view(request):
selected_runs = [latest]
break
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
continue
resp = render(request, 'tianxuan/globe.html', {
+4
View File
@@ -1,7 +1,10 @@
"""Geo extraction helpers and plan utilities."""
import traceback
from typing import Optional
from pathlib import Path
import json as _json
import logging
logger = logging.getLogger(__name__)
def _extract_lat(val):
@@ -45,6 +48,7 @@ def _read_auto_index():
return []
return data
except Exception:
logger.error("_read_auto_index failed: {}".format(traceback.format_exc()))
return []
+6 -4
View File
@@ -5,6 +5,7 @@ import traceback
import logging
from analysis.appdata import ensure_appdata_dir
from analysis.constants import MAX_CLUSTER_FEATURES, RANDOM_SEED
from analysis.models import AnalysisRun
from analysis.profile_util import profile
@@ -31,8 +32,7 @@ def _run_pipeline_worker(run_id, analysis_fn, **ctx):
try:
analysis_fn(run, ctx)
except Exception:
import traceback as _tb
tb = _tb.format_exc()
tb = traceback.format_exc()
logger.error(tb)
run.status = 'failed'
run.error_message = tb
@@ -208,6 +208,7 @@ def _background_process(run_id, upload_dir):
break
type_map[col] = pl.Int64 if all_int else pl.Float64
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass
if type_map:
@@ -229,7 +230,7 @@ def _background_process(run_id, upload_dir):
n_numeric = len(numeric_cols)
if n_numeric >= 2:
n_components = min(50, n_numeric - 1)
n_components = min(MAX_CLUSTER_FEATURES, n_numeric - 1)
try:
from sklearn.decomposition import TruncatedSVD # fmt: skip
import numpy as np # fmt: skip
@@ -239,7 +240,7 @@ def _background_process(run_id, upload_dir):
X = df_full.to_numpy()
X = np.nan_to_num(X, nan=0.0, posinf=0.0, neginf=0.0)
svd = TruncatedSVD(n_components=n_components, random_state=42)
svd = TruncatedSVD(n_components=n_components, random_state=RANDOM_SEED)
X_svd = svd.fit_transform(X)
svd_components = n_components
@@ -291,6 +292,7 @@ def _background_process(run_id, upload_dir):
from analysis.data_loader import drop_sqlite_table # fmt: skip
drop_sqlite_table(f'_data_{run_id}')
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
pass # Expected: drop_sqlite_table may fail if table already gone
run.status = 'failed'
run.error_message = tb
+4
View File
@@ -79,6 +79,7 @@ def tool_lab_run(request):
try:
body = _json.loads(request.body)
except Exception:
logger.error("tool_lab_run failed: {}".format(traceback.format_exc()))
return JsonResponse({'error': 'Invalid JSON body'}, status=400)
tool_name = body.get('tool')
@@ -120,6 +121,7 @@ def reload_run_data(request):
try:
body = _json.loads(request.body)
except Exception:
logger.error("reload_run_data failed: {}".format(traceback.format_exc()))
return JsonResponse({'error': 'Invalid JSON body'}, status=400)
display_id = body.get('display_id')
@@ -221,6 +223,7 @@ def tool_plan(request):
steps = len(content.get('steps', []))
plans.append({'name': p.stem, 'steps': steps})
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
plans.append({'name': p.stem, 'steps': 0})
# Add auto plans with markers
@@ -237,6 +240,7 @@ def tool_plan(request):
'pinned': auto_index_map.get(auto_name, False),
})
except Exception:
logger.error("unknown failed: {}".format(traceback.format_exc()))
auto_plans.append({
'name': auto_name.replace('.json', ''),
'steps': 0,