diff --git a/A 动机与模式/17 其它/对象属性 .py b/A 动机与模式/11 封装/2 对象封装/对象属性 .py similarity index 100% rename from A 动机与模式/17 其它/对象属性 .py rename to A 动机与模式/11 封装/2 对象封装/对象属性 .py diff --git a/A 动机与模式/12 语言特性/3 动态读写PyObject/3 反射/反射_函数.py b/A 动机与模式/12 语言特性/3 动态读写PyObject/2 反射/反射_函数.py similarity index 100% rename from A 动机与模式/12 语言特性/3 动态读写PyObject/3 反射/反射_函数.py rename to A 动机与模式/12 语言特性/3 动态读写PyObject/2 反射/反射_函数.py diff --git a/A 动机与模式/12 语言特性/3 动态读写PyObject/3 反射/反射_类.py b/A 动机与模式/12 语言特性/3 动态读写PyObject/2 反射/反射_类.py similarity index 100% rename from A 动机与模式/12 语言特性/3 动态读写PyObject/3 反射/反射_类.py rename to A 动机与模式/12 语言特性/3 动态读写PyObject/2 反射/反射_类.py diff --git a/A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/2 装饰器.py b/A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/2 装饰器.py deleted file mode 100644 index fe92b28..0000000 --- a/A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/2 装饰器.py +++ /dev/null @@ -1,34 +0,0 @@ -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 ) \ No newline at end of file diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/1 模板.py b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/1 模板.py new file mode 100644 index 0000000..cb6a8f4 --- /dev/null +++ b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/1 模板.py @@ -0,0 +1,64 @@ +''' +模板方法模式 (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) \ No newline at end of file diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/2 策略-工厂.py b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/2 策略-工厂.py new file mode 100644 index 0000000..6346f22 --- /dev/null +++ b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/2 策略-工厂.py @@ -0,0 +1,107 @@ +# -*- 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() \ No newline at end of file diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/观察者 1.py b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/3 观察者.py similarity index 100% rename from A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/观察者 1.py rename to A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/3 观察者.py diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/注册回调 1.py b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/注册回调 1.py deleted file mode 100644 index 0a77f49..0000000 --- a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/注册回调 1.py +++ /dev/null @@ -1,78 +0,0 @@ -''' -主逻辑将关键动作开放为注册服务 -各个组件注册到事件管理器相应服务 -事件管理器按顺序执行注册的事件 -''' - -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) \ No newline at end of file diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/注册回调 2.py b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/注册回调 2.py deleted file mode 100644 index 5e6da70..0000000 --- a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/注册回调 2.py +++ /dev/null @@ -1,105 +0,0 @@ -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() \ No newline at end of file diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/观察者 2.py b/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/观察者 2.py deleted file mode 100644 index 23544df..0000000 --- a/A 动机与模式/13 工程化考虑/2 松耦合/1 解耦模式/观察者 2.py +++ /dev/null @@ -1,68 +0,0 @@ -''' -基类 Subject 提供注册和提醒注册上的对象提醒机制 -每个派生组件做完自己的事情，调用基类的 notify 方法 -这样就可以实现一个简单的观察者模式，也可以说是订阅模式 -因为函数和参数混杂在一起传递，使得各个模块的处理结构其实是 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() \ No newline at end of file diff --git a/A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/1 装饰器.py b/A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/1 装饰器.py new file mode 100644 index 0000000..3e7c5ce --- /dev/null +++ b/A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/1 装饰器.py @@ -0,0 +1,52 @@ +''' +装饰器模式 (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) \ No newline at end of file diff --git a/A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/3 参数类型检查.py b/A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/2 参数类型检查.py similarity index 100% rename from A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/3 参数类型检查.py rename to A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/2 参数类型检查.py diff --git a/A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/1 反射装饰.py b/A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/3 反射装饰.py similarity index 90% rename from A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/1 反射装饰.py rename to A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/3 反射装饰.py index e5ec529..f691f64 100644 --- a/A 动机与模式/12 语言特性/3 动态读写PyObject/2 装饰/1 反射装饰.py +++ b/A 动机与模式/13 工程化考虑/2 松耦合/4 装饰器/3 反射装饰.py @@ -1,6 +1,9 @@ import time import cppy.cp_util as util +''' +用反射实现装饰器的效果 +''' # 工具函数 def extract_words(path_to_file): diff --git a/C 高性能编程/case_股票分析.md b/C 高性能编程/案例/股票分析/case_股票分析.md similarity index 100% rename from C 高性能编程/case_股票分析.md rename to C 高性能编程/案例/股票分析/case_股票分析.md diff --git a/C 高性能编程/case_股票分析.py b/C 高性能编程/案例/股票分析/case_股票分析.py similarity index 100% rename from C 高性能编程/case_股票分析.py rename to C 高性能编程/案例/股票分析/case_股票分析.py diff --git a/C 高性能编程/词频统计/1 抽象并发.py b/C 高性能编程/案例/词频统计/1 抽象并发.py similarity index 100% rename from C 高性能编程/词频统计/1 抽象并发.py rename to C 高性能编程/案例/词频统计/1 抽象并发.py diff --git a/C 高性能编程/词频统计/2 多线程.py b/C 高性能编程/案例/词频统计/2 多线程.py similarity index 100% rename from C 高性能编程/词频统计/2 多线程.py rename to C 高性能编程/案例/词频统计/2 多线程.py diff --git a/C 高性能编程/词频统计/3 多进程.py b/C 高性能编程/案例/词频统计/3 多进程.py similarity index 100% rename from C 高性能编程/词频统计/3 多进程.py rename to C 高性能编程/案例/词频统计/3 多进程.py diff --git a/D 走向工业级代码/04 测试驱动的开发/config/settings.yaml b/D 走向工业级代码/04 测试驱动/config/settings.yaml similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/config/settings.yaml rename to D 走向工业级代码/04 测试驱动/config/settings.yaml diff --git a/D 走向工业级代码/04 测试驱动的开发/readme.MD b/D 走向工业级代码/04 测试驱动/readme.MD similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/readme.MD rename to D 走向工业级代码/04 测试驱动/readme.MD diff --git a/D 走向工业级代码/04 测试驱动的开发/src/core.py b/D 走向工业级代码/04 测试驱动/src/core.py similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/src/core.py rename to D 走向工业级代码/04 测试驱动/src/core.py diff --git a/D 走向工业级代码/04 测试驱动的开发/src/file_io.py b/D 走向工业级代码/04 测试驱动/src/file_io.py similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/src/file_io.py rename to D 走向工业级代码/04 测试驱动/src/file_io.py diff --git a/D 走向工业级代码/04 测试驱动的开发/tests/integration/test_pipeline.py b/D 走向工业级代码/04 测试驱动/tests/integration/test_pipeline.py similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/tests/integration/test_pipeline.py rename to D 走向工业级代码/04 测试驱动/tests/integration/test_pipeline.py diff --git a/D 走向工业级代码/04 测试驱动的开发/tests/performance/test_benchmark.py b/D 走向工业级代码/04 测试驱动/tests/performance/test_benchmark.py similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/tests/performance/test_benchmark.py rename to D 走向工业级代码/04 测试驱动/tests/performance/test_benchmark.py diff --git a/D 走向工业级代码/04 测试驱动的开发/tests/unit/test_tokenizer.py b/D 走向工业级代码/04 测试驱动/tests/unit/test_tokenizer.py similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/tests/unit/test_tokenizer.py rename to D 走向工业级代码/04 测试驱动/tests/unit/test_tokenizer.py diff --git a/D 走向工业级代码/04 测试驱动的开发/测试驱动开发.ipynb b/D 走向工业级代码/04 测试驱动/测试驱动开发.ipynb similarity index 100% rename from D 走向工业级代码/04 测试驱动的开发/测试驱动开发.ipynb rename to D 走向工业级代码/04 测试驱动/测试驱动开发.ipynb