.gitignore

@@ -0,0 +1,4 @@
+log.txt
+/test
+/.venv
+__pycache__

A 代码模式/10 一盘大棋/1 最基础的写法.py

@@ -1,5 +1,4 @@
-# -*- coding: utf-8 -*-
-
+# 引入停用词表和测试文件的路径
 from cppy.cp_util import stopwordfilepath, testfilepath
 
 # 准备停用词表
@@ -49,10 +48,3 @@ for i in range(n):
 # 打印频率最高的前10个词
 for tf in word_freqs[:10]:
     print(tf[0], '-', tf[1])
-
-
-'''
-想到哪里写到哪里，只会Python语言最基础的语法，C语言编程风格 。
-所有逻辑，所有细节一盘大棋摆在面前。
-这样的代码,很难维护。后来的阅读者会进入一个一望无际的泥塘.
-'''

A 代码模式/10 一盘大棋/2 加入语言特性.py

@@ -23,8 +23,8 @@ with open(testfilepath, encoding='utf8') as f:
 # 打印前10个最常见的单词
 for word, freq in word_freqs.most_common(10):
     print(f"{word}-{freq}")
-    
 '''
+相比 A01
 使用collections.Counter来计数单词频率，从而简化了代码并提高了效率。
 使用enumerate来获取行号和行内容，使用set来存储停用词，都有助于提高代码的性能和可读性。
 使用most_common方法来获取最常见的单词，使输出更为简洁。

A 代码模式/10 一盘大棋/3 Hacker.py

@@ -8,9 +8,7 @@ words = re.findall('[a-z]{2,}',
 counts = collections.Counter(w for w in words if w not in stopwords)
 for (w, c) in counts.most_common(10):
     print(w, '-', c)
-    
 '''
 熟练的软件工程师，会如此简单完成任务
 后面的例子，我们必须变的啰嗦一些，不能用这种太 hacker 的写法
-我们假设是要解决一个相对复杂的问题
 '''

A 代码模式/11 基础结构/函数/1 全局变量.py

@@ -1,6 +1,3 @@
-'''
-给代码片段命名，打包成函数。用函数名概括问题求解的过程，方便阅读理解。
-'''
 import string
 from collections import Counter
 from cppy.cp_util import *

A 代码模式/11 基础结构/函数/2 临时变量传递.py

@@ -1,13 +1,7 @@
-'''
-消除全局变量，限定函数只有输入输出来和外界打交道。
-如果消除无用的中间变量，那么就是典型的数据处理流水线风格:我要在一个函数中，完成某个操作，然后调用下一个函数进行处理 。
-流水线风格阅读方便、提供栈性能优化的可能 。
-如果需要加参数，那么需要柯里化方法（即Python里面的闭包函数结构）
-'''
-
 import re
 from cppy.cp_util import *
 
+
 def extractwords(str_data):
     pattern = re.compile('[\W_]+')
     word_list = pattern.sub(' ', str_data).lower().split()

A 代码模式/11 基础结构/函数/3 常见风格.py

@@ -1,8 +1,3 @@
-'''
-把一些公共函数分离出去，保留中间变量来增加代码可读性。
-在 jupyter notebook 做分析常常如此。 
-'''
-
 import re
 from collections import Counter
 from cppy.cp_util import *

A 代码模式/11 基础结构/函数/4 递归.py

@@ -1,5 +1,3 @@
-''' 递归是基础编程思维产生的代码结构 。注意递归深度限制只能解决小规模问题 '''
-
 from cppy.cp_util import *
 from collections import Counter
 

A 代码模式/11 基础结构/对象化/1 类对象.py

@@ -1,10 +1,3 @@
-'''
-对象化的视角看待问题是抽象出问题中的实体。
-实体有方法、属性。程序执行过程，就是调动这些实体的交互行为 。
-作为彻底的面向对象的风格，主逻辑常常也写做一个类，init 产生各个工具，run 运行逻辑 。
-本例中实体可理解为几个工具。
-'''
-
 from collections import Counter
 from cppy.cp_util import *
 

A 代码模式/11 基础结构/对象化/2 字典对象.py

@@ -1,11 +1,5 @@
 from cppy.cp_util import *
 
-'''
-如果认为基于数据和函数的封装就是对象化编程。那么类并不是必要的，扩展的函数、字典也有这个能力。
-函数利用属性存数据，字典有的value是函数，有的是value是数据 。
-字典进行对象化程序设计，只是缺乏权限控制机制，而继承方面可以是通过一个代理模式包装实现。
-简单使用字典，也能满足对象的一些应用场景。而且更省资源
-'''
 
 def extract_words(obj, path_to_file):
     """
@@ -55,4 +49,4 @@ if __name__ == '__main__':
 
     # 获取排序后的单词频率并打印
     word_freqs = word_freqs_obj['sorted']()
-    print_word_freqs(word_freqs)
+    print_word_freqs(word_freqs)

A 代码模式/12 语言特性/2 暂停执行/1 生成器.py

@@ -1,7 +1,7 @@
 from cppy.cp_util import *
 
 #
-# 生成器 是一种简单异步实现
+# 生成器
 #
 def non_stop_words(testfilepath): 
     stopwords = get_stopwords()

A 代码模式/12 语言特性/2 暂停执行/2 异步.py

@@ -3,19 +3,10 @@ import aiofiles
 from collections import Counter
 from cppy.cp_util import *
 
-'''
-异步指程序运行期间具有挂起，然后等待唤醒，继续执行的能力。
-典型应用场景如数据 
-① 源源不断产生,比如网络监听； 
-② 太大,可能大于内存； 
-③ 有连续处理动作，当某一环得到想要结果，立刻终止，后续数据不需要处理 。
-
-本例，读文件的Io还是太快，的爬虫
-
-async 是当前版本 Python 官方异步编程的标准库
-aiofiles 是一个第三方库，提供异步文件操作的功能
-'''
 
+#
+# 协程: 有点复杂; 读文件的Io还是太快，的爬虫
+#
 async def read_file(file_path):
     async with aiofiles.open(file_path, 'r', encoding='utf-8') as file:
         content = await file.read()

A 代码模式/12 语言特性/3 运行期符号表/1 内省/1 内省.py

@@ -1,7 +1,5 @@
 from cppy.cp_util import *
 
-# 内省（Introspection）在编程中是指程序在运行时检查对象（变量）类型或属性的能力。
-# 这是一种动态地了解对象的方法，非常有利于调试、优化代码、实现泛型编程等场景。
 
 class Calculator:
     def frequencies(self,word_list):    

A 代码模式/12 语言特性/3 运行期符号表/2 装饰/1 反射装饰.py

@@ -1,9 +1,6 @@
 import time
 import cppy.cp_util as util
 
-'''
-用反射实现装饰器的效果
-'''
 
 # 工具函数
 def extract_words(path_to_file):    

A 代码模式/12 语言特性/3 运行期符号表/2 装饰/2 装饰器.py

@@ -0,0 +1,34 @@
+from collections import Counter
+import cppy.cp_util as util
+
+def word_frequency( top_n=10 ):
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            # 执行被装饰的函数
+            result = func(*args, **kwargs)
+            
+            # 初始化词频计数器
+            word_counts = Counter()
+            
+            # 分词并计数
+            for word in util.extract_str_words(result):
+                word_counts[word] += 1            
+            
+            # 输出所有词的频率最高的n个词            
+            most_common = word_counts.most_common(top_n)    
+            util.print_word_freqs( most_common )                    
+            
+            return result
+        
+        return wrapper
+    return decorator
+
+
+# 使用装饰器
+@word_frequency( top_n=10 )
+def read_file(file_path):
+    with open(file_path, 'r', encoding='utf-8') as file:
+        return file.read()
+
+
+read_file( util.testfilepath )

A 代码模式/12 语言特性/3 运行期符号表/2 装饰/3 参数类型检查.py

@@ -1,6 +1,3 @@
-
-# 自己设计类装饰器，类方法做进行类型申明和检查
-
 from collections import Counter
 from cppy.cp_util import *
 

A 代码模式/12 语言特性/3 运行期符号表/3 反射/反射_函数.py

@@ -1,8 +1,3 @@
-'''
-反射是一种动态地访问或修改对象状态和行为的能力，包括其属性、方法等。
-有被 Hacker 攻击的风险。
-'''
-
 import cppy.cp_util as util
 import operator
 

A 代码模式/13 多计算单元/数据共享/1 内存/1 队列.py

@@ -0,0 +1,56 @@
+import  threading, queue
+from cppy.cp_util import *
+from collections import Counter
+
+stop_words = get_stopwords()
+
+# 待处理数据放一个队列，多个线程轮流计数，最后合并统一计数
+class WordFrequencyCounter:
+    def __init__(self, input_file):
+        self.word_space = queue.Queue()
+        self.freq_space = queue.Queue()                
+        for chunk in get_chunks(input_file,3000):                        
+            self.word_space.put(chunk)        
+
+    def process_words(self):
+        while not self.word_space.empty():
+            try:
+                chunk = self.word_space.get_nowait()  # 不使用超时，持续获取数据
+            except queue.Empty:
+                break  # 队列为空，退出循环
+            # print(f"Worker thread ID: {threading.get_ident()}",len(chunk))  
+            words = [ w for w in chunk if w not in stop_words and len(w) >= 3 ]
+            word_freqs = Counter(words)
+            self.freq_space.put(dict(word_freqs))  # 将Counter对象转换为字典
+
+    def run(self):
+        workers = [ threading.Thread(target=self.process_words) for _ in range(5)]
+        for worker in workers:  worker.start()
+        for worker in workers:  worker.join()
+
+        word_freqs = Counter()  # 初始化一个空的Counter对象
+        while not self.freq_space.empty():
+            freqs = self.freq_space.get()            
+            if freqs:  # 确保freqs非空
+                word_freqs.update(freqs)  
+
+        print_word_freqs ( sort_dict (word_freqs) )
+        
+
+@timing_decorator
+def main(): 
+    counter = WordFrequencyCounter( testfilepath )
+    counter.run()
+
+if __name__ == '__main__':
+    main()
+
+'''
+在多线程之间传递数据，建议使用线程安全的队列，如queue.Queue或multiprocessing.Queue（后者也适用于多进程环境）。
+这些队列提供了线程安全的数据传输机制，可以避免竞态条件和数据损坏。
+全局变量不可预测
+
+multiprocessing.Queue 利用了操作系统提供的进程间通信（IPC, Inter-Process Communication）机制，具体实现取决于不同操作系统的支持。
+在Unix/Linux系统中，multiprocessing.Queue通常基于管道（pipes）、共享内存和/或消息队列等机制实现。
+而在Windows系统上，可能使用命名管道（named pipes）或者内存映射文件（memory-mapped files），以及某些版本的Windows特有的进程间同步对象如Mutexes、Semaphores和事件。
+'''    

A 代码模式/13 多计算单元/数据共享/2 本地服务/2 服务进程_不合并.py

@@ -0,0 +1,62 @@
+'''
+使用 multiprocessing.Manager:
+Manager 提供了一个可以在不同进程之间共享和修改的数据类型，如 list, dict, Namespace 等。
+它实际上是在背后启动了一个单独的服务器进程，其他进程通过代理来访问这些共享对象。
+
+使用 multiprocessing.Manager 来完成统计词频
+需要注意：
+ - Manager() 必须用函数包起来,不能按脚本随便放外面，否则会提示freeze_support
+ - 工作函数需要放到外面，不能做内部函数。否则会提示参数错误
+ - 无法在 Jupyter 类似环境运行 
+'''
+
+from cppy.cp_util import *
+from collections import Counter
+from multiprocessing import Manager, Process
+
+stop_words = get_stopwords()
+
+def process_chunk(shared_chunks,word_count):    
+    while True:
+        try:
+            chunk = shared_chunks.pop(0)  # 从共享列表中取出一个数据块        
+            if chunk is None: break # 如果取出的是None，表示所有数据块已处理完毕
+            words = extract_str_words(chunk)            
+            for word in words:
+                if word not in stop_words:
+                    word_count[word] = word_count.get(word, 0) + 1
+        except Exception as e:
+            print(e)
+            break
+
+
+@timing_decorator
+def main():         
+    # 创建一个Manager实例    
+    manager = Manager() 
+    shared_chunks = manager.list()     
+    word_count = manager.dict()
+
+    # 读取文件并按块大小分割，将块添加到共享列表中  
+    chunk_size = 1024 * 10  # 假设每个块是10KB，可以根据需要调整  
+    with open(testfilepath, 'r', encoding='utf-8') as f:  
+        while True:  
+            chunk = f.read(chunk_size)  
+            if not chunk: break  
+            shared_chunks.append(chunk)
+        shared_chunks.append(None)         
+    print('-------------------',len(shared_chunks))
+    processes = [ Process( target=process_chunk, 
+                          args=(shared_chunks,word_count)) 
+                    for _ in range( 4 ) ]  # 假设启动4个工作进程    
+    for p in processes:  p.start()    
+    for p in processes:  p.join()
+
+    # 将Manager类型的字典转换为普通的字典，以便使用Counter  
+    word_count = dict(word_count)  
+    word_freqs = Counter(word_count).most_common(10)      
+    print_word_freqs(word_freqs)    
+
+
+if __name__ == '__main__':
+    main()

A 代码模式/13 多计算单元/数据共享/2 本地服务/2 服务进程_分包.py

@@ -0,0 +1,42 @@
+'''
+使用 multiprocessing.Manager:
+Manager 提供了一个可以在不同进程之间共享和修改的数据类型，如 list, dict, Namespace 等。
+它实际上是在背后启动了一个单独的服务器进程，其他进程通过代理来访问这些共享对象。
+'''
+# 使用 multiprocessing.Manager 来完成统计词频
+# 怎么得到最快的一个结果，是一个试错过程：X程创建数目多少、分片的大小 ...
+
+from cppy.cp_util import *
+from collections import Counter
+from multiprocessing import Manager, Process
+
+stop_words = get_stopwords()
+
+def process_chunk(chunk,word_count):    
+    words = [ w for w in chunk if ( not w in stop_words ) and len(w) >= 3 ]    
+    for word in words: # 非常化时间
+        word_count[word] = word_count.get(word, 0) + 1
+    # word_count.update( Counter(words) ) # 类型不起作用
+
+@timing_decorator
+def main():   
+    manager = Manager()
+    word_count = manager.dict()
+
+    chunks = get_chunks(testfilepath,2800)    
+    print('-------------------',len(chunks))
+    processes = []
+    for chunk in chunks:
+        p = Process(target=process_chunk, 
+                    args=(chunk,word_count) )
+        processes.append(p)
+        p.start()
+
+    for p in processes:  p.join()
+
+    word_count = dict(word_count)  
+    word_freqs = Counter(word_count).most_common(10)      
+    print_word_freqs(word_freqs)    
+
+if __name__ == '__main__':
+    main()

A 代码模式/13 多计算单元/数据共享/3 数据库/数据库.py

@@ -1,10 +1,3 @@
-''' 
-以查询和独立组件并发操作数据为中心的思路，建数据库，围绕数据库做查询操作。
-现代开发模式中，编程语言很少直接用 SQL 操作数据库了，常常有个隔离具体数据库更方便编程的ORM层 ，
-用工具把表数据映射为对象(ORM) 
-createDb 创建数据对象，processData 处理数据，DataQuery 查询输出数据。相对“ 共享内存数据 ” ，它慢一些
-'''
-
 import sqlite3, os.path
 from cppy.cp_util import testfilepath,db_filename,extract_file_words
 

A 代码模式/13 多计算单元/数据共享/4 网络/restful/app.py

@@ -1,7 +1,3 @@
-'''
-REST 风格是把数据资源拓展到了互联网环境，用Web协议访问 。
-组件可以用任何环境任何语言开发, 只需要遵循共同的 rest 协议 。
-'''
 # -*- coding: utf-8 -*-
 from flask import Flask, request, jsonify, abort
 from functools import lru_cache

A 代码模式/13 多计算单元/数据分包/1 mapreduce.py

@@ -31,6 +31,3 @@ def main():
 if __name__ == '__main__':
     main()
 
-'''
-简单的利用函数式风格
-'''

A 代码模式/13 多计算单元/数据分包/2 多线程.py

@@ -0,0 +1,48 @@
+# -*- coding: utf-8 -*-
+from collections import Counter
+from cppy.cp_util import *
+from multiprocessing.pool import ThreadPool
+
+
+#
+# 多线程
+#
+stop_words = get_stopwords()
+
+def process_chunk(chunk):
+    # 过滤停用词    
+    words = [ w for w in chunk if ( not w in stop_words ) and len(w) >= 3 ]
+    return Counter(words)
+
+
+def merge_counts(counts_list):
+    """合并多个Counter对象的总和"""
+    return sum(counts_list, Counter())
+
+
+def thread_function(chunk, counts_list):
+    word_count = process_chunk(chunk)
+    counts_list.append(word_count)
+
+
+@timing_decorator
+def main():    
+    # 读数据，按1000个词一组分片
+    chunks = get_chunks(testfilepath,1000)
+
+    # 线程池    
+    pool = ThreadPool(len(chunks))  # 随意指定的线程数       
+
+    counts_list = pool.map(process_chunk, chunks)
+    pool.close()
+    pool.join()
+
+    # 合并计数
+    total_counts = merge_counts(counts_list)
+
+    # 输出最高频的n个词
+    print_word_freqs(total_counts.most_common(10))
+
+
+if __name__ == '__main__':
+    main()

A 代码模式/13 多计算单元/数据分包/3 多进程.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+import multiprocessing
+from collections import Counter
+from cppy.cp_util import *
+
+
+#
+# 多进程: 因为创建进程相比计算过程开销太大，结果最慢
+#
+stop_words = get_stopwords()
+
+def process_chunk(chunk):
+    # 过滤停用词    
+    words = [ w for w in chunk if ( not w in stop_words ) and len(w) >= 3 ]
+    return Counter(words)
+
+def merge_counts(counts_list):
+    """合并多个Counter对象的总和"""
+    return sum(counts_list, Counter())
+
+
+@timing_decorator
+def main():
+    # 读取文件内容，分割文件内容为多个块，每个块由一个进程处理    
+    chunks = get_chunks(testfilepath,1000)
+
+    # 使用多进程处理每个块
+    pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
+    counts_list = pool.map(process_chunk, chunks)
+    pool.close()
+    pool.join()
+
+    # 合并计数
+    total_counts = merge_counts(counts_list)
+
+    # 输出最高频的n个词
+    print_word_freqs(total_counts.most_common(10))
+
+
+if __name__ == '__main__':
+    main()   
+

A 代码模式/14 人机交互/1 终端/终端命令行/command_line_1.py

@@ -2,7 +2,6 @@ import sys
 import re
 from collections import Counter
 
-# 提供带参数的运行命令行环境
 # 使用  ： python command_line_1.py testfilepath 10
 
 # 清洗文本，移除标点符号并转换为小写
@@ -44,5 +43,4 @@ def main():
         print(f"Error: {e}")
 
 if __name__ == "__main__":
-    main()
-
+    main()

A 代码模式/14 人机交互/1 终端/终端命令行/command_line_2.py

@@ -1,9 +1,6 @@
 import re
 from collections import Counter
 
-# 提供一个命令行交互方式来驱动程序运行
-
-
 # 清洗文本，移除标点符号并转换为小写
 def clean_text(text):
     return re.sub(r'[^\w\s]', '', text).lower()

A 代码模式/14 人机交互/1 终端/终端菜单/terminal_menu.py

@@ -1,7 +1,3 @@
-
-# 命令行菜单驱动程序
-# while 循环体中出现一个 case 菜单，每个菜单调用对应函数 
-
 import os
 import cppy.cp_util as util
 

A 代码模式/14 人机交互/2 视窗/PyQT.py

@@ -1,6 +1,3 @@
-
-# 应用基于一个窗口，窗口基于操作系统的消息机制 
-
 import sys
 from PyQt5.QtWidgets import QApplication, QWidget, QPushButton, QVBoxLayout, QTextEdit, QFileDialog
 import cppy.cp_util as util

A 代码模式/14 人机交互/2 视窗/test.txt

@@ -0,0 +1,2 @@
+" my Some sure acquaintance or other, my dear, sure,other  I suppose; I am sure I do not
+know. sure "

A 代码模式/14 人机交互/3 Web/MVC/app.py

@@ -1,13 +1,3 @@
-'''
- Web的MVC三层结构是互联网最火的应用框架。
- 系统需要切分为模型、视图、控制三层。
- Python 中最简单的 Web服务器是 flask 
-    app.py - Flask应用的主文件，控制器（Controller）。
-    models.py - 模型（Model）文件，包含词频统计的逻辑。
-    templates/index.html - 视图（View）文件，用于展示结果。
- '''
-
-
 from flask import Flask, render_template, request
 from models import WordFrequencyModel
 

A 代码模式/15 工程化/1 复用/函数调用复用.py

@@ -1,11 +1,7 @@
 from collections import Counter
 from cppy.cp_util import *
 
-'''
-如果函数设计干净（没有修改全局变量），那么相同的调用参数就会得到必然相同的结果。
-【缓存】把每次计算输入、输出存起来，如果后面有相同的任务（调用参数相同），就不用重复计算了。
-'''
-
+# 缓存系统，使用字典来存储文件路径和对应的词频结果; 可以存到本地磁盘，就可重复利用计算结果
 word_freq_cache = {}
 
 def calculate_word_frequency(file_path):

A 代码模式/15 工程化/1 复用/对象复用.py

@@ -1,6 +1,4 @@
 # 创建对象是消耗资源的，如果发现对象已经存在，可以返回引用，不创造新对象 。设计模式中这个做法叫享元
-# 可以降低资源需求和提升响应速度。更常见的该模式使用场景是各种资源池。
-
 from cppy.cp_util import *
 
 #享元类

A 代码模式/15 工程化/2 松耦合/1 观察者/1 观察者.py

@@ -0,0 +1,64 @@
+'''
+入门级示例，是用来帮助理解其他例子
+
+把观察者挂到自己的处理队列上
+适当时机调用所有队列上的约定的观察者的 update 方法
+如果观察者有多个职能参与不同的任务链，不一定要统一命名update方法
+
+这是一个示例性质的原型，具体环境下需要调整
+'''
+
+import collections  
+from abc import ABC, abstractmethod  
+from cppy.cp_util import *
+
+# 定义观察者接口 ，在 Pyhon中并不是必须
+class Observer(ABC):
+    @abstractmethod
+    def update(self, word):
+        pass
+  
+# 定义具体观察者类，用于统计词频  
+class WordFrequencyObserver(Observer):  
+    def __init__(self):  
+        self.word_count = collections.Counter()          
+  
+    def update(self, word):  
+        self.word_count[word] += 1  
+
+  
+# 定义主题类  
+class WordSubject:  
+    def __init__(self):  
+        self.observers = []  
+  
+    def attach(self, observer):  
+        self.observers.append(observer)  
+  
+    def notify(self, word):  
+        for observer in self.observers:  
+            observer.update(word)  
+  
+  
+# 主函数  
+def main(testfilepath, top_n = 10 ):  
+    stopwords = get_stopwords()  
+    subject = WordSubject()  
+  
+    # 创建一个观察者并附加到主题  
+    observer = WordFrequencyObserver()  
+    subject.attach(observer)  
+  
+    # 处理文件      
+    wordlist = re_split( read_file(testfilepath) )
+    for word in wordlist:  
+        if word not in stopwords:  
+            subject.notify(word)  # 通知
+  
+    # 打印最高的N个词频  
+    top_words = observer.word_count.most_common(top_n) 
+    print_word_freqs(top_words)
+    
+  
+if __name__ == "__main__":          
+    main( testfilepath )

A 代码模式/15 工程化/2 松耦合/1 观察者/2 观察者.py

@@ -0,0 +1,69 @@
+
+'''
+本例的基本模式还是观察者
+基类 Subject 提供注册和提醒注册上的对象提醒机制
+
+因为函数和参数混杂在一起传递，使得各个模块的处理结构其实是 case by case
+'''
+
+from collections import Counter
+from typing import List
+from cppy.cp_util import *
+
+class Subject:
+    def register_handler(self, handler: callable, *args, **kwargs):
+        self.handler = handler
+        self.args = args
+        self.kwargs = kwargs
+
+    def notify(self, *args, **kwargs):
+        self.handler( self.data, *self.args, **self.kwargs)
+        
+
+
+# 组件一：TextLoader - 负责读取文本并过滤停用词
+class TextLoader(Subject):
+    def load_text(self, filename: str) -> List[str]:
+        return extract_file_words(filename)
+
+    def notify(self, *args, **kwargs):
+        filename = args[0]
+        self.data = self.load_text(filename)
+        super().notify(self.data, *args, **kwargs)        
+
+
+# 组件二：WordCounter - 计算词频
+class WordCounter(Subject):
+    def count_words(self, words: List[str]) -> dict:
+        return Counter(words)
+
+    def notify(self, *args, **kwargs ):
+        words = args[0] 
+        self.data = self.count_words(words)        
+        super().notify(self.data, *args, **kwargs)
+
+
+# 组件三：TopWordsPresenter - 排序并输出前10个词
+class TopWordsPresenter(Subject):    
+    def notify(self, words,*args, **kwargs):                
+        n = args[0]                   
+        top_words = words.most_common(n)
+        print_word_freqs( top_words )
+
+
+# 主程序逻辑
+def main():
+    loader = TextLoader()
+    counter = WordCounter()
+    presenter = TopWordsPresenter()
+
+    # 注册事件处理器
+    loader.register_handler(counter.notify)
+    counter.register_handler( presenter.notify,10 )
+
+    # 触发加载文本并开始流程
+    loader.notify(testfilepath)
+
+
+if __name__ == "__main__":
+    main()

A 代码模式/15 工程化/2 松耦合/1 观察者/3 注册回调.py

@@ -0,0 +1,86 @@
+################ 待整理 
+'''
+注册者 = 观察者
+
+每个组件提供注册消息接口和注册消息动作
+
+在其它单元上注册自己对于特定事件（消息）的响应函数
+同时负责自己的注册队列的序贯调用
+Python 中有一个Callable类型，可以用来判断是否是可以回调类型
+from typing import Callable
+
+这是一个示例性质的原型，具体分布式环境下需要调整
+'''
+
+from collections import defaultdict
+from cppy.cp_util import *
+
+#
+#  event_manager
+#
+class EventManager:
+    def __init__(self):
+        self.load_handlers = []  # 用于加载文件的事件处理器
+        self.process_handlers = []  # 用于处理数据的事件处理器
+        self.end_handlers = []  # 用于结束流程的事件处理器        
+
+    def register_load_event(self, handler):
+        self.load_handlers.append(handler)
+
+    def register_process_event(self, handler):
+        self.process_handlers.append(handler)
+
+    def register_end_event(self, handler):
+        self.end_handlers.append(handler)
+    
+    # 运行框架，按顺序执行注册的事件处理器
+    def run(self, file_path):
+        for handler in self.load_handlers:        handler(file_path)
+        for handler in self.process_handlers:     handler()
+        for handler in self.end_handlers:         handler()
+
+
+#
+# 功能组件
+#
+# 定义数据存储类，用于模拟文件内容的加载和处理
+class TextData:        
+    _word_event_handlers = []
+
+    def __init__( self, event_manager ):
+        self._stop_words = get_stopwords()                
+        event_manager.register_load_event(self.__load)
+        event_manager.register_process_event(self.__process_words)
+
+    def __load(self, path_to_file):        
+        self._data = re_split( read_file(path_to_file) )    
+
+    def __process_words(self):
+        for word in self._data:            
+            if word not in self._stop_words:
+                for handler in self._word_event_handlers:
+                    handler(word)
+
+    def register_word_event(self, handler):
+        self._word_event_handlers.append(handler)
+
+
+class WordFrequencyCounter:
+    def __init__(self, event_manager, data_storage):
+        self._word_freqs = defaultdict(int) # 存储单词频率
+        data_storage.register_word_event(self.__increment_count) # 注册单词事件
+        event_manager.register_end_event(self.__print_freqs) # 注册结束事件
+
+    def __increment_count(self, word):
+        self._word_freqs[word] += 1
+
+    def __print_freqs(self):
+        print_word_freqs ( sort_dict (self._word_freqs) )       
+
+
+if __name__ == '__main__':
+    em = EventManager()    
+    data_storage = TextData(em)
+    word_freq_counter = WordFrequencyCounter(em, data_storage)
+
+    em.run(testfilepath)

A 代码模式/15 工程化/2 松耦合/1 观察者/4 订阅发布.py

@@ -0,0 +1,107 @@
+################ 待整理 
+
+from cppy.cp_util import *
+'''
+订阅者 = 注册者 = 观察者
+
+注册回调的一个变体
+要点是：中心化、统一化
+
+为了简化消息订阅可能形成的复杂性，
+提供一个中心消息管理器统一负责消息的订阅和回调
+各个功能组件只是完成自己的功能，
+在中心管理器上订阅消息挂到自己响应的处理函数上
+
+总结，相比较的改变：
+- 注册的时候，通过提供一个类型字段标识不同消息
+- 其它实体不做注册和做回调，统一这两个功能到一个中心单元
+
+这是一个示例性质的原型，具体分布式环境下需要调整
+'''
+from collections import defaultdict
+#################################################
+# Event Manager
+#################################################
+class EventManager:
+    def __init__(self):
+        self._subs = defaultdict(list)
+
+    def subscribe(self, event_type, handler):
+        self._subs[event_type].append(handler)
+
+    def publish(self, event):
+        event_type = event[0]        
+        for handle in self._subs.get(event_type, []):  
+            handle(event)
+
+#################################################
+# Application Entities
+#################################################
+class DataStorage:    
+    def __init__(self, event_manager):
+        self._event_manager = event_manager
+        self._event_manager.subscribe('load', self._load)
+        self._event_manager.subscribe('start', self.produce_words)
+
+    def _load(self, event):        
+        self._data = extract_file_words( event[1] )
+
+    def produce_words(self, _):        
+        for word  in self._data:
+            self._event_manager.publish(('word', word ))
+        self._event_manager.publish(('eof', None))
+
+class StopWordFilter:    
+    def __init__(self, event_manager):        
+        self._event_manager = event_manager
+        self._event_manager.subscribe('load', self.load_stop_words)
+        self._event_manager.subscribe('word', self.filter_word)
+        self._stop_words = set()
+
+    def load_stop_words(self, _ ):        
+        self._stop_words = set( get_stopwords() )
+
+    def filter_word(self, event):
+        word = event[1]
+        if word not in self._stop_words:
+            self._event_manager.publish(('valid_word', word))
+
+class WordFrequencyCounter:    
+    def __init__(self, event_manager):        
+        self._event_manager = event_manager
+        self._event_manager.subscribe('valid_word', self.count_word)
+        self._event_manager.subscribe('print', self.print_freqs)
+        self._word_freqs = {}
+
+    def count_word(self, event):
+        word = event[1]
+        self._word_freqs[word] = self._word_freqs.get(word, 0) + 1        
+
+    def print_freqs(self, _ ):
+        print_word_freqs ( sort_dict (self._word_freqs) )      
+      
+class WordFrequencyApp:
+    def __init__(self, event_manager):
+        self._event_manager = event_manager
+        self._event_manager.subscribe('run', self.start_application)
+        self._event_manager.subscribe('eof', self.stop_application)
+
+    def start_application(self, event):
+        path_to_file = event[1]
+        self._event_manager.publish(('load', path_to_file))
+        self._event_manager.publish(('start', ))
+
+    def stop_application(self, _ ):
+        self._event_manager.publish(('print', ))
+
+
+def main():    
+    event_manager  = EventManager()
+    DataStorage( event_manager )
+    StopWordFilter( event_manager )
+    WordFrequencyCounter( event_manager )
+    WordFrequencyApp( event_manager )
+    event_manager.publish(('run', testfilepath ))
+
+if __name__ == "__main__":
+    main()    

A 代码模式/15 工程化/2 松耦合/1 观察者/readme.MD

@@ -0,0 +1,9 @@
+
+注册
+- 解耦合：通过回调函数，可以将不同部分的代码逻辑分离，降低模块之间的耦合度。
+- 主动通信：注册回调模式实现了下层模块与上层模块之间的主动通信。当下层模块发生特定事件或满足特定条件时，可以主动调用上层模块注册的回调函数，而不需要上层模块不停地轮询下层模块的状态。
+
+- 异步处理：回调函数常用于异步操作的响应处理，可以在主线程之外执行耗时操作，提升程序的效率和响应速度。
+- 简化设计：在某些情况下，使用回调函数可以避免复杂的控制流设计，使代码更加简洁明了。
+
+- 适应变化：随着项目的发展，需求可能会发生变化。注册回调模式使得在不影响现有代码的基础上，容易添加新功能或修改现有逻辑。

A 代码模式/15 工程化/2 松耦合/2 消息链/1 只有消息接口.py

@@ -0,0 +1,98 @@
+################ 待整理 
+''' 
+应用场景针对各个组件的 notify 方法发指令来驱动所有工作
+这是一个示例性质的原型，具体分布式环境下需要调整
+
+notify 用了四种写法，是和本主题无关的测试
+'''
+from cppy.cp_util import *
+from collections import defaultdict
+
+badmsg = lambda : exec ('''  raise Exception("Message not understood " , action ) ''')
+class fff:
+    def __init__(self, d):
+        self._data = defaultdict( badmsg )
+        self._data.update(d)
+
+    def __getitem__(self, key):        
+        return self._data[key]
+
+
+class DataStorageMod():    
+    def __init__(self):
+        self._data = []
+
+    def notify(self, action, *args):
+        return {
+            'init': lambda : self._init,
+            'words': lambda : self._words
+        }.get( action , badmsg )()(*args)
+
+    def _init(self, path_to_file):
+        self._data = re_split( read_file(path_to_file) )
+
+    def _words(self):
+        return self._data
+   
+
+class StopWordMod():    
+    _stop_words = []
+
+    def notify(self, action, *args):
+        return {  'init': self._init,
+          'is_stop_word': self._is_stop_word
+        }[ action ](*args)  
+ 
+    def _init(self):        
+        self._stop_words = get_stopwords()  
+
+    def _is_stop_word(self, wordx):
+        return wordx in self._stop_words
+
+
+class WordFrequencyMod():    
+    _word_freqs = {}
+
+    def notify(self, action, *args):        
+        return fff( {
+            'increment_count':  lambda : self._increment_count,
+            'sorted':           lambda : self._sorted
+        })[ action ]()(*args)
+ 
+    def _increment_count(self, word):    
+        self._word_freqs[word] = self._word_freqs.get(word,0) + 1
+        
+    def _sorted(self):        
+        return sort_dict(self._word_freqs)
+
+
+class ScenarioManager():
+
+    def notify(self, action, *args):
+        if action == 'init':
+            return self._init( *args)
+        elif action == 'run':
+            return self._run()
+        else:
+            raise Exception("Message not understood " + action )
+ 
+    def _init(self, path_to_file):
+        self._storage_manager = DataStorageMod()
+        self._stop_word_manager = StopWordMod()
+        self._word_freq_manager = WordFrequencyMod()
+        self._storage_manager.notify('init', path_to_file)
+        self._stop_word_manager.notify('init')
+
+    def _run(self):
+        for word  in self._storage_manager.notify('words'):
+            if not self._stop_word_manager.notify('is_stop_word', word):
+                self._word_freq_manager.notify('increment_count', word )
+
+        word_freqs = self._word_freq_manager.notify('sorted')        
+        print_word_freqs(word_freqs)        
+
+
+if __name__ == '__main__':
+    sm = ScenarioManager()
+    sm.notify('init', testfilepath)
+    sm.notify('run')

A 代码模式/15 工程化/2 松耦合/2 消息链/3 消息链.py

@@ -0,0 +1,102 @@
+'''
+后续组件挂载到前序组件后续链上
+仅提供 self.next_observer 的抽象关系
+后续组件接到指令和数据，自己决定动作
+
+理论上每个组件可以参与到多个生产队列
+
+本例使用了类来封装消息，相对于字符串理论上提供了更丰富的扩展可能
+这是一个示例性质的原型，具体环境下需要调整
+'''
+
+from collections import Counter  
+from typing import List,  Dict  
+from cppy.cp_util import *
+
+# 定义消息类型  
+class Message:  
+    def __init__(self, data):  
+        self.data = data  
+  
+class TokenizedText(Message):  
+    pass  
+  
+class FilteredText(Message):  
+    pass  
+  
+class WordFrequency(Message):  
+    pass  
+  
+# 定义观察者接口  
+class Observer:  
+    def notify(self, message: Message):  
+        pass  
+  
+# 切词订阅者  
+class TokenizerSubscriber(Observer):  
+    def __init__(self, next_observer: Observer):  
+        self.next_observer = next_observer  
+      
+    def notify(self, message: Message):  
+        if not isinstance(message.data, str):  
+            return  
+        tokenized_text = re_split(message.data)  
+        self.next_observer.notify(TokenizedText(tokenized_text))  
+  
+# 停用词订阅者  
+class StopWordsRemoverSubscriber(Observer):  
+    def __init__(self, next_observer: Observer, stop_words: List[str]):  
+        self.next_observer = next_observer  
+        self.stop_words = set(stop_words)  
+      
+    def notify(self, message: Message):  
+        if not isinstance(message, TokenizedText):  
+            return  
+        filtered_text = [word for word in message.data if word not in self.stop_words and len(word)>2 ]  
+        self.next_observer.notify(FilteredText(filtered_text))  
+  
+# 词频统计订阅者  
+class WordFrequencyCalculatorSubscriber(Observer):  
+    def __init__(self, next_observer: Observer):  
+        self.next_observer = next_observer          
+
+    def notify(self, message: Message):          
+        if not isinstance(message, FilteredText):  
+            return  
+        word_freq = Counter(message.data)          
+        self.next_observer.notify( WordFrequency(word_freq) )
+        
+  
+# 输出前N个词订阅者  
+class TopNWordsDisplaySubscriber(Observer):  
+    def __init__(self, n: int):  
+        self.n = n  
+      
+    def notify(self, message: Message):  
+        if not isinstance(message, WordFrequency):  
+            return          
+        print_word_freqs( message.data.most_common(self.n) )
+        
+  
+# 模拟发布者  
+def publish_text(text: str, observers: List[Observer]):  
+    for observer in observers:  
+        observer.notify(Message(text))  
+  
+# 主函数  
+def main():  
+    text = read_file()
+    
+    stop_words = get_stopwords()
+  
+    # 创建订阅者链  
+    display_subscriber = TopNWordsDisplaySubscriber( n=10 )  
+    freq_subscriber = WordFrequencyCalculatorSubscriber(display_subscriber)  
+    stop_words_subscriber = StopWordsRemoverSubscriber(freq_subscriber, stop_words)  
+    tokenizer_subscriber = TokenizerSubscriber(stop_words_subscriber)  
+  
+    # 发布文本  
+    publish_text(text, [tokenizer_subscriber])  
+  
+if __name__ == "__main__":  
+    main()

A 代码模式/15 工程化/2 松耦合/2 消息链/4 消息队列.py

@@ -1,9 +1,10 @@
+################ 待整理 
 ''' 
-每个对象运行于独立线程，每个对象只暴露一个队伍消息队列的接口。
-各个组件完全不能互操作，仅依靠队列发消息进行协作/
-多线程模式下消息队列来共享数据空间。相比较传统数据库共享方式访问更快
+多线程各个模块比较乱的但是协作序贯的完成了数据处理
+各个组件完全不能互操作，仅依靠队列发消息进行协作
 
-这种模式，改造下适合跨进程系统的后台响应对象设计.
+适合环节多，数据可分块，有IO-计算性能设计考量要求，让各个模块自己适应调整
+在某些情况下，可以避免复杂的控制流设计，使代码简洁
 '''
 
 from threading import Thread

A 代码模式/15 工程化/2 松耦合/4 插件/plugin.py

@@ -1,7 +1,3 @@
-
-# 插件模式提供一种个别扩展性开发和系统核心开发无关的松耦合结构。
-# 简单说，第三方开发者在没有核心框架源码下也能扩展或者改造系统功能
-
 import configparser, importlib.machinery
 from cppy.cp_util import *
 

A 代码模式/15 工程化/3 类型申明/参数类型申明.py

@@ -1,7 +1,3 @@
-
-# Python 作为弱类型语言希望拥有强类型语言类似的规范工整工程性的优点，牺牲一些代码的自由度。
-# 可以理解为更好的代码注释和更多的工程约束 。
-
 import cppy.cp_util as util
 
 

A 代码模式/15 工程化/4 对象接口/tf-14B.py

@@ -1,13 +1,9 @@
 import abc, re
 from cppy.cp_util import *
 
-'''
-如果在基类层面上进行设计，这样就能获得多个子类同一层面上的抽象操作。
-在一些语言里面，这种模式叫基于接口的编程，接口要求纯抽象类。
-Python 因为对象协议的缘故，所以抽象接口设计不是必须。
-沿用抽象接口设计是为工程性规整 。
-'''
-
+#
+# 接口
+#
 class IDataStorage (metaclass=abc.ABCMeta):  
     @abc.abstractmethod
     def words(self):
@@ -78,7 +74,8 @@ IWordFrequencyCounter.register(subclass=WordFrequencyManager)
 # 应用类
 #
 class WordFrequencyController:
-    def __init__(self, path_to_file):        
+    def __init__(self, path_to_file):
+        # self._storage = DataStorageManager1(path_to_file)
         self.storage = DataStorageManager2(path_to_file)
         self.stop_word_manager = StopWordManager()
         self.word_freq_counter = WordFrequencyManager()

A 代码模式/15 工程化/5 对象属性 .py

@@ -3,9 +3,6 @@ from dataclasses import dataclass
 from collections import Counter
 import re
 
-# 对象属性是现代 Python编程喜欢的风格
-# 这里使用了dataclass来简化代码 
-
 @dataclass
 class WordFrequency:
     text: str

A 代码模式/16 其它/2 异常/1 软件不能挂掉.py

@@ -2,7 +2,7 @@ import re, operator, string
 from cppy.cp_util import *
 
 #
-# 遇到异常，给出一个能让程序继续运行下去的默认值
+# The functions
 #
 def extract_words(path_to_file):
     try:

A 代码模式/16 其它/2 异常/2 时间停止在那一刻.py

@@ -1,9 +1,7 @@
 from collections import Counter
 from cppy.cp_util import *
 
-#
-# 遇到异常，退出程序
-#
+
 def extract_words(path_to_file):
     assert(type(path_to_file) is str), "Must be a string!" 
     assert(path_to_file), "Must be a non-empty string!" 

A 代码模式/16 其它/2 异常/3 所有错误的应对.py

@@ -1,8 +1,6 @@
 from cppy.cp_util import *
 
-#
-# 用断言从事发点给出出错的准确信息
-#
+
 def extractWords(path_to_file):
     assert(type(path_to_file) is str), "Must be a string" 
     assert(path_to_file), "Must be a non-empty string"         

A 代码模式/16 其它/3 状态机/81A.py

@@ -2,12 +2,6 @@
 import cppy.cp_util as util
 from collections import Counter
 
-'''
-状态机是计算机程序运行的基础理论。
-使用状态机的风格来处理文件并计算词频，我们可以将整个过程分解为一系列状态转移。
-每个状态代表处理过程中的一个阶段，比如“读取文件”、“分割单词”和“计算词频”等。
-这种方法在Python中并不常见，但它展示了如何使用状态机来管理程序的状态和流程    
-'''
 
 class WordFrequencyStateMachine:
     def __init__(self, file_path):

A 代码模式/readme.MD

@@ -0,0 +1,4 @@
+
+
+## 任务
+本项目的主要功能任务：做文本文件的分词，过滤常见词，求词频，并排序输出。

A 动机与模式/13 函数式风格/readme.md

@@ -1 +0,0 @@
-尾调用方式调用函数做连续的数据处理。代码好读，方便栈资源优化（一些语言已经实现了这种调用方式释放前面的函数栈，但Python当前还没做如此优化）。

A 动机与模式/14 异常/readme.md

@@ -1 +0,0 @@
-异常主要发生在参数传递和代码块执行过程。一种原则是：软件不能挂掉。检查参数合理性、检查代码块执行可能的错误，并进行合理结果补齐，保持程序继续运行【 1 软件不能挂掉 】，另外一种情况是发生异常就抛出然后终止程序【 2 时间停止在那一刻 】，或者由上层函数接住，集中统一处理。【 3 预判可能的错误 】。

A 动机与模式/15 工程化考虑/2 松耦合/1 解耦模式/1 模板.py

@@ -1,64 +0,0 @@
-'''
-模板方法模式 (Template Method Pattern)
-
-适用场景：定义词频统计的整体流程骨架，允许子类自定义某些步骤（如分词或输出格式）。
-
-如何应用：
-定义一个抽象类 WordFrequencyAnalyzer，包含固定的流程：读取文件 -> 分词 -> 统计 -> 输出。
-具体步骤（如分词或输出）由子类实现。
-
-优点：保证流程一致，子类只关注具体实现。
-
-知识点：  
-理解解继承与抽象类的使用。
-如何定义固定流程，允许子类灵活实现。强调流程与实现的分离。
-'''
-
-from abc import ABC, abstractmethod
-from typing import List, Dict
-from collections import Counter
-
-class WordFrequencyAnalyzer(ABC):
-    def analyze(self, directory: str, top_n: int = 10):
-        texts = self.read_texts(directory)
-        word_counts = self.count_words(texts)
-        self.print_results(word_counts, top_n)
-
-    @abstractmethod
-    def read_texts(self, directory: str) -> List[str]:
-        pass
-
-    @abstractmethod
-    def count_words(self, texts: List[str]) -> Dict[str, int]:
-        pass
-
-    @abstractmethod
-    def print_results(self, word_counts: Dict[str, int], top_n: int):
-        pass
-
-class SimpleAnalyzer(WordFrequencyAnalyzer):
-    def read_texts(self, directory: str) -> List[str]:
-        import os
-        texts = []
-        for filename in os.listdir(directory):
-            if filename.endswith(".txt"):
-                with open(os.path.join(directory, filename), "r", encoding="utf-8") as f:
-                    texts.append(f.read())
-        return texts
-
-    def count_words(self, texts: List[str]) -> Dict[str, int]:
-        import jieba
-        word_counts = Counter()
-        for text in texts:
-            words = jieba.cut(text)
-            word_counts.update([word for word in words if word.strip()])
-        return dict(word_counts)
-
-    def print_results(self, word_counts: Dict[str, int], top_n: int):
-        top_words = sorted(word_counts.items(), key=lambda x: x[1], reverse=True)[:top_n]
-        for word, count in top_words:
-            print(f"{word}: {count}")
-
-# 使用
-analyzer = SimpleAnalyzer()
-analyzer.analyze("data", 10)

A 动机与模式/15 工程化考虑/2 松耦合/1 解耦模式/2 策略-工厂.py

@@ -1,107 +0,0 @@
-# -*- coding: utf-8 -*-
-
-'''
-设计说明
-策略模式：定义 Tokenizer 接口，允许不同分词策略（如中文分词、英文分词）。
-工厂模式：TokenizerFactory 创建分词器，隔离具体实现。
-解耦：数据读取（DataReader）、词频统计（WordCounter）、结果输出（ResultPrinter）分离，各自独立。
-可扩展：支持添加新分词策略或输出格式。
-
-解耦：
-DataReader 只负责读取文件，与分词和统计无关。
-WordCounter 依赖抽象的 Tokenizer，不关心具体分词实现。
-ResultPrinter 只处理输出，格式可扩展（如输出到文件）。
-
-策略模式：
-Tokenizer 接口允许切换分词策略（如添加英文分词器）。
-ChineseTokenizer 是具体实现，易于替换。
-
-工厂模式：
-TokenizerFactory 封装分词器创建逻辑，调用方无需知道具体类。
-
-简洁性：代码结构清晰，模块职责单一。
-
-扩展性：
-可添加新分词器（如英文 EnglishTokenizer）。
-可扩展 ResultPrinter 支持不同输出格式。
-
-实践：可尝试添加英文分词器或新输出格式（如 CSV）。
-'''
-
-import os
-import jieba
-from collections import Counter
-from abc import ABC, abstractmethod
-from typing import List, Dict
-
-# 策略接口：分词器
-class Tokenizer(ABC):
-    @abstractmethod
-    def tokenize(self, text: str) -> List[str]:
-        pass
-
-# 具体策略：中文分词（使用 jieba）
-class ChineseTokenizer(Tokenizer):
-    def tokenize(self, text: str) -> List[str]:
-        return [word for word in jieba.cut(text) if word.strip()]
-
-# 工厂：创建分词器
-class TokenizerFactory:
-    @staticmethod
-    def create_tokenizer(language: str = "chinese") -> Tokenizer:
-        if language == "chinese":
-            return ChineseTokenizer()
-        raise ValueError(f"Unsupported language: {language}")
-
-# 数据读取模块
-class DataReader:
-    def __init__(self, directory: str):
-        self.directory = directory
-
-    def read_texts(self) -> List[str]:
-        texts = []
-        for filename in os.listdir(self.directory):
-            if filename.endswith(".txt"):
-                with open(os.path.join(self.directory, filename), "r", encoding="utf-8") as f:
-                    texts.append(f.read())
-        return texts
-
-# 词频统计模块
-class WordCounter:
-    def __init__(self, tokenizer: Tokenizer):
-        self.tokenizer = tokenizer
-
-    def count_words(self, texts: List[str]) -> Dict[str, int]:
-        word_counts = Counter()
-        for text in texts:
-            words = self.tokenizer.tokenize(text)
-            word_counts.update(words)
-        return dict(word_counts)
-
-# 结果输出模块
-class ResultPrinter:
-    @staticmethod
-    def print_top_words(word_counts: Dict[str, int], top_n: int = 10):
-        top_words = sorted(word_counts.items(), key=lambda x: x[1], reverse=True)[:top_n]
-        for word, count in top_words:
-            print(f"{word}: {count}")
-
-# 主程序
-def main():
-    # 配置
-    data_dir = "data"
-    top_n = 10
-
-    # 创建组件
-    reader = DataReader(data_dir)
-    tokenizer = TokenizerFactory.create_tokenizer("chinese")
-    counter = WordCounter(tokenizer)
-    printer = ResultPrinter()
-
-    # 执行流程
-    texts = reader.read_texts()
-    word_counts = counter.count_words(texts)
-    printer.print_top_words(word_counts, top_n)
-
-if __name__ == "__main__":
-    main()

A 动机与模式/15 工程化考虑/2 松耦合/1 解耦模式/3 观察者.py

@@ -1,105 +0,0 @@
-'''
-把观察者挂到自己的处理队列上
-'''
-
-import os
-import re
-import threading
-from queue import Queue
-from collections import Counter
-from abc import ABC, abstractmethod
-
-# 观察者接口
-class Observer(ABC):
-    @abstractmethod
-    def update(self, word_counts: Counter):
-        pass
-
-# 具体观察者：打印前 10 高频词
-class PrintTopWordsObserver(Observer):
-    def update(self, word_counts: Counter):
-        print("Top 10 高频词:")
-        for word, count in word_counts.most_common(10):
-            print(f"{word}: {count}")
-
-# 具体观察者：保存词频到文件
-class SaveToFileObserver(Observer):
-    def __init__(self, output_file):
-        self.output_file = output_file
-
-    def update(self, word_counts: Counter):
-        try:
-            with open(self.output_file, 'w', encoding='utf-8') as f:
-                for word, count in word_counts.most_common(10):
-                    f.write(f"{word}: {count}\n")
-            print(f"词频已保存到 {self.output_file}")
-        except Exception as e:
-            print(f"保存失败: {e}")
-
-# 词频统计器（主题）
-class WordFrequencyCounter:
-    def __init__(self):
-        self.observers = []
-        self.counter = Counter()
-        self.queue = Queue()
-        self.lock = threading.Lock()
-
-    def add_observer(self, observer: Observer):
-        self.observers.append(observer)
-
-    def remove_observer(self, observer: Observer):
-        self.observers.remove(observer)
-
-    def notify_observers(self):
-        for observer in self.observers:
-            observer.update(self.counter)
-
-    def process_file(self):
-        while True:
-            try:
-                file_path = self.queue.get_nowait()
-            except:
-                break
-            try:
-                with open(file_path, 'r', encoding='utf-8') as f:
-                    text = f.read().lower()
-                    words = re.findall(r'\b\w+\b', text)
-                    with self.lock:
-                        self.counter.update(words)
-            except Exception as e:
-                print(f"Error processing {file_path}: {e}")
-            finally:
-                self.queue.task_done()
-
-    def count_words(self, files, num_threads=4):
-        # 将文件路径放入队列
-        for file_path in files:
-            self.queue.put(file_path)
-        
-        # 创建并启动线程
-        threads = [threading.Thread(target=self.process_file) for _ in range(num_threads)]
-        for t in threads:
-            t.start()
-        for t in threads:
-            t.join()
-
-        # 通知所有观察者
-        self.notify_observers()
-
-def main():
-    # 获取文件列表
-    data_dir = 'data'
-    files = [os.path.join(data_dir, f) for f in os.listdir(data_dir) if f.endswith('.txt')]
-    
-    # 创建词频统计器
-    counter = WordFrequencyCounter()
-    
-    # 添加观察者
-    counter.add_observer(PrintTopWordsObserver())
-    counter.add_observer(SaveToFileObserver("word_frequency.txt"))
-    
-    # 统计词频并通知观察者
-    counter.count_words(files)
-
-if __name__ == '__main__':
-    main()

A 动机与模式/15 工程化考虑/2 松耦合/4 装饰器/1 装饰器.py

@@ -1,52 +0,0 @@
-'''
-装饰器模式 (Decorator Pattern)
-
-适用场景：动态为词频统计添加功能，如过滤停用词、忽略标点、或添加词性标注。
-如何应用：
-定义核心 WordCounter 接口，负责基本词频统计。
-
-使用装饰器类动态添加功能，如 StopWordFilterDecorator 过滤停用词。
-
-优点：功能扩展不修改核心逻辑，符合开闭原则。
-
-知识点：
-如何动态扩展功能，保持核心代码不变。
-体会组合优于继承的原则。
-
-'''
-
-from abc import ABC, abstractmethod
-from typing import List, Dict
-from collections import Counter
-
-class WordCounter(ABC):
-    @abstractmethod
-    def count_words(self, texts: List[str]) -> Dict[str, int]:
-        pass
-
-class BasicWordCounter(WordCounter):
-    def count_words(self, texts: List[str]) -> Dict[str, int]:
-        import jieba
-        word_counts = Counter()
-        for text in texts:
-            words = jieba.cut(text)
-            word_counts.update([word for word in words if word.strip()])
-        return dict(word_counts)
-
-class WordCounterDecorator(WordCounter, ABC):
-    def __init__(self, counter: WordCounter):
-        self.counter = counter
-
-class StopWordFilterDecorator(WordCounterDecorator):
-    def __init__(self, counter: WordCounter, stop_words: List[str]):
-        super().__init__(counter)
-        self.stop_words = stop_words
-
-    def count_words(self, texts: List[str]) -> Dict[str, int]:
-        word_counts = self.counter.count_words(texts)
-        return {word: count for word, count in word_counts.items() if word not in self.stop_words}
-
-# 使用
-counter = StopWordFilterDecorator(BasicWordCounter(), stop_words=["的", "了", "是"])
-word_counts = counter.count_words(["这是一段测试文本。"])
-print(word_counts)

A 动机与模式/readme.MD

@@ -1,8 +0,0 @@
-
-
-## 任务
-本项目的主要功能任务：对指定文件，读取文本，在停用词过滤后，按降序输出前10个高频词
-
-我们研究这个问题的解法在各种情景下的表现形式 。
-
-将 cppy整个目录转移到 anaconda3\Lib\site-packages 下面。里面放了一些共同的代码片段（函数），这样使得模式代码更短小简洁，便于更能集中理解各种模式的核心思想 。

B 面向对象设计模式/readme.md

@@ -1,24 +0,0 @@
-
-
-# 目标
-本节使用一个书城的各方面业务需求来展示面向对象的常见设计模式 。
-
-
-# 任务
-背景假设为一个综合书城，提供线上线下购买，还经营一个书吧、一个报告厅。
-
-
-# 说明
-面向对象的模式是把编程过程中的一些思路固定化，并给一个名字方便理解 。
-它是软件工程中一组经过验证的、可重复使用的代码写法 。
-所以，模式不是语法，而是编程思路 。
-这样做的好处是，统一大家的代码形式，提高代码可读性、可维护性、可扩展性 。
-那为啥，面向过程没有这么做 ？
-是因为这个思维提炼过程，充分利用了面向对象语言的特性：封装、继承、多态 。
-面向过程语言，没有这些特性，所以，面向过程程序设计一般不谈设计模式 。
-
-因为 Python 对象协议的机制，多态、接口概念发生了根本变化 。使得一些C++、Java 的模式没用了 。比如 “ 原型模式（Prototype）” ，可以使用copy.deepcopy()非常简便来创建 。另外，很多模式中继承关系也没必要了。但，下面很多示例中依旧保持了基类 。一是致敬经典，二是起到一个工程上更工整和代码注释的作用 。    
-
-
-# 应用场景
-面向对象设计模式在管理信息系统和图形用户界面系统应用比较广泛 。

B 高性能模式/000 普通做法.py

@@ -0,0 +1,74 @@
+"""
+根据提供的关键词列表，爬取天水市人民政府网站上指定日期内与关键词相关的新闻的标题，并将其存储至数据库中。
+
+考虑到相关因素，因此本代码只爬取前10页的新闻内容，即最多100条新闻作为测试。
+
+此方法为普通做法，即使用requests库通过Post请求爬取网页内容，再使用json提取新闻内容。
+
+注意：本代码中的关键词列表默认为['灾害']，日期范围默认为2018年1月1日至2018年12月31日。
+
+Args:
+    keywords: 用于搜索新闻的关键词列表
+    begin_date: 开始日期，用于搜索
+    end_date: 结束日期，用于搜索
+    size: 一次请求返回的新闻或政策的最大数量
+
+Examples:
+    ```
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)
+    ```
+"""
+import util
+import logging
+from typing import List
+
+import tqdm
+
+
+@util.timeit
+def main(keywords: List[str], begin_date: str, end_date: str, size: int = 10):
+    """
+    爬取与提供的关键词列表相关的新闻.
+
+    Args:
+        keywords: 用于搜索新闻的关键词列表
+        begin_date: 开始日期，用于搜索
+        end_date: 结束日期，用于搜索
+        size: 一次请求返回的新闻或政策的最大数量
+    """
+    logging.basicConfig(level=logging.INFO,
+                        format="%(asctime)s - %(levelname)s - %(message)s",
+                        datefmt="%Y-%m-%d %H:%M:%S",
+                        filename='log.txt',
+                        encoding='utf-8')
+
+    logging.info("开始运行普通爬取")
+
+    spider = util.Spider(keywords=keywords,
+                         begin_date=begin_date,
+                         end_date=end_date,
+                         size=size)
+
+    pbar = tqdm.tqdm(total=size * 10, desc='普通爬取进度', unit='条', ncols=80)
+    title_list = []
+    for keyword in keywords:
+        for current in range(1, 11):
+            logging.info(f'keyword: {keyword}, current: {current}')
+            config = spider.get_config(keyword, current)
+            data = spider.fetch(config)
+            title_list += spider.parse(data)
+            pbar.update(size)
+
+    spider.save(title_list)
+    pbar.close()
+    logging.info("爬取完成")
+
+
+if __name__ == "__main__":
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)

B 高性能模式/010 多进程.py

@@ -0,0 +1,86 @@
+"""
+根据提供的关键词列表，爬取天水市人民政府网站上指定日期内与关键词相关的新闻的标题，并将其存储至数据库中。
+
+考虑到相关因素，因此本代码只爬取前10页的新闻内容，即最多100条新闻作为测试。
+
+此方法为多进程做法，即使用多进程并发爬取网页内容，再使用json提取新闻内容。
+
+注意：本代码中的关键词列表默认为['灾害']，日期范围默认为2018年1月1日至2018年12月31日。
+
+Args:
+    keywords: 用于搜索新闻的关键词列表
+    begin_date: 开始日期，用于搜索
+    end_date: 结束日期，用于搜索
+    size: 一次请求返回的新闻或政策的最大数量
+
+Examples:
+    ```
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)
+    ```
+"""
+import util
+import logging
+from typing import List
+import multiprocessing
+
+import tqdm
+
+lock = multiprocessing.Lock()
+
+
+@util.timeit
+def main(keywords: List[str], begin_date: str, end_date: str, size: int = 10):
+    """
+    爬取与提供的关键词列表相关的新闻.
+
+    Args:
+        keywords: 用于搜索新闻的关键词列表
+        begin_date: 开始日期，用于搜索
+        end_date: 结束日期，用于搜索
+        size: 一次请求返回的新闻或政策的最大数量
+    """
+    logging.basicConfig(level=logging.INFO,
+                        format="%(asctime)s - %(levelname)s - %(message)s",
+                        datefmt="%Y-%m-%d %H:%M:%S",
+                        filename='log.txt',
+                        encoding='utf-8')
+
+    logging.info("开始运行普通做法")
+
+    spider = util.Spider(keywords=keywords,
+                         begin_date=begin_date,
+                         end_date=end_date,
+                         size=size)
+
+    title_list = []
+    pbar = tqdm.tqdm(total=size * 10, desc='多进程爬取进度', unit='条', ncols=80)
+
+    with multiprocessing.Pool(processes=5) as pool:
+        results = []
+        for keyword in keywords:
+            for current in range(1, 11):
+                logging.info(f'keyword: {keyword}, current: {current}')
+                config = spider.get_config(keyword, current)
+                results.append(pool.apply_async(spider.fetch, (config, )))
+
+        for result in results:
+            data = result.get()
+            title_list += spider.parse(data)
+
+            lock.acquire()
+            pbar.update(size)
+            lock.release()
+
+    spider.save(title_list)
+    pbar.close()
+    logging.info("爬取完成")
+
+
+if __name__ == "__main__":
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)

B 高性能模式/020 多线程.py

@@ -0,0 +1,89 @@
+"""
+根据提供的关键词列表，爬取天水市人民政府网站上指定日期内与关键词相关的新闻的标题，并将其存储至数据库中。
+
+考虑到相关因素，因此本代码只爬取前10页的新闻内容，即最多100条新闻作为测试。
+
+此方法为多线程做法，即使用多线程并行爬取网页内容，再使用json提取新闻内容。
+
+注意：本代码中的关键词列表默认为['灾害']，日期范围默认为2018年1月1日至2018年12月31日。
+
+Args:
+    keywords: 用于搜索新闻的关键词列表
+    begin_date: 开始日期，用于搜索
+    end_date: 结束日期，用于搜索
+    size: 一次请求返回的新闻或政策的最大数量
+
+Examples:
+    ```
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)
+    ```
+"""
+
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import threading
+import util
+import logging
+from typing import List
+
+import tqdm
+
+lock = threading.Lock()
+
+
+@util.timeit
+def main(keywords: List[str], begin_date: str, end_date: str, size: int = 10):
+    """
+    爬取与提供的关键词列表相关的新闻.
+
+    Args:
+        keywords: 用于搜索新闻的关键词列表
+        begin_date: 开始日期，用于搜索
+        end_date: 结束日期，用于搜索
+        size: 一次请求返回的新闻或政策的最大数量
+    """
+    logging.basicConfig(level=logging.INFO,
+                        format="%(asctime)s - %(levelname)s - %(message)s",
+                        datefmt="%Y-%m-%d %H:%M:%S",
+                        filename='log.txt',
+                        encoding='utf-8')
+
+    logging.info("开始运行多线程爬取")
+
+    spider = util.Spider(keywords=keywords,
+                         begin_date=begin_date,
+                         end_date=end_date,
+                         size=size)
+
+    pbar = tqdm.tqdm(total=size * 10, desc='多线程爬取进度', unit='条', ncols=80)
+    title_list = []
+    tasks = []
+    with ThreadPoolExecutor(max_workers=5) as executor:
+        for keyword in keywords:
+            for current in range(1, 11):
+                logging.info(f'keyword: {keyword}, current: {current}')
+
+                config = spider.get_config(keyword, current)
+                future = executor.submit(spider.fetch, config)
+                tasks.append(future)
+                # 更新进度条
+                lock.acquire()
+                pbar.update(size)
+                lock.release()
+
+        for future in as_completed(tasks):
+            data = future.result()
+            title_list += spider.parse(data)
+
+    spider.save(title_list)
+    pbar.close()
+    logging.info("爬取完成")
+
+
+if __name__ == "__main__":
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)

B 高性能模式/030 协程.py

@@ -0,0 +1,89 @@
+"""
+根据提供的关键词列表，爬取天水市人民政府网站上指定日期内与关键词相关的新闻的标题，并将其存储至数据库中。
+
+考虑到相关因素，因此本代码只爬取前10页的新闻内容，即最多100条新闻作为测试。
+
+此方法为协程做法，即使用gevent库通过协程并发爬取网页内容，再使用json提取新闻内容。
+
+注意：本代码中的关键词列表默认为['灾害']，日期范围默认为2018年1月1日至2018年12月31日。
+
+Args:
+    keywords: 用于搜索新闻的关键词列表
+    begin_date: 开始日期，用于搜索
+    end_date: 结束日期，用于搜索
+    size: 一次请求返回的新闻或政策的最大数量
+
+Examples:
+    ```
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)
+    ```
+"""
+
+import gevent
+from gevent import monkey
+
+# 打补丁，使标准库能够与gevent协同工作
+monkey.patch_all()
+
+import util
+import logging
+from typing import List
+
+import tqdm
+
+
+@util.timeit
+def main(keywords: List[str], begin_date: str, end_date: str, size: int = 10):
+    """
+    爬取与提供的关键词列表相关的新闻.
+
+    Args:
+        keywords: 用于搜索新闻的关键词列表
+        begin_date: 开始日期，用于搜索
+        end_date: 结束日期，用于搜索
+        size: 一次请求返回的新闻或政策的最大数量
+    """
+    logging.basicConfig(level=logging.INFO,
+                        format="%(asctime)s - %(levelname)s - %(message)s",
+                        datefmt="%Y-%m-%d %H:%M:%S",
+                        filename='log.txt',
+                        encoding='utf-8')
+
+    logging.info("开始运行协程爬取")
+
+    spider = util.Spider(keywords=keywords,
+                         begin_date=begin_date,
+                         end_date=end_date,
+                         size=size)
+
+    pbar = tqdm.tqdm(total=size * 10, desc='协程爬取进度', unit='条', ncols=80)
+    title_list = []
+
+    def fetch_and_parse(keyword, current):
+        logging.info(f'keyword: {keyword}, current: {current}')
+        config = spider.get_config(keyword, current)
+        data = spider.fetch(config)
+        titles = spider.parse(data)
+        title_list.extend(titles)
+        pbar.update(size)
+
+    jobs = [
+        gevent.spawn(fetch_and_parse, keyword, current) for keyword in keywords
+        for current in range(1, 11)
+    ]
+
+    gevent.joinall(jobs)
+
+    spider.save(title_list)
+    pbar.close()
+    logging.info("爬取完成")
+
+
+if __name__ == "__main__":
+    main(keywords=['灾害'],
+         begin_date='2018-01-01',
+         end_date='2018-12-31',
+         size=10)

B 高性能模式/040 异步.py

@@ -0,0 +1,85 @@
+"""
+根据提供的关键词列表，爬取天水市人民政府网站上指定日期内与关键词相关的新闻的标题，并将其存储至数据库中。
+
+考虑到相关因素，因此本代码只爬取前10页的新闻内容，即最多100条新闻作为测试。
+
+此方法为多线程做法，即使用异步并行爬取网页内容，再使用json提取新闻内容。
+
+注意：本代码中的关键词列表默认为['灾害']，日期范围默认为2018年1月1日至2018年12月31日。
+
+Args:
+    keywords: 用于搜索新闻的关键词列表
+    begin_date: 开始日期，用于搜索
+    end_date: 结束日期，用于搜索
+    size: 一次请求返回的新闻或政策的最大数量
+
+Examples:
+    ```
+    asyncio.run(
+        main_async(keywords=['灾害'],
+                   begin_date='2018-01-01',
+                   end_date='2018-12-31',
+                   size=10))
+    ```
+"""
+
+import asyncio
+import util
+import logging
+from typing import List
+import tqdm
+
+
+@util.timeit_async
+async def main_async(keywords: List[str],
+                     begin_date: str,
+                     end_date: str,
+                     size: int = 10):
+    """
+    使用异步方式爬取与提供的关键词列表相关的新闻.
+
+    Args:
+        keywords: 用于搜索新闻的关键词列表
+        begin_date: 开始日期，用于搜索
+        end_date: 结束日期，用于搜索
+        size: 一次请求返回的新闻或政策的最大数量
+    """
+    logging.basicConfig(level=logging.INFO,
+                        format="%(asctime)s - %(levelname)s - %(message)s",
+                        datefmt="%Y-%m-%d %H:%M:%S",
+                        filename='log.txt',
+                        encoding='utf-8')
+
+    logging.info("开始运行异步爬取")
+
+    spider = util.Spider(keywords=keywords,
+                         begin_date=begin_date,
+                         end_date=end_date,
+                         size=size)
+
+    pbar = tqdm.tqdm(total=size * 10, desc='异步爬取进度', unit='条', ncols=80)
+    title_list = []
+    tasks = []
+    for keyword in keywords:
+        for current in range(1, 11):
+            logging.info(f'keyword: {keyword}, current: {current}')
+            config = spider.get_config(keyword, current)
+            task = asyncio.create_task(spider.fetch_async(config))
+            tasks.append(task)
+
+    for task in asyncio.as_completed(tasks):
+        data = await task
+        title_list += spider.parse(data)
+        pbar.update(size)
+
+    spider.save(title_list)
+    pbar.close()
+    logging.info("爬取完成")
+
+
+if __name__ == "__main__":
+    asyncio.run(
+        main_async(keywords=['灾害'],
+                   begin_date='2018-01-01',
+                   end_date='2018-12-31',
+                   size=10))

B 高性能模式/readme.md

@@ -0,0 +1,25 @@
+
+
+# 目标
+本节使用一个爬虫任务来展示如何追求代码的性能 。
+充分理解线程、协程、进程、同步、异步、阻塞、非阻塞等概念，并能够根据具体场景选择合适的并发模型。
+主线问题：如何解决IO和计算速度不匹配、如何任务分解、分发和协作 。
+
+# 任务
+
+
+# 讨论分析
+普通做法连续进行了五次测试，时间分别为34.231s、34.091s、34.164s、34.226s、33.958s，平均时间为34.134s
+多进程（进程数=5）连续进行了五次测试，时间分别为7.719s、7.716s、7.690s、7.730s、7.711s，平均时间为7.7132s
+多线程（线程数=5）连续进行了五次测试，时间分别为7.185s、7.964s、6.983s、6.969s、7.035s，平均时间为7.2272s
+协程连续进行了五次测试，时间分别为3.775s、3.807s、3.733s、3.824s、3.744s，平均时间为3.776s
+异步连续进行了五次测试，时间分别为6.975s、7.675s、7.018s、7.032s、7.049s，平均时间为7.1498s
+注：为保证公平性，每一次Post请求后休眠3秒
+
+可以看出，协程的性能最好，普通做法的性能最差，多线程、多进程和异步的性能介于两者之间。
+考虑到多进程和多线程是故意开的5个进程和线程，而协程是单线程，所以协程的性能最好。
+另外，异步的性能最差，可能是由于异步的并发模型需要频繁地切换线程，导致性能下降。
+总的来说，协程的性能最好，多线程和多进程的性能介于两者之间，普通做法的性能最差。
+
+# 总结
+协程的性能最好，多线程和多进程的性能介于两者之间，普通做法的性能最差。