本文目录导读：
我来帮你实现一个基于XGBoost的文件内容模糊分类/分析的脚本，这里提供几个不同场景的示例：
分类脚本（中文文本）
import os
import jieba
import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.model_selection import train_test_split
import xgboost as xgb
from sklearn.metrics import classification_report
class FileContentClassifier:
def __init__(self):
self.vectorizer = TfidfVectorizer(max_features=5000,
tokenizer=self.tokenize,
analyzer='word')
self.model = xgb.XGBClassifier(
n_estimators=100,
max_depth=6,
learning_rate=0.1,
objective='multi:softprob',
random_state=42
)
def tokenize(self, text):
"""中文分词函数"""
return ' '.join(jieba.cut(text))
def load_files(self, directory):
"""加载目录下的文件"""
data = []
labels = []
for category in os.listdir(directory):
category_path = os.path.join(directory, category)
if os.path.isdir(category_path):
for filename in os.listdir(category_path):
file_path = os.path.join(category_path, filename)
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
data.append(content)
labels.append(category)
except:
print(f"无法读取文件: {file_path}")
return data, labels
def train(self, X, y):
"""训练模型"""
# 特征提取
X_features = self.vectorizer.fit_transform(X)
# 分割数据集
X_train, X_test, y_train, y_test = train_test_split(
X_features, y, test_size=0.2, random_state=42
)
# 训练模型
self.model.fit(X_train, y_train)
# 评估
y_pred = self.model.predict(X_test)
print(classification_report(y_test, y_pred))
def predict(self, text):
"""预测单个文本"""
features = self.vectorizer.transform([text])
return self.model.predict(features)[0]
# 使用示例
if __name__ == "__main__":
classifier = FileContentClassifier()
# 假设有分类好的文件目录
# classifier.load_files("./documents")
# 测试数据
sample_texts = [
"这是一篇关于机器学习的文章，讨论了神经网络和深度学习",
"今天天气真好，我们去公园散步吧"
]
sample_labels = ["技术", "生活"]
# 训练
classifier.train(sample_texts, sample_labels)
# 预测新内容
new_text = "人工智能正在改变世界"
result = classifier.predict(new_text)
print(f"预测结果: {result}")
模糊文件匹配和相似度分析
import os
import numpy as np
import xgboost as xgb
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import jieba
class FuzzyFileMatcher:
def __init__(self):
self.vectorizer = TfidfVectorizer(
tokenizer=self.tokenize,
analyzer='word',
max_features=1000,
min_df=2,
max_df=0.8,
sublinear_tf=True
)
self.file_features = None
self.file_contents = {}
self.anomaly_model = None
def tokenize(self, text):
"""分词处理"""
words = jieba.cut(text)
return ' '.join([w for w in words if len(w) > 1])
def load_and_vectorize(self, file_paths):
"""加载并向量化文件"""
contents = []
for path in file_paths:
try:
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
self.file_contents[path] = content
contents.append(content)
except:
print(f"无法读取: {path}")
# 向量化
self.file_features = self.vectorizer.fit_transform(contents)
return self.file_features
def find_similar_files(self, query_text, top_k=5):
"""查找相似文件（模糊匹配）"""
if self.file_features is None:
raise ValueError("请先加载文件")
# 向量化查询文本
query_vector = self.vectorizer.transform([query_text])
# 计算相似度
similarities = cosine_similarity(query_vector, self.file_features)[0]
# 获取最相似的文件
top_indices = np.argsort(similarities)[-top_k:][::-1]
results = []
for idx in top_indices:
if similarities[idx] > 0.1: # 相似度阈值
file_path = list(self.file_contents.keys())[idx]
results.append({
'file': file_path,
'similarity': similarities[idx],
'preview': self.file_contents[file_path][:100] + '...'
})
return results
def train_anomaly_detector(self, normal_texts, abnormal_texts):
"""训练异常检测模型"""
# 准备数据
all_texts = normal_texts + abnormal_texts
features = self.vectorizer.fit_transform(all_texts)
# 标签：正常=0，异常=1
y = [0] * len(normal_texts) + [1] * len(abnormal_texts)
# 训练XGBoost
self.anomaly_model = xgb.XGBClassifier(
n_estimators=50,
max_depth=4,
scale_pos_weight=len(normal_texts)/len(abnormal_texts),
random_state=42
)
self.anomaly_model.fit(features, y)
def detect_anomaly(self, text):
"""检测文本是否异常"""
if self.anomaly_model is None:
raise ValueError("请先训练异常检测器")
features = self.vectorizer.transform([text])
prediction = self.anomaly_model.predict(features)[0]
probability = self.anomaly_model.predict_proba(features)[0]
return {
'is_anomaly': bool(prediction),
'probability_anomaly': probability[1],
'probability_normal': probability[0]
}
# 使用示例
matcher = FuzzyFileMatcher()
# 加载文件
file_list = ["file1.txt", "file2.txt", "file3.txt"]
# matcher.load_and_vectorize(file_list)
# 查找相似内容
# query = "机器学习算法"
# results = matcher.find_similar_files(query)
# for r in results:
# print(f"文件: {r['file']}, 相似度: {r['similarity']:.2f}")
# 训练异常检测
normal_texts = ["正常文件内容", "日常文本"]
abnormal_texts = ["异常内容", "包含敏感词的内容"]
matcher.train_anomaly_detector(normal_texts, abnormal_texts)
# 检测新文本
test_text = "这是一段测试文本"
result = matcher.detect_anomaly(test_text)
print(f"是否异常: {result['is_anomaly']}")
print(f"异常概率: {result['probability_anomaly']:.2f}")
文件特征自动提取和分析
import os
import re
import hashlib
import numpy as np
import xgboost as xgb
from collections import Counter
import json
class FileFeatureAnalyzer:
def __init__(self):
self.model = None
self.feature_names = []
def extract_features(self, file_path):
"""提取文件特征"""
try:
with open(file_path, 'rb') as f:
content = f.read()
text = content.decode('utf-8', errors='ignore')
except:
return None
features = {}
# 基础统计特征
features['file_size'] = os.path.getsize(file_path)
features['char_count'] = len(text)
features['word_count'] = len(text.split())
features['line_count'] = text.count('\n')
# 文本复杂度特征
features['unique_words_ratio'] = len(set(text.split())) / max(len(text.split()), 1)
features['avg_word_length'] = np.mean([len(w) for w in text.split()]) if text.split() else 0
# 特殊字符统计
features['digit_ratio'] = sum(c.isdigit() for c in text) / max(len(text), 1)
features['punctuation_ratio'] = sum(c in '.,!?;:()[]{}""'' for c in text) / max(len(text), 1)
# 关键词特征（模糊匹配）
keywords_tech = ['算法', '数据', 'AI', '机器学习', '深度学习', '神经网络']
keywords_business = ['市场', '销售', '客户', '利润', '营收']
features['tech_keyword_count'] = sum(keyword in text for keyword in keywords_tech)
features['business_keyword_count'] = sum(keyword in text for keyword in keywords_business)
# 重复内容特征（模糊检测）
lines = text.split('\n')
line_freq = Counter(lines)
features['repeat_line_ratio'] = sum(1 for v in line_freq.values() if v > 1) / max(len(lines), 1)
# 熵值（信息量度量）
char_freq = Counter(text)
entropy = -sum((freq/len(text)) * np.log2(freq/len(text)) for freq in char_freq.values())
features['entropy'] = entropy
return features
def batch_extract_features(self, file_paths):
"""批量提取特征"""
all_features = []
valid_paths = []
for path in file_paths:
features = self.extract_features(path)
if features is not None:
all_features.append(features)
valid_paths.append(path)
self.feature_names = list(features.keys())
return np.array([[f[name] for name in self.feature_names] for f in all_features]), valid_paths
def train_vague_classifier(self, X, y):
"""训练模糊分类器"""
self.model = xgb.XGBClassifier(
n_estimators=50,
max_depth=3,
subsample=0.8,
colsample_bytree=0.8,
random_state=42
)
self.model.fit(X, y)
# 输出特征重要性
importance = self.model.feature_importances_
for name, imp in zip(self.feature_names, importance):
print(f"{name}: {imp:.3f}")
def vague_predict(self, features):
"""模糊预测"""
if self.model is None:
raise ValueError("请先训练模型")
prediction = self.model.predict(features.reshape(1, -1))[0]
probabilities = self.model.predict_proba(features.reshape(1, -1))[0]
return {
'prediction': prediction,
'confidence': max(probabilities),
'probability_distribution': list(probabilities)
}
# 使用示例
analyzer = FileFeatureAnalyzer()
# 提取特征
# sample_files = ["doc1.txt", "doc2.txt"]
# features, paths = analyzer.batch_extract_features(sample_files)
# 准备训练数据
# X_train = features
# y_train = [0, 1] # 0: 普通文件, 1: 重要文件
# 训练
# analyzer.train_vague_classifier(X_train, y_train)
# 预测新文件
# new_file = "new_doc.txt"
# new_features, _ = analyzer.batch_extract_features([new_file])
# result = analyzer.vague_predict(new_features[0])
# print(f"预测结果: {result}")
安装依赖
pip install xgboost scikit-learn pandas numpy jieba
使用建议
- 数据准备：收集足够多的标注数据
- 特征工程：根据业务需求调整特征
- 模型调优：调整XGBoost参数提高性能
- 模糊处理：设置合理的相似度阈值
- 增量学习：定期更新模型
这些脚本实现了文件内容的模糊分类、相似度匹配和特征分析，可以根据具体需求进行调整和优化。
