refactor: split analysis/views.py into views/ package (12 sub-modules)
- helpers.py: _extract_lat, _extract_lon, plan index utilities - dashboard.py: dashboard, run_list, run_detail, run_status_api, start_analysis - pipeline.py: _run_pipeline_worker, _background_process - clustering.py: cluster_overview, cluster_detail, _run_clustering_pipeline, _get_globe_flows - entity.py: entity_profile - upload.py: upload_page, upload_csv, upload_csv_batch, finalize_upload, delete_upload - auto.py: auto_page, run_llm_analysis_view - manual.py: manual_page, manual_run_analysis - globe.py: globe_view, _extract_flows_from_df - config.py: config_view, llm_test - log_viewer.py: log_viewer - tools.py: tool_lab, tool_lab_run, tool_plan, apply_filter, reload_run_data, retry_run All function logic preserved exactly. views/__init__.py re-exports all public functions for backward compatibility with urls.py and run_pipeline.py. Original views.py archived as views_backup.py.
This commit is contained in:
PM-pinou
@@ -0,0 +1,17 @@
|
||||
"""Analysis views package — re-exports all public view functions."""
|
||||
from .helpers import _extract_lat, _extract_lon
|
||||
from .dashboard import dashboard, run_list, run_detail, run_status_api, start_analysis
|
||||
from .pipeline import _run_pipeline_worker, _background_process
|
||||
from .clustering import cluster_overview, cluster_detail, _run_clustering_pipeline, _get_globe_flows
|
||||
from .entity import entity_profile
|
||||
from .upload import upload_page, upload_csv, upload_csv_batch, finalize_upload, delete_upload
|
||||
from .manual import manual_page, manual_run_analysis
|
||||
from .auto import auto_page, run_llm_analysis_view
|
||||
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__)
|
||||
@@ -0,0 +1,344 @@
|
||||
"""LLM auto analysis views."""
|
||||
import json
|
||||
import asyncio
|
||||
import threading
|
||||
import traceback
|
||||
import logging
|
||||
|
||||
from django.shortcuts import render, get_object_or_404
|
||||
from django.http import JsonResponse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from config import get_config
|
||||
from analysis.models import AnalysisRun
|
||||
from .pipeline import _run_pipeline_worker
|
||||
from .clustering import _run_clustering_pipeline
|
||||
from .helpers import _PLANS_DIR, _add_to_auto_index
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def auto_page(request):
|
||||
"""LLM-driven auto analysis page."""
|
||||
from analysis.session_store import SessionStore
|
||||
import json as _json
|
||||
|
||||
cfg = get_config()
|
||||
|
||||
# Get datasets from SessionStore + ALL runs from DB
|
||||
store = SessionStore()
|
||||
session_datasets = store.list_datasets() if hasattr(store, 'list_datasets') else []
|
||||
session_ds_map = {ds['dataset_id']: ds for ds in session_datasets}
|
||||
|
||||
# Build dataset entries from ALL AnalysisRun records
|
||||
all_runs = AnalysisRun.objects.all().order_by('-created_at')[:100]
|
||||
datasets_with_status = []
|
||||
|
||||
for run in all_runs:
|
||||
if run.run_type == 'upload':
|
||||
ds_id = f'upload_{run.display_id}'
|
||||
else:
|
||||
ds_id = f'run_{run.display_id}'
|
||||
|
||||
session_entry = session_ds_map.get(ds_id)
|
||||
|
||||
if session_entry:
|
||||
datasets_with_status.append({
|
||||
**session_entry,
|
||||
'dataset_id': ds_id,
|
||||
'run_status': run.status,
|
||||
'total_flows': run.total_flows,
|
||||
'display_id': run.display_id,
|
||||
'needs_reload': False,
|
||||
'csv_glob': run.csv_glob or '',
|
||||
'sqlite_table': run.sqlite_table or '',
|
||||
})
|
||||
elif run.csv_glob or run.sqlite_table:
|
||||
datasets_with_status.append({
|
||||
'dataset_id': ds_id,
|
||||
'run_status': run.status,
|
||||
'display_id': run.display_id,
|
||||
'total_flows': run.total_flows,
|
||||
'row_count': run.total_flows,
|
||||
'file_count': None,
|
||||
'columns': [],
|
||||
'column_count': 0,
|
||||
'svd_components': 0,
|
||||
'needs_reload': True,
|
||||
'csv_glob': run.csv_glob or '',
|
||||
'sqlite_table': run.sqlite_table or '',
|
||||
})
|
||||
|
||||
for ds_info in datasets_with_status:
|
||||
s = ds_info['run_status']
|
||||
if s in ('completed',):
|
||||
ds_info['status_label'] = '已分析'
|
||||
ds_info['badge_class'] = 'completed'
|
||||
elif s in ('failed',):
|
||||
ds_info['status_label'] = '错误'
|
||||
ds_info['badge_class'] = 'failed'
|
||||
elif s in ('pending', 'loading', 'profiling', 'aggregating', 'clustering', 'extracting'):
|
||||
ds_info['status_label'] = '分析中'
|
||||
ds_info['badge_class'] = 'analyzing'
|
||||
else:
|
||||
ds_info['status_label'] = '待分析'
|
||||
ds_info['badge_class'] = 'ready'
|
||||
|
||||
return render(request, 'tianxuan/auto.html', {
|
||||
'llm_configured': bool(cfg.llm.base_url and cfg.llm.api_key),
|
||||
'datasets_with_status_json': _json.dumps(datasets_with_status, ensure_ascii=False),
|
||||
})
|
||||
|
||||
|
||||
@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)
|
||||
body = json.loads(request.body)
|
||||
display_id = body.get('run_id')
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
pk = run.id
|
||||
run.run_type = 'auto'
|
||||
run.save(update_fields=['run_type'])
|
||||
head = body.get('head')
|
||||
|
||||
from analysis.session_store import SessionStore
|
||||
store = SessionStore()
|
||||
|
||||
# Determine dataset key format (consistent with manual_page)
|
||||
upload_ds_id = f'upload_{display_id}'
|
||||
run_ds_id = f'run_{display_id}'
|
||||
entry = store.get_dataset(upload_ds_id)
|
||||
if entry is None:
|
||||
entry = store.get_dataset(run_ds_id)
|
||||
|
||||
if entry is None:
|
||||
# Try primary key fallback
|
||||
upload_pk_id = f'upload_{pk}'
|
||||
entity_pk_id = f'entity_{pk}'
|
||||
entry = store.get_dataset(upload_pk_id)
|
||||
if entry is None:
|
||||
entry = store.get_dataset(entity_pk_id)
|
||||
|
||||
if entry is None:
|
||||
# Auto-reload from sqlite_table or csv_glob
|
||||
from analysis.data_loader import load_csv_directory, load_from_db
|
||||
lf = None
|
||||
schema = {}
|
||||
if run.sqlite_table:
|
||||
try:
|
||||
lf = load_from_db(run.sqlite_table)
|
||||
except Exception:
|
||||
pass
|
||||
if lf is None and run.csv_glob:
|
||||
try:
|
||||
lf, schema, _, _, _ = load_csv_directory(run.csv_glob)
|
||||
except Exception:
|
||||
pass
|
||||
if lf is None:
|
||||
return JsonResponse({'error': '数据集不存在，请先上传并等待预处理完成'})
|
||||
entry_key = upload_ds_id if run.run_type == 'upload' else run_ds_id
|
||||
store.store_dataset(entry_key, lf, schema=schema, metadata={
|
||||
'row_count': run.total_flows,
|
||||
'csv_glob': run.csv_glob or '',
|
||||
'sqlite_table': run.sqlite_table or '',
|
||||
})
|
||||
# Re-fetch entry
|
||||
entry = store.get_dataset(entry_key)
|
||||
|
||||
# Determine the active dataset_id
|
||||
for candidate in [upload_ds_id, run_ds_id, f'upload_{pk}', f'entity_{pk}']:
|
||||
if store.get_dataset(candidate):
|
||||
dataset_id = candidate
|
||||
break
|
||||
else:
|
||||
return JsonResponse({'error': '数据集不存在，请先上传并等待预处理完成'})
|
||||
|
||||
# ── Optional: apply filter before LLM pipeline ──
|
||||
filters = body.get('filters', [])
|
||||
if filters and len(filters) > 0:
|
||||
logic = body.get('logic', 'and')
|
||||
try:
|
||||
from analysis.tool_registry import _handle_filter_data
|
||||
filter_result = asyncio.run(_handle_filter_data(
|
||||
dataset_id=dataset_id,
|
||||
filters=filters,
|
||||
logic=logic,
|
||||
))
|
||||
if 'error' not in filter_result:
|
||||
filtered_id = filter_result.get('filtered_dataset_id', '')
|
||||
if filtered_id and store.get_dataset(filtered_id):
|
||||
dataset_id = filtered_id
|
||||
except Exception:
|
||||
logger.warning(f'[LLM auto] Filter failed (non-fatal): {traceback.format_exc()}')
|
||||
|
||||
def _analysis_fn(run, ctx):
|
||||
from tianxuan.llm_orchestrator import run_llm_pipeline, LLMConfig
|
||||
dataset_id = ctx['dataset_id']
|
||||
pk = ctx['pk']
|
||||
|
||||
def progress_callback(step, tool_name, result=None, meta=None):
|
||||
nonlocal run
|
||||
try:
|
||||
run = AnalysisRun.objects.get(id=pk)
|
||||
pct = min(10 + step * 6, 95)
|
||||
run.progress_pct = pct
|
||||
run.progress_msg = f'步骤 {step}: {tool_name}'
|
||||
|
||||
save_fields = ['progress_pct', 'progress_msg']
|
||||
if tool_name == 'thinking':
|
||||
# Accumulate LLM thinking text (appended to llm_thinking)
|
||||
if result:
|
||||
run.llm_thinking = (run.llm_thinking or '') + f'\n[{step}] {result}'
|
||||
save_fields = ['progress_pct', 'progress_msg', 'llm_thinking']
|
||||
elif tool_name == 'complete':
|
||||
pass # No extra tracking needed
|
||||
else:
|
||||
# Single tool call — always has result
|
||||
if isinstance(result, dict):
|
||||
summary = json.dumps(result, default=str, ensure_ascii=False)[:500]
|
||||
else:
|
||||
summary = str(result)[:500]
|
||||
run.run_log += f'[{step}] {tool_name}: {summary}...\n'
|
||||
|
||||
# One entry per tool call
|
||||
if meta and isinstance(meta, dict) and 'args' in meta:
|
||||
tool_calls = list(run.tool_calls_json or [])
|
||||
tool_calls.append({
|
||||
'step': step,
|
||||
'name': tool_name,
|
||||
'input': meta['args'],
|
||||
'output': result,
|
||||
})
|
||||
run.tool_calls_json = tool_calls
|
||||
save_fields = ['progress_pct', 'progress_msg', 'run_log', 'tool_calls_json']
|
||||
|
||||
run.save(update_fields=save_fields)
|
||||
except Exception as e:
|
||||
logger.warning('progress_callback error: %s', e)
|
||||
|
||||
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, 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"]}'
|
||||
run.save(update_fields=['status', 'progress_pct', 'progress_msg', 'run_log'])
|
||||
else:
|
||||
run.progress_pct = 55
|
||||
run.progress_msg = f'LLM分析完成: {result.get("result", "")[:200]}'
|
||||
run.run_log += f'\n[完成] {result.get("result", "")}\n[INFO] 开始自动聚类流程...'
|
||||
run.save(update_fields=['progress_pct', 'progress_msg', 'run_log'])
|
||||
|
||||
# ── Auto-run clustering pipeline after LLM analysis ──
|
||||
entity_ds_id = None
|
||||
tool_calls = run.tool_calls_json or []
|
||||
for tc in tool_calls:
|
||||
tool_name = tc.get('name', '')
|
||||
tool_input = tc.get('input', {})
|
||||
if tool_name in ('build_entity_profiles', 'run_clustering'):
|
||||
entity_ds_id = tool_input.get('dataset_id', '')
|
||||
if entity_ds_id and store.get_dataset(entity_ds_id):
|
||||
break
|
||||
entity_ds_id = None
|
||||
|
||||
# Fallback: try common dataset ID patterns
|
||||
if not entity_ds_id:
|
||||
candidates = [
|
||||
ctx.get('dataset_id', ''),
|
||||
f'entity_{pk}',
|
||||
f'entity_{run.display_id}',
|
||||
f'upload_{run.display_id}',
|
||||
]
|
||||
for cid in candidates:
|
||||
if cid and store.get_dataset(cid):
|
||||
entity_ds_id = cid
|
||||
break
|
||||
|
||||
if entity_ds_id:
|
||||
_run_clustering_pipeline(
|
||||
run=run,
|
||||
store=store,
|
||||
entity_ds_id=entity_ds_id,
|
||||
feature_columns=None,
|
||||
algorithm='agglomerative',
|
||||
min_cluster_size=5,
|
||||
run_umap=True,
|
||||
head=ctx.get('head'),
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
f'[LLM auto] No entity dataset found for clustering. '
|
||||
f'Tool calls: {[tc.get("name") for tc in tool_calls]}. '
|
||||
f'Marking run as completed without clustering.'
|
||||
)
|
||||
run.status = 'completed'
|
||||
run.progress_pct = 100
|
||||
run.progress_msg = 'LLM分析完成（无实体数据，跳过聚类）'
|
||||
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
||||
|
||||
# ── Auto-save workflow as loadable plan ──
|
||||
try:
|
||||
tool_calls_data = run.tool_calls_json or []
|
||||
steps = []
|
||||
for tc in tool_calls_data:
|
||||
tc_name = tc.get('name', '')
|
||||
if tc_name == 'thinking':
|
||||
continue
|
||||
steps.append({
|
||||
'tool_name': tc_name,
|
||||
'params': tc.get('input', {}),
|
||||
})
|
||||
if steps:
|
||||
plan_data = {
|
||||
'name': f'_auto_{run.display_id}',
|
||||
'steps': steps,
|
||||
'auto': True,
|
||||
'display_id': run.display_id,
|
||||
'created_at': str(run.created_at) if run.created_at else '',
|
||||
}
|
||||
plan_path = _PLANS_DIR / f'_auto_{run.display_id}.json'
|
||||
plan_path.write_text(
|
||||
json.dumps(plan_data, ensure_ascii=False, indent=2),
|
||||
encoding='utf-8',
|
||||
)
|
||||
_add_to_auto_index(f'_auto_{run.display_id}.json')
|
||||
logger.info(
|
||||
'[LLM auto] Saved auto plan %s (%d steps)',
|
||||
plan_path.name, len(steps),
|
||||
)
|
||||
except Exception as save_err:
|
||||
logger.warning('[LLM auto] Failed to save auto plan: %s', save_err)
|
||||
|
||||
except Exception:
|
||||
tb = traceback.format_exc()
|
||||
logger.error(tb)
|
||||
try:
|
||||
run = AnalysisRun.objects.get(id=pk)
|
||||
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 as e:
|
||||
logger.warning('save failed after pipeline error: %s', e)
|
||||
|
||||
t = threading.Thread(
|
||||
target=_run_pipeline_worker,
|
||||
args=(pk, _analysis_fn),
|
||||
kwargs=dict(dataset_id=dataset_id, head=head, pk=pk),
|
||||
daemon=True,
|
||||
)
|
||||
t.start()
|
||||
return JsonResponse({'status': 'started', 'run_id': run.display_id})
|
||||
@@ -0,0 +1,489 @@
|
||||
"""Cluster views: overview, detail, globe flows, and the clustering pipeline."""
|
||||
import json
|
||||
import traceback
|
||||
import logging
|
||||
|
||||
from django.shortcuts import render, get_object_or_404
|
||||
from django.http import Http404
|
||||
|
||||
from analysis.models import AnalysisRun, ClusterResult, EntityProfile
|
||||
from analysis import nl_describe
|
||||
from analysis.profile_util import profile
|
||||
|
||||
from .helpers import _extract_lat, _extract_lon
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def cluster_overview(request, display_id):
|
||||
"""Overview of all clusters for a run, with UMAP scatter data and geo scatter data."""
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
clusters = run.clusters.exclude(cluster_label=-1).order_by('-size').prefetch_related('features')
|
||||
noise = run.clusters.filter(cluster_label=-1).first()
|
||||
|
||||
# Build UMAP scatter data for Canvas / Three.js 3D
|
||||
entities = run.entities.exclude(embedding_x=None).exclude(embedding_y=None).only(
|
||||
'embedding_x', 'embedding_y', 'embedding_z', 'cluster_label', 'entity_value'
|
||||
)
|
||||
has_z = any(e.embedding_z is not None for e in entities)
|
||||
scatter_data = [
|
||||
{'x': e.embedding_x, 'y': e.embedding_y, 'z': e.embedding_z if has_z else 0,
|
||||
'label': e.cluster_label if e.cluster_label is not None else -1, 'entity': e.entity_value}
|
||||
for e in entities
|
||||
]
|
||||
|
||||
# Noise cluster: features + entity detail
|
||||
noise_features = []
|
||||
noise_summary = ''
|
||||
noise_entity_count = 0
|
||||
if noise:
|
||||
noise_features_qs = noise.features.all().order_by('-distinguishing_score')[:15]
|
||||
noise_features = [
|
||||
{'feature_name': f.feature_name, 'score': f.distinguishing_score, 'mean': f.mean,
|
||||
'std': f.std, 'p25': f.p25, 'p75': f.p75}
|
||||
for f in noise_features_qs
|
||||
]
|
||||
if noise_features:
|
||||
noise_summary = nl_describe.describe_cluster(noise_features)
|
||||
noise_entity_count = noise.size
|
||||
|
||||
# LLM workflow timeline for auto runs
|
||||
llm_timeline_json = ''
|
||||
if run.run_type == 'auto':
|
||||
llm_thinking = run.llm_thinking or ''
|
||||
tool_calls = run.tool_calls_json or []
|
||||
if tool_calls or llm_thinking:
|
||||
llm_timeline_json = json.dumps({
|
||||
'thinking': llm_thinking,
|
||||
'tool_calls': tool_calls,
|
||||
})
|
||||
|
||||
# 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 + NL summaries
|
||||
top_features = {}
|
||||
for c in clusters:
|
||||
features_qs = c.features.all().order_by('-distinguishing_score')[:10]
|
||||
feats = [
|
||||
{'feature_name': f.feature_name, 'score': f.distinguishing_score, 'mean': f.mean,
|
||||
'std': f.std}
|
||||
for f in features_qs
|
||||
]
|
||||
top_features[str(c.cluster_label)] = feats
|
||||
# Attach features + NL summary directly to cluster object for template
|
||||
c._top_features = feats
|
||||
c.nl_summary = nl_describe.describe_cluster(feats) if feats else ''
|
||||
|
||||
return render(request, 'analysis/cluster_overview.html', {
|
||||
'run': run,
|
||||
'clusters': clusters,
|
||||
'noise': noise,
|
||||
'scatter_data_json': json.dumps(scatter_data),
|
||||
'has_z': has_z,
|
||||
'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),
|
||||
'noise_features': noise_features,
|
||||
'noise_summary': noise_summary,
|
||||
'noise_entity_count': noise_entity_count,
|
||||
'llm_timeline_json': llm_timeline_json,
|
||||
# Globe flow data for the sidebar
|
||||
'globe_flows_json': json.dumps(_get_globe_flows(run)),
|
||||
})
|
||||
|
||||
|
||||
def _get_globe_flows(run, max_rows=500):
|
||||
"""Extract flow data for 3D globe visualization from a run."""
|
||||
from analysis.data_loader import load_csv_directory, load_from_db
|
||||
try:
|
||||
lf = None
|
||||
if run.sqlite_table:
|
||||
lf = load_from_db(run.sqlite_table)
|
||||
if lf is None and run.csv_glob:
|
||||
lf, _, _, _, _ = load_csv_directory(run.csv_glob)
|
||||
if lf is None:
|
||||
return []
|
||||
df = lf.head(max_rows).collect()
|
||||
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)
|
||||
bytes_col = next((c for c in df.columns if c.lower() in ('8ack', '8pak', '8byt', 'bytes_sent', 'bytes')), None)
|
||||
for row in df.iter_rows(named=True):
|
||||
sip = str(row.get(src_ip_col, '')) if src_ip_col else ''
|
||||
dip = str(row.get(dst_ip_col, '')) if dst_ip_col else ''
|
||||
if not sip or not dip:
|
||||
continue
|
||||
sg = geo_lookup(sip)
|
||||
dg = geo_lookup(dip)
|
||||
if not sg or not dg:
|
||||
continue
|
||||
tls_val = str(row.get(tls_col, '') or '') if tls_col else ''
|
||||
tls_ver = TLS_HEX_MAP.get(tls_val.replace(' ', ''), tls_val)
|
||||
flows.append({
|
||||
'slat': sg['lat'], 'slon': sg['lon'],
|
||||
'dlat': dg['lat'], 'dlon': dg['lon'],
|
||||
'tls': tls_ver,
|
||||
'bytes': float(row.get(bytes_col, 0) or 0) if bytes_col else 0,
|
||||
})
|
||||
return flows
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def cluster_detail(request, display_id, cluster_label):
|
||||
"""Detailed view of a single cluster with entity list, SVD features, and NL summary."""
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
try:
|
||||
cluster_label = int(cluster_label)
|
||||
except (ValueError, TypeError):
|
||||
raise Http404("Invalid cluster label")
|
||||
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()[:100]
|
||||
nl_summary = ''
|
||||
feats_list = [{'feature_name': f.feature_name, 'score': f.distinguishing_score,
|
||||
'mean': f.mean, 'std': f.std} for f in features[:10]]
|
||||
if feats_list:
|
||||
nl_summary = nl_describe.describe_cluster(feats_list)
|
||||
|
||||
# Build entity data with feature_json values
|
||||
entity_data = []
|
||||
for e in entities:
|
||||
entry = {'id': e.id, 'value': e.entity_value,
|
||||
'embedding_x': e.embedding_x, 'embedding_y': e.embedding_y}
|
||||
if e.feature_json:
|
||||
entry['features'] = {k: round(float(v), 4) if isinstance(v, (int, float)) else str(v)
|
||||
for k, v in e.feature_json.items()}
|
||||
entity_data.append(entry)
|
||||
|
||||
return render(request, 'analysis/cluster_detail.html', {
|
||||
'run': run,
|
||||
'cluster': cluster,
|
||||
'features': features,
|
||||
'entities': entities,
|
||||
'entity_data_json': json.dumps(entity_data),
|
||||
'nl_summary': nl_summary,
|
||||
})
|
||||
|
||||
|
||||
@profile
|
||||
def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorithm,
|
||||
min_cluster_size, run_umap=True, head=None):
|
||||
"""Unified clustering pipeline: clustering → feature extraction → UMAP → 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_umap: bool, whether to compute and save UMAP-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
|
||||
|
||||
try:
|
||||
from analysis.tool_registry import _handle_run_clustering, _handle_extract_features
|
||||
import polars as pl
|
||||
|
||||
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']
|
||||
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 — use actual LazyFrame dtypes (not stored schema dict
|
||||
# which may be stale after Utf8 coercion in _background_process).
|
||||
numeric_types = (pl.Int8, pl.Int16, pl.Int32, pl.Int64,
|
||||
pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64,
|
||||
pl.Float32, pl.Float64)
|
||||
|
||||
lf = entry['lazyframe']
|
||||
live_schema = lf.collect_schema()
|
||||
live_names = live_schema.names()
|
||||
live_dtypes = list(live_schema.dtypes())
|
||||
|
||||
if feature_columns:
|
||||
feature_cols = [
|
||||
c for c in feature_columns
|
||||
if c in live_names and live_dtypes[live_names.index(c)] in numeric_types
|
||||
]
|
||||
else:
|
||||
feature_cols = [
|
||||
live_names[i] for i, dt in enumerate(live_dtypes)
|
||||
if dt in numeric_types and not live_names[i].startswith('_')
|
||||
][:10]
|
||||
|
||||
# ── Step 1: Clustering ────────────────────────────────────────────────
|
||||
# If no numeric columns detected, pass empty list to let
|
||||
# _handle_run_clustering apply its Utf8→Float64 coercion fallback.
|
||||
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 2b: Cluster SVD feature extraction ──────────────────────
|
||||
run.progress_msg = '正在进行聚类SVD特征提取...'
|
||||
run.save(update_fields=['progress_msg'])
|
||||
try:
|
||||
from analysis.distance import cluster_svd_extract
|
||||
lf = entry['lazyframe']
|
||||
cluster_entry = store.get_cluster_result(cluster_id)
|
||||
labels_list = cluster_entry.get('labels', []) if cluster_entry else []
|
||||
|
||||
if labels_list and feature_cols:
|
||||
svd_results = cluster_svd_extract(lf, labels_list, feature_cols)
|
||||
|
||||
# Persist SVD features to ClusterFeature (method='cluster_svd')
|
||||
from analysis.models import ClusterFeature as CF
|
||||
from analysis.models import ClusterResult as CR
|
||||
for label, svd_info in svd_results.items():
|
||||
cr = CR.objects.filter(run=run, cluster_label=label).first()
|
||||
if cr is None:
|
||||
continue
|
||||
# Delete old zscore entries for this cluster, replace with SVD
|
||||
CF.objects.filter(cluster=cr, distinguishing_method='zscore').delete()
|
||||
for feat_name in svd_info.get('features', [])[:10]:
|
||||
idx = svd_info['features'].index(feat_name) if feat_name in svd_info['features'] else -1
|
||||
strength = (svd_info.get('feature_strength', [])[idx]
|
||||
if idx >= 0 and idx < len(svd_info.get('feature_strength', []))
|
||||
else None)
|
||||
CF.objects.get_or_create(
|
||||
cluster=cr,
|
||||
feature_name=feat_name,
|
||||
defaults={
|
||||
'mean': None, 'std': None, 'median': None,
|
||||
'p25': None, 'p75': None, 'missing_rate': None,
|
||||
'distinguishing_score': float(strength) if strength is not None else None,
|
||||
'distinguishing_method': 'cluster_svd',
|
||||
},
|
||||
)
|
||||
except Exception as svd_err:
|
||||
logger.warning(f'Cluster SVD extraction skipped: {svd_err}')
|
||||
|
||||
# ── Step 3: UMAP-3D embedding (2D fallback) ───────────────────────
|
||||
if run_umap:
|
||||
try:
|
||||
lf = entry['lazyframe']
|
||||
# UMAP: sample max 10K for training, batch-transform rest in 1K batches
|
||||
MAX_UMAP_TRAIN = 10000
|
||||
BATCH_SIZE = 1000
|
||||
try:
|
||||
n_total = lf.select(pl.len()).collect(streaming=True).item()
|
||||
except Exception:
|
||||
n_total = 0
|
||||
from sklearn.preprocessing import StandardScaler
|
||||
import umap
|
||||
import numpy as np
|
||||
|
||||
live_schema = lf.collect_schema()
|
||||
num_cols = [name for name, dt in zip(live_schema.names(), live_schema.dtypes())
|
||||
if dt in (pl.Int8, pl.Int16, pl.Int32, pl.Int64,
|
||||
pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64,
|
||||
pl.Float32, pl.Float64) and not name.startswith('_')]
|
||||
|
||||
if len(num_cols) >= 2:
|
||||
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_umap = lf.collect(streaming=True)
|
||||
else:
|
||||
df_umap = lf.collect(streaming=True)
|
||||
|
||||
umap_components = 3 # try 3D first
|
||||
coords = None
|
||||
reducer = None
|
||||
|
||||
if len(df_umap) > 2:
|
||||
for attempt_n in (3, 2):
|
||||
try:
|
||||
if n_total > MAX_UMAP_TRAIN:
|
||||
# Train UMAP on 10K sample, batch-transform full dataset
|
||||
mat_sample = df_sample.select(num_cols).to_numpy()
|
||||
scaler = StandardScaler().fit(mat_sample)
|
||||
mat_sample_scaled = np.nan_to_num(
|
||||
scaler.transform(mat_sample), nan=0.0)
|
||||
reducer = umap.UMAP(n_components=attempt_n,
|
||||
random_state=42,
|
||||
n_neighbors=15, min_dist=0.1,
|
||||
metric='euclidean')
|
||||
reducer.fit(mat_sample_scaled)
|
||||
# Batch transform in 1K-row batches
|
||||
coords_list = []
|
||||
for start in range(0, len(df_umap), BATCH_SIZE):
|
||||
end = min(start + BATCH_SIZE, len(df_umap))
|
||||
mat_batch = scaler.transform(
|
||||
df_umap[start:end].select(num_cols).to_numpy())
|
||||
mat_batch = np.nan_to_num(mat_batch, nan=0.0)
|
||||
coords_list.append(reducer.transform(mat_batch))
|
||||
coords = np.vstack(coords_list)
|
||||
else:
|
||||
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,
|
||||
n_neighbors=15, min_dist=0.1,
|
||||
metric='euclidean')
|
||||
coords = reducer.fit_transform(mat)
|
||||
umap_components = attempt_n
|
||||
break
|
||||
except Exception as e_umap:
|
||||
if attempt_n == 2:
|
||||
raise
|
||||
logger.warning(f'UMAP 3D failed ({e_umap}), falling back to 2D')
|
||||
|
||||
if coords is None:
|
||||
raise Exception('UMAP produced no output')
|
||||
|
||||
non_num = [c for c in df_umap.columns
|
||||
if c not in num_cols and not c.startswith('_')]
|
||||
ent_col = non_num[0] if non_num else df_umap.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_umap.iter_rows(named=True)):
|
||||
ev = str(row.get(ent_col, '')) or 'entity'
|
||||
ev = f'{ev}_{i}'
|
||||
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()
|
||||
has_z = umap_components >= 3 and coords.shape[1] >= 3
|
||||
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,
|
||||
embedding_z=float(coords[i, 2]) if has_z
|
||||
and i < len(coords) else None,
|
||||
))
|
||||
if profiles:
|
||||
EP.objects.bulk_create(profiles, ignore_conflicts=True, batch_size=1000)
|
||||
run.entity_count = len(profiles)
|
||||
run.save(update_fields=['entity_count'])
|
||||
except Exception as umap_err:
|
||||
logger.warning(f'UMAP embedding skipped: {umap_err}')
|
||||
|
||||
# Fallback: if entity_count not set (UMAP skipped), derive from cluster sizes
|
||||
if run.entity_count is None:
|
||||
from django.db.models import Sum
|
||||
total = run.clusters.aggregate(total=Sum('size'))['total'] or 0
|
||||
run.entity_count = total
|
||||
run.save(update_fields=['entity_count'])
|
||||
|
||||
# ── Done ──
|
||||
run.status = 'completed'
|
||||
run.progress_pct = 100
|
||||
run.progress_msg = '分析完成'
|
||||
run.save(update_fields=['status', 'progress_pct', 'progress_msg', 'entity_count'])
|
||||
|
||||
except Exception:
|
||||
tb = traceback.format_exc()
|
||||
logger.error(tb)
|
||||
try:
|
||||
run.status = 'failed'
|
||||
run.error_message = tb
|
||||
run.progress_msg = f'失败: {tb}'
|
||||
run.run_log += f'\n[ERROR] {tb}'
|
||||
run.save(update_fields=['status', 'error_message', 'progress_msg', 'run_log'])
|
||||
except Exception as e:
|
||||
logger.warning('save failed after pipeline error: %s', e)
|
||||
@@ -0,0 +1,130 @@
|
||||
"""Configuration views: config editor and LLM connectivity test."""
|
||||
import json
|
||||
import time
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import logging
|
||||
|
||||
from django.shortcuts import render
|
||||
from django.http import JsonResponse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from config import get_config, save_config, Config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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})
|
||||
@@ -0,0 +1,63 @@
|
||||
"""Dashboard views: home, run list, run detail, run status."""
|
||||
from django.shortcuts import render, get_object_or_404
|
||||
from django.http import JsonResponse
|
||||
|
||||
from analysis.models import AnalysisRun
|
||||
|
||||
|
||||
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 and optional ?type= filter."""
|
||||
run_type_filter = request.GET.get('type')
|
||||
page = int(request.GET.get('page', 1))
|
||||
per_page = 20
|
||||
|
||||
qs = AnalysisRun.objects.all()
|
||||
if run_type_filter in dict(AnalysisRun.RUN_TYPE_CHOICES):
|
||||
qs = qs.filter(run_type=run_type_filter)
|
||||
|
||||
total = qs.count()
|
||||
runs = qs[(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,
|
||||
'current_type': run_type_filter or '',
|
||||
})
|
||||
|
||||
|
||||
def run_detail(request, display_id):
|
||||
"""Details for a single analysis run."""
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
clusters = run.clusters.all().order_by('-size')
|
||||
return render(request, 'analysis/run_detail.html', {
|
||||
'run': run,
|
||||
'clusters': clusters,
|
||||
})
|
||||
|
||||
|
||||
def run_status_api(request, display_id):
|
||||
"""Return current run status as JSON."""
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
return JsonResponse({
|
||||
'id': run.display_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
|
||||
'llm_thinking': run.llm_thinking[-5000:], # last 5K chars
|
||||
'tool_calls': run.tool_calls_json,
|
||||
})
|
||||
|
||||
|
||||
def start_analysis(request):
|
||||
"""Placeholder: API endpoint to trigger analysis via MCP."""
|
||||
return JsonResponse({'status': 'not_implemented', 'message': 'Use MCP tools to trigger analysis'})
|
||||
@@ -0,0 +1,33 @@
|
||||
"""Entity profile view."""
|
||||
import json
|
||||
|
||||
from django.shortcuts import render, get_object_or_404
|
||||
|
||||
from analysis.models import EntityProfile
|
||||
|
||||
|
||||
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,
|
||||
})
|
||||
@@ -0,0 +1,210 @@
|
||||
"""3D Globe view and flow extraction."""
|
||||
import json
|
||||
import logging
|
||||
|
||||
from django.shortcuts import render
|
||||
|
||||
from analysis.models import AnalysisRun
|
||||
from .helpers import _extract_lat, _extract_lon
|
||||
|
||||
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.
|
||||
|
||||
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)
|
||||
time_col = next((c for c in df.columns if c.lower() in (
|
||||
'timestamp', 'time', '_time', 'ts', 'datetime', 'epoch',
|
||||
'created_at', 'created', 'event_time')), 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)
|
||||
|
||||
# Extract timestamp for animation ordering
|
||||
t_raw = row.get(time_col) if time_col else None
|
||||
t_val = None
|
||||
if t_raw is not None:
|
||||
if isinstance(t_raw, (int, float)):
|
||||
t_val = float(t_raw)
|
||||
elif hasattr(t_raw, 'timestamp'):
|
||||
t_val = t_raw.timestamp()
|
||||
else:
|
||||
try:
|
||||
t_val = float(t_raw)
|
||||
except (ValueError, TypeError):
|
||||
pass
|
||||
|
||||
flows.append({
|
||||
'slat': slat, 'slon': slon,
|
||||
'dlat': dlat, 'dlon': dlon,
|
||||
'tls': tls_label,
|
||||
'bytes': bs,
|
||||
'time': t_val,
|
||||
})
|
||||
|
||||
# Normalize timestamps to 0-1 and sort flows by time
|
||||
timed = [f for f in flows if f['time'] is not None]
|
||||
if len(timed) > 1:
|
||||
timed.sort(key=lambda f: f['time'])
|
||||
t_min, t_max = timed[0]['time'], timed[-1]['time']
|
||||
t_range = max(t_max - t_min, 1)
|
||||
for f in timed:
|
||||
f['time'] = (f['time'] - t_min) / t_range
|
||||
# Untimed flows get evenly spaced after timed ones
|
||||
untimed = [f for f in flows if f['time'] is None]
|
||||
for i, f in enumerate(untimed):
|
||||
f['time'] = min(1.0, (len(timed) + i) / max(len(flows) - 1, 1))
|
||||
flows = timed + untimed
|
||||
else:
|
||||
for i, f in enumerate(flows):
|
||||
f['time'] = i / max(len(flows) - 1, 1)
|
||||
|
||||
return flows
|
||||
|
||||
|
||||
def globe_view(request):
|
||||
from analysis.geoip import lookup as geo_lookup
|
||||
from analysis.data_loader import load_csv_directory, load_from_db
|
||||
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('display_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:
|
||||
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('display_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(display_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 = None
|
||||
if run.sqlite_table:
|
||||
lf = load_from_db(run.sqlite_table)
|
||||
if lf is None:
|
||||
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.display_id, run.csv_glob, exc)
|
||||
continue
|
||||
|
||||
if all_selected_failed or not flows:
|
||||
# Fallback: latest completed run with valid data (SQLite or CSV)
|
||||
for latest in AnalysisRun.objects.filter(status='completed').order_by('-id'):
|
||||
if not latest.csv_glob and not latest.sqlite_table:
|
||||
continue
|
||||
try:
|
||||
lf = None
|
||||
if latest.sqlite_table:
|
||||
lf = load_from_db(latest.sqlite_table)
|
||||
if lf is None:
|
||||
if not latest.csv_glob:
|
||||
continue
|
||||
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.display_id for r in selected_runs],
|
||||
})
|
||||
resp['Cache-Control'] = 'no-cache, no-store, must-revalidate'
|
||||
resp['Pragma'] = 'no-cache'
|
||||
resp['Expires'] = '0'
|
||||
return resp
|
||||
@@ -0,0 +1,71 @@
|
||||
"""Geo extraction helpers and plan utilities."""
|
||||
from typing import Optional
|
||||
from pathlib import Path
|
||||
import json as _json
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
# ── Plan index utilities (shared by tools.py and auto.py) ─────────────
|
||||
|
||||
_PLANS_DIR = Path(__file__).resolve().parent.parent.parent / '.omo' / 'plans'
|
||||
_PLANS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
_AUTO_INDEX_PATH = _PLANS_DIR / '_auto_index.json'
|
||||
_MAX_AUTO_PLANS = 10
|
||||
|
||||
|
||||
def _read_auto_index():
|
||||
"""Read the auto index file, returning list of {file, pinned}."""
|
||||
if not _AUTO_INDEX_PATH.exists():
|
||||
return []
|
||||
try:
|
||||
data = _json.loads(_AUTO_INDEX_PATH.read_text(encoding='utf-8'))
|
||||
if not isinstance(data, list):
|
||||
return []
|
||||
return data
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _write_auto_index(index):
|
||||
"""Write the auto index list."""
|
||||
_AUTO_INDEX_PATH.write_text(
|
||||
_json.dumps(index, ensure_ascii=False, indent=2),
|
||||
encoding='utf-8',
|
||||
)
|
||||
|
||||
|
||||
def _add_to_auto_index(filename):
|
||||
"""Prepend filename to auto index, evict unpinned entries when >10."""
|
||||
index = _read_auto_index()
|
||||
# Remove if already present
|
||||
index = [entry for entry in index if entry.get('file') != filename]
|
||||
# Prepend
|
||||
index.insert(0, {'file': filename, 'pinned': False})
|
||||
# Evict unpinned entries beyond max
|
||||
pinned_entries = [e for e in index if e.get('pinned')]
|
||||
unpinned_entries = [e for e in index if not e.get('pinned')]
|
||||
# Keep all pinned + fill remaining slots with unpinned
|
||||
result = pinned_entries + unpinned_entries[:_MAX_AUTO_PLANS - len(pinned_entries)]
|
||||
_write_auto_index(result)
|
||||
@@ -0,0 +1,36 @@
|
||||
"""Log viewer view."""
|
||||
from django.shortcuts import render
|
||||
from django.http import HttpResponse
|
||||
|
||||
|
||||
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})
|
||||
@@ -0,0 +1,191 @@
|
||||
"""Manual analysis page: workflow builder and run execution."""
|
||||
import json
|
||||
import threading
|
||||
import traceback
|
||||
import logging
|
||||
|
||||
from django.shortcuts import render, get_object_or_404
|
||||
from django.http import JsonResponse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from analysis.models import AnalysisRun
|
||||
from .pipeline import _run_pipeline_worker
|
||||
from .clustering import _run_clustering_pipeline
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def manual_page(request):
|
||||
"""Workflow builder: select dataset, add MCP tool steps, execute."""
|
||||
from analysis.session_store import SessionStore
|
||||
from analysis.tool_registry import get_tools_meta
|
||||
import json as _json
|
||||
|
||||
# Get all tools metadata
|
||||
tools = get_tools_meta()
|
||||
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)
|
||||
|
||||
meta_list = [{'name': t.name, 'description': t.description, 'inputSchema': t.inputSchema} for t in tools]
|
||||
|
||||
# Get datasets from SessionStore
|
||||
store = SessionStore()
|
||||
session_datasets = store.list_datasets() if hasattr(store, 'list_datasets') else []
|
||||
session_ds_map = {ds['dataset_id']: ds for ds in session_datasets}
|
||||
|
||||
# Build dataset entries from ALL AnalysisRun records (upload/manual/auto, any status)
|
||||
from analysis.models import AnalysisRun
|
||||
all_runs = AnalysisRun.objects.all().order_by('-created_at')[:100]
|
||||
datasets_with_status = []
|
||||
|
||||
for run in all_runs:
|
||||
# Determine how the dataset was keyed in SessionStore
|
||||
if run.run_type == 'upload':
|
||||
ds_id = f'upload_{run.display_id}'
|
||||
else:
|
||||
ds_id = f'run_{run.display_id}'
|
||||
|
||||
session_entry = session_ds_map.get(ds_id)
|
||||
|
||||
if session_entry:
|
||||
# Full data available in SessionStore — merge with run status
|
||||
datasets_with_status.append({
|
||||
**session_entry,
|
||||
'dataset_id': ds_id,
|
||||
'run_status': run.status,
|
||||
'total_flows': run.total_flows,
|
||||
'display_id': run.display_id,
|
||||
'needs_reload': False,
|
||||
})
|
||||
elif run.csv_glob or run.sqlite_table:
|
||||
# Data not in SessionStore but recoverable from disk/DB
|
||||
datasets_with_status.append({
|
||||
'dataset_id': ds_id,
|
||||
'run_status': run.status,
|
||||
'display_id': run.display_id,
|
||||
'total_flows': run.total_flows,
|
||||
'row_count': run.total_flows,
|
||||
'file_count': None,
|
||||
'column_count': 0,
|
||||
'columns': [],
|
||||
'svd_components': 0,
|
||||
'needs_reload': True,
|
||||
'csv_glob': run.csv_glob,
|
||||
'sqlite_table': run.sqlite_table or '',
|
||||
})
|
||||
|
||||
# Map statuses to user-facing labels for the filter tabs
|
||||
# ready → 待分析, completed → 已分析, failed → 错误
|
||||
# Everything else (pending/loading/profiling/aggregating/clustering/extracting) → analyzing → 分析中
|
||||
for ds_info in datasets_with_status:
|
||||
s = ds_info['run_status']
|
||||
if s in ('completed',):
|
||||
ds_info['status_label'] = '已分析'
|
||||
ds_info['badge_class'] = 'completed'
|
||||
elif s in ('failed',):
|
||||
ds_info['status_label'] = '错误'
|
||||
ds_info['badge_class'] = 'failed'
|
||||
elif s in ('pending', 'loading', 'profiling', 'aggregating', 'clustering', 'extracting'):
|
||||
ds_info['status_label'] = '分析中'
|
||||
ds_info['badge_class'] = 'analyzing'
|
||||
else:
|
||||
ds_info['status_label'] = '待分析'
|
||||
ds_info['badge_class'] = 'ready'
|
||||
|
||||
return render(request, 'tianxuan/manual.html', {
|
||||
'core_tools': core_tools,
|
||||
'diag_tools': diag_tools,
|
||||
'analysis_tools': analysis_tools,
|
||||
'other_tools': other_tools,
|
||||
'tools_meta': _json.dumps(meta_list, ensure_ascii=False),
|
||||
'datasets': datasets_with_status,
|
||||
'datasets_with_status_json': _json.dumps(datasets_with_status, ensure_ascii=False),
|
||||
})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def manual_run_analysis(request):
|
||||
"""AJAX POST: run clustering pipeline on the upload dataset."""
|
||||
if request.method != 'POST':
|
||||
return JsonResponse({'error': 'POST required'}, status=405)
|
||||
import json as _json
|
||||
body = _json.loads(request.body)
|
||||
display_id = body.get('run_id')
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
pk = run.id
|
||||
run.run_type = 'manual'
|
||||
run.save(update_fields=['run_type'])
|
||||
feature_columns = body.get('feature_columns')
|
||||
algorithm = body.get('algorithm', 'agglomerative')
|
||||
min_cluster_size = int(body.get('min_cluster_size', 5))
|
||||
head = body.get('head')
|
||||
cluster_mode = body.get('cluster_mode', 'raw')
|
||||
|
||||
def _analysis_fn(run, ctx):
|
||||
from analysis.session_store import SessionStore
|
||||
|
||||
feature_columns = ctx['feature_columns']
|
||||
algorithm = ctx['algorithm']
|
||||
min_cluster_size = ctx['min_cluster_size']
|
||||
head = ctx.get('head')
|
||||
pk = ctx['pk']
|
||||
cluster_mode = ctx.get('cluster_mode', 'raw')
|
||||
|
||||
store = SessionStore()
|
||||
try:
|
||||
# Use display_id for dataset key (consistent with _background_process)
|
||||
display_id = ctx.get('display_id', pk)
|
||||
upload_ds_id = f'upload_{display_id}'
|
||||
entry = store.get_dataset(upload_ds_id)
|
||||
if entry is None:
|
||||
# Fallback: try PK-based key (old pipeline)
|
||||
entry = store.get_dataset(f'upload_{pk}')
|
||||
if entry is None:
|
||||
# Fallback: try entity dataset (from older pipeline)
|
||||
entry = store.get_dataset(f'entity_{pk}')
|
||||
if entry is None:
|
||||
run.error_message = '请先上传并等待预处理完成'
|
||||
run.status = 'failed'
|
||||
run.save(update_fields=['status', 'error_message'])
|
||||
return
|
||||
|
||||
ds_id = upload_ds_id if store.get_dataset(upload_ds_id) else f'upload_{pk}'
|
||||
|
||||
# Use the unified clustering pipeline (clustering → extraction → UMAP)
|
||||
_run_clustering_pipeline(
|
||||
run=run, store=store, entity_ds_id=ds_id,
|
||||
feature_columns=feature_columns,
|
||||
algorithm=algorithm, min_cluster_size=min_cluster_size,
|
||||
run_umap=True, head=head,
|
||||
)
|
||||
except Exception:
|
||||
tb = traceback.format_exc()
|
||||
logger.error(tb)
|
||||
run.status = 'failed'
|
||||
run.error_message = tb
|
||||
run.progress_msg = f'失败: {tb}'
|
||||
run.run_log += f'\n[ERROR] {tb}'
|
||||
run.save(update_fields=['status', 'error_message', 'progress_msg', 'run_log'])
|
||||
|
||||
t = threading.Thread(
|
||||
target=_run_pipeline_worker,
|
||||
args=(pk, _analysis_fn),
|
||||
kwargs=dict(feature_columns=feature_columns, algorithm=algorithm,
|
||||
min_cluster_size=min_cluster_size, head=head, pk=pk,
|
||||
display_id=run.display_id, cluster_mode=cluster_mode),
|
||||
daemon=True,
|
||||
)
|
||||
t.start()
|
||||
return JsonResponse({'status': 'started', 'redirect': f'/runs/{run.display_id}/'})
|
||||
@@ -0,0 +1,298 @@
|
||||
"""Background pipeline: CSV loading, SVD, and pipeline worker."""
|
||||
import os
|
||||
import threading
|
||||
import traceback
|
||||
import logging
|
||||
|
||||
from analysis.appdata import ensure_appdata_dir
|
||||
from analysis.models import AnalysisRun
|
||||
from analysis.profile_util import profile
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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()
|
||||
|
||||
from analysis.db_utils import retry_on_lock
|
||||
from django.db import close_old_connections
|
||||
|
||||
close_old_connections()
|
||||
|
||||
@retry_on_lock()
|
||||
def _get_run():
|
||||
return AnalysisRun.objects.get(id=run_id)
|
||||
|
||||
run = _get_run()
|
||||
try:
|
||||
analysis_fn(run, ctx)
|
||||
except Exception:
|
||||
import traceback as _tb
|
||||
tb = _tb.format_exc()
|
||||
logger.error(tb)
|
||||
run.status = 'failed'
|
||||
run.error_message = tb
|
||||
run.run_log += f'\n[FATAL] {tb}'
|
||||
retry_on_lock()(lambda: run.save(update_fields=['status', 'error_message', 'run_log']))()
|
||||
|
||||
|
||||
@profile
|
||||
def _background_process(run_id, upload_dir):
|
||||
"""Background: stream-load CSVs → save to SQLite → ready for analysis."""
|
||||
import django
|
||||
os.environ['DJANGO_SETTINGS_MODULE'] = 'tianxuan.settings'
|
||||
django.setup()
|
||||
from analysis.models import AnalysisRun
|
||||
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
|
||||
|
||||
import glob
|
||||
import polars as pl
|
||||
csv_paths = sorted(glob.glob(str(upload_dir / '*.csv')))
|
||||
if not csv_paths:
|
||||
msg = '没有找到 CSV 文件'
|
||||
logger.error('[BACKGROUND] %s (run_id=%s)', msg, run_id)
|
||||
run.status = 'failed'
|
||||
run.error_message = msg
|
||||
run.save(update_fields=['status', 'error_message'])
|
||||
return
|
||||
|
||||
# ── Detect lat/lon columns for schema override ──
|
||||
import csv as _csv
|
||||
_LAT_LON_KW = ('latd', 'lond', 'latitude', 'longitude')
|
||||
lat_lon_overrides = {}
|
||||
try:
|
||||
with open(csv_paths[0], 'r', encoding='utf-8') as _f:
|
||||
_reader = _csv.reader(_f)
|
||||
_headers = next(_reader, [])
|
||||
for _col in _headers:
|
||||
_norm = _col.lower().replace('-', '_').replace('.', '_').replace(' ', '_')
|
||||
if any(kw in _norm for kw in _LAT_LON_KW):
|
||||
lat_lon_overrides[_col] = pl.Utf8
|
||||
logger.info('[BACKGROUND] Forcing lat/lon column %r to Utf8 for safe parsing', _col)
|
||||
except Exception:
|
||||
logger.warning('[BACKGROUND] Failed to detect lat/lon columns', exc_info=True)
|
||||
|
||||
total_files = len(csv_paths)
|
||||
logger.info('[BACKGROUND] Found %d CSV files in %s', total_files, upload_dir)
|
||||
# Adaptive batch size: aim for ~20 batches total, cap at 500 files per batch
|
||||
batch_size = max(1, min(500, total_files // 20 or 1))
|
||||
total_batches = (total_files + batch_size - 1) // batch_size
|
||||
|
||||
run.status = 'loading'; run.progress_pct = 10
|
||||
run.progress_msg = f'正在流式加载 {total_files} 个文件（共 {total_batches} 批）...'
|
||||
run.save(update_fields=['status', 'progress_pct', 'progress_msg'])
|
||||
|
||||
from analysis.data_loader import save_to_db, drop_sqlite_table, load_from_db, _dtype_to_sqlite
|
||||
|
||||
table_name = f'_data_{run_id}'
|
||||
|
||||
# ── Phase 1: Pre-scan all files for full union column set ──
|
||||
all_columns = []
|
||||
for _path in csv_paths:
|
||||
with open(_path, 'r', encoding='utf-8') as _f:
|
||||
_reader = _csv.reader(_f)
|
||||
_headers = next(_reader, [])
|
||||
for _h in _headers:
|
||||
if _h not in all_columns:
|
||||
all_columns.append(_h)
|
||||
reference_columns = sorted(all_columns, key=str.lower)
|
||||
schema = {col: 'Utf8' for col in all_columns}
|
||||
|
||||
# ── Phase 2: Create SQLite table once with all columns as TEXT ──
|
||||
import sqlite3 as _sqlite3
|
||||
from django.conf import settings as _dj_settings
|
||||
_db_path = _dj_settings.DATABASES['default']['NAME']
|
||||
_conn = _sqlite3.connect(_db_path)
|
||||
try:
|
||||
_col_defs = [f'"{_col}" TEXT' for _col in reference_columns]
|
||||
_conn.execute(f'DROP TABLE IF EXISTS "{table_name}"')
|
||||
_conn.execute(f'CREATE TABLE "{table_name}" ({", ".join(_col_defs)})')
|
||||
_conn.commit()
|
||||
finally:
|
||||
_conn.close()
|
||||
|
||||
# ── Phase 3: Batched collection → SQLite append ──
|
||||
schema_overrides = {col: pl.Utf8 for col in reference_columns}
|
||||
completed_batches = 0
|
||||
total_rows = 0
|
||||
for i in range(0, total_files, batch_size):
|
||||
batch = csv_paths[i:i+batch_size]
|
||||
# Scan each file INDIVIDUALLY — batch scan fails when files have different schemas
|
||||
dfs = []
|
||||
for fp in batch:
|
||||
lf = pl.scan_csv(fp, infer_schema_length=0,
|
||||
schema_overrides=schema_overrides)
|
||||
df = lf.collect(streaming=True)
|
||||
# Fill missing columns with NULL
|
||||
for col in reference_columns:
|
||||
if col not in df.columns:
|
||||
df = df.with_columns(pl.lit(None).alias(col))
|
||||
# Select all reference_columns in canonical order
|
||||
df = df.select(reference_columns)
|
||||
dfs.append(df)
|
||||
df_batch = pl.concat(dfs, how='diagonal_relaxed')
|
||||
|
||||
# Append to pre-created SQLite table
|
||||
save_to_db(run.id, df_batch.lazy(), mode='append')
|
||||
|
||||
# Delete temp files after successful SQLite write
|
||||
for f in batch:
|
||||
try:
|
||||
os.unlink(f)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
completed_batches += 1
|
||||
run.progress_pct = int((completed_batches / total_batches) * 80 + 10)
|
||||
run.progress_msg = f'正在加载数据... ({completed_batches}/{total_batches} 批)'
|
||||
run.save(update_fields=['progress_pct', 'progress_msg'])
|
||||
|
||||
# ── Phase 3: Reload from SQLite with correct types ──
|
||||
lf = load_from_db(table_name, schema_overrides=schema)
|
||||
if lf is None:
|
||||
raise RuntimeError(f'SQLite 表 {table_name} 加载失败')
|
||||
|
||||
# Apply _coerce_to_float on lat/lon columns (restore Float64 from Utf8 storage)
|
||||
from analysis.data_loader import _coerce_to_float as _coerce_to_float_fn
|
||||
for lat_lon_col in lat_lon_overrides:
|
||||
if lat_lon_col in lf.columns:
|
||||
lf = lf.with_columns(
|
||||
pl.col(lat_lon_col).map_batches(_coerce_to_float_fn, return_dtype=pl.Float64).alias(lat_lon_col)
|
||||
)
|
||||
|
||||
# ── Restore numeric types for Utf8 columns coerced during CSV load ──
|
||||
# All columns were forced to Utf8 for safe SQLite storage. Now try to
|
||||
# recover the original numeric types (Int64 / Float64) from the data.
|
||||
numeric_types = (pl.Int8, pl.Int16, pl.Int32, pl.Int64,
|
||||
pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64,
|
||||
pl.Float32, pl.Float64)
|
||||
# Get current schema to know which columns are still Utf8
|
||||
live_schema = lf.collect_schema()
|
||||
live_names = live_schema.names()
|
||||
live_dtypes = list(live_schema.dtypes())
|
||||
utf8_cols = [
|
||||
live_names[i] for i, dt in enumerate(live_dtypes)
|
||||
if dt == pl.Utf8 and not live_names[i].startswith('_')
|
||||
]
|
||||
if utf8_cols:
|
||||
# Infer types from a small sample to decide Float64 vs Int64
|
||||
sample_df = lf.select(utf8_cols).head(1000).collect(streaming=True)
|
||||
type_map: dict[str, pl.DataType] = {}
|
||||
for col in utf8_cols:
|
||||
try:
|
||||
# Try Float64 first (most general numeric type)
|
||||
casted = sample_df[col].cast(pl.Float64, strict=False)
|
||||
non_null = casted.is_not_null().sum()
|
||||
if non_null > len(casted) * 0.5:
|
||||
# Check if all non-null values are integers → use Int64
|
||||
all_int = True
|
||||
for v in casted.head(100).to_list():
|
||||
if v is not None and v != int(v):
|
||||
all_int = False
|
||||
break
|
||||
type_map[col] = pl.Int64 if all_int else pl.Float64
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if type_map:
|
||||
# Update schema dict to reflect restored types
|
||||
for col, dtype in type_map.items():
|
||||
schema[col] = str(dtype)
|
||||
# Apply casts to the LazyFrame
|
||||
casts = [pl.col(col).cast(dtype, strict=False) for col, dtype in type_map.items()]
|
||||
lf = lf.with_columns(casts)
|
||||
logger.info('[BACKGROUND] Restored numeric types for %d columns: %s',
|
||||
len(type_map), list(type_map.keys())[:10])
|
||||
|
||||
# ── SVD dimensionality reduction ──
|
||||
svd_components = 0
|
||||
# Only operate on columns with known numeric types in the schema
|
||||
numeric_dtypes = {'Int64', 'Float64', 'Int32', 'Float32',
|
||||
'Int8', 'Int16', 'UInt8', 'UInt16', 'UInt32', 'UInt64'}
|
||||
numeric_cols = [c for c, dt in schema.items() if dt in numeric_dtypes]
|
||||
n_numeric = len(numeric_cols)
|
||||
|
||||
if n_numeric >= 2:
|
||||
n_components = min(50, n_numeric - 1)
|
||||
try:
|
||||
from sklearn.decomposition import TruncatedSVD # fmt: skip
|
||||
import numpy as np # fmt: skip
|
||||
|
||||
# Collect numeric data for SVD (full materialise → transform → add back)
|
||||
df_full = lf.select(numeric_cols).collect(streaming=True)
|
||||
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)
|
||||
X_svd = svd.fit_transform(X)
|
||||
svd_components = n_components
|
||||
|
||||
# Build a new LazyFrame that includes _svd_* columns
|
||||
# Strategy: collect full LazyFrame once, add SVD columns, re-wrap as lazy
|
||||
df_all = lf.collect(streaming=True)
|
||||
for i in range(n_components):
|
||||
col_name = f'_svd_{i}'
|
||||
schema[col_name] = 'Float64'
|
||||
df_all = df_all.with_columns(
|
||||
pl.Series(col_name, X_svd[:, i])
|
||||
)
|
||||
lf = df_all.lazy()
|
||||
|
||||
logger.info('[BACKGROUND] SVD: %d components from %d numeric columns '
|
||||
'(explained variance ratio sum=%.4f)',
|
||||
svd_components, n_numeric,
|
||||
float(svd.explained_variance_ratio_.sum()))
|
||||
except Exception as e:
|
||||
logger.warning('[BACKGROUND] SVD failed (non-fatal): %s', e)
|
||||
elif n_numeric == 1:
|
||||
logger.info('[BACKGROUND] SVD skipped: only 1 numeric column (%s)',
|
||||
numeric_cols[0])
|
||||
|
||||
# Get row count
|
||||
df_count = lf.select(pl.len()).collect(streaming=True)
|
||||
total_rows = df_count[0, 0] if df_count.height > 0 else 0
|
||||
|
||||
run.sqlite_table = table_name
|
||||
run.total_flows = total_rows
|
||||
run.save(update_fields=['sqlite_table', 'total_flows'])
|
||||
|
||||
ds_id = f'upload_{run.display_id}'
|
||||
store.store_dataset(ds_id, lf, schema=schema, metadata={
|
||||
'row_count': total_rows, 'file_count': total_files, 'upload_dir': str(upload_dir),
|
||||
'csv_glob': str(upload_dir / '*.csv'),
|
||||
'sqlite_table': table_name,
|
||||
'svd_components': svd_components,
|
||||
})
|
||||
|
||||
# ── Phase 3: Data ready for filtering/clustering via MCP tools ──
|
||||
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)
|
||||
# Drop partial SQLite table on failure (idempotent)
|
||||
try:
|
||||
from analysis.data_loader import drop_sqlite_table # fmt: skip
|
||||
drop_sqlite_table(f'_data_{run_id}')
|
||||
except Exception:
|
||||
pass # Expected: drop_sqlite_table may fail if table already gone
|
||||
run.status = 'failed'
|
||||
run.error_message = tb
|
||||
run.run_log += f'\n[ERROR] {tb}'
|
||||
run.save(update_fields=['status', 'error_message', 'run_log'])
|
||||
@@ -0,0 +1,377 @@
|
||||
"""MCP Tool Lab, plan management, and utility views."""
|
||||
import json
|
||||
import asyncio
|
||||
import threading
|
||||
import traceback
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from django.shortcuts import render, get_object_or_404, redirect
|
||||
from django.http import JsonResponse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from analysis.appdata import ensure_appdata_dir
|
||||
from analysis.models import AnalysisRun
|
||||
from .pipeline import _background_process
|
||||
from .helpers import (
|
||||
_PLANS_DIR, _AUTO_INDEX_PATH, _MAX_AUTO_PLANS,
|
||||
_read_auto_index, _write_auto_index, _add_to_auto_index,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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)
|
||||
|
||||
# ── Backend validation: check required params + dataset existence ──
|
||||
from analysis.tool_registry import handle_call, get_tools_meta
|
||||
from analysis.session_store import SessionStore
|
||||
tools_meta = get_tools_meta()
|
||||
tool_schema = None
|
||||
for t in tools_meta:
|
||||
if t.name == tool_name:
|
||||
tool_schema = t.inputSchema
|
||||
break
|
||||
if tool_schema:
|
||||
required_fields = tool_schema.get('required', [])
|
||||
for field in required_fields:
|
||||
val = params.get(field)
|
||||
if val is None or (isinstance(val, str) and val.strip() == ''):
|
||||
return JsonResponse({'error': f'Missing required parameter: {field}'}, status=400)
|
||||
# If field is a dataset_id, verify it exists in the store
|
||||
if 'dataset' in field.lower():
|
||||
store = SessionStore()
|
||||
if not store.get_dataset(str(val)):
|
||||
return JsonResponse({'error': f'Dataset not found: {val}'}, status=400)
|
||||
|
||||
result = asyncio.run(handle_call(tool_name, params))
|
||||
return JsonResponse({'status': 'ok', 'result': result})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def reload_run_data(request):
|
||||
"""Reload a run's data from csv_glob or sqlite_table into SessionStore."""
|
||||
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)
|
||||
|
||||
display_id = body.get('display_id')
|
||||
if not display_id:
|
||||
return JsonResponse({'error': 'Missing display_id'}, status=400)
|
||||
|
||||
from analysis.models import AnalysisRun
|
||||
from analysis.data_loader import load_csv_directory, load_from_db
|
||||
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
|
||||
# Determine dataset_id (same keying as manual_page)
|
||||
ds_id = f'upload_{display_id}' if run.run_type == 'upload' else f'run_{display_id}'
|
||||
|
||||
from analysis.session_store import SessionStore
|
||||
store = SessionStore()
|
||||
# Check if already loaded
|
||||
existing = store.get_dataset(ds_id)
|
||||
if existing is not None:
|
||||
return JsonResponse({'status': 'ok', 'dataset_id': ds_id, 'reloaded': False})
|
||||
|
||||
# Try loading from sqlite_table first, then csv_glob
|
||||
lf = None
|
||||
schema = {}
|
||||
row_count = 0
|
||||
file_count = 0
|
||||
|
||||
if run.sqlite_table:
|
||||
try:
|
||||
lf = load_from_db(run.sqlite_table)
|
||||
if lf is not None:
|
||||
import polars as pl
|
||||
row_count = lf.select(pl.len()).collect(streaming=True)[0, 0]
|
||||
except Exception as exc:
|
||||
logging.getLogger(__name__).warning(f'Failed to load from sqlite_table: {exc}')
|
||||
|
||||
if lf is None and run.csv_glob:
|
||||
try:
|
||||
from pathlib import Path
|
||||
csv_path = Path(run.csv_glob)
|
||||
if csv_path.is_file() or csv_path.is_dir() or csv_path.parent.is_dir():
|
||||
lf, schema, row_count, file_count, _ = load_csv_directory(run.csv_glob)
|
||||
except Exception as exc:
|
||||
logging.getLogger(__name__).warning(f'Failed to load from csv_glob: {exc}')
|
||||
|
||||
if lf is None:
|
||||
return JsonResponse({'error': 'No data source available (upload dir or sqlite table gone)'}, status=404)
|
||||
|
||||
store.store_dataset(ds_id, lf, schema=schema, metadata={
|
||||
'row_count': row_count,
|
||||
'file_count': file_count,
|
||||
'csv_glob': run.csv_glob or '',
|
||||
'sqlite_table': run.sqlite_table or '',
|
||||
})
|
||||
|
||||
return JsonResponse({
|
||||
'status': 'ok',
|
||||
'dataset_id': ds_id,
|
||||
'reloaded': True,
|
||||
'row_count': row_count,
|
||||
'columns': list(schema.keys()),
|
||||
})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def tool_plan(request):
|
||||
"""Save, list, load, delete, or pin/unpin an analysis plan."""
|
||||
import json as _j
|
||||
if request.method == 'GET':
|
||||
name = request.GET.get('name')
|
||||
if name:
|
||||
# Auto plans stored with _auto_ prefix
|
||||
if name.startswith('_auto_'):
|
||||
path = _PLANS_DIR / f'{name}.json'
|
||||
else:
|
||||
path = _PLANS_DIR / f'{name}.json'
|
||||
if not path.exists():
|
||||
return JsonResponse({'error': f'Plan "{name}" not found'}, status=404)
|
||||
try:
|
||||
data = _j.loads(path.read_text(encoding='utf-8'))
|
||||
return JsonResponse(data)
|
||||
except Exception as e:
|
||||
return JsonResponse({'error': str(e)}, status=500)
|
||||
|
||||
# List all plans (regular + auto)
|
||||
auto_index = _read_auto_index()
|
||||
auto_index_map = {e.get('file', ''): e.get('pinned', False) for e in auto_index}
|
||||
auto_names = set()
|
||||
plans = []
|
||||
for p in sorted(_PLANS_DIR.glob('*.json')):
|
||||
# Skip index file and auto plans (handled separately)
|
||||
if p.name == '_auto_index.json':
|
||||
continue
|
||||
if p.name.startswith('_auto_'):
|
||||
auto_names.add(p.name)
|
||||
continue
|
||||
try:
|
||||
content = _j.loads(p.read_text(encoding='utf-8'))
|
||||
steps = len(content.get('steps', []))
|
||||
plans.append({'name': p.stem, 'steps': steps})
|
||||
except Exception:
|
||||
plans.append({'name': p.stem, 'steps': 0})
|
||||
|
||||
# Add auto plans with markers
|
||||
auto_plans = []
|
||||
for auto_name in sorted(auto_names):
|
||||
try:
|
||||
path = _PLANS_DIR / auto_name
|
||||
content = _j.loads(path.read_text(encoding='utf-8'))
|
||||
steps = len(content.get('steps', []))
|
||||
auto_plans.append({
|
||||
'name': auto_name.replace('.json', ''),
|
||||
'steps': steps,
|
||||
'auto': True,
|
||||
'pinned': auto_index_map.get(auto_name, False),
|
||||
})
|
||||
except Exception:
|
||||
auto_plans.append({
|
||||
'name': auto_name.replace('.json', ''),
|
||||
'steps': 0,
|
||||
'auto': True,
|
||||
'pinned': auto_index_map.get(auto_name, False),
|
||||
})
|
||||
# Interleave: auto plans first, then regular
|
||||
return JsonResponse({'plans': auto_plans + plans})
|
||||
|
||||
elif request.method == 'POST':
|
||||
try:
|
||||
body = _j.loads(request.body)
|
||||
# Pin/unpin toggle for auto plans
|
||||
if 'pin' in body and body.get('name', '').startswith('_auto_'):
|
||||
filename = body['name'].replace('.json', '') + '.json' \
|
||||
if not body['name'].endswith('.json') else body['name']
|
||||
index = _read_auto_index()
|
||||
new_pinned = bool(body['pin'])
|
||||
found = False
|
||||
for entry in index:
|
||||
if entry.get('file') == filename:
|
||||
entry['pinned'] = new_pinned
|
||||
found = True
|
||||
break
|
||||
if not found and new_pinned:
|
||||
index.append({'file': filename, 'pinned': True})
|
||||
_write_auto_index(index)
|
||||
return JsonResponse({
|
||||
'status': 'pinned' if new_pinned else 'unpinned',
|
||||
'name': body['name'],
|
||||
'pinned': new_pinned,
|
||||
})
|
||||
|
||||
# Regular save
|
||||
name = body.get('name', '').strip()
|
||||
steps = body.get('steps', [])
|
||||
if not name:
|
||||
return JsonResponse({'error': 'Missing plan name'}, status=400)
|
||||
path = _PLANS_DIR / f'{name}.json'
|
||||
path.write_text(_j.dumps({'name': name, 'steps': steps}, ensure_ascii=False, indent=2), encoding='utf-8')
|
||||
return JsonResponse({'status': 'saved', 'name': name, 'steps': len(steps)})
|
||||
except Exception as e:
|
||||
return JsonResponse({'error': str(e)}, status=500)
|
||||
|
||||
elif request.method == 'DELETE':
|
||||
name = request.GET.get('name', '').strip()
|
||||
if not name:
|
||||
return JsonResponse({'error': 'Missing plan name'}, status=400)
|
||||
path = _PLANS_DIR / f'{name}.json'
|
||||
if path.exists():
|
||||
path.unlink()
|
||||
# Also remove from auto index if applicable
|
||||
if name.startswith('_auto_'):
|
||||
filename = name + '.json' if not name.endswith('.json') else name
|
||||
index = _read_auto_index()
|
||||
index = [e for e in index if e.get('file') != filename]
|
||||
_write_auto_index(index)
|
||||
return JsonResponse({'status': 'deleted', 'name': name})
|
||||
|
||||
return JsonResponse({'error': 'Method not allowed'}, status=405)
|
||||
|
||||
|
||||
# ── Filter endpoint for manual analysis page ──────────────────────────
|
||||
|
||||
@csrf_exempt
|
||||
def apply_filter(request):
|
||||
"""POST {dataset_id, filters, logic} → apply filter_data tool → return filtered_X id."""
|
||||
if request.method != 'POST':
|
||||
return JsonResponse({'error': 'POST required'}, status=405)
|
||||
import json as _json
|
||||
body = _json.loads(request.body)
|
||||
dataset_id = body.get('dataset_id')
|
||||
filters = body.get('filters', [])
|
||||
logic = body.get('logic', 'and')
|
||||
|
||||
if not dataset_id:
|
||||
return JsonResponse({'error': 'dataset_id is required'})
|
||||
if not filters:
|
||||
return JsonResponse({'error': 'filters is required'})
|
||||
|
||||
try:
|
||||
from analysis.tool_registry import _handle_filter_data
|
||||
result = asyncio.run(_handle_filter_data(
|
||||
dataset_id=dataset_id,
|
||||
filters=filters,
|
||||
logic=logic,
|
||||
))
|
||||
if 'error' in result:
|
||||
return JsonResponse({'error': result['error']})
|
||||
|
||||
filtered_id = result.get('filtered_dataset_id', '')
|
||||
row_count = result.get('row_count', 0)
|
||||
|
||||
# Use the unique flt_xxx ID from _handle_filter_data — never rename
|
||||
# to filtered_{num} which would collide on repeated filters of the same source.
|
||||
return JsonResponse({
|
||||
'dataset_id': filtered_id,
|
||||
'row_count': row_count,
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f'[apply_filter] {traceback.format_exc()}')
|
||||
return JsonResponse({'error': str(e)})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def retry_run(request, display_id):
|
||||
"""Reset a failed run and re-trigger based on run_type."""
|
||||
if request.method != 'POST':
|
||||
return JsonResponse({'error': 'POST required'}, status=405)
|
||||
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
if run.status != 'failed':
|
||||
return JsonResponse({'error': '只能重试失败的分析运行'}, status=400)
|
||||
|
||||
run.status = 'pending'
|
||||
run.error_message = ''
|
||||
run.progress_pct = 0
|
||||
run.progress_msg = ''
|
||||
run.save(update_fields=['status', 'error_message', 'progress_pct', 'progress_msg'])
|
||||
|
||||
if run.run_type == 'upload' and run.csv_glob:
|
||||
upload_dir = Path(run.csv_glob).parent
|
||||
expected_base = ensure_appdata_dir() / 'data' / 'uploads'
|
||||
if not str(upload_dir.resolve()).startswith(str(expected_base.resolve())):
|
||||
return JsonResponse({'error': 'Invalid upload path'}, status=400)
|
||||
if upload_dir.is_dir():
|
||||
t = threading.Thread(
|
||||
target=_background_process,
|
||||
args=(run.id, upload_dir),
|
||||
daemon=True,
|
||||
)
|
||||
t.start()
|
||||
else:
|
||||
run.error_message = '上传数据目录已不存在，请重新上传'
|
||||
run.status = 'failed'
|
||||
run.save(update_fields=['status', 'error_message'])
|
||||
|
||||
return redirect('analysis:run_detail', display_id=run.display_id)
|
||||
@@ -0,0 +1,133 @@
|
||||
"""Upload views: CSV upload, batch, finalize, delete."""
|
||||
import threading
|
||||
import logging
|
||||
import shutil
|
||||
from pathlib import Path
|
||||
|
||||
from django.shortcuts import render, get_object_or_404
|
||||
from django.http import JsonResponse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
|
||||
from analysis.appdata import ensure_appdata_dir
|
||||
from analysis.models import AnalysisRun
|
||||
from .pipeline import _background_process
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
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'})
|
||||
logger.info('[UPLOAD] Received %d files', len(files))
|
||||
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')
|
||||
from analysis.appdata import ensure_appdata_dir
|
||||
base = ensure_appdata_dir() / '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))
|
||||
# Create run with status='pending' — background processing deferred to finalize endpoint
|
||||
run = AnalysisRun.objects.create(
|
||||
csv_glob=str(upload_dir / '*.csv'),
|
||||
status='pending',
|
||||
run_type='upload',
|
||||
)
|
||||
# Deferred: _background_process will be started by finalize_upload endpoint
|
||||
return JsonResponse({'run_id': run.display_id, 'files': saved, 'status': 'pending'})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def upload_csv_batch(request, display_id):
|
||||
"""Append files to existing upload run. Does NOT re-trigger _background_process."""
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
files = request.FILES.getlist('files')
|
||||
if not files:
|
||||
return JsonResponse({'error': 'No files uploaded'})
|
||||
upload_dir = Path(run.csv_glob).parent if run.csv_glob else None
|
||||
if upload_dir:
|
||||
expected_base = ensure_appdata_dir() / 'data' / 'uploads'
|
||||
if not str(upload_dir.resolve()).startswith(str(expected_base.resolve())):
|
||||
return JsonResponse({'error': 'Invalid upload path'}, status=400)
|
||||
if not upload_dir or not upload_dir.exists():
|
||||
return JsonResponse({'error': 'Upload directory not found'})
|
||||
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))
|
||||
# NO background process re-trigger — first batch handles it
|
||||
return JsonResponse({'run_id': run.display_id, 'files': len(saved), 'status': 'ok'})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def finalize_upload(request, display_id):
|
||||
"""Finalize batch upload: validate upload_dir, set status='loading', start background processing."""
|
||||
if request.method != 'POST':
|
||||
return JsonResponse({'error': 'POST required'}, status=405)
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
if run.status != 'pending':
|
||||
return JsonResponse({'error': f'Run status is {run.status}, expected pending'}, status=400)
|
||||
upload_dir = Path(run.csv_glob).parent if run.csv_glob else None
|
||||
if not upload_dir or not upload_dir.exists():
|
||||
return JsonResponse({'error': 'Upload directory not found'}, status=400)
|
||||
run.status = 'loading'
|
||||
run.save(update_fields=['status'])
|
||||
t = threading.Thread(target=_background_process, args=(run.id, upload_dir), daemon=True)
|
||||
t.start()
|
||||
return JsonResponse({'run_id': run.display_id, 'status': 'loading'})
|
||||
|
||||
|
||||
@csrf_exempt
|
||||
def delete_upload(request, display_id):
|
||||
"""Delete uploaded files, SQLite data table, and AnalysisRun.
|
||||
|
||||
Handles all run statuses (loading, profiling, clustering, completed, failed).
|
||||
Gracefully tolerates null csv_glob and missing/errored SQLite tables.
|
||||
"""
|
||||
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
||||
|
||||
# 1. Drop the SQLite data table if one exists
|
||||
if run.sqlite_table:
|
||||
try:
|
||||
from analysis.data_loader import drop_sqlite_table # fmt: skip
|
||||
drop_sqlite_table(run.sqlite_table)
|
||||
except Exception as exc:
|
||||
logger.warning('[DELETE] failed to drop SQLite table %s: %s', run.sqlite_table, exc)
|
||||
|
||||
# 2. Remove the upload directory if csv_glob is set
|
||||
if run.csv_glob:
|
||||
p = Path(run.csv_glob).parent
|
||||
expected_base = ensure_appdata_dir() / 'data' / 'uploads'
|
||||
if not str(p.resolve()).startswith(str(expected_base.resolve())):
|
||||
return JsonResponse({'error': 'Invalid upload path'}, status=400)
|
||||
if p.exists():
|
||||
shutil.rmtree(p, ignore_errors=True)
|
||||
|
||||
# 3. Delete the ORM record
|
||||
run.delete()
|
||||
return JsonResponse({'deleted': True})
|
||||
Reference in New Issue
Block a user