Python3 SQLite3从JSON加载PowerTracker数据至数据表的优化问询

阿华AIGC实验室

2026-5-11

高效将JSON传感器数据导入SQLite DATA表的Python方案

核心思路：用`executemany()`实现批量插入

Python的sqlite3模块自带的executemany()方法完美解决你的需求——它能一次性批量插入多条数据，比循环调用execute()更高效、更符合Python风格。关键是先做好JSON字段与DATA表列的映射匹配，再把设备数据转换成适配表结构的格式。

步骤1：定义字段映射关系

先把JSON返回的字段和DATA表中需要用到的列一一对应（注意两者的命名差异）：

field_mapping = {
    "deviceid": "EUI64",
    "timestamp": "TimeStamp",
    "alias": "alias",
    "modid": "modelident",
    "model": "modelname",
    "signalstrength": "rssi",
    "voltage": "voltage",
    "current": "current",
    "frequency": "frequency",
    "powerfactor": "powerfactory",
    "activepower": "activepower",
    "apparentpower": "apparentpower",
    "mainenergy": "mainenergy",
    "negativeenergy": "nenergy"
}

注：只映射你需要用到的字段即可，未映射的列会默认填充NULL，完全适配你的DATA表结构

步骤2：转换JSON数据为插入格式

遍历devices列表，把每个设备的JSON字典转换成符合映射规则的格式，同时处理数据类型转换（比如JSON里的字符串转数值）：

def convert_device_data(device):
    converted = {}
    for json_key, db_col in field_mapping.items():
        value = device.get(json_key)
        # 根据表列类型做数据转换
        if db_col in ["TimeStamp", "rssi"]:
            converted[db_col] = int(value) if value else None
        elif db_col in ["voltage", "current", "frequency", "powerfactory", 
                       "activepower", "apparentpower", "mainenergy", "nenergy"]:
            converted[db_col] = float(value) if value else None
        else:
            converted[db_col] = value
    # 补充表中需要的默认值（比如devicetype，可根据你的实际需求调整）
    converted["devicetype"] = 0
    return converted

# 假设你已经获取并解析好JSON数据到data_dict变量
devices_data = [convert_device_data(dev) for dev in data_dict["devices"]]

步骤3：批量插入到DATA表

用命名占位符:column_name构建INSERT语句，结合executemany()完成批量插入，既清晰又避免字段顺序错误：

import sqlite3

# 连接到你的SQLite数据库
conn = sqlite3.connect("your_power_data.db")
cursor = conn.cursor()

# 动态生成INSERT语句（只包含需要插入的列）
columns = ", ".join(devices_data[0].keys())
placeholders = ", ".join([f":{col}" for col in devices_data[0].keys()])
insert_query = f"INSERT INTO DATA ({columns}) VALUES ({placeholders})"

# 执行批量插入
cursor.executemany(insert_query, devices_data)

# 提交事务并关闭连接
conn.commit()
conn.close()

额外优化建议

避免重复数据：如果担心同一时间戳的重复插入，可以给TimeStamp和EUI64添加联合唯一约束，或者把INSERT语句改成INSERT OR IGNORE INTO ...

错误处理：添加异常捕获确保数据安全：

try:
    cursor.executemany(insert_query, devices_data)
    conn.commit()
except sqlite3.Error as e:
    print(f"数据库插入出错: {e}")
    conn.rollback()
finally:
    conn.close()