从列表字典提取RecFrag数据导入MySQL及datetime序列化问题解决

一、直接跳过JSON转换，直接插入MySQL（推荐） #

既然最终目标是将数据存入MySQL，完全不需要先转JSON，直接提取数据后用参数化查询插入即可，同时自动处理datetime和NaN值：

import mysql.connector
from datetime import datetime
import pandas as pd  # 用于判断NaN

# 模拟你的原始采集数据
raw_data = ([{'TableNbr': 9, 'BegRecNbr': 9222359, 'TableName': b'Initial', 'IsOffset': 0, 'NbrOfRecs': 1, 'ByteOffset': None, 'RecFrag': [{'RecNbr': 9222359, 'TimeOfRec': datetime(2023, 4, 28, 1, 2, 17), 'Fields': {b'BattV': 11.994667053222656, b'PTemp_C': 23.523090362548828, b'BP_kPa': 100.7260971069336, b'Rain_mm': 0.0, b'Rain_mm_2': 0.0, b'AirTC': -39.08104705810547, b'AirTC2': 22.446022033691406, b'RH': 0.8168454170227051, b'RH2': 99.92742156982422, b'SlrkW': 0.000201545815798454, b'SlrkW_2': 0.0, b'SlrMJ': 4.0309163296115e-07, b'SlrMJ_2': 0.0, b'WS_ms': 0.0, b'WindDir': 83.36723327636719, b'Enc_RH': 58.200233459472656, b'T107_C': 26.024932861328125, b'SlrkW_3': pd.NA, b'Raw_mV': pd.NA, b'CS320_Temp': pd.NA, b'CS320_X': pd.NA, b'CS320_Y': pd.NA, b'CS320_Z': pd.NA, b'SlrMJ_3': pd.NA}}]}], 0)

# 提取核心数据
rec_frag = raw_data[0][0]['RecFrag'][0]
record_id = rec_frag['RecNbr']
record_time = rec_frag['TimeOfRec']
sensor_fields = rec_frag['Fields']

# 连接MySQL（替换为你的数据库信息）
db_conn = mysql.connector.connect(
    host='localhost',
    user='your_username',
    password='your_password',
    database='sensor_db'
)
cursor = db_conn.cursor()

# 构造插入SQL（字段名对应你的数据表结构）
insert_sql = """
INSERT INTO sensor_records (Record, TimeStamp, BattV, PTemp_C, BP_kPa, Rain_mm, Rain_mm_2, AirTC, AirTC2, RH, RH2, SlrkW, SlrkW_2, SlrMJ, SlrMJ_2, WS_ms, WindDir, Enc_RH, T107_C, SlrkW_3, Raw_mV, CS320_Temp, CS320_X, CS320_Y, CS320_Z, SlrMJ_3)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""

# 整理参数，将bytes类型的键转成字符串，同时把NaN替换为MySQL支持的NULL
def handle_nan(val):
    return None if pd.isna(val) else val

params = (
    record_id,
    record_time,
    handle_nan(sensor_fields[b'BattV']),
    handle_nan(sensor_fields[b'PTemp_C']),
    handle_nan(sensor_fields[b'BP_kPa']),
    handle_nan(sensor_fields[b'Rain_mm']),
    handle_nan(sensor_fields[b'Rain_mm_2']),
    handle_nan(sensor_fields[b'AirTC']),
    handle_nan(sensor_fields[b'AirTC2']),
    handle_nan(sensor_fields[b'RH']),
    handle_nan(sensor_fields[b'RH2']),
    handle_nan(sensor_fields[b'SlrkW']),
    handle_nan(sensor_fields[b'SlrkW_2']),
    handle_nan(sensor_fields[b'SlrMJ']),
    handle_nan(sensor_fields[b'SlrMJ_2']),
    handle_nan(sensor_fields[b'WS_ms']),
    handle_nan(sensor_fields[b'WindDir']),
    handle_nan(sensor_fields[b'Enc_RH']),
    handle_nan(sensor_fields[b'T107_C']),
    handle_nan(sensor_fields[b'SlrkW_3']),
    handle_nan(sensor_fields[b'Raw_mV']),
    handle_nan(sensor_fields[b'CS320_Temp']),
    handle_nan(sensor_fields[b'CS320_X']),
    handle_nan(sensor_fields[b'CS320_Y']),
    handle_nan(sensor_fields[b'CS320_Z']),
    handle_nan(sensor_fields[b'SlrMJ_3'])
)

# 执行插入并提交
cursor.execute(insert_sql, params)
db_conn.commit()

# 关闭连接
cursor.close()
db_conn.close()

二、解决JSON序列化datetime的问题 #

如果确实需要将数据转成JSON，可通过自定义序列化函数处理datetime类型：

import json
from datetime import datetime

# 自定义序列化函数，将datetime转为ISO标准字符串
def serialize_datetime(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    raise TypeError(f"Type {type(obj)} not serializable")

# 提取数据并构造字典
rec_frag = raw_data[0][0]['RecFrag'][0]
data_dict = {
    'Record': rec_frag['RecNbr'],
    'TimeStamp': rec_frag['TimeOfRec'],
    # 将bytes键转成字符串
    **{key.decode(): val for key, val in rec_frag['Fields'].items()}
}

# 转JSON时指定自定义序列化函数
json_data = json.dumps(data_dict, default=serialize_datetime)
print(json_data)

三、用Pandas简化数据处理与序列化 #

借助Pandas可以自动处理datetime和NaN的序列化，同时快速整理数据：

import pandas as pd

# 提取数据转为DataFrame
rec_frag = raw_data[0][0]['RecFrag'][0]
df = pd.DataFrame({
    'Record': [rec_frag['RecNbr']],
    'TimeStamp': [rec_frag['TimeOfRec']],
    **{key.decode(): [val] for key, val in rec_frag['Fields'].items()}
})

# 转JSON，自动处理datetime和NaN
json_data = df.to_json(orient='records', date_format='iso')
print(json_data)

# 也可以直接用Pandas写入MySQL
# df.to_sql('sensor_records', con=db_conn, if_exists='append', index=False)

内容的提问来源于stack exchange，提问作者Ale Durán