本文目录导读：
基于文本相似度的聚类
使用 Python 和 TF-IDF + K-means
import os
import re
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import KMeans
from sklearn.preprocessing import normalize
import numpy as np
def read_files(directory):
"""读取目录下所有文件"""
documents = []
filenames = []
for filename in os.listdir(directory):
filepath = os.path.join(directory, filename)
if os.path.isfile(filepath):
try:
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
documents.append(content)
filenames.append(filename)
except:
print(f"无法读取文件: {filename}")
return documents, filenames
def preprocess_text(text):
"""文本预处理"""
# 转为小写
text = text.lower()
# 去除标点符号
text = re.sub(r'[^\w\s]', '', text)
# 去除多余空白
text = re.sub(r'\s+', ' ', text).strip()
return text
def rough_clustering(directory, n_clusters=3):
"""粗糙聚类主函数"""
# 读取文件
documents, filenames = read_files(directory)
if len(documents) < 2:
print("文件数量不足，无法聚类")
return
# 预处理
processed_docs = [preprocess_text(doc) for doc in documents]
# TF-IDF 向量化
vectorizer = TfidfVectorizer(
max_features=1000, # 限制特征数量
stop_words='english', # 英文停用词
max_df=0.8, # 文档频率上限
min_df=2 # 文档频率下限
)
# 转换为特征向量
X = vectorizer.fit_transform(processed_docs)
# 使用 K-means 聚类
kmeans = KMeans(
n_clusters=n_clusters,
random_state=42,
n_init=10
)
kmeans.fit(X)
# 输出结果
clusters = {}
for idx, label in enumerate(kmeans.labels_):
if label not in clusters:
clusters[label] = []
clusters[label].append(filenames[idx])
print("聚类结果：")
for cluster_id, files in clusters.items():
print(f"\n簇 {cluster_id + 1}:")
for file in files:
print(f" - {file}")
return kmeans.labels_
# 使用示例
if __name__ == "__main__":
rough_clustering("./documents", n_clusters=3)
基于关键词匹配的简单聚类
def keyword_based_clustering(files, keywords_groups):
"""
基于关键词组的简单聚类
Args:
files: 文件字典 {文件名: 内容}
keywords_groups: 关键词组列表 [[kw1, kw2], [kw3, kw4], ...]
"""
clusters = {i: [] for i in range(len(keywords_groups))}
for filename, content in files.items():
content_lower = content.lower()
# 计算与每个关键词组的匹配度
scores = []
for group in keywords_groups:
score = sum(1 for kw in group if kw.lower() in content_lower)
scores.append(score)
# 分配到匹配度最高的组
if max(scores) > 0: # 至少匹配一个关键词
best_group = scores.index(max(scores))
clusters[best_group].append(filename)
else:
# 未匹配的文件单独分组
if -1 not in clusters:
clusters[-1] = []
clusters[-1].append(filename)
return clusters
# 使用示例
keywords_groups = [
['python', '编程', '代码', '开发'],
['数据', '分析', '统计', '机器学习'],
['文档', '报告', '说明', '手册']
]
files = {
'file1.txt': 'Python编程入门教程',
'file2.txt': '数据分析报告',
'file3.txt': '用户使用手册'
}
result = keyword_based_clustering(files, keywords_groups)
使用 N-gram 相似度聚类
from collections import Counter
from typing import List, Set
import math
def ngram_similarity(text1: str, text2: str, n: int = 3) -> float:
"""计算两个文本的 N-gram 相似度"""
def get_ngrams(text: str, n: int) -> Set[str]:
text = text.lower()
return set(text[i:i+n] for i in range(len(text) - n + 1))
ngrams1 = get_ngrams(text1, n)
ngrams2 = get_ngrams(text2, n)
if not ngrams1 or not ngrams2:
return 0.0
intersection = ngrams1 & ngrams2
union = ngrams1 | ngrams2
return len(intersection) / len(union)
def ngram_clustering(files: dict, threshold: float = 0.3) -> dict:
"""
基于 N-gram 相似度的聚类
Args:
files: 文件字典 {文件名: 内容}
threshold: 相似度阈值
"""
clusters = {}
assigned = set()
cluster_id = 0
filenames = list(files.keys())
for i, f1 in enumerate(filenames):
if f1 in assigned:
continue
# 创建新簇
clusters[cluster_id] = [f1]
assigned.add(f1)
for j in range(i+1, len(filenames)):
f2 = filenames[j]
if f2 in assigned:
continue
# 计算相似度
similarity = ngram_similarity(files[f1], files[f2])
if similarity >= threshold:
clusters[cluster_id].append(f2)
assigned.add(f2)
cluster_id += 1
return clusters
# 使用示例
files = {
'doc1.txt': 'This is a programming document',
'doc2.txt': 'This document is about coding',
'doc3.txt': 'Statistical analysis of data',
'doc4.txt': 'Data analysis and statistics'
}
result = ngram_clustering(files, threshold=0.3)
使用 Bash 脚本实现简单聚类
#!/bin/bash
# 基于关键词的简单文件聚类
cluster_by_keywords() {
local directory="$1"
local output_dir="clustered_files"
mkdir -p "$output_dir"
# 定义关键词组
declare -A clusters
clusters["programming"]="python|java|code|programming"
clusters["data"]="data|analysis|statistics|machine learning"
clusters["documentation"]="manual|guide|documentation|help"
# 遍历文件
for file in "$directory"/*; do
if [ -f "$file" ]; then
filename=$(basename "$file")
content=$(cat "$file" | tr '[:upper:]' '[:lower:]')
matched=false
for cluster in "${!clusters[@]}"; do
keywords="${clusters[$cluster]}"
if echo "$content" | grep -qE "$keywords"; then
mkdir -p "$output_dir/$cluster"
cp "$file" "$output_dir/$cluster/"
matched=true
break
fi
done
# 未匹配的文件放入其他
if [ "$matched" = false ]; then
mkdir -p "$output_dir/other"
cp "$file" "$output_dir/other/"
fi
fi
done
echo "聚类完成！结果保存在 $output_dir 目录"
}
# 使用示例
cluster_by_keywords "/path/to/documents"
完整示例：综合方案
import os
import re
from pathlib import Path
import json
class FuzzyCluster:
"""模糊聚类器"""
def __init__(self, similarity_threshold=0.3):
self.threshold = similarity_threshold
self.clusters = {}
def read_files(self, directory):
"""读取目录下所有文本文件"""
files = {}
for filepath in Path(directory).glob('*.txt'):
try:
with open(filepath, 'r', encoding='utf-8') as f:
files[filepath.name] = f.read()
except Exception as e:
print(f"跳过文件 {filepath.name}: {e}")
return files
def simple_similarity(self, text1, text2):
"""简单相似度计算（基于共现词）"""
words1 = set(re.findall(r'\w+', text1.lower()))
words2 = set(re.findall(r'\w+', text2.lower()))
if not words1 or not words2:
return 0
intersection = words1 & words2
union = words1 | words2
return len(intersection) / len(union)
def cluster(self, directory):
"""执行聚类"""
files = self.read_files(directory)
filenames = list(files.keys())
if len(filenames) < 2:
return {"0": filenames}
# 简单的层次聚类
clusters = {}
assigned = set()
cluster_id = 0
for i, f1 in enumerate(filenames):
if f1 in assigned:
continue
clusters[cluster_id] = [f1]
assigned.add(f1)
for j in range(i+1, len(filenames)):
f2 = filenames[j]
if f2 in assigned:
continue
similarity = self.simple_similarity(files[f1], files[f2])
if similarity >= self.threshold:
clusters[cluster_id].append(f2)
assigned.add(f2)
cluster_id += 1
return clusters
def save_results(self, output_file="clusters.json"):
"""保存聚类结果"""
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(self.clusters, f, ensure_ascii=False, indent=2)
# 使用示例
if __name__ == "__main__":
clusterer = FuzzyCluster(similarity_threshold=0.2)
result = clusterer.cluster("./documents")
print("\n聚类结果：")
for cluster_id, files in result.items():
print(f"簇 {cluster_id}:")
for file in files:
print(f" - {file}")
clusterer.save_results()
使用方法
-
安装依赖（如果使用 Python 版本）：
pip install scikit-learn numpy
-
准备文件：将需要聚类的文件放在一个目录中
-
运行脚本：
python cluster_files.py
调整建议
- 相似度阈值：根据需求调整，值越低聚类越粗糙
- 聚类数量：K-means 需要指定数量，其他方法自动确定
- 特征提取：可以添加自定义的停用词列表
- 性能优化：大量文件时考虑分批处理
这些方法提供不同程度的模糊聚类,选择哪种取决于你的具体需求和数据特点。
