Python异常与文件

异常处理和文件操作是脚本、Web、数据处理、AI 工程都会频繁使用的基础能力。零基础学习时不要只记 try except 语法，而要理解：异常是错误在调用链上的传播机制，文件操作是程序和外部世界交换数据的入口。

学完本页后，你应该能做到：

1. 看懂异常从底层函数一路传播到入口层的过程。
2. 知道什么时候捕获异常，什么时候继续抛给上层。
3. 会设计业务异常，而不是到处抛 Exception。
4. 会安全读取文本、JSON、CSV、大文件。
5. 知道乱码、文件不存在、权限不足、磁盘满、内存暴涨怎么排查。
6. 能写一个商业项目里常见的“导入文件预校验”脚本。

为什么要学异常和文件

程序运行时一定会遇到不可控情况：用户输入不合法、文件不存在、网络中断、JSON 格式错误、磁盘权限不足。如果不处理异常，程序会直接崩溃；如果乱捕获异常，问题会被隐藏，线上排查会更困难。

文件操作也是一样。很多脚本都要读取配置、处理 CSV、保存日志或导出结果。如果不显式指定编码、不及时释放资源、不检查文件是否存在，就容易出现乱码、文件占用、数据丢失等问题。

商业系统里常见的文件场景：

场景	典型文件	风险
医疗设备资产导入	CSV、Excel	字段缺失、编码错误、重复资产编号
订单批量对账	CSV、TXT	大文件一次性读入导致内存飙升
AI 数据集清洗	JSONL、CSV	单行脏数据导致整批任务中断
配置加载	JSON、YAML、ENV	文件不存在或格式错误导致服务启动失败
报表导出	CSV、JSON	写到一半失败导致用户拿到半截文件

异常不是 if 的替代品

if 用来处理业务上可预期的分支，异常用来表达当前函数无法继续完成任务。

例如：

def register(username: str):
    if not username:
        return {"success": False, "message": "用户名不能为空"}

这里用户名为空是正常业务分支，用 if 很合适。

再看文件读取：

from pathlib import Path


def load_config(path: Path) -> str:
    return path.read_text(encoding="utf-8")

如果文件不存在、权限不足、编码错误，load_config 自己通常无法修复，只能把异常抛给上层，由入口层决定提示用户、终止启动、切换默认配置或报警。

如果把所有异常都用 if 返回错误码，调用者每一步都要检查返回值，很容易漏掉。异常的价值就是：只要当前层无法处理，就让错误沿调用栈向上传播。

try except else finally

基础结构如下：

try:
    value = int("abc")
except ValueError as e:
    print("转换失败", e)
else:
    print("没有异常时执行")
finally:
    print("无论是否异常都会执行")

执行规则：

关键字	什么时候执行	常见用途
try	放可能出错的代码	读取文件、解析 JSON、调用接口
except	捕获指定异常	给用户友好提示、记录日志、转成业务异常
else	try 没有异常时执行	只在成功后提交、写结果、继续处理
finally	无论成功失败都执行	关闭资源、释放锁、清理临时文件

flowchart TD
    A["进入 try"] --> B["执行可能失败的代码"]
    B --> C{"是否抛出异常"}
    C -->|否| D["执行 else"]
    C -->|是| E["查找匹配的 except"]
    E --> F{"找到匹配处理器"}
    F -->|是| G["执行 except"]
    F -->|否| H["继续向上层传播"]
    D --> I["执行 finally"]
    G --> I
    H --> I
    I --> J["离开当前代码块"]

注意：finally 不是“最后一定成功”的意思，它只是保证清理逻辑会运行。如果 finally 中又抛异常，可能会覆盖原始异常，所以 finally 里不要写复杂业务。

异常传播原理

Python 函数调用会形成调用栈。底层函数抛出异常后，解释器会在当前函数找匹配的 except；找不到就退出当前函数，回到调用它的上层继续找。

def read_file():
    raise FileNotFoundError("data.csv 不存在")


def import_data():
    read_file()


def main():
    try:
        import_data()
    except FileNotFoundError as e:
        print("导入失败:", e)


main()

传播过程：

flowchart TD
    A["main 调用 import_data"] --> B["import_data 调用 read_file"]
    B --> C["read_file 抛出 FileNotFoundError"]
    C --> D{"read_file 内是否捕获"}
    D -->|否| E["返回 import_data 查找 except"]
    E --> F{"import_data 内是否捕获"}
    F -->|否| G["返回 main 查找 except"]
    G --> H{"main 是否捕获"}
    H -->|是| I["输出友好错误"]
    H -->|否| J["程序终止并打印 traceback"]

这个机制解释了两个重要问题：

1. 底层代码不一定要捕获异常。底层更适合抛出清晰的异常，让上层统一处理。
2. 入口层必须兜底。CLI、Web Controller、定时任务入口、消息消费者入口都应该做统一异常处理，否则一次异常可能导致任务失败或请求直接 500。

捕获异常的原则

不要这样写：

try:
    data = int("abc")
except Exception:
    pass

这样会让错误消失。线上只会看到结果不对，却没有任何异常信息。

更好的写法：

try:
    data = int("abc")
except ValueError as exc:
    print(f"数字格式错误: {exc}")
    raise

捕获规则：

原则	原因
捕获具体异常	避免把编程错误、系统错误也吞掉
能处理才捕获	当前层只记录但不处理时，应该 raise 继续抛出
入口层统一兜底	入口层负责转成人能看懂的错误、日志和告警
不要静默忽略	否则数据错了也不知道在哪里错
不要用异常控制正常流程	高频正常分支用异常会让代码难懂、性能也更差

raise 和 raise from

raise 用来抛出异常。raise from 用来保留“底层异常”和“业务异常”的因果关系。

场景：底层文件不存在，但业务上想表达“导入文件加载失败”。

from pathlib import Path


class ImportFileError(Exception):
    """导入文件加载失败。"""


def load_import_file(path: Path) -> str:
    try:
        return path.read_text(encoding="utf-8")
    except FileNotFoundError as exc:
        raise ImportFileError(f"导入文件不存在: {path}") from exc
    except UnicodeDecodeError as exc:
        raise ImportFileError(f"导入文件不是 UTF-8 编码: {path}") from exc

为什么要 from exc？

1. 用户看到的是业务错误：导入文件加载失败。
2. 开发者在 traceback 里还能看到根因：FileNotFoundError 或 UnicodeDecodeError。
3. 排查时不会丢失底层现场。

如果不使用 raise from，也能抛业务异常，但因果链不清晰，复杂系统里很难定位根因。

自定义异常体系

小脚本可以直接抛内置异常。商业项目建议有一套业务异常层次，方便入口层统一处理。

class AppError(Exception):
    """应用可预期异常基类。"""


class ValidationError(AppError):
    """用户输入或导入数据不合法。"""


class ImportFileError(AppError):
    """导入文件读取、解析失败。"""


class ExternalServiceError(AppError):
    """第三方接口或外部系统异常。"""

入口层可以这样处理：

def main():
    try:
        run_job()
    except AppError as exc:
        print(f"任务失败: {exc}")
    except Exception as exc:
        print(f"系统异常，需要开发排查: {exc}")
        raise

为什么要把可预期业务异常和未知异常分开？

类型	例子	处理方式
可预期业务异常	用户文件缺字段	提示用户修改数据
外部依赖异常	接口超时	重试、降级、告警
编程错误	None.xxx、下标越界	保留 traceback，开发修复代码

如果所有异常都用 Exception("失败")，入口层无法区分是用户数据问题、依赖问题还是代码 bug。

with 和上下文管理器

文件、网络连接、锁、数据库连接都属于资源。资源使用完必须释放。with 的价值是：即使中间发生异常，也能执行清理逻辑。

from pathlib import Path

path = Path("demo.txt")
with path.open("r", encoding="utf-8") as file:
    content = file.read()

上面代码近似等价于：

file = open("demo.txt", "r", encoding="utf-8")
try:
    content = file.read()
finally:
    file.close()

自定义上下文管理器：

class ManagedResource:
    def __enter__(self):
        print("打开资源")
        return self

    def __exit__(self, exc_type, exc, traceback):
        print("释放资源")
        return False


with ManagedResource() as resource:
    print("使用资源")

__exit__ 的返回值很关键：

返回值	含义
False	不吞异常，异常继续向外传播
True	吞掉异常，外层看不到异常

一般不要在 __exit__ 返回 True，否则错误会被隐藏。

文件路径：优先 pathlib

pathlib 比字符串拼接路径更安全清晰。

from pathlib import Path

base_dir = Path(__file__).resolve().parent
data_file = base_dir / "data" / "demo.json"

常用方法：

方法	作用
Path.cwd()	当前工作目录
Path(__file__).parent	当前代码文件所在目录
path.exists()	是否存在
path.is_file()	是否普通文件
path.is_dir()	是否目录
path.mkdir(parents=True, exist_ok=True)	创建目录
path.resolve()	转成绝对路径

注意：当前工作目录不一定等于代码所在目录。你在 IDE、命令行、定时任务里启动同一个脚本，Path.cwd() 可能不同。配置文件、模板文件建议基于 __file__ 定位。

编码和乱码

文本文件本质上是字节。encoding 告诉 Python 应该按什么规则把字节解码成字符串。

from pathlib import Path

text = Path("users.csv").read_text(encoding="utf-8")

如果文件实际是 GBK，而你按 UTF-8 读，就可能抛：

UnicodeDecodeError: 'utf-8' codec can't decode byte ...

处理建议：

1. 新项目统一 UTF-8。
2. 读写文件都显式指定 encoding="utf-8"。
3. 遇到历史文件，先确认来源系统编码，不要盲猜。
4. errors="replace" 可以临时定位问题，但不要默认用于正式数据处理，因为它会把无法识别的字符替换掉，可能导致数据被悄悄改坏。

诊断示例：

from pathlib import Path

path = Path("legacy.csv")
try:
    text = path.read_text(encoding="utf-8")
except UnicodeDecodeError:
    sample = path.read_bytes()[:200]
    print("前 200 个字节:", sample)
    raise

大文件处理

不要对大文件直接 read() 或 read_text()。

# 不推荐：10GB 文件会尝试一次性读入内存
content = Path("large.log").read_text(encoding="utf-8")

推荐逐行读取：

from pathlib import Path

path = Path("large.log")
with path.open("r", encoding="utf-8") as file:
    for line_no, line in enumerate(file, start=1):
        if "ERROR" in line:
            print(line_no, line.strip())

处理二进制大文件时可以分块：

from pathlib import Path

chunk_size = 1024 * 1024
with Path("video.bin").open("rb") as src:
    while chunk := src.read(chunk_size):
        print("读取块大小:", len(chunk))

为什么逐行或分块更稳？

flowchart TD
    A["10GB 文件"] --> B{"一次性 read"}
    B --> C["需要约 10GB 以上内存"]
    C --> D["可能 OOM 或机器变慢"]
    A --> E{"逐行/分块读取"}
    E --> F["内存只保存当前行或当前块"]
    F --> G["适合日志扫描、导入预处理、数据清洗"]

JSON 文件

写入 JSON：

import json
from pathlib import Path

data = {"name": "Tom", "age": 18}
Path("user.json").write_text(
    json.dumps(data, ensure_ascii=False, indent=2),
    encoding="utf-8",
)

读取 JSON：

import json
from pathlib import Path

text = Path("user.json").read_text(encoding="utf-8")
data = json.loads(text)

解析外部 JSON 要处理格式错误和字段缺失：

import json
from pathlib import Path


class ValidationError(Exception):
    pass


def load_user(path: Path) -> dict:
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValidationError(f"JSON 格式错误: {path}") from exc

    if "name" not in data:
        raise ValidationError("缺少 name 字段")
    return data

为什么不能只 json.loads 后直接用？因为外部输入永远不可信。用户可能少字段、字段类型不对、文件内容不完整。解析成功只代表 JSON 语法正确，不代表业务数据正确。

CSV 文件

CSV 是商业系统最常见的批量导入导出格式。推荐使用标准库 csv，不要自己按逗号 split，因为字段里可能包含逗号、引号、换行。

读取：

import csv
from pathlib import Path

path = Path("assets.csv")
with path.open("r", encoding="utf-8", newline="") as file:
    reader = csv.DictReader(file)
    for row in reader:
        print(row["asset_code"], row["asset_name"])

写入：

import csv
from pathlib import Path

rows = [
    {"asset_code": "A001", "asset_name": "心电监护仪"},
    {"asset_code": "A002", "asset_name": "输液泵"},
]

with Path("result.csv").open("w", encoding="utf-8", newline="") as file:
    writer = csv.DictWriter(file, fieldnames=["asset_code", "asset_name"])
    writer.writeheader()
    writer.writerows(rows)

newline="" 是为了让 csv 模块自己处理换行，否则在 Windows 上可能出现空行。

原子写入

报表、清洗结果、配置文件写入时，最怕写到一半失败，留下半截文件。更稳的方式是：先写临时文件，成功后再替换目标文件。

from pathlib import Path


def atomic_write_text(path: Path, content: str):
    tmp_path = path.with_suffix(path.suffix + ".tmp")
    tmp_path.write_text(content, encoding="utf-8")
    tmp_path.replace(path)

流程：

flowchart TD
    A["准备写 result.csv"] --> B["先写 result.csv.tmp"]
    B --> C{"临时文件写入成功"}
    C -->|否| D["保留旧 result.csv"]
    C -->|是| E["replace 替换目标文件"]
    E --> F["用户看到完整新文件"]

如果直接写目标文件，中途磁盘满或进程被杀，目标文件可能损坏。

商业 Demo：医疗资产 CSV 导入预校验

需求：用户上传资产 CSV，系统需要先做预校验，输出两份文件：

1. clean_assets.csv：校验通过的数据。
2. error_assets.csv：校验失败的数据和失败原因。

输入字段：

字段	说明
asset_code	资产编号，必填，不能重复
asset_name	资产名称，必填
department	科室，必填
price	原值，必须是大于等于 0 的数字

完整 Demo：

import csv
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path


class AppError(Exception):
    pass


class ImportFileError(AppError):
    pass


class ValidationError(AppError):
    pass


@dataclass(frozen=True)
class AssetRow:
    asset_code: str
    asset_name: str
    department: str
    price: Decimal


def parse_price(value: str) -> Decimal:
    try:
        price = Decimal(value)
    except (InvalidOperation, TypeError) as exc:
        raise ValidationError("price 必须是数字") from exc

    if price < 0:
        raise ValidationError("price 不能小于 0")
    return price


def validate_row(row: dict, seen_codes: set[str]) -> AssetRow:
    asset_code = (row.get("asset_code") or "").strip()
    asset_name = (row.get("asset_name") or "").strip()
    department = (row.get("department") or "").strip()
    price_text = (row.get("price") or "").strip()

    if not asset_code:
        raise ValidationError("asset_code 不能为空")
    if asset_code in seen_codes:
        raise ValidationError(f"asset_code 重复: {asset_code}")
    if not asset_name:
        raise ValidationError("asset_name 不能为空")
    if not department:
        raise ValidationError("department 不能为空")

    price = parse_price(price_text)
    seen_codes.add(asset_code)
    return AssetRow(asset_code, asset_name, department, price)


def read_assets(path: Path) -> tuple[list[AssetRow], list[dict]]:
    if not path.exists():
        raise ImportFileError(f"文件不存在: {path}")
    if not path.is_file():
        raise ImportFileError(f"不是普通文件: {path}")

    clean_rows: list[AssetRow] = []
    error_rows: list[dict] = []
    seen_codes: set[str] = set()

    try:
        with path.open("r", encoding="utf-8", newline="") as file:
            reader = csv.DictReader(file)
            required = {"asset_code", "asset_name", "department", "price"}
            if not reader.fieldnames or not required.issubset(reader.fieldnames):
                raise ImportFileError(f"CSV 表头必须包含: {sorted(required)}")

            for line_no, row in enumerate(reader, start=2):
                try:
                    clean_rows.append(validate_row(row, seen_codes))
                except ValidationError as exc:
                    row["line_no"] = line_no
                    row["error"] = str(exc)
                    error_rows.append(row)
    except UnicodeDecodeError as exc:
        raise ImportFileError("文件编码必须是 UTF-8") from exc

    return clean_rows, error_rows


def write_clean(path: Path, rows: list[AssetRow]):
    with path.open("w", encoding="utf-8", newline="") as file:
        writer = csv.DictWriter(
            file,
            fieldnames=["asset_code", "asset_name", "department", "price"],
        )
        writer.writeheader()
        for row in rows:
            writer.writerow(
                {
                    "asset_code": row.asset_code,
                    "asset_name": row.asset_name,
                    "department": row.department,
                    "price": str(row.price),
                }
            )


def write_errors(path: Path, rows: list[dict]):
    if not rows:
        path.write_text("", encoding="utf-8")
        return

    fieldnames = list(rows[0].keys())
    with path.open("w", encoding="utf-8", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)


def main():
    input_path = Path("assets.csv")
    output_dir = Path("output")
    output_dir.mkdir(exist_ok=True)

    try:
        clean_rows, error_rows = read_assets(input_path)
        write_clean(output_dir / "clean_assets.csv", clean_rows)
        write_errors(output_dir / "error_assets.csv", error_rows)
        print(f"校验通过 {len(clean_rows)} 行，失败 {len(error_rows)} 行")
    except AppError as exc:
        print(f"导入失败: {exc}")


if __name__ == "__main__":
    main()

这个 Demo 体现了几个工程原则：

1. read_assets 负责文件读取和表头校验。
2. validate_row 负责单行字段校验。
3. ValidationError 表示某一行业务数据错误，不影响整批继续校验。
4. ImportFileError 表示文件级错误，通常整批都不能继续。
5. 所有文件都显式指定 encoding="utf-8" 和 newline=""。

导入流程：

flowchart TD
    A["用户上传 assets.csv"] --> B["检查文件存在和类型"]
    B --> C["按 UTF-8 打开 CSV"]
    C --> D["检查表头"]
    D --> E["逐行读取"]
    E --> F{"单行是否通过校验"}
    F -->|是| G["加入 clean_rows"]
    F -->|否| H["记录 line_no 和 error"]
    G --> I{"是否还有下一行"}
    H --> I
    I -->|有| E
    I -->|没有| J["写 clean_assets.csv"]
    J --> K["写 error_assets.csv"]
    K --> L["返回导入预校验结果"]

常见坑

问题	错误写法	后果	正确做法
吞异常	except Exception: pass	数据错了但没有日志	捕获具体异常，必要时重新抛出
不指定编码	open("a.csv")	不同机器表现不同	open(..., encoding="utf-8")
一次性读大文件	read_text()	内存暴涨	逐行或分块读取
手写 CSV 解析	line.split(",")	遇到引号、逗号就错	使用 csv.DictReader
写目标文件到一半失败	直接覆盖原文件	文件损坏	先写临时文件再替换
相对路径混乱	依赖当前目录	定时任务找不到文件	基于配置或 __file__ 定位

生产排查流程

文件处理任务失败时，不要只看“失败”两个字，要按层次排查。

flowchart TD
    A["文件任务失败"] --> B{"文件是否存在"}
    B -->|否| C["检查上传路径、挂载目录、定时任务工作目录"]
    B -->|是| D{"是否有读取权限"}
    D -->|否| E["检查文件权限、运行用户、容器挂载权限"]
    D -->|是| F{"是否编码错误"}
    F -->|是| G["确认来源系统编码，统一转 UTF-8"]
    F -->|否| H{"是否格式错误"}
    H -->|是| I["检查 JSON/CSV 表头、分隔符、坏行"]
    H -->|否| J{"是否文件过大"}
    J -->|是| K["改为逐行/分块处理，观察内存"]
    J -->|否| L["查看完整 traceback 和业务日志"]

具体排查清单：

现象	重点检查
FileNotFoundError	当前工作目录、相对路径、文件是否上传成功
PermissionError	运行用户、目录权限、容器卷挂载读写权限
UnicodeDecodeError	文件真实编码、是否混入特殊字符
JSONDecodeError	是否少括号、少引号、文件写到一半
CSV 列读取不到	表头是否一致、是否有 BOM、分隔符是否不是逗号
内存突然升高	是否一次性 read()、是否把所有行都缓存到列表
输出文件损坏	是否直接覆盖写、进程是否中途退出、磁盘是否满

面试标准回答

try except else finally 的执行顺序是什么？

标准回答：先执行 try，如果没有异常就执行 else，最后执行 finally；如果有异常，就查找匹配的 except，执行完后仍然执行 finally。finally 常用于释放资源，不适合写复杂业务逻辑。

追问：finally 一定会执行吗？

回答：正常情况下会执行，例如异常、return 前都会执行。但进程被强杀、解释器崩溃、机器断电这类极端情况无法保证。

raise 和 raise from 有什么区别？

标准回答：raise 是抛出或重新抛出异常；raise from 会把当前业务异常和底层异常建立因果关系。项目中常用于把 FileNotFoundError、UnicodeDecodeError 转成业务异常，同时保留根因 traceback。

为什么推荐使用 with 打开文件？

标准回答：with 基于上下文管理器协议，会在退出代码块时自动调用 __exit__，等价于在 finally 中关闭资源。即使读取过程中抛异常，也能释放文件句柄，避免文件占用和资源泄漏。

如何处理大文件？

标准回答：不能一次性 read() 到内存。文本大文件用逐行读取，二进制大文件用固定大小分块读取。如果需要聚合统计，尽量边读边处理，避免把所有数据放进列表。

为什么不能捕获 Exception 后什么都不做？

标准回答：这样会吞掉错误，导致调用方以为成功，实际数据已经不可靠。正确做法是捕获具体异常，记录上下文，当前层能处理就处理，不能处理就 raise 继续抛给上层。

JSON 解析成功是否代表数据一定正确？

标准回答：不代表。JSON 解析成功只说明语法合法，不说明业务字段完整、类型正确、值合法。项目中还要校验必填字段、字段类型、枚举值和业务范围。

关联知识点

• Python 数据结构：理解 list、dict、set 在导入校验中的作用。
• Python 函数与模块：理解函数拆分、入口函数、循环导入问题。
• Python 面向对象：理解 dataclass、异常类、分层设计。
• Python 日志与调试：学习如何记录 traceback、定位线上问题。
• Python 测试：为文件导入校验编写单元测试。