diff --git a/VERSION b/VERSION index 89fb16f..46882bc 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -v1.1.4 \ No newline at end of file +v1.1.6 \ No newline at end of file diff --git a/analysis/views.py b/analysis/views.py index 9f82b23..b8ab028 100644 --- a/analysis/views.py +++ b/analysis/views.py @@ -444,7 +444,7 @@ def _background_process(run_id, upload_dir): for _h in _headers: if _h not in all_columns: all_columns.append(_h) - reference_columns = all_columns + reference_columns = sorted(all_columns, key=str.lower) schema = {col: 'Utf8' for col in all_columns} # ── Phase 2: Create SQLite table once with all columns as TEXT ── @@ -506,6 +506,50 @@ def _background_process(run_id, upload_dir): pl.col(lat_lon_col).map_batches(_coerce_to_float_fn, return_dtype=pl.Float64).alias(lat_lon_col) ) + # ── Restore numeric types for Utf8 columns coerced during CSV load ── + # All columns were forced to Utf8 for safe SQLite storage. Now try to + # recover the original numeric types (Int64 / Float64) from the data. + numeric_types = (pl.Int8, pl.Int16, pl.Int32, pl.Int64, + pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64, + pl.Float32, pl.Float64) + # Get current schema to know which columns are still Utf8 + live_schema = lf.collect_schema() + live_names = live_schema.names() + live_dtypes = list(live_schema.dtypes()) + utf8_cols = [ + live_names[i] for i, dt in enumerate(live_dtypes) + if dt == pl.Utf8 and not live_names[i].startswith('_') + ] + if utf8_cols: + # Infer types from a small sample to decide Float64 vs Int64 + sample_df = lf.select(utf8_cols).head(1000).collect(streaming=True) + type_map: dict[str, pl.DataType] = {} + for col in utf8_cols: + try: + # Try Float64 first (most general numeric type) + casted = sample_df[col].cast(pl.Float64, strict=False) + non_null = casted.is_not_null().sum() + if non_null > len(casted) * 0.5: + # Check if all non-null values are integers → use Int64 + all_int = True + for v in casted.head(100).to_list(): + if v is not None and v != int(v): + all_int = False + break + type_map[col] = pl.Int64 if all_int else pl.Float64 + except Exception: + pass + + if type_map: + # Update schema dict to reflect restored types + for col, dtype in type_map.items(): + schema[col] = str(dtype) + # Apply casts to the LazyFrame + casts = [pl.col(col).cast(dtype, strict=False) for col, dtype in type_map.items()] + lf = lf.with_columns(casts) + logger.info('[BACKGROUND] Restored numeric types for %d columns: %s', + len(type_map), list(type_map.keys())[:10]) + # 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 @@ -676,6 +720,7 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith try: from analysis.tool_registry import _handle_run_clustering, _handle_extract_features + import polars as pl entry = store.get_dataset(entity_ds_id) if entry is None: @@ -689,7 +734,6 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith # ── Downsampling ── if head is not None and head > 0: lf = entry['lazyframe'] - import polars as pl try: row_count = lf.select(pl.len()).collect(streaming=True).item() except Exception: @@ -708,24 +752,31 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith entry = store.get_dataset(entity_ds_id) # refresh row_count = head - # Resolve feature columns - numeric_types = {'Int8', 'Int16', 'Int32', 'Int64', - 'UInt8', 'UInt16', 'UInt32', 'UInt64', - 'Float32', 'Float64'} + # Resolve feature columns — use actual LazyFrame dtypes (not stored schema dict + # which may be stale after Utf8 coercion in _background_process). + numeric_types = (pl.Int8, pl.Int16, pl.Int32, pl.Int64, + pl.UInt8, pl.UInt16, pl.UInt32, pl.UInt64, + pl.Float32, pl.Float64) + + lf = entry['lazyframe'] + live_schema = lf.collect_schema() + live_names = live_schema.names() + live_dtypes = list(live_schema.dtypes()) if feature_columns: - feature_cols = [c for c in feature_columns if c in schema] + feature_cols = [ + c for c in feature_columns + if c in live_names and live_dtypes[live_names.index(c)] in numeric_types + ] else: feature_cols = [ - c for c in schema.keys() - if not c.startswith('_') - and schema[c].split('(')[0].strip() in numeric_types + live_names[i] for i, dt in enumerate(live_dtypes) + if dt in numeric_types and not live_names[i].startswith('_') ][:10] - if not feature_cols: - raise Exception('No valid feature columns available for clustering') - - # ── Step 1: Clustering ────────────────────────────────────────────── + # ── Step 1: Clustering ──────────────────────────────────────────────── + # If no numeric columns detected, pass empty list to let + # _handle_run_clustering apply its Utf8→Float64 coercion fallback. run.status = 'clustering' run.progress_pct = 70 run.progress_msg = '正在进行聚类分析...' @@ -772,7 +823,6 @@ def _run_clustering_pipeline(run, store, entity_ds_id, feature_columns, algorith # ── Step 3: PCA-2D embedding (downsampled for large data) ───────── if run_pca: try: - import polars as pl lf = entry['lazyframe'] # Sample max 50K rows for PCA to avoid memory blowup on 10M entities MAX_PCA = 50000 diff --git a/scripts/gen_test_data.py b/scripts/gen_test_data.py index ba3841c..d39a6f4 100644 --- a/scripts/gen_test_data.py +++ b/scripts/gen_test_data.py @@ -1,11 +1,11 @@ """Generate synthetic TLS flow test data matching TlsDB.csv column spec. -Coverage rates simulate real TLS traffic: IP/port always present, -lat/lon/ISP often blank, TLS fingerprint fields sometimes blank, -obscure fields rarely present. +Outputs 1998 CSV files (0.csv .. 1997.csv) with realistic TLS flow data. +Columns match TlsDB.csv exactly; per-file column order is randomized. +Columns that are all-blank in a given file are auto-dropped from that file's header. Usage: - runtime\\python\\python.exe scripts\\gen_test_data.py [--rows N] + runtime\\python\\python.exe scripts\\gen_test_data.py [--files N] [--rows N] [--output-dir DIR] """ import random import csv @@ -15,14 +15,15 @@ from datetime import datetime, timedelta random.seed(42) -# All columns from TlsDB.csv in realistic order +# ── Column spec (exactly matches TlsDB.csv — 'row' removed because not in TlsDB) ── + COLUMNS = [ ':ips', ':ipd', ':prs', ':prd', 'scnt', 'dcnt', ':ips.latd', ':ips.lond', ':ipd.latd', ':ipd.lond', ':ips.ispn', ':ipd.ispn', ':ips.orgn', ':ipd.orgn', ':ips.city', ':ipd.city', ':ips.anon', ':ipd.anon', ':ips.doma', ':ipd.doma', - 'server-ip', 'client-ip', 'row', 'time', 'timestamp', + 'server-ip', 'client-ip', 'time', 'timestamp', '1ipp', '4dbn', 'tabl', '4ksz', 'cnrs', 'isrs', 'cnam', '0ver', 'snam', '4dur', '8seq', '2tmo', 'name', 'source-node', 'cipher-suite', 'ecdhe-named-curve', @@ -63,6 +64,7 @@ SERVICES = ['mail.example.com','api.example.com','cdn.example.com','auth.example SRC_NODES = ['packet_capture','ssl_logs','netflow','zeek','suricata'] NODE_NAMES = ['wan-link','core-02','gw-09','edge-01','backbone-03'] + def load_coverage(tlsdb_path=None): """Parse TlsDB.csv and return {column_name: coverage_float}. @@ -82,30 +84,40 @@ def load_coverage(tlsdb_path=None): coverage[name] = 1.0 return coverage + COVERAGE = load_coverage() def _rand_ip(pool): return random.choice(pool) + def _rand_hex(n): return ' '.join(f'{random.randint(0,255):02x}' for _ in range(n)) + def _maybe(rate): """Return True with given probability (0.0-1.0).""" return random.random() < rate + def _blank_or(rate, value, blank_marker=''): """Return value with rate probability, else blank_marker.""" return value if _maybe(rate) else blank_marker + def _blank_or_plus(rate, value): - """Return value with rate probability, else '+' (null marker). - Signature matches _blank_or(rate, value, blank_marker=...).""" + """Return value with rate probability, else '+' (null marker).""" return value if _maybe(rate) else '+' + def generate_row(row_id, base_time): - """Generate a single TLS flow row matching all TlsDB.csv columns.""" + """Generate a single TLS flow row matching all TlsDB.csv columns. + + row_id is a global sequential counter used to derive: + - time: base_time.strftime('%Y-%m-%d %H:%M:%S.') + f'{row_id % 1000000:06d}' + - timestamp: round(base_time.timestamp() + row_id * 0.000001, 6) + """ src_ip = _rand_ip(SRC_IPS) dst_ip = _rand_ip(DST_IPS) src_port = random.choice(PORTS) @@ -121,7 +133,11 @@ def generate_row(row_id, base_time): curve_name = random.choice(NAMED_CURVES_TEXT) service = random.choice(SERVICES) cert_cn = service if _maybe(0.7) else f'*.{random.choice(SERVICES).split(".")[-2:]}' - ts = int(base_time.timestamp()) + row_id + + # (a) time: YYYY-MM-DD HH:MM:SS.XXXXXX + time_str = base_time.strftime('%Y-%m-%d %H:%M:%S.') + f'{row_id % 1000000:06d}' + # (b) timestamp: float with 6 decimals + ts = round(base_time.timestamp() + row_id * 0.000001, 6) # Row-local lat/lon with realistic blank rates src_lat = round(random.uniform(-60, 60), 6) @@ -135,7 +151,7 @@ def generate_row(row_id, base_time): ':ipd': dst_ip, ':prs': src_port, ':prd': dst_port, - 'time': base_time.strftime('%H:%M:%S'), + 'time': time_str, 'timestamp': ts, # ── TLS fingerprint (very high coverage) ── @@ -164,12 +180,11 @@ def generate_row(row_id, base_time): # ── TCP / network info (medium coverage) ── '1ipp': str(random.randint(1, 50)), - 'row': _blank_or(1.0, ''), # TLS协议含义 - 'tabl': random.choice(['TlsC','TlsS']), # 100% coverage + 'tabl': random.choice(['TlsC', 'TlsS']), 'name': _blank_or(COVERAGE.get('name', 1.0), random.choice(NODE_NAMES)), 'source-node': _blank_or(COVERAGE.get('source-node', 1.0), random.choice(SRC_NODES)), - # ── Session resumption (boolean: coverage rate = probability of +) ── + # ── Session resumption (boolean) ── 'cnrs': _blank_or_plus(COVERAGE.get('cnrs', 1.0), '+'), 'isrs': _blank_or_plus(COVERAGE.get('isrs', 1.0), '+'), @@ -195,117 +210,82 @@ def generate_row(row_id, base_time): ':ips.doma': _blank_or(COVERAGE.get(':ips.doma', 1.0), random.choice(SERVICES)), ':ipd.doma': _blank_or(COVERAGE.get(':ipd.doma', 1.0), random.choice(SERVICES)), - # ── Two-letter abbreviation columns ── - 'scnt': _blank_or(COVERAGE.get('scnt', 1.0), src_country), - 'dcnt': _blank_or(COVERAGE.get('dcnt', 1.0), dst_country), - 'crcc': _blank_or(COVERAGE.get('crcc', 1.0), random.choice(['OK','ER','--'])), - '@iph': _blank_or(COVERAGE.get('@iph', 1.0), random.choice(['A','B','C','D'])), - 'eiph': _blank_or_plus(COVERAGE.get('eiph', 1.0), '+'), # boolean: + = True, + null marker = absent + # ── Two-letter / abbreviation columns ── + # (c) scnt/dcnt: .lower() → e.g. 'us', 'cn' + 'scnt': _blank_or(COVERAGE.get('scnt', 1.0), src_country.lower()), + 'dcnt': _blank_or(COVERAGE.get('dcnt', 1.0), dst_country.lower()), + 'crcc': _blank_or(COVERAGE.get('crcc', 1.0), random.choice(['OK', 'ER', '--'])), + '@iph': _blank_or(COVERAGE.get('@iph', 1.0), random.choice(['A', 'B', 'C', 'D'])), + 'eiph': _blank_or_plus(COVERAGE.get('eiph', 1.0), '+'), # ── Obscure / rarely populated fields ── 'orga': _blank_or(COVERAGE.get('orga', 1.0), random.choice(ORGS)), 'orgu': _blank_or(COVERAGE.get('orgu', 1.0), str(random.randint(1000, 99999))), - - # ── TLS protocol-level (always present) ── } def main(): - parser = argparse.ArgumentParser(description='Generate TLS flow test data matching TlsDB.csv') - parser.add_argument('--rows', type=int, default=10000, help='Number of rows') - parser.add_argument('--output', type=str, default='data/test_flows.csv') - parser.add_argument('--missing-cols', action='store_true', - help='Generate 2-3 files with 20-40%% columns randomly missing (varied per file)') - parser.add_argument('--missing-col-pct', type=str, default='20-40', - help='Range of columns to drop (e.g. "20-40" or "30")') + parser = argparse.ArgumentParser( + description='Generate TLS flow test data matching TlsDB.csv — ' + '1998 files (0.csv..1997.csv) with randomized column order per file' + ) + parser.add_argument('--files', type=int, default=1998, + help='Number of CSV files (default: 1998)') + parser.add_argument('--rows', type=int, default=100, + help='Rows per file (default: 100)') + parser.add_argument('--output-dir', type=str, default='data/test_csvs', + help='Output directory (default: data/test_csvs)') + parser.add_argument('--seed', type=int, default=42, + help='Random seed for data generation (default: 42)') args = parser.parse_args() - columns = COLUMNS - - output_path = Path(args.output) - output_path.parent.mkdir(parents=True, exist_ok=True) + random.seed(args.seed) + out_dir = Path(args.output_dir) + out_dir.mkdir(parents=True, exist_ok=True) base_time = datetime(2026, 6, 1, 0, 0, 0) - if args.missing_cols: - # Parse missing-col-pct range - _pct_str = args.missing_col_pct - if '-' in _pct_str: - _parts = _pct_str.split('-') - _min_pct = float(_parts[0]) / 100.0 - _max_pct = float(_parts[1]) / 100.0 - else: - _min_pct = _max_pct = float(_pct_str) / 100.0 + total_rows = 0 - # Columns that must NOT be dropped (IP/port identity columns) - _keep_cols = {':ips', ':ipd', ':prs', ':prd', 'server-ip', 'client-ip'} - _droppable = [c for c in columns if c not in _keep_cols] + for file_idx in range(args.files): + # File-specific RNG for column-order randomization (g) + file_seed = args.seed + file_idx + file_rng = random.Random(file_seed) - # Split rows into 2-3 files - total_rows = args.rows - num_files = random.randint(2, 3) - _split_sizes = [total_rows // num_files] * num_files - _split_sizes[-1] += total_rows % num_files # distribute remainder + # Generate all rows for this file + rows = [] + for row_i in range(args.rows): + row = generate_row(total_rows + row_i, base_time) + rows.append(row) - _row_offset = 0 - for _fi in range(num_files): - _n_rows = _split_sizes[_fi] - if _n_rows == 0: - continue + # (f) Auto-drop columns that are ALL blank ('' or '+') across every row + non_blank_cols = set() + for row in rows: + for col in COLUMNS: + val = row.get(col, '') + if val != '' and val != '+': + non_blank_cols.add(col) + file_columns = [c for c in COLUMNS if c in non_blank_cols] - # Randomly drop 20-40% of droppable columns for this file - _drop_ratio = random.uniform(_min_pct, _max_pct) - _n_drop = max(1, int(len(_droppable) * _drop_ratio)) - _dropped = set(random.sample(_droppable, _n_drop)) - _file_columns = [c for c in columns if c not in _dropped] + # (g) Randomize column order per file — high+low coverage interleaved + shuffled = file_columns[:] + file_rng.shuffle(shuffled) - # Output file name: insert _partN before .csv - _stem = output_path.stem - _out = output_path.parent / f'{_stem}_part{_fi + 1}.csv' + # (e) Write: 0.csv, 1.csv, ... 1997.csv + out_path = out_dir / f'{file_idx}.csv' + with open(out_path, 'w', newline='', encoding='utf-8') as f: + writer = csv.DictWriter(f, fieldnames=shuffled, extrasaction='ignore') + writer.writeheader() + for row in rows: + writer.writerow(row) - _blank_counts = {col: 0 for col in _file_columns} - with open(_out, 'w', newline='', encoding='utf-8') as f: - writer = csv.DictWriter(f, fieldnames=_file_columns, extrasaction='ignore') - writer.writeheader() - for row_i in range(_n_rows): - row_time = base_time + timedelta(seconds=random.randint(0, 86400 * 30)) - row = generate_row(_row_offset + row_i, row_time) - # Filter row to file columns - _filtered = {k: v for k, v in row.items() if k in _file_columns} - writer.writerow(_filtered) - for col in _file_columns: - if _filtered.get(col, '') == '' or _filtered.get(col, '') == '+': - _blank_counts[col] += 1 + total_rows += args.rows - _row_offset += _n_rows - print(f'Generated {_n_rows} rows → {_out} ({len(_file_columns)} columns, dropped {_n_drop}/{len(_droppable)} droppable)') - print(f' Kept: {sorted(_keep_cols & set(_file_columns))}') - print(f' Dropped: {sorted(_dropped)[:10]}...' if len(_dropped) > 10 else f' Dropped: {sorted(_dropped)}') + if (file_idx + 1) % 100 == 0 or file_idx == 0 or file_idx == args.files - 1: + print(f' [{file_idx + 1}/{args.files}] {total_rows} rows, ' + f'{len(shuffled)} cols, seed={file_seed}') - print(f'\nTotal: {total_rows} rows across {num_files} files with varied column sets.') - return - - # ── Standard mode (single file, all columns) ── - blank_counts = {col: 0 for col in columns} - total_rows = args.rows - - with open(output_path, 'w', newline='', encoding='utf-8') as f: - writer = csv.DictWriter(f, fieldnames=columns) - writer.writeheader() - for i in range(total_rows): - row_time = base_time + timedelta(seconds=random.randint(0, 86400 * 30)) - row = generate_row(i, row_time) - writer.writerow(row) - for col in columns: - if row.get(col, '') == '' or row.get(col, '') == '+': - blank_counts[col] += 1 - - print(f'Generated {total_rows} rows → {output_path}') - print('\nCoverage report (blank/null rate per column):') - for col in columns: - rate = blank_counts[col] / total_rows * 100 - bar_len = int((100 - rate) / 5) - bar = '#' * bar_len + '.' * (20 - bar_len) - print(f' {col:20s} {bar} {100-rate:5.1f}% present') + print(f'\nDone: {args.files} files × {args.rows} rows = {total_rows} total rows') + print(f'Output: {out_dir.resolve()}') if __name__ == '__main__': diff --git a/update.bat b/update.bat index 9d8e742..a7bd4f5 100644 --- a/update.bat +++ b/update.bat @@ -10,7 +10,7 @@ echo. :: Auto-detect update package set ZIP_FILE=%1 if "%ZIP_FILE%"=="" ( - for %%f in (tianxuan_update_*.zip) do set ZIP_FILE=%%f + for /f "delims=" %%f in ('dir /b tianxuan_update_*.zip 2^>nul') do set ZIP_FILE=%%f ) if "%ZIP_FILE%"=="" ( echo [ERROR] 未找到任何 tianxuan_update_*.zip 文件