PM-pinou
105 lines
5.0 KiB
Python
105 lines
5.0 KiB
Python
"""Django management command: ``python manage.py run_pipeline``
|
|
|
|
Runs the analysis pipeline from CLI: load → cluster → extract features.
|
|
Usage: uv run python manage.py run_pipeline <csv_glob> [--algo hdbscan|kmeans]
|
|
"""
|
|
import os, sys, json, asyncio
|
|
from django.core.management.base import BaseCommand, CommandError
|
|
from django.utils import timezone
|
|
|
|
class Command(BaseCommand):
|
|
help = 'Run the analysis pipeline from CLI (load → cluster → extract)'
|
|
|
|
def add_arguments(self, parser):
|
|
parser.add_argument('csv_glob', type=str, help='Glob pattern for CSV files')
|
|
parser.add_argument('--algo', type=str, default='hdbscan', choices=['hdbscan', 'kmeans'],
|
|
help='Clustering algorithm (default: hdbscan)')
|
|
parser.add_argument('--output', type=str, default=None,
|
|
help='Output directory for exports (default: no export)')
|
|
|
|
def handle(self, *args, **options):
|
|
import django; django.setup()
|
|
from analysis.models import AnalysisRun
|
|
from analysis.views import _run_pipeline_worker
|
|
|
|
csv_glob = options['csv_glob']
|
|
algo = options['algo']
|
|
output = options['output']
|
|
|
|
# Create AnalysisRun record first (_run_pipeline_worker needs it)
|
|
run_id = AnalysisRun.objects.create(
|
|
csv_glob=csv_glob, status='running',
|
|
).id
|
|
|
|
# Capture self for console output inside the closure
|
|
_self = self
|
|
|
|
def _analysis_fn(run, ctx):
|
|
import warnings
|
|
warnings.filterwarnings('ignore', category=RuntimeWarning, module='sklearn')
|
|
from analysis.session_store import SessionStore
|
|
from analysis.data_loader import load_csv_directory
|
|
|
|
store = SessionStore()
|
|
store.drop_all()
|
|
|
|
# ── Step 1: Load CSV ─────────────────────────────────────────────
|
|
_self.stdout.write(f'[1/3] 加载 CSV: {csv_glob}')
|
|
lf, schema, row_count, file_count, memory_mb = load_csv_directory(csv_glob)
|
|
ds_id = 'manual_ds'
|
|
store.store_dataset(ds_id, lf, schema=schema, metadata={
|
|
'row_count': row_count, 'file_count': file_count, 'csv_glob': csv_glob,
|
|
})
|
|
_self.stdout.write(_self.style.SUCCESS(f' → {row_count} rows, {file_count} files, {memory_mb:.1f} MB'))
|
|
|
|
# Auto-select numeric columns for clustering
|
|
numeric_types = {'Int8', 'Int16', 'Int32', 'Int64',
|
|
'UInt8', 'UInt16', 'UInt32', 'UInt64',
|
|
'Float32', 'Float64'}
|
|
feature_cols = [
|
|
c for c in schema.keys()
|
|
if not c.startswith('_')
|
|
and schema[c].split('(')[0].strip() in numeric_types
|
|
][:10]
|
|
|
|
# ── Step 2: Clustering + Extraction + UMAP (shared pipeline) ─────
|
|
_self.stdout.write(f'[2/3] 聚类 ({algo})')
|
|
_self.stdout.write(f' → 聚类特征: {feature_cols}')
|
|
run.total_flows = row_count
|
|
run.save(update_fields=['total_flows'])
|
|
from analysis.views import _run_clustering_pipeline
|
|
_run_clustering_pipeline(
|
|
run=run, store=store, entity_ds_id=ds_id,
|
|
feature_columns=feature_cols,
|
|
algorithm=algo, min_cluster_size=5,
|
|
run_umap=True,
|
|
)
|
|
if run.status == 'failed':
|
|
_self.stderr.write(_self.style.ERROR(f' ✗ 分析失败: {run.error_message}'))
|
|
return
|
|
_self.stdout.write(_self.style.SUCCESS(
|
|
f' → {run.cluster_count} 个簇'
|
|
))
|
|
|
|
# ── Export ───────────────────────────────────────────────────────
|
|
if output:
|
|
from analysis.tool_registry import _handle_export_results
|
|
os.makedirs(output, exist_ok=True)
|
|
asyncio.run(_handle_export_results(result_id=ds_id, output_path=output, format='csv', overwrite=True))
|
|
_self.stdout.write(_self.style.SUCCESS(f' → 导出到 {output}'))
|
|
|
|
_run_pipeline_worker(run_id, _analysis_fn)
|
|
|
|
# ── Read final status and print summary ─────────────────────────────
|
|
run = AnalysisRun.objects.get(id=run_id)
|
|
if run.status == 'failed':
|
|
err_msg = (run.error_message or '')[:500]
|
|
self.stderr.write(self.style.ERROR(f'\n*** 分析失败! Run ID: #{run.display_id}'))
|
|
if err_msg:
|
|
self.stderr.write(self.style.ERROR(f' {err_msg}'))
|
|
return
|
|
self.stdout.write(self.style.SUCCESS(
|
|
f'\n*** 分析完成! Run ID: #{run.display_id}'
|
|
f' 查看 Django: http://127.0.0.1:8000/runs/{run.display_id}/'
|
|
))
|