基础篇 / 第 8 章

第 8 章 · 异常处理与文件操作

异常是什么：错误的两种流派

程序中的错误分两大类，处理方式截然不同：

# error_basics.py —— 错误是如何产生的
# 1) 语法错误：漏写冒号（写在注释里以免中断脚本）
# def broken()
#     pass

# 2) 运行时错误
print("开始")
result = 10 / 0        # ZeroDivisionError: division by zero
print("永远不会执行到")   # 程序在这里就终止了

如果让异常「自由传播」，最外层会打印一大段 traceback（回溯）并以非零状态码退出。对交互式脚本这没问题，但对 Web 服务、批处理任务，必须捕获、记录、降级。

Python 的异常体系根植于一个清晰的三分：

# 异常体系速览（省略部分中间层）
BaseException
├── KeyboardInterrupt        # Ctrl+C，用户中断
├── SystemExit               # sys.exit()
├── GeneratorExit            # 生成器被关闭
└── Exception                 # 普通可捕获异常的根
    ├── ArithmeticError
    │   ├── ZeroDivisionError
    │   ├── OverflowError
    │   └── FloatingPointError
    ├── LookupError
    │   ├── IndexError       # 列表越界
    │   └── KeyError         # 字典键不存在
    ├── AttributeError       # 属性不存在
    ├── ValueError           # 值类型对但值不合理
    ├── TypeError            # 操作/类型不匹配
    ├── OSError              # 操作系统错误（文件、网络）
    │   ├── FileNotFoundError
    │   ├── PermissionError
    │   └── IsADirectoryError
    ├── NameError             # 名字未定义
    ├── ImportError
    ├── RecursionError
    └── ...（还有约 60 个内置异常）

try:
    1 / 0
except Exception as e:      # Exception 捕获绝大多数「可恢复」错误
    print("ok")
# 注意：KeyboardInterrupt 和 SystemExit 不属于 Exception，
# 因此 except Exception 不会误捕获用户的 Ctrl+C。

try/except/else/finally 完整语义

Python 的异常处理有四个子句，不只是两个，执行顺序有严格规定：

# try_full.py —— 四子句的完整演示
import sys


def read_config(path):
    """演示四子句的分工：try 只放可能失败的那一行。"""
    try:
        handle = open(path, encoding="utf-8")   # 只有「打开」可能失败
    except FileNotFoundError:
        print(f"  [except] 文件不存在：{path}，使用默认值")
        return {"debug": False}
    except PermissionError as e:
        print(f"  [except] 权限不足：{e}")
        raise RuntimeError(f"无法读取配置 {path}") from e   # 异常链，保留因果
    except OSError as e:
        print(f"  [except] 其他 OS 错误：{e}")
        raise
    else:
        # 只有 try 完全成功才会走这里；解析放在这里，解析失败不会被上面的 OS 异常误捕
        try:
            with handle:
                content = handle.read()
        except UnicodeDecodeError as e:
            print(f"  [内层] 编码错误：{e}")
            return {"debug": False}
        print("  [else] 读取成功，进入解析阶段")
        return parse(content)
    finally:
        print("  [finally] 清理动作总会执行")


def parse(text):
    data = {}
    for line in text.splitlines():
        if "=" in line:
            k, _, v = line.partition("=")
            data[k.strip()] = v.strip()
    return data


print("场景 1：正常路径")
print(" 结果：", read_config("config_ok.txt"))

print("\n场景 2：文件不存在")
print(" 结果：", read_config("not_exist.txt"))

print("\n--- finally 在 return 之前也会执行 ---")
def f():
    try:
        return "try 的返回值"
    finally:
        print("  finally 先打印，然后才真正 return")


print(" 结果：", f())

print("\n--- else 的价值 ---")
def parse_number(s):
    """else 分支让「转换」不污染 try 的异常捕获范围。"""
    try:
        n = int(s)          # 可能抛 ValueError
    except ValueError:
        return None
    else:
        print(f"  成功解析：{n}")
        return n
    finally:
        print(f"  处理完毕：{s!r}")


print(parse_number("42"))
print(parse_number("abc"))

sys.stdout.flush()

捕获异常的几种写法

# catch_patterns.py —— 捕获方式对比
def demo():
    # 1) 捕获后忽略：慎用，吞掉信息等于埋雷
    try:
        int("abc")
    except ValueError:
        pass

    # 2) 捕获并转成有意义的返回值（最常用）
    try:
        return int("abc")
    except ValueError:
        return 0

    # 3) 捕获多个类型：逗号分隔（写成一个元组）
    try:
        [][10]
    except (IndexError, KeyError) as e:
        print(f"  容器错误：{type(e).__name__}")

    # 4) 分级捕获：先具体，再宽泛
    try:
        {}["missing"]
    except IndexError:
        print("  索引错误")
    except KeyError:
        print("  键错误")

    # 5) 捕获并重新抛出（加上下文）
    try:
        int("abc")
    except ValueError as e:
        raise ValueError(f"字段 config.port 不是合法整数：{e}") from e

    # 6) 断言式用法：raise 主动抛错
    n = -1
    if n < 0:
        raise ValueError("n 不能为负数")

    # 7) 异常分组（Python 3.11+）：更干净的分支
    try:
        int("abc")
    except* ValueError as eg:
        print(f"  分组捕获到 {len(eg.exceptions)} 个 ValueError")


if __name__ == "__main__":
    import sys
    print("demo() 返回：", demo())
    # demo() 返回： 0
    try:
        demo()
    except ValueError as e:
        print("最外层收到：", e)
        print("原因链：", type(e.__cause__).__name__)

自定义异常与异常层级设计

内置异常不够表达业务语义时，应该定义自己的异常，并按业务层级组织成树，让调用方能用一个大类统一捕获、用小类精确判断。

# custom_exceptions.py —— 自定义异常的规范写法
class AppError(Exception):
    """本应用所有异常的根类。调用方可 except AppError 一网打尽。"""


class ValidationError(AppError):
    """参数校验失败。"""


class NotFoundError(AppError):
    """资源不存在。"""


class ConfigError(AppError):
    """配置相关错误。"""

    def __init__(self, key, reason, source=None):
        super().__init__(f"配置项 {key!r} 无效：{reason}")
        self.key = key
        self.reason = reason
        self.source = source     # 结构化字段，便于调用方自行处理


class ProcessingError(AppError):
    """处理流程失败，保留原始异常。"""

    def __init__(self, message, cause):
        super().__init__(message)
        self.cause = cause


# ---- 演示 ----
def load_config(data):
    """按约定解析配置，失败抛自定义异常。"""
    result = {}
    for line in data.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            raise ConfigError(line, "缺少 '=' 分隔符", source=data)
        key, _, raw = line.partition("=")
        key, raw = key.strip(), raw.strip()
        if raw.lower() in ("true", "false"):
            result[key] = raw.lower() == "true"
        elif raw.isdigit():
            result[key] = int(raw)
        else:
            result[key] = raw
    return result


def find_user(db, uid):
    if uid not in db:
        raise NotFoundError(f"用户 {uid} 不存在")
    return db[uid]


def process(uid, raw_config):
    try:
        cfg = load_config(raw_config)
        user = find_user({"1": "王小明", "2": "李小红"}, uid)
    except AppError as e:
        raise ProcessingError(f"处理用户 {uid} 失败", cause=e) from e
    return {"user": user, "config": cfg}


try:
    process("99", "debug=true\nport=8080")
except NotFoundError as e:
    print("精确捕获 NotFoundError：", e)
except ConfigError as e:
    print("精确捕获 ConfigError：", e, "| key =", e.key)
except ProcessingError as e:
    print("统一捕获 ProcessingError：", e)
    print("  内层原因：", e.cause, "（类型：", type(e.cause).__name__, "）")
except AppError as e:
    print("兜底捕获 AppError：", e)

print()
print("正常情况：", process("1", "debug=false\nport=8080\nname=服务A"))
# 预期输出：
# 统一捕获 ProcessingError： 处理用户 99 失败
#   内层原因： 用户 99 不存在 （类型： NotFoundError ）
#
# 正常情况： {'user': '王小明', 'config': {'debug': False, 'port': 8080, 'name': '服务A'}}

自定义异常的规范清单

断言与 raise

# assert_vs_raise.py —— assert 不是校验
def divide(a, b):
    assert b != 0, "除数不能为 0"       # ❌ 用于用户输入校验
    return a / b


def divide_safe(a, b):
    if b == 0:
        raise ValueError("除数不能为 0")   # ✅ 真正的校验
    return a / b


# assert 会在 python -O 优化模式下被完全移除！
# python -O script.py  -> divide(1, 0) 不报错，返回 inf 或抛 ZeroDivisionError
# 所以：assert 用于「开发者不变量」，raise 用于「运行时可预期的错误」

with 语句与上下文管理器

with 用来管理「需要成对获取与释放」的资源：文件、锁、数据库连接、临时目录。无论正常结束、抛异常还是 return，释放动作保证执行一次。

# with_why.py —— 为什么必须有 with
# 错误写法 1：忘记关闭，句柄泄漏
f = open("data.txt", encoding="utf-8")
content = f.read()
# f.close()   <-- 万一中间抛异常，这行不会执行

# 错误写法 2：虽然有 finally，但 close 不接受保护
f = open("data.txt", encoding="utf-8")
try:
    content = f.read()
finally:
    f.close()      # 如果 close() 本身抛异常，前一个异常会被掩盖

# 正确写法
with open("data.txt", encoding="utf-8") as f:
    content = f.read()
# 出了 with 块，f 自动关闭（即使块内抛异常）

__enter__ 与 __exit__ 的工作原理

# with_protocol.py —— 上下文管理器的两个魔术方法
class Resource:
    """演示上下文管理器协议。"""

    def __init__(self, name):
        self.name = name
        self.opened = False

    def __enter__(self):
        print(f"  → __enter__：打开 {self.name}")
        self.opened = True
        return self          # 返回值成为 as 后的名字

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"  ← __exit__：关闭 {self.name}（异常类型={exc_type}）")
        self.opened = False
        # 返回 True 会「吞掉」异常（不推荐）；返回 False/None 让异常继续传播
        return False

    def use(self):
        if not self.opened:
            raise RuntimeError("资源未打开")
        print(f"    使用 {self.name}")


print("场景 1：正常")
with Resource("数据库连接") as res:
    res.use()

print("\n场景 2：块内抛异常")
try:
    with Resource("文件句柄") as res:
        res.use()
        raise ValueError("处理过程中出错")
except ValueError as e:
    print("  异常仍然向外传播：", e)

print("\n场景 3：__exit__ 返回 True 会吞掉异常")
class SwallowError(Resource):
    def __exit__(self, *args):
        super().__exit__(*args)
        return True        # 抑制异常


with SwallowError("会吞错误的资源"):
    raise ValueError("这条异常不会传播出去")
print("  看，我还在这里，异常被 __exit__ 抑制了")

contextlib 提供的快捷写法

# contextlib.py —— 不必总手写 __enter__/__exit__
import contextlib
import io
import time


# 1) @contextmanager：把生成器包装成上下文管理器
@contextlib.contextmanager
def timer(label):
    """计时上下文管理器。"""
    start = time.perf_counter()
    try:
        yield f"{label} 开始"
    finally:
        elapsed = time.perf_counter() - start
        print(f"  [{label}] 耗时 {elapsed * 1000:.3f} ms")


with timer("数据处理") as info:
    print(" ", info)
    sum(range(100000))


# 2) @contextlib.suppress：故意忽略异常（比 try/except/pass 更清晰）
with contextlib.suppress(ZeroDivisionError, TypeError):
    print("  1 / 0 =", 1 / 0)     # 被抑制
    print("  这里不会执行")
print("  继续正常流程")


# 3) contextlib.redirect_stdout / redirect_stderr：临时重定向输出
buffer = io.StringIO()
with contextlib.redirect_stdout(buffer):
    print("这段输出被收进 buffer")
print("  捕获到的内容：", repr(buffer.getvalue().strip()))


# 4) contextlib.ExitStack：动态管理多个资源（适合数量不定的情况）
@contextlib.contextmanager
def resource(name):
    print(f"  打开 {name}")
    try:
        yield name
    finally:
        print(f"  关闭 {name}")


def process_many(names):
    # ExitStack 保证无论中途是否出错，所有已打开的资源都会正确关闭
    with contextlib.ExitStack() as stack:
        for n in names:
            stack.enter_context(resource(n))
        print("  全部就绪，开始处理")


process_many(["连接A", "连接B", "连接C"])


# 5) 多个资源用逗号分隔（括号可省略，但建议保留以便换行）
with open("f1.txt", "w", encoding="utf-8") as a, \
     open("f2.txt", "w", encoding="utf-8") as b:
    a.write("内容A\n")
    b.write("内容B\n")
print("  多资源 with 执行完毕")

文件打开：open() 的参数全景

open() 是 Python 内置函数，返回一个文本或二进制文件对象。

# open_signature.py —— open 的完整用法
# open(file, mode='r', buffering=-1, encoding=None, errors=None, newline=None, closefd=True)

# 常用模式表
MODES = {
    "r":  "只读（默认）。文件必须存在，否则 FileNotFoundError",
    "w":  "写入。先清空文件；文件不存在则创建",
    "a":  "追加。在文件末尾写入；不存在则创建",
    "x":  "独占创建。文件已存在则 FileExistsError（防覆盖的安全写法）",
    "b":  "二进制模式（可与上面组合，如 'rb'、'wb'）",
    "+":  "读写模式（可组合，如 'r+'、'w+'）",
}
for k, v in MODES.items():
    print(f"  {k:3s} -> {v}")

# 文本 vs 二进制
text = open("demo.txt", "w", encoding="utf-8")
text.write("hello")
text.close()

with open("demo.txt", "r", encoding="utf-8") as f:
    print("\n文本模式读到：", repr(f.read()), type(f.read()))

with open("demo.txt", "rb") as f:
    data = f.read()
    print("二进制模式读到：", data, type(data))
    # 预期输出：b'hello' <class 'bytes'>

# r+ / w+ / a+ 的区别
with open("demo.txt", "r+", encoding="utf-8") as f:
    print("r+ 内容：", f.read())          # 读，不清空

with open("demo.txt", "a+", encoding="utf-8") as f:
    f.write(" world")
    f.seek(0)                            # 追加模式指针在末尾，读之前要归零
    print("a+ 内容：", f.read())          # hello world

编码、换行与 Unicode

文本文件本质是「字节序列 + 编码规则」。读错编码就会看到乱码或抛出 UnicodeDecodeError。

# encoding_demo.py —— 编码与换行
import codecs

# 1) 为什么一定要显式写 encoding
#    不同系统的默认编码不同（Windows 中文版 cp936，Linux 通常 utf-8），
#    不写 encoding 就是「碰运气」，跨机器就会炸。
with open("utf8_demo.txt", "w", encoding="utf-8") as f:
    f.write("中文内容\n第二行\n")

with open("utf8_demo.txt", "r", encoding="utf-8") as f:
    print("utf-8 读取：", f.read().splitlines())

# 2) 读错编码的后果
raw = open("utf8_demo.txt", "rb").read()
print("原始字节：", raw[:8])
try:
    raw.decode("gbk")
except UnicodeDecodeError as e:
    print("用 gbk 解码失败：", type(e).__name__)
print("用 gbk 容错解码：", raw.decode("gbk", errors="replace")[:8])

# 3) errors 参数的取值
text = "hello"
for err in ("strict", "ignore", "replace", "backslashreplace"):
    try:
        b = text.encode("ascii")
        r = b.decode("ascii", errors=err)
    except UnicodeEncodeError:
        r = "（无法用 ascii 编码）"
    print(f"  errors={err:16s} -> {r}")

print("\n中文测试：", "中文".encode("utf-8"))
for err in ("strict", "ignore", "replace"):
    try:
        print(f"  utf-8 bytes 用 {err:8s} 解码 ->",
              "中文".encode("utf-8").decode("utf-8", errors=err))
    except UnicodeError as e:
        print(f"  {err} 失败：", e)

# 4) newline 参数：控制换行符的统一
#    newline=None（默认）读写都做「通用换行符」转换
#    newline='\n'  读时不转换，写时用 \n
#    newline=''    读时保留原始换行符（csv 模块推荐）
for nl in (None, "", "\n", "\r\n"):
    with open("nl_test.txt", "w", encoding="utf-8", newline=nl) as f:
        f.write("a\nb\n")
    with open("nl_test.txt", "rb") as f:
        raw = f.read()
    with open("nl_test.txt", "r", encoding="utf-8", newline=nl) as f:
        lines = f.readlines()
    print(f"  newline={str(nl):6s} 落盘字节={raw} 读出={lines}")

# 预期输出（关键行）：
#   newline=None   落盘字节=b'a\nb\n'       读出=['a\n', 'b\n']
#   newline='\r\n' 落盘字节=b'a\r\nb\r\n'   读出=['a\r\n', 'b\r\n']

# 5) 判断文件编码（BOM 探测）
for path in ("utf8_demo.txt",):
    with open(path, "rb") as f:
        head = f.read(4)
    detected = "utf-8-sig（有BOM）" if codecs.BOM_UTF8 in head else "utf-8（无BOM）"
    print(f"\n{path} 开头字节 {head} -> {detected}")

文件读取：五种方式

# read_methods.py —— 读取方式对比
# 准备测试文件
with open("sample.txt", "w", encoding="utf-8") as f:
    f.write("第一行内容\n第二行内容\n第三行内容\n\n第五行内容\n")

print("文件大小：", __import__("os").path.getsize("sample.txt"), "字节\n")

# 方式 1：read() 一次性全部读入字符串
with open("sample.txt", "r", encoding="utf-8") as f:
    data = f.read()
print("read() 长度：", len(data), "| 字符数（不含换行）：", len(data.splitlines()))
print("前 20 字：", repr(data[:20]))

# 方式 2：readline() 读一行（文件末尾不自动加换行）
with open("sample.txt", "r", encoding="utf-8") as f:
    print("\nreadline() 三次：")
    for i in range(3):
        print(f"  第{i+1}行：", repr(f.readline()))

# 方式 3：readlines() 读所有行成 list（每行保留 \n）
with open("sample.txt", "r", encoding="utf-8") as f:
    lines = f.readlines()
print("\nreadlines()：", lines)

# 方式 4：for 逐行迭代 —— 最推荐，内存友好且写法简洁
with open("sample.txt", "r", encoding="utf-8") as f:
    print("\nfor 逐行（自动 strip）：")
    for i, line in enumerate(f, start=1):
        print(f"  {i}: {line.strip()!r}")

# 方式 5：read(n) / readline(size) 分块读取
with open("sample.txt", "r", encoding="utf-8") as f:
    print("\n分块读取（每次 6 个字符）：")
    chunks = []
    while True:
        chunk = f.read(6)
        if not chunk:      # 空字符串表示读到文件末尾
            break
        chunks.append(chunk)
    print("  ", chunks)

# 方式 6：跳过空行与注释
with open("sample.txt", "r", encoding="utf-8") as f:
    content = [ln.strip() for ln in f if ln.strip()]
print("\n非空行：", content)

# 方式 7：一次读入字典（.read().splitlines()）
with open("sample.txt", encoding="utf-8") as f:
    print("行列表：", f.read().splitlines())

文件写入

# write_methods.py —— 写入方式与缓冲
import os

# 1) 一次性写入（w 模式会先清空文件！）
with open("out1.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")
    f.writelines(["第三行\n", "第四行\n"])     # writelines 接可迭代对象

with open("out1.txt", encoding="utf-8") as f:
    print("out1 内容：", f.read().splitlines())

# 2) print(..., file=f) —— 让 print 写进文件
with open("out2.txt", "w", encoding="utf-8") as f:
    for i in range(1, 4):
        print(f"第 {i} 行", file=f)          # 少了 file= 就写到屏幕上了

with open("out2.txt", encoding="utf-8") as f:
    print("out2 内容：", f.read().splitlines())

# 3) 追加模式 a
with open("out1.txt", "a", encoding="utf-8") as f:
    f.write("追加的行\n")

with open("out1.txt", encoding="utf-8") as f:
    print("\n追加后 out1：", f.read().splitlines())

# 4) 定位写入 r+ / seek / tell
with open("out1.txt", "r+", encoding="utf-8") as f:
    print("当前位置 tell()：", f.tell())     # 0
    f.seek(0, os.SEEK_END)                    # 移到文件末尾
    print("移到末尾 tell()：", f.tell())
    f.seek(0, os.SEEK_SET)                    # 回到开头（SEEK_SET=0）
    print("回到开头 tell()：", f.tell())
    f.write("!!!")                            # 覆盖前 3 个字节
    f.seek(0)
    print("覆盖后前 20 字：", repr(f.read(20)))

# 5) 安全创建：x 模式（文件已存在就报错，避免误覆盖）
try:
    with open("out1.txt", "x", encoding="utf-8") as f:
        f.write("不该写入")
except FileExistsError:
    print("\nx 模式拦截了覆盖操作：文件已存在")

# 6) 缓冲与 flush
with open("buffered.txt", "w", encoding="utf-8", buffering=1) as f:  # buffering=1 行缓冲
    f.write("第一行\n")
    print("行缓冲下，flush 前文件大小：", os.path.getsize("buffered.txt"))
    f.write("第二行\n")
    print("写入两行后：", os.path.getsize("buffered.txt"))
    f.flush()      # 强制把缓冲区写入磁盘

print("\n最终 out1 内容：")
with open("out1.txt", encoding="utf-8") as f:
    for line in f:
        print("  ", line.rstrip())

路径处理：pathlib 与 os.path

硬编码 "data/users.txt" 在不同系统会指向不同位置。Python 用 pathlib 提供面向对象的、跨平台的路径方案。

# pathlib_demo.py —— 推荐的路径处理方式
import os
from pathlib import Path

# 1) 构造路径：/ 运算符拼接（不是除法！）
base = Path("/home/user/project") / "data"
cfg = base / "config.ini"
print("路径：", cfg)
print("父目录：", cfg.parent)
print("文件名：", cfg.name, "| 不含扩展名：", cfg.stem, "| 后缀：", cfg.suffix)
print("全部父级：", list(cfg.parents))
print("是绝对路径？", cfg.is_absolute(), "| 是否存在？", cfg.exists(), "| 是文件？", cfg.is_file())

# 2) ~ 展开、相对路径解析
print("家目录：", Path.home())
print("当前目录：", Path.cwd())

# 3) 遍历与查找
proj = Path.cwd()
for p in sorted(proj.glob("*.html")):
    print("  html:", p.name)

for p in proj.glob("chapter-0*.html"):
    print("  chapter:", p.name)

# 递归查找（rglob）
for p in list(proj.rglob("assets/*"))[:5]:
    print("  assets:", p.relative_to(proj))

# 4) 创建目录
outdir = proj / "output" / "logs"
outdir.mkdir(parents=True, exist_ok=True)   # parents：连父目录一起建；exist_ok：已存在不报错
print("目录已创建：", outdir.exists())

sample = outdir / "demo.log"
sample.write_text("第一行\n第二行\n", encoding="utf-8")     # 一行写文件
print("write_text 后大小：", sample.stat().st_size, "字节")

with sample.open("a", encoding="utf-8") as f:
    f.write("第三行\n")
print("追加后内容：", sample.read_text(encoding="utf-8").splitlines())

# 5) 重命名、移动、删除
target = outdir / "demo_v2.log"
sample.rename(target)                       # 改名/移动
print("rename 后：", target.name, target.exists())

target.unlink()                              # 删除单个文件
print("unlink 后存在？", target.exists())
outdir.rmdir()                               # 删除空目录
print("rmdir 后存在？", outdir.exists())

# 6) 临时目录的上下文管理
import tempfile

with tempfile.TemporaryDirectory() as tmp:
    p = Path(tmp) / "cache.bin"
    p.write_bytes(b"\x00\x01\x02")
    print("临时文件：", p.name, "| 存在？", p.exists(), "| 大小：", p.stat().st_size)
# 出了 with，临时目录及其内容已被自动删除

# 7) 老式 os.path 对照
print("\n--- os.path 对照 ---")
print(os.path.join("a", "b", "c.txt"))
print(os.path.exists("chapter-06.html"), os.path.getsize("chapter-06.html"))
print(os.path.splitext("data.tar.gz"))       # ('data.tar', '.gz')
print(os.path.basename("/a/b/c.txt"), os.path.dirname("/a/b/c.txt"))

# 新旧对照表
rows = [
    ("os.path.join(a, b)",     "Path(a) / b"),
    ("os.path.basename(p)",    "Path(p).name"),
    ("os.path.dirname(p)",     "Path(p).parent"),
    ("os.path.splitext(p)",    "Path(p).stem, Path(p).suffix"),
    ("os.path.exists(p)",      "Path(p).exists()"),
    ("os.path.isdir(p)",       "Path(p).is_dir()"),
    ("os.path.abspath(p)",     "Path(p).resolve()"),
]
print()
for old, new in rows:
    print(f"  {old:28s} -> {new}")

# 预期输出示例：
#   路径： /home/user/project/data/config.ini
#   父目录： /home/user/project/data
#   文件名： config.ini | 不含扩展名： config | 后缀： .ini

实战：CSV 与 JSON 文件

JSON

# json_demo.py —— JSON 读写
import json
from pathlib import Path

# 1) 写：ensure_ascii=False 让中文不被转成 \uXXXX
config = {
    "服务名": "订单中心",
    "端口": 8080,
    "调试": False,
    "数据库": {"主机": "127.0.0.1", "名": "orders", "连接池": 10},
    "允许的来源": ["https://a.com", "https://b.com"],
}

Path("config.json").write_text(
    json.dumps(config, ensure_ascii=False, indent=2, sort_keys=True),
    encoding="utf-8",
)
print("写出的 JSON：")
print(Path("config.json").read_text(encoding="utf-8"))

# 2) 读
data = json.loads(Path("config.json").read_text(encoding="utf-8"))
print("\n读回：", data["服务名"], "| 端口：", data["数据库"]["主机"], type(data["数据库"]["主机"]))

# 3) 文件对象直读直写
with open("config.json", encoding="utf-8") as f:
    obj = json.load(f)
with open("config_copy.json", "w", encoding="utf-8") as f:
    json.dump(obj, f, ensure_ascii=False, indent=2)
print("json.dump 写出的字节数：", Path("config_copy.json").stat().st_size)

# 4) 容错解析：允许注释与尾逗号（应对人写的配置文件）
text = """
{
  // 数据库配置
  "host": "localhost",   /* 行内注释 */
  "port": 5432,
}
"""
try:
    json.loads(text)
except json.JSONDecodeError as e:
    print("\n严格解析失败：", e)

loose = json.loads(text, object_pairs_hook=lambda pairs: dict(pairs))
print("（object_pairs_hook 只能处理重复键，注释仍不支持）->", loose)

# 5) 类型转换陷阱：JSON 里没有 datetime，所有日期都是字符串
raw = '{"created": "2026-01-15T10:30:00"}'
parsed = json.loads(raw)
from datetime import datetime
dt = datetime.fromisoformat(parsed["created"])     # 手动转换
print("\n解析出时间：", dt, type(dt).__name__)

# 6) dumps 的常用参数
print("\n紧凑格式：", json.dumps({"a": 1, "b": [1, 2]}, separators=(",", ":")))
print("排序+中文：", json.dumps({"b": 1, "a": "中"}, ensure_ascii=False, sort_keys=True))
print("自定义默认类型：",
      json.dumps({"d": datetime(2026, 1, 1)}, default=str))

CSV

# csv_demo.py —— CSV 读写
import csv
from pathlib import Path

rows = [
    {"订单号": "SO-001", "客户": "王小明", "金额": 199.90, "状态": "已发货"},
    {"订单号": "SO-002", "客户": "李小红", "金额": 89.00,  "状态": "待付款"},
    {"订单号": "SO-003", "客户": "张三丰", "金额": 1299.50, "状态": "已完成"},
]

# 1) 写（DictWriter 更可读，字段顺序显式）
fieldnames = ["订单号", "客户", "金额", "状态"]
with open("orders.csv", "w", encoding="utf-8-sig", newline="") as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()               # 写表头
    writer.writerows(rows)             # 批量写

print("写出的 CSV（utf-8-sig 便于 Excel 打开中文）：")
print(Path("orders.csv").read_text(encoding="utf-8-sig"))

# 2) 读（DictReader）
with open("orders.csv", "r", encoding="utf-8-sig", newline="") as f:
    reader = csv.DictReader(f)
    print("字段名：", reader.fieldnames)
    for row in reader:
        print(f"  {row['订单号']} | {row['客户']} | 金额={float(row['金额']):.2f} | {row['状态']}")

# 3) 处理缺失字段与多余列
with open("orders.csv", encoding="utf-8-sig", newline="") as f:
    reader = csv.DictReader(f, restkey="额外列", restval="(空)")
    for i, row in enumerate(reader):
        if row.get("额外列") or row.get("备注") is None:
            print(f"  行 {i} 异常：额外列={row['额外列']} 缺失备注={row['备注']}")

# 4) 统计：汇总金额
total = 0.0
with open("orders.csv", encoding="utf-8-sig", newline="") as f:
    for row in csv.DictReader(f):
        if row["状态"] != "待付款":
            total += float(row["金额"])
print("\n已发货+已完成的总金额：", round(total, 2))

# 5) 把列表写成 CSV 字符串（配合 StringIO / 网络上传）
import io
buf = io.StringIO()
writer = csv.DictWriter(buf, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
csv_text = buf.getvalue()
print("\n内存中的 CSV 文本：")
print(csv_text)

# 6) 处理包含分隔符/换行/引号的字段
tricky = [{"name": 'Smith, John', "note": "包含,逗号和\"引号\"", "desc": "第一行\n第二行"}]
buf2 = io.StringIO()
w = csv.DictWriter(buf2, fieldnames=["name", "note", "desc"])
w.writeheader()
w.writerows(tricky)
back = list(csv.DictReader(io.StringIO(buf2.getvalue())))
print("含特殊字符的字段往返一致？", back == tricky)
print("原样还原：", back[0])

# 7) TSV（制表符分隔）
with open("orders.tsv", "w", encoding="utf-8", newline="") as f:
    w = csv.DictWriter(f, fieldnames=fieldnames, delimiter="\t")
    w.writeheader()
    w.writerows(rows[:1])
print("\nTSV 首行：", Path("orders.tsv").read_text(encoding="utf-8").splitlines()[0])

处理大文件与特殊文件

流式处理

# bigfile.py —— 内存友好的大文件处理
import io
import os
from pathlib import Path

# 1) 用 StringIO 模拟一个超大文件，不需要真的造磁盘数据
fake_big = StringIO = io.StringIO()
for i in range(1, 100001):
    fake_big.write(f"2026-01-01 12:00:00 INFO 用户 {i} 登录成功\n")
fake_big.seek(0)

# 流式统计：无论文件多大，内存占用恒定
line_count = 0
error_count = 0
char_total = 0
for line in fake_big:
    line_count += 1
    char_total += len(line)
    if "ERROR" in line:
        error_count += 1

print(f"总行数={line_count}  含ERROR行={error_count}  总字符={char_total}")
# 总行数=100000  含ERROR行=0  总字符=4900000

# 2) 生成器管道：逐行转换，不把整个文件读进内存
fake_big.seek(0)


def clean_lines(f):
    for raw in f:
        yield raw.rstrip("\n")


def parse_log(f):
    for line in clean_lines(f):
        parts = line.split(" ", 4)
        if len(parts) == 5:
            yield {"ts": f"{parts[0]} {parts[1]}", "level": parts[2], "msg": parts[4]}


fake_big.seek(0)
for record in parse_log(fake_big):
    if record["level"] == "INFO":
        first_msg = record["msg"]
        break
print("第一条 INFO：", first_msg)

# 3) itertools.islice：只取前 N 行而不影响文件游标
import itertools
fake_big.seek(0)
first_5 = [ln.strip() for ln in itertools.islice(fake_big, 5)]
print("前 5 行：")
for ln in first_5:
    print("  ", ln)

# 4) 真实场景：边写边统计的大日志
log_path = Path("app.log")
with log_path.open("w", encoding="utf-8") as f:
    for i in range(1, 10001):
        f.write(f"line {i}: payload={'x' * 20}\n")

file_size = log_path.stat().st_size
mem_lines = len(log_path.read_text(encoding="utf-8").splitlines())   # 一次性读（内存吃紧）
stream_lines = sum(1 for _ in log_path.open(encoding="utf-8"))        # 流式（恒定内存）
print(f"\n文件 {file_size} 字节；read+splitlines 得 {mem_lines} 行；流式统计得 {stream_lines} 行")

# 5) 追加日志（实际生产做法）
import datetime
with log_path.open("a", encoding="utf-8") as f:
    ts = datetime.datetime.now().isoformat(timespec="seconds")
    f.write(f"{ts} INFO 服务启动完成 pid=1234\n")

# 6) 按行偏移读取超大文件（如数据库导出）
log_path.write_text("".join(f"{i}\n" for i in range(1000000)), encoding="utf-8")
offset = 0
with log_path.open("r", encoding="utf-8") as f:
    for i, line in enumerate(f):
        if i >= 3:
            break
        offset = f.tell()
print("读完前 4 行，指针位置：", offset, "字节")

二进制文件与其他格式

# binary_and_formats.py —— 二进制与特殊文件
from pathlib import Path
import struct

# 1) 二进制读写
Path("data.bin").write_bytes(bytes(range(256)))
raw = Path("data.bin").read_bytes()
print("二进制读取：前 8 字节 =", raw[:8])
print("单个字节：", raw[10], "| 切片类型：", type(raw[10:20]).__name__)

# 2) struct：按二进制格式打包/解包（写协议、解析文件头）
# 格式 '>2sI'：大端、2 字节字符串、4 字节无符号整数
packed = struct.pack(">2sI", b"AB", 123456)
print("struct.pack：", packed, "长度", len(packed))
name, num = struct.unpack(">2sI", packed)
print("struct.unpack：", name, num)

# 3) 十六进制查看二进制内容
Path("head.bin").write_bytes(b"PK\x03\x04rest of the data")
print("前 16 字节 hex：", Path("head.bin").read_bytes()[:16].hex(" "))
print("十六进制文本视图：", Path("head.bin").read_bytes()[:8].hex(" "))

# 4) 逐块处理大二进制文件
big = Path("chunk.bin")
big.write_bytes(b"x" * (2 * 1024 * 1024))       # 2MB
CHUNK = 64 * 1024
total = 0
with big.open("rb") as f:
    while chunk := f.read(CHUNK):        # 海象运算符（Python 3.8+）
        total += len(chunk)
print(f"分块读取 2MB 文件，合计 {total} 字节，每块 {CHUNK // 1024}KB")

# 5) PNG/JPEG 等图片：纯二进制
#    （需要第三方库 Pillow，见第 9 章依赖管理）
# from PIL import Image
# img = Image.open("photo.png")
# print(img.size, img.mode)
# img.thumbnail((200, 200))
# img.save("thumb.png")

# 6) 压缩包
import zipfile

with zipfile.ZipFile("archive.zip", "w", zipfile.ZIP_DEFLATED) as zf:
    zf.writestr("hello.txt", "你好，压缩包！")
    zf.write("chunk.bin", arcname="data/chunk.bin")

with zipfile.ZipFile("archive.zip") as zf:
    print("\n压缩包内容：", zf.namelist())
    print("读取 hello.txt：", zf.read("hello.txt").decode("utf-8"))
    print("testzip() 检查：", zf.testzip())     # None 表示无损坏

# 7) 临时文件
import tempfile

with tempfile.NamedTemporaryFile("w+", suffix=".log", encoding="utf-8",
                                 delete=False) as tf:
    temp_name = tf.name
    tf.write("临时日志\n")
    tf.flush()
    tf.seek(0)
    print("\n临时文件内容：", tf.read().strip())
print("delete=False，所以文件仍在：", Path(temp_name).exists())
Path(temp_name).unlink()          # 手动清理
print("手动删除后存在？", Path(temp_name).exists())

实战：日志文件轮转与归档

# log_manager.py —— 完整实战：读取、追加、轮转、归档日志
import os
import shutil
from datetime import datetime, timedelta
from pathlib import Path


class LogManager:
    """日志文件管理器：追加写入 + 按大小轮转 + 自动归档。"""

    def __init__(self, log_dir, name="app", max_bytes=1024 * 100, keep_days=7):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
        self.name = name
        self.max_bytes = max_bytes
        self.keep_days = keep_days
        self.current = self.log_dir / f"{name}.log"

    # ---------- 写入 ----------
    def write(self, level, message):
        line = f"{datetime.now():%Y-%m-%d %H:%M:%S} {level:5s} {message}\n"
        # 先检查是否需要轮转
        self._rotate_if_needed()
        with self.current.open("a", encoding="utf-8") as f:
            f.write(line)
        return line.strip()

    def _rotate_if_needed(self):
        if self.current.exists() and self.current.stat().st_size >= self.max_bytes:
            stamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            archive = self.log_dir / f"{self.name}_{stamp}.log"
            self.current.rename(archive)
            print(f"  [轮转] {self.current.name} -> {archive.name}")

    # ---------- 读取 ----------
    def tail(self, n=5):
        """读取最后 n 行（流式，不占大内存）。"""
        if not self.current.exists():
            return []
        lines = []
        with self.current.open("r", encoding="utf-8") as f:
            for line in f:
                lines.append(line.rstrip("\n"))
                if len(lines) > n:
                    lines.pop(0)      # 简单滑动窗口
        return lines

    def search(self, keyword):
        """在所有日志中查找关键字，返回 (文件, 行号, 内容)。"""
        hits = []
        for path in sorted(self.log_dir.glob(f"{self.name}*.log")):
            with path.open("r", encoding="utf-8") as f:
                for lineno, line in enumerate(f, start=1):
                    if keyword in line:
                        hits.append((path.name, lineno, line.rstrip("\n")))
        return hits

    def iter_records(self, level=None):
        """生成器：逐条产出解析后的日志记录。"""
        for path in sorted(self.log_dir.glob(f"{self.name}*.log")):
            with path.open("r", encoding="utf-8") as f:
                for lineno, line in enumerate(f, start=1):
                    try:
                        ts, lvl, msg = line.split(" ", 2)
                    except ValueError:
                        yield {"file": path.name, "line": lineno,
                               "raw": line.rstrip("\n"), "error": "格式不正确"}
                        continue
                    if level and lvl.strip() != level:
                        continue
                    yield {"file": path.name, "line": lineno,
                           "ts": ts, "level": lvl.strip(), "msg": msg.strip()}

    # ---------- 清理 ----------
    def cleanup(self, keep_days=None):
        """删除过期日志，返回删除的文件列表。"""
        keep = timedelta(days=keep_days if keep_days is not None else self.keep_days)
        cutoff = datetime.now() - keep
        removed = []
        for path in self.log_dir.glob(f"{self.name}*.log"):
            if path == self.current:
                continue
            mtime = datetime.fromtimestamp(path.stat().st_mtime)
            if mtime < cutoff:
                path.unlink()
                removed.append(path.name)
        return removed

    def stats(self):
        files = list(self.log_dir.glob(f"{self.name}*.log"))
        return {
            "文件数": len(files),
            "总大小(KB)": round(sum(p.stat().st_size for p in files) / 1024, 2),
            "当前文件": self.current.name,
        }


if __name__ == "__main__":
    import tempfile

    with tempfile.TemporaryDirectory() as tmp:
        mgr = LogManager(tmp, name="demo", max_bytes=400, keep_days=1)

        print("=== 写入（触发多次轮转）===")
        for i in range(1, 9):
            mgr.write("INFO", f"用户 {i} 登录成功，来源 IP 192.168.1.{i}")
        mgr.write("ERROR", "数据库连接超时")
        mgr.write("DEBUG", "缓存命中率 92.3%")

        print("\n=== 统计 ===")
        print(mgr.stats())

        print("\n=== tail 最后 3 行 ===")
        for line in mgr.tail(3):
            print("  ", line)

        print("\n=== 搜索关键字 'ERROR' ===")
        for fname, lineno, text in mgr.search("ERROR"):
            print(f"  {fname}:{lineno}  {text}")

        print("\n=== 生成器：只取 ERROR 级别 ===")
        for rec in mgr.iter_records(level="ERROR"):
            print(f"  {rec['file']}:{rec['line']}  {rec['ts']} {rec['msg']}")

        print("\n=== 清理（保留 365 天，本次不会删）===")
        print("  删除：", mgr.cleanup(keep_days=365) or "（无）")

预期输出：

=== 写入（触发多次轮转）===
  [轮转] demo.log -> demo_20260101_120001.log
  [轮转] demo.log -> demo_20260101_120004.log

=== 统计 ===
{'文件数': 3, '总大小(KB)': 0.0, '当前文件': 'demo.log'}

=== tail 最后 3 行 ===
   2026-01-01 12:00:09 DEBUG 缓存命中率 92.3%
   ...（实际以运行为准）

=== 搜索关键字 'ERROR' ===
  demo_20260101_120004.log:7  2026-01-01 12:00:07 ERROR 数据库连接超时

本章小结

• 异常体系：BaseException → Exception → 具体异常。KeyboardInterrupt/SystemExit 不属于 Exception，绝不要用裸 except 捕获。
• 四个子句：try（尽量窄）→ except（最具体）→ else（成功后执行，缩小 try 范围）→ finally（必执行）。
• 捕获原则：只捕获能处理的异常；需要上下文时用 raise X from e；服务层兜底用 except Exception 但必须记日志。
• 自定义异常：继承业务基类、名字以 Error 结尾、携带结构化字段、形成层级树。参数校验用 raise 而不是 assert（-O 会优化掉 assert）。
• with 语句基于 __enter__/__exit__，保证资源只释放一次；contextlib.contextmanager、suppress、ExitStack 是常用快捷方式。
• open 模式：r 读/w 清空写/a 追加/x 独占创建；b 二进制、+ 读写。非法组合直接 ValueError。
• 编码与换行：一律显式写 encoding="utf-8"；跨系统统一用 "\n"；CSV 必须 newline=""，给 Excel 读的 CSV 用 utf-8-sig。
• 读取方式：for line in f 是首选（恒定内存），不要用 f.readlines() 处理大文件。
• 路径处理：用 pathlib.Path 而非字符串拼接；/ 拼路径、.glob()/.rglob() 遍历、.mkdir(parents=True, exist_ok=True) 建目录。
• 格式读写：JSON 用 json.dump/load + ensure_ascii=False；CSV 永远用 csv 模块而非 split(",")；压缩包用 zipfile。

练习题

1. 四子句语义。写一个函数 safe_divide(a, b)，用完整的 try/except/else/finally 实现：除零返回 None（except）、类型错误抛 TypeError（except）、成功走 else 打印结果、finally 打印「已处理」。分别用 6 / 3、6 / 0、"6" / 3 调用并解释输出顺序。
2. 异常转换链。定义 DataError（基类）、ParseError、SchemaError。写 parse_record(line) 解析 "id,name,score" 格式：字段数不对抛 SchemaError，score 不是数字抛 ParseError，都继承 DataError。用 except DataError 批量处理并用 e.args 或自定义字段区分类型，确保 traceback 里能看到 from 链。
3. 禁止裸 except。给下面代码挑错并重写：try: result = 10/0; except: print("错了"); finally: print("结束")。至少指出三个问题，并说明为什么 except Exception 也仍需谨慎。
4. 自定义上下文管理器。用 @contextlib.contextmanager 写 @timer(name)（打印进入/耗时，含 finally），再用 __enter__/__exit__ 手写一个等价类版本 TimerClass。测试正常退出、抛异常退出两种路径，确认清理都执行了。
5. 文件读写基础。完成：① 用 x 模式创建 test.txt，第二次创建应失败并被捕获；② 以 a 追加三行；③ 用 r+ 把第一行改成大写（用 seek + write）；④ 用 for 逐行打印所有行号；⑤ 删掉文件并确认。
6. 编码实验。① 写一个 utf-8 文件并用 gbk 读取，观察 UnicodeDecodeError；② 用 errors="replace" 和 errors="ignore" 再读，比较结果差异；③ 写一个带 BOM 的文件（utf-8-sig），用 utf-8 和 utf-8-sig 分别读取，比较首字符；④ 用 newline 参数在 \n 与 \r\n 之间转换并用二进制模式验证落盘字节。
7. 大文件流式处理。生成 100 万行的日志文件（2026-01-01 HH:MM:SS LEVEL message 格式），用流式方式统计：各级别行数、平均行长、最长的 3 条消息、包含关键字的行数。要求内存占用不随文件大小线性增长，并对比 readlines() 版本的耗时差异。
8. pathlib 全家桶。用 Path 完成：创建 data/2026/01 多级目录、写入三个文件、用 glob 列出 *.json、用 rglob 递归找所有 .log、把 .tmp 后缀批量改名为 .bak、删除超过 7 天未修改的文件、计算目录总大小。最后打印一棵目录树（自写递归）。
9. CSV 实战。读取 orders.csv（含中文、逗号、引号、换行的复杂字段），完成：按状态分组统计订单数与总金额；找出金额最高的 3 单并排序；把统计结果用 DictWriter 写回新 CSV（utf-8-sig）。验证含特殊字符的字段往返一致。
10. JSON 配置管理器。实现 ConfigManager 类：load() 读 JSON（文件不存在时自动生成默认配置并记录 created=True）、get(key, default) 支持点号路径（"db.host"）、set(key, value) 自动建父级并写回、merge(dict) 深合并、save() 原子写入（先写 .tmp 再 replace）。测试全部方法并验证文件内容。
11. 综合实战：全文检索工具。实现命令行风格的 grep：grep(pattern, path, recursive=False, ignore_case=False, show_line_number=True)。要求：用 argparse 解析参数；递归模式用 os.walk 或 Path.rglob；逐文件流式搜索；捕获 UnicodeDecodeError 跳过非文本文件并记录；输出统一格式 文件名:行号:内容；--count 模式只输出每个文件的匹配数；--stats 输出扫描文件数/匹配行数/耗时。最后在自己站点的 chapter-*.html 上试跑并验证结果正确。