PM-pinou
1266 lines
52 KiB
Python
1266 lines
52 KiB
Python
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import threading
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.http import HttpResponse, JsonResponse
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
|
|
from config import get_config, save_config, Config
|
|
from .models import AnalysisRun, ClusterResult, EntityProfile
|
|
|
|
|
|
def dashboard(request):
|
|
"""Home page: recent runs overview."""
|
|
runs = AnalysisRun.objects.all()[:20]
|
|
return render(request, 'analysis/dashboard.html', {'runs': runs})
|
|
|
|
|
|
def run_list(request):
|
|
"""List all analysis runs with pagination."""
|
|
page = int(request.GET.get('page', 1))
|
|
per_page = 20
|
|
total = AnalysisRun.objects.count()
|
|
runs = AnalysisRun.objects.all()[(page - 1) * per_page: page * per_page]
|
|
return render(request, 'analysis/run_list.html', {
|
|
'runs': runs,
|
|
'page': page,
|
|
'total_pages': (total + per_page - 1) // per_page if total else 1,
|
|
})
|
|
|
|
|
|
def run_detail(request, run_id):
|
|
"""Details for a single analysis run."""
|
|
run = get_object_or_404(AnalysisRun, id=run_id)
|
|
clusters = run.clusters.all().order_by('-size')
|
|
return render(request, 'analysis/run_detail.html', {
|
|
'run': run,
|
|
'clusters': clusters,
|
|
})
|
|
|
|
|
|
@csrf_exempt
|
|
def retry_analysis(request, run_id):
|
|
"""Retry a failed analysis run. Creates a new run with same parameters."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
|
|
original = get_object_or_404(AnalysisRun, id=run_id)
|
|
if original.status != 'failed':
|
|
return JsonResponse({'error': '只能重试失败的分析'}, status=400)
|
|
|
|
# Create new run copying original parameters
|
|
new_run = AnalysisRun.objects.create(
|
|
csv_glob=original.csv_glob,
|
|
entity_column=original.entity_column,
|
|
status='pending',
|
|
total_flows=original.total_flows,
|
|
)
|
|
|
|
def _retry_fn(run, ctx):
|
|
from analysis.data_loader import load_csv_directory
|
|
from analysis.session_store import SessionStore
|
|
from analysis.entity_detector import detect_entity_column
|
|
from analysis.entity_aggregator import aggregate_by_entity
|
|
|
|
store = SessionStore()
|
|
try:
|
|
lf, schema, row_count, file_count, memory_mb = load_csv_directory(run.csv_glob)
|
|
ds_id = f'retry_{run.id}'
|
|
store.store_dataset(ds_id, lf, schema=schema, metadata={
|
|
'row_count': row_count, 'file_count': file_count, 'csv_glob': run.csv_glob,
|
|
})
|
|
run.total_flows = row_count
|
|
run.save(update_fields=['total_flows'])
|
|
|
|
# Detect entity column
|
|
result = detect_entity_column(ds_id)
|
|
entity_col = result.get('recommended')
|
|
if not entity_col:
|
|
raise Exception('Could not detect entity column')
|
|
|
|
run.entity_column = entity_col
|
|
run.save(update_fields=['entity_column'])
|
|
|
|
# Aggregate
|
|
agg_lf, feature_columns = aggregate_by_entity(lf, entity_col, schema)
|
|
df_agg = agg_lf.collect(streaming=True)
|
|
entity_ds_id = f'entity_{run.id}'
|
|
store.store_dataset(entity_ds_id, agg_lf, schema=dict(zip(feature_columns, ['']*len(feature_columns))))
|
|
run.entity_count = len(df_agg)
|
|
run.save(update_fields=['entity_count'])
|
|
|
|
# Run clustering pipeline (same defaults as manual analysis)
|
|
_run_clustering_pipeline(
|
|
run=run, store=store, entity_ds_id=entity_ds_id,
|
|
feature_columns=None, algorithm='hdbscan',
|
|
min_cluster_size=5, run_pca=True,
|
|
)
|
|
except Exception as e:
|
|
run.status = 'failed'
|
|
run.error_message = str(e)
|
|
run.save(update_fields=['status', 'error_message'])
|
|
|
|
t = threading.Thread(target=_run_pipeline_worker, args=(new_run.id, _retry_fn), daemon=True)
|
|
t.start()
|
|
|
|
return JsonResponse({'status': 'started', 'run_id': new_run.id, 'redirect': f'/runs/{new_run.id}/'})
|
|
|
|
|
|
def _extract_lat(val):
|
|
"""Try to extract a latitude from *val* (number or str)."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
v = float(val)
|
|
return None if v < -90 or v > 90 else v
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _extract_lon(val):
|
|
"""Try to extract a longitude from *val* (number or str)."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
v = float(val)
|
|
return None if v < -180 or v > 180 else v
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def cluster_overview(request, run_id):
|
|
"""Overview of all clusters for a run, with PCA scatter data and geo scatter data."""
|
|
run = get_object_or_404(AnalysisRun, id=run_id)
|
|
clusters = run.clusters.exclude(cluster_label=-1).order_by('-size').prefetch_related('features')
|
|
noise = run.clusters.filter(cluster_label=-1).first()
|
|
|
|
# Build PCA scatter data for Chart.js
|
|
entities = run.entities.exclude(embedding_x=None).exclude(embedding_y=None).only('embedding_x', 'embedding_y', 'cluster_label', 'entity_value')
|
|
scatter_data = [
|
|
{'x': e.embedding_x, 'y': e.embedding_y, 'label': e.cluster_label, 'entity': e.entity_value}
|
|
for e in entities
|
|
]
|
|
|
|
# Build geo scatter data from feature_json (lat/lon)
|
|
geo_data = []
|
|
geo_skipped = 0
|
|
for e in run.entities.all().only('feature_json', 'cluster_label', 'entity_value'):
|
|
fj = e.feature_json
|
|
if not fj:
|
|
continue
|
|
lat = _extract_lat(fj.get('avg_latitude'))
|
|
lon = _extract_lon(fj.get('avg_longitude'))
|
|
if lat is not None and lon is not None:
|
|
geo_data.append({
|
|
'x': lon,
|
|
'y': lat,
|
|
'label': e.cluster_label if e.cluster_label is not None else -1,
|
|
'entity': e.entity_value,
|
|
})
|
|
else:
|
|
geo_skipped += 1
|
|
|
|
# Cluster sizes for bar chart
|
|
cluster_sizes = [{'label': c.cluster_label, 'size': c.size, 'silhouette': c.silhouette_score}
|
|
for c in clusters]
|
|
|
|
# Top features per cluster from ClusterFeature
|
|
top_features = {}
|
|
for c in clusters:
|
|
features_qs = c.features.all().order_by('-distinguishing_score')[:10]
|
|
top_features[str(c.cluster_label)] = [
|
|
{'feature_name': f.feature_name, 'score': f.distinguishing_score, 'mean': f.mean}
|
|
for f in features_qs
|
|
]
|
|
|
|
return render(request, 'analysis/cluster_overview.html', {
|
|
'run': run,
|
|
'clusters': clusters,
|
|
'noise': noise,
|
|
'scatter_data_json': json.dumps(scatter_data),
|
|
'geo_data_json': json.dumps(geo_data),
|
|
'geo_skipped': geo_skipped,
|
|
'geo_count': len(geo_data),
|
|
'cluster_sizes_json': json.dumps(cluster_sizes),
|
|
'top_features_json': json.dumps(top_features),
|
|
})
|
|
|
|
|
|
def cluster_detail(request, run_id, cluster_label):
|
|
"""Detailed view of a single cluster with entity list."""
|
|
run = get_object_or_404(AnalysisRun, id=run_id)
|
|
cluster = get_object_or_404(ClusterResult, run=run, cluster_label=cluster_label)
|
|
features = cluster.features.all().order_by('-distinguishing_score')[:50]
|
|
entities = cluster.entities.all()[:50]
|
|
|
|
return render(request, 'analysis/cluster_detail.html', {
|
|
'run': run,
|
|
'cluster': cluster,
|
|
'features': features,
|
|
'entities': entities,
|
|
})
|
|
|
|
|
|
def entity_profile(request, entity_id):
|
|
"""Detailed profile for a single entity."""
|
|
entity = get_object_or_404(EntityProfile, id=entity_id)
|
|
run = entity.run
|
|
|
|
# Compute deviation from cluster mean
|
|
deviations = {}
|
|
if entity.feature_json and entity.cluster:
|
|
cluster_features = {
|
|
f.feature_name: {'mean': f.mean, 'std': f.std}
|
|
for f in entity.cluster.features.all()
|
|
}
|
|
for col, val in entity.feature_json.items():
|
|
if col in cluster_features and cluster_features[col].get('std') and cluster_features[col]['std'] > 1e-10:
|
|
deviations[col] = {
|
|
'value': val,
|
|
'cluster_mean': cluster_features[col]['mean'],
|
|
'zscore': (val - cluster_features[col]['mean']) / cluster_features[col]['std'],
|
|
}
|
|
|
|
return render(request, 'analysis/entity_profile.html', {
|
|
'entity': entity,
|
|
'run': run,
|
|
'deviations': deviations,
|
|
})
|
|
|
|
|
|
def start_analysis(request):
|
|
"""Placeholder: API endpoint to trigger analysis via MCP."""
|
|
return JsonResponse({'status': 'not_implemented', 'message': 'Use MCP tools to trigger analysis'})
|
|
|
|
|
|
def config_view(request):
|
|
"""Config page: GET renders current config, POST saves updates."""
|
|
cfg = get_config()
|
|
|
|
if request.method == 'POST':
|
|
# Build a new Config from POST data (fall back to current values)
|
|
server = Config.Server(
|
|
host=request.POST.get('server_host', cfg.server.host),
|
|
port=int(request.POST.get('server_port', cfg.server.port)),
|
|
debug=request.POST.get('server_debug') == 'true',
|
|
)
|
|
data = Config.Data(
|
|
schema_strict=request.POST.get('data_schema_strict') == 'true',
|
|
recursive=request.POST.get('data_recursive') == 'true',
|
|
)
|
|
clustering = Config.Clustering(
|
|
algorithm=request.POST.get('clustering_algorithm', cfg.clustering.algorithm),
|
|
min_cluster_size=int(request.POST.get('clustering_min_cluster_size', cfg.clustering.min_cluster_size)),
|
|
random_state=cfg.clustering.random_state,
|
|
)
|
|
llm = Config.LLM(
|
|
enabled=request.POST.get('llm_enabled') == 'true',
|
|
base_url=request.POST.get('llm_base_url', cfg.llm.base_url),
|
|
api_key=request.POST.get('llm_api_key', cfg.llm.api_key),
|
|
model=request.POST.get('llm_model', cfg.llm.model),
|
|
)
|
|
cfg = Config(server=server, data=data, clustering=clustering, llm=llm)
|
|
save_config(cfg)
|
|
# Re-render with saved flag
|
|
return render(request, 'tianxuan/config.html', {'config': cfg, 'saved': True})
|
|
|
|
return render(request, 'tianxuan/config.html', {'config': cfg})
|
|
|
|
|
|
@csrf_exempt
|
|
def llm_test(request):
|
|
"""LLM connectivity test: POST JSON {base_url, api_key, model} → chat/completions."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'success': False, 'message': '仅支持 POST 请求', 'latency_ms': 0}, status=405)
|
|
|
|
try:
|
|
body = json.loads(request.body)
|
|
except json.JSONDecodeError:
|
|
return JsonResponse({'success': False, 'message': '无效的 JSON 请求体', 'latency_ms': 0})
|
|
|
|
base_url = (body.get('base_url') or '').rstrip('/')
|
|
api_key = body.get('api_key') or ''
|
|
model = body.get('model') or ''
|
|
|
|
if not base_url:
|
|
return JsonResponse({'success': False, 'message': 'Base URL 不能为空', 'latency_ms': 0})
|
|
if not api_key:
|
|
return JsonResponse({'success': False, 'message': 'API Key 不能为空', 'latency_ms': 0})
|
|
if not model:
|
|
return JsonResponse({'success': False, 'message': 'Model 不能为空', 'latency_ms': 0})
|
|
|
|
url = f'{base_url}/chat/completions'
|
|
payload = json.dumps({
|
|
'model': model,
|
|
'messages': [{'role': 'user', 'content': 'respond with ok'}],
|
|
'max_tokens': 5,
|
|
}).encode('utf-8')
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=payload,
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'Bearer {api_key}',
|
|
},
|
|
method='POST',
|
|
)
|
|
|
|
start = time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
if 'choices' in data and len(data['choices']) > 0:
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=True latency={elapsed}ms')
|
|
return JsonResponse({
|
|
'success': True,
|
|
'message': f'连接成功，模型 {model} 返回正常',
|
|
'latency_ms': elapsed,
|
|
})
|
|
else:
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=True latency={elapsed}ms')
|
|
return JsonResponse({
|
|
'success': True,
|
|
'message': f'连接成功，但响应中无 choices（原始响应已记录）',
|
|
'latency_ms': elapsed,
|
|
})
|
|
except urllib.error.HTTPError as e:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
try:
|
|
detail = json.loads(e.read().decode('utf-8', errors='replace'))
|
|
msg = detail.get('error', {}).get('message', str(e))
|
|
except Exception:
|
|
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})
|
|
except urllib.error.URLError as e:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
reason = str(e.reason) if e.reason else '连接失败'
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': f'连接失败: {reason}', 'latency_ms': elapsed})
|
|
except TimeoutError:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': '请求超时（5秒）', 'latency_ms': elapsed})
|
|
except Exception as e:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': f'未知错误: {str(e)}', 'latency_ms': elapsed})
|
|
|
|
|
|
# ── Upload page ─────────────────────────────────────────────────────────
|
|
|
|
def upload_page(request):
|
|
"""CSV upload page."""
|
|
runs = AnalysisRun.objects.all().order_by('-created_at')[:20]
|
|
return render(request, 'tianxuan/upload.html', {'runs': runs})
|
|
|
|
|
|
@csrf_exempt
|
|
def upload_csv(request):
|
|
"""Handle CSV upload AJAX POST."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
files = request.FILES.getlist('files')
|
|
if not files:
|
|
return JsonResponse({'error': 'No files uploaded'})
|
|
if len(files) > 10000:
|
|
return JsonResponse({'error': '一次最多上传 10000 个文件'})
|
|
for f in files:
|
|
if f.size > 500 * 1024 * 1024:
|
|
return JsonResponse({'warning': f'文件 {f.name} 超过 500MB，建议分割后上传'})
|
|
total_size = sum(f.size for f in files)
|
|
if total_size > 50 * 1024 * 1024 * 1024:
|
|
return JsonResponse({'error': '总大小超过 50GB 限制'})
|
|
from django.utils import timezone
|
|
ts = timezone.now().strftime('%Y%m%d_%H%M%S')
|
|
import platform
|
|
if platform.system() == 'Windows':
|
|
base = Path(os.environ.get('APPDATA', '')) / 'TianXuan' / 'data' / 'uploads'
|
|
else:
|
|
base = Path.home() / '.tianxuan' / 'data' / 'uploads'
|
|
upload_dir = base / ts
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
saved = []
|
|
for f in files:
|
|
path = upload_dir / f.name
|
|
with open(path, 'wb+') as dst:
|
|
for chunk in f.chunks():
|
|
dst.write(chunk)
|
|
saved.append(str(path))
|
|
# Auto-process in background
|
|
run = AnalysisRun.objects.create(
|
|
csv_glob=str(upload_dir / '*.csv'),
|
|
status='loading',
|
|
)
|
|
t = threading.Thread(target=_background_process, args=(run.id, upload_dir), daemon=True)
|
|
t.start()
|
|
return JsonResponse({'run_id': run.id, 'files': saved, 'status': 'loading'})
|
|
|
|
|
|
def delete_upload(request, run_id):
|
|
"""Delete uploaded files and AnalysisRun."""
|
|
run = get_object_or_404(AnalysisRun, id=run_id)
|
|
import shutil
|
|
upload_dir = None
|
|
if run.csv_glob:
|
|
p = Path(run.csv_glob).parent
|
|
if p.exists():
|
|
shutil.rmtree(p, ignore_errors=True)
|
|
run.delete()
|
|
return JsonResponse({'deleted': True})
|
|
|
|
|
|
import traceback, logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _background_process(run_id, upload_dir):
|
|
"""Background: load → detect → aggregate → ready."""
|
|
import django
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = 'tianxuan.settings'
|
|
django.setup()
|
|
from analysis.models import AnalysisRun
|
|
from analysis.data_loader import load_csv_directory
|
|
from analysis.entity_detector import detect_entity_column
|
|
from analysis.entity_aggregator import aggregate_by_entity
|
|
from analysis.session_store import SessionStore
|
|
|
|
run = AnalysisRun.objects.get(id=run_id)
|
|
store = SessionStore()
|
|
try:
|
|
# Check upload directory still exists (may have been temp-cleaned)
|
|
if not upload_dir.is_dir():
|
|
msg = '上传数据目录已不存在，请重新上传'
|
|
logger.error('[BACKGROUND] %s (run_id=%s, upload_dir=%s)', msg, run_id, upload_dir)
|
|
run.status = 'failed'
|
|
run.error_message = msg
|
|
run.save(update_fields=['status', 'error_message'])
|
|
return
|
|
|
|
run.status = 'loading'; run.progress_pct = 10; run.progress_msg = '正在加载数据...'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
lf, schema, row_count, file_count, mem = load_csv_directory(
|
|
str(upload_dir / '*.csv'), schema_strict=False
|
|
)
|
|
ds_id = f'upload_{run_id}'
|
|
store.store_dataset(ds_id, lf, schema=schema, metadata={
|
|
'row_count': row_count, 'file_count': file_count, 'upload_dir': str(upload_dir),
|
|
'csv_glob': str(upload_dir / '*.csv'),
|
|
})
|
|
run.total_flows = row_count; run.save(update_fields=['total_flows'])
|
|
|
|
run.status = 'profiling'; run.progress_pct = 30; run.progress_msg = '正在分析数据特征...'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
entity_result = detect_entity_column(ds_id)
|
|
entity_col = entity_result.get('recommended', '')
|
|
entity_cols_multi = entity_result.get('recommended_multi', [])
|
|
# Store as comma-separated (multi) or single-column
|
|
if entity_cols_multi:
|
|
entity_cols = entity_cols_multi[:2] # top 2
|
|
run.entity_column = ','.join(entity_cols)
|
|
else:
|
|
entity_cols = [entity_col] if entity_col else []
|
|
run.entity_column = entity_col
|
|
run.save(update_fields=['entity_column'])
|
|
|
|
if entity_cols:
|
|
run.status = 'aggregating'; run.progress_pct = 50; run.progress_msg = '正在构建实体画像...'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
agg_lf, features = aggregate_by_entity(lf, entity_cols, schema)
|
|
df_agg = agg_lf.collect(streaming=True)
|
|
entity_ds_id = f'entity_{run_id}'
|
|
store.store_dataset(entity_ds_id, agg_lf,
|
|
schema=dict(zip(features, [''] * len(features))),
|
|
metadata={'row_count': len(df_agg), 'entity_column': ','.join(entity_cols),
|
|
'csv_glob': str(upload_dir / '*.csv')})
|
|
run.entity_count = len(df_agg)
|
|
run.entity_column = ','.join(entity_cols)
|
|
run.save(update_fields=['entity_column', 'entity_count'])
|
|
|
|
run.status = 'ready'; run.progress_pct = 60; run.progress_msg = '数据准备就绪'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
except Exception:
|
|
tb = traceback.format_exc()
|
|
logger.error(tb)
|
|
run.status = 'failed'
|
|
run.run_log += f'\n[ERROR] {tb}'
|
|
run.save(update_fields=['run_log'])
|
|
run.error_message = tb
|
|
run.save(update_fields=['status', 'error_message'])
|
|
|
|
|
|
# ── Manual analysis page ───────────────────────────────────────────────
|
|
|
|
def manual_page(request):
|
|
"""4-step manual analysis wizard."""
|
|
step = int(request.GET.get('step', 1))
|
|
runs = AnalysisRun.objects.filter(status='ready').order_by('-created_at')[:20]
|
|
ctx = {'step': step, 'runs': runs}
|
|
# Step 2: pass entity candidates and schema for the selected run
|
|
if step >= 2:
|
|
run_id = request.GET.get('run_id')
|
|
if run_id:
|
|
try:
|
|
run = AnalysisRun.objects.get(id=int(run_id))
|
|
ctx['selected_run'] = run
|
|
# Try to get entity detection candidates from SessionStore
|
|
from analysis.session_store import SessionStore
|
|
from analysis.entity_detector import detect_entity_column
|
|
store = SessionStore()
|
|
ds_id = f'upload_{run_id}'
|
|
entry = store.get_dataset(ds_id)
|
|
if entry is not None:
|
|
schema = entry.get('schema', {})
|
|
ctx['schema_keys'] = list(schema.keys())
|
|
# Build numeric-only feature column options with dtype labels
|
|
numeric_types = {'Int8', 'Int16', 'Int32', 'Int64',
|
|
'UInt8', 'UInt16', 'UInt32', 'UInt64',
|
|
'Float32', 'Float64'}
|
|
ctx['feature_options'] = [
|
|
{'name': k, 'dtype': v}
|
|
for k, v in schema.items()
|
|
if v.split('(')[0].strip() in numeric_types
|
|
]
|
|
# Detect entity candidates
|
|
det_result = detect_entity_column(ds_id)
|
|
if not det_result.get('error'):
|
|
ctx['entity_candidates'] = det_result.get('candidates', [])
|
|
# Pre-select stored entity columns (possibly comma-separated multi)
|
|
entity_cols_str = run.entity_column
|
|
if entity_cols_str:
|
|
ctx['selected_entity_cols'] = entity_cols_str.split(',')
|
|
except (AnalysisRun.DoesNotExist, ValueError):
|
|
pass
|
|
# Step 3: pass config defaults
|
|
from config import get_config
|
|
cfg = get_config()
|
|
ctx['default_algo'] = cfg.clustering.algorithm
|
|
ctx['default_min_size'] = cfg.clustering.min_cluster_size
|
|
return render(request, 'tianxuan/manual.html', ctx)
|
|
|
|
|
|
# ── Pipeline worker ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def _run_pipeline_worker(run_id, analysis_fn, **ctx):
|
|
"""Unified background worker: setup / ORM / progress / error handling."""
|
|
import django
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = 'tianxuan.settings'
|
|
os.environ['DJANGO_ALLOW_ASYNC_UNSAFE'] = 'true'
|
|
django.setup()
|
|
|
|
run = AnalysisRun.objects.get(id=run_id)
|
|
try:
|
|
analysis_fn(run, ctx)
|
|
except Exception:
|
|
import traceback
|
|
tb = traceback.format_exc()
|
|
logger.error(tb)
|
|
run.status = 'failed'
|
|
run.error_message = tb
|
|
run.run_log += f'\n[FATAL] {tb}'
|
|
run.save(update_fields=['status', 'error_message', 'run_log'])
|
|
|
|
|
|
# ── Shared clustering pipeline ─────────────────────────────────────────
|
|
|
|
def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorithm,
|
|
min_cluster_size, run_pca=True, head=None):
|
|
"""Unified clustering pipeline: clustering → feature extraction → PCA → ORM save.
|
|
|
|
Args:
|
|
run: AnalysisRun ORM object.
|
|
store: SessionStore instance.
|
|
entity_ds_id: dataset ID in SessionStore pointing to entity-aggregated data.
|
|
feature_columns: list of feature column names (None = auto-detect).
|
|
algorithm: 'hdbscan' or 'kmeans'.
|
|
min_cluster_size: int for HDBSCAN.
|
|
run_pca: bool, whether to compute and save PCA-2D embeddings.
|
|
head: optional int, downsample to at most this many rows (min 100).
|
|
"""
|
|
import warnings
|
|
warnings.filterwarnings('ignore', category=RuntimeWarning, module='sklearn')
|
|
import asyncio, traceback, logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
try:
|
|
from analysis.tool_registry import _handle_run_clustering, _handle_extract_features
|
|
|
|
entry = store.get_dataset(entity_ds_id)
|
|
if entry is None:
|
|
run.status = 'failed'
|
|
run.error_message = 'Entity dataset not found in session store'
|
|
run.save(update_fields=['status', 'error_message'])
|
|
return
|
|
|
|
schema = entry.get('schema', {})
|
|
|
|
# ── Downsampling ──
|
|
if head is not None and head > 0:
|
|
lf = entry['lazyframe']
|
|
import polars as pl
|
|
try:
|
|
row_count = lf.select(pl.len()).collect(streaming=True).item()
|
|
except Exception:
|
|
row_count = 0
|
|
|
|
if row_count > head:
|
|
head = max(head, 100) # ensure minimum rows for clustering
|
|
run.progress_msg = f'正在采样 {head}/{row_count} 行...'
|
|
run.save(update_fields=['progress_msg'])
|
|
lf = lf.head(head)
|
|
store.store_dataset(
|
|
entity_ds_id, lf,
|
|
schema=schema,
|
|
metadata=entry.get('metadata', {}),
|
|
)
|
|
entry = store.get_dataset(entity_ds_id) # refresh
|
|
row_count = head
|
|
|
|
# Resolve feature columns
|
|
numeric_types = {'Int8', 'Int16', 'Int32', 'Int64',
|
|
'UInt8', 'UInt16', 'UInt32', 'UInt64',
|
|
'Float32', 'Float64'}
|
|
entity_cols_set = set()
|
|
if hasattr(run, 'entity_column') and run.entity_column:
|
|
entity_cols_set = {c.strip() for c in run.entity_column.split(',')}
|
|
|
|
if feature_columns:
|
|
feature_cols = [c for c in feature_columns if c in schema and c not in entity_cols_set]
|
|
else:
|
|
feature_cols = [
|
|
c for c in schema.keys()
|
|
if not c.startswith('_') and c not in entity_cols_set
|
|
and schema[c].split('(')[0].strip() in numeric_types
|
|
][:10]
|
|
|
|
if not feature_cols:
|
|
raise Exception('No valid feature columns available for clustering')
|
|
|
|
# ── Step 1: Clustering ──────────────────────────────────────────────
|
|
run.status = 'clustering'
|
|
run.progress_pct = 70
|
|
run.progress_msg = '正在进行聚类分析...'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
|
|
result = asyncio.run(_handle_run_clustering(
|
|
dataset_id=entity_ds_id,
|
|
cluster_columns=feature_cols,
|
|
algorithm=algorithm,
|
|
params={'min_cluster_size': min_cluster_size},
|
|
random_state=42,
|
|
))
|
|
|
|
if 'error' in result:
|
|
err = result['error']
|
|
if 'numeric' in err.lower():
|
|
available = [
|
|
f"{k}({v})" for k, v in schema.items()
|
|
if v.split('(')[0].strip() in numeric_types and not k.startswith('_')
|
|
]
|
|
err += f"。可用数值列: {available}"
|
|
raise Exception(err)
|
|
|
|
cluster_id = result.get('cluster_result_id', '')
|
|
run.cluster_count = result.get('n_clusters', 0)
|
|
run.save(update_fields=['cluster_count'])
|
|
|
|
# ── Step 2: Feature extraction ──────────────────────────────────────
|
|
run.status = 'extracting'
|
|
run.progress_pct = 90
|
|
run.progress_msg = '正在提取聚类特征...'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
|
|
feat_result = asyncio.run(_handle_extract_features(
|
|
dataset_id=entity_ds_id,
|
|
cluster_result_id=cluster_id,
|
|
top_k=10, method='zscore', save_to_db=True,
|
|
run_id=run.id,
|
|
))
|
|
|
|
if 'error' in feat_result:
|
|
raise Exception(feat_result['error'])
|
|
|
|
# ── Step 3: PCA-2D embedding (downsampled for large data) ─────────
|
|
if run_pca:
|
|
try:
|
|
import polars as pl
|
|
lf = entry['lazyframe']
|
|
# Sample max 50K rows for PCA to avoid memory blowup on 10M entities
|
|
MAX_PCA = 50000
|
|
try:
|
|
n_total = lf.select(pl.len()).collect(streaming=True).item()
|
|
except Exception:
|
|
n_total = 0
|
|
if n_total > MAX_PCA:
|
|
lf_pca = lf.sample(n=MAX_PCA, seed=42)
|
|
run.progress_msg = f'PCA 采样 {MAX_PCA}/{n_total} 实体...'
|
|
run.save(update_fields=['progress_msg'])
|
|
else:
|
|
lf_pca = lf
|
|
df_pca = lf_pca.collect(streaming=True)
|
|
from sklearn.preprocessing import StandardScaler
|
|
from sklearn.decomposition import PCA
|
|
import numpy as np
|
|
|
|
num_cols = [c for c in df_pca.columns
|
|
if df_pca[c].dtype in (
|
|
pl.Int8, pl.Int16, pl.Int32, pl.Int64,
|
|
pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64,
|
|
pl.Float32, pl.Float64)]
|
|
|
|
if len(num_cols) >= 2 and len(df_pca) > 2:
|
|
mat = np.nan_to_num(StandardScaler().fit_transform(
|
|
df_pca.select(num_cols).to_numpy()), nan=0.0)
|
|
coords = PCA(n_components=2, random_state=42).fit_transform(mat)
|
|
|
|
non_num = [c for c in df_pca.columns if c not in num_cols and not c.startswith('_')]
|
|
ent_col = non_num[0] if non_num else df_pca.columns[0]
|
|
|
|
cluster_entry = store.get_cluster_result(cluster_id)
|
|
labels_list = cluster_entry.get('labels', []) if cluster_entry else []
|
|
|
|
from analysis.models import EntityProfile as EP
|
|
from analysis.models import ClusterResult as CR
|
|
|
|
profiles = []
|
|
cr_cache = {}
|
|
for i, row in enumerate(df_pca.iter_rows(named=True)):
|
|
ev = str(row.get(ent_col, ''))
|
|
if not ev:
|
|
continue
|
|
lbl = labels_list[i] if i < len(labels_list) else -1
|
|
cache_key = f'{run.id}_{lbl}'
|
|
if cache_key not in cr_cache:
|
|
cr_cache[cache_key] = CR.objects.filter(run=run, cluster_label=lbl).first()
|
|
profiles.append(EP(
|
|
entity_value=ev, run=run, cluster_label=lbl,
|
|
cluster=cr_cache[cache_key],
|
|
embedding_x=float(coords[i, 0]) if i < len(coords) else None,
|
|
embedding_y=float(coords[i, 1]) if i < len(coords) else None,
|
|
))
|
|
if profiles:
|
|
EP.objects.bulk_create(profiles, ignore_conflicts=True, batch_size=1000)
|
|
except Exception as pca_err:
|
|
logger.warning(f'PCA-2D embedding skipped: {pca_err}')
|
|
|
|
# ── Done ──
|
|
run.status = 'completed'
|
|
run.progress_pct = 100
|
|
run.progress_msg = '分析完成'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
|
|
|
except Exception:
|
|
tb = traceback.format_exc()
|
|
logger.error(tb)
|
|
try:
|
|
run.status = 'failed'
|
|
run.progress_msg = f'失败: {tb}'
|
|
run.run_log += f'\n[ERROR] {tb}'
|
|
run.save(update_fields=['run_log'])
|
|
run.error_message = tb
|
|
run.save(update_fields=['status', 'progress_msg', 'error_message'])
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
# ── Manual analysis page ───────────────────────────────────────────────
|
|
|
|
@csrf_exempt
|
|
def run_analysis(request):
|
|
"""AJAX POST: run analysis pipeline and return result page URL."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
import json
|
|
body = json.loads(request.body)
|
|
run_id = body.get('run_id')
|
|
entity_col = body.get('entity_column', '')
|
|
entity_columns = body.get('entity_columns')
|
|
feature_columns = body.get('feature_columns')
|
|
algorithm = body.get('algorithm', 'hdbscan')
|
|
min_cluster_size = int(body.get('min_cluster_size', 5))
|
|
head = body.get('head')
|
|
|
|
def _analysis_fn(run, ctx):
|
|
from analysis.session_store import SessionStore
|
|
from analysis.entity_aggregator import aggregate_by_entity
|
|
|
|
entity_columns = ctx['entity_columns']
|
|
feature_columns = ctx['feature_columns']
|
|
algorithm = ctx['algorithm']
|
|
min_cluster_size = ctx['min_cluster_size']
|
|
head = ctx.get('head')
|
|
|
|
store = SessionStore()
|
|
try:
|
|
if entity_columns:
|
|
upload_ds_id = f'upload_{run_id}'
|
|
upload_entry = store.get_dataset(upload_ds_id)
|
|
if upload_entry is None:
|
|
run.error_message = '请先上传并等待预处理完成'
|
|
run.status = 'failed'
|
|
run.save(update_fields=['status', 'error_message'])
|
|
return
|
|
upload_lf = upload_entry['lazyframe']
|
|
upload_schema = upload_entry.get('schema', {})
|
|
agg_lf, agg_features = aggregate_by_entity(upload_lf, entity_columns, upload_schema)
|
|
df_agg = agg_lf.collect(streaming=True)
|
|
agg_schema = {n: str(d) for n, d in zip(df_agg.columns, [df_agg[c].dtype for c in df_agg.columns])}
|
|
entity_ds_id = f'entity_{run_id}_custom'
|
|
store.store_dataset(
|
|
dataset_id=entity_ds_id,
|
|
lazyframe=df_agg.lazy(),
|
|
schema=agg_schema,
|
|
metadata={'row_count': len(df_agg), 'entity_columns': entity_columns,
|
|
'csv_glob': run.csv_glob},
|
|
)
|
|
run.entity_column = ','.join(entity_columns)
|
|
run.entity_count = len(df_agg)
|
|
run.save(update_fields=['entity_column', 'entity_count'])
|
|
else:
|
|
entity_ds_id = f'entity_{run_id}'
|
|
entry = store.get_dataset(entity_ds_id)
|
|
if entry is None:
|
|
run.error_message = '请先上传并等待预处理完成'
|
|
run.status = 'failed'
|
|
run.save(update_fields=['status', 'error_message'])
|
|
return
|
|
|
|
# Use the unified clustering pipeline (clustering → extraction → PCA)
|
|
from analysis.views import _run_clustering_pipeline
|
|
_run_clustering_pipeline(
|
|
run=run, store=store, entity_ds_id=entity_ds_id,
|
|
feature_columns=feature_columns,
|
|
algorithm=algorithm, min_cluster_size=min_cluster_size,
|
|
run_pca=True, head=head,
|
|
)
|
|
except Exception:
|
|
tb = traceback.format_exc()
|
|
logger.error(tb)
|
|
run.status = 'failed'
|
|
run.progress_msg = f'失败: {tb}'
|
|
run.run_log += f'\n[ERROR] {tb}'
|
|
run.save(update_fields=['run_log'])
|
|
run.error_message = tb
|
|
run.save(update_fields=['status', 'progress_msg', 'error_message'])
|
|
|
|
t = threading.Thread(
|
|
target=_run_pipeline_worker,
|
|
args=(run_id, _analysis_fn),
|
|
kwargs=dict(entity_col=entity_col, entity_columns=entity_columns,
|
|
feature_columns=feature_columns, algorithm=algorithm,
|
|
min_cluster_size=min_cluster_size, head=head),
|
|
daemon=True,
|
|
)
|
|
t.start()
|
|
return JsonResponse({'status': 'started', 'redirect': f'/runs/{run_id}/'})
|
|
|
|
|
|
|
|
|
|
# ── LLM auto analysis page ─────────────────────────────────────────────
|
|
|
|
|
|
@csrf_exempt
|
|
def run_llm_analysis_view(request):
|
|
"""AJAX POST: run LLM-driven analysis pipeline."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
import json
|
|
body = json.loads(request.body)
|
|
run_id = body.get('run_id')
|
|
entity_col = body.get('entity_column', '')
|
|
head = body.get('head')
|
|
|
|
from analysis.session_store import SessionStore
|
|
store = SessionStore()
|
|
upload_ds_id = f'upload_{run_id}'
|
|
entry = store.get_dataset(upload_ds_id)
|
|
|
|
if entry is None:
|
|
# Try entity dataset as fallback
|
|
entity_ds_id = f'entity_{run_id}'
|
|
entry = store.get_dataset(entity_ds_id)
|
|
if entry is None:
|
|
return JsonResponse({'error': '数据集不存在，请先上传并等待预处理完成'})
|
|
|
|
dataset_id = upload_ds_id if store.get_dataset(upload_ds_id) else f'entity_{run_id}'
|
|
|
|
def _analysis_fn(run, ctx):
|
|
from tianxuan.llm_orchestrator import run_llm_pipeline, LLMConfig
|
|
dataset_id = ctx['dataset_id']
|
|
entity_col = ctx['entity_col']
|
|
|
|
def progress_callback(step, tool_name, result):
|
|
nonlocal run
|
|
try:
|
|
run = AnalysisRun.objects.get(id=run_id)
|
|
pct = min(10 + step * 6, 95)
|
|
run.progress_pct = pct
|
|
run.progress_msg = f'步骤 {step}: {tool_name}'
|
|
if result:
|
|
summary = str(result)
|
|
run.run_log += f'[{step}] {tool_name}: {summary}\n'
|
|
else:
|
|
run.run_log += f'[{step}] {tool_name}\n'
|
|
run.save(update_fields=['progress_pct', 'progress_msg', 'run_log'])
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
from config import get_config
|
|
cfg = get_config()
|
|
llm_cfg = LLMConfig(cfg.llm.base_url, cfg.llm.api_key, cfg.llm.model)
|
|
run.run_log = '启动 LLM 分析...\n'
|
|
run.status = 'loading'
|
|
run.progress_pct = 5
|
|
run.progress_msg = 'LLM 初始化中...'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg', 'run_log'])
|
|
|
|
result = run_llm_pipeline(dataset_id, entity_col, config=llm_cfg, callback=progress_callback)
|
|
|
|
if 'error' in result:
|
|
run.status = 'failed'
|
|
run.progress_msg = f'失败: {result["error"]}'
|
|
run.run_log += f'\n[ERROR] {result["error"]}'
|
|
else:
|
|
run.status = 'completed'
|
|
run.progress_pct = 100
|
|
run.progress_msg = 'LLM 分析完成'
|
|
run.run_log += f'\n[完成] {result.get("result", "")}'
|
|
run.save(update_fields=['status', 'progress_pct', 'progress_msg', 'run_log'])
|
|
except Exception:
|
|
tb = traceback.format_exc()
|
|
logger.error(tb)
|
|
try:
|
|
run = AnalysisRun.objects.get(id=run_id)
|
|
run.status = 'failed'
|
|
run.progress_msg = f'失败: {tb}'
|
|
run.run_log += f'\n[ERROR] {tb}'
|
|
run.error_message = tb
|
|
run.save(update_fields=['status', 'progress_msg', 'run_log', 'error_message'])
|
|
except Exception:
|
|
pass
|
|
|
|
t = threading.Thread(
|
|
target=_run_pipeline_worker,
|
|
args=(run_id, _analysis_fn),
|
|
kwargs=dict(dataset_id=dataset_id, entity_col=entity_col, head=head),
|
|
daemon=True,
|
|
)
|
|
t.start()
|
|
return JsonResponse({'status': 'started', 'run_id': run_id})
|
|
|
|
|
|
def auto_page(request):
|
|
"""LLM-driven auto analysis page."""
|
|
runs = AnalysisRun.objects.filter(status='ready').order_by('-created_at')[:20]
|
|
cfg = get_config()
|
|
return render(request, 'tianxuan/auto.html', {
|
|
'runs': runs,
|
|
'llm_configured': bool(cfg.llm.base_url and cfg.llm.api_key),
|
|
})
|
|
|
|
|
|
# ── Progress API ───────────────────────────────────────────────────────
|
|
|
|
def run_status_api(request, run_id):
|
|
"""Return current run status as JSON."""
|
|
run = get_object_or_404(AnalysisRun, id=run_id)
|
|
return JsonResponse({
|
|
'id': run.id,
|
|
'status': run.status,
|
|
'entity_count': run.entity_count,
|
|
'cluster_count': run.cluster_count,
|
|
'error_message': run.error_message,
|
|
'progress_pct': run.progress_pct,
|
|
'progress_msg': run.progress_msg,
|
|
'run_log': run.run_log[-2000:], # last 2000 chars
|
|
})
|
|
|
|
|
|
def log_viewer(request):
|
|
"""Display last 200 lines of the app log (DEBUG only).
|
|
With ?pos=N, returns text: line_count\\nnew_content for AJAX polling.
|
|
"""
|
|
from django.conf import settings
|
|
if not settings.DEBUG:
|
|
return HttpResponse('Log viewer only available in DEBUG mode', status=403)
|
|
log_path = settings.BASE_DIR / 'logs' / 'tianxuan.log'
|
|
pos = request.GET.get('pos')
|
|
if pos is not None:
|
|
# AJAX polling mode: return new content since byte position
|
|
try:
|
|
pos = int(pos)
|
|
except ValueError:
|
|
pos = 0
|
|
if log_path.exists():
|
|
with open(log_path, 'r', encoding='utf-8') as f:
|
|
total = f.seek(0, 2) # seek to end
|
|
if pos >= total:
|
|
return HttpResponse(f'{total}\n', content_type='text/plain')
|
|
f.seek(pos)
|
|
new_content = f.read()
|
|
return HttpResponse(f'{total}\n{new_content}', content_type='text/plain')
|
|
return HttpResponse('0\n', content_type='text/plain')
|
|
|
|
# HTML page mode
|
|
lines = []
|
|
if log_path.exists():
|
|
with open(log_path, 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()[-200:]
|
|
return render(request, 'tianxuan/log_viewer.html', {'lines': lines})
|
|
|
|
# ── 3D Globe view ─────────────────────────────────────────────────────────
|
|
|
|
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.
|
|
|
|
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)
|
|
src_lon_col = next((c for c in df.columns if c.lower() in (
|
|
':ips.lond', ':ips_lond', ':ips_lon', 'src_longitude', 'src_lon', 'source_longitude', 'source_lon')), None)
|
|
dst_lat_col = next((c for c in df.columns if c.lower() in (
|
|
':ipd.latd', ':ipd_latd', ':ipd_lat', 'dst_latitude', 'dst_lat', 'dest_latitude', 'dest_lat')), None)
|
|
dst_lon_col = next((c for c in df.columns if c.lower() in (
|
|
':ipd.lond', ':ipd_lond', ':ipd_lon', 'dst_longitude', 'dst_lon', 'dest_longitude', 'dest_lon')), None)
|
|
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)
|
|
bytes_col = next((c for c in df.columns if c.lower() in ('8ack', '8pak', '8byt', 'bytes_sent', 'bytes')), None)
|
|
|
|
flows = []
|
|
for row in df.iter_rows(named=True):
|
|
slat = slon = dlat = dlon = None
|
|
|
|
if src_lat_col and src_lon_col:
|
|
slat = _extract_lat(row.get(src_lat_col))
|
|
slon = _extract_lon(row.get(src_lon_col))
|
|
if dst_lat_col and dst_lon_col:
|
|
dlat = _extract_lat(row.get(dst_lat_col))
|
|
dlon = _extract_lon(row.get(dst_lon_col))
|
|
|
|
if (slat is None or slon is None) and src_ip_col:
|
|
sg = geo_lookup(str(row.get(src_ip_col, '')))
|
|
if sg:
|
|
slat, slon = sg['lat'], sg['lon']
|
|
if (dlat is None or dlon is None) and dst_ip_col:
|
|
dg = geo_lookup(str(row.get(dst_ip_col, '')))
|
|
if dg:
|
|
dlat, dlon = dg['lat'], dg['lon']
|
|
|
|
if slat is None or slon is None or dlat is None or dlon is None:
|
|
continue
|
|
|
|
tls_val = str(row.get(tls_col or '', ''))
|
|
# Check hex format first (e.g. "03 03" or "0303")
|
|
hex_key = tls_val.replace(' ', '').strip()
|
|
if hex_key in TLS_HEX_MAP:
|
|
tls_label = TLS_HEX_MAP[hex_key]
|
|
elif '1.3' in tls_val:
|
|
tls_label = 'TLSv1.3'
|
|
elif '1.2' in tls_val:
|
|
tls_label = 'TLSv1.2'
|
|
else:
|
|
tls_label = 'other'
|
|
|
|
bs = float(row.get(bytes_col or '', 0) or 0)
|
|
flows.append({
|
|
'slat': slat, 'slon': slon,
|
|
'dlat': dlat, 'dlon': dlon,
|
|
'tls': tls_label,
|
|
'bytes': bs,
|
|
})
|
|
return flows
|
|
|
|
|
|
def globe_view(request):
|
|
import json
|
|
from analysis.geoip import lookup as geo_lookup
|
|
from analysis.data_loader import load_csv_directory
|
|
from analysis.models import AnalysisRun
|
|
from analysis.session_store import SessionStore
|
|
|
|
MAX_ROWS_PER_RUN = 300 # 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')
|
|
if data_param:
|
|
store = SessionStore()
|
|
flows = []
|
|
all_runs = list(AnalysisRun.objects.filter(status='completed').order_by('-id').values('id', 'csv_glob'))
|
|
try:
|
|
entry = store.get_dataset(data_param)
|
|
if entry is not None:
|
|
lf = entry['lazyframe']
|
|
df = lf.head(MAX_ROWS_PER_RUN).collect()
|
|
flows = _extract_flows_from_df(df, MAX_ROWS_PER_RUN)
|
|
except Exception as exc:
|
|
import logging
|
|
logging.getLogger(__name__).warning(f'globe ?data= param failed: {exc}')
|
|
return render(request, 'tianxuan/globe.html', {
|
|
'geo_flows': json.dumps(flows),
|
|
'all_runs': all_runs,
|
|
'selected_ids': [],
|
|
})
|
|
|
|
# Get all completed runs for the selector
|
|
all_runs = list(AnalysisRun.objects.filter(status='completed').order_by('-id').values('id', 'csv_glob'))
|
|
|
|
# Get selected run IDs from query params (multiple checkboxes)
|
|
selected_ids_list = request.GET.getlist('runs')
|
|
selected_runs = []
|
|
if selected_ids_list:
|
|
for sid in selected_ids_list:
|
|
sid = sid.strip()
|
|
if sid.isdigit():
|
|
try:
|
|
r = AnalysisRun.objects.get(id=int(sid), status='completed')
|
|
selected_runs.append(r)
|
|
except AnalysisRun.DoesNotExist:
|
|
pass
|
|
|
|
# Check if user-selected runs have valid files; if not, fall through
|
|
all_selected_failed = bool(selected_ids_list)
|
|
fallback_run = None
|
|
|
|
if not selected_runs:
|
|
all_selected_failed = False
|
|
|
|
flows = []
|
|
|
|
for run in selected_runs:
|
|
try:
|
|
lf, schema, rc, fc, mem = load_csv_directory(run.csv_glob)
|
|
df = lf.head(MAX_ROWS_PER_RUN).collect()
|
|
flows.extend(_extract_flows_from_df(df, MAX_ROWS_PER_RUN))
|
|
all_selected_failed = False
|
|
break # stop after first successful load
|
|
except Exception as exc:
|
|
logger.warning('[GLOBE] Failed to load run %s (%s): %s', run.id, run.csv_glob, exc)
|
|
continue
|
|
|
|
if all_selected_failed or not flows:
|
|
# Fallback: latest completed run with valid files
|
|
for latest in AnalysisRun.objects.filter(status='completed').order_by('-id'):
|
|
if not latest.csv_glob:
|
|
continue
|
|
glob_path = Path(latest.csv_glob)
|
|
try:
|
|
if '/*' in latest.csv_glob:
|
|
if glob_path.parent.is_dir():
|
|
lf, schema, rc, fc, mem = load_csv_directory(latest.csv_glob)
|
|
df = lf.head(MAX_ROWS_PER_RUN).collect()
|
|
flows.extend(_extract_flows_from_df(df, MAX_ROWS_PER_RUN))
|
|
selected_runs = [latest]
|
|
break
|
|
elif glob_path.exists():
|
|
lf, schema, rc, fc, mem = load_csv_directory(latest.csv_glob)
|
|
df = lf.head(MAX_ROWS_PER_RUN).collect()
|
|
flows.extend(_extract_flows_from_df(df, MAX_ROWS_PER_RUN))
|
|
selected_runs = [latest]
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
resp = render(request, 'tianxuan/globe.html', {
|
|
'geo_flows': json.dumps(flows),
|
|
'all_runs': all_runs,
|
|
'selected_ids': [r.id for r in selected_runs],
|
|
})
|
|
resp['Cache-Control'] = 'no-cache, no-store, must-revalidate'
|
|
resp['Pragma'] = 'no-cache'
|
|
resp['Expires'] = '0'
|
|
return resp
|
|
|
|
|
|
# ── MCP Tool Lab — manual tool execution ─────────────────────────────
|
|
|
|
|
|
def tool_lab(request):
|
|
"""List all available MCP tools with descriptions and input schemas."""
|
|
from analysis.tool_registry import get_tools_meta
|
|
tools = get_tools_meta()
|
|
# Group tools by category
|
|
core_names = {'profile_data', 'build_entity_profiles', 'compute_scores',
|
|
'run_clustering', 'extract_features', 'detect_anomalies',
|
|
'visualize_anomalies'}
|
|
diag_names = {'validate_data', 'explore_distributions', 'find_outliers',
|
|
'diagnose_clustering', 'compare_datasets', 'export_debug_sample',
|
|
'repair_schema'}
|
|
analysis_names = {'analyze_patterns', 'analyze_temporal', 'analyze_tls_health',
|
|
'analyze_geo_distribution', 'analyze_entity_detail'}
|
|
|
|
core_tools, diag_tools, analysis_tools, other_tools = [], [], [], []
|
|
for t in tools:
|
|
if t.name in core_names:
|
|
core_tools.append(t)
|
|
elif t.name in diag_names:
|
|
diag_tools.append(t)
|
|
elif t.name in analysis_names:
|
|
analysis_tools.append(t)
|
|
else:
|
|
other_tools.append(t)
|
|
|
|
# Get list of datasets for parameter autofill
|
|
from analysis.session_store import SessionStore
|
|
store = SessionStore()
|
|
ds_list = store.list_datasets() if hasattr(store, 'list_datasets') else []
|
|
|
|
import json as _json
|
|
# Serialize tools metadata for JS
|
|
meta_list = []
|
|
for t in tools:
|
|
meta_list.append({
|
|
'name': t.name,
|
|
'description': t.description,
|
|
'inputSchema': t.inputSchema,
|
|
})
|
|
return render(request, 'analysis/tool_lab.html', {
|
|
'core_tools': core_tools,
|
|
'diag_tools': diag_tools,
|
|
'analysis_tools': analysis_tools,
|
|
'other_tools': other_tools,
|
|
'datasets': ds_list,
|
|
'tools_meta': _json.dumps(meta_list, ensure_ascii=False),
|
|
})
|
|
|
|
|
|
@csrf_exempt
|
|
def tool_lab_run(request):
|
|
"""Execute a single MCP tool via AJAX POST."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
import json as _json
|
|
try:
|
|
body = _json.loads(request.body)
|
|
except Exception:
|
|
return JsonResponse({'error': 'Invalid JSON body'}, status=400)
|
|
|
|
tool_name = body.get('tool')
|
|
params = body.get('params', {})
|
|
if not tool_name:
|
|
return JsonResponse({'error': 'Missing tool name'}, status=400)
|
|
|
|
from analysis.tool_registry import handle_call
|
|
import asyncio
|
|
try:
|
|
result = asyncio.run(handle_call(tool_name, params))
|
|
return JsonResponse({'status': 'ok', 'result': result})
|
|
except ValueError as e:
|
|
return JsonResponse({'error': str(e)}, status=400)
|
|
except Exception as e:
|
|
import traceback
|
|
return JsonResponse({'error': str(e), 'traceback': traceback.format_exc()}, status=500)
|
|
|