#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 数据清洗脚本 - 用于清洗和预处理网络安全训练数据 适用于 Qwen3.5-0.8B-HOS 训练流程 功能： 1. 去重（MinHash + SimHash） 2. 去噪（移除低质量数据） 3. 格式标准化（统一为 Alpaca 格式） 4. 隐私脱敏（移除敏感信息） """ import json import os import re import hashlib from typing import Dict, List, Optional from datasketch import MinHash, MinHashLSH from simhash import Simhash # 配置 DATA_DIR = "datasets" OUTPUT_DIR = "datasets/cleaned" def compute_minhash(text: str, num_perm: int = 128) -> MinHash: """计算文本的 MinHash 签名""" m = MinHash(num_perm=num_perm) for word in text.split(): m.update(word.encode('utf-8')) return m def compute_simhash(text: str, hashbits: int = 64) -> Simhash: """计算文本的 SimHash 指纹""" return Simhash(text, hashbits=hashbits) def deduplicate_minhash(data: List[Dict], threshold: float = 0.8) -> List[Dict]: """使用 MinHash 进行去重""" print(f" MinHash 去重 (阈值: {threshold})...") lsh = MinHashLSH(threshold=threshold, num_perm=128) unique_data = [] duplicates = 0 for idx, item in enumerate(data): text = f"{item.get('instruction', '')} {item.get('input', '')} {item.get('output', '')}" m = compute_minhash(text) key = f"doc_{idx}" if not lsh.query(m): lsh.insert(key, m) unique_data.append(item) else: duplicates += 1 print(f" 原始: {len(data)}, 去重后: {len(unique_data)}, 移除: {duplicates}") return unique_data def deduplicate_simhash(data: List[Dict], threshold: int = 3) -> List[Dict]: """使用 SimHash 进行去重""" print(f" SimHash 去重 (汉明距离阈值: {threshold})...") unique_data = [] seen_hashes = [] duplicates = 0 for item in data: text = f"{item.get('instruction', '')} {item.get('input', '')} {item.get('output', '')}" sh = compute_simhash(text) is_duplicate = False for seen_sh in seen_hashes: if sh.distance(seen_sh) <= threshold: is_duplicate = True duplicates += 1 break if not is_duplicate: seen_hashes.append(sh) unique_data.append(item) print(f" 原始: {len(data)}, 去重后: {len(unique_data)}, 移除: {duplicates}") return unique_data def remove_noise(data: List[Dict]) -> List[Dict]: """去噪 - 移除低质量数据""" print(" 去噪处理...") clean_data = [] removed = 0 for item in data: instruction = item.get('instruction', '') input_text = item.get('input', '') output = item.get('output', '') # 过滤条件 # 1. 移除空内容 if not instruction or not output: removed += 1 continue # 2. 移除过短的内容 if len(output.strip()) < 10: removed += 1 continue # 3. 移除明显无关的内容 noise_patterns = [ r'^\s*$', # 纯空白 r'^(test|测试|demo|示例)\s*$', # 测试内容 r'^(TODO|FIXME|HACK)', # 代码注释 ] is_noise = False for pattern in noise_patterns: if re.search(pattern, output, re.IGNORECASE): is_noise = True break if is_noise: removed += 1 continue clean_data.append(item) print(f" 原始: {len(data)}, 去噪后: {len(clean_data)}, 移除: {removed}") return clean_data def standardize_format(data: List[Dict]) -> List[Dict]: """格式标准化 - 统一为 Alpaca 格式""" print(" 格式标准化...") standardized = [] for item in data: # 确保必要的字段存在 standardized_item = { 'instruction': item.get('instruction', '').strip(), 'input': item.get('input', '').strip(), 'output': item.get('output', '').strip() } # 清理多余空白 for key in standardized_item: standardized_item[key] = re.sub(r'\s+', ' ', standardized_item[key]).strip() standardized.append(standardized_item) print(f" 标准化完成: {len(standardized)} 条记录") return standardized def desensitize(data: List[Dict]) -> List[Dict]: """隐私脱敏 - 移除敏感信息""" print(" 隐私脱敏...") desensitized = [] # 敏感信息正则表达式 sensitive_patterns = { 'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', 'phone': r'\b1[3-9]\d{9}\b', # 中国大陆手机号 'id_card': r'\b\d{17}[\dXx]\b', # 身份证号 'ip': r'\b(?:\d{1,3}\.){3}\d{1,3}\b', # IP 地址 'password': r'(password|密码|pwd)\s*[:=]\s*\S+', # 密码 'api_key': r'(api[_-]?key|apikey)\s*[:=]\s*[\w-]+', # API Key 'token': r'(token|access[_-]?token)\s*[:=]\s*[\w.-]+', # Token } for item in data: new_item = item.copy() for key in ['instruction', 'input', 'output']: text = new_item.get(key, '') # 替换敏感信息 for pattern_name, pattern in sensitive_patterns.items(): if pattern_name in ['email', 'phone', 'id_card']: text = re.sub(pattern, f'<{pattern_name.upper()}>', text) elif pattern_name in ['password', 'api_key', 'token']: text = re.sub(pattern, f'<{pattern_name.upper()}>', text, flags=re.IGNORECASE) new_item[key] = text desensitized.append(new_item) print(f" 脱敏完成: {len(desensitized)} 条记录") return desensitized def process_dataset(input_file: str, output_file: str): """处理单个数据集文件""" print(f"\n处理: {input_file}") # 读取数据 data = [] with open(input_file, 'r', encoding='utf-8') as f: for line in f: if line.strip(): data.append(json.loads(line)) print(f" 原始数据: {len(data)} 条") # 1. 去重 (MinHash) data = deduplicate_minhash(data, threshold=0.8) # 2. 去重 (SimHash) data = deduplicate_simhash(data, threshold=3) # 3. 去噪 data = remove_noise(data) # 4. 格式标准化 data = standardize_format(data) # 5. 隐私脱敏 data = desensitize(data) # 保存结果 os.makedirs(os.path.dirname(output_file), exist_ok=True) with open(output_file, 'w', encoding='utf-8') as f: for item in data: f.write(json.dumps(item, ensure_ascii=False) + '\n') print(f" 处理后数据: {len(data)} 条") print(f" 保存至: {output_file}") def main(): """主函数""" print("============================================") print(" 数据清洗脚本") print(" Qwen3.5-0.8B-HOS") print("============================================") # 创建输出目录 os.makedirs(OUTPUT_DIR, exist_ok=True) # 处理所有数据集 datasets = [ 'cybersecurity_hq.jsonl', 'cybersecurity_sharegpt.jsonl', 'cnvd_vulnerability.jsonl', 'merged_cybersec.jsonl' ] for dataset in datasets: input_file = os.path.join(DATA_DIR, dataset) output_file = os.path.join(OUTPUT_DIR, dataset) if os.path.exists(input_file): process_dataset(input_file, output_file) else: print(f"\n跳过: {input_file} (文件不存在)") print("\n============================================") print(" 数据清洗完成!") print(f" 输出目录: {OUTPUT_DIR}") print("============================================") if __name__ == "__main__": main()