脚本如何实现文件内容模糊颗粒度

本文目录导读：

1. 文本内容的模糊匹配（模糊搜索）
2. 文件内容的模糊处理（模糊化/脱敏）
3. 文件比较的模糊接口
4. 模糊搜索的索引实现
5. 批量文件处理框架

的模糊颗粒度处理,主要包含几种不同的方法和应用场景。

的模糊匹配（模糊搜索）

Python 实现 - 使用 fuzzywuzzy

from fuzzywuzzy import fuzz, process
def fuzzy_search_in_file(file_path, search_term, threshold=80):
    """
    在文件中进行模糊搜索
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        content = f.read()
        lines = content.split('\n')
    results = []
    for i, line in enumerate(lines, 1):
        # 计算相似度
        similarity = fuzz.partial_ratio(search_term.lower(), line.lower())
        if similarity >= threshold:
            results.append({
                'line': i,
                'content': line,
                'similarity': similarity
            })
    return sorted(results, key=lambda x: x['similarity'], reverse=True)
# 使用示例
file_path = 'data.txt'
search_term = 'example text'
matches = fuzzy_search_in_file(file_path, search_term, threshold=75)
for match in matches:
    print(f"行 {match['line']}: {match['content']} (相似度: {match['similarity']}%)")

更快的实现 - 使用 rapidfuzz

from rapidfuzz import fuzz, process
def fast_fuzzy_search(file_path, search_term, limit=5):
    """
    高性能模糊搜索
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        lines = [line.strip() for line in f if line.strip()]
    # 获取最佳匹配项
    results = process.extract(
        search_term, 
        lines, 
        scorer=fuzz.partial_ratio,
        limit=limit
    )
    return results
# 使用示例
results = fast_fuzzy_search('data.txt', 'search query', limit=10)
for text, score, index in results:
    print(f"匹配: {text[:50]}... (得分: {score})")

的模糊处理（模糊化/脱敏）

基于正则的模糊化

import re
import random
class ContentFuzzer:
    def __init__(self, blur_level=0.3):
        self.blur_level = blur_level  # 0-1 的模糊程度
    def blur_email(self, email):
        """模糊化邮箱"""
        local, domain = email.split('@')
        if len(local) > 2:
            visible = int(len(local) * (1 - self.blur_level))
            blurred = local[:visible] + '*' * (len(local) - visible)
            return f"{blurred}@{domain}"
        return email
    def blur_phone(self, phone):
        """模糊化手机号"""
        if len(phone) >= 7:
            visible = int(7 * (1 - self.blur_level))
            return phone[:visible] + '*' * (len(phone) - visible)
        return phone
    def blur_text(self, text):
        """模糊化文本中的敏感信息"""
        # 邮箱模糊化
        text = re.sub(r'\b[\w\.-]+@[\w\.-]+\.\w{2,4}\b', 
                      lambda m: self.blur_email(m.group()), text)
        # 手机号模糊化
        text = re.sub(r'\b1[3-9]\d{9}\b', 
                      lambda m: self.blur_phone(m.group()), text)
        # 身份证号模糊化
        text = re.sub(r'\b\d{18}\b|\b\d{17}X\b', 
                      lambda m: '*' * 18, text)
        return text
    def process_file(self, input_path, output_path):
        """处理文件"""
        with open(input_path, 'r', encoding='utf-8') as f:
            content = f.read()
        blurred_content = self.blur_text(content)
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(blurred_content)
# 使用示例
fuzzer = ContentFuzzer(blur_level=0.5)
fuzzer.process_file('original.txt', 'blurred.txt')

文件比较的模糊接口

相似度计算的实现

import difflib
import hashlib
class FileSimilarity:
    @staticmethod
    def content_similarity(file1_path, file2_path):
        """计算两个文件的相似度"""
        with open(file1_path, 'r', encoding='utf-8') as f1, \
             open(file2_path, 'r', encoding='utf-8') as f2:
            content1 = f1.read()
            content2 = f2.read()
        # 使用 SequenceMatcher
        matcher = difflib.SequenceMatcher(None, content1, content2)
        similarity_ratio = matcher.ratio()
        # 使用最长公共子序列
        lcs_length = len(matcher.find_longest_match(0, len(content1), 
                                                      0, len(content2)))
        lcs_similarity = 2 * lcs_length / (len(content1) + len(content2))
        return {
            'ratio_similarity': similarity_ratio,
            'lcs_similarity': lcs_similarity,
            'hash_similarity': FileSimilarity.hash_similarity(content1, content2)
        }
    @staticmethod
    def hash_similarity(content1, content2):
        """基于哈希的相似度"""
        # 分块哈希比较
        chunk_size = 64
        chunks1 = set()
        chunks2 = set()
        for i in range(0, len(content1), chunk_size):
            chunk = content1[i:i+chunk_size]
            chunks1.add(hashlib.md5(chunk.encode()).hexdigest())
        for i in range(0, len(content2), chunk_size):
            chunk = content2[i:i+chunk_size]
            chunks2.add(hashlib.md5(chunk.encode()).hexdigest())
        # 计算 Jaccard 相似度
        intersection = chunks1 & chunks2
        union = chunks1 | chunks2
        return len(intersection) / len(union) if union else 1.0
# 使用示例
result = FileSimilarity.content_similarity('file1.txt', 'file2.txt')
print(f"序列相似度: {result['ratio_similarity']:.2%}")
print(f"LCS相似度: {result['lcs_similarity']:.2%}")
print(f"哈希相似度: {result['hash_similarity']:.2%}")

模糊搜索的索引实现

基于倒排索引的模糊搜索

from collections import defaultdict
import re
class FuzzyIndex:
    def __init__(self, ngram_size=2):
        self.ngram_size = ngram_size
        self.index = defaultdict(set)
    def _get_ngrams(self, word):
        """生成 n-gram"""
        word = f"#{word}#"  # 添加边界标记
        ngrams = []
        for i in range(len(word) - self.ngram_size + 1):
            ngrams.append(word[i:i+self.ngram_size])
        return ngrams
    def build_index(self, file_path):
        """构建模糊搜索索引"""
        with open(file_path, 'r', encoding='utf-8') as f:
            for line_num, line in enumerate(f, 1):
                words = re.findall(r'\w+', line.lower())
                for word in words:
                    ngrams = self._get_ngrams(word)
                    for ngram in ngrams:
                        self.index[ngram].add((line_num, word))
    def search(self, query, threshold=0.5):
        """模糊搜索"""
        query_ngrams = self._get_ngrams(query.lower())
        matches = defaultdict(int)
        # 统计共享 n-gram 的数量
        for ngram in query_ngrams:
            if ngram in self.index:
                for line_num, word in self.index[ngram]:
                    matches[(line_num, word)] += 1
        # 计算相似度
        results = []
        for (line_num, word), shared_ngrams in matches.items():
            # Jaccard 相似度
            total_ngrams = len(query_ngrams) + len(self._get_ngrams(word))
            similarity = 2 * shared_ngrams / total_ngrams
            if similarity >= threshold:
                results.append({
                    'line': line_num,
                    'word': word,
                    'similarity': similarity
                })
        return sorted(results, key=lambda x: x['similarity'], reverse=True)
# 使用示例
index = FuzzyIndex(ngram_size=3)
index.build_index('document.txt')
results = index.search('example', threshold=0.3)
for r in results:
    print(f"行 {r['line']}: {r['word']} (相似度: {r['similarity']:.2%})")

批量文件处理框架

import os
from concurrent.futures import ThreadPoolExecutor
class FileFuzzyProcessor:
    def __init__(self, blur_level=0.3, thread_count=4):
        self.blur_level = blur_level
        self.thread_count = thread_count
    def process_single_file(self, input_path, output_dir):
        """处理单个文件"""
        # 创建输出目录
        os.makedirs(output_dir, exist_ok=True)
        filename = os.path.basename(input_path)
        output_path = os.path.join(output_dir, f"blurred_{filename}")
        # 读取和模糊处理
        with open(input_path, 'r', encoding='utf-8') as f:
            content = f.read()
        blurred_content = self.blur_content(content)
        # 保存
        with open(output_path, 'w', encoding='utf-8') as f:
            f.write(blurred_content)
        return output_path
    def blur_content(self, content):
        """模糊内容的具体实现"""
        # 随机替换一些字符
        if random.random() < self.blur_level:
            content = list(content)
            positions = random.sample(range(len(content)), 
                                    int(len(content) * 0.1 * self.blur_level))
            for pos in positions:
                if content[pos].isalnum():
                    content[pos] = 'X'
            content = ''.join(content)
        return content
    def process_directory(self, input_dir, output_dir):
        """处理整个目录"""
        files = []
        for root, _, filenames in os.walk(input_dir):
            for filename in filenames:
                if filename.endswith('.txt'):
                    files.append(os.path.join(root, filename))
        # 多线程处理
        with ThreadPoolExecutor(max_workers=self.thread_count) as executor:
            futures = []
            for file_path in files:
                future = executor.submit(self.process_single_file, 
                                       file_path, output_dir)
                futures.append(future)
            results = []
            for future in futures:
                results.append(future.result())
        return results
# 使用示例
processor = FileFuzzyProcessor(blur_level=0.3, thread_count=4)
results = processor.process_directory('input_folder', 'output_folder')
print(f"处理了 {len(results)} 个文件")

选择合适的模糊技术取决于具体需求：精确度、性能、内存使用和实时性要求。