{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "8a36de81",
   "metadata": {},
   "source": [
    "#  更优雅的执行流"
   ]
  },
  {
   "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",
    "- 添加缓存装饰器，避免重复分词相同文件。\n",
    "- 使用 functools.reduce 重写 get_top_words，尝试不同排序逻辑。\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
}
