本文目录导读：
模糊粗糙增量学习的方法和脚本思路。
基础哈希比较法（最简单）
import hashlib
import json
import os
class RoughHashLearner:
def __init__(self, state_file="learning_state.json"):
self.state_file = state_file
self.learned_hashes = self.load_state()
def load_state(self):
if os.path.exists(self.state_file):
with open(self.state_file, 'r') as f:
return json.load(f)
return {}
def save_state(self):
with open(self.state_file, 'w') as f:
json.dump(self.learned_hashes, f)
def get_fuzzy_hash(self, content):
# 模糊哈希：按块计算，只保留关键特征
chunks = [content[i:i+64] for i in range(0, len(content), 64)]
chunk_hashes = []
for chunk in chunks[:10]: # 只取前10个块
chunk_hashes.append(hashlib.md5(chunk.encode()).hexdigest()[:8])
return ':'.join(chunk_hashes)
def learn(self, filepath):
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
fuzzy_hash = self.get_fuzzy_hash(content)
filename = os.path.basename(filepath)
if filename not in self.learned_hashes:
self.learned_hashes[filename] = []
# 检查是否已学习过相似内容
for existing_hash in self.learned_hashes[filename]:
if self.similarity(fuzzy_hash, existing_hash) > 0.7:
return False, "Similar content already learned"
self.learned_hashes[filename].append({
'hash': fuzzy_hash,
'size': len(content),
'timestamp': os.path.getmtime(filepath)
})
self.save_state()
return True, "New content learned"
def similarity(self, hash1, hash2):
parts1 = hash1.split(':')
parts2 = hash2.split(':')
matches = sum(1 for p1, p2 in zip(parts1, parts2) if p1 == p2)
return matches / max(len(parts1), len(parts2))
滑动窗口指纹法
class SlidingWindowLearner:
def __init__(self, window_size=40, threshold=3):
self.window_size = window_size
self.threshold = threshold
self.fingerprints = set()
def extract_fingerprints(self, content):
fingerprints = set()
for i in range(len(content) - self.window_size + 1):
window = content[i:i+self.window_size]
# 使用滚动哈希或简单哈希
fp = hash(window) & 0xFFFFFFFF # 取低32位
# 采样：只取特定模式的指纹
if fp % 100 < 10: # 约10%的采样率
fingerprints.add(fp)
return fingerprints
def is_new_content(self, content):
new_fps = self.extract_fingerprints(content)
common = new_fps & self.fingerprints
similarity = len(common) / max(len(new_fps), 1)
return similarity < 0.3 # 相似度低于30%认为是新内容
def learn(self, filepath):
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
if self.is_new_content(content):
new_fps = self.extract_fingerprints(content)
self.fingerprints.update(new_fps)
return True, "New content patterns learned"
return False, "Similar patterns already exist"
关键词权重法
from collections import Counter
import re
class KeywordWeightLearner:
def __init__(self):
self.keyword_weights = {} # keyword -> weight
self.total_tokens = 0
def extract_keywords(self, content):
# 分词并提取关键词（这里简单处理，实际可用jieba等）
words = re.findall(r'\b[a-zA-Z]{3,}\b', content.lower())
return Counter(words)
def learn(self, filepath):
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
keywords = self.extract_keywords(content)
total_new = len(keywords)
if total_new < 10:
return False, "File too short to extract meaningful patterns"
# 更新关键词权重
for word, count in keywords.items():
if word not in self.keyword_weights:
self.keyword_weights[word] = 0
self.keyword_weights[word] += count
self.total_tokens += sum(keywords.values())
return True, f"Learned {total_new} keywords"
def get_similarity_score(self, content1, content2):
keywords1 = self.extract_keywords(content1)
keywords2 = self.extract_keywords(content2)
# 计算余弦相似度
common = set(keywords1.keys()) & set(keywords2.keys())
if not common:
return 0
dot_product = sum(keywords1[w] * keywords2[w] for w in common)
norm1 = sum(k**2 for k in keywords1.values()) ** 0.5
norm2 = sum(k**2 for k in keywords2.values()) ** 0.5
return dot_product / (norm1 * norm2) if norm1 * norm2 > 0 else 0
简单朴素贝叶斯学习器
import math
from collections import defaultdict
class NaiveBayesLearner:
def __init__(self, smooth_factor=1.0):
self.class_counts = defaultdict(int)
self.feature_counts = defaultdict(lambda: defaultdict(int))
self.total_docs = 0
self.smooth_factor = smooth_factor
def extract_features(self, content):
# 使用简单的n-gram作为特征
words = re.findall(r'\b\w+\b', content.lower())
features = set()
for i in range(len(words) - 2):
features.add(' '.join(words[i:i+3])) # trigram
return features
def learn(self, filepath, label="default"):
with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
content = f.read()
features = self.extract_features(content)
self.class_counts[label] += 1
self.total_docs += 1
for feature in features:
self.feature_counts[label][feature] += 1
return True, f"Learned {len(features)} features"
def classify(self, content):
features = self.extract_features(content)
scores = {}
for label in self.class_counts:
prior = math.log(self.class_counts[label] / self.total_docs)
likelihood = 0
for feature in features:
count = self.feature_counts[label].get(feature, 0)
total = sum(self.feature_counts[label].values())
# 使用拉普拉斯平滑
prob = (count + self.smooth_factor) / (total + self.smooth_factor * len(self.feature_counts))
likelihood += math.log(prob)
scores[label] = prior + likelihood
return max(scores, key=scores.get)
综合使用示例
def main():
# 初始化各种学习器
hash_learner = RoughHashLearner()
window_learner = SlidingWindowLearner()
keyword_learner = KeywordWeightLearner()
bayes_learner = NaiveBayesLearner()
# 批量学习文件
import glob
files = glob.glob("data/*.txt")
for filepath in files[:10]: # 学习前10个文件
print(f"\n学习文件: {filepath}")
# 多种学习方法
success1, msg1 = hash_learner.learn(filepath)
success2, msg2 = window_learner.learn(filepath)
success3, msg3 = keyword_learner.learn(filepath)
success4, msg4 = bayes_learner.learn(filepath)
print(f" 哈希法: {msg1}")
print(f" 滑动窗口法: {msg2}")
print(f" 关键词权重法: {msg3}")
print(f" 朴素贝叶斯: {msg4}")
# 新增文件检测
if len(files) > 10:
new_file = files[-1]
print(f"\n检测新文件: {new_file}")
with open(new_file, 'r', encoding='utf-8', errors='ignore') as f:
new_content = f.read()
score = keyword_learner.get_similarity_score(
new_content,
open(filepath).read()
)
print(f" 与已学习文件的相似度: {score:.2%}")
if __name__ == "__main__":
main()
使用建议
-
选择合适方法：
- 用关键词权重法
- 二进制文件：用哈希比较法
- 代码文件：用滑动窗口法
- 分类任务：用朴素贝叶斯
-
优化存储：
- 使用数据库替代JSON存储状态
- 定期清理过期学习数据
-
性能优化：
- 对大文件进行采样处理
- 采用异步处理机制
这些方法都实现了"模糊粗糙"的增量学习,可以根据实际场景调整参数和组合使用。
