本篇是《Python 从入门到初级》第 13 / 20 篇。示例使用 Python 3.12 兼容语法,完整目录结构与数据见配套源码包。
前面的 CSV 脚本已经能生成报告,但如果另一个程序也要统计订单,不应该再复制整段代码。本篇把“读数据”和“做汇总”拆成模块,为后面的数据库、命令行和测试复用。

1. 模块、包和入口分别是什么
一个 .py 文件可以是模块;包把相关模块组织到目录中;入口负责接收用户命令并安排调用。
这里使用传统的带 __init__.py 的包结构:
examples/
├── lesson13.py
├── data/orders.csv
└── ops_tools/
├── __init__.py
└── orders.py
__init__.py 本例可以只放一句包说明。不要把自己的文件命名为 csv.py、json.py 或 pathlib.py,否则可能遮住要导入的标准库模块。
2. 先给可复用模块写清输入约定
订单模块接受含必需表头的 CSV,要求编号不重复、分类非空、金额为非负整数分,状态只能为 paid、pending、cancelled。空数据表允许,但表头必须存在。
相较于第 11 篇,它不再对错误记录“算了再说”,而是在读取边界拒绝不符合约定的数据,让调用者决定提示、停止还是处理其他文件。
保存 ops_tools/orders.py:
import csv
import json
from pathlib import Path
REQUIRED_FIELDS = {"order_id", "category", "amount_cents", "status"}
STATUSES = {"paid", "pending", "cancelled"}
def read_orders(path):
orders = []
seen_ids = set()
with Path(path).open(newline="", encoding="utf-8-sig") as handle:
reader = csv.DictReader(handle)
if reader.fieldnames is None or not REQUIRED_FIELDS.issubset(reader.fieldnames):
raise ValueError("CSV 缺少必需表头")
for row in reader:
line_number = reader.line_num
if None in row or any(row.get(field) is None for field in REQUIRED_FIELDS):
raise ValueError(f"第 {line_number} 行字段数量不匹配")
order_id = row["order_id"].strip()
category = row["category"].strip()
status = row["status"].strip()
if not order_id or not category:
raise ValueError(f"第 {line_number} 行编号或分类为空")
if order_id in seen_ids:
raise ValueError(f"第 {line_number} 行订单编号重复")
if status not in STATUSES:
raise ValueError(f"第 {line_number} 行状态不受支持")
try:
amount = int(row["amount_cents"])
except ValueError as exc:
raise ValueError(f"第 {line_number} 行金额必须是整数分") from exc
if amount < 0:
raise ValueError(f"第 {line_number} 行金额不能为负数")
seen_ids.add(order_id)
orders.append({"order_id": order_id, "category": category,
"amount_cents": amount, "status": status})
return orders
def summarize(orders):
paid = [order for order in orders if order["status"] == "paid"]
categories = {}
for order in paid:
name = order["category"]
categories[name] = categories.get(name, 0) + order["amount_cents"]
return {
"total_orders": len(orders),
"paid_orders": len(paid),
"paid_amount_cents": sum(order["amount_cents"] for order in paid),
"by_category_cents": dict(sorted(categories.items())),
}
def save_json(summary, path):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("x", encoding="utf-8") as handle:
json.dump(summary, handle, ensure_ascii=False, indent=2)
handle.write("\n")
3. 三个函数为什么分开
read_orders() 负责文件和数据验证,返回清洗过的记录列表;summarize() 负责对符合约定的记录计算汇总;save_json() 负责输出。
这样测试计算时不必总读真实文件,换成数据库或 HTTP 入口时,也有清晰的复用边界。summarize() 仍要求调用方提供符合字段约定的记录,并不是接受任何 Python 对象的万能转换器。
save_json 使用 x 模式,输出文件存在时拒绝覆盖。失败写入不具备自动回滚或原子替换能力,后面的工具会明确报告错误;这仍是一个入门实现,不是事务式文件存储。
4. 编写一个很薄的入口
保存并运行 lesson13.py:
from pathlib import Path
from ops_tools.orders import read_orders, summarize
def main():
base = Path(__file__).resolve().parent
orders = read_orders(base / "data" / "orders.csv")
summary = summarize(orders)
print("订单数:", summary["total_orders"])
print("已支付金额(分):", summary["paid_amount_cents"])
if __name__ == "__main__":
main()
预期订单数为 5,已支付金额为 38300 分。入口知道数据在哪,模块知道怎样处理数据;两者没有相互抢占职责。
from ops_tools.orders import ... 从包中导入所需名字。Python 的模块加载机制见 官方 Modules 教程。运行时应保留目录结构,不要只把入口文件单独复制走。
5. if __name__ 的作用
当文件作为主程序运行时,__name__ 为 "__main__";作为模块导入时,它通常是模块名。把启动动作放在这个判断下面,可以让“导入函数”和“执行工具”分开。
模块顶层的其他语句仍会在首次导入执行,所以不要在顶层无条件调用接口、复制文件或删除数据。本例模块只定义常量和函数。
6. 怎样确认导入的是哪份代码
在项目根目录可以查看模块位置:
python -c "import ops_tools.orders as m; print(m.__file__)"
遇到 ModuleNotFoundError,先查当前项目目录、解释器和包结构。不要立即把各种目录硬塞进 sys.path,或把项目文件复制到系统安装目录。
练习:给数据增加重复编号,预期读入时被明确拒绝;恢复原始文件,再导入 summarize 计算空列表,预期数量和金额为 0。它们检验的是模块的输入约定,而不只是 import 是否成功。