Python 完整订单脚本图解：从 JSON 输入到汇总输出

把文件、JSON、容器、判断、循环、函数和错误处理串成一份可运行订单脚本，从三条输入记录得到 2 单、328.0 的已支付汇总。

前八章分别拆开了变量、容器、条件、循环、函数、文件和错误路径。这一章不再增加新语法，而是把它们装进一份可以独立运行的订单脚本：从 orders.json 读取三条记录，校验并规范字段，筛选已支付订单，统计数量和金额，最后写出 paid_orders_summary.json。

固定输入中，A001 和 A003 已支付，因此脚本的基准结果必须是：2 单，总金额 328.0。

先看整条数据管道

脚本只做五件事，每一步都有明确的输入和输出：

这条管道刻意保持单一：一个输入文件、一个脚本、一个输出文件。它不需要配置层、数据库或第三方数据框架。

准备固定输入文件

在脚本所在目录创建 orders.json：

[
  {"order_no": "A001", "sku": "K161", "amount": 129.0, "status": "已支付"},
  {"order_no": "A002", "sku": "K162", "amount": 88.0, "status": "已退款"},
  {"order_no": "A003", "sku": "K161", "amount": 199.0, "status": "已支付"}
]

四个字段各有职责：

完整脚本：复制后可以直接运行

把下面代码保存为 process_orders.py，与 orders.json 放在同一目录：

import json
from pathlib import Path


# A · 输入与输出位置
INPUT_PATH = Path("orders.json")
OUTPUT_PATH = Path("paid_orders_summary.json")


# B · 从 JSON 文件取得一批原始订单
def load_orders(path):
    with path.open("r", encoding="utf-8") as file:
        orders = json.load(file)

    if not isinstance(orders, list):
        raise TypeError("orders.json 顶层必须是列表")

    return orders


# C · 校验并规范一条订单
def normalize_order(raw_order):
    if not isinstance(raw_order, dict):
        raise TypeError("每条订单必须是字典")

    required_fields = {"order_no", "sku", "amount", "status"}
    missing_fields = required_fields - raw_order.keys()

    if missing_fields:
        missing_text = ", ".join(sorted(missing_fields))
        raise KeyError(f"缺少字段：{missing_text}")

    return {
        "order_no": str(raw_order["order_no"]).strip(),
        "sku": str(raw_order["sku"]).strip().upper(),
        "amount": float(raw_order["amount"]),
        "status": str(raw_order["status"]).strip(),
    }


# D · 筛选已支付订单并生成汇总
def build_summary(clean_orders, problems):
    paid_order_nos = []
    paid_count = 0
    paid_amount = 0.0

    for order in clean_orders:
        if order["status"] != "已支付":
            continue

        paid_order_nos.append(order["order_no"])
        paid_count = paid_count + 1
        paid_amount = paid_amount + order["amount"]

    return {
        "paid_order_nos": paid_order_nos,
        "paid_count": paid_count,
        "paid_amount": round(paid_amount, 2),
        "problem_count": len(problems),
        "problems": problems,
    }


# E · 写出结果，并把前面四段串起来
def write_summary(path, summary):
    with path.open("w", encoding="utf-8") as file:
        json.dump(summary, file, ensure_ascii=False, indent=2)


def main():
    raw_orders = load_orders(INPUT_PATH)
    clean_orders = []
    problems = []

    for raw_order in raw_orders:
        try:
            clean_order = normalize_order(raw_order)
        except (KeyError, TypeError, ValueError) as error:
            order_no = (
                raw_order.get("order_no", "未知订单")
                if isinstance(raw_order, dict)
                else "未知订单"
            )
            problems.append({
                "order_no": order_no,
                "error_type": type(error).__name__,
                "message": str(error),
            })
            continue

        clean_orders.append(clean_order)

    summary = build_summary(clean_orders, problems)
    write_summary(OUTPUT_PATH, summary)

    print("已写出：", OUTPUT_PATH)
    print("已支付订单：", summary["paid_count"])
    print("已支付金额：", summary["paid_amount"])
    print("问题记录：", summary["problem_count"])


if __name__ == "__main__":
    main()

在该目录运行：

python process_orders.py

如果环境使用 python3 或 py，替换命令里的 python 即可。正常输出：

已写出： paid_orders_summary.json
已支付订单： 2
已支付金额： 328.0
问题记录： 0

五段代码分别负责什么

完整代码不应被看成一整堵符号。注释 A～E 把它分成五个职责，每段只回答一个问题：

main() 是整条管道的装配点。它没有把所有逻辑重新写一遍，只负责按顺序传递结果：raw_orders → clean_orders / problems → summary → 输出文件。

三条订单逐条经过脚本

clean_orders 和 problems 一开始都是空列表。每条记录先经过 normalize_order()；规范成功后才有资格参与状态筛选。

注意“校验成功”和“业务状态命中”是两回事。A002 的字段完全合法，所以它进入 clean_orders；只是状态为“已退款”，因此没有进入已支付统计。

输入文件与输出文件怎样对应

脚本不会把完整输入原样复制到结果文件。输入面向“逐条处理”，输出面向“继续使用汇总”：

实际写出的 JSON 是：

{
  "paid_order_nos": [
    "A001",
    "A003"
  ],
  "paid_count": 2,
  "paid_amount": 328.0,
  "problem_count": 0,
  "problems": []
}

这个文件可以继续交给报表、通知或其他脚本；本章不再增加这些下游步骤。

补充路径：一条金额异常怎样离开主线

基准输入没有坏数据。现在只做一次补充实验：把 A003 的金额从 199.0 改成字符串 "未知"，其他字段不变。

[
  {"order_no": "A001", "sku": "K161", "amount": 129.0, "status": "已支付"},
  {"order_no": "A002", "sku": "K162", "amount": 88.0, "status": "已退款"},
  {"order_no": "A003", "sku": "K161", "amount": "未知", "status": "已支付"}
]

同一份 process_orders.py 不需要修改。A003 到达 float(raw_order["amount"]) 时产生 ValueError，被 main() 里针对单条记录的处理器接住，写入 problems，然后继续结束本轮。

这次输出文件中的核心字段会类似下面（具体错误文字可能随 Python 版本略有不同）：

{
  "paid_order_nos": ["A001"],
  "paid_count": 1,
  "paid_amount": 129.0,
  "problem_count": 1,
  "problems": [
    {
      "order_no": "A003",
      "error_type": "ValueError",
      "message": "could not convert string to float: '未知'"
    }
  ]
}

错误路径的关键不是“程序没有停”，而是结果仍然诚实：A003 没有被算进金额，同时订单号、异常类型和原因被明确留下。把 orders.json 恢复成基准数据后再次运行，结果就会回到 2 单、328.0。

用结果反查每个阶段

运行完成后，不要只看“有没有生成文件”。按下面顺序验证：

1. paid_order_nos 是否只包含 A001、A003。
2. paid_count 是否等于列表长度 2。
3. paid_amount 是否等于 129.0 + 199.0 = 328.0。
4. 基准数据的 problem_count 是否为 0。
5. 把 A003 金额改坏后，它是否出现在 problems，并从统计中消失。