如何将Webhook接收字符串拆分后的列表按列标识追加至CSV文件？

我正在开发一款Flask应用，用于每隔几分钟接收Webhook传来的字符串。计划拆分该字符串，提取变量后追加至CSV文件实现数据累积，想知道如何将拆分后列表的第一部分设为列标题，第二部分设为对应值？

当前代码 #

# 从WebHook接收的字符串
string = 'https://example.com/test?Date_Event=2013-12-31T15:21:32Z&DeviceID=Item1&Variable=11.1'
# 移除不需要的URL部分
string1 = string.replace("https://example.com/test?","")
# 用&分割剩余字符串
res = string1.split('&')
# 循环提取列名和值
for x in res:
    datapoint = x.split('=')
    print(datapoint)

当前输出 #

['Date_Event', '2013-12-31T15:21:32Z']
['DeviceID', 'Item1']
['Variable', '11.1']

期望输出（CSV格式对应效果） #

Date_Event           | DeviceID | Variable
------------------------------------------
2013-12-31T15:21:32Z | Item1    | 11.1
...

解决方案 #

1. 更优雅的URL参数解析

不用手动replace和split，直接用Python标准库urllib.parse解析URL参数，能稳定处理特殊字符：

from urllib.parse import urlparse, parse_qs

string = 'https://example.com/test?Date_Event=2013-12-31T15:21:32Z&DeviceID=Item1&Variable=11.1'
parsed_url = urlparse(string)
params = parse_qs(parsed_url.query)
# 转换为键值对（parse_qs返回的是列表，取第一个元素）
data_dict = {k: v[0] for k, v in params.items()}

2. 分离表头与数据行

从字典中提取表头（键）和对应值（值）：

headers = list(data_dict.keys())
values = list(data_dict.values())

3. 写入CSV文件（支持追加）

使用Python内置的csv模块，判断文件是否存在：不存在则写入表头，存在则直接追加数据行，确保格式一致：

import csv
import os

csv_file = 'webhook_data.csv'

# 判断文件是否存在
file_exists = os.path.isfile(csv_file)

with open(csv_file, 'a', newline='', encoding='utf-8') as f:
    writer = csv.DictWriter(f, fieldnames=headers)
    # 首次写入时添加表头
    if not file_exists:
        writer.writeheader()
    # 写入当前数据行
    writer.writerow(data_dict)

整合到Flask接口

把逻辑放到Flask的路由中，处理Webhook请求：

from flask import Flask, request
from urllib.parse import parse_qs
import csv
import os

app = Flask(__name__)
CSV_FILE = 'webhook_data.csv'

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    # 假设Webhook的字符串在请求参数中，根据实际传入方式调整
    webhook_string = request.args.get('data')
    if not webhook_string:
        return "Missing data", 400
    
    # 解析参数
    parsed_params = parse_qs(webhook_string.split('?')[1])
    data_dict = {k: v[0] for k, v in parsed_params.items()}
    headers = list(data_dict.keys())

    # 写入CSV
    file_exists = os.path.isfile(CSV_FILE)
    with open(CSV_FILE, 'a', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=headers)
        if not file_exists:
            writer.writeheader()
        writer.writerow(data_dict)
    
    return "Data saved", 200

if __name__ == '__main__':
    app.run(debug=True)

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