第九章:异常处理与日志:让脚本在真实数据面前不崩溃
本文针对科研脚本中常见的数据处理崩溃问题,提出了系统的异常处理与日志记录方案。主要内容包括: 异常分类策略:区分可恢复错误(如编码问题)、不可恢复错误(如关键文件缺失)和需要上报的风险。 结构化异常处理:推荐使用try/except/else/finally结构,强调捕获具体异常而非裸捕获。 三类高频问题处理: 文件不存在采用fail-fast策略 编码问题提供多编码尝试方案 数据质量问题建议跳过

第九章:异常处理与日志:让脚本在真实数据面前不崩溃
写科研脚本最常见的“崩溃现场”不是算法太难,而是现实太脏:
- CSV 里突然多了一列分隔符
- TXT 里混进了不可见字符,编码炸裂
- JSON 少了一个逗号,整批数据无法加载
- 路径写错、文件缺失、权限不够
- 你只用
print(),出了问题只能“重跑 + 猜”
这一章的目标不是教你背异常类型,而是建立一个可复用的防崩工作流:
让错误可预期、可定位、可复盘;让脚本即使失败,也能“体面地失败”。
1. 先建立一个共识:异常不是敌人,异常是信息
Python 的异常机制本质上是在告诉你:哪个假设被现实推翻了。而你的任务是把这些“被推翻的假设”分类处理:
- 可恢复:跳过坏行、替换缺失值、降级策略
- 不可恢复:关键文件缺失、配置错误、数据结构完全不合法(应立即失败)
- 需要上报:运行完成但存在风险(需要日志与告警)
2. try/except/else/finally:把“处理范围”写清楚
Python 的 try/except/else/finally 不是语法花活,它提供了一种清晰的工程结构:
try:只放“可能失败的最小代码块”except:只捕获你确实知道怎么处理的异常else:只有在 try 成功时才执行,避免把“正常逻辑”意外纳入捕获范围finally:无论成功失败都要执行(释放资源、关闭连接等)(Python documentation)
一个推荐的结构如下:
try:
# 只放:可能出错的语句
data = load_json(path)
except FileNotFoundError:
# 你确实能处理:例如提示用户、切换默认路径
...
except ValueError as e:
# 你确实能处理:例如给出更清晰的错误信息
...
else:
# try 成功后再做:后续逻辑
process(data)
finally:
# 必须执行:清理动作
close_resources()
关键原则:捕获具体异常,不要裸捕获。
裸写 except Exception: 很容易把真正的 bug 吞掉,导致错误“悄无声息地变坏”。
3. 文科科研的三类高频崩溃:文件、编码、数据质量
下面是你在 CSV/JSON/TXT 处理中最常遇到的三类异常,以及建议的处理策略。
3.1 文件不存在:FileNotFoundError(不可恢复,应该 fail fast)
from pathlib import Path
def require_file(path: Path) -> Path:
if not path.exists():
raise FileNotFoundError(f"找不到文件:{path}")
return path
这是典型的不可恢复错误:没有原始数据,一切分析都失去意义。正确做法是:尽早失败,给出清晰报错。
3.2 编码问题:UnicodeDecodeError(可恢复,但要明确策略)
对科研而言,最稳的做法是:
- 原始文件尽量统一
utf-8 - 对“历史遗留”或 Excel 导出,允许尝试
utf-8-sig - 不要静默吞掉乱码:至少记录日志,方便复盘
def read_text_safely(path: Path) -> str:
for enc in ("utf-8", "utf-8-sig"):
try:
return path.read_text(encoding=enc)
except UnicodeDecodeError:
continue
raise UnicodeDecodeError("utf-8", b"", 0, 1, f"无法解码:{path}")
3.3 数据质量:坏行、缺列、分隔符混乱(可恢复,建议“跳过 + 记录”)
如果你用 pandas.read_csv 读到“坏行”(字段数不一致),现在推荐用 on_bad_lines 指定策略(error/warn/skip 或 callable),它是对旧参数的替代方案。(Pandas)
import pandas as pd
df = pd.read_csv(
"data/raw/survey.csv",
encoding="utf-8-sig",
on_bad_lines="warn" # 或 "skip"
)
这里要做的不是“追求零坏行”,而是:你必须知道坏行发生过,并且能定位坏行比例与原因(这就是日志的价值)。
4. JSON 的两类典型失败:结构不合法与编码不合法
json.load/json.loads 失败时,常见异常包括 JSONDecodeError 与 UnicodeDecodeError。(Python documentation)
推荐策略:
- JSON 解析失败:记录文件名与失败位置(行/列),并中止(多数情况下不可恢复)
- 输出 JSON:统一
ensure_ascii=False+indent,便于人工审阅与版本对比(diff)(Python documentation)
import json
from pathlib import Path
def load_json(path: Path):
with path.open("r", encoding="utf-8") as f:
return json.load(f)
def dump_json(obj, path: Path):
with path.open("w", encoding="utf-8") as f:
json.dump(obj, f, ensure_ascii=False, indent=2, sort_keys=True)
5. 为什么 print 不够:你需要“可检索的证据链”
print() 的问题在于它不可控、不可分级、不可轮转、不可统一格式。
而 logging 提供了模块化组件:logger、handler、formatter、filter,能够把日志输出到不同目的地并统一格式。(Python documentation)
你要把日志当成科研脚本的“黑匣子记录”:
- 你跑了什么版本的脚本
- 输入文件是什么
- 清洗规则是什么
- 跳过了多少坏行
- 输出文件在哪里
- 运行耗时与异常堆栈是什么
6. 最小可用日志闭环:控制台 + 文件 + 轮转
对于个人科研项目,我建议你从一个最小闭环开始:
- 控制台:运行时即时反馈
- 文件:长期留存,便于复盘
- 轮转:避免日志无限增大(RotatingFileHandler / TimedRotatingFileHandler)(Python documentation)
import logging
from logging.handlers import RotatingFileHandler
from pathlib import Path
def setup_logger(log_dir: Path) -> logging.Logger:
log_dir.mkdir(parents=True, exist_ok=True)
log_path = log_dir / "run.log"
logger = logging.getLogger("research")
logger.setLevel(logging.INFO)
fmt = logging.Formatter(
"%(asctime)s | %(levelname)s | %(name)s | %(message)s"
)
# 控制台
sh = logging.StreamHandler()
sh.setFormatter(fmt)
# 文件(轮转)
fh = RotatingFileHandler(
log_path, maxBytes=5 * 1024 * 1024, backupCount=5, encoding="utf-8"
)
fh.setFormatter(fmt)
# 避免重复添加 handler(在 notebook/重复运行时常见)
if not logger.handlers:
logger.addHandler(sh)
logger.addHandler(fh)
return logger
7. 把“防崩”写成结构:入口边界 + 领域异常 + 统一退出
你需要一个“异常边界”(exception boundary):
业务逻辑内部尽量抛出明确异常;最外层统一捕获、记录、并给出可读结果。
8. 一份可复用的“科研脚本骨架”
下面这个骨架可以直接复制到你的 src/run_pipeline.py,作为每个项目的默认起点。
from pathlib import Path
import time
import logging
import json
import pandas as pd
class DataQualityError(Exception):
"""数据质量问题:字段缺失、关键列为空、比例异常等。"""
def main():
ROOT = Path("research_project")
RAW = ROOT / "data" / "raw"
PROCESSED = ROOT / "data" / "processed"
LOGS = ROOT / "logs"
PROCESSED.mkdir(parents=True, exist_ok=True)
logger = setup_logger(LOGS)
t0 = time.time()
try:
input_csv = RAW / "survey.csv"
if not input_csv.exists():
raise FileNotFoundError(f"缺少输入文件:{input_csv}")
logger.info("开始读取 CSV:%s", input_csv)
df = pd.read_csv(
input_csv,
encoding="utf-8-sig",
on_bad_lines="warn"
)
required_cols = {"name", "age"}
missing = required_cols - set(df.columns)
if missing:
raise DataQualityError(f"缺少关键字段:{missing}")
# 简单清洗示例
df["name"] = df["name"].astype(str).str.strip()
df = df.dropna(subset=["name"])
out_path = PROCESSED / "survey_clean.csv"
df.to_csv(out_path, index=False, encoding="utf-8")
logger.info("输出完成:%s | 行数=%d", out_path, len(df))
except (FileNotFoundError, DataQualityError) as e:
logger.error("可解释失败:%s", e)
return 1
except Exception:
# 任何未预料的异常:保留堆栈,便于定位
logger.exception("未预料异常(请查看堆栈)")
return 2
finally:
cost = time.time() - t0
logger.info("运行结束:耗时 %.2fs", cost)
return 0
上面有两个关键点:
- 可解释失败(已知问题):用
logger.error给出清晰原因 - 不可解释失败(未知问题):用
logger.exception记录完整堆栈,便于排错(这是工程上最省时间的做法)
9. 本章“质量门槛”:10 条脚本不崩的检查口令
每次你写一个新脚本,上线前至少过一遍:
- 输入路径是否校验(存在性/权限)
- 读文件是否指定 encoding
- 数据是否做关键字段检查
- 坏行策略是否明确(error/warn/skip)(Pandas)
- 清洗规则是否记录在日志中
- 输出是否写入 processed,而不是覆盖 raw
- 是否有异常边界(main 捕获)
- 是否有日志文件(可复盘)(Python documentation)
- 是否输出运行摘要(行数、跳过数、耗时)
- 是否能在“缺文件/坏 JSON/坏行 CSV”三种情况下给出可读报错(Python documentation)
下一篇:
更多推荐



所有评论(0)