本文目录导读：
基于词频的相似度分析
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()
使用建议
-
选择合适的方法：
- 短文本：使用Jaccard相似度
- 长文本：使用TF-IDF余弦相似度
- 多语言：使用Word2Vec或BERT嵌入
-
性能优化：
- 预处理缓存
- 并行处理
- 增量更新
-
应用场景：
- 文档去重
- 抄袭检测
- 主题分类
- 信息检索
根据你的具体需求选择合适的分析方法,建议从简单的TF-IDF开始，逐步增加复杂度。
