From 92e1305c07dd530be1b26a595f2d742f3bac4c80 Mon Sep 17 00:00:00 2001
From: PM-pinou <2504420230@qq.com>
Date: Fri, 24 Jul 2026 22:55:09 +0800
Subject: [PATCH] =?UTF-8?q?fix(upload):=20per-file=20background=20thread?=
 =?UTF-8?q?=20=E2=80=94=20each=20CSV=20processed+deleted=20immediately,=20?=
 =?UTF-8?q?no=20batch=20accumulation?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 analysis/views/upload.py | 74 ++++++++++++++++++++++++++++++++++------
 1 file changed, 63 insertions(+), 11 deletions(-)

diff --git a/analysis/views/upload.py b/analysis/views/upload.py
index abebf46..4313bcc 100644
--- a/analysis/views/upload.py
+++ b/analysis/views/upload.py
@@ -43,20 +43,72 @@ def upload_csv(request):
     upload_dir = base / ts
     upload_dir.mkdir(parents=True, exist_ok=True)
     saved = []
+    run_ids = []
     for f in files:
-        path = upload_dir / f.name
-        with open(path, 'wb+') as dst:
+        # 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)
-        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'})
+        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."""
+    import django
+    import os
+    os.environ['DJANGO_SETTINGS_MODULE'] = 'tianxuan.settings'
+    os.environ['DJANGO_ALLOW_ASYNC_UNSAFE'] = 'true'
+    django.setup()
+    from analysis.models import AnalysisRun
+    from analysis.data_loader._sqlite import save_to_db, drop_sqlite_table
+    from analysis.data_loader._csv import load_csv_directory
+    from analysis.session_store import SessionStore
+    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 = f'_data_{run_id}'
+        save_to_db(lf, table_name, schema)
+        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