用 Python iterparse 流式转换 XML 到 CSV：释放节点，保留完整报告

python3 --version
mkdir xml-stream-demo
cd xml-stream-demo
cat > inventory.xml <<'XML'
<?xml version="1.0" encoding="UTF-8"?>
<inventory xmlns="urn:demo:inventory">
  <item sku="001"><name>螺母, M6</name><quantity>12</quantity></item>
  <item sku="002"><name>Washer &amp; spacer</name><quantity>0</quantity></item>
</inventory>
XML

"""Stream trusted local inventory XML into CSV. Python 3.9+."""
import argparse
import csv
import os
from pathlib import Path
import sys
import tempfile
import xml.etree.ElementTree as ET

NS = "{urn:demo:inventory}"


def field(item, name):
    matches = item.findall(NS + name)
    if len(matches) != 1 or len(matches[0]) != 0:
        raise ValueError(f"expected one plain-text {name}")
    value = (matches[0].text or "").strip()
    if not value:
        raise ValueError(f"empty {name}")
    return value


def convert(source, target):
    writer = csv.writer(target, lineterminator="\n")
    writer.writerow(["sku", "name", "quantity"])
    root = None
    depth = count = 0
    for event, node in ET.iterparse(source, events=("start", "end")):
        if event == "start":
            depth += 1
            if depth == 1:
                root = node
                if root.tag != NS + "inventory":
                    raise ValueError("expected inventory root with demo namespace")
            continue
        if depth == 2:
            if node.tag != NS + "item":
                raise ValueError("expected item directly under inventory")
            sku = node.get("sku", "").strip()
            if not sku:
                raise ValueError("missing sku attribute")
            name = field(node, "name")
            quantity = field(node, "quantity")
            if not quantity.isascii() or not quantity.isdecimal():
                raise ValueError("quantity must contain ASCII digits only")
            writer.writerow([sku, name, int(quantity)])
            count += 1
            root.remove(node)
            node.clear()
        depth -= 1
    if count == 0:
        raise ValueError("inventory contains no items")
    return count


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("input", type=Path)
    parser.add_argument("output", type=Path)
    args = parser.parse_args()
    temporary = None
    try:
        if args.input.resolve() == args.output.resolve():
            raise ValueError("input and output must be different paths")
        with args.input.open("rb") as source:
            with tempfile.NamedTemporaryFile(
                mode="w", encoding="utf-8", newline="",
                dir=args.output.parent, prefix=".xml-csv-", delete=False,
            ) as target:
                temporary = Path(target.name)
                count = convert(source, target)
        os.replace(temporary, args.output)
    except (OSError, ValueError, ET.ParseError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)
    print(f"OK: {count} items")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python3 xml_to_csv.py --help
python3 xml_to_csv.py inventory.xml inventory.csv

OK: 2 items

sku,name,quantity
001,"螺母, M6",12
002,Washer & spacer,0

python3 - <<'PY_CHECK'
import csv
from pathlib import Path
import subprocess
import sys

with open("inventory.csv", encoding="utf-8", newline="") as source:
    assert list(csv.reader(source)) == [
        ["sku", "name", "quantity"],
        ["001", "螺母, M6", "12"],
        ["002", "Washer & spacer", "0"],
    ]
before = Path("inventory.csv").read_bytes()
original = Path("inventory.xml").read_bytes()
Path("truncated.xml").write_bytes(original.replace(b"</inventory>", b""))
result = subprocess.run(
    [sys.executable, "xml_to_csv.py", "truncated.xml", "inventory.csv"],
    capture_output=True, text=True,
)
assert result.returncode == 1, result
assert result.stdout == "" and result.stderr.startswith("ERROR:")
assert Path("inventory.csv").read_bytes() == before
assert Path("inventory.xml").read_bytes() == original
assert not list(Path(".").glob(".xml-csv-*"))
print("PASS: CSV round-trip; truncated XML preserves old output")
PY_CHECK

python3 - <<'PY_DATA'
with open("many.xml", "w", encoding="utf-8") as target:
    target.write('<inventory xmlns="urn:demo:inventory">\n')
    for index in range(100000):
        target.write(
            f'<item sku="{index:06d}"><name>Part {index}</name>'
            '<quantity>1</quantity></item>\n'
        )
    target.write('</inventory>\n')
PY_DATA
python3 xml_to_csv.py many.xml many.csv
python3 - <<'PY_COUNT'
import csv
with open("many.csv", encoding="utf-8", newline="") as source:
    rows = csv.reader(source)
    assert next(rows) == ["sku", "name", "quantity"]
    assert sum(1 for _ in rows) == 100000
print("PASS: 100000 CSV records")
PY_COUNT