PM-pinou
1936 lines
80 KiB
Python
1936 lines
80 KiB
Python
import json
|
|
import time
|
|
import urllib.request
|
|
import urllib.error
|
|
import threading
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from analysis.appdata import ensure_appdata_dir
|
|
|
|
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.http import HttpResponse, JsonResponse
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
|
|
from config import get_config, save_config, Config
|
|
from .models import AnalysisRun, ClusterResult, EntityProfile
|
|
|
|
|
|
def dashboard(request):
|
|
"""Home page: recent runs overview."""
|
|
runs = AnalysisRun.objects.all()[:20]
|
|
return render(request, 'analysis/dashboard.html', {'runs': runs})
|
|
|
|
|
|
def run_list(request):
|
|
"""List all analysis runs with pagination 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 _extract_lat(val):
|
|
"""Try to extract a latitude from *val* (number or str)."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
v = float(val)
|
|
return None if v < -90 or v > 90 else v
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def _extract_lon(val):
|
|
"""Try to extract a longitude from *val* (number or str)."""
|
|
if val is None:
|
|
return None
|
|
try:
|
|
v = float(val)
|
|
return None if v < -180 or v > 180 else v
|
|
except (ValueError, TypeError):
|
|
return None
|
|
|
|
|
|
def cluster_overview(request, 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
|
|
entities = run.entities.exclude(embedding_x=None).exclude(embedding_y=None).only('embedding_x', 'embedding_y', 'cluster_label', 'entity_value')
|
|
scatter_data = [
|
|
{'x': e.embedding_x, 'y': e.embedding_y, 'label': e.cluster_label, 'entity': e.entity_value}
|
|
for e in entities
|
|
]
|
|
|
|
# Build geo scatter data from feature_json (lat/lon)
|
|
geo_data = []
|
|
geo_skipped = 0
|
|
for e in run.entities.all().only('feature_json', 'cluster_label', 'entity_value'):
|
|
fj = e.feature_json
|
|
if not fj:
|
|
continue
|
|
lat = _extract_lat(fj.get('avg_latitude'))
|
|
lon = _extract_lon(fj.get('avg_longitude'))
|
|
if lat is not None and lon is not None:
|
|
geo_data.append({
|
|
'x': lon,
|
|
'y': lat,
|
|
'label': e.cluster_label if e.cluster_label is not None else -1,
|
|
'entity': e.entity_value,
|
|
})
|
|
else:
|
|
geo_skipped += 1
|
|
|
|
# Cluster sizes for bar chart
|
|
cluster_sizes = [{'label': c.cluster_label, 'size': c.size, 'silhouette': c.silhouette_score}
|
|
for c in clusters]
|
|
|
|
# Top features per cluster from ClusterFeature
|
|
top_features = {}
|
|
for c in clusters:
|
|
features_qs = c.features.all().order_by('-distinguishing_score')[:10]
|
|
top_features[str(c.cluster_label)] = [
|
|
{'feature_name': f.feature_name, 'score': f.distinguishing_score, 'mean': f.mean}
|
|
for f in features_qs
|
|
]
|
|
|
|
return render(request, 'analysis/cluster_overview.html', {
|
|
'run': run,
|
|
'clusters': clusters,
|
|
'noise': noise,
|
|
'scatter_data_json': json.dumps(scatter_data),
|
|
'geo_data_json': json.dumps(geo_data),
|
|
'geo_skipped': geo_skipped,
|
|
'geo_count': len(geo_data),
|
|
'cluster_sizes_json': json.dumps(cluster_sizes),
|
|
'top_features_json': json.dumps(top_features),
|
|
})
|
|
|
|
|
|
def cluster_detail(request, display_id, cluster_label):
|
|
"""Detailed view of a single cluster with entity list."""
|
|
run = get_object_or_404(AnalysisRun, display_id=display_id)
|
|
cluster = get_object_or_404(ClusterResult, run=run, cluster_label=cluster_label)
|
|
features = cluster.features.all().order_by('-distinguishing_score')[:50]
|
|
entities = cluster.entities.all()[:50]
|
|
|
|
return render(request, 'analysis/cluster_detail.html', {
|
|
'run': run,
|
|
'cluster': cluster,
|
|
'features': features,
|
|
'entities': entities,
|
|
})
|
|
|
|
|
|
def entity_profile(request, entity_id):
|
|
"""Detailed profile for a single entity."""
|
|
entity = get_object_or_404(EntityProfile, id=entity_id)
|
|
run = entity.run
|
|
|
|
# Compute deviation from cluster mean
|
|
deviations = {}
|
|
if entity.feature_json and entity.cluster:
|
|
cluster_features = {
|
|
f.feature_name: {'mean': f.mean, 'std': f.std}
|
|
for f in entity.cluster.features.all()
|
|
}
|
|
for col, val in entity.feature_json.items():
|
|
if col in cluster_features and cluster_features[col].get('std') and cluster_features[col]['std'] > 1e-10:
|
|
deviations[col] = {
|
|
'value': val,
|
|
'cluster_mean': cluster_features[col]['mean'],
|
|
'zscore': (val - cluster_features[col]['mean']) / cluster_features[col]['std'],
|
|
}
|
|
|
|
return render(request, 'analysis/entity_profile.html', {
|
|
'entity': entity,
|
|
'run': run,
|
|
'deviations': deviations,
|
|
})
|
|
|
|
|
|
def start_analysis(request):
|
|
"""Placeholder: API endpoint to trigger analysis via MCP."""
|
|
return JsonResponse({'status': 'not_implemented', 'message': 'Use MCP tools to trigger analysis'})
|
|
|
|
|
|
def config_view(request):
|
|
"""Config page: GET renders current config, POST saves updates."""
|
|
cfg = get_config()
|
|
|
|
if request.method == 'POST':
|
|
# Build a new Config from POST data (fall back to current values)
|
|
server = Config.Server(
|
|
host=request.POST.get('server_host', cfg.server.host),
|
|
port=int(request.POST.get('server_port', cfg.server.port)),
|
|
debug=request.POST.get('server_debug') == 'true',
|
|
)
|
|
data = Config.Data(
|
|
schema_strict=request.POST.get('data_schema_strict') == 'true',
|
|
recursive=request.POST.get('data_recursive') == 'true',
|
|
)
|
|
clustering = Config.Clustering(
|
|
algorithm=request.POST.get('clustering_algorithm', cfg.clustering.algorithm),
|
|
min_cluster_size=int(request.POST.get('clustering_min_cluster_size', cfg.clustering.min_cluster_size)),
|
|
random_state=cfg.clustering.random_state,
|
|
)
|
|
llm = Config.LLM(
|
|
enabled=request.POST.get('llm_enabled') == 'true',
|
|
base_url=request.POST.get('llm_base_url', cfg.llm.base_url),
|
|
api_key=request.POST.get('llm_api_key', cfg.llm.api_key),
|
|
model=request.POST.get('llm_model', cfg.llm.model),
|
|
)
|
|
cfg = Config(server=server, data=data, clustering=clustering, llm=llm)
|
|
save_config(cfg)
|
|
# Re-render with saved flag
|
|
return render(request, 'tianxuan/config.html', {'config': cfg, 'saved': True})
|
|
|
|
return render(request, 'tianxuan/config.html', {'config': cfg})
|
|
|
|
|
|
@csrf_exempt
|
|
def llm_test(request):
|
|
"""LLM connectivity test: POST JSON {base_url, api_key, model} → chat/completions."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'success': False, 'message': '仅支持 POST 请求', 'latency_ms': 0}, status=405)
|
|
|
|
try:
|
|
body = json.loads(request.body)
|
|
except json.JSONDecodeError:
|
|
return JsonResponse({'success': False, 'message': '无效的 JSON 请求体', 'latency_ms': 0})
|
|
|
|
base_url = (body.get('base_url') or '').rstrip('/')
|
|
api_key = body.get('api_key') or ''
|
|
model = body.get('model') or ''
|
|
|
|
if not base_url:
|
|
return JsonResponse({'success': False, 'message': 'Base URL 不能为空', 'latency_ms': 0})
|
|
if not api_key:
|
|
return JsonResponse({'success': False, 'message': 'API Key 不能为空', 'latency_ms': 0})
|
|
if not model:
|
|
return JsonResponse({'success': False, 'message': 'Model 不能为空', 'latency_ms': 0})
|
|
|
|
url = f'{base_url}/chat/completions'
|
|
payload = json.dumps({
|
|
'model': model,
|
|
'messages': [{'role': 'user', 'content': 'respond with ok'}],
|
|
'max_tokens': 5,
|
|
}).encode('utf-8')
|
|
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=payload,
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'Bearer {api_key}',
|
|
},
|
|
method='POST',
|
|
)
|
|
|
|
start = time.monotonic()
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
data = json.loads(resp.read().decode('utf-8'))
|
|
if 'choices' in data and len(data['choices']) > 0:
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=True latency={elapsed}ms')
|
|
return JsonResponse({
|
|
'success': True,
|
|
'message': f'连接成功，模型 {model} 返回正常',
|
|
'latency_ms': elapsed,
|
|
})
|
|
else:
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=True latency={elapsed}ms')
|
|
return JsonResponse({
|
|
'success': True,
|
|
'message': f'连接成功，但响应中无 choices（原始响应已记录）',
|
|
'latency_ms': elapsed,
|
|
})
|
|
except urllib.error.HTTPError as e:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
try:
|
|
detail = json.loads(e.read().decode('utf-8', errors='replace'))
|
|
msg = detail.get('error', {}).get('message', str(e))
|
|
except Exception:
|
|
msg = str(e)
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': f'HTTP {e.code}: {msg}', 'latency_ms': elapsed})
|
|
except urllib.error.URLError as e:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
reason = str(e.reason) if e.reason else '连接失败'
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': f'连接失败: {reason}', 'latency_ms': elapsed})
|
|
except TimeoutError:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': '请求超时（5秒）', 'latency_ms': elapsed})
|
|
except Exception as e:
|
|
elapsed = int((time.monotonic() - start) * 1000)
|
|
logger.info(f'[LLM_TEST] base_url={base_url} model={model} success=False latency={elapsed}ms')
|
|
return JsonResponse({'success': False, 'message': f'未知错误: {str(e)}', 'latency_ms': elapsed})
|
|
|
|
|
|
# ── Upload page ─────────────────────────────────────────────────────────
|
|
|
|
def upload_page(request):
|
|
"""CSV upload page."""
|
|
runs = AnalysisRun.objects.all().order_by('-created_at')[:20]
|
|
return render(request, 'tianxuan/upload.html', {'runs': runs})
|
|
|
|
|
|
@csrf_exempt
|
|
def upload_csv(request):
|
|
"""Handle CSV upload AJAX POST."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
files = request.FILES.getlist('files')
|
|
if not files:
|
|
return JsonResponse({'error': 'No files uploaded'})
|
|
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)
|
|
import logging
|
|
import shutil
|
|
|
|
_logger = logging.getLogger(__name__)
|
|
|
|
# 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})
|
|
|
|
|
|
import traceback, logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
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'])
|
|
|
|
|
|
# ── Manual analysis page ───────────────────────────────────────────────
|
|
|
|
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()
|
|
datasets = store.list_datasets() if hasattr(store, 'list_datasets') else []
|
|
|
|
# Get upload AnalysisRun statuses for dataset badges
|
|
from analysis.models import AnalysisRun
|
|
upload_runs = AnalysisRun.objects.filter(run_type='upload').order_by('-created_at')[:50]
|
|
run_status_map = {
|
|
f'upload_{r.display_id}': {
|
|
'status': r.status,
|
|
'total_flows': r.total_flows,
|
|
'display_id': r.display_id,
|
|
}
|
|
for r in upload_runs
|
|
}
|
|
|
|
# Build datasets_with_status: each dataset annotated with run status
|
|
datasets_with_status = []
|
|
for ds in datasets:
|
|
run_info = run_status_map.get(ds['dataset_id'])
|
|
if run_info:
|
|
datasets_with_status.append({
|
|
**ds,
|
|
'run_status': run_info['status'],
|
|
'total_flows': run_info['total_flows'],
|
|
'display_id': run_info['display_id'],
|
|
})
|
|
else:
|
|
datasets_with_status.append({**ds, 'run_status': 'ready', 'total_flows': None, 'display_id': None})
|
|
|
|
# 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'] = '已分析'
|
|
elif s in ('failed',):
|
|
ds_info['status_label'] = '错误'
|
|
elif s in ('pending', 'loading', 'profiling', 'aggregating', 'clustering', 'extracting'):
|
|
ds_info['status_label'] = '分析中'
|
|
else:
|
|
ds_info['status_label'] = '待分析'
|
|
|
|
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,
|
|
'datasets_with_status_json': _json.dumps(datasets_with_status, ensure_ascii=False),
|
|
})
|
|
|
|
|
|
# ── Pipeline worker ─────────────────────────────────────────────────────────
|
|
|
|
|
|
def _run_pipeline_worker(run_id, analysis_fn, **ctx):
|
|
"""Unified background worker: setup / ORM / progress / error handling."""
|
|
import django
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = 'tianxuan.settings'
|
|
os.environ['DJANGO_ALLOW_ASYNC_UNSAFE'] = 'true'
|
|
django.setup()
|
|
|
|
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
|
|
tb = traceback.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']))()
|
|
|
|
|
|
# ── Shared clustering pipeline ─────────────────────────────────────────
|
|
|
|
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, traceback, logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
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 3: UMAP-2D embedding (sampled training for large data) ─────────
|
|
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)
|
|
|
|
if len(df_umap) > 2:
|
|
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=2, 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=2, random_state=42,
|
|
n_neighbors=15, min_dist=0.1,
|
|
metric='euclidean')
|
|
coords = reducer.fit_transform(mat)
|
|
|
|
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()
|
|
profiles.append(EP(
|
|
entity_value=ev, run=run, cluster_label=lbl,
|
|
cluster=cr_cache[cache_key],
|
|
embedding_x=float(coords[i, 0]) if i < len(coords) else None,
|
|
embedding_y=float(coords[i, 1]) if i < len(coords) else None,
|
|
))
|
|
if profiles:
|
|
EP.objects.bulk_create(profiles, ignore_conflicts=True, batch_size=1000)
|
|
run.entity_count = len(profiles)
|
|
run.save(update_fields=['entity_count'])
|
|
except Exception as umap_err:
|
|
logger.warning(f'UMAP-2D 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)
|
|
|
|
|
|
@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
|
|
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', 'hdbscan')
|
|
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)
|
|
from analysis.views import _run_clustering_pipeline
|
|
_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}/'})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ── 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
|
|
import asyncio
|
|
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)
|
|
|
|
# Re-store with filtered_X ID pattern
|
|
if dataset_id.startswith('upload_'):
|
|
display_suffix = dataset_id.split('_', 1)[1]
|
|
new_id = f'filtered_{display_suffix}'
|
|
else:
|
|
new_id = filtered_id
|
|
|
|
from analysis.session_store import SessionStore
|
|
store = SessionStore()
|
|
entry = store.get_dataset(filtered_id)
|
|
if entry and new_id != filtered_id:
|
|
store.store_dataset(
|
|
dataset_id=new_id,
|
|
lazyframe=entry['lazyframe'],
|
|
schema=entry.get('schema', {}),
|
|
metadata=dict(entry.get('metadata', {}), parent_dataset_id=dataset_id),
|
|
parent_id=dataset_id,
|
|
)
|
|
new_id_to_use = new_id
|
|
else:
|
|
new_id_to_use = filtered_id
|
|
|
|
return JsonResponse({
|
|
'dataset_id': new_id_to_use,
|
|
'row_count': row_count,
|
|
})
|
|
except Exception as e:
|
|
import traceback
|
|
logger.error(f'[apply_filter] {traceback.format_exc()}')
|
|
return JsonResponse({'error': str(e)})
|
|
|
|
|
|
# ── LLM auto analysis page ─────────────────────────────────────────────
|
|
|
|
|
|
@csrf_exempt
|
|
def run_llm_analysis_view(request):
|
|
"""AJAX POST: run LLM-driven analysis pipeline."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
import json
|
|
body = json.loads(request.body)
|
|
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()
|
|
# Use display_id for dataset key (consistent with _background_process)
|
|
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)
|
|
upload_ds_id = f'upload_{pk}'
|
|
entry = store.get_dataset(upload_ds_id)
|
|
|
|
if entry is None:
|
|
# Try entity dataset as fallback
|
|
entity_ds_id = f'entity_{pk}'
|
|
entry = store.get_dataset(entity_ds_id)
|
|
if entry is None:
|
|
return JsonResponse({'error': '数据集不存在，请先上传并等待预处理完成'})
|
|
|
|
dataset_id = upload_ds_id if store.get_dataset(upload_ds_id) else f'entity_{pk}'
|
|
|
|
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
|
|
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:
|
|
# Real tool call — log it
|
|
if result:
|
|
if isinstance(result, dict):
|
|
# Format dict as JSON, truncate large values
|
|
import json as _json
|
|
summary = _json.dumps(result, default=str, ensure_ascii=False)[:500]
|
|
else:
|
|
summary = str(result)[:500]
|
|
run.run_log += f'[{step}] {tool_name}: {summary}...\n'
|
|
else:
|
|
run.run_log += f'[{step}] {tool_name}\n'
|
|
|
|
# Capture structured tool call data
|
|
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='hdbscan',
|
|
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:
|
|
import json as _json
|
|
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})
|
|
|
|
|
|
def auto_page(request):
|
|
"""LLM-driven auto analysis page."""
|
|
cfg = get_config()
|
|
selected_run = None
|
|
run_id = request.GET.get('run_id')
|
|
if run_id:
|
|
try:
|
|
selected_run = AnalysisRun.objects.get(display_id=int(run_id))
|
|
except (ValueError, AnalysisRun.DoesNotExist):
|
|
pass
|
|
return render(request, 'tianxuan/auto.html', {
|
|
'llm_configured': bool(cfg.llm.base_url and cfg.llm.api_key),
|
|
'selected_run': selected_run,
|
|
})
|
|
|
|
|
|
# ── Progress API ───────────────────────────────────────────────────────
|
|
|
|
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 log_viewer(request):
|
|
"""Display last 200 lines of the app log (DEBUG only).
|
|
With ?pos=N, returns text: line_count\\nnew_content for AJAX polling.
|
|
"""
|
|
from django.conf import settings
|
|
if not settings.DEBUG:
|
|
return HttpResponse('Log viewer only available in DEBUG mode', status=403)
|
|
log_path = settings.BASE_DIR / 'logs' / 'tianxuan.log'
|
|
pos = request.GET.get('pos')
|
|
if pos is not None:
|
|
# AJAX polling mode: return new content since byte position
|
|
try:
|
|
pos = int(pos)
|
|
except ValueError:
|
|
pos = 0
|
|
if log_path.exists():
|
|
with open(log_path, 'r', encoding='utf-8') as f:
|
|
total = f.seek(0, 2) # seek to end
|
|
if pos >= total:
|
|
return HttpResponse(f'{total}\n', content_type='text/plain')
|
|
f.seek(pos)
|
|
new_content = f.read()
|
|
return HttpResponse(f'{total}\n{new_content}', content_type='text/plain')
|
|
return HttpResponse('0\n', content_type='text/plain')
|
|
|
|
# HTML page mode
|
|
lines = []
|
|
if log_path.exists():
|
|
with open(log_path, 'r', encoding='utf-8') as f:
|
|
lines = f.readlines()[-200:]
|
|
return render(request, 'tianxuan/log_viewer.html', {'lines': lines})
|
|
|
|
# ── 3D Globe view ─────────────────────────────────────────────────────────
|
|
|
|
def _extract_flows_from_df(df, MAX_ROWS):
|
|
"""Extract flow data (lat/lon/TLS/bytes) from a Polars DataFrame."""
|
|
from analysis.geoip import lookup as geo_lookup
|
|
TLS_HEX_MAP = {'0303': 'TLSv1.2', '0304': 'TLSv1.3', '0302': 'TLSv1.1', '0301': 'TLSv1.0'}
|
|
# NOTE: Duplicated in type_classifier.py — keep both in sync.
|
|
|
|
src_lat_col = next((c for c in df.columns if c.lower() in (
|
|
':ips.latd', ':ips_latd', ':ips_lat', 'src_latitude', 'src_lat', 'source_latitude', 'source_lat')), None)
|
|
src_lon_col = next((c for c in df.columns if c.lower() in (
|
|
':ips.lond', ':ips_lond', ':ips_lon', 'src_longitude', 'src_lon', 'source_longitude', 'source_lon')), None)
|
|
dst_lat_col = next((c for c in df.columns if c.lower() in (
|
|
':ipd.latd', ':ipd_latd', ':ipd_lat', 'dst_latitude', 'dst_lat', 'dest_latitude', 'dest_lat')), None)
|
|
dst_lon_col = next((c for c in df.columns if c.lower() in (
|
|
':ipd.lond', ':ipd_lond', ':ipd_lon', 'dst_longitude', 'dst_lon', 'dest_longitude', 'dest_lon')), None)
|
|
src_ip_col = next((c for c in df.columns if 'src_ip' in c.lower() or ':ips' in c.lower()), None)
|
|
dst_ip_col = next((c for c in df.columns if 'dst_ip' in c.lower() or ':ipd' in c.lower()), None)
|
|
tls_col = next((c for c in df.columns if c.lower() in ('0ver', 'tls_version', '_tlsver', 'version')), None)
|
|
bytes_col = next((c for c in df.columns if c.lower() in ('8ack', '8pak', '8byt', 'bytes_sent', 'bytes')), None)
|
|
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):
|
|
import json
|
|
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:
|
|
import logging
|
|
logging.getLogger(__name__).warning(f'globe ?data= param failed: {exc}')
|
|
return render(request, 'tianxuan/globe.html', {
|
|
'geo_flows': json.dumps(flows),
|
|
'all_runs': all_runs,
|
|
'selected_ids': [],
|
|
})
|
|
|
|
# Get all completed runs for the selector
|
|
all_runs = list(AnalysisRun.objects.filter(status='completed').order_by('-id').values('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
|
|
|
|
|
|
# ── MCP Tool Lab — manual tool execution ─────────────────────────────
|
|
|
|
|
|
def tool_lab(request):
|
|
"""List all available MCP tools with descriptions and input schemas."""
|
|
from analysis.tool_registry import get_tools_meta
|
|
tools = get_tools_meta()
|
|
# Group tools by category
|
|
core_names = {'profile_data', 'build_entity_profiles', 'compute_scores',
|
|
'run_clustering', 'extract_features', 'detect_anomalies',
|
|
'visualize_anomalies'}
|
|
diag_names = {'validate_data', 'explore_distributions', 'find_outliers',
|
|
'diagnose_clustering', 'compare_datasets', 'export_debug_sample',
|
|
'repair_schema'}
|
|
analysis_names = {'analyze_patterns', 'analyze_temporal', 'analyze_tls_health',
|
|
'analyze_geo_distribution', 'analyze_entity_detail'}
|
|
|
|
core_tools, diag_tools, analysis_tools, other_tools = [], [], [], []
|
|
for t in tools:
|
|
if t.name in core_names:
|
|
core_tools.append(t)
|
|
elif t.name in diag_names:
|
|
diag_tools.append(t)
|
|
elif t.name in analysis_names:
|
|
analysis_tools.append(t)
|
|
else:
|
|
other_tools.append(t)
|
|
|
|
# Get list of datasets for parameter autofill
|
|
from analysis.session_store import SessionStore
|
|
store = SessionStore()
|
|
ds_list = store.list_datasets() if hasattr(store, 'list_datasets') else []
|
|
|
|
import json as _json
|
|
# Serialize tools metadata for JS
|
|
meta_list = []
|
|
for t in tools:
|
|
meta_list.append({
|
|
'name': t.name,
|
|
'description': t.description,
|
|
'inputSchema': t.inputSchema,
|
|
})
|
|
return render(request, 'analysis/tool_lab.html', {
|
|
'core_tools': core_tools,
|
|
'diag_tools': diag_tools,
|
|
'analysis_tools': analysis_tools,
|
|
'other_tools': other_tools,
|
|
'datasets': ds_list,
|
|
'tools_meta': _json.dumps(meta_list, ensure_ascii=False),
|
|
})
|
|
|
|
|
|
@csrf_exempt
|
|
def tool_lab_run(request):
|
|
"""Execute a single MCP tool via AJAX POST."""
|
|
if request.method != 'POST':
|
|
return JsonResponse({'error': 'POST required'}, status=405)
|
|
import json as _json
|
|
try:
|
|
body = _json.loads(request.body)
|
|
except Exception:
|
|
return JsonResponse({'error': 'Invalid JSON body'}, status=400)
|
|
|
|
tool_name = body.get('tool')
|
|
params = body.get('params', {})
|
|
if not tool_name:
|
|
return JsonResponse({'error': 'Missing tool name'}, status=400)
|
|
|
|
from analysis.tool_registry import handle_call
|
|
import asyncio
|
|
result = asyncio.run(handle_call(tool_name, params))
|
|
return JsonResponse({'status': 'ok', 'result': result})
|
|
|
|
|
|
# ── Save / Load / Delete analysis plans ─────────────────────────────
|
|
|
|
from pathlib import Path
|
|
_PLANS_DIR = Path(__file__).resolve().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}."""
|
|
import json as _j
|
|
if not _AUTO_INDEX_PATH.exists():
|
|
return []
|
|
try:
|
|
data = _j.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."""
|
|
import json as _j
|
|
_AUTO_INDEX_PATH.write_text(
|
|
_j.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)
|
|
|
|
|
|
@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)
|
|
|
|
|
|
@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)
|
|
|