fix(upload): per-file background thread — each CSV processed+deleted immediately, no batch accumulation
This commit is contained in:
@@ -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
|
||||
fsize = fpath.stat().st_size
|
||||
saved.append(str(fpath))
|
||||
|
||||
# Create a run for this single file
|
||||
run = AnalysisRun.objects.create(
|
||||
csv_glob=str(upload_dir / '*.csv'),
|
||||
status='pending',
|
||||
csv_glob=str(fpath),
|
||||
status='loading',
|
||||
run_type='upload',
|
||||
)
|
||||
# Deferred: _background_process will be started by finalize_upload endpoint
|
||||
return JsonResponse({'run_id': run.display_id, 'files': saved, 'status': 'pending'})
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user