Code - Python进阶语法实战

Python进阶语法实战 - 详细知识点拆解与可执行练习步骤

核心目标：掌握Python进阶语法核心用法，能够结合企业实际场景（日志处理、数据库连接）完成实战开发，满足性能与规范要求。

一、核心知识点拆解

本部分按「基础知识点讲解→核心定义→本质作用→实现方式→实战场景」结构拆解，每个知识点配套方法说明与示例代码。

（一）装饰器

1. 基础知识点讲解

装饰器是Python中一种高阶函数的高级应用，依赖于函数的可调用性和闭包特性，能够在不修改原函数代码的前提下，对函数的功能进行扩展或增强。常用于日志记录、权限校验、接口限流、性能统计等场景，是企业级开发中实现代码复用和功能解耦的重要手段。

2. 核心定义

装饰器（Decorator）：一种接受函数作为参数，并返回一个新函数的可调用对象（通常是函数），新函数会在原函数执行前后添加额外逻辑，且不改变原函数的定义和调用方式。

3. 本质作用

实现「功能增强」与「业务逻辑」的解耦，将通用功能（如日志、限流）抽离为独立组件，降低代码冗余，提高代码可维护性和可扩展性。

4. 实现方式

• 基础装饰器（无参数）：通过嵌套函数实现，外层函数接受原函数作为参数，内层函数包裹原函数执行逻辑，最后返回内层函数。
• 带参数装饰器：需要三层嵌套函数，最外层函数接收装饰器参数，中间层接收原函数，内层函数执行增强逻辑，通过闭包传递参数。
• 装饰器语法糖：使用 @装饰器名 放在函数定义上方，简化装饰器的调用。
• 类装饰器：通过类的 __call__ 方法实现，当实例化类并装饰函数时，调用实例会触发 __call__ 方法执行增强逻辑。

5. 实战场景：开发日志装饰器（支持日志分级、错误邮件提醒）

（1）实战用到的方法说明

• logging 模块：Python内置日志模块，支持日志分级（DEBUG/INFO/WARNING/ERROR/CRITICAL）、日志格式自定义、文件输出等。
• smtplib 与 email 模块：用于发送邮件，当出现ERROR级以上日志时触发邮件提醒。
• functools.wraps：用于保留原函数的元信息（如函数名、文档字符串），避免装饰器导致原函数信息丢失。
• 闭包特性：用于保存装饰器参数（如日志级别、邮件配置），供内层函数使用。

（2）示例代码

import logging
import smtplib
from email.mime.text import MIMEText
from functools import wraps
from typing import Callable

# 1. 配置日志基础设置（符合企业规范）
def setup_logger(logger_name: str = "business_logger") -> logging.Logger:
    logger = logging.getLogger(logger_name)
    logger.setLevel(logging.DEBUG)  # 基础日志级别，低于此级别的日志不记录
    
    # 避免重复添加处理器
    if logger.handlers:
        return logger
    
    # 控制台处理器（输出INFO及以上级别）
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.INFO)
    
    # 文件处理器（输出DEBUG及以上级别，按天分割）
    file_handler = logging.handlers.TimedRotatingFileHandler(
        "business.log", when="D", interval=1, backupCount=7, encoding="utf-8"
    )
    file_handler.setLevel(logging.DEBUG)
    
    # 企业规范日志格式：时间-日志级别-函数名-日志信息
    log_format = "%(asctime)s - %(levelname)s - %(funcName)s - %(message)s"
    formatter = logging.Formatter(log_format)
    
    console_handler.setFormatter(formatter)
    file_handler.setFormatter(formatter)
    
    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    return logger

# 2. 邮件发送工具函数
def send_error_email(subject: str, content: str, smtp_config: dict) -> None:
    """当出现ERROR级日志时发送邮件提醒"""
    msg = MIMEText(content, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = smtp_config["sender"]
    msg["To"] = smtp_config["receiver"]
    
    try:
        with smtplib.SMTP_SSL(smtp_config["smtp_server"], smtp_config["smtp_port"]) as smtp:
            smtp.login(smtp_config["username"], smtp_config["password"])
            smtp.send_message(msg)
        print("错误邮件发送成功")
    except Exception as e:
        print(f"错误邮件发送失败：{str(e)}")

# 3. 支持日志分级和邮件提醒的装饰器
def log_decorator(level: int = logging.INFO, smtp_config: dict = None) -> Callable:
    """
    日志装饰器
    :param level: 日志级别
    :param smtp_config: 邮件配置字典，格式：{
        "smtp_server": "smtp.xxx.com",
        "smtp_port": 465,
        "sender": "xxx@xxx.com",
        "receiver": "xxx@xxx.com",
        "username": "xxx@xxx.com",
        "password": "xxx"
    }
    :return: 装饰后的函数
    """
    logger = setup_logger()
    
    def decorator(func: Callable) -> Callable:
        @wraps(func)  # 保留原函数元信息
        def wrapper(*args, **kwargs):
            # 记录函数执行前日志
            logger.log(level, f"函数 {func.__name__} 开始执行，参数：args={args}, kwargs={kwargs}")
            try:
                result = func(*args, **kwargs)
                # 记录函数执行成功日志
                logger.log(level, f"函数 {func.__name__} 执行成功，返回值：{result}")
                return result
            except Exception as e:
                # 记录错误日志
                error_msg = f"函数 {func.__name__} 执行失败，错误信息：{str(e)}"
                logger.error(error_msg, exc_info=True)  # exc_info=True 记录异常堆栈
                # 若配置邮件，发送错误邮件
                if smtp_config:
                    send_error_email(f"【错误告警】函数 {func.__name__} 执行异常", error_msg, smtp_config)
                raise  # 重新抛出异常，不影响原函数异常处理逻辑
        return wrapper
    return decorator

# 4. 测试用例
if __name__ == "__main__":
    # 邮件配置（需替换为实际配置）
    test_smtp_config = {
        "smtp_server": "smtp.163.com",
        "smtp_port": 465,
        "sender": "your_email@163.com",
        "receiver": "target_email@xxx.com",
        "username": "your_email@163.com",
        "password": "your_email_password"  # 163邮箱需使用授权码
    }
    
    @log_decorator(level=logging.DEBUG, smtp_config=test_smtp_config)
    def add(a: int, b: int) -> int:
        return a + b
    
    @log_decorator(level=logging.INFO, smtp_config=test_smtp_config)
    def divide(a: int, b: int) -> float:
        return a / b
    
    # 测试正常执行
    add(1, 2)
    # 测试异常执行（触发错误日志和邮件提醒）
    divide(1, 0)
    

（二）生成器/迭代器

1. 基础知识点讲解

迭代器和生成器是Python中处理可迭代对象的核心机制，主要用于高效处理大量数据（如超大日志、海量文件）。迭代器是实现了迭代协议的对象，生成器是一种特殊的迭代器，通过简洁的语法生成，能够延迟生成数据（即需要时才产生数据），大幅降低内存占用。

2. 核心定义

• 可迭代对象（Iterable）：能够被 for 循环遍历的对象，如列表、元组、字符串、字典等，实现了 __iter__ 方法。
• 迭代器（Iterator）：实现了 __iter__ 方法和 __next__ 方法的对象，能够通过 next() 函数逐个获取元素，遍历结束时抛出 StopIteration 异常。
• 生成器（Generator）：一种特殊的迭代器，通过 yield 关键字定义的函数，调用该函数时不执行函数体，而是返回一个生成器对象，每次调用 next() 或迭代时，执行到 yield处暂停并返回值，再次调用时从暂停处继续执行。

3. 本质作用

延迟计算（惰性求值），避免一次性将大量数据加载到内存，提高内存使用效率，适用于处理超大文件、海量日志、无限序列等场景。

4. 实现方式

• 迭代器实现：

  • 方式1：自定义类，实现 __iter__（返回自身）和 __next__（定义元素获取逻辑）方法。
  • 方式2：使用 iter() 函数将可迭代对象转换为迭代器。

• 生成器实现：

  • 方式1：生成器函数，使用yield 关键字，函数调用返回生成器对象。
  • 方式2：生成器表达式，语法类似列表推导式，将 [] 改为 ()，直接生成生成器对象。

5. 实战场景：用生成器读取济南某车企100万行设备日志并筛选故障记录

（1）实战用到的方法说明

• open() 函数：用于打开文件，默认以文本模式读取，支持上下文管理器（with 语句）自动关闭文件。
• 生成器函数：通过 yield 逐行读取日志，避免一次性加载100万行数据到内存。
• 字符串处理方法：strip()（去除换行符和空格）、find() 或 in 关键字（筛选故障记录，如包含“故障”“异常”“错误”等关键词）。
• time 模块：用于统计日志处理时间，确保≤3分钟。
• psutil 模块：用于监控内存占用，确保≤100MB（需提前安装：pip install psutil）。

（2）示例代码

import time
import psutil
import os

def read_large_log(file_path: str) -> str:
    """
    生成器函数：逐行读取超大日志文件
    :param file_path: 日志文件路径
    :yield: 每一行日志内容（去除换行符）
    """
    # 使用with语句打开文件，自动关闭，避免资源泄露
    with open(file_path, "r", encoding="utf-8") as f:
        for line in f:  # 文件对象本身是迭代器，逐行读取
            yield line.strip()  # 去除换行符和前后空格，返回一行日志

def filter_fault_records(log_generator, fault_keywords: list = None) -> str:
    """
    筛选故障记录的生成器
    :param log_generator: 日志生成器（read_large_log返回的对象）
    :param fault_keywords: 故障关键词列表，默认包含常见故障标识
    :yield: 包含故障关键词的日志记录
    """
    if fault_keywords is None:
        fault_keywords = ["故障", "异常", "错误", "告警", "失效", "故障码"]
    for log in log_generator:
        # 检查日志是否包含任意一个故障关键词
        if any(keyword in log for keyword in fault_keywords):
            yield log

def save_fault_records(fault_generator, save_path: str) -> None:
    """
    保存筛选出的故障记录到文件
    :param fault_generator: 故障日志生成器
    :param save_path: 故障日志保存路径
    """
    # 若保存目录不存在，创建目录
    save_dir = os.path.dirname(save_path)
    if not os.path.exists(save_dir):
        os.makedirs(save_dir)
    
    # 逐行写入故障日志，避免一次性写入大量数据
    with open(save_path, "w", encoding="utf-8") as f:
        for fault_log in fault_generator:
            f.write(fault_log + "\n")

def monitor_resource() -> dict:
    """监控当前进程的内存占用"""
    process = psutil.Process(os.getpid())
    memory_info = process.memory_info()
    return {
        "memory_usage_mb": memory_info.rss / 1024 / 1024,  # 内存占用（MB）
        "cpu_usage": process.cpu_percent(interval=0.1)  # CPU使用率（%）
    }

# 测试用例（模拟100万行日志处理）
if __name__ == "__main__":
    # 1. 准备：若没有实际日志文件，可先生成模拟日志（100万行）
    mock_log_path = "jn_auto_equipment.log"
    fault_save_path = "jn_auto_fault_records.log"
    
    # 生成模拟日志（仅用于测试，实际使用时替换为真实日志路径）
    print("生成模拟100万行设备日志...")
    with open(mock_log_path, "w", encoding="utf-8") as f:
        for i in range(1000000):
            if i % 1000 == 0:  # 每1000行生成1条故障日志
                f.write(f"2026-01-15 {i//3600:02d}:{i%3600//60:02d}:{i%60:02d} 设备{str(i%100).zfill(3)} 故障码：E{i%100} 温度异常\n")
            else:
                f.write(f"2026-01-15 {i//3600:02d}:{i%3600//60:02d}:{i%60:02d} 设备{str(i%100).zfill(3)} 运行正常 温度：{25 + i%10}℃\n")
    
    # 2. 开始处理日志，统计时间和内存
    start_time = time.time()
    
    # 读取日志生成器
    log_gen = read_large_log(mock_log_path)
    # 筛选故障日志生成器
    fault_gen = filter_fault_records(log_gen)
    # 保存故障日志
    save_fault_records(fault_gen, fault_save_path)
    
    # 统计处理时间
    end_time = time.time()
    process_time = end_time - start_time
    
    # 监控内存占用
    resource_info = monitor_resource()
    
    # 输出结果
    print(f"日志处理完成！")
    print(f"处理时间：{process_time:.2f} 秒（要求≤180秒）")
    print(f"内存占用：{resource_info['memory_usage_mb']:.2f} MB（要求≤100MB）")
    print(f"故障日志已保存到：{fault_save_path}")
    
    # 验证故障日志数量（模拟数据应生成1000条故障日志）
    with open(fault_save_path, "r", encoding="utf-8") as f:
        fault_count = len(f.readlines())
    print(f"筛选出的故障日志数量：{fault_count} 条")
    

（三）上下文管理器

1. 基础知识点讲解

上下文管理器是Python中用于管理资源（如文件、数据库连接、网络连接）的机制，能够确保资源在使用后被正确释放，避免资源泄露。核心是通过 with 语句实现资源的自动获取和释放，即使在使用过程中出现异常，也能保证资源被正确清理。

2. 核心定义

上下文管理器（Context Manager）：实现了上下文管理协议的对象，即同时实现 __enter__ 方法和 __exit__ 方法。__enter__ 方法在进入 with 代码块时执行，用于获取资源并返回资源对象；__exit__ 方法在退出 with 代码块时执行（无论是否发生异常），用于释放资源。

3. 本质作用

自动化资源管理，简化资源获取和释放的代码逻辑，避免因忘记释放资源或异常导致的资源泄露（如数据库连接未关闭、文件句柄未释放）。

4. 实现方式

• 类实现（自定义上下文管理器）：创建类并实现 __enter__ 和 __exit__ 方法。__exit__ 方法接收3个参数（exc_type, exc_val, exc_tb），分别表示异常类型、异常值、异常追踪信息，若返回 True 则表示捕获并处理异常，不向外抛出。
• contextlib.contextmanager 装饰器实现（）：使用 contextlib 模块的contextmanager 装饰器，将一个生成器函数转换为上下文管理器。生成器函数中，yield 之前的代码对应 __enter__ 方法的逻辑，用于获取资源；yield 之后的代码对应 __exit__ 方法的逻辑，用于释放资源。

5. 实战场景：封装MySQL连接上下文管理器（支持断连自动重试3次）

（1）实战用到的方法说明

• pymysql 模块：Python连接MySQL数据库的第三方库，提供数据库连接、游标、执行SQL等功能（需提前安装：pip install pymysql）。
• contextlib.contextmanager 装饰器：用于快速实现上下文管理器，简化代码。
• 重试机制：通过循环实现断连自动重试，最多重试3次，每次重试前添加短暂延迟（避免频繁重试）。
• 异常处理：捕获 pymysql 相关异常（如连接异常、执行异常），确保异常时资源正常释放。

（2）示例代码

import pymysql
from pymysql.err import OperationalError, ProgrammingError
from contextlib import contextmanager
import time

@contextmanager
def mysql_context_manager(
    host: str,
    port: int,
    user: str,
    password: str,
    db: str,
    charset: str = "utf8mb4",
    retry_count: int = 3,
    retry_delay: float = 1.0
) -> pymysql.cursors.Cursor:
    """
    MySQL连接上下文管理器，支持断连自动重试
    :param host: MySQL主机地址
    :param port: MySQL端口
    :param user: 用户名
    :param password: 密码
    :param db: 数据库名
    :param charset: 字符集
    :param retry_count: 重试次数
    :param retry_delay: 重试延迟（秒）
    :yield: 数据库游标对象（用于执行SQL）
    """
    conn = None
    cursor = None
    current_retry = 0
    
    # 连接重试逻辑
    while current_retry < retry_count:
        try:
            # 1. 建立数据库连接（__enter__ 逻辑）
            conn = pymysql.connect(
                host=host,
                port=port,
                user=user,
                password=password,
                db=db,
                charset=charset,
                cursorclass=pymysql.cursors.DictCursor  # 返回字典类型结果
            )
            cursor = conn.cursor()
            print(f"MySQL连接成功（重试次数：{current_retry}）")
            yield cursor  # 返回游标对象给with语句
            # 若执行到此处，说明with代码块正常执行，提交事务
            conn.commit()
            print("事务提交成功")
            break  # 连接成功，退出重试循环
        except OperationalError as e:
            # 连接异常，进行重试
            current_retry += 1
            print(f"MySQL连接失败（错误：{str(e)}），即将进行第 {current_retry} 次重试...")
            if current_retry >= retry_count:
                print(f"重试次数耗尽（共{retry_count}次），连接失败")
                raise  # 重试耗尽，抛出异常
            time.sleep(retry_delay)  # 重试前延迟
        except ProgrammingError as e:
            # SQL语法错误等，不重试，直接回滚事务并抛出异常
            print(f"SQL执行错误：{str(e)}，事务回滚")
            if conn:
                conn.rollback()
            raise
        except Exception as e:
            # 其他异常，回滚事务并抛出
            print(f"未知错误：{str(e)}，事务回滚")
            if conn:
                conn.rollback()
            raise
        finally:
            # 2. 释放资源（__exit__ 逻辑），无论是否发生异常都执行
            if cursor:
                cursor.close()
                print("游标已关闭")
            if conn:
                conn.close()
                print("MySQL连接已关闭")

# 测试用例
if __name__ == "__main__":
    # 数据库配置（需替换为实际配置）
    mysql_config = {
        "host": "localhost",
        "port": 3306,
        "user": "root",
        "password": "your_mysql_password",
        "db": "auto_equipment_db",
        "charset": "utf8mb4"
    }
    
    # 1. 测试正常连接和执行SQL
    try:
        with mysql_context_manager(**mysql_config) as cursor:
            # 创建设备日志表（若不存在）
            create_table_sql = """
            CREATE TABLE IF NOT EXISTS equipment_log (
                id INT AUTO_INCREMENT PRIMARY KEY,
                log_time DATETIME NOT NULL,
                device_id VARCHAR(20) NOT NULL,
                log_content VARCHAR(500) NOT NULL,
                fault_flag TINYINT(1) NOT NULL DEFAULT 0 COMMENT '0-正常，1-故障'
            ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
            """
            cursor.execute(create_table_sql)
            print("设备日志表创建成功（或已存在）")
            
            # 插入一条测试数据
            insert_sql = """
            INSERT INTO equipment_log (log_time, device_id, log_content, fault_flag)
            VALUES ('2026-01-15 10:30:00', 'DEV001', '设备运行正常，温度28℃', 0);
            """
            cursor.execute(insert_sql)
            print("测试数据插入成功")
            
            # 查询数据
            select_sql = "SELECT * FROM equipment_log WHERE device_id = 'DEV001'"
            cursor.execute(select_sql)
            result = cursor.fetchall()
            print(f"查询结果：{result}")
    except Exception as e:
        print(f"正常测试执行失败：{str(e)}")
    
    # 2. 测试连接失败重试（故意写错端口）
    print("\n" + "="*50 + "\n测试连接失败重试机制...")
    error_config = mysql_config.copy()
    error_config["port"] = 3307  # 错误端口
    try:
        with mysql_context_manager(**error_config) as cursor:
            pass
    except Exception as e:
        print(f"重试测试执行结果：{str(e)}")
    

（四）高阶函数

1. 基础知识点讲解

高阶函数是Python函数式编程的核心，指满足以下两个条件之一的函数：① 接受一个或多个函数作为参数；② 返回一个函数。高阶函数能够实现函数的灵活组合和复用，常用于批量数据转换、过滤、排序等场景，简化数据处理逻辑。

2. 核心定义

高阶函数（Higher-Order Function）：可以操作函数的函数，即把函数作为参数传递，或把函数作为返回值返回。Python内置的高阶函数有 map()、filter()、sorted()、reduce()（需从 functools 导入）等。

3. 本质作用

将数据处理逻辑与函数本身解耦，通过传递不同的函数参数，实现对不同数据的多样化处理（如批量转换数据格式、批量过滤数据），提高代码的灵活性和复用性。

4. 实现方式

• 自定义高阶函数：定义函数时，参数列表中包含函数类型参数，或函数体内返回一个新函数。

• 使用内置高阶函数：

  • map(func, iterable)：将函数 func 应用于可迭代对象iterable 的每个元素，返回一个迭代器，包含每次函数调用的结果。
  • filter(func, iterable)：用函数 func 判断可迭代对象 iterable 的每个元素，返回一个迭代器，包含所有使 func 返回 True 的元素。
  • sorted(iterable, key=None, reverse=False)：对可迭代对象排序，key 参数接受一个函数，用于指定排序的依据（如按元素的某个属性排序）。
  • reduce(func, iterable[, initial])：将函数 func 应用于可迭代对象 iterable 的元素，累积计算得到一个结果（func 需接受两个参数，返回一个结果）。

5. 实战场景：批量数据转换（车企设备日志数据格式化）

（1）实战用到的方法说明

• map() 函数：批量将原始日志字符串转换为结构化字典数据。
• filter() 函数：批量过滤无效日志（如格式错误、缺失关键信息的日志）。
• 字符串处理函数：split()（分割日志字段）、strip()（去除空格）、datetime.strptime()（将字符串时间转换为datetime对象）。
• 匿名函数（lambda）：用于简化简单函数的定义，作为高阶函数的参数。

（2）示例代码

from datetime import datetime
from functools import reduce

# 1. 定义数据转换函数
def parse_log(log_line: str) -> dict or None:
    """
    解析单条设备日志，转换为结构化字典
    原始日志格式：2026-01-15 10:30:00 设备001 运行正常 温度28℃
    结构化格式：{
        "log_time": datetime对象,
        "device_id": str,
        "status": str,
        "temperature": int  # 温度数值（提取℃前的数字）
    }
    :param log_line: 原始日志字符串
    :return: 结构化字典，格式错误返回None
    """
    try:
        # 按空格分割日志（注意：日志内容中的空格需特殊处理，此处假设字段间为单个空格）
        parts = log_line.split()
        if len(parts) < 5:
            print(f"无效日志（字段不足）：{log_line}")
            return None
        
        # 提取各字段
        log_time_str = f"{parts[0]} {parts[1]}"
        device_id = parts[2]
        status = parts[3]
        temperature_str = parts[4]
        
        # 转换时间格式
        log_time = datetime.strptime(log_time_str, "%Y-%m-%d %H:%M:%S")
        
        # 提取温度数值（去除℃）
        temperature = int(temperature_str.replace("℃", ""))
        
        return {
            "log_time": log_time,
            "device_id": device_id,
            "status": status,
            "temperature": temperature
        }
    except Exception as e:
        print(f"日志解析失败（{str(e)}）：{log_line}")
        return None

# 2. 定义无效日志过滤函数
def is_valid_log(parsed_log: dict or None) -> bool:
    """
    过滤无效日志（解析失败的日志）
    :param parsed_log: 解析后的日志字典
    :return: 有效返回True，无效返回False
    """
    return parsed_log is not None

# 3. 定义批量转换函数（高阶函数封装）
def batch_convert_logs(log_lines: list) -> list:
    """
    批量转换日志数据：解析+过滤
    :param log_lines: 原始日志列表
    :return: 结构化的有效日志列表
    """
    # 使用map批量解析日志
    parsed_logs = map(parse_log, log_lines)
    # 使用filter批量过滤无效日志
    valid_logs = filter(is_valid_log, parsed_logs)
    # 转换为列表返回（迭代器转换为列表，方便后续使用）
    return list(valid_logs)

# 4. 定义批量统计函数（使用reduce）
def batch_stat_temperature(valid_logs: list) -> dict:
    """
    批量统计设备温度：最高温、最低温、平均温
    :param valid_logs: 结构化的有效日志列表
    :return: 温度统计字典
    """
    if not valid_logs:
        return {"max_temp": None, "min_temp": None, "avg_temp": None}
    
    # 使用reduce计算最高温
    max_temp = reduce(lambda x, y: x if x["temperature"] > y["temperature"] else y, valid_logs)["temperature"]
    # 使用reduce计算最低温
    min_temp = reduce(lambda x, y: x if x["temperature"] < y["temperature"] else y, valid_logs)["temperature"]
    # 计算平均温
    total_temp = reduce(lambda x, y: x + y["temperature"], valid_logs, 0)
    avg_temp = round(total_temp / len(valid_logs), 2)
    
    return {
        "max_temp": max_temp,
        "min_temp": min_temp,
        "avg_temp": avg_temp
    }

# 测试用例
if __name__ == "__main__":
    # 原始日志数据（模拟批量日志）
    raw_logs = [
        "2026-01-15 10:30:00 设备001 运行正常 温度28℃",
        "2026-01-15 10:31:00 设备002 运行正常 温度30℃",
        "2026-01-15 10:32:00 设备003 故障 温度45℃",  # 故障日志
        "2026-01-15 10:33:00 设备004 运行正常 温度26℃",
        "无效日志内容",  # 无效日志
        "2026-01-15 10:34:00 设备005 运行正常 温度29℃",
        "2026-01-15 10:35:00 设备006 异常 温度38℃"   # 异常日志
    ]
    
    # 1. 批量转换日志
    valid_logs = batch_convert_logs(raw_logs)
    print("批量转换后的有效日志：")
    for log in valid_logs:
        print(log)
    
    # 2. 批量统计温度
    temp_stat = batch_stat_temperature(valid_logs)
    print("\n温度统计结果：")
    print(f"最高温度：{temp_stat['max_temp']}℃")
    print(f"最低温度：{temp_stat['min_temp']}℃")
    print(f"平均温度：{temp_stat['avg_temp']}℃")
    
    # 3. 扩展：按温度排序（使用sorted高阶函数）
    sorted_logs = sorted(valid_logs, key=lambda x: x["temperature"], reverse=True)
    print("\n按温度降序排序后的日志：")
    for log in sorted_logs:
        print(log)
    

二、可执行练习步骤

本部分按「前置准备→分步操作→验证方法→问题排查」结构设计，覆盖第1周3个核心实战任务，严格贴合学习目标要求。

练习一：开发日志装饰器（支持日志分级、错误邮件提醒）

1. 前置准备

• 环境配置：确保已安装Python（3.7+）、VS Code，在VS Code中创建新的Python文件（命名为 log_decorator.py）。
• 依赖安装：本练习使用内置模块（logging、smtplib、email、functools），无需额外安装依赖。
• 邮件配置：准备一个可用的邮箱（如163、QQ邮箱），获取SMTP服务器地址、端口和授权码（163邮箱授权码需在网页端设置中开启）。
• 参考资料：《流畅的Python》第5章（装饰器相关内容）。

2. 分步操作

1. 步骤1：配置基础日志（符合企业规范）

   • 编写 setup_logger 函数，配置日志级别为 DEBUG，添加控制台处理器（输出 INFO 及以上）和文件处理器（按天分割，保留7天）。
   • 定义企业规范日志格式："%(asctime)s - %(levelname)s - %(funcName)s - %(message)s"，包含时间、日志级别、函数名、日志信息。

2. 步骤2：实现邮件发送工具函数

   • 编写 send_error_email 函数，接收邮件主题、内容和配置，使用 smtplib.SMTP_SSL 连接SMTP服务器，发送邮件。
   • 处理邮件发送异常，确保发送失败时能打印错误信息。

3. 步骤3：实现带参数的日志装饰器

   • 定义 log_decorator 函数，接收日志级别和邮件配置参数，返回装饰器函数。
   • 在装饰器内部嵌套 wrapper 函数，使用 functools.wraps 保留原函数元信息。
   • 在 wrapper 函数中，记录函数执行前后的日志，捕获异常并记录错误日志，若配置邮件则触发错误邮件提醒。

4. 步骤4：编写测试用例验证

   • 定义2个测试函数（如 add 加法函数、divide 除法函数），使用 @log_decorator 语法糖装饰，配置不同的日志级别和邮件参数。
   • 调用测试函数，包括正常执行和异常执行（如 divide(1, 0) 触发除零错误）。

3. 验证方法

• 查看控制台输出：正常执行时能看到 INFO/DEBUG 级日志，异常执行时能看到 ERROR 级日志和异常堆栈。
• 查看日志文件：项目目录下生成 business.log 文件，包含所有 DEBUG 及以上级别日志，按天分割。
• 验证邮件提醒：异常执行时，目标邮箱能收到错误告警邮件，包含函数名和错误信息。

4. 问题排查

• 日志文件未生成：检查日志处理器是否添加到logger对象，确保 setup_logger 函数中没有重复添加处理器的逻辑（避免多次调用时处理器重复）。
• 邮件发送失败：检查SMTP服务器地址、端口是否正确（如163邮箱SMTP服务器为 smtp.163.com，端口465）；确认授权码正确，而非邮箱登录密码；检查网络连接，确保能访问SMTP服务器。
• 原函数元信息丢失：确保在 wrapper 函数上添加了 @wraps(func) 装饰器。
• 异常未捕获：检查 wrapper 函数中的 try-except 块是否覆盖了函数执行的完整逻辑，确保异常能被捕获并记录。

练习二：用生成器读取100万行设备日志并筛选故障记录

1. 前置准备

• 环境配置：继续使用VS Code，创建新文件log_generator.py。
• 依赖安装：安装 psutil 模块（监控内存）：pip install psutil。
• 日志数据准备：若没有真实的100万行日志，使用代码生成模拟日志（参考示例代码中的模拟日志生成逻辑）。
• 参考资料：《流畅的Python》第6章（生成器与迭代器相关内容）。

2. 分步操作

1. 步骤1：实现逐行读取日志的生成器

   • 编写 read_large_log 生成器函数，使用 with open() 打开日志文件，逐行读取并 yield 去除换行符的日志内容。

2. 步骤2：实现故障日志筛选生成器

   • 编写 filter_fault_records 函数，接收日志生成器和故障关键词列表，遍历日志并筛选包含任意关键词的故障日志，yield 故障日志。
   • 默认故障关键词：["故障", "异常", "错误", "告警", "失效", "故障码"]。

3. 步骤3：实现故障日志保存功能

   • 编写 save_fault_records 函数，接收故障日志生成器和保存路径，逐行写入故障日志到文件，避免一次性加载大量数据。
   • 添加目录创建逻辑，确保保存目录不存在时能自动创建。

4. 步骤4：添加资源监控和时间统计

   • 编写 monitor_resource 函数，使用 psutil.Process 获取当前进程的内存占用（转换为MB）和CPU使用率。
   • 在主程序中，记录日志处理的开始和结束时间，计算处理耗时。

5. 步骤5：执行测试并验证性能指标

   • 生成100万行模拟日志（或使用真实日志），调用上述函数完成日志读取、筛选、保存。
   • 输出处理时间和内存占用，验证是否满足要求（处理时间≤3分钟，内存占用≤100MB）。

3. 验证方法

• 性能指标验证：处理100万行日志时，处理时间≤180秒，内存占用≤100MB（通过 monitor_resource 函数输出验证）。
• 结果验证：查看保存的故障日志文件，确认筛选出的日志均包含故障关键词，无遗漏或误筛。
• 资源泄露验证：处理完成后，检查进程内存是否正常释放（可通过任务管理器/活动监视器查看Python进程内存占用）。

4. 问题排查

• 内存占用过高：检查是否在生成器之外使用了一次性读取全量数据的操作（如read()或readlines()），确保所有日志处理逻辑均基于生成器逐行处理；若仍过高，可检查是否存在未关闭的文件句柄，确保所有文件操作都使用with语句自动管理资源。
• 处理时间过长：优化字符串匹配逻辑，避免在循环内使用复杂的正则表达式或多次字符串切割；若日志文件较大，可考虑关闭文件编码校验（明确日志编码格式时），或使用更高效的文件读取模式（如二进制模式读取后再解码）。
• 故障日志筛选遗漏/误筛：检查故障关键词列表是否完整，补充业务场景下的特殊故障标识（如特定故障码“E001”“E002”等）；优化匹配逻辑，处理日志中关键词被空格分隔的情况（如“故障 代码”），可使用正则表达式re.search(r'故障|异常|错误', log)增强匹配灵活性。
• 模拟日志生成过慢：若生成100万行模拟日志耗时较长，可减少循环内的字符串格式化操作，提前拼接固定格式字符串，仅替换变量部分；或直接使用现成的大日志文件进行测试。

练习三：封装MySQL连接上下文管理器（支持断连自动重试）

1. 前置准备

• 环境配置：在VS Code中创建新文件mysql_context.py，确保本地或远程MySQL服务正常运行。
• 依赖安装：安装pymysql模块（MySQL连接工具）：pip install pymysql。
• 数据库准备：提前创建测试数据库（如auto_equipment_db），无需手动创表（示例代码含建表逻辑）；获取MySQL服务的主机地址、端口、用户名、密码等配置信息。
• 参考资料：《流畅的Python》第7章（上下文管理器相关内容）、pymysql官方文档。

2. 分步操作

1. 步骤1：导入依赖模块导入pymysql模块及相关异常类（OperationalError连接异常、ProgrammingErrorSQL语法异常）。
2. 导入contextlib.contextmanager装饰器，用于快速实现上下文管理器；导入time模块用于重试延迟。
3. 步骤2：实现带重试机制的MySQL上下文管理器定义mysql_context_manager函数，添加@contextmanager装饰器，接收MySQL连接配置参数（主机、端口、用户名、密码等）及重试参数（重试次数、延迟时间）。
4. 在函数内部初始化数据库连接（conn）和游标（cursor）为None，设置重试计数器current_retry。
5. 使用while循环实现重试逻辑：尝试通过pymysql.connect()建立数据库连接，创建游标对象，yield游标给with语句块（对应__enter__逻辑）。
6. 若连接成功，执行with代码块后提交事务；若捕获OperationalError（连接异常），则重试计数器加1，延迟后重新连接，重试耗尽则抛出异常。
7. 捕获ProgrammingError（SQL语法错误）或其他异常时，回滚事务并直接抛出异常（不重试）。
8. 在finally块中释放资源：关闭游标和数据库连接（对应__exit__逻辑），确保无论是否异常都能释放资源。
9. 步骤3：编写测试用例验证定义MySQL连接配置字典，填写实际的数据库连接信息（主机、端口、用户名、密码、数据库名等）。
10. 测试正常连接场景：使用with mysql_context_manager(**mysql_config) as cursor，执行创表、插入数据、查询数据的SQL语句，验证事务提交和数据操作正确性。
11. 测试连接失败重试场景：故意修改MySQL端口（如将3306改为3307），执行with语句，验证是否能触发3次重试，重试耗尽后抛出异常并打印错误信息。
12. 测试SQL语法错误场景：执行语法错误的SQL语句（如SELECT * FROM non_existent_table），验证是否能回滚事务并抛出ProgrammingError。

3. 验证方法

• 正常连接验证：执行测试用例后，登录MySQL数据库，查看equipment_log表是否创建成功，插入的测试数据是否存在，查询结果是否正确。
• 重试机制验证：修改错误端口后，控制台应打印“MySQL连接失败，即将进行第X次重试”的日志，共打印3次重试信息后抛出连接失败异常。
• 异常处理验证：执行错误SQL时，控制台打印“SQL执行错误，事务回滚”，数据库中无无效数据插入，确保事务一致性。
• 资源释放验证：测试完成后，通过MySQL命令show processlist查看连接数，确认无残留的未关闭连接（即上下文管理器正确释放资源）。

4. 问题排查

• MySQL连接失败（无重试）：检查重试逻辑是否正确包裹pymysql.connect()代码块，确保OperationalError异常被正确捕获；确认retry_count参数未被误设为0。
• 事务提交失败/数据未插入：检查yield语句后是否有conn.commit()逻辑，确保with代码块正常执行时会提交事务；若存在异常回滚，需确认异常类型是否被正确分类（如SQL语法错误应回滚，连接异常无需回滚）。
• 游标关闭失败：在finally块中添加判断逻辑，确保仅在游标和连接不为None时执行关闭操作（避免对None对象调用close()方法）。
• 中文乱码问题：确认MySQL数据库、表的字符集均为utf8mb4，且上下文管理器中charset参数设置为utf8mb4，避免插入中文日志时出现乱码。
• 重试延迟无效：检查time.sleep(retry_delay)是否在catch块内、重试计数器递增之后，确保每次重试前都会延迟指定时间。

三、知识点关联与学习建议

1. 核心知识点关联梳理

本周学习的装饰器、生成器/迭代器、上下文管理器、高阶函数并非孤立存在，在企业实战中常组合使用：

• 装饰器 + 生成器：可通过装饰器为生成器函数添加日志记录、性能统计功能（如统计生成器逐行处理数据的耗时）。
• 上下文管理器 + 生成器：contextlib.contextmanager装饰器本质就是通过生成器简化上下文管理器实现，避免手动编写__enter__和__exit__方法。
• 高阶函数 + 生成器：map()、filter()等高阶函数可直接作用于生成器对象，实现数据的流式处理（如先通过生成器读取日志，再用map()解析日志，最后用filter()筛选故障记录）。
• 上下文管理器 + 装饰器：可通过装饰器为数据库连接上下文管理器添加重试次数、超时控制等增强功能，提高连接稳定性。

2. 学习建议与进阶方向

• 基础巩固：每个知识点至少独立完成2个不同场景的实战案例（如装饰器除了日志记录，还可实现接口权限校验；生成器除了日志处理，还可实现海量数据的分批导出）。
• 源码阅读：阅读Python内置模块源码（如contextlib、functools.wraps），理解高阶语法的底层实现逻辑，加深对闭包、协议（迭代协议、上下文协议）的理解。
• 进阶拓展：学习第三方库中的进阶应用，如tenacity库（基于装饰器的重试机制）、pandas库中的迭代器（处理超大CSV文件）、SQLAlchemy中的上下文管理器（数据库会话管理）。
• 实战复盘：完成本周3个核心练习后，尝试整合为一个完整项目（如“车企设备日志处理系统”）：用生成器读取日志→高阶函数解析筛选→上下文管理器连接MySQL保存结果→装饰器记录整个流程的日志和异常邮件提醒。

3. 常见误区规避

• 装饰器：忘记使用functools.wraps保留原函数元信息，导致help(func)或func.__name__获取错误信息，影响调试和文档生成。
• 生成器：误以为生成器函数调用时会执行函数体，实际需通过next()或迭代触发执行；避免在生成器中使用return语句（会抛出StopIteration异常，影响正常迭代）。
• 上下文管理器：手动实现时遗漏__exit__方法中的异常处理逻辑，导致异常时资源无法释放；使用contextlib.contextmanager时，yield语句后未处理资源释放（如未关闭文件、数据库连接）。
• 高阶函数：过度使用lambda函数导致代码可读性下降，复杂逻辑建议定义独立函数作为参数；混淆map()、filter()的返回值（均为迭代器，需转换为列表才能直接查看结果）。

四、总结

本周核心是掌握Python进阶语法的“工程化应用”能力，重点在于理解各知识点的“本质作用”（如装饰器实现解耦、生成器优化内存、上下文管理器管理资源），而非仅记忆语法格式。通过企业真实场景（日志处理、数据库连接）的实战练习，需达到“能灵活组合知识点解决实际问题”的目标，为后续AI项目开发（如海量训练数据处理、模型训练过程监控）打下基础。后续学习中，应持续强化“用合适的语法解决对应的工程问题”的思维，避免为了使用进阶语法而过度设计代码。

（注：文档部分内容可能由 AI 生成）