From efc6e3ab9d8cd1ea4d5c37b6f142b551b7421002 Mon Sep 17 00:00:00 2001 From: PM-pinou <2504420230@qq.com> Date: Mon, 20 Jul 2026 19:40:49 +0800 Subject: [PATCH] v8: fix server startup, CSV parse, upload flow, load_from_db types, integration test --- AGENTS.md | 1 + README.md | 6 ++ analysis/data_loader.py | 45 ++++++++++++- analysis/views.py | 74 +++++++++++++++++---- build.bat | 37 +++++++++++ scripts/build_update.py | 46 +++++++++++-- scripts/start_server.py | 4 ++ tests/test_integration.py | 132 ++++++++++++++++++++++++++++++++++++++ tianxuan/wsgi.py | 10 +++ 9 files changed, 334 insertions(+), 21 deletions(-) create mode 100644 build.bat create mode 100644 tests/test_integration.py diff --git a/AGENTS.md b/AGENTS.md index d715f99..935df7c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -93,6 +93,7 @@ tianxuan/ │ ├── run.bat # 用户启动 (含PYTHONUTF8=1) ├── shell.bat # Django shell +├── TlsDB.csv # 参考表头名清单（非实际数据） ├── README.md ├── AGENTS.md └── manage.py diff --git a/README.md b/README.md index 59ea618..189a9d6 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,12 @@ 基于 Polars + scikit-learn + Three.js 的 TLS 流数据分析工具链。支持 CSV 批量导入、自动实体检测、多维聚合、聚类分析、特征提取、**3D 地球流量可视化**。提供 Django Web 操作界面和 MCP 协议接口，支持 32B 大语言模型编排分析流程。**全离线运行**，所有 JS 库/地图贴图/GeoIP 数据库均已内嵌。 +## TlsDB.csv — 参考表头名清单 + +`TlsDB.csv` 是 TLS 流量数据的**参考表头名清单**，包含常见的列名、含义和数据类型对照表。 +**此文件为参考对照表，非实际数据文件。** +实际 CSV 文件的列名可能与表头名不同，分析工具会自动匹配。 + --- ## 目录 diff --git a/analysis/data_loader.py b/analysis/data_loader.py index 0e82504..9865c1e 100644 --- a/analysis/data_loader.py +++ b/analysis/data_loader.py @@ -235,9 +235,39 @@ def save_to_db(run_id: int, lf: pl.LazyFrame, mode: str = 'replace') -> str: return table_name -def load_from_db(table_name: str) -> Optional[pl.LazyFrame]: +_STR_TO_DTYPE = { + 'Float64': pl.Float64, 'Float32': pl.Float32, + 'Int64': pl.Int64, 'Int32': pl.Int32, 'Int16': pl.Int16, 'Int8': pl.Int8, + 'UInt64': pl.UInt64, 'UInt32': pl.UInt32, 'UInt16': pl.UInt16, 'UInt8': pl.UInt8, + 'Utf8': pl.Utf8, 'String': pl.Utf8, 'Categorical': pl.Categorical, + 'Boolean': pl.Boolean, 'Bool': pl.Boolean, + 'Date': pl.Date, 'Datetime': pl.Datetime, 'Time': pl.Time, +} + + +def _str_to_polars_dtype(s: str) -> pl.DataType: + """Convert a Polars dtype string (e.g. ``'Float64'``, ``'Int32'``) to a ``pl.DataType``. + + Strips trailing metadata like ``Datetime(time_unit='us', time_zone=None)``. + """ + base = s.split('(')[0].strip() + return _STR_TO_DTYPE.get(base, pl.Utf8) + + +def load_from_db( + table_name: str, + schema_overrides: dict[str, pl.DataType | str] | None = None, +) -> Optional[pl.LazyFrame]: """Load a SQLite table into a LazyFrame. + Args: + table_name: SQLite table name. + schema_overrides: Optional dict of ``{column_name: dtype}`` to cast + columns after loading. Values can be ``pl.DataType`` instances + (e.g. ``pl.Float64``) or dtype strings (e.g. ``'Float64'``). + Use when SQLite ``TEXT`` columns need to be restored to + ``Float64`` / ``Int64`` etc. + Returns ``None`` if the table does not exist. """ from django.conf import settings @@ -259,6 +289,19 @@ def load_from_db(table_name: str) -> Optional[pl.LazyFrame]: if not rows: return pl.DataFrame().lazy() df = pl.from_dicts([dict(r) for r in rows]) + + # Apply type overrides to restore original column types + if schema_overrides: + casts = {} + for col, dtype in schema_overrides.items(): + if col not in df.columns: + continue + dtype_obj = dtype if isinstance(dtype, pl.DataType) else _str_to_polars_dtype(dtype) + if df[col].dtype != dtype_obj: + casts[col] = pl.col(col).cast(dtype_obj) + if casts: + df = df.with_columns(list(casts.values())) + return df.lazy() finally: conn.close() diff --git a/analysis/views.py b/analysis/views.py index 72280ff..6e79eeb 100644 --- a/analysis/views.py +++ b/analysis/views.py @@ -339,6 +339,7 @@ def upload_csv(request): return JsonResponse({'run_id': run.display_id, 'files': saved, 'status': 'loading'}) +@csrf_exempt def delete_upload(request, display_id): """Delete uploaded files, SQLite data table, and AnalysisRun. @@ -405,6 +406,22 @@ def _background_process(run_id, upload_dir): 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) # 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)) @@ -417,34 +434,63 @@ def _background_process(run_id, upload_dir): from analysis.data_loader import save_to_db, drop_sqlite_table, load_from_db table_name = f'_data_{run_id}' + + # ── Phase 1: Scan first batch ONLY for schema consistency ── + first_batch = csv_paths[:batch_size] + first_lf = pl.scan_csv(first_batch, low_memory=True, + schema_overrides=lat_lon_overrides) + first_schema = dict(first_lf.collect_schema()) + schema = {name: str(dtype) for name, dtype in first_schema.items()} + reference_columns = list(schema.keys()) # canonical column order & set + + # ── Phase 2: Batched collection → SQLite write ── + table_created = False completed_batches = 0 - schema = {} total_rows = 0 - - # ── Phase 1: Stream CSVs into SQLite in batches ───────────────── for i in range(0, total_files, batch_size): - batch_files = csv_paths[i:i + batch_size] - mode = 'replace' if i == 0 else 'append' + batch = csv_paths[i:i+batch_size] + batch_lf = pl.scan_csv(batch, low_memory=True, + schema_overrides=lat_lon_overrides) + df_batch = batch_lf.collect(streaming=True) - batch_lf = pl.scan_csv(batch_files, low_memory=True) + # Schema alignment: ensure columns match reference order + for col in reference_columns: + if col not in df_batch.columns: + df_batch = df_batch.with_columns(pl.lit(None).alias(col)) + extra = [c for c in df_batch.columns if c not in reference_columns] + if extra: + df_batch = df_batch.drop(extra) + df_batch = df_batch.select(reference_columns) - # Capture schema from first batch (efficient — header-only scan) - if not schema: - raw_schema = batch_lf.collect_schema() - schema = {name: str(dtype) for name, dtype in raw_schema.items()} + mode = 'replace' if not table_created else 'append' + save_to_db(run.id, df_batch.lazy(), mode=mode) + table_created = True - save_to_db(run.id, batch_lf, mode=mode) + # 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.progress_msg = f'正在加载数据... ({completed_batches}/{total_batches} 批)' run.save(update_fields=['progress_pct', 'progress_msg']) - # ── Phase 2: Load from SQLite for analysis ───────────────────── - lf = load_from_db(table_name) + # ── 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) + ) + # 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 diff --git a/build.bat b/build.bat new file mode 100644 index 0000000..013fa8c --- /dev/null +++ b/build.bat @@ -0,0 +1,37 @@ +@echo off +chcp 65001 >nul +cd /d "%~dp0" +echo ================================================ +echo 天璇 一键构建工具 +echo ================================================ +echo. +echo 本工具会自动： +echo 1. 提交所有未提交的代码变更 +echo 2. 创建版本标签 (VERSION) +echo 3. 构建增量更新包 +echo. + +:: Check portable Python exists +if not exist "runtime\python\python.exe" ( + echo [INFO] 使用系统 Python... + set PYTHON=python +) else ( + set PYTHON=runtime\python\python.exe +) + +%PYTHON% scripts/build_update.py --auto +if %ERRORLEVEL% neq 0 ( + echo. + echo [ERROR] 构建失败，请检查错误信息 + pause + exit /b %ERRORLEVEL% +) + +echo. +echo ================================================ +echo ✅ 一键构建完成！ +echo ================================================ +echo. +echo 更新包已生成，复制到离线机后双击 update.bat +echo. +pause diff --git a/scripts/build_update.py b/scripts/build_update.py index 054f250..0029ee5 100644 --- a/scripts/build_update.py +++ b/scripts/build_update.py @@ -10,6 +10,7 @@ only modified/new files, ready to be applied on the offline machine. Output: tianxuan_update_-.zip """ +import argparse import hashlib import json import os @@ -166,17 +167,47 @@ def detect_runtime_changes(project_dir: Path) -> bool: return False -def build_update_package(): +def build_update_package(auto_mode=False): print("=" * 50) print(" 天璇 增量更新包构建工具") print("=" * 50) # Step 1: Check working tree if not is_working_tree_clean(): - print("\n[ERROR] 工作树有未提交的变更！") - print(" 请先提交或贮藏 (git stash) 后再构建") - subprocess.run(['git', 'status', '--short'], cwd=PROJECT_DIR) - sys.exit(1) + if auto_mode: + # Step A: Auto-commit any uncommitted changes + print("\n[AUTO] 检测到未提交的变更，自动提交中...") + subprocess.run(['git', 'add', '-A'], cwd=PROJECT_DIR, check=True) + status = subprocess.run( + ['git', 'status', '--short'], + capture_output=True, text=True, cwd=PROJECT_DIR + ) + file_count = len([l for l in status.stdout.strip().split('\n') if l.strip()]) + version = get_current_version() + commit_msg = f"auto: update for v{version}" + subprocess.run( + ['git', 'commit', '-m', commit_msg], + cwd=PROJECT_DIR, check=True + ) + print(f" ✅ 已提交 {file_count} 个文件: {commit_msg}") + else: + print("\n[ERROR] 工作树有未提交的变更！") + print(" 请先提交或贮藏 (git stash) 后再构建") + subprocess.run(['git', 'status', '--short'], cwd=PROJECT_DIR) + sys.exit(1) + + # Step 1.5: Auto-tag (auto mode only) + if auto_mode: + current_ver = get_current_version() + result = subprocess.run( + ['git', 'tag', '-l', current_ver], + capture_output=True, text=True, cwd=PROJECT_DIR + ) + if current_ver not in result.stdout: + subprocess.run(['git', 'tag', current_ver], cwd=PROJECT_DIR, check=True) + print(f" ✅ 已创建 tag: {current_ver}") + else: + print(f" ℹ️ Tag {current_ver} 已存在，跳过") # Step 2: Determine versions current_version = get_current_version() @@ -301,4 +332,7 @@ def build_update_package(): if __name__ == '__main__': - build_update_package() + parser = argparse.ArgumentParser(description='Build incremental update package for 天璇') + parser.add_argument('--auto', '-a', action='store_true', help='One-click mode: auto-commit + auto-tag + build') + args = parser.parse_args() + build_update_package(auto_mode=args.auto) diff --git a/scripts/start_server.py b/scripts/start_server.py index 7125ee1..64db19c 100644 --- a/scripts/start_server.py +++ b/scripts/start_server.py @@ -13,6 +13,10 @@ import sys import webbrowser from pathlib import Path +_project_root = Path(__file__).resolve().parent.parent +if str(_project_root) not in sys.path: + sys.path.insert(0, str(_project_root)) + from config import get_config diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..cf83db7 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,132 @@ +""" +Full-stack integration test. Run: python tests/test_integration.py +IMPORTANT: Update this file whenever frontend HTML/API changes. +Pure Python — no curl, no requests required. +""" +import subprocess, time, json, sys, threading, os, uuid, urllib.request, urllib.error + +PORT = 18765 +BASE = f"http://127.0.0.1:{PORT}" +errors = [] +_lock = threading.Lock() + +def fail(msg): + with _lock: errors.append(msg) + +def check(cond, msg): + if not cond: fail(msg) + +def start_server(): + manage_py = os.path.join(os.path.dirname(__file__), '..', 'manage.py') + proc = subprocess.Popen( + [sys.executable, manage_py, 'runserver', f'127.0.0.1:{PORT}', '--noreload'], + stderr=subprocess.PIPE, stdout=subprocess.PIPE, text=True, encoding='utf-8', + ) + def monitor(): + for line in proc.stderr: + stripped = line.strip() + if 'ERROR' in stripped.upper() and 'GET' not in stripped and 'POST' not in stripped: + fail(f"STDERR: {stripped}") + threading.Thread(target=monitor, daemon=True).start() + return proc + +def build_multipart_formdata(fields): + """Build multipart/form-data body and content-type boundary. + fields: {name: (filename, content_type, data_bytes)} + """ + boundary = uuid.uuid4().hex + lines = [] + for name, (filename, content_type, data) in fields.items(): + lines.append(f'--{boundary}'.encode()) + lines.append(f'Content-Disposition: form-data; name="{name}"; filename="{filename}"'.encode()) + lines.append(f'Content-Type: {content_type}'.encode()) + lines.append(b'') + lines.append(data) + lines.append(f'--{boundary}--'.encode()) + return b'\r\n'.join(lines), f'multipart/form-data; boundary={boundary}' + +def http_get(path, max_retries=10): + """GET request with retry on connection refused.""" + for attempt in range(max_retries): + try: + r = urllib.request.urlopen(f'{BASE}{path}', timeout=5) + body = r.read() + code = r.status + r.close() + return code, body + except urllib.error.URLError as e: + if 'refused' in str(e).lower() and attempt < max_retries - 1: + time.sleep(1) + continue + return 502, b'Connection refused' + +def http_post_multipart(path, fields): + """POST multipart form data, returns (status_code, body_bytes).""" + body_bytes, content_type = build_multipart_formdata(fields) + req = urllib.request.Request(f'{BASE}{path}', data=body_bytes, + headers={'Content-Type': content_type}) + try: + r = urllib.request.urlopen(req, timeout=30) + return r.status, r.read() + except urllib.error.HTTPError as e: + return e.code, e.read() + +def main(): + proc = start_server() + time.sleep(3) + + # Test 1: Homepage + code, _ = http_get('/') + check(code == 200, f"GET / → {code}") + + # Test 2: Upload CSV with '+' in lat/lon + csv_content = ( + ":ips,:ipd,0ver,:prd,:prs,:ips.latd,:ips.lond\n" + "1.2.3.4,5.6.7.8,0303,443,80,35.0,139.0\n" + "9.10.11.12,13.14.15.16,0304,8080,443,+,+\n" + ) + fields = {'files': ('test.csv', 'text/csv', csv_content.encode('utf-8'))} + code, body = http_post_multipart('/upload/csv/', fields) + check(code == 200, f"POST /upload/csv/ → {code}") + data = json.loads(body) + run_id = data.get('run_id') + check(run_id is not None, "No run_id in upload response") + + # Test 3: Poll run status until ready + for _ in range(30): + code, body = http_get(f'/runs/{run_id}/status/') + check(code == 200, f"GET /runs/{run_id}/status/ → {code}") + s = json.loads(body) + if s['status'] in ('ready', 'completed', 'failed'): + check(s['status'] != 'failed', f"Run failed: {s.get('error_message','')}") + break + time.sleep(1) + + # Test 4: Run list page + code, _ = http_get('/runs/') + check(code == 200, f"GET /runs/ → {code}") + + # Test 5: Globe page + code, _ = http_get('/globe/') + check(code == 200, f"GET /globe/ → {code}") + + # Test 6: Config page + code, _ = http_get('/config/') + check(code == 200, f"GET /config/ → {code}") + + # Test 7: Delete run — @csrf_exempt means 200 (or 500), not 403, not 404 + code, _ = http_post_multipart(f'/runs/{run_id}/delete/', {}) + check(code != 404, f"POST /runs/{run_id}/delete/ → {code} (should not be 404)") + + # Cleanup + proc.terminate() + proc.wait() + + if errors: + print("INTEGRATION TEST FAILED:") + for e in errors: print(f" {e}") + sys.exit(1) + print("ALL INTEGRATION TESTS PASSED") + +if __name__ == "__main__": + main() diff --git a/tianxuan/wsgi.py b/tianxuan/wsgi.py index 0da3fff..e5d3c4a 100644 --- a/tianxuan/wsgi.py +++ b/tianxuan/wsgi.py @@ -5,6 +5,16 @@ os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'tianxuan.settings') application = get_wsgi_application() +# ── Auto-migrate on fresh DB ────────────────────────────────────── +# A brand-new database has zero tables — migrate + import tls_ref_data. +from django.db import connection +from django.core.management import call_command +with connection.cursor() as cursor: + cursor.execute("SELECT count(*) FROM sqlite_master WHERE type='table'") + if cursor.fetchone()[0] == 0: + call_command('migrate') + call_command('import_tlsdb') + # ── DB integrity check ──────────────────────────────────────────── # Uses settings DB path (may be %APPDATA%/TianXuan/db.sqlite3) from django.conf import settings