使用Python3将指定嵌套结构XML数据转换为CSV表格的方法求助

问题背景 #

使用Python 3.8版本，需要将指定结构的XML气象观测数据转换为CSV格式表格，现有网上案例不匹配该XML结构，新手无法直接适配。

示例XML #

<?xml version="1.0"?>
<observations>
<station tz="Australia/Brisbane" description="Brisbane" forecast-district-id="QLD_PW015" lon="153.0389" lat="-27.4808" type="AWS" stn-height="8.13" stn-name="BRISBANE" bom-id="040913" wmo-id="94576">
<period wind-src="metar_10" time-local="2021-07-11T09:30:00+10:00" time-utc="2021-07-10T23:30:00+00:00" index="0">
<level type="surface" index="0">
<element type="apparent_temp" units="Celsius">14.0</element>
<element type="delta_t" units="Celsius">4.4</element>
<element type="gust_kmh" units="km/h">19</element>
<element type="wind_gust_spd" units="knots">10</element>
<element type="air_temperature" units="Celsius">16.6</element>
<element type="dew_point" units="Celsius">7.8</element>
<element type="pres" units="hPa">1022.0</element>
<element type="msl_pres" units="hPa">1022.0</element>
<element type="qnh_pres" units="hPa">1022.0</element>
<element type="rain_hour" units="mm">0.0</element>
<element type="rain_ten" units="mm">0.0</element>
<element type="rel-humidity" units="%">56</element>
<element type="wind_dir">WSW</element>
<element type="wind_dir_deg" units="deg">244</element>
<element type="wind_spd_kmh" units="km/h">11</element>
<element type="wind_spd" units="knots">6</element>
</level>
</period>
</station>
<station tz="Australia/Brisbane" description="Brisbane Airport" forecast-district-id="QLD_PW015" lon="153.1292" lat="-27.3917" type="AWS" stn-height="4.51" stn-name="BRISBANE AERO" bom-id="040842" wmo-id="94578">
<period wind-src="metar_10" time-local="2021-07-11T09:30:00+10:00" time-utc="2021-07-10T23:30:00+00:00" index="0">
<level type="surface" index="0">
<element type="apparent_temp" units="Celsius">11.5</element>
<element type="cloud">Clear</element>
<element type="cloud_oktas">0</element>
<element type="delta_t" units="Celsius">4.1</element>
<element type="gust_kmh" units="km/h">33</element>
<element type="wind_gust_spd" units="knots">18</element>
<element type="air_temperature" units="Celsius">16.7</element>
<element type="dew_point" units="Celsius">8.7</element>
<element type="pres" units="hPa">1021.8</element>
<element type="msl_pres" units="hPa">1021.8</element>
<element type="qnh_pres" units="hPa">1021.9</element>
<element type="rain_hour" units="mm">0.0</element>
<element type="rain_ten" units="mm">0.0</element>
<element type="rel-humidity" units="%">59</element>
<element type="vis_km" units="km">10</element>
<element type="wind_dir">SW</element>
<element type="wind_dir_deg" units="deg">218</element>
<element type="wind_spd_kmh" units="km/h">26</element>
<element type="wind_spd" units="knots">14</element>
</level>
</period>
</station>
</observations>

预期输出CSV #

description,pres,dew_point,rel-humidity,wind_spd_kmh
Brisbane,1022.0,7.8,56,11
Brisbane Airport,1021.8,8.7,59,26

原有脚本问题 #

• 没有导入csv模块，运行会报错
• 多层嵌套循环逻辑冗余，效率极低
• 循环内反复打开CSV文件，写入效率差
• 未处理缺失字段，部分站点没有对应属性时会出现列错位
• 缩进错误，无法直接运行

修正后可直接运行代码 #

import csv
from xml.dom import minidom

# 配置部分：修改这里的路径和要提取的字段即可
XML_PATH = "你的XML文件路径.xml"
CSV_PATH = "输出的CSV路径.csv"
# 按你需要的输出顺序定义要提取的字段，第一个是station的属性，后面是element的type属性值
EXTRACT_FIELDS = [
    "description",
    "pres",
    "dew_point",
    "rel-humidity",
    "wind_spd_kmh"
]

if __name__ == "__main__":
    # 读取XML文件
    dom = minidom.parse(XML_PATH)
    stations = dom.getElementsByTagName("station")
    all_rows = []

    for station in stations:
        # 先拿station属性里的字段值
        row = {}
        for field in EXTRACT_FIELDS:
            # 先尝试从station的属性里取
            if station.hasAttribute(field):
                row[field] = station.getAttribute(field)
            else:
                # 取不到默认设为空，后面再找element里的
                row[field] = ""
        
        # 遍历一次所有element节点，匹配需要的字段
        elements = station.getElementsByTagName("element")
        for elem in elements:
            elem_type = elem.getAttribute("type")
            if elem_type in EXTRACT_FIELDS:
                # 拿到element的文本值
                row[elem_type] = elem.firstChild.nodeValue.strip()
        
        # 按定义的字段顺序拼成列表
        sorted_row = [row[field] for field in EXTRACT_FIELDS]
        all_rows.append(sorted_row)

    # 一次性写入CSV
    with open(CSV_PATH, mode="w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        # 写表头
        writer.writerow(EXTRACT_FIELDS)
        # 写内容
        writer.writerows(all_rows)

使用说明 #

把XML_PATH和CSV_PATH改成你自己的文件路径即可运行，如果需要增减输出字段，直接修改EXTRACT_FIELDS列表就行，不需要改核心逻辑。

内容的提问来源于stack exchange，提问作者D B