脚本如何实现文件内容相关性分析

本文目录导读：

1. 基于词频的相似度分析
2. 基于N-gram的相似度分析
3. 主题模型分析（LDA）
4. 快速文件对比（命令行工具）
5. 完整分析工具
6. 使用建议

基于词频的相似度分析

Python实现（TF-IDF + 余弦相似度）

import os
import re
from collections import Counter
from math import log, sqrt
import json
class FileSimilarityAnalyzer:
    def __init__(self):
        self.documents = {}
        self.tfidf_matrix = {}
        self.idf = {}
    def preprocess_text(self, text):
        """文本预处理"""
        # 转小写、去标点
        text = text.lower()
        text = re.sub(r'[^\w\s]', '', text)
        # 分词
        words = text.split()
        # 去除停用词（可根据需要扩展）
        stop_words = {'the', 'a', 'an', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for'}
        return [word for word in words if word not in stop_words and len(word) > 2]
    def load_files(self, file_paths):
        """加载文件"""
        for file_path in file_paths:
            try:
                with open(file_path, 'r', encoding='utf-8') as f:
                    content = f.read()
                filename = os.path.basename(file_path)
                self.documents[filename] = self.preprocess_text(content)
                print(f"已加载: {filename}")
            except Exception as e:
                print(f"加载失败 {file_path}: {e}")
    def compute_idf(self):
        """计算IDF值"""
        total_docs = len(self.documents)
        # 统计每个词出现在多少文档中
        word_doc_count = Counter()
        for words in self.documents.values():
            unique_words = set(words)
            word_doc_count.update(unique_words)
        # 计算IDF
        self.idf = {
            word: log((total_docs + 1) / (count + 1)) + 1
            for word, count in word_doc_count.items()
        }
    def compute_tf(self, words):
        """计算TF值"""
        word_count = Counter(words)
        total_words = len(words)
        return {word: count / total_words for word, count in word_count.items()}
    def compute_tfidf(self, filename):
        """计算TF-IDF向量"""
        words = self.documents[filename]
        tf = self.compute_tf(words)
        tfidf = {}
        for word, tf_value in tf.items():
            if word in self.idf:
                tfidf[word] = tf_value * self.idf[word]
        return tfidf
    def cosine_similarity(self, vec1, vec2):
        """计算余弦相似度"""
        # 找出共同词汇
        common_words = set(vec1.keys()) & set(vec2.keys())
        if not common_words:
            return 0.0
        # 计算点积
        dot_product = sum(vec1[word] * vec2[word] for word in common_words)
        # 计算向量模长
        norm1 = sqrt(sum(v ** 2 for v in vec1.values()))
        norm2 = sqrt(sum(v ** 2 for v in vec2.values()))
        if norm1 == 0 or norm2 == 0:
            return 0.0
        return dot_product / (norm1 * norm2)
    def analyze_similarity(self):
        """分析所有文件之间的相似度"""
        # 计算所有文档的TF-IDF
        for filename in self.documents:
            self.tfidf_matrix[filename] = self.compute_tfidf(filename)
        # 计算相似度矩阵
        filenames = list(self.documents.keys())
        similarity_matrix = {}
        for i in range(len(filenames)):
            for j in range(i + 1, len(filenames)):
                similarity = self.cosine_similarity(
                    self.tfidf_matrix[filenames[i]],
                    self.tfidf_matrix[filenames[j]]
                )
                pair = (filenames[i], filenames[j])
                similarity_matrix[pair] = similarity
        return similarity_matrix
# 使用示例
analyzer = FileSimilarityAnalyzer()
analyzer.load_files(['file1.txt', 'file2.txt', 'file3.txt'])
analyzer.compute_idf()
results = analyzer.analyze_similarity()
# 输出结果
print("\n文件相似度分析结果：")
for (file1, file2), similarity in sorted(results.items(), key=lambda x: x[1], reverse=True):
    print(f"{file1} <-> {file2}: {similarity:.4f}")

基于N-gram的相似度分析

class NGramSimilarityAnalyzer:
    def __init__(self, n=2):
        self.n = n
    def extract_ngrams(self, text):
        """提取N-gram"""
        text = text.lower()
        ngrams = []
        for i in range(len(text) - self.n + 1):
            ngrams.append(text[i:i + self.n])
        return ngrams
    def jaccard_similarity(self, set1, set2):
        """计算Jaccard相似度"""
        intersection = len(set1 & set2)
        union = len(set1 | set2)
        return intersection / union if union > 0 else 0
    def analyze_files(self, file_paths):
        """分析文件相似度"""
        file_ngrams = {}
        for file_path in file_paths:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            file_ngrams[file_path] = set(self.extract_ngrams(content))
        # 计算相似度
        results = []
        files = list(file_ngrams.keys())
        for i in range(len(files)):
            for j in range(i + 1, len(files)):
                similarity = self.jaccard_similarity(
                    file_ngrams[files[i]], 
                    file_ngrams[files[j]]
                )
                results.append((files[i], files[j], similarity))
        return sorted(results, key=lambda x: x[2], reverse=True)
# 使用示例
ngram_analyzer = NGramSimilarityAnalyzer(n=3)
results = ngram_analyzer.analyze_files(['file1.txt', 'file2.txt', 'file3.txt'])
for file1, file2, similarity in results:
    print(f"{file1} <-> {file2}: {similarity:.4f}")

主题模型分析（LDA）

from gensim import corpora, models
import jieba
class LDAAnalyzer:
    def __init__(self, num_topics=5):
        self.num_topics = num_topics
        self.dictionary = None
        self.model = None
    def preprocess_chinese(self, text):
        """中文文本预处理"""
        words = jieba.cut(text)
        # 去停用词
        stop_words = set(['的', '了', '在', '是', '我', '有', '和', '就', '不', '人', '都', '一', '一个', '上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好', '自己', '这'])
        return [word for word in words if word not in stop_words and len(word) > 1]
    def train_model(self, documents):
        """训练LDA模型"""
        # 分词并创建词袋
        texts = [self.preprocess_chinese(doc) for doc in documents]
        self.dictionary = corpora.Dictionary(texts)
        corpus = [self.dictionary.doc2bow(text) for text in texts]
        # 训练LDA模型
        self.model = models.LdaModel(
            corpus=corpus,
            id2word=self.dictionary,
            num_topics=self.num_topics,
            passes=10
        )
        return corpus, texts
    def get_document_topics(self, text):
        """获取文档主题分布"""
        words = self.preprocess_chinese(text)
        bow = self.dictionary.doc2bow(words)
        return self.model.get_document_topics(bow)
# 使用示例
lda = LDAAnalyzer(num_topics=3)
documents = ["文档内容1", "文档内容2", "文档内容3"]
corpus, texts = lda.train_model(documents)
# 显示主题
print("主题分布：")
for topic_id, topic_words in lda.model.print_topics(num_words=5):
    print(f"主题 {topic_id}: {topic_words}")

快速文件对比（命令行工具）

Shell脚本（diff-based）

#!/bin/bash
# 文件相似度分析脚本
analyze_files() {
    local file1="$1"
    local file2="$2"
    # 检查文件是否存在
    if [ ! -f "$file1" ] || [ ! -f "$file2" ]; then
        echo "文件不存在"
        return 1
    fi
    # 计算行数
    lines1=$(wc -l < "$file1")
    lines2=$(wc -l < "$file2")
    # 计算共同行数
    common_lines=$(comm -12 <(sort "$file1") <(sort "$file2") | wc -l)
    # 计算相似度（基于共同行）
    total_lines=$((lines1 + lines2))
    if [ $total_lines -gt 0 ]; then
        similarity=$(echo "scale=4; $common_lines * 2 / $total_lines" | bc)
        echo "文件1: $file1 ($lines1 行)"
        echo "文件2: $file2 ($lines2 行)"
        echo "共同行数: $common_lines"
        echo "相似度: $similarity"
    fi
}
# 批量分析
analyze_directory() {
    local dir="$1"
    local files=($(find "$dir" -type f -name "*.txt"))
    for i in "${!files[@]}"; do
        for j in $(seq $((i+1)) $((${#files[@]}-1))); do
            echo "---"
            analyze_files "${files[$i]}" "${files[$j]}"
        done
    done
}
# 使用
analyze_directory "./documents/"

完整分析工具

#!/usr/bin/env python3
import os
import argparse
from typing import Dict, List, Tuple
class ComprehensiveAnalyzer:
    def __init__(self):
        self.methods = {}
    def register_method(self, name, func):
        self.methods[name] = func
    def analyze(self, file_paths: List[str], method: str = 'all') -> Dict:
        """综合分析"""
        results = {}
        if method == 'all':
            for name, func in self.methods.items():
                try:
                    results[name] = func(file_paths)
                except Exception as e:
                    print(f"方法 {name} 执行失败: {e}")
        else:
            if method in self.methods:
                results[method] = self.methods[method](file_paths)
            else:
                print(f"未知分析方法: {method}")
        return results
def main():
    parser = argparse.ArgumentParser(description='文件内容相关性分析工具')
    parser.add_argument('files', nargs='+', help='要分析的文件')
    parser.add_argument('--method', choices=['tfidf', 'ngram', 'lda', 'all'], 
                       default='all', help='分析方法')
    args = parser.parse_args()
    # 创建分析器并执行分析
    analyzer = ComprehensiveAnalyzer()
    results = analyzer.analyze(args.files, args.method)
    # 输出结果
    for method, result in results.items():
        print(f"\n=== {method} 分析结果 ===")
        print(result)
if __name__ == "__main__":
    main()

使用建议

1. 选择合适的方法：

   • 短文本：使用Jaccard相似度
   • 长文本：使用TF-IDF余弦相似度
   • 多语言：使用Word2Vec或BERT嵌入

2. 性能优化：

   • 预处理缓存
   • 并行处理
   • 增量更新

3. 应用场景：

   • 文档去重
   • 抄袭检测
   • 主题分类
   • 信息检索

根据你的具体需求选择合适的分析方法,建议从简单的TF-IDF开始，逐步增加复杂度。