HJQ/tianxuan
2
1
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

release: v1.1.6 — update.bat wildcard fix + schema column-order fix + gen_test format alignment + pipeline verify

Browse Source
This commit is contained in:
PM-pinou
2026-07-22 11:39:35 +08:00
parent db82f58ff2
commit 9ca5da1237
4 changed files with 152 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
VERSION
+1 -1
View File
@@ -1 +1 @@
v1.1.4
v1.1.6
analysis/views.py
+65 -15
View File
@@ -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
scripts/gen_test_data.py
+85 -105
View File
@@ -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__':
update.bat
+1 -1
View File
@@ -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 文件