接下来将系统讲解 Python 的基础语法和四大核心数据结构（list、tuple、dict、set）以及它们特有的推导式。这是 Python 入门的重中之重，掌握它们就可以写绝大多数脚本了。

---

一、Python 基础语法（三大特色）

1. 动态类型（Dynamic Typing）

• 变量不需要声明类型，可以直接赋值，Python 会在运行时自动推断类型。
• 同一个变量可以先后指向不同类型的对象。

x = 10          # 此时 x 是整数
print(type(x))  # <class 'int'>

x = "Hello"     # 现在 x 变成字符串（允许）
print(type(x))  # <class 'str'>

x = [1, 2, 3]   # 又变成列表

注意：动态类型很方便，但也容易导致运行时类型错误（TypeError）。编写代码时建议保持变量用途单一。

2. 分号（;）可选，但几乎不用

• Python 语句末尾不需要分号，直接换行即可。
• 分号可以用于在一行内写多条语句，但强烈不推荐，会降低可读性。

# 正确写法
a = 1
b = 2
c = a + b

# 虽然合法，但别这样写
a = 1; b = 2; c = a + b   # 丑陋，不 Pythonic

3. 缩进决定代码块（Indentation Matters）

• Python 使用缩进（通常 4 个空格）来区分代码块，而不是 {} 或 end。
• 同一代码块内的语句必须拥有相同的缩进量。

# if 语句块
score = 85
if score >= 60:
    print("及格了")        # 缩进 4 空格
    print("继续努力")      # 同一块，相同缩进
else:
    print("不及格")        # else 与 if 对齐，其下属语句缩进

# 循环
for i in range(3):
    print(i)              # 循环体缩进
print("循环结束")          # 非循环体，无缩进

常见错误：混用 Tab 和空格，或者缩进不一致 → IndentationError。建议编辑器设置「Tab 转为空格」。

---

二、四大核心数据结构

1. 列表（list）—— 有序、可变

• 特点：任意类型元素、可变（增删改）、有序、可重复。
• 创建：[1, 2, 3] 或 list((1,2,3))
• 常用操作：索引、切片、append()、extend()、insert()、remove()、pop()、sort()、reverse() 等。

fruits = ["apple", "banana", "cherry"]
fruits.append("orange")      # 末尾添加
fruits[1] = "blueberry"      # 修改
del fruits[0]                # 删除
print(fruits[1:])            # 切片 ['blueberry', 'orange']

列表推导式（List Comprehension）

• 语法：[表达式 for 变量 in 可迭代对象 if 条件]
• 作用：用一行循环生成新列表，比传统 for 循环更简洁、更快。

# 传统写法
squares = []
for x in range(10):
    squares.append(x**2)

# 列表推导式
squares = [x**2 for x in range(10)]

# 带条件：筛选偶数
even_squares = [x**2 for x in range(10) if x % 2 == 0]
print(even_squares)   # [0, 4, 16, 36, 64]

---

2. 元组（tuple）—— 有序、不可变

• 特点：元素不可修改（类似只读列表），可哈希（可作为字典的键）。
• 创建：(1, 2, 3) 或 1,2,3（括号可省略）。
• 常用操作：索引、切片、count()、index()。

point = (10, 20)
print(point[0])      # 10
# point[0] = 5       ❌ 报错，元组不可变

# 单元素元组需要加逗号
t = (5,)             # 正确，否则 (5) 只是整数

元组相关推导式？—— 没有元组推导式，但有生成器表达式

• 语法：(表达式 for 变量 in 可迭代对象 if 条件) 返回一个生成器（惰性求值）。
• 若需元组，可以用 tuple() 包裹生成器表达式。

# 生成器表达式（不立即生成所有值）
gen = (x**2 for x in range(10))
print(type(gen))     # <class 'generator'>

# 转为元组
t = tuple(x**2 for x in range(10))
print(t)             # (0, 1, 4, 9, 16, 25, 36, 49, 64, 81)

实战提示：当数据不需要修改时优先用元组，占用内存更小、性能略高，且可做字典的键。

---

3. 字典（dict）—— 键值对映射

• 特点：无序（Python 3.7+ 实际保留插入顺序，但不要依赖）、键必须不可变（通常用字符串、数字、元组）、值任意。
• 创建：{"name": "Alice", "age": 25} 或 dict(name="Alice", age=25)
• 常用操作：通过键访问/赋值、keys()、values()、items()、get()、pop()、update()。

person = {"name": "Bob", "age": 30}
print(person["name"])           # Bob
person["city"] = "New York"     # 添加新键值对
del person["age"]               # 删除键
print(person.get("salary", 0))  # 安全获取，不存在返回 0

字典推导式

• 语法：{key_expr: value_expr for 变量 in 可迭代对象 if 条件}

# 将列表转为字典：索引作为键，元素作为值
names = ["Alice", "Bob", "Charlie"]
index_dict = {i: name for i, name in enumerate(names)}
print(index_dict)   # {0: 'Alice', 1: 'Bob', 2: 'Charlie'}

# 筛选：只保留值大于 1 的键值对
squares = {x: x**2 for x in range(6) if x > 1}
print(squares)      # {2: 4, 3: 9, 4: 16, 5: 25}

---

4. 集合（set）—— 无序、不重复

• 特点：元素唯一、无序、可变。常用于去重和数学集合运算（并、交、差）。
• 创建：{1, 2, 3} 或 set([1,2,3])。注意空集合必须用 set()，因为 {} 是空字典。
• 常用操作：add()、remove()、discard()、union（|）、intersection（&）、difference（-）等。

a = {1, 2, 3}
a.add(4)                # {1,2,3,4}
a.remove(2)             # {1,3,4}
b = {3, 4, 5}
print(a & b)            # 交集 {3,4}
print(a | b)            # 并集 {1,3,4,5}

集合推导式

• 语法：{表达式 for 变量 in 可迭代对象 if 条件}

# 从列表中快速去重并生成集合
nums = [1, 2, 2, 3, 4, 4, 5]
unique = {x for x in nums}
print(unique)          # {1, 2, 3, 4, 5}

# 筛选条件：只保留偶数
even_set = {x for x in range(10) if x % 2 == 0}
print(even_set)        # {0, 2, 4, 6, 8}

---

三、四种推导式对比表

类型	写法示例	结果类型	适用场景
列表推导式	[x*2 for x in range(5)]	list	生成新列表，常用
元组“推导式”	tuple(x*2 for x in range(5))	tuple	需要不可变序列时
字典推导式	{x: x*2 for x in range(5)}	dict	从可迭代对象构造字典
集合推导式	{x*2 for x in range(5)}	set	去重或构造集合

记住：没有直接的元组推导式，因为 (expr for ...) 是生成器表达式，不是元组。

---

四、综合练习：用推导式解决实际问题

题目：有一个包含重复单词的列表 words = ["apple", "banana", "apple", "cherry", "banana", "date"]，要求：

1. 生成一个列表，包含每个单词的长度。
2. 生成一个集合，包含所有单词的首字母。
3. 生成一个字典，键为单词，值为该单词的长度（去重后）。

参考答案：

words = ["apple", "banana", "apple", "cherry", "banana", "date"]

# 1. 每个单词的长度
lengths = [len(w) for w in words]
print(lengths)   # [5, 6, 5, 6, 6, 4]

# 2. 首字母集合
first_letters = {w[0] for w in words}
print(first_letters)   # {'a', 'b', 'c', 'd'}

# 3. 单词->长度字典（去重，保留第一次出现的长度）
word_len = {w: len(w) for w in words}   # 重复键会被后面的覆盖，但长度相同无影响
print(word_len)        # {'apple': 5, 'banana': 6, 'cherry': 6, 'date': 4}

---

五、总结与建议

知识点	关键点
动态类型	变量不绑定类型，灵活但要小心运行时的类型错误。
无分号	每行一条语句，清晰易读。
缩进	强制统一风格，可读性高；切勿混用 Tab 和空格。
列表	有序、可变；最常用。
元组	有序、不可变；可做字典键。
字典	键值映射；查询效率极高。
集合	无序、唯一；去重和集合运算。
推导式	简洁高效，但不要写过于复杂的嵌套，可读性优先。

掌握了这些，你已经可以流畅地读写大多数 Python 代码了。下一步建议学习函数定义、文件操作和异常处理。

非常好！函数定义、文件操作和异常处理是 Python 实战编程的三大基石。掌握它们，你就能写出结构清晰、健壮、可复用的程序。下面我分三个部分系统讲解，每部分都配有示例和最佳实践。

---

一、函数定义（Function Definition）

函数用于封装重复使用的代码块，提高可读性和可维护性。

1. 基本语法

def 函数名(参数1, 参数2, ...):
    """文档字符串（可选，用于说明函数作用）"""
    函数体
    return 返回值   # 若无 return，返回 None

示例：

def greet(name):
    """向指定用户打招呼"""
    return f"Hello, {name}!"

msg = greet("Alice")
print(msg)   # Hello, Alice!

2. 参数类型详解

参数类型	写法	说明	示例
位置参数	def f(a, b)	按位置传递，必须传	f(1, 2)
默认参数	def f(a, b=10)	调用时可省略，默认值在函数定义时计算	f(1) → b=10
关键字参数	def f(a, b)	调用时指定参数名，顺序可换	f(b=2, a=1)
可变位置参数	def f(*args)	接收任意多个位置参数，打包为元组	f(1,2,3) → args=(1,2,3)
可变关键字参数	def f(**kwargs)	接收任意多个关键字参数，打包为字典	f(x=1, y=2) → kwargs={‘x’:1, ‘y’:2}

综合示例：

def demo(a, b=1, *args, **kwargs):
    print(f"a={a}, b={b}, args={args}, kwargs={kwargs}")

demo(10, 20, 30, 40, name="Tom", age=18)
# 输出: a=10, b=20, args=(30, 40), kwargs={'name': 'Tom', 'age': 18}

3. 返回值

• 可以返回任意类型（单个值、多个值自动打包为元组）。
• 没有 return 或只写 return，函数返回 None。

def add_sub(a, b):
    return a + b, a - b   # 返回元组

result = add_sub(10, 5)
print(result)   # (15, 5)
sum_val, diff_val = add_sub(10, 5)  # 解包赋值

4. 作用域（Scope）

• 局部变量：函数内部赋值，只在函数内可见。
• 全局变量：在函数外定义，函数内若要修改需用 global 关键字（不推荐滥用）。

x = 100          # 全局变量

def func():
    global x     # 声明要修改全局 x
    x = 200      # 修改全局变量
    y = 10       # 局部变量
    print(f"内部：x={x}, y={y}")

func()
print(f"外部：x={x}")   # 200

5. lambda 表达式（匿名函数）

适合简单的单行函数，无需 def。

square = lambda x: x ** 2
print(square(5))   # 25

# 常用于排序的 key
students = [("Alice", 25), ("Bob", 20), ("Charlie", 23)]
students.sort(key=lambda s: s[1])   # 按年龄排序

6. 最佳实践

• 函数名使用小写加下划线（snake_case）。
• 添加文档字符串（docstring），方便查看帮助。
• 函数的职责单一，不要过长（建议不超过 30 行）。
• 避免修改全局变量，尽量通过参数和返回值传递数据。

---

二、文件操作（File I/O）

Python 通过内置 open() 函数操作文件，支持文本和二进制模式。

1. 打开文件：open(file, mode, encoding)

模式	含义	文件指针位置
'r'	只读（默认）	文件开头
'w'	写入（会覆盖原内容或创建新文件）	文件开头
'a'	追加（写入到末尾）	文件末尾
'x'	独占创建，文件已存在则报错	文件开头
'b'	二进制模式（常与 r/w 结合，如 'rb'）	-
'+'	读写模式（如 'r+'）	取决于前面的模式

指定编码（重要）：

f = open("data.txt", "r", encoding="utf-8")

2. 读取文件

# 方法1：读取全部内容
with open("test.txt", "r", encoding="utf-8") as f:
    content = f.read()        # 整个文件为一个字符串
    print(content)

# 方法2：按行读取（适合大文件）
with open("test.txt", "r", encoding="utf-8") as f:
    for line in f:            # 每次迭代一行
        print(line.strip())   # strip 去掉换行符

# 方法3：读取所有行到列表
with open("test.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()     # 列表，每个元素带换行符

3. 写入文件

# 覆盖写入
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("Hello, World!\n")
    f.write("第二行内容")

# 追加写入
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("\n这是追加的行")

4. 使用 with 语句（推荐）

with 会自动关闭文件，即使发生异常也会正确关闭。比手动 f.close() 更安全。

# 推荐写法
with open("file.txt", "r") as f:
    data = f.read()
# 离开 with 块后，文件自动关闭

5. 操作二进制文件（如图片）

# 读取图片
with open("image.jpg", "rb") as f:
    binary_data = f.read()

# 写入二进制
with open("copy.jpg", "wb") as f:
    f.write(binary_data)

6. 文件指针操作（seek, tell）

• tell()：返回当前指针位置（字节数）。
• seek(offset, whence)：移动指针。whence=0 从开头（默认），1 从当前位置，2 从末尾。

with open("data.txt", "rb") as f:
    print(f.tell())    # 0
    f.seek(5)          # 移动到第 5 字节
    print(f.tell())    # 5
    f.seek(-3, 2)      # 从末尾向前移动 3 字节

7. 常见文件操作模块 os 和 pathlib

import os
import shutil

os.rename("old.txt", "new.txt")   # 重命名
os.remove("temp.txt")             # 删除文件
os.mkdir("mydir")                 # 创建目录
shutil.copy("src.txt", "dst.txt") # 复制文件

8. 最佳实践

• 总是使用 with 管理文件资源。
• 明确指定 encoding，避免跨平台乱码（尤其 Windows）。
• 处理大文件时逐行读取，不要一次 read() 整个文件。
• 检查文件是否存在可使用 os.path.exists() 或 try-except。

---

三、异常处理（Exception Handling）

异常处理让程序能够优雅地应对运行时错误，而不是直接崩溃。

1. 基本结构：try-except-else-finally

try:
    # 可能抛出异常的代码
    num = int(input("请输入数字: "))
    result = 10 / num
except ValueError:
    print("那不是有效的整数！")
except ZeroDivisionError:
    print("除数不能为 0！")
except Exception as e:          # 捕获所有其他异常（不推荐滥用）
    print(f"未知错误: {e}")
else:
    # 仅在 try 块没有异常时执行
    print(f"计算结果是 {result}")
finally:
    # 无论是否发生异常都会执行（常用于资源清理）
    print("执行完毕")

2. 常见内置异常

异常	触发场景
ValueError	类型转换失败、传入无效值
TypeError	对不支持的类型进行操作
ZeroDivisionError	除以 0
IndexError	列表索引越界
KeyError	字典键不存在
FileNotFoundError	打开不存在的文件
AttributeError	对象没有该属性/方法

3. 手动抛出异常：raise

def divide(a, b):
    if b == 0:
        raise ValueError("除数不能为 0")
    return a / b

try:
    divide(10, 0)
except ValueError as e:
    print(e)

4. 自定义异常（继承 Exception）

class NegativeNumberError(Exception):
    """自定义异常：负数不被允许"""
    pass

def check_positive(n):
    if n < 0:
        raise NegativeNumberError(f"负数 {n} 无效")
    return n

5. 异常处理的最佳实践

• 捕获具体异常，不要只用裸 except:（会捕获键盘中断等系统异常）。
• 不要过度使用异常：能通过 if 预判的常见情况（如文件存在性）尽量先判断，异常应处理真正“异常”的情况。
• 保持 try 块尽可能小，只包含可能出错的代码。
• finally 用于释放资源（如关闭文件、网络连接），即使 try 中有 return 也会执行。
• 记录异常日志（logging 模块）便于调试。

6. 结合文件操作的异常处理示例

def read_file_safely(filename):
    try:
        with open(filename, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        print(f"文件 {filename} 不存在")
        return None
    except PermissionError:
        print(f"没有权限读取 {filename}")
        return None
    except Exception as e:
        print(f"读取文件时发生其他错误: {e}")
        return None

---

四、综合实战：函数 + 文件 + 异常

需求：编写函数 process_numbers(filename)，读取一个文本文件中的整数（每行一个），返回它们的和与平均值。如果文件不存在或内容格式错误，要妥善处理。

def process_numbers(filename):
    """读取文件中的整数，返回总和和平均值"""
    numbers = []
    try:
        with open(filename, "r", encoding="utf-8") as f:
            for line_no, line in enumerate(f, 1):
                line = line.strip()
                if not line:         # 忽略空行
                    continue
                try:
                    num = int(line)
                except ValueError:
                    print(f"警告：第 {line_no} 行不是整数，已跳过")
                    continue
                numbers.append(num)
    except FileNotFoundError:
        print(f"错误：文件 '{filename}' 未找到")
        return None, None
    except Exception as e:
        print(f"读取文件时发生未知错误: {e}")
        return None, None

    if not numbers:
        print("文件中没有有效整数")
        return 0, 0

    total = sum(numbers)
    average = total / len(numbers)
    return total, average

# 使用示例
total, avg = process_numbers("data.txt")
if total is not None:
    print(f"总和: {total}, 平均值: {avg:.2f}")

---

五、总结与下一步

主题	核心要点
函数	def，参数类型（位置/默认/可变/关键字），返回值，作用域，lambda
文件操作	open 模式（r/w/a/b/+），with 语句，读写方法（read/write/readlines），编码指定，os 模块
异常处理	try-except-else-finally，捕获具体异常，raise，自定义异常，不滥用异常

下一步建议：学习模块与包（import）、迭代器与生成器、装饰器等高级特性。如果你有特定场景想练习（例如统计单词频率、批量处理文件、编写命令行工具等）

非常好！你已经掌握了基础语法、数据结构和函数/文件/异常处理，现在向进阶特性迈进：模块与包、迭代器与生成器、装饰器。这些特性会让你的代码更优雅、高效、可维护。

---

一、模块与包（Module & Package）

1. 模块（Module）

• 每个 .py 文件就是一个模块，模块名即文件名（不含 .py）。
• 通过 import 导入模块，使用其函数、类、变量。

# 文件 math_utils.py
def add(a, b):
    return a + b

PI = 3.14159

# 另一个文件 main.py
import math_utils
print(math_utils.add(3, 5))          # 8
print(math_utils.PI)                 # 3.14159

# 或者
from math_utils import add, PI
print(add(3, 5))

# 或者别名
import math_utils as mu

2. 包（Package）

• 包是包含多个模块的目录，目录下必须有 __init__.py 文件（可以为空，表示该目录是一个包）。
• 包允许层级命名空间。

my_package/
    __init__.py
    math_utils.py
    string_utils.py
    sub_package/
        __init__.py
        another.py

导入方式：

import my_package.math_utils
from my_package import string_utils
from my_package.sub_package import another

3. 常用导入形式及特点

导入形式	优点	缺点
import module	命名空间清晰，避免冲突	访问时需带模块名
from module import func	调用方便	可能覆盖同名变量
from module import *	极不推荐，污染命名空间	难以追踪来源
import module as alias	简化长模块名	-

4. 模块搜索路径

• 当前脚本所在目录
• PYTHONPATH 环境变量中的目录
• Python 标准库目录
• site-packages（第三方库）

可通过 sys.path 查看和修改搜索路径。

5. 常用标准库模块

• sys：系统相关参数、路径
• os：操作系统接口
• json：JSON 处理
• re：正则表达式
• collections：额外数据结构（defaultdict, Counter, deque）
• itertools：高效的迭代工具
• functools：高阶函数工具（partial, wraps, lru_cache）
• random：随机数
• datetime：日期时间

---

二、迭代器与生成器（Iterator & Generator）

1. 迭代器（Iterator）

• 实现了 __iter__() 和 __next__() 方法的对象。
• 可被 for 循环遍历，也能用 next() 手动获取下一个元素。
• 迭代器只能前进一次，无法重置。

# 手动实现一个迭代器
class CountDown:
    def __init__(self, start):
        self.current = start

    def __iter__(self):
        return self

    def __next__(self):
        if self.current <= 0:
            raise StopIteration
        val = self.current
        self.current -= 1
        return val

for num in CountDown(5):
    print(num)   # 5 4 3 2 1

2. 生成器（Generator）

• 生成器函数：使用 yield 关键字的函数，调用时返回生成器对象（一种特殊的迭代器）。
• 生成器表达式：类似列表推导式，但用圆括号 (expr for item in iterable)。
• 惰性求值：仅在需要时生成下一个值，节省内存（尤其适合处理大文件或无限序列）。

# 生成器函数：斐波那契数列
def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

for val in fibonacci(10):
    print(val, end=' ')   # 0 1 1 2 3 5 8 13 21 34

# 生成器表达式：计算平方和，不立即生成列表
squares_gen = (x**2 for x in range(1000000))  # 几乎不占内存
print(sum(squares_gen))                       # 累加时逐个生成

3. 生成器的优势场景

• 读取超大文件（例如几个GB的日志），逐行 yield 处理。
• 产生无限序列（如随机数、时间戳）。
• 实现协程（yield 也可接收值，进阶内容）。

# 逐行读取大文件（经典用法）
def read_large_file(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        for line in f:
            yield line.strip()

for line in read_large_file('huge.log'):
    process(line)   # 逐行处理，内存占用恒定

---

三、装饰器（Decorator）

1. 什么是装饰器

• 一种高阶函数，接收一个函数作为参数，返回一个新函数（通常增强原函数功能）。
• 语法糖 @decorator 让代码更简洁。
• 本质：func = decorator(func)

2. 最简单的装饰器（无参数）

def timer(func):
    import time
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 耗时 {end-start:.4f} 秒")
        return result
    return wrapper

@timer
def slow_function():
    import time
    time.sleep(1)
    return "Done"

slow_function()   # 输出: slow_function 耗时 1.0023 秒

3. 保留原函数元数据：functools.wraps

装饰器会覆盖原函数的 __name__、__doc__ 等属性，用 wraps 修复。

from functools import wraps

def logger(func):
    @wraps(func)          # 保持原函数信息
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@logger
def add(a, b):
    """求和函数"""
    return a + b

print(add.__name__)   # add (而不是 wrapper)
print(add.__doc__)    # 求和函数

4. 带参数的装饰器

需要三层嵌套：最外层接收装饰器参数，内层接收函数，再内层接收函数参数。

def repeat(times):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(3)
def greet(name):
    print(f"Hello {name}")

greet("Alice")   # 打印三次

5. 类装饰器（了解即可）

使用类的 __call__ 方法实现装饰器。

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"调用 {self.func.__name__} 次数: {self.count}")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello():
    print("Hello")

say_hello()  # 次数1
say_hello()  # 次数2

6. 常见应用场景

• 计时（性能分析）
• 日志记录（自动打印函数调用信息）
• 权限校验（如用户登录检查）
• 缓存（functools.lru_cache）
• 重试机制（网络请求失败自动重试）
• 事务管理（数据库操作自动提交/回滚）

---

四、实战练习题（你要求的“特定场景”）

下面为你设计 3 个实战练习题，分别练习模块/包、生成器、装饰器，并综合运用。

练习1：批量处理文件的命令行工具（模块与包 + 文件操作）

描述：编写一个 Python 包 filetools，包含两个模块：

• counter.py：提供函数 count_words(filename)，统计一个文本文件中的单词总数（按空格拆分，忽略大小写）。
• converter.py：提供函数 convert_to_uppercase(filename, output_filename)，将文件内容全部转为大写并写入新文件。

然后在包外写一个脚本 batch_cli.py，使用 argparse（标准库）或简单的 input 交互，实现以下功能：

• 用户输入一个目录路径，程序递归查找该目录下所有 .txt 文件。
• 对每个 .txt 文件，输出其单词总数。
• 询问用户是否转换为大写，若是，在文件名前加 UPPER_ 并保存。

考察点：包的创建、模块导入、递归文件遍历、argparse 基础、文件读写。

练习2：超大日志分析工具（生成器）

描述：有一个服务器日志文件 access.log，每行格式如下：

2025-01-15 10:30:45 192.168.1.1 GET /index.html 200
2025-01-15 10:31:02 192.168.1.2 POST /api/login 404
...

假设文件大小可能超过内存（几GB）。要求：

1. 编写一个生成器 read_log_lines(file_path)，逐行 yield 日志行。
2. 编写另一个生成器 filter_by_status(log_lines, status_code)，接收日志行生成器，yield 出状态码等于 status_code 的行。
3. 编写一个函数 count_ips(status_code)，使用上面两个生成器，统计特定状态码下每个 IP 的出现次数，返回字典。
4. 测试：统计 404 状态码的 IP 前 10 名。

考察点：生成器惰性求值、生成器链式处理、内存友好。

练习3：带缓存和超时的 API 调用装饰器

描述：模拟一个需要调用远程 API 的函数 fetch_user_info(user_id)（可先用 time.sleep() 模拟延迟，随机返回 JSON 数据）。这个函数调用很昂贵（每次 0.5 秒）。

要求编写一个装饰器 @cache_with_ttl(ttl_seconds=60)，实现以下功能：

• 对于相同的 user_id，在 ttl_seconds 内直接返回缓存结果，不实际调用函数。
• 超过 TTL 后，重新调用函数并更新缓存。
• 支持任意参数和关键字参数（不仅 user_id）。
• 用 functools.wraps 保留原函数元数据。

考察点：带参数的装饰器、闭包、字典缓存、时间判断、*args, **kwargs。

附加练习：命令行彩蛋

用装饰器实现一个 @retry(max_attempts=3, delay=1)，装饰一个可能失败的函数（比如随机抛出异常）。当函数失败时自动重试，达到最大次数后抛出最后一次异常。这是实际项目中常见的装饰器应用。

---

五、学习路径建议

1. 模块与包：练习将自己写的多个 .py 文件组织成包，并在不同目录下导入。
2. 生成器：找一个大文件（比如几万行 CSV），练习逐行处理而不是一次加载到内存。
3. 装饰器：先写无参数的装饰器（计时、日志），再尝试带参数的装饰器，最后阅读 functools.lru_cache 源码。

如果你对某个练习题想先看答案或需要提示，可以随时问我。完成这些练习后，你就能自信地阅读和编写高级 Python 代码了。

下一步可以学习：并发编程（threading/asyncio）、上下文管理器（with 自定义）、元类（了解即可）等。 🚀

🚀

很好！你已经进入了 Python 高级进阶阶段。下面我将分三个部分为你详细讲解：并发编程（threading / asyncio）、上下文管理器（自定义 with）、元类（了解即可）。每个部分都有概念解释、代码示例和最佳实践。

---

一、并发编程（Concurrent Programming）

并发可以让程序同时执行多个任务，提高效率。Python 主要提供两种方案：多线程（threading） 和 异步 I/O（asyncio）。另外还有多进程（multiprocessing），但这里暂不展开。

1. threading（多线程）—— 适合 I/O 密集型任务

• 原理：在单个进程中创建多个线程，共享全局变量。由于 GIL（全局解释器锁） 的存在，Python 的多线程无法并行计算，但可以在 I/O 等待时切换线程，适合网络请求、文件读写等 I/O 密集型任务。
• 常用模块：threading
• 注意：线程间共享数据需要加锁（Lock）避免竞争条件。

基础用法

import threading
import time

def worker(name, delay):
    for i in range(3):
        print(f"线程 {name} 执行第 {i} 次")
        time.sleep(delay)

# 创建线程
t1 = threading.Thread(target=worker, args=("A", 1))
t2 = threading.Thread(target=worker, args=("B", 1.5))

# 启动线程
t1.start()
t2.start()

# 等待线程结束
t1.join()
t2.join()
print("所有线程完成")

使用锁保护共享数据

counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:          # 自动获取和释放锁
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
print(counter)   # 应该等于 1,000,000

线程池（concurrent.futures.ThreadPoolExecutor）—— 更高级接口

from concurrent.futures import ThreadPoolExecutor
import requests

def fetch_url(url):
    return requests.get(url).status_code

urls = ["https://python.org", "https://google.com", "https://github.com"]
with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(fetch_url, urls)
    for status in results:
        print(status)

2. asyncio（异步 I/O）—— 单线程并发，适合高并发 I/O

• 原理：通过事件循环（event loop）和协程（coroutine）实现单线程异步。遇到 I/O 操作时“让出”控制权，事件循环去执行其他协程，I/O 完成后恢复执行。无需锁，但需要所有 I/O 操作为异步库（如 aiohttp 代替 requests）。
• 关键词：async def、await、asyncio.run()、asyncio.create_task()
• 优势：比线程更轻量，能支持数万并发连接。

基础示例

import asyncio

async def say_hello(delay, name):
    await asyncio.sleep(delay)   # 模拟异步 I/O
    print(f"Hello, {name}")

async def main():
    task1 = asyncio.create_task(say_hello(2, "Alice"))
    task2 = asyncio.create_task(say_hello(1, "Bob"))
    await task1
    await task2

asyncio.run(main())   # 1秒后打印 Bob，2秒后打印 Alice

异步 HTTP 请求（需要安装 aiohttp）

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return response.status

async def main():
    urls = ["https://python.org", "https://google.com"]
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        print(results)

asyncio.run(main())

什么时候用 threading，什么时候用 asyncio？

场景	推荐方案
大量并发网络请求（爬虫、Web API 调用）	asyncio + aiohttp
少量长时间 I/O 操作（如文件读写、数据库查询）	threading 或 ThreadPoolExecutor
CPU 密集型计算（如图像处理、数值计算）	多进程（multiprocessing）

---

二、上下文管理器（Context Manager）与自定义 with

上下文管理器用于自动管理资源（如文件、锁、数据库连接），确保进入和退出时执行特定操作。最常用的就是 with open(...) as f。

1. 实现上下文管理器的两种方式

方式一：使用类，实现 __enter__ 和 __exit__ 方法

class ManagedFile:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        # 返回 True 表示异常已处理，不会向外抛出；默认返回 None (即 False)
        return False

# 使用
with ManagedFile("test.txt", "w") as f:
    f.write("Hello, context manager!")

方式二：使用 contextlib.contextmanager 装饰器（生成器风格，更简洁）

from contextlib import contextmanager

@contextmanager
def managed_file(filename, mode):
    f = open(filename, mode)
    try:
        yield f          # 将文件对象提供给 with 语句体
    finally:
        f.close()        # 无论是否异常，都关闭文件

with managed_file("test.txt", "r") as f:
    content = f.read()
    print(content)

2. 其他常见上下文管理器

• threading.Lock：with lock: ... 自动获取和释放锁。
• decimal.localcontext()：临时改变精度。
• pandas.option_context()：临时修改显示选项。

3. 同时管理多个资源

with open("in.txt", "r") as f_in, open("out.txt", "w") as f_out:
    f_out.write(f_in.read())

4. 自定义上下文管理器的应用场景

• 计时器（自动记录代码块耗时）
• 数据库事务（进入时开始事务，退出时提交或回滚）
• 改变标准输出（临时重定向 print）
• 临时修改环境变量

计时器示例：

import time
from contextlib import contextmanager

@contextmanager
def timer(name):
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"[{name}] 耗时 {elapsed:.4f} 秒")

with timer("sleep"):
    time.sleep(1.2)
# 输出：[sleep] 耗时 1.2002 秒

---

三、元类（Metaclass）—— 了解即可

元类是 Python 中最深奥的特性之一，它用来创建类。简单说：类是实例的模板，而元类是类的模板。绝大多数情况下你不需要自己编写元类，但理解其机制有助于阅读框架源码（如 Django、SQLAlchemy）。

1. 基础概念

• type 是 Python 的内置元类，所有类（包括 object）都是由 type 创建的。
• 可以使用 class MyClass(metaclass=some_metaclass) 指定元类。

# 普通定义
class Foo:
    pass

# 等价于手动调用 type
Foo = type('Foo', (), {})

# type 的三个参数：类名、父类元组、属性字典

2. 自定义元类（通常继承 type）

class MyMeta(type):
    def __new__(cls, name, bases, dct):
        # 在类创建之前修改或添加属性
        dct['version'] = 1.0
        return super().__new__(cls, name, bases, dct)

    def __call__(cls, *args, **kwargs):
        # 控制类的实例化过程
        print(f"Creating instance of {cls.__name__}")
        return super().__call__(*args, **kwargs)

class MyClass(metaclass=MyMeta):
    pass

print(MyClass.version)   # 1.0
obj = MyClass()          # 打印 "Creating instance of MyClass"

3. 实际应用场景

• 框架中的 ORM：根据类定义自动生成数据库表映射。
• 单例模式：用元类确保一个类只有一个实例。
• 自动注册子类：当定义子类时自动将其加入某个列表。

单例元类示例：

class SingletonMeta(type):
    _instances = {}
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Database(metaclass=SingletonMeta):
    def __init__(self):
        print("初始化数据库连接")

db1 = Database()   # 打印一次
db2 = Database()
print(db1 is db2)  # True

4. 为什么说“了解即可”？

• 元类容易导致代码难以理解和调试。
• 绝大多数问题可以用更简单的方式（类装饰器、继承、类方法）解决。
• 只有在编写框架或库时才会需要，业务代码中尽量避免。

---

四、综合实战：并发 + 上下文管理器 + 装饰器

需求：编写一个装饰器 @async_timeit，用于异步函数（协程），打印执行耗时。同时，用自定义上下文管理器记录日志。

import asyncio
import time
from functools import wraps
from contextlib import contextmanager

# 1. 自定义上下文管理器：计时
@contextmanager
def timer_context(name):
    start = time.perf_counter()
    yield
    elapsed = time.perf_counter() - start
    print(f"[{name}] 耗时 {elapsed:.4f} 秒")

# 2. 装饰器：异步函数计时
def async_timeit(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        with timer_context(func.__name__):
            return await func(*args, **kwargs)
    return wrapper

# 3. 异步函数示例
@async_timeit
async def fetch_data():
    await asyncio.sleep(1)
    return "数据"

async def main():
    result = await fetch_data()
    print(result)

asyncio.run(main())
# 输出：
# [fetch_data] 耗时 1.0023 秒
# 数据

---

五、学习建议与总结

主题	学习优先级	核心要点
并发编程（threading）	重要	GIL、I/O 密集型、锁、线程池
并发编程（asyncio）	重要	协程、事件循环、异步 I/O
上下文管理器	必备	with、__enter__/__exit__、contextmanager 装饰器
元类	了解即可	type、控制类创建、框架底层

下一步：

• 多进程：multiprocessing 绕过 GIL 进行 CPU 密集型计算。
• 类型注解：typing 模块提高代码可维护性。
• 测试：unittest / pytest 编写单元测试。
• 打包发布：setuptools 将自己的模块发布到 PyPI。

可以针对并发编程或上下文管理器设计几个实战练习题（例如：异步爬虫、线程安全的日志记录器、自动重试的上下文管理器等）