当前位置：首页 > article >正文

python基础入门：3.5实战：词频统计工具

article 2025/2/10 12:50:55

Python词频统计终极指南：字典与排序的完美结合

import re
from collections import defaultdict

def word_frequency_analysis(file_path, top_n=10):
    """
    完整的词频统计解决方案
    :param file_path: 文本文件路径
    :param top_n: 显示前N个高频词
    :return: 排序后的词频列表
    """
    # 文本预处理管道
    with open(file_path, 'r', encoding='utf-8') as f:
        text = f.read().lower()  # 统一小写
        words = re.findall(r'\b[a-z]+\b', text)  # 提取纯单词

    # 使用defaultdict简化计数
    word_counts = defaultdict(int)
    for word in words:
        word_counts[word] += 1

    # 多级排序：先按频率降序，再按字母顺序
    sorted_words = sorted(word_counts.items(),
                         key=lambda x: (-x[1], x[0]))
  
    # 格式化输出
    print(f"{" 单词 ":<15}{" 频率 ":<10}")
    print("-"*25)
    for word, count in sorted_words[:top_n]:
        print(f"{word:<15}{count:<10}")
    return sorted_words

# 使用示例
analysis_result = word_frequency_analysis("sample.txt", top_n=15)

核心实现解析

1. 文本预处理优化

# 进阶正则表达式处理
text = """
Python's 3.10 version introduced pattern matching!
Don't-miss-this-feature.
"""

# 处理步骤分解
cleaned = re.sub(r"[^\w\s-]", "", text.lower())  # 保留连字符
words = re.findall(r"\b[\w-]+\b", cleaned)  # 匹配带连字符的单词
# 结果：["python's", "version", ...] → 需要进一步过滤

# 最终优化版正则
words = re.findall(r"\b(?![\d_]+)[a-z'-]+\b", text.lower())

预处理流程图：

原始文本 → 小写转换 → 标点过滤 → 单词提取 → 有效词过滤

2. 统计方法对比

# 方法1：原生字典
counts = {}
for word in words:
    counts[word] = counts.get(word, 0) + 1

# 方法2：defaultdict
counts = defaultdict(int)
for word in words:
    counts[word] += 1

# 方法3：Counter
from collections import Counter
counts = Counter(words)

性能对比表：

方法	时间复杂度	内存使用	可读性
原生字典	O(n)	低	中等
defaultdict	O(n)	低	高
Counter	O(n)	低	最高

进阶排序技巧

1. 多级排序实现

# 主要排序规则：频率降序 → 次要规则：字母升序
sorted_words = sorted(word_counts.items(),
                      key=lambda x: (-x[1], x[0]))

# 等效写法
sorted_words = sorted(word_counts.items(),
                      key=lambda x: (x[1], x[0]),
                      reverse=True)
sorted_words = sorted(sorted_words,
                      key=lambda x: x[0])

2. 频率分布直方图

def plot_frequency(sorted_list):
    """使用matplotlib可视化结果"""
    import matplotlib.pyplot as plt
  
    words, counts = zip(*sorted_list[:20])
    plt.figure(figsize=(12, 6))
    plt.barh(words[::-1], counts[::-1])  # 降序排列显示
    plt.xlabel('Frequency')
    plt.title('Top 20 Frequent Words')
    plt.tight_layout()
    plt.show()

plot_frequency(analysis_result)

企业级功能扩展

1. 停用词过滤

def load_stopwords(file='stopwords.txt'):
    with open(file) as f:
        return set(line.strip() for line in f)

stopwords = load_stopwords()
filtered_words = [w for w in words if w not in stopwords]

2. 词干提取

from nltk.stem import PorterStemmer
stemmer = PorterStemmer()

processed_words = [stemmer.stem(w) for w in filtered_words]

3. 多文件批处理

import glob

def batch_analysis(folder_path):
    total_counts = defaultdict(int)
    for file in glob.glob(f"{folder_path}/*.txt"):
        with open(file) as f:
            words = re.findall(r'\b[a-z]+\b', f.read().lower())
            for word in words:
                total_counts[word] += 1
    return sorted(total_counts.items(), key=lambda x: -x[1])

性能优化策略

1. 内存优化

# 生成器处理大文件
def process_large_file(file_path):
    with open(file_path) as f:
        for line in f:
            words = re.findall(r'\b[a-z]+\b', line.lower())
            yield from words

# 流式处理
counts = defaultdict(int)
for word in process_large_file("big_data.txt"):
    counts[word] += 1

2. 多线程加速

from concurrent.futures import ThreadPoolExecutor

def parallel_count(file_chunks):
    with ThreadPoolExecutor() as executor:
        results = executor.map(count_chunk, file_chunks)
    # 合并结果
    total = defaultdict(int)
    for partial in results:
        for k, v in partial.items():
            total[k] += v
    return total

扩展应用：

结合TF-IDF算法实现关键词提取
实时文本流分析（使用滑动窗口）
情感分析中的特征词统计
自动补全系统的词频基础

# 实时分析示例
from collections import deque

class StreamingAnalyzer:
    def __init__(self, window_size=1000):
        self.window = deque(maxlen=window_size)
        self.counts = defaultdict(int)
  
    def add_text(self, text):
        words = re.findall(r'\b[a-z]+\b', text.lower())
        for word in words:
            # 处理过期词
            if len(self.window) == self.window.maxlen:
                old_word = self.window.popleft()
                self.counts[old_word] -= 1
                if self.counts[old_word] == 0:
                    del self.counts[old_word]
            # 更新新词
            self.window.append(word)
            self.counts[word] += 1
  
    def get_top_words(self, n=10):
        return sorted(self.counts.items(), key=lambda x: -x[1])[:n]