脚本如何实现文件内容模糊粗糙增量学习

本文目录导读：

1. 基础哈希比较法（最简单）
2. 滑动窗口指纹法
3. 关键词权重法
4. 简单朴素贝叶斯学习器
5. 综合使用示例
6. 使用建议

模糊粗糙增量学习的方法和脚本思路。

基础哈希比较法（最简单）

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()

使用建议

1. 选择合适方法：

   • 用关键词权重法
   • 二进制文件：用哈希比较法
   • 代码文件：用滑动窗口法
   • 分类任务：用朴素贝叶斯

2. 优化存储：

   • 使用数据库替代JSON存储状态
   • 定期清理过期学习数据

3. 性能优化：

   • 对大文件进行采样处理
   • 采用异步处理机制

这些方法都实现了"模糊粗糙"的增量学习,可以根据实际场景调整参数和组合使用。