【Python】函数式编程2

第四章：高级应用与设计模式

4.1 利用高阶函数构建领域特定语言（Domain Specific Languages, DSLs）

在软件开发中，我们常常面临一个挑战：如何让业务专家或非程序员能够以更自然、更直观的方式表达他们的需求或定义复杂的规则。传统的命令式代码往往过于详细和技术化，难以被领域专家理解或修改。而**领域特定语言（Domain Specific Languages, DSLs）**正是为了解决这个问题而生。

4.1.1 什么是领域特定语言（DSLs）？

DSLs 是一种专门为特定应用领域设计的计算机语言。与通用编程语言（如Python、Java）不同，DSLs 的语法和语义都围绕着一个特定的领域概念进行设计，使得在该领域内的问题表达变得更加简洁、直观和高效。

DSLs 的分类：

外部 DSLs（External DSLs）：

拥有独立的语法解析器和解释器。
例如：SQL（数据库查询）、HTML（网页结构）、正则表达式（文本模式匹配）、YAML/JSON（数据序列化）。
优点：语法可以完全定制，与宿主语言无关。
缺点：需要额外的解析和工具支持，开发成本较高。

内部 DSLs（Internal DSLs）：

在现有宿主通用编程语言（如 Python）的基础上构建。
通过利用宿主语言的语法特性和库来模拟领域特定的表达方式。
优点：无需开发新的解析器，直接利用宿主语言的全部能力和工具链；开发成本较低。
缺点：受限于宿主语言的语法，可能不如外部 DSL 那样“领域特定”。

本节我们将专注于如何在 Python 中使用函数式编程的理念和技术来构建内部 DSLs。

4.1.2 函数式编程与内部 DSLs 的契合点

函数式编程范式与构建内部 DSLs 有着天然的契合。主要原因在于：

高阶函数（Higher-Order Functions）：函数可以作为参数传递，也可以作为返回值。这使得我们可以创建“操作组合器”，将简单的函数组合成复杂的行为，而这些组合器本身可以构成 DSL 的语法元素。
函数组合（Function Composition）：将多个小函数像管道一样连接起来，形成一个数据流。这与 DSL 中描述的工作流或数据转换管道非常吻合。
柯里化（Currying）/ 偏函数应用（Partial Application）：允许我们预设函数的部分参数，创建出更具体的函数。这有助于构建 DSL 中可配置的操作。
不可变性与纯函数：DSLs 往往需要描述一系列声明式的操作或规则，而不是命令式的状态修改。纯函数和不可变数据能够确保 DSL 描述的行为是可预测和无副作用的。
声明式风格：函数式编程鼓励声明式编程，即描述“做什么”而不是“怎么做”。这与 DSL 旨在让用户声明其意图而非实现细节的目标一致。

4.1.3 案例分析：构建一个简单的日志处理和分析 DSL

让我们考虑一个常见的场景：我们需要对日志文件进行处理、过滤、转换和分析。这通常涉及多个步骤，例如：

读取日志
根据时间、级别或关键词过滤日志
解析日志行（例如，提取时间戳、消息、用户ID、CPU使用率等）
对解析后的数据进行进一步处理（例如，计算平均CPU使用率、统计错误数量）
将结果输出

如果每次都要编写一套命令式代码来完成这些任务，会非常繁琐且难以修改。我们可以尝试构建一个简单的内部 DSL 来简化这个过程。

DSLs 的目标：
我们的目标是让用户能够以如下方式描述日志处理流程：

# 伪代码：期望的 DSL 语法
pipeline = (
    read_log_file("app.log")
    .filter_by_level("ERROR")
    .parse_json_line()
    .calculate_average_cpu()
    .save_to_report("error_report.txt")
)

# 稍后运行这个管道
# run_pipeline(pipeline)

这看起来像是一个链式调用的 API，但其底层是函数式组合。

实现步骤：

定义基础操作函数（原子操作）：这些是构成 DSL 的最小单元，每个函数都执行一个具体的、单一的日志处理任务。
创建“管道”或“流”的抽象：一个对象或函数，能够持有当前的数据，并允许我们链式调用下一个操作。这通常会用到高阶函数。
构建操作组合器：让用户能够以声明式的方式组合这些操作。

具体实现：

我们从一个 LogData 类开始，它将承载在管道中流动的数据，并提供链式方法来构建 DSL。

import json # 导入json模块，用于处理JSON数据
import re # 导入re模块，用于正则表达式操作
from typing import List, Dict, Any, Callable, TypeVar # 导入类型提示
from functools import wraps # 导入wraps装饰器，用于保留函数元数据

# 定义一个类型变量，表示在管道中流动的数据类型
# 这里我们假设数据可以是任意类型，但在实际中会更具体
T = TypeVar('T') # 定义类型变量T
U = TypeVar('U') # 定义类型变量U

# 辅助函数：compose - 用于函数组合 (与之前章节类似)
def compose(*functions: Callable[..., Any]) -> Callable[..., Any]: # 定义一个函数，用于组合多个函数
    """
    函数组合器。从右到左应用函数。
    compose(f, g, h)(x) 等同于 f(g(h(x)))
    """
    if not functions: # 如果没有传入函数
        return lambda x: x # 返回一个恒等函数
    
    def composed(*args, **kwargs): # 定义组合后的函数
        result = functions[-1](*args, **kwargs) # 首先应用最右边的函数
        for f in reversed(functions[:-1]): # 遍历剩余的函数（从右往左）
            result = f(result) # 将上一个函数的结果作为输入，应用当前函数
        return result # 返回最终结果
    return composed # 返回组合后的函数

# 日志数据容器与 DSL 构建器
class LogProcessingPipeline: # 定义LogProcessingPipeline类，用于构建日志处理管道
    """
    负责封装日志数据，并提供链式方法来构建日志处理 DSL。
    每个方法都返回一个新的 LogProcessingPipeline 实例，以保持不变性。
    """
    def __init__(self, data: List[Dict[str, Any]] = None): # 类的初始化方法
        self._data: List[Dict[str, Any]] = data if data is not None else [] # 存储处理中的日志数据，默认为空列表
        self._actions: List[Callable[[List[Dict[str, Any]]], List[Dict[str, Any]]]] = [] # 存储待执行的操作列表
        self._initial_source: Callable[[], List[str]] = lambda: [] # 存储初始数据源函数，默认为空

    def _clone(self, new_data: List[Dict[str, Any]] = None) -> 'LogProcessingPipeline': # 内部方法：克隆当前管道实例
        """
        克隆当前管道实例，并可选择地传入新的数据。
        这确保了每个链式操作都返回一个新实例，保持不可变性。
        """
        new_pipeline = LogProcessingPipeline(new_data if new_data is not None else self._data[:]) # 创建新实例，复制当前数据
        new_pipeline._actions = self._actions[:] # 复制当前操作列表
        new_pipeline._initial_source = self._initial_source # 复制初始数据源
        return new_pipeline # 返回新管道实例

    # --- 数据源操作 (副作用，在执行时发生) ---
    def read_file(self, filepath: str) -> 'LogProcessingPipeline': # 读取文件，作为管道的初始数据源
        """
        设置日志文件的读取操作作为管道的起始。
        返回一个新的 LogProcessingPipeline 实例。
        这个操作本身不直接返回数据，而是设置了数据源。
        """
        def _read_action(): # 定义实际的读取动作
            print(f"[DSL] 正在读取文件: {
     
              filepath}...") # 打印提示信息
            try: # 尝试执行代码
                with open(filepath, 'r', encoding='utf-8') as f: # 以UTF-8编码打开文件
                    return f.readlines() # 读取所有行并返回
            except FileNotFoundError: # 捕获文件未找到错误
                print(f"[DSL-错误] 文件未找到: {
     
              filepath}") # 打印错误信息
                return [] # 返回空列表
            except Exception as e: # 捕获其他异常
                print(f"[DSL-错误] 读取文件时发生错误: {
     
              e}") # 打印错误信息
                return [] # 返回空列表
        
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._initial_source = _read_action # 设置新的初始数据源
        return new_pipeline # 返回新管道实例

    def from_list(self, log_lines: List[str]) -> 'LogProcessingPipeline': # 从列表中加载数据
        """
        从一个字符串列表设置管道的起始数据。
        """
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._initial_source = lambda: log_lines # 设置数据源为传入的列表
        return new_pipeline # 返回新管道实例

    # --- 转换操作 (纯函数，在 execute 时按序应用) ---

    def parse_json_lines(self) -> 'LogProcessingPipeline': # 解析JSON格式的日志行
        """
        将管道中的每行字符串尝试解析为 JSON 对象。
        不成功的行将被过滤掉。
        返回一个新的 LogProcessingPipeline 实例。
        """
        def _parse_action(lines: List[str]) -> List[Dict[str, Any]]: # 定义解析动作
            print("[DSL] 正在解析 JSON 行...") # 打印提示信息
            parsed_data = [] # 初始化解析后的数据列表
            for line in lines: # 遍历每一行
                try: # 尝试执行代码
                    data = json.loads(line) # 将行解析为JSON对象
                    parsed_data.append(data) # 添加到解析后的数据列表
                except json.JSONDecodeError: # 捕获JSON解码错误
                    # print(f"  [DSL-警告] 跳过无效的 JSON 行: {line.strip()}") # 打印警告信息（可选）
                    pass # 静默跳过无效行
            return parsed_data # 返回解析后的数据

        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_parse_action) # 添加解析动作到操作列表
        return new_pipeline # 返回新管道实例
    
    def parse_regex_lines(self, pattern: str, keys: List[str]) -> 'LogProcessingPipeline': # 使用正则表达式解析日志行
        """
        使用正则表达式解析日志行。
        pattern: 正则表达式字符串，应包含命名捕获组。
        keys: 捕获组的名称列表，用于构建字典的键。
        """
        regex_compiled = re.compile(pattern) # 编译正则表达式
        def _parse_action(lines: List[str]) -> List[Dict[str, Any]]: # 定义解析动作
            print(f"[DSL] 正在使用正则表达式 '{
     
              pattern}' 解析行...") # 打印提示信息
            parsed_data = [] # 初始化解析后的数据列表
            for line in lines: # 遍历每一行
                match = regex_compiled.match(line) # 尝试匹配正则表达式
                if match: # 如果匹配成功
                    data = {
   
            key: match.group(key) for key in keys if key in match.groupdict()} # 提取捕获组并构建字典
                    parsed_data.append(data) # 添加到解析后的数据列表
            return parsed_data # 返回解析后的数据

        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_parse_action) # 添加解析动作到操作列表
        return new_pipeline # 返回新管道实例

    def filter_by_field(self, field: str, value: Any) -> 'LogProcessingPipeline': # 根据字段值过滤
        """
        根据指定字段的值进行过滤。
        只保留 field 字段等于 value 的日志项。
        """
        def _filter_action(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 定义过滤动作
            print(f"[DSL] 正在按字段 '{
     
              field}' = '{
     
              value}' 过滤...") # 打印提示信息
            return [item for item in data if item.get(field) == value] # 过滤并返回匹配的项
        
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_filter_action) # 添加过滤动作到操作列表
        return new_pipeline # 返回新管道实例

    def filter_by_condition(self, predicate: Callable[[Dict[str, Any]], bool]) -> 'LogProcessingPipeline': # 根据自定义条件过滤
        """
        根据自定义的条件（一个接收字典并返回布尔值的函数）过滤日志项。
        """
        def _filter_action(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 定义过滤动作
            print(f"[DSL] 正在按自定义条件过滤...") # 打印提示信息
            return [item for item in data if predicate(item)] # 过滤并返回满足条件的项
        
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_filter_action) # 添加过滤动作到操作列表
        return new_pipeline # 返回新管道实例

    def select_fields(self, fields: List[str]) -> 'LogProcessingPipeline': # 选择特定字段
        """
        只保留日志项中指定的字段。
        """
        def _select_action(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 定义选择动作
            print(f"[DSL] 正在选择字段: {
     
              ', '.join(fields)}...") # 打印提示信息
            return [{
   
            field: item.get(field) for field in fields if field in item} for item in data] # 为每个项创建新字典，只包含指定字段
        
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_select_action) # 添加选择动作到操作列表
        return new_pipeline # 返回新管道实例
    
    def transform_field(self, field: str, transform_func: Callable[[Any], Any]) -> 'LogProcessingPipeline': # 转换字段值
        """
        对指定字段的值应用一个转换函数。
        """
        def _transform_action(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 定义转换动作
            print(f"[DSL] 正在转换字段 '{
     
              field}'...") # 打印提示信息
            new_data = [] # 初始化新数据列表
            for item in data: # 遍历每个项
                new_item = item.copy() # 复制当前项
                if field in new_item: # 如果字段存在
                    new_item[field] = transform_func(new_item[field]) # 应用转换函数
                new_data.append(new_item) # 添加到新数据列表
            return new_data # 返回新数据列表
        
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_transform_action) # 添加转换动作到操作列表
        return new_pipeline # 返回新管道实例

    # --- 聚合操作 (会改变数据的结构，通常是管道的最后一步) ---
    def aggregate_count(self, group_by_field: str = None) -> 'LogProcessingPipeline': # 计数聚合
        """
        对日志项进行计数聚合。
        如果指定 group_by_field，则按该字段分组计数。
        否则，统计总数。
        返回的数据结构将变为 [{'group': 'value', 'count': N}] 或 [{'total_count': N}]。
        """
        def _aggregate_action(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 定义聚合动作
            print(f"[DSL] 正在进行计数聚合 (按 {
     
              group_by_field if group_by_field else '总数'}) ...") # 打印提示信息
            if not data: # 如果没有数据
                return [{
   
            'total_count': 0}] # 返回总数为0
            
            if group_by_field: # 如果指定了分组字段
                counts = {
   
            } # 初始化计数字典
                for item in data: # 遍历每个项
                    key = item.get(group_by_field, "N/A") # 获取分组键，默认为“N/A”
                    counts[key] = counts.get(key, 0) + 1 # 计数加1
                return [{
   
            group_by_field: k, 'count': v} for k, v in counts.items()] # 返回分组计数结果
            else: # 否则，统计总数
                return [{
   
            'total_count': len(data)}] # 返回总数
        
        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_aggregate_action) # 添加聚合动作到操作列表
        return new_pipeline # 返回新管道实例

    def aggregate_average(self, field_to_avg: str, group_by_field: str = None) -> 'LogProcessingPipeline': # 平均值聚合
        """
        计算指定字段的平均值聚合。
        如果指定 group_by_field，则按该字段分组计算平均值。
        """
        def _aggregate_action(data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: # 定义聚合动作
            print(f"[DSL] 正在计算字段 '{
     
              field_to_avg}' 的平均值聚合 (按 {
     
              group_by_field if group_by_field else '总平均'}) ...") # 打印提示信息
            if not data: # 如果没有数据
                return [{
   
            f'average_{
     
              field_to_avg}': 0.0}] # 返回平均值为0
            
            if group_by_field: # 如果指定了分组字段
                groups_sum = {
   
            } # 初始化分组和字典
                groups_count = {
   
            } # 初始化分组计数字典
                for item in data: # 遍历每个项
                    key = item.get(group_by_field, "N/A") # 获取分组键
                    value = item.get(field_to_avg) # 获取要平均的字段值
                    
                    if isinstance(value, (int, float)): # 如果值是数字
                        groups_sum[key] = groups_sum.get(key, 0.0) + value # 累加和
                        groups_count[key] = groups_count.get(key, 0) + 1 # 累加计数
                
                results = [] # 初始化结果列表
                for k, total_sum in groups_sum.items(): # 遍历分组和
                    count = groups_count.get(k, 0) # 获取计数
                    avg = total_sum / count if count > 0 else 0.0 # 计算平均值
                    results.append({
   
            group_by_field: k, f'average_{
     
              field_to_avg}': avg}) # 添加到结果列表
                return results # 返回分组平均值结果
            else: # 否则，计算总平均
                total_sum = 0.0 # 初始化总和
                count = 0 # 初始化计数
                for item in data: # 遍历每个项
                    value = item.get(field_to_avg) # 获取字段值
                    if isinstance(value, (int, float)): # 如果值是数字
                        total_sum += value # 累加总和
                        count += 1 # 计数加1
                avg = total_sum / count if count > 0 else 0.0 # 计算平均值
                return [{
   
            f'total_average_{
     
              field_to_avg}': avg}] # 返回总平均值

        new_pipeline = self._clone() # 克隆当前管道
        new_pipeline._actions.append(_aggregate_action) # 添加聚合动作到操作列表
        return new_pipeline # 返回新管道实例

    # --- 输出操作 (副作用，在执行时发生) ---
    def save_to_file(self, output_filepath: str) -> 'LogProcessingPipeline': # 保存结果到文件
        """
        设置一个操作，将最终处理结果保存到文件。
        注意：这个操作在 execute 时才执行。
        """
        def _save_action(final_data: List[Dict[str, Any]]): # 定义保存动作
            print(f"[DSL] 正在将结果保存到文件: {
     
              output_filepath}...") # 打印提示信息
            try: # 尝试执行代码
                with open(output_filepath, 'w', encoding='utf-8')