521_2

.ipynb_checkpoints/readme-checkpoint.MD

@@ -1,14 +0,0 @@
-
-## 代码为啥要这样写，我要这样写代码
-
-A 代码模式
-用一个简单任务，展示各种软件工程需求（完成任务简单、可读性强、可复用高、维护成本低等）下的代码写法
-
-B 面向对象设计模式
-用一个业务场景复现面向对象的经典设计模式
-
-C 高性能模式
-考虑执行时间快，资源占用少的一些思路、办法和结论
-
-D 制造工业级代码
-问题同 A ，以构建工业级的代码为目标，用多种方式做了优化提升演示

C 高性能编程/readme.md

@@ -0,0 +1,44 @@
+
+从计算机系统结构的角度，提高 Python 任务执行速度的核心在于：减少解释器开销（编译/JIT）、提升并行性（多核/GPU）、优化内存访问（缓存友好）、降低 I/O 瓶颈以及适配硬件特性等。当前主要办法如下：
+
+
+### 计算单元层面利用多核并行计算
+对于 CPU 密集型任务，使用多进程，每个进程拥有独立的 Python 解释器和内存空间，运行在独立的内核上，实现并行计算。
+
+
+### I/O 层面减少等待时间
+- 异步编程：针对I/O请求等待，手工实现任务切换，完成并发执行.
+- 多线程：解释器自动完成I/O请求的线程切换 。
+- 批量处理，减少I/O请求数量 。
+
+
+### 编译层面减少解释器开销
+- 使用 JIT 编译器：Just-In-Time（JIT）编译可以在运行时将Python代码编译成机器码，从而提升执行速度 。PyPy 是一种替代 CPython 的实现，PyPy 的 JIT 引擎可以分析代码执行路径，优化频繁调用的函数，充分利用处理器架构。
+- **Cython 编译**：Cython 允许开发者为 Python 代码添加 C 类型注解，并编译为 C 代码，再由 C 编译器生成机器码。Cython 特别适合静态类型优化场景。
+
+
+### 利用Python的解释器特性
+-  **使用内置数据类型和函数**：内置的数据类型（如列表、字典、集合等）和函数通常经过高度优化。
+-  **选择合适的数据结构**：例如，一些类型执行一些操作更快，一些类型更省空间。
+-  **减少全局变量的使用**：访问全局变量通常比局部变量慢，因为它们需要在更大的作用域中查找。
+-  **减少函数调用**，可降低堆栈操作开销。
+-  使用列表推导式替代循环，降低频繁创建和销毁临时对象的开销。
+-  使用生成器而不是列表来处理大数据集，以减少内存占用。
+-  使用XX池或预分配资源。
+
+
+### 使用第三方高性能库
+-  NumPy/Pandas 用 C/C++ 编写并经过优化,使用连续内存块存储数据，向量化操作比显式的Python循环更高效。
+-  SIMD 指令加速，NumPy、Numba、Pandas/SciPy 都使用了 SIMD。Cython 可以直接用 C 代码使用 SIMD 。
+- `gzip` 模块可压缩数据，减少网络传输的数据量，提高网络传输速度。
+- `mmap` 模块实现内存映射文件，在处理超大文件、优化I/O性能以及进程间通信方面具有显著优势。
+- `functools.lru_cache` 缓存计算结果，避免重复计算 。
+
+
+### 使用性能分析工具
+如 cProfile 、Py-Spy、timeit 或 line_profiler
+
+## 总结
+具体实施时，应结合性能分析工具定位瓶颈，并根据任务特点选择合适的策略 。
+当然计算设备方面也可以简单提升：多机、更快的 CPU、更多核的CPU、更多的内存、更快的存储、增加 GPU/FPGA/TPU 。
+此外，随着Python社区的发展，新的技术和工具不断涌现，开发者应持续关注最新进展，以便更好地优化自己的代码 。

D Plus/.ipynb_checkpoints/01 特殊执行方式的语言特性.ipynb-checkpoint.ipynb

@@ -1,312 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "markdown",
-   "id": "86405617-889a-40c2-a895-7b51fb14b65d",
-   "metadata": {},
-   "source": [
-    "# 教学目标\n",
-    "\n",
-    "- 在词频统计案例中引入装饰器和函数式编程 。\n",
-    "- 分析这些特性和模式如何进一步优化代码质量（可读性、可维护性、可扩展性、复用性）。\n",
-    "- 探讨高级特性在案例中的适用性与局限性。"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "e6a6a633-d3af-4778-815c-4490dff5f624",
-   "metadata": {},
-   "source": [
-    "## 第一部分：引入装饰器\n",
-    "\n",
-    "装饰器可用于在不修改函数代码的情况下添加功能。适合日志记录、性能分析、错误处理等场景。"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "3a5c7d69-d445-4a9c-bb48-7fde0a36c646",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# 为 TextAnalyzer 类添加一个装饰器，用于记录方法执行时间。\n",
-    "\n",
-    "import os\n",
-    "import jieba\n",
-    "from collections import Counter\n",
-    "import yaml\n",
-    "import time\n",
-    "import functools\n",
-    "\n",
-    "def timing_decorator(func):\n",
-    "    \"\"\"装饰器：记录函数执行时间\"\"\"\n",
-    "    @functools.wraps(func)\n",
-    "    def wrapper(*args, **kwargs):\n",
-    "        start_time = time.time()\n",
-    "        result = func(*args, **kwargs)\n",
-    "        end_time = time.time()\n",
-    "        print(f\"{func.__name__} took {end_time - start_time:.4f} seconds\")\n",
-    "        return result\n",
-    "    return wrapper\n",
-    "\n",
-    "class TextAnalyzer:\n",
-    "    def __init__(self, config_path='config.yaml'):\n",
-    "        with open(config_path, 'r', encoding='utf-8') as f:\n",
-    "            config = yaml.safe_load(f)\n",
-    "        self.data_dir = config['data_dir']\n",
-    "        self.top_n = config['top_n']\n",
-    "        self.stop_words_file = config['stop_words_file']\n",
-    "        self.output_file = config['output_file']\n",
-    "        self.word_count = Counter()\n",
-    "        self.stop_words = self.load_stop_words()\n",
-    "\n",
-    "    def load_stop_words(self):\n",
-    "        \"\"\"加载停用词\"\"\"\n",
-    "        try:\n",
-    "            with open(self.stop_words_file, 'r', encoding='utf-8') as f:\n",
-    "                return set(line.strip() for line in f if line.strip())\n",
-    "        except Exception as e:\n",
-    "            print(f\"Error loading stop words: {e}\")\n",
-    "            return set()\n",
-    "\n",
-    "    @timing_decorator\n",
-    "    def read_file(self, file_path):\n",
-    "        \"\"\"读取文件内容\"\"\"\n",
-    "        try:\n",
-    "            with open(file_path, 'r', encoding='utf-8') as f:\n",
-    "                return f.read()\n",
-    "        except Exception as e:\n",
-    "            print(f\"Error reading {file_path}: {e}\")\n",
-    "            return \"\"\n",
-    "\n",
-    "    @timing_decorator\n",
-    "    def tokenize(self, text):\n",
-    "        \"\"\"中文分词并过滤停用词\"\"\"\n",
-    "        words = jieba.lcut(text)\n",
-    "        return [word for word in words if word not in self.stop_words]\n",
-    "\n",
-    "    def process_file(self, file_path):\n",
-    "        \"\"\"处理单个文件\"\"\"\n",
-    "        if file_path.endswith('.txt'):\n",
-    "            text = self.read_file(file_path)\n",
-    "            words = self.tokenize(text)\n",
-    "            self.word_count.update(words)\n",
-    "\n",
-    "    def process_directory(self):\n",
-    "        \"\"\"处理目录下所有文件\"\"\"\n",
-    "        for file in os.listdir(self.data_dir):\n",
-    "            file_path = os.path.join(self.data_dir, file)\n",
-    "            self.process_file(file_path)\n",
-    "\n",
-    "    def get_top_words(self):\n",
-    "        \"\"\"获取前 N 高频词\"\"\"\n",
-    "        return self.word_count.most_common(self.top_n)\n",
-    "\n",
-    "    def save_results(self, top_words):\n",
-    "        \"\"\"保存结果到文件\"\"\"\n",
-    "        with open(self.output_file, 'w', encoding='utf-8') as f:\n",
-    "            for word, count in top_words:\n",
-    "                f.write(f\"{word}: {count}\\n\")\n",
-    "\n",
-    "    def run(self):\n",
-    "        \"\"\"执行词频统计并保存结果\"\"\"\n",
-    "        self.process_directory()\n",
-    "        top_words = self.get_top_words()\n",
-    "        self.save_results(top_words)\n",
-    "        for word, count in top_words:\n",
-    "            print(f\"{word}: {count}\")"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "4dcabfd9-b8f9-4796-a060-9d9f6689e92f",
-   "metadata": {},
-   "source": [
-    "### 装饰器分析\n",
-    "\n",
-    "功能：timing_decorator 记录 read_file 和 tokenize 方法的执行时间，帮助分析性能瓶颈（如分词耗时较长）。\n",
-    "\n",
-    "工程质量提升：\n",
-    "   - 可维护性：无需修改原方法代码即可添加性能监控，符合开闭原则,维护更方便。\n",
-    "   - 可读性：装饰器将性能监控逻辑与业务逻辑分离，代码更清晰。\n",
-    "   - 复用性：timing_decorator 可复用于其他方法或项目。\n",
-    "\n",
-    "局限性：装饰器增加少量性能开销，需谨慎用于高频调用的函数。"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "8fcbe48d-de8f-4387-9be3-f05f88553029",
-   "metadata": {},
-   "source": [
-    "## 第二部分：引入函数式编程\n",
-    "\n",
-    "函数式编程（如高阶函数、lambda、map/reduce）强调无变量污染、数据转换简洁性。在词频统计案例中，函数式编程可用于：\n",
-    "- 数据处理：使用 map 和 filter 处理文件和单词。\n",
-    "- 词频统计：使用 reduce 合并词频。\n",
-    "- 管道式处理：通过函数组合实现数据流处理。"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "1a6970b2-7488-43e3-ae9f-0174ff9b4b57",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# 函数式处理文件和词频\n",
-    "\n",
-    "import os\n",
-    "import jieba\n",
-    "from collections import Counter\n",
-    "import yaml\n",
-    "from functools import reduce\n",
-    "from typing import List, Tuple\n",
-    "\n",
-    "def timing_decorator(func):\n",
-    "    \"\"\"装饰器：记录函数执行时间\"\"\"\n",
-    "    import time\n",
-    "    import functools\n",
-    "    @functools.wraps(func)\n",
-    "    def wrapper(*args, **kwargs):\n",
-    "        start_time = time.time()\n",
-    "        result = func(*args, **kwargs)\n",
-    "        end_time = time.time()\n",
-    "        print(f\"{func.__name__} took {end_time - start_time:.4f} seconds\")\n",
-    "        return result\n",
-    "    return wrapper\n",
-    "\n",
-    "class TextAnalyzer:\n",
-    "    def __init__(self, config_path='config.yaml'):\n",
-    "        with open(config_path, 'r', encoding='utf-8') as f:\n",
-    "            config = yaml.safe_load(f)\n",
-    "        self.data_dir = config['data_dir']\n",
-    "        self.top_n = config['top_n']\n",
-    "        self.stop_words_file = config['stop_words_file']\n",
-    "        self.output_file = config['output_file']\n",
-    "        self.stop_words = self.load_stop_words()\n",
-    "\n",
-    "    def load_stop_words(self) -> set:\n",
-    "        \"\"\"加载停用词\"\"\"\n",
-    "        try:\n",
-    "            with open(self.stop_words_file, 'r', encoding='utf-8') as f:\n",
-    "                return set(line.strip() for line in f if line.strip())\n",
-    "        except Exception as e:\n",
-    "            print(f\"Error loading stop words: {e}\")\n",
-    "            return set()\n",
-    "\n",
-    "    @timing_decorator\n",
-    "    def read_file(self, file_path: str) -> str:\n",
-    "        \"\"\"读取文件内容\"\"\"\n",
-    "        try:\n",
-    "            with open(file_path, 'r', encoding='utf-8') as f:\n",
-    "                return f.read()\n",
-    "        except Exception as e:\n",
-    "            print(f\"Error reading {file_path}: {e}\")\n",
-    "            return \"\"\n",
-    "\n",
-    "    def tokenize(self, text: str) -> List[str]:\n",
-    "        \"\"\"中文分词并过滤停用词（函数式）\"\"\"\n",
-    "        return list(filter(lambda w: w not in self.stop_words, jieba.lcut(text)))\n",
-    "\n",
-    "    def process_file(self, file_path: str) -> Counter:\n",
-    "        \"\"\"处理单个文件，返回词频 Counter\"\"\"\n",
-    "        if file_path.endswith('.txt'):\n",
-    "            text = self.read_file(file Couple(path)\n",
-    "            words = self.tokenize(text)\n",
-    "            return Counter(words)\n",
-    "        return Counter()\n",
-    "\n",
-    "    def process_directory(self) -> Counter:\n",
-    "        \"\"\"处理目录下所有文件（函数式）\"\"\"\n",
-    "        file_paths = (os.path.join(self.data_dir, f) for f in os.listdir(self.data_dir))\n",
-    "        counters = map(self.process_file, file_paths)\n",
-    "        return reduce(lambda c1, c2: c1 + c2, counters, Counter())\n",
-    "\n",
-    "    def get_top_words(self, word_count: Counter) -> List[Tuple[str, int]]:\n",
-    "        \"\"\"获取前 N 高频词\"\"\"\n",
-    "        return word_count.most_common(self.top_n)\n",
-    "\n",
-    "    def save_results(self, top_words: List[Tuple[str, int]]):\n",
-    "        \"\"\"保存结果到文件\"\"\"\n",
-    "        with open(self.output_file, 'w', encoding='utf-8') as f:\n",
-    "            for word, count in top_words:\n",
-    "                f.write(f\"{word}: {count}\\n\")\n",
-    "\n",
-    "    def run(self):\n",
-    "        \"\"\"执行词频统计并保存结果\"\"\"\n",
-    "        word_count = self.process_directory()\n",
-    "        top_words = self.get_top_words(word_count)\n",
-    "        self.save_results(top_words)\n",
-    "        for word, count in top_words:\n",
-    "            print(f\"{word}: {count}\")"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "6ce3b7c3-f099-4e2c-b415-18b0e3ab492a",
-   "metadata": {},
-   "source": [
-    "### 函数式编程分析\n",
-    "\n",
-    "改进：\n",
-    "- map：在 process_directory 中，使用 map(self.process_file, file_paths) 并行处理文件路径，生成词频 Counter 列表。\n",
-    "- reduce：使用 reduce(lambda c1, c2: c1 + c2, counters, Counter()) 合并所有文件的词频，简洁且无副作用。\n",
-    "- filter：在 tokenize 中，使用 filter(lambda w: w not in self.stop_words, ...) 过滤停用词，替代列表推导式。\n",
-    "- 生成器：file_paths 使用生成器表达式，减少内存占用。\n",
-    "\n",
-    "工程质量提升：\n",
-    "- 可读性：函数式编程使数据处理逻辑更简洁，管道式处理清晰表达数据流（文件路径 -> 词频 -> 合并）。\n",
-    "- 性能：生成器和 map 优化内存使用，适合处理大量文件。\n",
-    "- 可维护性：函数式代码无副作用，易于测试和调试。\n",
-    "- 适用场景：适合数据转换和批量处理（如文件读取、词频合并）。\n",
-    "- 简洁性：map、reduce 等使数据处理逻辑更紧凑。\n",
-    "- 内存效率：生成器和惰性求值优化内存使用。\n",
-    "- 结合并发可显著提升效率。\n",
-    "\n",
-    "适用场景：数据流处理（如文件处理、词频合并）、无状态操作。\n",
-    "\n",
-    "局限性：\n",
-    "- 函数式代码对初学者可能不够直观，需熟悉 map、reduce 等概念。\n",
-    "- 对于复杂逻辑，函数式编程可能增加调试难度。"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "458e18ec-b536-4860-9e12-d0bf5ed9d876",
-   "metadata": {},
-   "source": [
-    "#  练习\n",
-    "\n",
-    "实践练习：\n",
-    "- 添加日志装饰器，记录每次文件处理的详细信息。\n",
-    "- 使用 functools.reduce 重写 get_top_words，尝试不同排序逻辑。\n",
-    "\n",
-    "扩展任务：\n",
-    "- 添加缓存装饰器，避免重复分词相同文件。\n",
-    "- 实现函数式管道，将文件读取、分词、统计串联为单一流。"
-   ]
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.12.7"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 5
-}

D Plus/.ipynb_checkpoints/99 工业级代码-checkpoint.ipynb

@@ -1,665 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "b6bc2a3c-3b15-4bc5-83a2-adeae3b7b4d0",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "## 项目结构\n",
-    "\n",
-    "word_frequency_project/\n",
-    "│\n",
-    "├── data/                      # 小说文本存放目录\n",
-    "│   ├── novel1.txt\n",
-    "│   ├── novel2.txt\n",
-    "│   └── ...\n",
-    "├── src/                       # 源代码目录\n",
-    "│   ├── __init__.py\n",
-    "│   ├── config.py             # 配置文件\n",
-    "│   ├── data_loader.py        # 数据加载模块\n",
-    "│   ├── preprocessor.py       # 文本预处理模块\n",
-    "│   ├── word_counter.py       # 词频统计模块\n",
-    "│   ├── output_formatter.py   # 输出格式化模块\n",
-    "│   └── main.py              # 主程序入口\n",
-    "├── tests/                     # 单元测试目录\n",
-    "│   ├── __init__.py\n",
-    "│   ├── test_data_loader.py\n",
-    "│   ├── test_preprocessor.py\n",
-    "│   ├── test_word_counter.py\n",
-    "│   └── test_output_formatter.py\n",
-    "├── requirements.txt           # 依赖文件\n",
-    "└── README.md                 # 项目说明"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "d0b55f2e-24ba-49da-8d11-f0f5eea611b0",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "src/config.py\n",
-    "定义配置项，便于扩展和修改。\n",
-    "'''\n",
-    "\n",
-    "import os\n",
-    "\n",
-    "class Config:\n",
-    "    DATA_DIR = \"data\"\n",
-    "    TOP_N_WORDS = 10\n",
-    "    STOP_WORDS = {\"的\", \"了\", \"是\", \"在\", \"和\", \"我\", \"你\", \"他\", \"她\"}  # 示例停用词\n",
-    "    ENCODING = \"utf-8\"\n",
-    "    LOG_LEVEL = \"INFO\"\n",
-    "\n",
-    "    @classmethod\n",
-    "    def get_data_dir(cls):\n",
-    "        return os.path.join(os.path.dirname(__file__), \"..\", cls.DATA_DIR)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "e5bdcdf0-16a2-4dda-85f1-d018c6370aee",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "src/data_loader.py\n",
-    "负责加载小说文本，支持目录扫描和文件读取，提供扩展点以支持不同格式。\n",
-    "'''\n",
-    "\n",
-    "import os\n",
-    "import logging\n",
-    "from src.config import Config\n",
-    "\n",
-    "class DataLoader:\n",
-    "    def __init__(self):\n",
-    "        self.data_dir = Config.get_data_dir()\n",
-    "        logging.basicConfig(level=Config.LOG_LEVEL)\n",
-    "        self.logger = logging.getLogger(__name__)\n",
-    "\n",
-    "    def load_texts(self):\n",
-    "        \"\"\"加载 data 目录下的所有文本文件\"\"\"\n",
-    "        texts = []\n",
-    "        try:\n",
-    "            for filename in os.listdir(self.data_dir):\n",
-    "                if filename.endswith(\".txt\"):\n",
-    "                    file_path = os.path.join(self.data_dir, filename)\n",
-    "                    with open(file_path, \"r\", encoding=Config.ENCODING) as f:\n",
-    "                        texts.append(f.read())\n",
-    "                    self.logger.info(f\"Loaded file: {filename}\")\n",
-    "            if not texts:\n",
-    "                self.logger.warning(\"No text files found in data directory\")\n",
-    "            return texts\n",
-    "        except Exception as e:\n",
-    "            self.logger.error(f\"Error loading files: {str(e)}\")\n",
-    "            raise"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "786e7ffa-82bc-46b9-8ffc-444d6796b87b",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "src/preprocessor.py\n",
-    "文本预处理模块，负责分词和清理，支持扩展以添加更多预处理逻辑。\n",
-    "'''\n",
-    "\n",
-    "import jieba\n",
-    "import re\n",
-    "from src.config import Config\n",
-    "\n",
-    "def timing_decorator(func):\n",
-    "    \"\"\"装饰器：记录方法执行时间\"\"\"\n",
-    "    import time\n",
-    "    def wrapper(*args, **kwargs):\n",
-    "        start = time.time()\n",
-    "        result = func(*args, **kwargs)\n",
-    "        end = time.time()\n",
-    "        print(f\"{func.__name__} took {end - start:.2f} seconds\")\n",
-    "        return result\n",
-    "    return wrapper\n",
-    "\n",
-    "class TextPreprocessor:\n",
-    "    def __init__(self):\n",
-    "        self.stop_words = Config.STOP_WORDS\n",
-    "\n",
-    "    @timing_decorator\n",
-    "    def preprocess(self, text):\n",
-    "        \"\"\"预处理：分词、去除停用词和非中文字符\"\"\"\n",
-    "        # 移除非中文字符\n",
-    "        text = re.sub(r\"[^\\u4e00-\\u9fff]\", \" \", text)\n",
-    "        # 分词\n",
-    "        words = jieba.cut(text)\n",
-    "        # 过滤停用词和空字符\n",
-    "        return [word for word in words if word.strip() and word not in self.stop_words]"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "4edd5ca7-4ba7-4446-b93e-2cfd83efca2e",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "src/word_counter.py\n",
-    "词频统计模块，使用单例模式确保全局唯一计数器。\n",
-    "'''\n",
-    "\n",
-    "from collections import Counter\n",
-    "from typing import List, Dict\n",
-    "\n",
-    "class Singleton:   为啥需要单例？\n",
-    "    \"\"\"单例模式装饰器\"\"\"\n",
-    "    def __init__(self, cls):\n",
-    "        self._cls = cls\n",
-    "        self._instance = None\n",
-    "\n",
-    "    def __call__(self, *args, **kwargs):\n",
-    "        if self._instance is None:\n",
-    "            self._instance = self._cls(*args, **kwargs)\n",
-    "        return self._instance\n",
-    "\n",
-    "@Singleton\n",
-    "class WordCounter:\n",
-    "    def __init__(self):\n",
-    "        self.counter = Counter()\n",
-    "\n",
-    "    def count_words(self, words: List[str]) -> None:\n",
-    "        \"\"\"更新词频统计\"\"\"\n",
-    "        self.counter.update(words)\n",
-    "\n",
-    "    def get_top_n(self, n: int = 10) -> Dict[str, int]:\n",
-    "        \"\"\"获取前 N 个高频词\"\"\"\n",
-    "        return dict(self.counter.most_common(n))\n",
-    "\n",
-    "    def reset(self):\n",
-    "        \"\"\"重置计数器\"\"\"\n",
-    "        self.counter.clear()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "41af3e0e-3153-4d23-9a9f-65b566b384e8",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "src/output_formatter.py\n",
-    "输出格式化模块，支持多种输出格式，便于扩展。\n",
-    "'''\n",
-    "\n",
-    "from typing import Dict\n",
-    "\n",
-    "class OutputFormatter:\n",
-    "    @staticmethod\n",
-    "    def format_json(data: Dict[str, int]) -> str:\n",
-    "        import json\n",
-    "        return json.dumps(data, ensure_ascii=False, indent=2)\n",
-    "\n",
-    "    @staticmethod\n",
-    "    def format_text(data: Dict[str, int]) -> str:\n",
-    "        return \"\\n\".join(f\"{word}: {count}\" for word, count in data.items())"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "6596162c-fd42-4b32-b328-9987568b3846",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "src/main.py\n",
-    "主程序入口，协调各模块工作。\n",
-    "'''\n",
-    "\n",
-    "from src.data_loader import DataLoader\n",
-    "from src.preprocessor import TextPreprocessor\n",
-    "from src.word_counter import WordCounter\n",
-    "from src.output_formatter import OutputFormatter\n",
-    "from src.config import Config\n",
-    "\n",
-    "def main():\n",
-    "    # 初始化模块\n",
-    "    loader = DataLoader()\n",
-    "    preprocessor = TextPreprocessor()\n",
-    "    counter = WordCounter()\n",
-    "    formatter = OutputFormatter()\n",
-    "\n",
-    "    # 加载文本\n",
-    "    texts = loader.load_texts()\n",
-    "\n",
-    "    # 预处理并统计词频\n",
-    "    for text in texts:\n",
-    "        words = preprocessor.preprocess(text)\n",
-    "        counter.count_words(words)\n",
-    "\n",
-    "    # 获取结果\n",
-    "    top_words = counter.get_top_n(Config.TOP_N_WORDS)\n",
-    "\n",
-    "    # 输出结果\n",
-    "    print(\"=== Top 10 Words (Text Format) ===\")\n",
-    "    print(formatter.format_text(top_words))\n",
-    "    print(\"\\n=== Top 10 Words (JSON Format) ===\")\n",
-    "    print(formatter.format_json(top_words))\n",
-    "\n",
-    "if __name__ == \"__main__\":\n",
-    "    main()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "36a32f17-5ce3-46e2-a563-f151454f6342",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "tests/test_data_loader.py\n",
-    "单元测试示例，确保数据加载模块的正确性。\n",
-    "'''\n",
-    "\n",
-    "import unittest\n",
-    "import os\n",
-    "from src.data_loader import DataLoader\n",
-    "from src.config import Config\n",
-    "\n",
-    "class TestDataLoader(unittest.TestCase):\n",
-    "    def setUp(self):\n",
-    "        self.loader = DataLoader()\n",
-    "        # 创建临时测试文件\n",
-    "        self.test_file = os.path.join(Config.get_data_dir(), \"test_novel.txt\")\n",
-    "        with open(self.test_file, \"w\", encoding=Config.ENCODING) as f:\n",
-    "            f.write(\"这是一个测试文本\")\n",
-    "\n",
-    "    def test_load_texts(self):\n",
-    "        texts = self.loader.load_texts()\n",
-    "        self.assertGreater(len(texts), 0)\n",
-    "        self.assertIn(\"这是一个测试文本\", texts)\n",
-    "\n",
-    "    def tearDown(self):\n",
-    "        if os.path.exists(self.test_file):\n",
-    "            os.remove(self.test_file)\n",
-    "\n",
-    "if __name__ == \"__main__\":\n",
-    "    unittest.main()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "1f550544-f0f4-4f0c-bdb7-9928b6820bdf",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "tests/test_preprocessor.py\n",
-    "测试文本预处理模块。\n",
-    "'''\n",
-    "\n",
-    "import unittest\n",
-    "from src.preprocessor import TextPreprocessor\n",
-    "\n",
-    "class TestTextPreprocessor(unittest.TestCase):\n",
-    "    def setUp(self):\n",
-    "        self.preprocessor = TextPreprocessor()\n",
-    "\n",
-    "    def test_preprocess(self):\n",
-    "        text = \"这是一个测试文本，包含了123和一些符号！\"\n",
-    "        words = self.preprocessor.preprocess(text)\n",
-    "        expected = [\"测试\", \"文本\", \"包含\", \"一些\", \"符号\"]\n",
-    "        self.assertEqual(words, expected)\n",
-    "\n",
-    "if __name__ == \"__main__\":\n",
-    "    unittest.main()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "8fb8b4cd-0b27-426a-9556-8f21227c5374",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "tests/test_word_counter.py\n",
-    "测试词频统计模块。\n",
-    "'''\n",
-    "import unittest\n",
-    "from src.word_counter import WordCounter\n",
-    "\n",
-    "class TestWordCounter(unittest.TestCase):\n",
-    "    def setUp(self):\n",
-    "        self.counter = WordCounter()\n",
-    "\n",
-    "    def test_count_words(self):\n",
-    "        self.counter.count_words([\"测试\", \"文本\", \"测试\"])\n",
-    "        result = self.counter.get_top_n(2)\n",
-    "        expected = {\"测试\": 2, \"文本\": 1}\n",
-    "        self.assertEqual(result, expected)\n",
-    "\n",
-    "    def test_reset(self):\n",
-    "        self.counter.count_words([\"测试\"])\n",
-    "        self.counter.reset()\n",
-    "        self.assertEqual(self.counter.get_top_n(1), {})\n",
-    "\n",
-    "if __name__ == \"__main__\":\n",
-    "    unittest.main()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "4b7507dc-b693-4dbf-9a21-5f2833d13d0e",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "'''\n",
-    "requirements.txt\n",
-    "列出项目依赖。\n",
-    "''''\n",
-    "jieba==0.42.1"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "573c4ddd-800e-4b59-9e20-a87d6a2b14cd",
-   "metadata": {},
-   "source": [
-    "'''\n",
-    "README.md\n",
-    "提供项目说明和使用方法。\n",
-    "'''\n",
-    "# Word Frequency Analysis Project\n",
-    "\n",
-    "## Overview\n",
-    "This project processes 100 novels in the `data` directory, counts word frequencies, and outputs the top 10 words. It demonstrates software engineering principles like modularity, design patterns, and unit testing.\n",
-    "\n",
-    "## Setup\n",
-    "1. Install dependencies: `pip install -r requirements.txt`\n",
-    "2. Place novel files (.txt) in the `data` directory.\n",
-    "3. Run the program: `python src/main.py`\n",
-    "\n",
-    "## Testing\n",
-    "Run tests: `python -m unittest discover tests`\n",
-    "\n",
-    "## Extensibility\n",
-    "- Add new preprocessors in `preprocessor.py`.\n",
-    "- Support new output formats in `output_formatter.py`.\n",
-    "- Modify configurations in `config.py`."
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "4bd74972-f9c4-4ac9-a557-de4198889047",
-   "metadata": {},
-   "source": [
-    "## 使用方法\n",
-    "\n",
-    "准备环境：\n",
-    "pip install -r requirements.txt\n",
-    "\n",
-    "准备数据：\n",
-    "- 在 data 目录下放入 100 个 .txt 小说文件（需为 UTF-8 编码）。\n",
-    "- 确保安装 jieba 分词库。\n",
-    "\n",
-    "运行程序：\n",
-    "python src/main.py\n",
-    "\n",
-    "运行测试：\n",
-    "python -m unittest discover tests"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "16f7a973-7c49-4d11-ab3f-457d4622e5e6",
-   "metadata": {},
-   "source": [
-    "## 扩展建议\n",
-    "\n",
-    "- 支持多语言：在 TextPreprocessor 中添加英文分词（如使用 nltk 或 spacy）。\n",
-    "- 数据库存储：将词频结果保存到数据库（如 SQLite），在 WordCounter 中添加存储方法。\n",
-    "- 并行处理：使用 multiprocessing 加速大文件处理。\n",
-    "- 可视化：在 OutputFormatter 中添加图表输出（如使用 matplotlib）。\n",
-    "- 配置文件：将 Config 改为从外部 JSON/YAML 文件加载。"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "b2ad0efb-4c7c-4f98-a809-ce6cdcefdb34",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "## 设计说明\n",
-    "\n",
-    "模块化设计：\n",
-    "- 各模块（DataLoader, TextPreprocessor, WordCounter, OutputFormatter）职责单一，符合单一职责原则（SRP）。\n",
-    "- 模块间通过明确接口交互，易于替换或扩展。\n",
-    "\n",
-    "设计模式：\n",
-    "- 单例模式：WordCounter 使用单例模式，确保全局唯一计数器。\n",
-    "- 策略模式：OutputFormatter 支持多种输出格式（JSON、Text），易于添加新格式。\n",
-    "- 装饰器模式：timing_decorator 用于性能监控，便于扩展其他功能（如日志记录）。\n",
-    "\n",
-    "可扩展性：\n",
-    "- Config 类集中管理配置，便于调整参数（如停用词、输出数量）。\n",
-    "- DataLoader 支持动态扫描目录，新增文件无需改动代码。\n",
-    "- TextPreprocessor 可扩展以支持其他分词工具或预处理规则。\n",
-    "\n",
-    "单元测试：\n",
-    "- 每个模块都有对应的测试用例，确保功能正确性。\n",
-    "- 使用 unittest 框架，支持持续集成。\n",
-    "\n",
-    "语言特性利用：\n",
-    "- 使用 Python 的装饰器（timing_decorator）记录方法执行时间。\n",
-    "- 利用类型注解（typing 模块）提高代码可读性。\n",
-    "- 异常处理和日志记录（logging）增强鲁棒性。\n",
-    "\n",
-    "教学用途：\n",
-    "- 包含常见工程化实践：模块化、测试驱动开发、配置管理。\n",
-    "- 提供扩展点（如支持英文分词、数据库存储），便于学生实践。"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "b1aac488-3a98-418c-8201-e7f77c392a1f",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# text_analyzer.py\n",
-    "\n",
-    "import os\n",
-    "import jieba\n",
-    "from collections import Counter\n",
-    "import yaml\n",
-    "from contextlib import contextmanager\n",
-    "from typing import List, Tuple\n",
-    "from abc import ABC, abstractmethod\n",
-    "\n",
-    "@contextmanager\n",
-    "def file_reader(file_path: str):\n",
-    "    try:\n",
-    "        with open(file_path, 'r', encoding='utf-8') as f:\n",
-    "            yield f.read()\n",
-    "    except Exception as e:\n",
-    "        print(f\"Error reading {file_path}: {e}\")\n",
-    "        yield \"\"\n",
-    "\n",
-    "class Tokenizer(ABC):\n",
-    "    @abstractmethod\n",
-    "    def tokenize(self, text: str, stop_words: set) -> List[str]:\n",
-    "        pass\n",
-    "\n",
-    "class JiebaTokenizer(Tokenizer):\n",
-    "    def tokenize(self, text: str, stop_words: set) -> List[str]:\n",
-    "        for word in jieba.lcut(text):\n",
-    "            if word not in stop_words:\n",
-    "                yield word\n",
-    "\n",
-    "class SimpleTokenizer(Tokenizer):\n",
-    "    def tokenize(self, text: str, stop_words: set) -> List[str]:\n",
-    "        for word in text.split():\n",
-    "            if word not in stop_words:\n",
-    "                yield word\n",
-    "\n",
-    "class TokenizerFactory:\n",
-    "    @staticmethod\n",
-    "    def create_tokenizer(name: str) -> Tokenizer:\n",
-    "        return {'jieba': JiebaTokenizer(), 'simple': SimpleTokenizer()}.get(name, JiebaTokenizer())\n",
-    "\n",
-    "class OutputObserver(ABC):\n",
-    "    @abstractmethod\n",
-    "    def update(self, top_words: List[Tuple[str, int]]):\n",
-    "        pass\n",
-    "\n",
-    "class ConsoleOutput(OutputObserver):\n",
-    "    def update(self, top_words: List[Tuple[str, int]]):\n",
-    "        for word, count in top_words:\n",
-    "            print(f\"{word}: {count}\")\n",
-    "\n",
-    "class FileOutput(OutputObserver):\n",
-    "    def __init__(self, output_file: str):\n",
-    "        self.output_file = output_file\n",
-    "    def update(self, top_words: List[Tuple[str, int]]):\n",
-    "        with open(self.output_file, 'w', encoding='utf-8') as f:\n",
-    "            for word, count in top_words:\n",
-    "                f.write(f\"{word}: {count}\\n\")\n",
-    "\n",
-    "class TextAnalyzer:\n",
-    "    def __init__(self, config_path='config.yaml'):\n",
-    "        with open(config_path, 'r', encoding='utf-8') as f:\n",
-    "            config = yaml.safe_load(f)\n",
-    "        self.data_dir = config['data_dir']\n",
-    "        self.top_n = config['top_n']\n",
-    "        self.stop_words_file = config['stop_words_file']\n",
-    "        self.output_file = config['output_file']\n",
-    "        self.stop_words = self.load_stop_words()\n",
-    "        self.word_count = Counter()\n",
-    "        self.tokenizer = TokenizerFactory.create_tokenizer(config.get('tokenizer', 'jieba'))\n",
-    "        self.observers = [ConsoleOutput(), FileOutput(self.output_file)]\n",
-    "\n",
-    "    def load_stop_words(self) -> set:\n",
-    "        with file_reader(self.stop_words_file) as content:\n",
-    "            return set(line.strip() for line in content.splitlines() if line.strip())\n",
-    "\n",
-    "    def process_file(self, file_path: str):\n",
-    "        if file_path.endswith('.txt'):\n",
-    "            with file_reader(file_path) as text:\n",
-    "                words = self.tokenizer.tokenize(text, self.stop_words)\n",
-    "                self.word_count.update(words)\n",
-    "\n",
-    "    def process_directory(self):\n",
-    "        for file in os.listdir(self.data_dir):\n",
-    "            file_path = os.path.join(self.data_dir, file)\n",
-    "            self.process_file(file_path)\n",
-    "\n",
-    "    def get_top_words(self) -> List[Tuple[str, int]]:\n",
-    "        return self.word_count.most_common(self.top_n)\n",
-    "\n",
-    "    def notify_observers(self, top_words: List[Tuple[str, int]]):\n",
-    "        for observer in self.observers:\n",
-    "            observer.update(top_words)\n",
-    "\n",
-    "    def run(self):\n",
-    "        self.process_directory()\n",
-    "        top_words = self.get_top_words()\n",
-    "        self.notify_observers(top_words)"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "d5c689f4-e363-4327-9dc4-15c7157d4288",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "# main.py\n",
-    "\n",
-    "from text_analyzer import TextAnalyzer\n",
-    "\n",
-    "def main():\n",
-    "    analyzer = TextAnalyzer()\n",
-    "    analyzer.run()\n",
-    "\n",
-    "if __name__ == '__main__':\n",
-    "    main()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "cc1d9fb1-3bb5-4f71-aeb3-e304511f4785",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "## 结论\n",
-    "\n",
-    "通过引入上下文管理器、生成器、元编程、策略模式、观察者模式和工厂模式，词频统计代码在可扩展性、可维护性和复用性上进一步提升。\n",
-    "这些特性和模式使代码更模块化、灵活，适合大型项目，同时保持清晰的工程结构。结合之前的装饰器和函数式编程，代码已达到工程化水平。\n",
-    "\n",
-    "若需深入，可以进一步考虑其它性能特性."
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "id": "7244afd0-4405-402a-b9be-75f5d7ff883c",
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "## 进一步练习\n",
-    "\n",
-    "实践练习：\n",
-    "- 实现新分词器（如 thulac）并通过策略模式或工厂模式集成。\n",
-    "- 添加新观察者（如 JSON 输出）。\n",
-    "\n",
-    "使用生成器实现流式词频统计，比较内存占用。\n",
-    "实现缓存机制，缓存已处理文件的分词结果。\n",
-    "\n",
-    "添加命令行接口（argparse），动态配置 top_n 和 tokenizer。"
-   ]
-  },
-  {
-   "cell_type": "markdown",
-   "id": "09c10307-f162-4b36-85b6-6bc01d0001e0",
-   "metadata": {},
-   "source": [
-    "## 综合实现（整合特性与模式）\n",
-    "\n",
-    "整合上下文管理器、生成器、策略模式和观察者模式的最终实现（部分代码展示）。"
-   ]
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "Python 3 (ipykernel)",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 3
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython3",
-   "version": "3.12.7"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 5
-}