脚本如何实现文件内容均值漂移聚类

wen 实用脚本 2026-07-08 182

本文目录导读：

脚本如何实现文件内容均值漂移聚类

Python 实现文本均值漂移聚类
基于数值特征的均值漂移聚类
自定义均值漂移算法实现
命令行工具版本
注意事项

Python 实现文本均值漂移聚类

import numpy as np
from sklearn.cluster import MeanShift
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import StandardScaler
class TextMeanShiftCluster:
    def __init__(self, bandwidth=None):
        self.vectorizer = TfidfVectorizer(
            max_features=1000,
            stop_words='english'
        )
        self.cluster = MeanShift(bandwidth=bandwidth)
        self.scaler = StandardScaler()
    def read_file(self, filepath):
        """读取文件内容"""
        with open(filepath, 'r', encoding='utf-8') as f:
            content = f.read()
        return content
    def process_lines(self, filepath):
        """按行处理文件内容"""
        lines = []
        with open(filepath, 'r', encoding='utf-8') as f:
            for line in f:
                line = line.strip()
                if line:
                    lines.append(line)
        return lines
    def process_documents(self, filepath, delimiter='\n\n'):
        """按段落处理文件内容"""
        content = self.read_file(filepath)
        documents = content.split(delimiter)
        return [doc.strip() for doc in documents if doc.strip()]
    def cluster(self, text_data):
        """执行均值漂移聚类"""
        # 文本向量化
        X = self.vectorizer.fit_transform(text_data).toarray()
        # 标准化
        X_scaled = self.scaler.fit_transform(X)
        # 聚类
        labels = self.cluster.fit_predict(X_scaled)
        return labels, X_scaled
# 使用示例
def cluster_file_content(filepath):
    clusterer = TextMeanShiftCluster()
    # 读取文档
    documents = clusterer.process_documents(filepath)
    if len(documents) < 2:
        print("文档数量太少，无法聚类")
        return
    # 执行聚类
    labels, features = clusterer.cluster(documents)
    # 输出结果
    n_clusters = len(set(labels)) - (1 if -1 in labels else 0)
    print(f"聚类数量: {n_clusters}")
    for i, (doc, label) in enumerate(zip(documents, labels)):
        print(f"文档 {i+1}: 类别 {label}")
        # 只显示前50个字符
        print(f"  内容预览: {doc[:50]}...")
        print()
# 主程序
if __name__ == "__main__":
    # 处理单个文件
    # cluster_file_content('path/to/your/file.txt')
    # 或者完整示例
    sample_file = 'sample.txt'
    # 创建示例文件
    with open(sample_file, 'w', encoding='utf-8') as f:
        f.write("Python是一门流行的编程语言\n\n")
        f.write("机器学习是人工智能的重要分支\n\n")
        f.write("Python的库非常丰富\n\n")
        f.write("深度学习是机器学习的一种方法\n\n")
        f.write("数据科学需要Python和R语言\n\n")
    # 执行聚类
    cluster_file_content(sample_file)

基于数值特征的均值漂移聚类

import numpy as np
from sklearn.cluster import MeanShift
import matplotlib.pyplot as plt
class NumericFileCluster:
    def __init__(self, bandwidth=None):
        self.clusterer = MeanShift(bandwidth=bandwidth)
    def read_numeric_data(self, filepath, delimiter=','):
        """读取数值数据"""
        data = []
        with open(filepath, 'r') as f:
            for line in f:
                if line.strip() and not line.startswith('#'):
                    values = [float(x) for x in line.strip().split(delimiter)]
                    data.append(values)
        return np.array(data)
    def cluster(self, data):
        """执行聚类"""
        labels = self.clusterer.fit_predict(data)
        cluster_centers = self.clusterer.cluster_centers_
        return labels, cluster_centers
    def visualize(self, data, labels):
        """可视化聚类结果（仅适用于2D数据）"""
        if data.shape[1] == 2:
            plt.scatter(data[:, 0], data[:, 1], c=labels, cmap='viridis')
            plt.title('Mean Shift Clustering Result')
            plt.colorbar()
            plt.show()
        else:
            print(f"数据维度 {data.shape[1]}，无法可视化")
# 使用示例
def cluster_numeric_file(filepath):
    clusterer = NumericFileCluster(bandwidth=0.5)
    data = clusterer.read_numeric_data(filepath)
    labels, centers = clusterer.cluster(data)
    # 输出结果
    print(f"数据点数量: {len(data)}")
    print(f"聚类中心数量: {len(centers)}")
    print(f"聚类标签: {labels}")
    return labels, centers

自定义均值漂移算法实现

import numpy as np
from collections import defaultdict
class CustomMeanShift:
    def __init__(self, bandwidth=1.0, max_iter=100):
        self.bandwidth = bandwidth
        self.max_iter = max_iter
    def gaussian_kernel(self, distance):
        """高斯核函数"""
        return np.exp(-0.5 * (distance / self.bandwidth) ** 2)
    def shift_point(self, point, points):
        """计算一个点的均值漂移向量"""
        # 计算所有点到当前点的距离
        distances = np.linalg.norm(points - point, axis=1)
        # 应用核函数
        weights = self.gaussian_kernel(distances)
        # 计算加权平均
        weighted_sum = np.sum(points * weights[:, np.newaxis], axis=0)
        weight_sum = np.sum(weights)
        return weighted_sum / weight_sum
    def fit(self, data):
        """执行均值漂移聚类"""
        # 初始化每个点为候选聚类中心
        centroids = data.copy()
        # 迭代漂移
        for iteration in range(self.max_iter):
            new_centroids = []
            for centroid in centroids:
                # 计算新的中心点
                new_center = self.shift_point(centroid, data)
                new_centroids.append(new_center)
            new_centroids = np.array(new_centroids)
            # 检查收敛
            shift_distance = np.linalg.norm(centroids - new_centroids, axis=1)
            if np.max(shift_distance) < 1e-3:
                break
            centroids = new_centroids
        # 合并相近的聚类中心
        unique_centroids = self.merge_centroids(centroids)
        # 分配标签
        labels = self.assign_labels(data, unique_centroids)
        self.cluster_centers_ = unique_centroids
        self.labels_ = labels
        return labels
    def merge_centroids(self, centroids, min_distance=0.5):
        """合并相近的聚类中心"""
        merged = []
        used = set()
        for i, c1 in enumerate(centroids):
            if i in used:
                continue
            group = [c1]
            used.add(i)
            for j, c2 in enumerate(centroids[i+1:], i+1):
                if j not in used:
                    distance = np.linalg.norm(c1 - c2)
                    if distance < min_distance:
                        group.append(c2)
                        used.add(j)
            # 计算组内平均作为新的中心
            merged.append(np.mean(group, axis=0))
        return np.array(merged)
    def assign_labels(self, data, centroids):
        """为每个数据点分配标签"""
        labels = []
        for point in data:
            # 计算到所有中心的距离
            distances = [np.linalg.norm(point - c) for c in centroids]
            # 分配最近的标签
            label = np.argmin(distances)
            labels.append(label)
        return np.array(labels)
# 文本特征提取的自定义实现
class TextFeatureExtractor:
    def __init__(self, max_features=1000):
        self.max_features = max_features
        self.vocabulary = {}
    def extract_features(self, texts):
        """提取文本特征（词袋模型）"""
        # 构建词汇表
        word_freq = defaultdict(int)
        for text in texts:
            words = text.lower().split()
            for word in words:
                if len(word) > 2:  # 过滤短词
                    word_freq[word] += 1
        # 选择最频繁的词
        sorted_words = sorted(word_freq.items(), 
                            key=lambda x: x[1], 
                            reverse=True)[:self.max_features]
        self.vocabulary = {word: idx for idx, (word, _) in enumerate(sorted_words)}
        # 构建特征矩阵
        features = []
        for text in texts:
            vector = np.zeros(len(self.vocabulary))
            words = text.lower().split()
            for word in words:
                if word in self.vocabulary:
                    vector[self.vocabulary[word]] += 1
            features.append(vector)
        return np.array(features)
# 完整的使用示例
def complete_example():
    # 创建示例数据
    sample_texts = [
        "Python编程语言很流行",
        "Java也是一种主流语言",
        "JavaScript用于网页开发",
        "深度学习需要大量数据",
        "机器学习是AI的核心",
        "数据科学包含统计知识"
    ]
    # 提取特征
    extractor = TextFeatureExtractor(max_features=100)
    features = extractor.extract_features(sample_texts)
    # 自定义均值漂移聚类
    custom_ms = CustomMeanShift(bandwidth=0.8)
    labels = custom_ms.fit(features)
    # 输出结果
    for text, label in zip(sample_texts, labels):
        print(f"文本: {text} -> 类别: {label}")
# 运行示例
if __name__ == "__main__":
    complete_example()

命令行工具版本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sys
import argparse
import numpy as np
from sklearn.cluster import MeanShift
def main():
    parser = argparse.ArgumentParser(description='文件内容均值漂移聚类')
    parser.add_argument('file', help='输入文件路径')
    parser.add_argument('--bandwidth', type=float, default=None, 
                       help='聚类带宽参数')
    parser.add_argument('--delimiter', default='\n\n', 
                       help='文档分隔符')
    parser.add_argument('--max-features', type=int, default=100, 
                       help='最大特征数量')
    parser.add_argument('--output', help='输出文件路径')
    args = parser.parse_args()
    # 这里可以添加处理逻辑
    print(f"处理文件: {args.file}")
    print(f"带宽参数: {args.bandwidth or 'auto'}")
if __name__ == "__main__":
    main()