tianxuan/analysis/views/upload.py

"""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 = []
    run_ids = []
    for f in files:
        # Each file gets its own run and its own processing thread
        fpath = upload_dir / f.name
        with open(fpath, 'wb+') as dst:
            for chunk in f.chunks():
                dst.write(chunk)
        fsize = fpath.stat().st_size
        saved.append(str(fpath))

        # Create a run for this single file
        run = AnalysisRun.objects.create(
            csv_glob=str(fpath),
            status='loading',
            run_type='upload',
        )
        run_ids.append(run.display_id)

        # Start processing thread immediately (per-file, not batched)
        t = threading.Thread(
            target=_process_single_file,
            args=(run.id, str(fpath)),
            daemon=True,
        )
        t.start()

    return JsonResponse({'run_ids': run_ids, 'files': saved, 'status': 'processing'})


def _process_single_file(run_id, filepath):
    """Load a single CSV into SQLite, then delete it. Runs in background thread inside Django."""
    from analysis.data_loader._sqlite import save_to_db, drop_sqlite_table
    from analysis.data_loader._csv import load_csv_directory
    import logging
    logger = logging.getLogger(__name__)

    try:
        run = AnalysisRun.objects.get(id=run_id)
        lf, schema, row_count, file_count, mem = load_csv_directory(filepath)
        table_name = save_to_db(run.id, lf)
        run.sqlite_table = table_name
        run.total_flows = row_count
        run.status = 'ready'
        run.save(update_fields=['sqlite_table', 'total_flows', 'status'])

        # Delete CSV file immediately
        Path(filepath).unlink(missing_ok=True)
        logger.info('[UPLOAD] File %s → SQLite table %s (%d rows), CSV deleted',
                    filepath, table_name, row_count)
    except Exception as exc:
        logger.error('[UPLOAD] Failed to process %s: %s', filepath, exc)
        try:
            run = AnalysisRun.objects.get(id=run_id)
            run.status = 'failed'
            run.error_message = str(exc)
            run.save(update_fields=['status', 'error_message'])
        except Exception:
            pass


@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})