forked from gary/ems
2
0
Fork 0
sun_ems/ems_c/kernel/kit_db.c

3815 lines
148 KiB
C
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

/*****************************************************************************
* @copyright 2024-2024, . POWER SUPPLY CO., LTD.
* @file kit_db.c
* @brief sqlite3数据库操作方法
* @author Gary
* @date 2024/09/10
* @remark 去掉sqlite3_get_table即将过期的方法补充了设备、协议和点位表的查询方法
*****************************************************************************/
#include <string.h>
#include <dirent.h>
#include <sys/stat.h>
#include <pthread.h>
#include "utextend.h"
#include "kit_log.h"
#include "kit_string.h"
#include "kit_db.h"
#define CONFIG_SQLITE_PATH "/opt/company_ems/db/sqlite3.db" // 配置数据库
#define BUSINESS_SQLITE_PATH "/opt/company_ems/db/business.db" // 业务数据库
static sqlite3 *config_db; // 全局变量,数据库句柄
static sqlite3 *business_db; // 全局变量,业务数据库句柄
pthread_mutex_t config_db_mutex = PTHREAD_MUTEX_INITIALIZER; // 全局变量,配置数据库线程锁
pthread_mutex_t business_db_mutex = PTHREAD_MUTEX_INITIALIZER; // 全局变量,业务数据库线程锁
// 设备配置信息icd
static UT_icd ems_dev_config_icd =
{
sizeof(ems_dev_config_t),
NULL,
NULL,
NULL,
};
// 设备信息的icd
static UT_icd dev_icd =
{
sizeof(dev_info_t), // 设备结构体大小
NULL, // 初始化函数
NULL, // 复制函数
NULL, // 销毁函数
};
// 点位信息icd
static UT_icd point_icd =
{
sizeof(point_t),
NULL,
NULL,
NULL,
};
// 时序数据的icd
static UT_icd timed_data_config_icd =
{
sizeof(timed_data_config_t),
NULL,
NULL,
NULL,
};
// 故障数据的icd
static UT_icd faults_data_config_icd =
{
sizeof(faults_data_config_t),
NULL,
NULL,
NULL,
};
// 北向配置信息icd
static UT_icd north_config_icd =
{
sizeof(north_config_t),
NULL,
NULL,
NULL,
};
// 削峰填谷日期配置信息icd
static UT_icd pv_date_config_icd =
{
sizeof(pv_date_config_t),
NULL,
NULL,
NULL,
};
// 设备拓扑结构信息icd
static UT_icd topology_icd =
{
sizeof(topology_t),
NULL,
NULL,
NULL,
};
static UT_icd advanced_setting_icd =
{
sizeof(advanced_setting_t),
NULL,
NULL,
NULL,
};
// MQTT断点记录信息icd
static UT_icd break_record_icd =
{
sizeof(break_record_t),
NULL,
NULL,
NULL,
};
// DIDO逻辑设置信息的icd
static UT_icd di_do_logic_set_icd =
{
sizeof(di_do_logic_set_t),
NULL,
NULL,
NULL,
};
/*****************************************************************************
* @brief 检查文件夹是否存在
* @param[in] path文件夹路径
* @return 存在-true不存在-false
*****************************************************************************/
static bool file_exists(const char *const path)
{
struct stat buf = {0};
if (-1 != stat(path, &buf) && S_ISDIR(buf.st_mode))
{
return true;
}
else
{
return false;
}
}
/*****************************************************************************
* @brief 拼接路径,将源路径拼接到目标路径的指定位置
* @param[in] dst: 目标路径字符串
* @param[in] offset: 拼接的起始位置(即从目标路径的何处开始拼接)
* @param[in] max_size: 目标路径的最大长度(缓冲区大小)
* @param[in] src: 源路径字符串
* @return 返回拼接后的路径长度;如果拼接失败,则返回 -1
*****************************************************************************/
static int path_cat(char *dst, int offset, int max_size, const char *src)
{
int len = strlen(src);
// 检查目标缓冲区是否足够容纳新拼接的路径
if (offset + len >= max_size)
{
return -1;
}
// 将源路径复制到目标路径的指定位置
strcpy(dst + offset, src);
return offset + len;
}
/*****************************************************************************
* @brief 读取文件内容并存储为字符串
* @param[in] path: 文件路径
* @param[out] sql: 指向读取的字符串的指针,内存由该函数分配,调用者需负责释放
* @return 0 表示成功,非 0 表示失败
*****************************************************************************/
static int read_file_string(const char *path, char **sql)
{
FILE *fp = fopen(path, "r");
if (fp == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "打开文件'%s'失败", path);
return 1; // 打开文件失败
}
// 获取文件大小
fseek(fp, 0, SEEK_END);
long size = ftell(fp);
rewind(fp);
// 分配足够的内存来存储文件内容
*sql = malloc(size + 1);
if (*sql == NULL)
{
fclose(fp);
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
return 1; // 内存分配失败
}
// 读取文件内容到内存中并检查读取结果
size_t read_size = fread(*sql, 1, size, fp);
if (read_size != size)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "读取文件内容失败");
free(*sql);
fclose(fp);
return 1; // 读取文件内容失败
}
(*sql)[size] = '\0'; // 添加字符串终止符
fclose(fp);
return 0;
}
/*****************************************************************************
* @brief 判断字符串是否以指定后缀结尾
* @param[in] str: 要检查的字符串
* @param[in] suffix: 指定的后缀
* @return 如果字符串以指定后缀结尾则返回 true否则返回 false
*****************************************************************************/
static bool ends_with(const char *str, const char *suffix)
{
// 如果字符串的长度小于后缀的长度,直接返回 false
if (strlen(str) < strlen(suffix))
{
return false;
}
// 比较字符串的结尾部分与后缀是否相等
return strcmp(str + strlen(str) - strlen(suffix), suffix) == 0;
}
/*****************************************************************************
* @brief 比较两个 schema 文件名,用于排序
* @param[in] a: 第一个文件名
* @param[in] b: 第二个文件名
* @return 比较结果,用于排序函数
*****************************************************************************/
static int schema_sort_cmp(const void *a, const void *b)
{
return strcmp(*(char **)a, *(char **)b);
}
/*****************************************************************************
* @brief 收集目录中的所有 SQL 文件
* @param[in] dir: 目录路径
* @return 包含所有 SQL 文件名的 UT_array 指针;如果失败,则返回 NULL
*****************************************************************************/
static UT_array *collect_schemas(const char *dir)
{
DIR *dirp = NULL;
struct dirent *dent = NULL;
UT_array *files = NULL;
// 打开目录
if ((dirp = opendir(dir)) == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "文件夹打开失败: %s", dir);
return NULL;
}
// 初始化 UT_array 用于存储文件名
utarray_new(files, &ut_str_icd);
// 遍历目录中的所有文件,筛选出以 ".sql" 结尾的文件
while (NULL != (dent = readdir(dirp)))
{
if (ends_with(dent->d_name, ".sql"))
{
char *file = strdup(dent->d_name); // strdup 自动分配并拷贝字符串
if (file == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "strdup 失败");
continue;
}
// 将副本指针推入 UT_array
utarray_push_back(files, &file);
}
}
// 关闭目录
closedir(dirp);
return files;
}
/*****************************************************************************
* @brief 执行目录中的所有 SQL 文件
* @param[in] dir: 存储 SQL 文件的目录路径
* @param[in] db: 数据库句柄
* @return 0 表示成功,非 0 表示失败
*****************************************************************************/
static int execute_sql_files(const char *dir, sqlite3 *db)
{
char *sql = NULL;
int ret = 0;
if (!file_exists(dir))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "sql 文件 `%s` 不存在", dir);
return 1;
}
// 获取目录下的所有 SQL 文件
UT_array *files = collect_schemas(dir);
if (NULL == files)
{
return 1; // 目录读取失败
}
// 对文件名进行排序
utarray_sort(files, schema_sort_cmp);
// 遍历每个 SQL 文件
utarray_foreach(files, char **, file)
{
char *path = NULL;
path = calloc(PATH_MAX_SIZE, sizeof(char));
if (NULL == path)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
return 1;
}
// 拼接文件路径
if (PATH_MAX_SIZE <= path_cat(path, 0, PATH_MAX_SIZE, dir) ||
PATH_MAX_SIZE <= path_cat(path, strlen(dir), PATH_MAX_SIZE, *file))
{ // 修正了路径拼接问题
KITLOG(LOG_KIT_EN, ERROR_EN, "路径过长: %s", path);
free(path);
return 1;
}
// 读取 SQL 文件内容
ret = read_file_string(path, &sql);
if (0 != ret)
{
free(path);
continue; // 读取失败,跳过该文件
}
// 执行 SQL 语句
char *err_msg = NULL;
ret = sqlite3_exec(db, sql, NULL, NULL, &err_msg);
if (SQLITE_OK != ret)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "执行 %s 失败: (%d)%s", path, ret, err_msg);
sqlite3_free(err_msg);
ret = 1;
free(path);
free(sql);
continue;
}
free(path);
free(sql);
}
// 释放文件列表内存
utarray_free(files);
return ret;
}
/*****************************************************************************
* @brief 初始化 SQLite 数据库
* @param[in] schema_dir: 存储 SQL 文件的目录路径
* @param[in] db: 待初始化数据库
* @return 0 表示成功1 表示失败
*****************************************************************************/
static int init_db_handle(const char *schema_dir, sqlite3 *db)
{
int ret = 0;
// 设置忙时超时时间
ret = sqlite3_busy_timeout(db, 120 * 1000);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "设置忙时超时失败: %s", sqlite3_errmsg(db));
sqlite3_close(db);
return 1;
}
// 启用外键支持
ret = sqlite3_exec(db, "PRAGMA foreign_keys=ON", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "数据库外键启动失败: %s", sqlite3_errmsg(db));
sqlite3_close(db);
return 1;
}
// 设置 WAL 模式
ret = sqlite3_exec(db, "PRAGMA journal_mode=WAL", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "数据库 journal_mode WAL 启动失败: %s", sqlite3_errmsg(db));
sqlite3_close(db);
return 1;
}
#ifdef INIT_DB
// 执行目录中的所有 SQL 文件
ret = execute_sql_files(schema_dir, db);
if (ret != 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "执行 SQL 文件失败");
sqlite3_close(db);
return 1;
}
#endif
return 0;
}
/*****************************************************************************
* @brief 根据北向上报下发设备Id获取点位信息
* @param[in] upDisDevDbId: 上报下发设备Id
* @param[out] p_upDisDev: 用于存储北向配置信息的指针, 包含上报点位和下发设备信息
* @return 0-成功 1-失败
*****************************************************************************/
static int kit_get_updis_dev_point(int upDisDevDbId, up_dis_dev_t *p_upDisDev)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
char *query = "SELECT udd.dev_type, d.code, udp.cloud_code, "
"udp.cloud_name, udp.point_code "
"FROM up_dis_dev udd "
"LEFT JOIN device d ON udd.dev_id = d.id "
"LEFT JOIN up_dis_point udp ON udd.id = udp.up_dis_id "
"WHERE udd.id = ?";
if (sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL) != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询失败: %s", sqlite3_errmsg(config_db));
return 1;
}
if (sqlite3_bind_int(stmt, 1, upDisDevDbId) != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定upDisDevDbId失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1;
}
int step;
while ((step = sqlite3_step(stmt)) == SQLITE_ROW)
{
up_dis_point_t *temp = realloc(p_upDisDev->upDisPointArr, (p_upDisDev->upDisPointNum + 1) * sizeof(up_dis_point_t));
if (temp == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
ret = 1;
break;
}
p_upDisDev->upDisPointArr = temp;
up_dis_point_t *point = &p_upDisDev->upDisPointArr[p_upDisDev->upDisPointNum++];
point->devType = (dev_type_e)sqlite3_column_int(stmt, 0);
point->devId = sqlite3_column_int(stmt, 1);
snprintf((char *)point->cloudCode, MAX_CODE_BUF_LEN, "%s", sqlite3_column_text(stmt, 2));
snprintf((char *)point->cloudName, MAX_NAME_BUF_LEN, "%s", sqlite3_column_text(stmt, 3));
point->pointId = sqlite3_column_int(stmt, 4);
}
if (step != SQLITE_DONE && ret == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询上报下发点位信息失败: %s", sqlite3_errmsg(config_db));
ret = 1;
}
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 根据北向配置ID获取上报下发配置信息
* @param[in] northConfigDbId: 北向配置表数据库ID
* @param[out] northConfig: 用于存储北向配置信息的指针, 包含上报点位和下发点位信息
* @return 0-成功 1-失败
*****************************************************************************/
static int kit_get_north_config_by_id(int northConfigDbId, north_config_t *northConfig)
{
int ret = 0;
int upDisDevId;
sqlite3_stmt *stmt = NULL;
char *query = "SELECT udd.up_dis_config_type, udd.dev_type, d.code, "
"udd.id "
"FROM up_dis_dev udd "
"LEFT JOIN device d ON udd.dev_id = d.id "
"WHERE udd.north_config_id = ?";
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL)) // config_db: 全局数据库连接
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1;
}
// 绑定参数到预编译语句
if (SQLITE_OK != sqlite3_bind_int(stmt, 1, northConfigDbId))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 northConfigId 失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1; // 返回错误代码
}
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step)
{
// 获取上报下发配置类型
up_dis_config_type_e upDisConfigType = sqlite3_column_int(stmt, 0);
// 根据配置类型分配内存并存储点位信息
switch (upDisConfigType)
{
case kUp_Dis_Cfg_Type_Up: // 上报点位
northConfig->upDevNum++;
// 分配内存以存储新的上报设备信息
up_dis_dev_t *tempUp = (up_dis_dev_t *)realloc(northConfig->upDevArr, northConfig->upDevNum * sizeof(up_dis_dev_t));
if (tempUp == NULL)
{
// 内存分配失败,记录错误信息并返回
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
ret = 1;
goto error;
}
northConfig->upDevArr = tempUp;
// 获取上报配置信息
northConfig->upDevArr[northConfig->upDevNum - 1].devType = (dev_type_e)sqlite3_column_int(stmt, 1); // 设备类型
northConfig->upDevArr[northConfig->upDevNum - 1].devId = sqlite3_column_int(stmt, 2); // 设备编号
upDisDevId = sqlite3_column_int(stmt, 3); // 上报下发设备Id
kit_get_updis_dev_point(upDisDevId, &northConfig->upDevArr[northConfig->upDevNum - 1]);
break;
case kUp_Dis_Cfg_Type_Dis_Point: // 下发控制点位
northConfig->disDevNum++;
// 分配内存以存储新的下发设备信息
up_dis_dev_t *tempDis = (up_dis_dev_t *)realloc(northConfig->disDevArr, northConfig->disDevNum * sizeof(up_dis_dev_t));
if (tempDis == NULL)
{
// 内存分配失败,记录错误信息并返回
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
ret = 1;
goto error;
}
northConfig->disDevArr = tempDis;
// 获取下发配置信息
northConfig->disDevArr[northConfig->disDevNum - 1].devType = (dev_type_e)sqlite3_column_int(stmt, 1); // 设备类型
northConfig->disDevArr[northConfig->disDevNum - 1].devId = sqlite3_column_int(stmt, 2); // 设备编号
upDisDevId = sqlite3_column_int(stmt, 3); // 上报下发设备Id
kit_get_updis_dev_point(upDisDevId, &northConfig->disDevArr[northConfig->upDevNum - 1]);
break;
default:
KITLOG(LOG_KIT_EN, ERROR_EN, "未知的上报下发配置类型: %d", upDisConfigType);
ret = 1;
break;
}
if (ret != 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "填充上报下发信息失败");
sqlite3_finalize(stmt);
return 1;
}
step = sqlite3_step(stmt);
}
if (step != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询上报下发配置信息失败: %s", sqlite3_errmsg(config_db));
ret = 1;
}
error:
// 清理资源
sqlite3_finalize(stmt);
return ret;
}
/* ############################################################################### 配置数据库操作 ############################################################################### */
/*****************************************************************************
* @brief 初始化 SQLite 数据库
* @param[in] schema_dir: 存储 SQL 文件的目录路径
* @return 0-成功 1-失败
*****************************************************************************/
int kit_init_db(const char *schema_dir)
{
int ret = 0;
// 打开配置数据库,如果不存在则创建
// ret = sqlite3_open(CONFIG_SQLITE_PATH, &config_db);
ret = sqlite3_open_v2(CONFIG_SQLITE_PATH, &config_db, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_WAL, NULL);
if (SQLITE_OK != ret)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "打开数据库 `%s` 失败: %s", CONFIG_SQLITE_PATH, sqlite3_errstr(ret));
return 1;
}
char config_schema_dir[PATH_MAX_SIZE]; // 假设路径不会超过 PATH_MAX_SIZE 字符
snprintf(config_schema_dir, PATH_MAX_SIZE, "%s/config/", schema_dir);
ret = init_db_handle(config_schema_dir, config_db);
if (0 != ret)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "初始化数据库 `%s` 失败", CONFIG_SQLITE_PATH);
return 1;
}
// 打开业务数据库,如果不存在则创建
// ret = sqlite3_open(BUSINESS_SQLITE_PATH, &business_db);
ret = sqlite3_open_v2(BUSINESS_SQLITE_PATH, &business_db, SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE | SQLITE_OPEN_WAL, NULL);
if (SQLITE_OK != ret)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "打开数据库 `%s` 失败: %s", BUSINESS_SQLITE_PATH, sqlite3_errstr(ret));
return 1;
}
char business_schema_dir[PATH_MAX_SIZE]; // 假设路径不会超过 PATH_MAX_SIZE 字符
snprintf(business_schema_dir, PATH_MAX_SIZE, "%s/business/", schema_dir);
ret = init_db_handle(business_schema_dir, business_db);
if (0 != ret)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "初始化数据库 `%s` 失败", BUSINESS_SQLITE_PATH);
return 1;
}
#ifdef INIT_DB
kit_sync_dev_data(); // 同步设备信息
kit_sync_point_data(); // 同步点位信息
#endif
return 0;
}
/*****************************************************************************
* @brief 释放 SQLite 数据库连接
* @return 0 表示成功1 表示失败
*****************************************************************************/
int kit_uninit_db()
{
if (config_db != NULL)
{
sqlite3_close(config_db);
config_db = NULL;
return 0;
}
if (business_db != NULL)
{
sqlite3_close(business_db);
business_db = NULL;
return 0;
}
return 1;
}
/*****************************************************************************
* @brief 从 SQLite 数据库中获取数据表数据。使用准备好的语句Prepared Statement从 SQLite数据库中
* 获取数据表数据,避免使用 `sqlite3_get_table`,因为它是一个过时的接口,存在安全和性能问题。
* @param[out] outData: 用于存储查询结果的结构体指针。
* @param[in] tableName: 数据表名称。
* @param[in] columnsName: 要查询的列名数组,可以为 NULL 或空数组,表示查询所有列。
* @param[in] columnNum: 要查询的列名数量。
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_db_data(db_table_data_t *outData, const char *tableName, const char *columnsName[], int columnNum)
{
sqlite3_stmt *stmt = NULL;
int ret = 0;
char *sql = NULL;
char *sql_columns = NULL;
// 构建 SQL 查询语句
if (columnsName == NULL || columnNum == 0)
{
sql = sqlite3_mprintf("SELECT * FROM %s", tableName);
}
else
{
sql_columns = kit_join_text(columnNum, columnsName, ",");
if (sql_columns == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "构建列名失败");
return 1;
}
sql = sqlite3_mprintf("SELECT %s FROM %s", sql_columns, tableName);
free(sql_columns); // 释放拼接好的列名字符串
}
if (sql == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "构建 SQL 语句失败: sqlite3_mprintf 失败");
return 1;
}
// 准备 SQL 语句
sqlite3_free(sql); // 释放 SQL 字符串
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备语句失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt); // 结束语句
return 1;
}
// 获取查询结果
outData->nrow = 0;
outData->ncolumn = sqlite3_column_count(stmt); // 获取列数
// 如果有列,继续处理
if (outData->ncolumn > 0)
{
// 先检查有多少行
while ((ret = sqlite3_step(stmt)) == SQLITE_ROW)
{
outData->nrow++;
}
// 如果没有找到行
if (outData->nrow == 0)
{
KITLOG(LOG_KIT_EN, INFO_EN, "没有找到匹配的行");
sqlite3_finalize(stmt);
return 0;
}
// 为结果表分配内存
outData->dbresult = (char **)sqlite3_malloc(outData->nrow * outData->ncolumn * sizeof(char *));
if (outData->dbresult == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "分配结果表内存失败");
sqlite3_finalize(stmt);
return 1;
}
// 重置语句,重新开始读取
sqlite3_reset(stmt);
// 读取数据行并存储结果
int row = 0;
while ((ret = sqlite3_step(stmt)) == SQLITE_ROW)
{
for (int i = 0; i < outData->ncolumn; i++)
{
const char *text = (const char *)sqlite3_column_text(stmt, i);
outData->dbresult[row * outData->ncolumn + i] = text ? sqlite3_mprintf("%s", text) : NULL;
}
row++;
}
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "执行语句失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1;
}
}
// 结束语句
sqlite3_finalize(stmt);
return 0;
}
/*****************************************************************************
* @brief 释放 `db_table_data_t` 结构体中分配的内存。
* 该函数用于释放 `kitGetDbData` 函数获取的查询结果数据所占用的内存。
* 它会依次释放 `dbresult`、`errmsg` 和 `db` 指针指向的内存。
* @param data 指向 `db_table_data_t` 结构体的指针。
*****************************************************************************/
void kit_free_table_data(db_table_data_t *data)
{
if (data == NULL)
{
// 如果 `data` 指针为空,则直接返回。
return;
}
// 释放 `dbresult` 指向的内存,如果 `dbresult` 不为空。
if (data->dbresult != NULL)
{
sqlite3_free_table(data->dbresult);
data->dbresult = NULL;
}
// 释放 `errmsg` 指向的内存,如果 `errmsg` 不为空。
if (data->errmsg != NULL)
{
sqlite3_free(data->errmsg);
data->errmsg = NULL;
}
// 函数执行完毕,返回。
return;
}
/*****************************************************************************
* @brief: 执行多条 SQL 语句,函数内关闭连接。
* 该函数用于执行多个 SQL 语句,并处理执行过程中的错误。
* 它会依次执行传入的 SQL 语句数组,并在执行失败时记录错误信息。
* 函数执行完毕后会关闭数据库连接。
* @param[in] tableName: 数据表名称(未使用,可以忽略)。
* @param[in] sql: 包含多个 SQL 语句的字符串数组。
* @param[in] sql_num: SQL 语句的数量。
* @return 0-成功 1-失败
*****************************************************************************/
int kit_exec_db_data(char *tableName, char *sql[], int sql_num)
{
sqlite3 *db = NULL; // 数据库连接指针
int ret = 0; // sqlite3_exec 函数返回值
char *zErrMsg = 0; // 错误信息指针
// 循环执行每个 SQL 语句
for (int i = 0; i < sql_num; i++)
{
// printf("%s\n",sql[i]); // 可选:打印每个 SQL 语句
ret = sqlite3_exec(db, sql[i], NULL, 0, &zErrMsg);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "执行 SQL 语句[%d] 失败: %s.", i, zErrMsg);
sqlite3_free(zErrMsg); // 释放错误信息内存
return 1;
}
}
sqlite3_free(zErrMsg); // 释放错误信息内存
return 0; // 返回执行结果标志
}
/*****************************************************************************
* @brief 从数据库查询电站数据
* @param[in] p_station电站信息的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_station_db_data(station_t *p_station)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT name, address, lng, lat, time_zone FROM station DESC LIMIT 1");
if (query == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
return 1;
}
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1; // 返回错误代码
}
int step = sqlite3_step(stmt);
if (step == SQLITE_ROW)
{
snprintf((char *)p_station->name, MAX_NAME_BUF_LEN, "%s", (const char *)sqlite3_column_text(stmt, 0)); // 电站名称
snprintf((char *)p_station->address, MAX_ADDR_BUF_LEN, "%s", (const char *)sqlite3_column_text(stmt, 1)); // 电站地址
p_station->lng = sqlite3_column_double(stmt, 2); // 经度
p_station->lat = sqlite3_column_double(stmt, 3); // 纬度
snprintf((char *)p_station->timeZone, MAX_CODE_BUF_LEN, "%s", (const char *)sqlite3_column_text(stmt, 4)); // 时区
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return (step == SQLITE_ROW) ? 0 : 1; // 返回成功或失败代码
}
/*****************************************************************************
* @brief 从数据库获取设备信息
* @param[out] devPointInfo: 用于存储设备信息的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_dev_db_data(UT_array **devPointInfo)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *query; // 查询语句
utarray_new(*devPointInfo, &dev_icd); // 创建新的 UT_array 用于存储设备点位信息
// 构建查询语句
query = "SELECT id, type, code, protocol_type, name, "
"sort_order, template_id "
"FROM device WHERE 1=1 "; // 基础查询语句
// 添加结果排序
query = sqlite3_mprintf("%s ORDER BY type, protocol_type, id ASC", query); // 按 id 升序排序
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从数据库中获取每一列的值
dev_info_t dev_info = {0}; // 初始化设备点位结构体
dev_info.devDbId = sqlite3_column_int(stmt, 0);
dev_info.devType = sqlite3_column_int(stmt, 1);
dev_info.devId = sqlite3_column_int(stmt, 2);
dev_info.protocolType = sqlite3_column_int(stmt, 3);
snprintf((char *)dev_info.devName, MAX_NAME_BUF_LEN, "%s", (const char *)sqlite3_column_text(stmt, 4));
dev_info.sortOrder = sqlite3_column_int(stmt, 5);
dev_info.templateId = sqlite3_column_int(stmt, 6);
// 将设备点位信息添加到 UT_array 中
utarray_push_back(*devPointInfo, &dev_info);
// 继续处理下一行
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return 0; // 成功返回
error:
sqlite3_finalize(stmt); // 确保释放预编译语句
utarray_free(*devPointInfo); // 释放 UT_array
devPointInfo = NULL; // 设置为 NULL
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 将 config_db 中的 device 表数据复制到 business_db 中的 device 表
* @return 0-成功 1-失败
*****************************************************************************/
int kit_sync_dev_data()
{
sqlite3_stmt *stmt_select = NULL; // 用于 SELECT 语句
sqlite3_stmt *stmt_insert = NULL; // 用于 INSERT 语句
const char *sql_delete;
const char *sql_select;
const char *sql_insert;
int ret;
pthread_mutex_lock(&config_db_mutex); // 给配置库操作上锁
pthread_mutex_lock(&business_db_mutex); // 给业务库操作上锁
// 清空 business_db 中的 device 表
sql_delete = "DELETE FROM device";
ret = sqlite3_exec(business_db, sql_delete, NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "清空 business_db 中的 device 表失败: %s", sqlite3_errmsg(business_db));
goto error;
}
// 从 config_db 中选择数据
sql_select = "SELECT id, sn, type, code, name, protocol_type, template_id, sort_order, remark, status FROM device";
ret = sqlite3_prepare_v2(config_db, sql_select, -1, &stmt_select, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", sql_select, sqlite3_errmsg(config_db));
goto error;
}
// 准备插入语句
sql_insert = "INSERT INTO device (id, sn, type, code, name, protocol_type, template_id, sort_order, remark, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
ret = sqlite3_prepare_v2(business_db, sql_insert, -1, &stmt_insert, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备插入语句 `%s` 失败: %s", sql_insert, sqlite3_errmsg(business_db));
goto error;
}
// 循环插入数据
while ((ret = sqlite3_step(stmt_select)) == SQLITE_ROW)
{
// 绑定参数
sqlite3_bind_int(stmt_insert, 1, sqlite3_column_int(stmt_select, 0));
sqlite3_bind_text(stmt_insert, 2, (const char *)sqlite3_column_text(stmt_select, 1), -1, SQLITE_STATIC);
sqlite3_bind_int(stmt_insert, 3, sqlite3_column_int(stmt_select, 2));
sqlite3_bind_int(stmt_insert, 4, sqlite3_column_int(stmt_select, 3));
sqlite3_bind_text(stmt_insert, 5, (const char *)sqlite3_column_text(stmt_select, 4), -1, SQLITE_STATIC);
sqlite3_bind_int(stmt_insert, 6, sqlite3_column_int(stmt_select, 5));
sqlite3_bind_int(stmt_insert, 7, sqlite3_column_int(stmt_select, 6));
sqlite3_bind_int(stmt_insert, 8, sqlite3_column_int(stmt_select, 7));
sqlite3_bind_text(stmt_insert, 9, (const char *)sqlite3_column_text(stmt_select, 8), -1, SQLITE_STATIC);
sqlite3_bind_int(stmt_insert, 10, sqlite3_column_int(stmt_select, 9));
// 执行插入
ret = sqlite3_step(stmt_insert);
sqlite3_reset(stmt_insert); // 重置语句以进行下一次插入
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
goto error;
}
}
// 检查是否有错误发生在sqlite3_step(stmt_select)中
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询数据失败: %s", sqlite3_errmsg(config_db));
goto error;
}
// 清理资源
sqlite3_finalize(stmt_select);
sqlite3_finalize(stmt_insert);
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 0;
error:
sqlite3_finalize(stmt_select);
sqlite3_finalize(stmt_insert);
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 1;
}
/*****************************************************************************
* @brief 从数据库获取协议信息。
* 该函数从数据库中获取与指定设备 ID 关联的协议信息。
* 它使用准备好的语句Prepared Statement来执行查询并绑定设备 ID 参数。
* 函数会返回获取到的协议信息,并处理查询过程中的错误。
* @param[in] templateType: 模板类别0设备配置, 1北向配置
* @param[in] relatedId: 关联数据库 ID。
* @param[out] protoContent: 用于存储协议信息的字符数组。
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_protocol_db_data(const template_type_e templateType, const uint64_t relatedId, char protoContent[MAX_CONFIG_CONTENT_LEN])
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
char *query = sqlite3_mprintf("SELECT content FROM protocol WHERE 1=1 AND template_type=? AND related_id=? LIMIT 1");
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query); // 释放查询字符串
return 1; // 返回错误代码
}
// 绑定参数到预编译语句
if (SQLITE_OK != sqlite3_bind_int(stmt, 1, templateType))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 template_type 失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
sqlite3_free(query); // 释放查询字符串
return 1; // 返回错误代码
}
if (SQLITE_OK != sqlite3_bind_int64(stmt, 2, relatedId))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 related_id 失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
sqlite3_free(query); // 释放查询字符串
return 1; // 返回错误代码
}
// 执行查询并处理结果
int step = sqlite3_step(stmt);
if (step == SQLITE_ROW || step == SQLITE_DONE)
{
snprintf(protoContent, MAX_CONFIG_CONTENT_LEN, "%s", sqlite3_column_text(stmt, 0));
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
sqlite3_free(query); // 释放查询字符串
return 1; // 返回错误代码
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
sqlite3_free(query); // 释放查询字符串
return 0; // 成功返回
}
/*****************************************************************************
* @brief 从数据库获取点位信息
* @param[in] templateId: 模板Id
* @param[out] points: 用于存储点位信息的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_point_db_data(const uint64_t templateId, UT_array **points)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *query; // 查询语句
utarray_new(*points, &point_icd); // 创建新的 UT_array 用于存储设备点位信息
// 构建查询语句
query = "SELECT group_type, code, name, data_type, "
"attribute, func, reg_addr, bit_pos, bit_num, endian, "
"precision, ratio, offset, is_persisted, storage_interval, "
"mutate_bound, default_value, min_value, max_value, dev_type,id "
"FROM point WHERE 1=1 "; // 基础查询语句
// 检查并添加设备Id条件
query = sqlite3_mprintf("%s AND template_id=?", query); // 动态添加设备Id条件
// 添加结果排序
query = sqlite3_mprintf("%s ORDER BY func, reg_addr ASC", query); // 按 id 升序排序
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 绑定参数到预编译语句
int param_index = 1; // 参数索引
if (SQLITE_OK != sqlite3_bind_int64(stmt, param_index++, templateId))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 `%s` 的参数 dev_id 失败: %s", query, sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// printf("%s\n",query);
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从数据库中获取每一列的值
point_t point_info = {0}; // 初始化设备点位结构体
point_info.groupType = (point_group_type_e)sqlite3_column_int(stmt, 0);
point_info.pointId = (uint16_t)sqlite3_column_int(stmt, 1);
snprintf((char *)point_info.pointName, MAX_POINT_NAME_LEN, "%s", (const char *)sqlite3_column_text(stmt, 2));
point_info.dataType = (data_type_e)sqlite3_column_int(stmt, 3);
point_info.attribute = (point_attribute_e)sqlite3_column_int(stmt, 4);
point_info.func = (uint32_t)sqlite3_column_int(stmt, 5);
point_info.regAddr = (uint16_t)sqlite3_column_int(stmt, 6);
point_info.bitPos = (uint8_t)sqlite3_column_int(stmt, 7);
point_info.bitNum = (uint8_t)sqlite3_column_int(stmt, 8);
point_info.endian = (data_endian_e)sqlite3_column_int(stmt, 9);
point_info.precision = sqlite3_column_double(stmt, 10);
point_info.ratio = sqlite3_column_double(stmt, 11);
point_info.offset = sqlite3_column_double(stmt, 12);
point_info.isPersisted = (uint8_t)sqlite3_column_int(stmt, 13);
point_info.storageInterval = (storage_interval_type_e)sqlite3_column_int(stmt, 14);
point_info.mutateBound = sqlite3_column_double(stmt, 15);
point_info.defaultValue = sqlite3_column_double(stmt, 16);
point_info.minValue = sqlite3_column_double(stmt, 17);
point_info.maxValue = sqlite3_column_double(stmt, 18);
point_info.devType = sqlite3_column_int(stmt, 19);
point_info.pointDbId = sqlite3_column_int(stmt, 20);
// point_info.devDbId = sqlite3_column_int(stmt, 20);
// 将设备点位信息添加到 UT_array 中
utarray_push_back(*points, &point_info);
// 继续处理下一行
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return 0; // 成功返回
error:
sqlite3_finalize(stmt); // 确保释放预编译语句
utarray_free(*points); // 释放 UT_array
points = NULL; // 设置为 NULL
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 从 config_db 获取点位信息并同步到 business_db
* @param[out] 无
* @return 0-成功 1-失败
*****************************************************************************/
int kit_sync_point_data()
{
sqlite3_stmt *stmt_config = NULL; // config_db 中的预编译语句句柄
sqlite3_stmt *stmt_business = NULL; // business_db 中的预编译语句句柄
const char *query_config; // 从 config_db 查询点位信息的 SQL 语句
const char *query_clear = "DELETE FROM point"; // 清空 business_db 中的 point 表的 SQL 语句
int step; // 查询结果步骤
pthread_mutex_lock(&config_db_mutex); // 给配置库操作上锁
pthread_mutex_lock(&business_db_mutex); // 给业务库操作上锁
// 清空 business_db 中的 point 表
if (SQLITE_OK != sqlite3_exec(business_db, query_clear, NULL, NULL, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "清空 business_db 中的 point 表失败: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 1; // 处理错误
}
// 构建查询语句从 config_db 获取所有点位信息
query_config = "SELECT template_id, protocol_type, dev_type, group_type, code, name, "
"data_type, attribute, func, reg_addr, bit_pos, bit_num, endian, precision, "
"ratio, offset, is_persisted, storage_interval, mutate_bound, default_value, "
"min_value, max_value, unit, is_show FROM point ORDER BY id ASC";
// 准备从 config_db 执行查询语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query_config, -1, &stmt_config, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query_config, sqlite3_errmsg(config_db));
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 1; // 处理错误
}
// 准备向 business_db 插入数据的 SQL 语句
const char *insert_query = "INSERT INTO point (template_id, protocol_type, dev_type, group_type, "
"code, name, data_type, attribute, func, reg_addr, bit_pos, bit_num, endian, "
"precision, ratio, offset, is_persisted, storage_interval, mutate_bound, "
"default_value, min_value, max_value, unit, is_show) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
// 准备向 business_db 插入数据
if (SQLITE_OK != sqlite3_prepare_v2(business_db, insert_query, -1, &stmt_business, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备插入 `%s` 失败: %s", insert_query, sqlite3_errmsg(business_db));
sqlite3_finalize(stmt_config); // 确保释放预编译语句
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 1; // 处理错误
}
// 执行查询并插入数据
step = sqlite3_step(stmt_config);
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从 config_db 中获取每一列的值
int param_index = 1; // 参数索引
// 绑定参数到插入语句
if (SQLITE_OK != sqlite3_bind_int64(stmt_business, param_index++, sqlite3_column_int64(stmt_config, 0))) // template_id
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 1))) // protocol_type
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 2))) // dev_type
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 3))) // group_type
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 4))) // code
goto error;
if (SQLITE_OK != sqlite3_bind_text(stmt_business, param_index++, sqlite3_column_text(stmt_config, 5), -1, SQLITE_STATIC)) // name
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 6))) // data_type
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 7))) // attribute
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 8))) // func
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 9))) // reg_addr
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 10))) // bit_pos
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 11))) // bit_num
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 12))) // endian
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 13))) // precision
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 14))) // ratio
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 15))) // offset
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 16))) // is_persisted
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 17))) // storage_interval
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 18))) // mutate_bound
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 19))) // default_value
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 20))) // min_value
goto error;
if (SQLITE_OK != sqlite3_bind_double(stmt_business, param_index++, sqlite3_column_double(stmt_config, 21))) // max_value
goto error;
if (SQLITE_OK != sqlite3_bind_text(stmt_business, param_index++, sqlite3_column_text(stmt_config, 22), -1, SQLITE_STATIC)) // unit
goto error;
if (SQLITE_OK != sqlite3_bind_int(stmt_business, param_index++, sqlite3_column_int(stmt_config, 23))) // is_show
goto error;
// 执行插入操作
if (SQLITE_DONE != sqlite3_step(stmt_business))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
goto error;
}
// 重置插入语句以便于重新绑定参数
sqlite3_reset(stmt_business);
// 继续处理下一行
step = sqlite3_step(stmt_config);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt_config); // 释放 config_db 预编译语句
sqlite3_finalize(stmt_business); // 释放 business_db 预编译语句
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 0; // 成功返回
error:
sqlite3_finalize(stmt_config); // 确保释放预编译语句
sqlite3_finalize(stmt_business); // 确保释放预编译语句
pthread_mutex_unlock(&config_db_mutex); // 给配置库操作释放锁
pthread_mutex_unlock(&business_db_mutex); // 给业务库操作释放锁
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 从数据库获取读取时序数据配置
* @param[out] timedDataConfigs: 用于指定存储间隔类型的的时序数据配置的数据
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_tdata_config(UT_array **timedDataConfigs)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT p.id, p.code, p.name, p.storage_interval, d.name, d.type, d.id, d.code FROM point p LEFT JOIN device d ON p.template_id = d.template_id ORDER BY p.storage_interval ASC");
int ret;
ret = sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1;
}
utarray_new(*timedDataConfigs, &timed_data_config_icd);
while ((ret = sqlite3_step(stmt)) == SQLITE_ROW)
{
timed_data_config_t data_info = {0};
data_info.pointDbId = sqlite3_column_int(stmt, 0);
data_info.pointId = sqlite3_column_int(stmt, 1);
snprintf((char *)data_info.pointName, MAX_POINT_NAME_LEN, "%s", (const char *)sqlite3_column_text(stmt, 2));
data_info.storageInterval = (storage_interval_type_e)sqlite3_column_int(stmt, 3);
snprintf((char *)data_info.devName, MAX_NAME_BUF_LEN, "%s", (const char *)sqlite3_column_text(stmt, 4));
data_info.devType = sqlite3_column_int(stmt, 5);
data_info.devDbId = sqlite3_column_int(stmt, 6);
data_info.devId = sqlite3_column_int(stmt, 7);
utarray_push_back(*timedDataConfigs, &data_info);
}
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
sqlite3_free(query);
utarray_free(*timedDataConfigs);
*timedDataConfigs = NULL;
return 1;
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return 0;
}
/*****************************************************************************
* @brief 从数据库获取故障点位配置
* @param[out] faultsDataConfigs: 用于存储故障配置点位的数据
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_faults_config(UT_array **faultsDataConfigs)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT p.id, p.code, p.name, d.name, d.type, d.id, d.code FROM point p LEFT JOIN device d ON p.template_id = d.template_id WHERE p.group_type = ? ORDER BY p.storage_interval ASC");
int ret;
ret = sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1;
}
// 绑定参数
if (SQLITE_OK != sqlite3_bind_int(stmt, 1, kPoint_SOE))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 group_type 失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1; // 返回错误代码
}
utarray_new(*faultsDataConfigs, &faults_data_config_icd);
while ((ret = sqlite3_step(stmt)) == SQLITE_ROW)
{
timed_data_config_t data_info = {0};
data_info.pointDbId = sqlite3_column_int(stmt, 0);
data_info.pointId = sqlite3_column_int(stmt, 1);
snprintf((char *)data_info.pointName, MAX_POINT_NAME_LEN, "%s", (const char *)sqlite3_column_text(stmt, 2));
snprintf((char *)data_info.devName, MAX_NAME_BUF_LEN, "%s", (const char *)sqlite3_column_text(stmt, 3));
data_info.devType = sqlite3_column_int(stmt, 4);
data_info.devDbId = sqlite3_column_int(stmt, 5);
data_info.devId = sqlite3_column_int(stmt, 6);
utarray_push_back(*faultsDataConfigs, &data_info);
}
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
sqlite3_free(query);
utarray_free(*faultsDataConfigs);
*faultsDataConfigs = NULL;
return 1;
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return 0;
}
/*****************************************************************************
* @brief 从数据库获取EMS设备配置信息
* @param[out] emsDevConfigs: 用于设备配置信息的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_config_db_data(UT_array **emsDevConfigs)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *query; // 查询语句
utarray_new(*emsDevConfigs, &ems_dev_config_icd); // 创建新的 UT_array 用于存储设备点位信息
// 构建查询语句
query = "SELECT id, type, content "
"FROM ems_device_config WHERE 1=1 "; // 基础查询语句
// 添加结果排序
query = sqlite3_mprintf("%s ORDER BY type ASC", query); // 按 type 升序排序
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从数据库中获取每一列的值
ems_dev_config_t dev_config = {0}; // 初始化设备配置结构体
dev_config.dbId = (uint16_t)sqlite3_column_int(stmt, 0);
dev_config.type = (ems_dev_config_type_e)sqlite3_column_int(stmt, 1);
snprintf((char *)dev_config.content, MAX_CONFIG_CONTENT_LEN, "%s", (const char *)sqlite3_column_text(stmt, 2));
// 将设备配置信息添加到 UT_array 中
utarray_push_back(*emsDevConfigs, &dev_config);
// 继续处理下一行
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return 0; // 成功返回
error:
sqlite3_finalize(stmt); // 确保释放预编译语句
utarray_free(*emsDevConfigs); // 释放 UT_array
emsDevConfigs = NULL; // 设置为 NULL
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 根据指定配置类型Type从数据库获取 EMS 设备配置信息
* @param[in] type: 配置类型
* @param[out] emsDevConfig: 用于存储设备配置信息的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_config_by_type(ems_dev_config_type_e type, ems_dev_config_t *emsDevConfig)
{
int ret = 0; // 返回代码
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *query; // 查询语句
// 构建查询语句
query = "SELECT content "
"FROM ems_device_config WHERE type = ? LIMIT 1"; // 根据 TYPE 查询
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1; // 返回错误代码
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 1, (int)type)) // 使用 sqlite3_bind_int
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 `%s` 的参数 type 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_finalize(stmt); // 确保释放预编译语句
return 1; // 返回错误代码
}
// 执行查询并处理结果
int step = sqlite3_step(stmt);
if (SQLITE_ROW == step) // 如果找到结果
{
// 从数据库中获取每一列的值
snprintf((char *)emsDevConfig->content, MAX_CONFIG_CONTENT_LEN, "%s", (const char *)sqlite3_column_text(stmt, 0));
}
else if (SQLITE_DONE == step) // 如果没有找到结果
{
KITLOG(LOG_KIT_EN, ERROR_EN, "未找到配置类型 Type 为 %d 的设备配置", type);
ret = 1; // 设置返回代码为失败
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
ret = 1; // 设置返回代码为失败
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return ret; // 返回结果
}
/*****************************************************************************
* @brief 根据协议类型获取北向配置信息
* @param[out] northConfigs: 用于存储北向配置信息的数组, 使用 UT_array 存储
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_north_config_arr(UT_array **northConfigs)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
const char *query = "SELECT nc.id, nc.protocol_type, nc.config_type "
"FROM north_config nc ";
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL)) // config_db: 全局数据库连接
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
KITPTF(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1;
}
utarray_new(*northConfigs, &north_config_icd); // 创建新的 UT_array 用于北向信息
// 执行查询
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step)
{
north_config_t northConfig = {0};
// 获取北向配置信息
northConfig.dbId = sqlite3_column_int(stmt, 0);
northConfig.protocolType = (protocol_type_slave_e)sqlite3_column_int(stmt, 1);
northConfig.configType = (north_config_type_e)sqlite3_column_int(stmt, 2);
// 获取上报下发配置信息
ret = kit_get_north_config_by_id(sqlite3_column_int(stmt, 0), &northConfig);
if (ret != 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "获取上报下发配置信息失败");
goto error; // 处理错误
}
// 将北向配置信息添加到 UT_array 中
utarray_push_back(*northConfigs, &northConfig);
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询北向配置信息失败: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt);
return ret;
error:
// 清理资源
sqlite3_finalize(stmt);
utarray_free(*northConfigs); // 释放 UT_array
northConfigs = NULL;
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 获取所有削峰填谷的日期配置
* @param[out] pvDateConfigs: 削峰填谷的日期配置数组, 使用 UT_array 存储
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_pv_date_cfg_db_data(UT_array **pvDateConfigs)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
// SQL 查询语句,获取 pv_date_config 表中的数据
const char *query = "SELECT pdc.pv_time_config_id, pdc.start_date, pdc.end_date "
"FROM pv_date_config pdc "
"ORDER BY pdc.start_date";
// 创建新的 UT_array 用于存储削峰填谷的日期配置, 使用 pv_date_config_icd 初始化
utarray_new(*pvDateConfigs, &pv_date_config_icd);
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL)) // global_db: 全局数据库连接
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1;
}
// 执行查询并处理结果
int step = sqlite3_step(stmt); // 执行查询
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从数据库中获取每一列的值
pv_date_config_t pv_date_config_info = {0}; // 削峰填谷的日期配置结构体
// 获取开始日期和结束日期
int snprintf_ret1 = snprintf((char *)pv_date_config_info.startDate, sizeof(pv_date_config_info.startDate), "%s", (const char *)sqlite3_column_text(stmt, 1));
int snprintf_ret2 = snprintf((char *)pv_date_config_info.endDate, sizeof(pv_date_config_info.endDate), "%s", (const char *)sqlite3_column_text(stmt, 2));
// 检查 snprintf 是否成功
if (snprintf_ret1 < 0 || snprintf_ret2 < 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "格式化日期失败");
goto error;
}
// 根据 pv_time_config_id 获取时间配置信息
if (kit_get_pv_time_cfg_db_data(sqlite3_column_int(stmt, 0), &pv_date_config_info) != 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "获取时间配置信息失败");
goto error;
}
// 将设备点位信息添加到 UT_array 中
utarray_push_back(*pvDateConfigs, &pv_date_config_info);
// 继续处理下一行
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return ret;
error:
// 处理查询过程中出现的错误
sqlite3_finalize(stmt); // 确保释放预编译语句
utarray_free(*pvDateConfigs); // 释放 UT_array
pvDateConfigs = NULL; // 设置为 NULL
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 根据削峰填谷时间配置的数据库Id获取削峰填谷时间执行配置
* @param[in] pvTimeCfgDbId: 削峰填谷时间配置的数据库Id
* @param[out] pvDateConfig: 用于存储削峰填谷的日期配置, 包含一个 pv_time_config_t 类型的数组
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_pv_time_cfg_db_data(int pvDateCfgDbId, pv_date_config_t *pvDateConfig)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
// SQL 查询语句,根据 time_section_config_id 获取 pv_time_exe 表中的数据
char *query = "SELECT pte.start_time, pte.end_time, pte.power "
"FROM pv_time_exe pte "
"WHERE pte.time_section_config_id = ? "
"ORDER BY pte.start_time";
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL)) // config_db: 全局数据库连接
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1;
}
// 绑定参数
if (SQLITE_OK != sqlite3_bind_int(stmt, 1, pvDateCfgDbId))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定 pvDateCfgDbId 失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1; // 返回错误代码
}
// 执行查询并处理结果
int step = sqlite3_step(stmt); // 执行查询
while (SQLITE_ROW == step) // 遍历每一行结果
{
pvDateConfig->timeCfgLen++;
// 重新分配内存以存储新的时间配置信息
pv_time_config_t *temp = (pv_time_config_t *)realloc(pvDateConfig->timeCfgTab, pvDateConfig->timeCfgLen * sizeof(pv_time_config_t));
if (temp == NULL)
{
// 内存分配失败,记录错误信息并返回
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
ret = 1;
goto error;
}
pvDateConfig->timeCfgTab = temp;
// 获取时间配置信息
pvDateConfig->timeCfgTab[pvDateConfig->timeCfgLen - 1].startTime = sqlite3_column_int(stmt, 0);
pvDateConfig->timeCfgTab[pvDateConfig->timeCfgLen - 1].endTime = sqlite3_column_int(stmt, 1);
// pvDateConfig->timeCfgTab[pvDateConfig->timeCfgLen - 1].workMode = sqlite3_column_int(stmt, 2);
pvDateConfig->timeCfgTab[pvDateConfig->timeCfgLen - 1].power = sqlite3_column_double(stmt, 2);
step = sqlite3_step(stmt);
}
// 检查查询是否完成
if (step != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询削峰填谷时间配置执行信息失败: %s", sqlite3_errmsg(config_db));
ret = 1;
}
error:
// 清理资源
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 设置所有削峰填谷的日期配置
* @param[out] pvDateTimeConfigs: 削峰填谷的日期配置、时间配置数组, 使用 UT_array 存储
* @return 0-成功 1-失败
*****************************************************************************/
int kit_set_pv_date_time_cfg(UT_array **pvDateTimeConfigs)
{
int ret = 0;
sqlite3_stmt* stmt_date = NULL, * stmt_time = NULL, * stmt_exe = NULL;
// SQL 删除语句,删除 pv_date_config 表中的数据
const char* sql_delete_date = "DELETE FROM pv_date_config";
// SQL 查询语句,删除 pv_time_config 表中的数据
const char* sql_delete_time = "DELETE FROM pv_time_config";
// SQL 查询语句,删除 pv_time_exe 表中的数据
const char* sql_delete_exe = "DELETE FROM pv_time_exe";
// SQL 插入语句,插入 pv_date_config 表中的数据
const char* sql_insert_date = "INSERT INTO pv_date_config(pv_time_config_id, start_date, end_date, remark) VALUES (?, ?, ?, ?)";
// SQL 插入语句,插入 pv_time_config 表中的数据
const char* sql_insert_time = "INSERT INTO pv_time_config(name, remark) VALUES (?, ?)";
// SQL 插入语句,插入 pv_time_exe 表中的数据
const char* sql_insert_exe = "INSERT INTO pv_time_exe(time_section_config_id, start_time, end_time, power, remark) VALUES (?, ?, ?, ?,?)";
pthread_mutex_lock(&config_db_mutex);
// 开始事务
ret = sqlite3_exec(config_db, "BEGIN TRANSACTION", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "开启事务失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
// 清空表数据
if (sqlite3_exec(config_db, sql_delete_date, NULL, NULL, NULL) != SQLITE_OK ||
sqlite3_exec(config_db, sql_delete_time, NULL, NULL, NULL) != SQLITE_OK ||
sqlite3_exec(config_db, sql_delete_exe, NULL, NULL, NULL) != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "清空表失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
// 准备插入语句
if (sqlite3_prepare_v2(config_db, sql_insert_date, -1, &stmt_date, NULL) != SQLITE_OK ||
sqlite3_prepare_v2(config_db, sql_insert_time, -1, &stmt_time, NULL) != SQLITE_OK ||
sqlite3_prepare_v2(config_db, sql_insert_exe, -1, &stmt_exe, NULL) != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "预编译插入语句失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
// 遍历需要插入的数据
utarray_foreach(*pvDateTimeConfigs, pv_date_config_t *, p_pvDateConfig)
{
char timeConfigName[50]; // 存储时间配置名称
snprintf(timeConfigName, sizeof(timeConfigName), "%s-%s", p_pvDateConfig->startDate, p_pvDateConfig->endDate);
// 插入 pv_time_config
sqlite3_bind_text(stmt_time, 1, timeConfigName, -1, SQLITE_STATIC);
if (sqlite3_step(stmt_time) != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入 pv_time_config 表失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
long long timeConfigId = sqlite3_last_insert_rowid(config_db); // 获取最后插入行的 ID
sqlite3_reset(stmt_time);
// 插入 pv_time_exe
for (int i = 0; i < p_pvDateConfig->timeCfgLen; i++)
{
sqlite3_bind_int64(stmt_exe, 1, timeConfigId);
sqlite3_bind_int(stmt_exe, 2, p_pvDateConfig->timeCfgTab[i].startTime);
sqlite3_bind_int(stmt_exe, 3, p_pvDateConfig->timeCfgTab[i].endTime);
sqlite3_bind_double(stmt_exe, 4, p_pvDateConfig->timeCfgTab[i].power);
if (sqlite3_step(stmt_exe) != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入 pv_time_exe 表失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
sqlite3_reset(stmt_exe);
}
// 插入 pv_date_config
sqlite3_bind_int64(stmt_date, 1, timeConfigId);
sqlite3_bind_text(stmt_date, 2, (const char *)p_pvDateConfig->startDate, -1, SQLITE_STATIC);
sqlite3_bind_text(stmt_date, 3, (const char *)p_pvDateConfig->endDate, -1, SQLITE_STATIC);
if (sqlite3_step(stmt_date) != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入 pv_date_config 表失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
sqlite3_reset(stmt_date);
}
// 提交事务
if (sqlite3_exec(config_db, "COMMIT TRANSACTION", NULL, NULL, NULL) != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(config_db));
goto cleanup;
}
ret = 0; // 成功
cleanup:
if (ret != 0)
{
// 回滚事务
sqlite3_exec(config_db, "ROLLBACK TRANSACTION", NULL, NULL, NULL);
}
// 释放资源
sqlite3_finalize(stmt_date);
sqlite3_finalize(stmt_time);
sqlite3_finalize(stmt_exe);
pthread_mutex_unlock(&config_db_mutex);
// 释放 UT_array 资源
if (ret != 0)
{
utarray_free(*pvDateTimeConfigs);
*pvDateTimeConfigs = NULL;
}
return ret;
}
/*****************************************************************************
* @brief 从数据库获取EMS拓扑结构信息
* @param[out] topologys: 用于存储ems拓扑信息结构体的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_topology_db_data(UT_array **topologies)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *query; // 查询语句
query = "SELECT tpy.id, tpy.parent_id, tpy.menu_tree, tpy.sort_order, "
"CASE "
" WHEN tpy.menu_tree = 0 THEN tpy.name "
" WHEN tpy.menu_tree = 1 THEN dev.name ELSE NULL "
"END AS name, "
"CASE "
" WHEN tpy.menu_tree = 0 THEN NULL "
" WHEN tpy.menu_tree = 1 THEN dev.sn ELSE NULL "
"END AS sn, "
"CASE "
" WHEN tpy.menu_tree = 0 THEN NULL "
" WHEN tpy.menu_tree = 1 THEN dev.type ELSE NULL "
"END AS dev_type "
"FROM "
"topology tpy "
"LEFT JOIN device dev ON tpy.dev_id = dev.id";
utarray_new(*topologies, &topology_icd); // 创建新的 UT_array 用于存储ems拓扑信息
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_finalize(stmt); // 确保释放预编译语句
return 1; // 返回错误代码
}
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从数据库中获取每一列的值
topology_t topology_info = {0}; // 初始化拓扑结构的结构体
topology_info.dbId = sqlite3_column_int(stmt, 0);
if (sqlite3_column_type(stmt, 1) == SQLITE_NULL)
{
topology_info.parentId = -999;
}
else
{
topology_info.parentId = sqlite3_column_int(stmt, 1);
}
topology_info.menuTree = (menu_tree_e)sqlite3_column_int(stmt, 2);
topology_info.sortOrder = sqlite3_column_int(stmt, 3);
const char *name = (const char *)sqlite3_column_text(stmt, 4);
snprintf((char *)topology_info.name, MAX_NAME_BUF_LEN, "%s", name ? name : "");
const char *sn = (const char *)sqlite3_column_text(stmt, 5);
snprintf((char *)topology_info.sn, MAX_CODE_BUF_LEN, "%s", sn ? sn : "");
topology_info.devType = (dev_type_e)sqlite3_column_int(stmt, 6);
// 将设备点位信息添加到 UT_array 中
utarray_push_back(*topologies, &topology_info);
// 继续处理下一行
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return 0; // 成功返回
error:
sqlite3_finalize(stmt); // 确保释放预编译语句
utarray_free(*topologies); // 释放 UT_array
topologies = NULL; // 设置为 NULL
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 从数据库里获取保护算法的控制参数
* @param[in] p_protectAlgo保护算法结构体指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_protect_algorithm(protect_algorithm_t *p_protectAlgo)
{
sqlite3_stmt *stmt = NULL;
char *query = "SELECT transformer_capacity, ol_warn_limit_value, ol_shutdown_value, "
"max_power, demand_switch, target_demand, de_warn_limit_value, "
"de_shutdown_value, backflow_switch, bf_warn_limit_value, "
"bf_shutdown_value, soc_forbid_chg, soc_forbid_dischg, "
"id "
"FROM protect_algorithm "
"ORDER BY id DESC "
"LIMIT 1";
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1; // 返回错误代码
}
int step = sqlite3_step(stmt);
if (SQLITE_ROW == step)
{
p_protectAlgo->transCapacity = sqlite3_column_double(stmt, 0); // 变压器总容量(KW)
p_protectAlgo->olWarnLimitVal = sqlite3_column_double(stmt, 1); // 过载预警限流值(KW)
p_protectAlgo->olShutdownVal = sqlite3_column_double(stmt, 2); // 过载关机值(KW)
p_protectAlgo->maxPower = sqlite3_column_double(stmt, 3); // 充放电功率上限(KW)
p_protectAlgo->demandSwitch = sqlite3_column_int(stmt, 4); // 防需量开关
p_protectAlgo->targetDemand = sqlite3_column_double(stmt, 5); // 目标需量
p_protectAlgo->deWarnLimitVal = sqlite3_column_double(stmt, 6); // 需量预警限流值(KW)
p_protectAlgo->deShutdownVal = sqlite3_column_double(stmt, 7); // 需量关机值(KW)
p_protectAlgo->backflowSwitch = sqlite3_column_int(stmt, 8); // 防逆流开关
p_protectAlgo->bfWarnLimitVal = sqlite3_column_double(stmt, 9); // 防逆流预警限流值(KW)
p_protectAlgo->bfShutdownVal = sqlite3_column_double(stmt, 10); // 防逆流关机值(KW)
p_protectAlgo->socForbidChg = sqlite3_column_int(stmt, 11); // 禁充SOC [55%-100%]
p_protectAlgo->socForbidDischg = sqlite3_column_int(stmt, 12); // 禁放SOC [0%-45%]
p_protectAlgo->dbId = sqlite3_column_int(stmt, 13); // 数据库Id
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败或者无数据,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1; // 返回错误代码
}
sqlite3_finalize(stmt);
return 0; // 返回成功代码
}
/*****************************************************************************
* @brief 更新数据库的保护算法数据
* @param[in] p_protectAlgo保护算法结构体指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_set_protect_algorithm(protect_algorithm_t *p_protectAlgo)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
char *sql = sqlite3_mprintf(
"UPDATE protect_algorithm SET transformer_capacity = ?, ol_warn_limit_value = ?, ol_shutdown_value = ?, "
"max_power = ?, demand_switch = ?, target_demand = ?, de_warn_limit_value = ?, "
"de_shutdown_value = ?, backflow_switch = ?, bf_warn_limit_value = ?, "
"bf_shutdown_value = ?, soc_forbid_chg = ?, soc_forbid_dischg = ? WHERE id = ?");
if (SQLITE_OK != sqlite3_prepare_v2(config_db, sql, -1, &stmt, NULL))
{
fprintf(stderr, "SQL语句预编译失败: %s\n", sqlite3_errmsg(config_db));
sqlite3_free(sql);
return 1;
}
sqlite3_bind_double(stmt, 1, p_protectAlgo->transCapacity);
sqlite3_bind_double(stmt, 2, p_protectAlgo->olWarnLimitVal);
sqlite3_bind_double(stmt, 3, p_protectAlgo->olShutdownVal);
sqlite3_bind_double(stmt, 4, p_protectAlgo->maxPower);
sqlite3_bind_int(stmt, 5, p_protectAlgo->demandSwitch);
sqlite3_bind_double(stmt, 6, p_protectAlgo->targetDemand);
sqlite3_bind_double(stmt, 7, p_protectAlgo->deWarnLimitVal);
sqlite3_bind_double(stmt, 8, p_protectAlgo->deShutdownVal);
sqlite3_bind_int(stmt, 9, p_protectAlgo->backflowSwitch);
sqlite3_bind_double(stmt, 10, p_protectAlgo->bfWarnLimitVal);
sqlite3_bind_double(stmt, 11, p_protectAlgo->bfShutdownVal);
sqlite3_bind_int(stmt, 12, p_protectAlgo->socForbidChg);
sqlite3_bind_int(stmt, 13, p_protectAlgo->socForbidDischg);
sqlite3_bind_int(stmt, 14, p_protectAlgo->dbId);
pthread_mutex_lock(&config_db_mutex);
int step = sqlite3_step(stmt);
if (step != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败");
if (step == SQLITE_ERROR)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "%s", sqlite3_errmsg(config_db));
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "错误代码: %d", step);
}
ret = 1;
goto cleanup;
}
if (sqlite3_changes(config_db) == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "没有更新任何行。请检查 dbId 是否正确");
ret = 1;
goto cleanup;
}
cleanup:
pthread_mutex_unlock(&config_db_mutex);
sqlite3_finalize(stmt);
sqlite3_free(sql);
return ret;
}
/*****************************************************************************
* @brief 从数据库里获取调试模式算法的控制参数
* @param[in] p_debugAlgo调试模式算法结构体指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_debug_algorithm(debug_algorithm_t *p_debugAlgo)
{
sqlite3_stmt *stmt = NULL;
char *query = "SELECT active_power, reactive_power, pcs_switch, id, protect_switch "
"FROM debug_algorithm "
"ORDER BY id DESC "
"LIMIT 1";
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1; // 返回错误代码
}
int step = sqlite3_step(stmt);
if (SQLITE_ROW == step)
{
p_debugAlgo->activePower = sqlite3_column_double(stmt, 0); // 目标有功功率(KW)
p_debugAlgo->reactivePower = sqlite3_column_double(stmt, 1); // 目标无功功率(KW)
p_debugAlgo->pcsSwitch = sqlite3_column_int(stmt, 2); // 充放电开关
p_debugAlgo->dbId = sqlite3_column_int(stmt, 3); // 数据库Id
p_debugAlgo->protectSwitch = sqlite3_column_int(stmt, 4); // 保护算法开关
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
return 1; // 返回错误代码
}
sqlite3_finalize(stmt);
return 0; // 返回成功代码
}
/*****************************************************************************
* @brief 更新数据库的调试模式算法数据
* @param[in] p_debugAlgo调试模式算法结构体指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_set_debug_algorithm(debug_algorithm_t *p_debugAlgo)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
const char *sql = sqlite3_mprintf("UPDATE debug_algorithm SET active_power = ?, reactive_power = ?, pcs_switch = ?, protect_switch = ? WHERE id = ? ");
if (SQLITE_OK != sqlite3_prepare_v2(config_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(config_db));
return 1;
}
sqlite3_bind_double(stmt, 1, p_debugAlgo->activePower);
sqlite3_bind_double(stmt, 2, p_debugAlgo->reactivePower);
sqlite3_bind_int(stmt, 3, p_debugAlgo->pcsSwitch);
sqlite3_bind_int(stmt, 4, p_debugAlgo->protectSwitch);
sqlite3_bind_int(stmt, 5, p_debugAlgo->dbId);
pthread_mutex_lock(&config_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (step != SQLITE_DONE)
{
if (step == SQLITE_ERROR)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s", sqlite3_errmsg(config_db));
}
else if (step == SQLITE_ROW)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "UPDATE操作中出现意外的SQLITE_ROW返回值");
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL更新失败错误代码: %d", step);
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 检查是否实际更新了任何行。SQLITE_DONE 并不保证这一点。
if (sqlite3_changes(config_db) == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "没有更新任何行。请检查 dbId 是否正确。");
pthread_mutex_unlock(&config_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 从数据库查询功率分配数据
* @param[in] p_allocatePower功率分配的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_allocate_power(allocate_power_t *p_allocatePower)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT pcs_num, chg_direction, id, allocate_mode FROM allocate_power ORDER BY id DESC LIMIT 1");
if (query == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
return 1;
}
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1; // 返回错误代码
}
int step = sqlite3_step(stmt);
if (step == SQLITE_ROW)
{
p_allocatePower->pcsNum = sqlite3_column_int(stmt, 0); // pcs数量
p_allocatePower->chgDirection = sqlite3_column_int(stmt, 1); // 充放电方向1(正放 负充) -1(正充 负放)
p_allocatePower->dbId = sqlite3_column_int(stmt, 2); // 数据库Id
p_allocatePower->allocateMode = sqlite3_column_int(stmt, 3); // 下发功率开关总功率0分相功率1
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败或者无数据,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return (step == SQLITE_ROW) ? 0 : 1; // 返回成功或失败代码
}
/*****************************************************************************
* @brief 更新数据库的功率分配数据
* @param[in] p_allocatePower功率分配的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_set_allocate_power(allocate_power_t *p_allocatePower)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
const char *sql = sqlite3_mprintf("UPDATE allocate_power SET pcs_num = ?, chg_direction = ?, allocate_mode = ? WHERE id = ? ");
if (SQLITE_OK != sqlite3_prepare_v2(config_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(config_db));
return 1;
}
sqlite3_bind_int(stmt, 1, p_allocatePower->pcsNum);
sqlite3_bind_int(stmt, 2, p_allocatePower->chgDirection);
sqlite3_bind_int(stmt, 3, p_allocatePower->allocateMode);
sqlite3_bind_int(stmt, 4, p_allocatePower->dbId);
pthread_mutex_lock(&config_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (step != SQLITE_DONE)
{
if (step == SQLITE_ERROR)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s", sqlite3_errmsg(config_db));
}
else if (step == SQLITE_ROW)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "UPDATE操作中出现意外的SQLITE_ROW返回值");
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL更新失败错误代码: %d", step);
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 检查是否实际更新了任何行。SQLITE_DONE 并不保证这一点。
if (sqlite3_changes(config_db) == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "没有更新任何行。请检查 dbId 是否正确。");
pthread_mutex_unlock(&config_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 从数据库查询工作模式数据
* @param[in] p_workModeSet工作模式的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_work_mode_set(work_mode_set_t *p_workModeSet)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT work_mode, id FROM work_mode_set DESC LIMIT 1");
if (query == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
return 1;
}
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1; // 返回错误代码
}
int step = sqlite3_step(stmt);
if (step == SQLITE_ROW)
{
p_workModeSet->workMode = (work_mode_e)sqlite3_column_int(stmt, 0); // 工作模式
p_workModeSet->dbId = sqlite3_column_int(stmt, 1); // 数据库Id
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return (step == SQLITE_ROW) ? 0 : 1; // 返回成功或失败代码
}
/*****************************************************************************
* @brief 更新数据库的工作模式数据
* @param[in] p_workModeSet工作模式的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_set_work_mode(work_mode_set_t *p_workModeSet)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
const char *sql = sqlite3_mprintf("UPDATE work_mode_set SET work_mode = ? WHERE id = ? ");
if (SQLITE_OK != sqlite3_prepare_v2(config_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(config_db));
return 1;
}
sqlite3_bind_int(stmt, 1, (uint8_t)p_workModeSet->workMode);
sqlite3_bind_int(stmt, 2, p_workModeSet->dbId);
pthread_mutex_lock(&config_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (step != SQLITE_DONE)
{
if (step == SQLITE_ERROR)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s", sqlite3_errmsg(config_db));
}
else if (step == SQLITE_ROW)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "UPDATE操作中出现意外的SQLITE_ROW返回值");
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL更新失败错误代码: %d", step);
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 检查是否实际更新了任何行。SQLITE_DONE 并不保证这一点。
if (sqlite3_changes(config_db) == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "没有更新任何行。请检查 dbId 是否正确。");
pthread_mutex_unlock(&config_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 从数据库查询DIDO逻辑设置数据
* @param[in] diDoSetsDIDO逻辑设置结构体的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_di_do_set_arr(UT_array **diDoSets)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *query; // 查询语句
query = "SELECT seq, type, cabinet_code, strategy_seq "
"FROM di_do_set ";
// 准备 SQL 语句
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
return 1;
}
utarray_new(*diDoSets, &di_do_logic_set_icd); // 创建新的 UT_array 用于存储DIDO逻辑设置信息
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step) // 遍历每一行结果
{
// 从数据库中获取每一列的值
di_do_logic_set_t di_do_set_logic_info = {0}; // 初始化DIDO逻辑设置的结构体
di_do_set_logic_info.dIDOseq = sqlite3_column_int(stmt, 0);
di_do_set_logic_info.dIDOType = (di_do_type_e)sqlite3_column_int(stmt, 1);
di_do_set_logic_info.cabinetCode = sqlite3_column_int(stmt, 2);
if (di_do_set_logic_info.dIDOType == kDi)
{
di_do_set_logic_info.strategySeqDi = (diFuc_e)sqlite3_column_int(stmt, 3);
}
else
{
di_do_set_logic_info.strategySeqDo = (doFuc_e)sqlite3_column_int(stmt, 3);
}
// 将DIDO逻辑设置添加到 UT_array 中
utarray_push_back(*diDoSets, &di_do_set_logic_info);
// 继续处理下一行
step = sqlite3_step(stmt);
}
// 检查是否完成所有行的处理
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询未完成,错误信息: %s", sqlite3_errmsg(config_db));
goto error; // 处理错误
}
// 清理资源
sqlite3_finalize(stmt); // 释放预编译语句
return 0; // 成功返回
error:
sqlite3_finalize(stmt); // 确保释放预编译语句
utarray_free(*diDoSets); // 释放 UT_array
diDoSets = NULL; // 设置为 NULL
return 1; // 返回错误代码
}
/*****************************************************************************
* @brief 从数据库里获取温控算法的控制参数
* @param[in] p_tempCtrlAlgo温控算法结构体指针
* @return 0-成功非0-失败(具体错误码可通过日志或其他机制获取)
*****************************************************************************/
int kit_get_temp_ctrl_algorithm(temp_ctrl_algorithm_t *p_tempCtrlAlgo)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
if (p_tempCtrlAlgo == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "传入的结构体指针为空");
return 1; // 返回错误代码
}
char *query = "SELECT mod, work_cool_trig_temp, work_cool_hys, "
"work_heat_trig_temp, work_heat_hys, standby_cool_trig_temp, "
"standby_cool_hys, standby_heat_trig_temp, standby_heat_hys, "
"standby_max_limit, standby_min_limit, standby_maxback_limit, "
"standby_minback_limit, temp_cmd_normal, temp_cmd_heat, temp_cmd_cold, "
"temp_cmd_open, temp_cmd_close, id "
"FROM temp_ctrl_algorithm "
"ORDER BY id DESC "
"LIMIT 1";
int rc = sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL);
if (rc != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
ret = rc; // 记录sqlite3的错误码
goto cleanup;
}
int step = sqlite3_step(stmt);
if (step == SQLITE_ROW)
{
p_tempCtrlAlgo->mod = (temperature_ctrl_mod_e)sqlite3_column_int(stmt, 0);
p_tempCtrlAlgo->workCoolTrigTemp = (float)sqlite3_column_double(stmt, 1);
p_tempCtrlAlgo->workCoolHys = (float)sqlite3_column_double(stmt, 2);
p_tempCtrlAlgo->workHeatTrigTemp = (float)sqlite3_column_double(stmt, 3);
p_tempCtrlAlgo->workHeatHys = (float)sqlite3_column_double(stmt, 4);
p_tempCtrlAlgo->standbyCoolTrigTemp = (float)sqlite3_column_double(stmt, 5);
p_tempCtrlAlgo->standbyCoolHys = (float)sqlite3_column_double(stmt, 6);
p_tempCtrlAlgo->standbyHeatTrigTemp = (float)sqlite3_column_double(stmt, 7);
p_tempCtrlAlgo->standbyHeatHys = (float)sqlite3_column_double(stmt, 8);
p_tempCtrlAlgo->standbyMaxLimit = (float)sqlite3_column_double(stmt, 9);
p_tempCtrlAlgo->standbyMinLimit = (float)sqlite3_column_double(stmt, 10);
p_tempCtrlAlgo->standbyMaxbackLimit = (float)sqlite3_column_double(stmt, 11);
p_tempCtrlAlgo->standbyMinbackLimit = (float)sqlite3_column_double(stmt, 12);
p_tempCtrlAlgo->tempCmdNormal = sqlite3_column_int(stmt, 13);
p_tempCtrlAlgo->tempCmdHeat = sqlite3_column_int(stmt, 14);
p_tempCtrlAlgo->tempCmdCold = sqlite3_column_int(stmt, 15);
p_tempCtrlAlgo->tempCmdOpen = sqlite3_column_int(stmt, 16);
p_tempCtrlAlgo->tempCmdClose = sqlite3_column_int(stmt, 17);
p_tempCtrlAlgo->dbId = sqlite3_column_int64(stmt, 18);
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
ret = step; // 记录sqlite3的错误码
}
cleanup:
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 更新数据库的温控算法数据
* @param[in] p_tempCtrlAlgo温控算法结构体指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_set_temp_ctrl_algorithm(temp_ctrl_algorithm_t *p_tempCtrlAlgo)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
// 使用互斥锁保护数据库访问,防止多线程竞争
pthread_mutex_lock(&config_db_mutex);
// SQL语句使用占位符防止SQL注入并根据实际情况选择合适的更新策略 (UPDATE or INSERT)
char *query = "UPDATE temp_ctrl_algorithm SET mod = ?, work_cool_trig_temp = ?, work_cool_hys = ?, "
"work_heat_trig_temp = ?, work_heat_hys = ?, standby_cool_trig_temp = ?, "
"standby_cool_hys = ?, standby_heat_trig_temp = ?, standby_heat_hys = ?, "
"standby_max_limit = ?, standby_min_limit = ?, standby_maxback_limit = ?, "
"standby_minback_limit = ?, temp_cmd_normal = ?, temp_cmd_heat = ?, temp_cmd_cold = ?, "
"temp_cmd_open = ?, temp_cmd_close = ? "
"WHERE id = ?;"; // 更新最新的记录
// 准备SQL语句
int rc = sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL);
if (rc != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备SQL语句失败: %s", sqlite3_errmsg(config_db));
ret = 1;
goto unlock_and_return;
}
// 绑定参数
ret = sqlite3_bind_int(stmt, 1, p_tempCtrlAlgo->mod); // 温度控制模式
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 2, p_tempCtrlAlgo->workCoolTrigTemp); // 制冷触发温度
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 3, p_tempCtrlAlgo->workCoolHys); // 制冷回差
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 4, p_tempCtrlAlgo->workHeatTrigTemp); // 制热触发温度
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 5, p_tempCtrlAlgo->workHeatHys); // 制热回差
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 6, p_tempCtrlAlgo->standbyCoolTrigTemp); // 制冷触发温度
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 7, p_tempCtrlAlgo->standbyCoolHys); // 制冷回差
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 8, p_tempCtrlAlgo->standbyHeatTrigTemp); // 制热触发温度
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 9, p_tempCtrlAlgo->standbyHeatHys); // 制热回差
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 10, p_tempCtrlAlgo->standbyMaxLimit); // 开启自循环高温
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 11, p_tempCtrlAlgo->standbyMinLimit); // 开启自循环低温
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 12, p_tempCtrlAlgo->standbyMaxbackLimit); // 自循环高温回温
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_double(stmt, 13, p_tempCtrlAlgo->standbyMinbackLimit); // 自循环低温回温
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_int(stmt, 14, p_tempCtrlAlgo->tempCmdNormal); // 循环控制字
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_int(stmt, 15, p_tempCtrlAlgo->tempCmdHeat); // 制热控制字
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_int(stmt, 16, p_tempCtrlAlgo->tempCmdCold); // 制冷控制字
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_int(stmt, 17, p_tempCtrlAlgo->tempCmdOpen); // 开控制字
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_int(stmt, 18, p_tempCtrlAlgo->tempCmdClose); // 关控制字
if (ret != SQLITE_OK)
goto error;
ret = sqlite3_bind_int64(stmt, 19, p_tempCtrlAlgo->dbId); // 数据库Id
if (ret != SQLITE_OK)
goto error;
// 执行SQL语句
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "执行SQL语句失败: %s", sqlite3_errmsg(config_db));
goto error;
}
// 检查是否实际更新了任何行
if (sqlite3_changes(config_db) == 0)
{
KITLOG(LOG_KIT_EN, WARN_EN, "没有更新任何行。请检查数据是否存在。");
ret = 1;
}
sqlite3_finalize(stmt);
goto unlock_and_return;
error:
sqlite3_finalize(stmt);
ret = 1;
unlock_and_return:
pthread_mutex_unlock(&config_db_mutex); // 解锁
return ret;
}
/*****************************************************************************
* @brief 从数据库获取高级设置
* @param[out] advancedSettings: 用于存储高级设置的数据
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_advanced_setting(UT_array **advancedSettings)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT key, value FROM advanced_setting ORDER BY key ASC");
int ret;
ret = sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1;
}
utarray_new(*advancedSettings, &advanced_setting_icd);
while ((ret = sqlite3_step(stmt)) == SQLITE_ROW)
{
advanced_setting_t setting_info = {0};
snprintf((char *)setting_info.key, MAX_KEY_LEN, "%s", (const char *)sqlite3_column_text(stmt, 0));
snprintf((char *)setting_info.value, MAX_VALUE_LEN, "%s", (const char *)sqlite3_column_text(stmt, 1));
utarray_push_back(*advancedSettings, &setting_info);
}
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败: %s", sqlite3_errmsg(config_db));
sqlite3_finalize(stmt);
sqlite3_free(query);
utarray_free(*advancedSettings);
*advancedSettings = NULL;
return 1;
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return 0;
}
/*****************************************************************************
* @brief 从数据库查询升级记录数据
* @param[in] p_upgradeRecord升级记录的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_get_upgrade_record(upgrade_record_t *p_upgradeRecord)
{
sqlite3_stmt *stmt = NULL;
char *query = sqlite3_mprintf("SELECT id, path FROM upgrade_record Where status = 0 LIMIT 1");
if (query == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
return 1;
}
if (SQLITE_OK != sqlite3_prepare_v2(config_db, query, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "准备查询 `%s` 失败: %s", query, sqlite3_errmsg(config_db));
sqlite3_free(query);
return 1; // 返回错误代码
}
int step = sqlite3_step(stmt);
if (step == SQLITE_ROW)
{
p_upgradeRecord->dbId = sqlite3_column_int(stmt, 0); // 数据库Id
snprintf((char *)p_upgradeRecord->path, MAX_UPGRADE_PATH_LEN, "%s", sqlite3_column_text(stmt, 1)); // 升级包文件路径
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "查询失败,错误码: %d, 错误信息: %s", step, sqlite3_errmsg(config_db));
}
sqlite3_finalize(stmt);
sqlite3_free(query);
return (step == SQLITE_ROW) ? 0 : 1; // 返回成功或失败代码
}
/*****************************************************************************
* @brief 更新数据库的升级记录数据
* @param[in] p_upgradeRecord升级记录的指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_update_upgrade_record(upgrade_record_t *p_upgradeRecord)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
const char *sql = sqlite3_mprintf("UPDATE upgrade_record SET status = ?, err_msg = ? WHERE id = ? ");
if (SQLITE_OK != sqlite3_prepare_v2(config_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(config_db));
return 1;
}
sqlite3_bind_int(stmt, 1, (uint8_t)p_upgradeRecord->status);
sqlite3_bind_text(stmt, 2, (const char *)p_upgradeRecord->errMsg, -1, SQLITE_STATIC);
sqlite3_bind_int(stmt, 3, p_upgradeRecord->dbId);
pthread_mutex_lock(&config_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (step != SQLITE_DONE)
{
if (step == SQLITE_ERROR)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s", sqlite3_errmsg(config_db));
}
else if (step == SQLITE_ROW)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "UPDATE操作中出现意外的SQLITE_ROW返回值");
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL更新失败错误代码: %d", step);
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 检查是否实际更新了任何行。SQLITE_DONE 并不保证这一点。
if (sqlite3_changes(config_db) == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "没有更新任何行。请检查 dbId 是否正确。");
pthread_mutex_unlock(&config_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&config_db_mutex); // 解锁
sqlite3_finalize(stmt);
return ret;
}
/* ############################################################################### 业务数据库操作 ############################################################################### */
/*****************************************************************************
* @brief 批量插入时序数据存储
* @param[in] timedDatas: 用于存储备份数据的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_insert_timed_data(UT_array *timedDatas)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *sql = "INSERT INTO timed_data_storage (dev_type, dev_id, point_id, timed_value, storage_interval, created_time) VALUES (?, ?, ?, ?, ?, strftime('%Y-%m-%d %H:%M:%S', 'now'))";
int ret = 0;
int total_count = utarray_len(timedDatas); // 总记录数
int batch_size = 100; // 每批插入的条数
int i;
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
return 1;
}
// 预编译SQL语句
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
return 1;
}
// 批量插入
for (i = 0; i < total_count; i++)
{
timed_data_storage_t *timedData = (timed_data_storage_t *)utarray_eltptr(timedDatas, i);
// 绑定参数
sqlite3_bind_int(stmt, 1, timedData->devType);
sqlite3_bind_int(stmt, 2, timedData->devDbId);
sqlite3_bind_int(stmt, 3, timedData->pointDbId);
// sqlite3_bind_text(stmt, 4, (const char *)timedData->devName, -1, SQLITE_STATIC);
// sqlite3_bind_text(stmt, 5, (const char *)timedData->pointName, -1, SQLITE_STATIC);
sqlite3_bind_double(stmt, 4, timedData->timedValue);
sqlite3_bind_int(stmt, 5, timedData->storageInterval);
// 执行插入
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 重置语句,以便下次使用
sqlite3_reset(stmt);
// 每 batch_size 条记录提交一次事务
if ((i + 1) % batch_size == 0)
{
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 重新开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法重新开始事务: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
}
}
// 提交剩余数据
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex); // 解锁
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 释放预编译语句
sqlite3_finalize(stmt);
utarray_free(timedDatas);
return 0; // 成功
}
/*****************************************************************************
* @brief 根据日期时间条件删除该日期时间之前timed_data_storage的数据
* @param[in] storage_interval存储周期
* @param[in] date_time_string日期时间
* @return 0-成功 1-失败
*****************************************************************************/
int kit_del_timed_data_by_date(int storage_interval, const char *date_time_string)
{
sqlite3_stmt *stmt = NULL; // SQLite语句句柄
int ret = 0; // 返回值
char sql[256]; // SQL语句缓冲区
// 参数校验确保日期字符串不为NULL
if (date_time_string == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "日期时间字符串为NULL");
return 1;
}
// 使用sqlite3_snprintf构建SQL语句防止SQL注入
sqlite3_snprintf(sizeof(sql), sql,
"DELETE FROM timed_data_storage WHERE storage_interval = %d AND created_time < '%q';",
storage_interval, date_time_string);
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务,确保数据一致性
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1; // 直接返回,不继续执行
}
// 准备SQL语句避免直接使用sqlite3_exec
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句准备失败: %s SQL: %s", sqlite3_errmsg(business_db), sql);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 执行SQL语句
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE) // SQLITE_DONE表示成功执行完语句
{
KITLOG(LOG_KIT_EN, ERROR_EN, "删除数据失败: %s SQL: %s", sqlite3_errmsg(business_db), sql);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt); // 释放SQLite语句句柄
return 1;
}
// 提交事务,确保操作生效
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
sqlite3_finalize(stmt); // 释放SQLite语句句柄
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
// 释放SQLite语句句柄避免内存泄漏
sqlite3_finalize(stmt);
return 0; // 返回0表示成功
}
/*****************************************************************************
* @brief 批量插入故障数据存储
* @param[in] faultsDatas: 用于存储故障数据的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_insert_faults_data(UT_array *faultsDatas)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *sql = "INSERT INTO faults_record (dev_type, dev_id, point_id, faults_value, created_time) VALUES (?, ?, ?, ?, strftime('%Y-%m-%d %H:%M:%S', 'now'))";
int ret = 0;
int total_count = utarray_len(faultsDatas); // 总记录数
int batch_size = 100; // 每批插入的条数
int i;
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
return 1;
}
// 预编译SQL语句
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
return 1;
}
// 批量插入
for (i = 0; i < total_count; i++)
{
faults_data_storage_t *faultsData = (faults_data_storage_t *)utarray_eltptr(faultsDatas, i);
// 绑定参数
sqlite3_bind_int(stmt, 1, faultsData->devType);
sqlite3_bind_int(stmt, 2, faultsData->devDbId);
sqlite3_bind_int(stmt, 3, faultsData->pointDbId);
sqlite3_bind_double(stmt, 4, faultsData->faultsValue);
// 执行插入
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 重置语句,以便下次使用
sqlite3_reset(stmt);
// 每 batch_size 条记录提交一次事务
if ((i + 1) % batch_size == 0)
{
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 重新开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法重新开始事务: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
}
}
// 提交剩余数据
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex); // 解锁
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 释放预编译语句
sqlite3_finalize(stmt);
utarray_free(faultsDatas);
return 0; // 成功
}
/*****************************************************************************
* @brief 根据日期时间条件删除该日期时间之前faults_data_storage的数据
* @param[in] date_time_string日期时间
* @return 0-成功 1-失败
*****************************************************************************/
int kit_del_faults_data_by_date(const char *date_time_string)
{
sqlite3_stmt *stmt = NULL; // SQLite语句句柄
int ret = 0; // 返回值
char sql[256]; // SQL语句缓冲区
// 参数校验确保日期字符串不为NULL
if (date_time_string == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "日期时间字符串为NULL");
return 1;
}
// 使用sqlite3_snprintf构建SQL语句防止SQL注入
sqlite3_snprintf(sizeof(sql), sql, "DELETE FROM faults_record WHERE created_time < '%q';", date_time_string);
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务,确保数据一致性
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1; // 直接返回,不继续执行
}
// 准备SQL语句避免直接使用sqlite3_exec
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句准备失败: %s SQL: %s", sqlite3_errmsg(business_db), sql);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 执行SQL语句
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE) // SQLITE_DONE表示成功执行完语句
{
KITLOG(LOG_KIT_EN, ERROR_EN, "删除数据失败: %s SQL: %s", sqlite3_errmsg(business_db), sql);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt); // 释放SQLite语句句柄
return 1;
}
// 提交事务,确保操作生效
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
sqlite3_finalize(stmt); // 释放SQLite语句句柄
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
// 释放SQLite语句句柄避免内存泄漏
sqlite3_finalize(stmt);
return 0; // 返回0表示成功
}
/*******************************************************************************
* @brief 删除指定时间字符串之前的断点保存数据
* @param[in] date_time_string 日期时间字符串,格式为"YYYY-MM-DD HH:MM:SS"
* @return 0-成功 1-失败
*****************************************************************************/
int kit_del_break_data_by_date(const char *date_time_string)
{
char sql1[256], sql2[256]; // SQL语句缓冲区
sqlite3_stmt *stmt1 = NULL, *stmt2 = NULL; // SQLite语句句柄
int ret = 0; // 返回值
// 参数校验确保日期时间字符串不为NULL
if (date_time_string == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "日期时间字符串为NULL");
return 1;
}
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务,确保数据一致性
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1; // 直接返回,不继续执行
}
// 使用sqlite3_snprintf构建SQL语句防止SQL注入
sqlite3_snprintf(sizeof(sql1), sql1, "DELETE FROM break_record WHERE created_time < ?;");
sqlite3_snprintf(sizeof(sql2), sql2, "DELETE FROM break_data_storage WHERE created_time < ?;");
// 准备并执行第一个 DELETE 语句
// 准备SQL语句
ret = sqlite3_prepare_v2(business_db, sql1, -1, &stmt1, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句准备失败: %s SQL: %s", sqlite3_errmsg(business_db), sql1);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 绑定参数
ret = sqlite3_bind_text(stmt1, 1, date_time_string, -1, SQLITE_STATIC);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定参数失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt1);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 执行SQL语句
ret = sqlite3_step(stmt1);
if (ret != SQLITE_DONE) // SQLITE_DONE表示成功执行完语句
{
KITLOG(LOG_KIT_EN, ERROR_EN, "删除数据失败: %s SQL: %s", sqlite3_errmsg(business_db), sql1);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt1); // 释放SQLite语句句柄
return 1;
}
// 准备并执行第二个 DELETE 语句
// 准备SQL语句
ret = sqlite3_prepare_v2(business_db, sql2, -1, &stmt2, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句准备失败: %s SQL: %s", sqlite3_errmsg(business_db), sql2);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 绑定参数
ret = sqlite3_bind_text(stmt2, 1, date_time_string, -1, SQLITE_STATIC);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "绑定参数失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt2);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 执行SQL语句
ret = sqlite3_step(stmt2);
if (ret != SQLITE_DONE) // SQLITE_DONE表示成功执行完语句
{
KITLOG(LOG_KIT_EN, ERROR_EN, "删除数据失败: %s SQL: %s", sqlite3_errmsg(business_db), sql2);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt2); // 释放SQLite语句句柄
return 1;
}
// 提交事务,确保操作生效
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
// 释放SQLite语句句柄避免内存泄漏
sqlite3_finalize(stmt1);
return 0; // 返回0表示成功
}
/*****************************************************************************
* @brief 查询MQTT断开记录根据isUploaded字段筛选。
* @param[in] isUploaded: 要查询的is_uploaded字段值。如果为255则查询所有记录。
* @param[out] breakRecord: 用于存储查询结果的UT_array指针。
* @return 0-成功 1-失败
*****************************************************************************/
int kit_query_break_records(uint8_t isUploaded, UT_array **breakRecord)
{
sqlite3_stmt *stmt = NULL;
char *sql;
if (isUploaded == 255)
{
sql = sqlite3_mprintf("SELECT id, reason, is_uploaded FROM break_record ORDER BY created_time ASC;");
}
else
{
sql = sqlite3_mprintf("SELECT id, reason, is_uploaded FROM break_record WHERE is_uploaded = %d ORDER BY created_time ASC;", isUploaded);
}
if (SQLITE_OK != sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
sqlite3_free(sql);
return 1;
}
sqlite3_free(sql);
utarray_new(*breakRecord, &break_record_icd); // 初始化UT_array
// 执行查询并处理结果
int step = sqlite3_step(stmt);
while (SQLITE_ROW == step)
{
break_record_t record = {0};
/*
//永远不可能为空
if (&record == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "内存分配失败");
utarray_free(*breakRecord);
sqlite3_finalize(stmt);
return 1;
}
*/
record.dbId = sqlite3_column_int64(stmt, 0);
strncpy((char *)record.reason, (const char *)sqlite3_column_text(stmt, 1), MAX_NAME_BUF_LEN - 1);
record.reason[MAX_NAME_BUF_LEN - 1] = '\0'; // 确保字符串结尾为\0
record.isUploaded = sqlite3_column_int(stmt, 2);
utarray_push_back(*breakRecord, &record);
step = sqlite3_step(stmt);
}
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL执行失败: %s", sqlite3_errmsg(business_db));
utarray_free(*breakRecord);
sqlite3_finalize(stmt);
return 1;
}
sqlite3_finalize(stmt);
return 0;
}
/*****************************************************************************
* @brief 插入MQTT断开记录包括中断原因、是否已上传以及断开开始和结束时间。返回插入记录的ID。
* @param[in] breakRecord: 包含MQTT断开信息的结构体指针该结构体必须包含中断原因(reason)
* @return 插入记录的ID-1-插入记录失败包括SQL预编译失败和SQL执行失败
*****************************************************************************/
long long kit_insert_break_record(break_record_t *breakRecord)
{
sqlite3_stmt *stmt = NULL;
const char *sql = "INSERT INTO break_record(reason, is_uploaded, start_time) VALUES(?, 0, strftime('%Y-%m-%d %H:%M:%S', 'now'));"; // 注意分号
if (SQLITE_OK != sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
return -1; // 返回-1表示失败
}
sqlite3_bind_text(stmt, 1, (const char *)breakRecord->reason, -1, SQLITE_STATIC);
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return -1; // 返回-1表示失败
}
long long last_insert_rowid = sqlite3_last_insert_rowid(business_db); // 获取最后插入行的rowid
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return last_insert_rowid; // 返回插入记录的ID
}
/*****************************************************************************
* @brief 更新MQTT断开记录的上传状态。此函数更新数据库中MQTT断开记录的`is_uploaded`字段。
* @param[in] breakRecord: 指向包含MQTT断开信息的结构体指针。该结构体必须包含断开记录在数据库中的ID (`dbId`) 和新的上传状态 (`isUploaded`)。
* @return 0-成功更新记录1-更新记录失败包括SQL语句预编译失败、SQL语句执行失败或未找到匹配的记录。如果SQL语句执行成功但没有更新任何行例如`dbId`不存在则返回1并记录警告日志。
*****************************************************************************/
int kit_update_break_record(break_record_t *breakRecord)
{
int ret = 0;
sqlite3_stmt *stmt = NULL;
const char *sql = sqlite3_mprintf("UPDATE break_record SET is_uploaded = ? WHERE id = ?");
if (SQLITE_OK != sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(business_db));
return 1;
}
sqlite3_bind_int(stmt, 1, breakRecord->isUploaded);
sqlite3_bind_int64(stmt, 2, breakRecord->dbId);
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (step != SQLITE_DONE)
{
if (step == SQLITE_ERROR)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s", sqlite3_errmsg(business_db));
}
else if (step == SQLITE_ROW)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "UPDATE操作中出现意外的SQLITE_ROW返回值");
}
else
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL更新失败错误代码: %d", step);
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 检查是否实际更新了任何行。SQLITE_DONE 并不保证这一点。
if (sqlite3_changes(business_db) == 0)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "没有更新任何行。请检查 dbId 是否正确。");
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return ret;
}
/*****************************************************************************
* @brief 批量插入MQTT断联时的历史数据存储
* @param[in] breakDbIdMQTT断联的数据库Id
* @param[in] timedDatas: 用于历史数据存储的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_insert_break_data_storage(uint64_t breakDbId, UT_array **timedDatas)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *sql = "INSERT INTO break_data_storage(break_id, content, is_uploaded, created_time) VALUES (?, ?, 0, strftime('%Y-%m-%d %H:%M:%S', 'now'))";
int ret = 0;
int total_count = utarray_len(*timedDatas); // 总记录数
int batch_size = 100; // 每批插入的条数
int i;
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
return 1;
}
// 预编译SQL语句
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
return 1;
}
// 批量插入
for (i = 0; i < total_count; i++)
{
break_data_storage_t *timedData = (break_data_storage_t *)utarray_eltptr(*timedDatas, i);
// 绑定参数
sqlite3_bind_int64(stmt, 1, breakDbId);
sqlite3_bind_text(stmt, 2, (const char *)timedData->content, -1, SQLITE_STATIC);
// 执行插入
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 重置语句,以便下次使用
sqlite3_reset(stmt);
// 每 batch_size 条记录提交一次事务
if ((i + 1) % batch_size == 0)
{
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 重新开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法重新开始事务: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
}
}
// 提交剩余数据
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex); // 解锁
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 释放预编译语句
sqlite3_finalize(stmt);
return 0; // 成功
}
/*****************************************************************************
* @brief 根据断开记录ID和行号范围查询断开数据。
* @param[in] breakDbId: MQTT断开记录的数据库ID。
* @param[out] timedDatas: 用于存储查询数据的UT_array指针。必须在调用之前初始化。
* @return 0-成功1-失败(数据库错误或其他问题)。
*****************************************************************************/
int kit_query_break_data(uint64_t breakDbId, UT_array **timedDatas)
{
sqlite3_stmt *stmt = NULL;
char *sql = NULL;
int ret = 0;
// 构造SQL查询语句使用LIMIT子句进行分页
// 将 SELECT 子句改为选择 id 和 content 两个字段
sql = sqlite3_mprintf("SELECT id, content FROM break_data_storage WHERE break_id = %d AND is_uploaded = 0 ORDER BY created_time LIMIT 100", breakDbId);
if (SQLITE_OK != sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(business_db));
sqlite3_free(sql);
return 1;
}
while ((ret = sqlite3_step(stmt)) == SQLITE_ROW)
{
break_data_storage_t timedData = {0};
snprintf((char *)timedData.content, MAX_JSON_STR_LEN, "%s", (uint8_t *)sqlite3_column_text(stmt, 1));
uint64_t id = sqlite3_column_int(stmt, 0);
timedData.breakDbId = id;
utarray_push_back(*timedDatas, &timedData);
}
if (ret != SQLITE_DONE && ret != SQLITE_ROW)
{ // 添加对非SQLITE_DONE和SQLITE_ROW情况的检查
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s, 错误码: %d", sqlite3_errmsg(business_db), ret);
sqlite3_finalize(stmt);
sqlite3_free(sql);
return 1;
}
sqlite3_finalize(stmt);
sqlite3_free(sql);
return 0;
}
/*****************************************************************************
* @brief 批量更新断开数据的上传状态,使用事务保证数据一致性。
* @param[in] break_id: MQTT断开记录的break_id。
* @param[in] ids: storage的id数组。
* @param[in] length: storage的id数组长度。
* @return 0-成功1-失败
*****************************************************************************/
int kit_update_break_data(uint16_t break_id, uint64_t ids[], int length)
{
if (length <= 0 || ids == NULL)
{
// 如果传入的id数组为空或长度为0, 无需执行更新。
return 0;
}
sqlite3_stmt *stmt = NULL;
int ret = 0;
int i = 0;
char *sql = NULL;
char *in_clause = NULL;
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务
if (SQLITE_OK != sqlite3_exec(business_db, "BEGIN TRANSACTION", NULL, NULL, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "开始事务失败: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex);
return 1;
}
// 构建 IN 子句字符串
in_clause = sqlite3_mprintf("(");
for (i = 0; i < length; i++)
{
in_clause = sqlite3_mprintf("%s%s?", in_clause, (i > 0) ? "," : "");
}
in_clause = sqlite3_mprintf("%s)", in_clause);
// 创建 SQL 语句
sql = sqlite3_mprintf("UPDATE break_data_storage SET is_uploaded = 1 WHERE break_id = ? AND id IN %s", in_clause);
sqlite3_free(in_clause);
if (SQLITE_OK != sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句预编译失败: %s", sqlite3_errmsg(business_db));
sqlite3_free(sql);
sqlite3_exec(business_db, "ROLLBACK TRANSACTION", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex);
return 1;
}
sqlite3_free(sql);
// 绑定 break_id 参数
sqlite3_bind_int(stmt, 1, break_id);
// 绑定 ID 参数
for (i = 0; i < length; i++)
{
sqlite3_bind_int64(stmt, i + 2, ids[i]); // IDs 从第二个参数开始
}
// 执行 SQL 语句
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句执行失败: %s, 错误码: %d", sqlite3_errmsg(business_db), ret);
sqlite3_finalize(stmt);
sqlite3_exec(business_db, "ROLLBACK TRANSACTION", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex);
return 1;
}
sqlite3_finalize(stmt);
// 提交事务
if (SQLITE_OK != sqlite3_exec(business_db, "COMMIT TRANSACTION", NULL, NULL, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex);
return 1;
}
pthread_mutex_unlock(&business_db_mutex);
return 0;
}
/*****************************************************************************
* @brief 批量插入插入功率数据
* @param[in] powerDatas: 用于存储备份数据的 UT_array 指针
* @return 0-成功 1-失败
*****************************************************************************/
int kit_insert_power_data(UT_array *powerDatas)
{
sqlite3_stmt *stmt = NULL; // 预编译语句句柄
const char *sql = "INSERT INTO power_data_storage(dev_type, dev_code, point_code, value, created_time) VALUES(?, ?, ?, ?, strftime('%Y-%m-%d %H:%M:%S', 'now'));";
int ret = 0;
int total_count = utarray_len(powerDatas); // 总记录数
int batch_size = 100; // 每批插入的条数
int i;
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 预编译SQL语句
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 批量插入
for (i = 0; i < total_count; i++)
{
power_data_storage_t *powerData = (power_data_storage_t *)utarray_eltptr(powerDatas, i);
// 绑定参数
sqlite3_bind_int(stmt, 1, powerData->devType);
sqlite3_bind_int(stmt, 2, powerData->devId);
sqlite3_bind_int(stmt, 3, powerData->pointId);
sqlite3_bind_double(stmt, 4, powerData->value);
// 执行插入
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 重置语句,以便下次使用
sqlite3_reset(stmt);
// 每 batch_size 条记录提交一次事务
if ((i + 1) % batch_size == 0)
{
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
// 重新开始事务
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法重新开始事务: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 1;
}
}
}
// 提交剩余数据
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
pthread_mutex_unlock(&business_db_mutex); // 解锁
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1;
}
// 释放预编译语句
sqlite3_finalize(stmt);
utarray_free(powerDatas);
return 0; // 成功
}
/*****************************************************************************
* @brief 根据日期时间条件删除该日期时间之前power_data_storage的数据
* @param[in] date_time_string日期时间
* @return 0-成功 1-失败
*****************************************************************************/
int kit_del_power_data_by_date(const char *date_time_string)
{
sqlite3_stmt *stmt = NULL; // SQLite语句句柄
int ret = 0; // 返回值
char sql[256]; // SQL语句缓冲区
// 参数校验确保日期字符串不为NULL
if (date_time_string == NULL)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "日期时间字符串为NULL");
return 1;
}
// 使用sqlite3_snprintf构建SQL语句防止SQL注入
sqlite3_snprintf(sizeof(sql), sql, "DELETE FROM power_data_storage WHERE created_time < '%q';", date_time_string);
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
// 开始事务,确保数据一致性
ret = sqlite3_exec(business_db, "BEGIN TRANSACTION;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "无法开始事务: %s", sqlite3_errmsg(business_db));
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1; // 直接返回,不继续执行
}
// 准备SQL语句避免直接使用sqlite3_exec
ret = sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL语句准备失败: %s SQL: %s", sqlite3_errmsg(business_db), sql);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
return 1;
}
// 执行SQL语句
ret = sqlite3_step(stmt);
if (ret != SQLITE_DONE) // SQLITE_DONE表示成功执行完语句
{
KITLOG(LOG_KIT_EN, ERROR_EN, "删除数据失败: %s SQL: %s", sqlite3_errmsg(business_db), sql);
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt); // 释放SQLite语句句柄
return 1;
}
// 提交事务,确保操作生效
ret = sqlite3_exec(business_db, "COMMIT;", NULL, NULL, NULL);
if (ret != SQLITE_OK)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "提交事务失败: %s", sqlite3_errmsg(business_db));
sqlite3_exec(business_db, "ROLLBACK;", NULL, NULL, NULL); // 回滚事务
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt); // 释放SQLite语句句柄
return 1;
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
// 释放SQLite语句句柄避免内存泄漏
sqlite3_finalize(stmt);
return 0; // 返回0表示成功
}
/*****************************************************************************
* @brief 获取指定表中特定dbId的记录数量
* @param[in] db: SQLite3 数据库连接对象
* @param[in] tabName: 查询表的名称
* @param[in] dbid: 用于查询记录的dbId值
* @return 记录数量(成功返回记录数量;失败返回-1
*****************************************************************************/
int kit_get_record_count_by_dbid(sqlite3 *db, const char *tabName, uint64_t dbid)
{
sqlite3_stmt *stmt;
char sql[256];
int result = 0;
int rc;
// 使用 snprintf 合成 SQL 查询字符串
snprintf(sql, sizeof(sql), "SELECT COUNT(*) FROM %s WHERE db_id = ?", tabName);
// 预编译SQL语句
rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
if (rc != SQLITE_OK)
{
printf("Failed to prepare statement: %s\n", sqlite3_errmsg(db));
return -1;
}
// 绑定参数
sqlite3_bind_int64(stmt, 1, dbid);
// 执行SQL查询
if ((rc = sqlite3_step(stmt)) == SQLITE_ROW)
{
result = sqlite3_column_int(stmt, 0);
}
else
{
printf("Failed to execute query: %s\n", sqlite3_errmsg(db));
}
// 清理和释放资源
sqlite3_finalize(stmt);
return result;
}
/*****************************************************************************
* @brief 插入北向断开记录,记录最新连接状态。
* @param[in] connectStatus: 连接状态
* @return 0-成功 1-失败。
*****************************************************************************/
int kit_insert_lost_contact_record(const char* connectStatus)
{
sqlite3_stmt *stmt = NULL;
const char *sql = "INSERT INTO lost_contact_record(connect_status, created_time) VALUES(?, strftime('%Y-%m-%d %H:%M:%S', 'now'));";
if (SQLITE_OK != sqlite3_prepare_v2(business_db, sql, -1, &stmt, NULL))
{
KITLOG(LOG_KIT_EN, ERROR_EN, "SQL预编译失败: %s", sqlite3_errmsg(business_db));
return 1; // 返回1表示失败
}
sqlite3_bind_text(stmt, 1, connectStatus, -1, SQLITE_STATIC);
pthread_mutex_lock(&business_db_mutex); // 加锁,确保单线程访问
int step = sqlite3_step(stmt);
if (SQLITE_DONE != step)
{
KITLOG(LOG_KIT_EN, ERROR_EN, "插入数据失败: %s", sqlite3_errmsg(business_db));
sqlite3_finalize(stmt);
return 1; // 返回1表示失败
}
pthread_mutex_unlock(&business_db_mutex); // 解锁
sqlite3_finalize(stmt);
return 0;
}