lisonwu/dcapp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新 insert update接口

Browse Source
This commit is contained in:
why
2025-02-27 14:10:15 +08:00
parent 583ade8cf3
commit b352aeacef
5 changed files with 615 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -24,6 +24,7 @@ add_executable(dcapp
${PROJECT_SOURCE_DIR}/datacenter/src/mqttClient.cpp
${PROJECT_SOURCE_DIR}/datacenter/src/ToolUtils.cpp
${PROJECT_SOURCE_DIR}/datacenter/bizService/F_InsertRecord.cpp
${PROJECT_SOURCE_DIR}/datacenter/bizService/F_UpdateRecord.cpp
${DC_SRC} ${DS_SRC} ${DQ_SRC})
datacenter/bizService/F_InsertRecord.cpp
+278
View File
@@ -0,0 +1,278 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-21 14:03:54
* @LastEditors: why
* @LastEditTime: 2025-02-27 11:59:36
*/
#include <iostream>
#include <chrono>
#include <iomanip>
#include <sstream>
#include <string>
#include <ctime>
#include "json.hpp"
#include "F_InsertRecord.h"
#include "bizservice.h"
#include "sqlite3/sqlite3.h"
using namespace std;
using json = nlohmann::json ;// for convenience
typedef unsigned char BYTE;
typedef struct topic{
std::string src;
std::string des;
std::string sid;
std::string fid;
} topic_type;
REGISTER_BIZSERVICE(InsertRecord);
InsertRecord::InsertRecord(void){
handpara = nullptr;
initpara = nullptr;
}
InsertRecord::~InsertRecord(void){
}
void InsertRecord::Init(initparam_t *param){
initpara = param;
}
void hexstr_to_str(void *dst, const void *src, int src_len)
{
int i;
BYTE low, high, ch;
BYTE *dst_ptr = (BYTE *) dst;
const BYTE *src_ptr = (const BYTE *) src;
for (i = 0; i < src_len / 2; i++)
{
ch = *src_ptr++;
if (ch >= '0' && ch <= '9')
high = ch - '0';
else if (ch >= 'A' && ch <= 'F')
high = ch - 'A' + 10;
else if (ch >= 'a' && ch <= 'f')
high = ch - 'a' + 10;
else
break;
ch = *src_ptr++;
if (ch >= '0' && ch <= '9')
low = ch - '0';
else if (ch >= 'A' && ch <= 'F')
low = ch - 'A' + 10;
else if (ch >= 'a' && ch <= 'f')
low = ch - 'a' + 10;
else
break;
*dst_ptr++ = high * 16 + low; //hex
}
}
void str_to_hexstr(void *dst, const void *src, int src_len)
{
int i;
BYTE low, high, ch;
char *dst_ptr = (char *) dst;
const char *src_ptr = (const char *) src;
const char *str = "0123456789abcdef";
for (i = 0; i < src_len; i++)
{
ch = *src_ptr++;
high = (ch >> 4) & 0x0f;
low = ch & 0x0f;
*dst_ptr++ = str[high];
*dst_ptr++ = str[low];
}
}
int encode_blob(void *dst, int max_len, const void *src, int src_len)
{
int dst_len = 0;
// may be changed to base64 encode
if (max_len >= src_len * 2 + 1)
{
dst_len = src_len * 2;
str_to_hexstr(dst, src, src_len);
}
((BYTE *) dst)[dst_len] = 0x00;
return dst_len;
}
int decode_blob(void *dst, int max_len, const void *src)
{
// may be changed to base64 decode
int src_len, dst_len = 0;
src_len = strlen((char *) src);
if (max_len >= src_len / 2)
{
dst_len = src_len / 2;
hexstr_to_str(dst, src, src_len);
}
return dst_len;
}
std::string InsertRecord::now_to_string() {
// 获取系统时钟时间点
auto time_now = std::chrono::system_clock::now();
// 将时间点转成time_t类型的值 (秒)
std::time_t tt = std::chrono::system_clock::to_time_t(time_now);
// 转换成tm结构体形式
struct tm* ptm = gmtime(&tt); // 或者 localtime(&tt) 如果想要本地时间
ptm->tm_hour += 8; //utc -> cst
// 计算出毫秒部分
long milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(
time_now.time_since_epoch()
).count() % 1000;
char buffer[40];
strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%S", ptm);
// 添加上毫秒和时区信息
sprintf(buffer + strlen(buffer), ".%03ld+0800", milliseconds);
return std::string(buffer);
}
time_t InsertRecord::string_to_now(std::string stamp) {
struct tm tm={};
std::istringstream ss(stamp);
ss >> std::get_time(&tm,"%Y-%m-%dT%H:%M:%S");
if(ss.fail()){
return -1;
}
return mktime(&tm);
}
int InsertRecord::SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result){
sqlite3 *db = (sqlite3 *)handpara;
char *errMsg = nullptr;
const char *sql;
int rc;
int resp = 0;
// 开启事务
if (sqlite3_exec(db, "BEGIN TRANSACTION;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务开启失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
return -1;
}
// 执行批量插入操作
sqlite3_stmt *stmt;
int idx = 0;
for(const auto &k:sql_list){
std::cout << "cmd:"<<k.c_str()<<std::endl;
if (sqlite3_prepare_v2(db, k.c_str(), -1, &stmt, nullptr) != SQLITE_OK) {
std::cerr << "准备语句失败: " << sqlite3_errmsg(db) << std::endl;
continue;
}
rc = sqlite3_step(stmt);
if (rc != SQLITE_DONE) {
std::cerr << "插入失败: " << sqlite3_errmsg(db) << std::endl;
result[idx] = "-1";
}else
result[idx] = "0"; //SUCCESS
sqlite3_reset(stmt); // 重置语句
}
// 提交事务
if (sqlite3_exec(db, "COMMIT;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务提交失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
resp = -1;
}
// 清理
sqlite3_finalize(stmt);
return resp;
}
int InsertRecord::HandleMsg(const msg_t msg, msg_t &outMsg,handparam_t *param){
topic.clear();
topic_split(msg.topic,'/');
json value,resp;
int ret = 0;
int afn_flag = 0;
std::vector<std::string> sql_list;
std::vector<std::string> result;
sql_list.clear();
result.clear();
std::string table_name("dayfrozen");
try{
value = json::parse(msg.payload);
}
catch(const std::exception &e){
std::cout << "topic "<<msg.topic.c_str()<<" parse error:"<<e.what();
return -1;//format fail
}
printf("insert request:%s \n",msg.payload.c_str());
if(!value.is_object())return -1;
if(value.contains("token")){
resp["token"] = value["token"];
}
if(value.contains("time")){
;// value["time"]
}
if(value.contains("data") && value["data"].contains("dataarray")){
json subvalue = value["data"]["dataarray"];
if(subvalue.is_array()){
int idx = 0;
for(const auto &k:subvalue){
std::stringstream ss;
if(k["afn"].get<std::string>() != "0D"){
afn_flag = -1;
continue;
}
//根据afn/idclass 确定数据库表位置 待添加
ss << "insert into " <<table_name.c_str() <<" (idx,metid,dataid,frozentime,createdtime,source,crcaux,databuff) values ("\
<< 0 << ","\
<<std::stoi(k["metid"].get<std::string>())<<","\
<<k["dataid"].get<std::string>().c_str()<<","\
<<string_to_now(k["frozentime"].get<std::string>())<<","\
<<time(NULL)<<","\
<<std::stoi(k["crcaux"].get<std::string>())<<","\
<<k["databuff"].get<std::string>().c_str()<<");";
printf("insert sql:%s \n",ss.str().c_str());
sql_list.push_back(ss.str());
ss.clear();
}
}
}
// ret = SQLBatch_exc(sql_list,result);
resp["time"]=now_to_string();
if(ret == -1 || afn_flag == -1)
resp["ack"]="false";
else if(ret == 0)
resp["ack"]="true";
resp["data"]["dataidex"]=result;
std::stringstream ss;
ss << topic[1].c_str()<<"/"<<topic[0].c_str()<<"/"\
<<topic[2].c_str()<<"/"<<topic[3].c_str();
outMsg.topic = ss.str();
outMsg.timestamp = time(NULL);
outMsg.payload = resp.dump();
std::cout << "resp InsertRecord payload:"<<resp.dump().c_str()<<std::endl;
return 1;
}
std::string InsertRecord::GetFID(){
std::string value("F-InsertRecord");
return value;
}
void InsertRecord::topic_split(const std::string& str, char delimiter) {
std::string token; // 临时存储每个子字符串
std::istringstream tokenStream(str); // 使用字符串流处理输入字符串
// 按分隔符逐个读取子字符串
while (std::getline(tokenStream, token, delimiter)) {
topic.push_back(token); // 将子字符串添加到结果中
}
}
datacenter/bizService/F_InsertRecord.h
+28
View File
@@ -0,0 +1,28 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-21 14:04:30
* @LastEditors: why
* @LastEditTime: 2025-02-26 15:13:26
*/
#include <string>
#include <vector>
#include "bizservice.h"
class InsertRecord : public IBizService
{
public:
void Init(initparam_t *param);
int HandleMsg(const msg_t msg,msg_t &outMsg, handparam_t *param);
std::string GetFID();
InsertRecord();
~InsertRecord();
private:
handparam_t * handpara;
initparam_t * initpara;
std::vector<std::string> topic;//0：dst 1:src 2:sid 3:fid
void topic_split(const std::string& str, char delimiter);
int SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result);
std::string now_to_string();
time_t string_to_now(std::string stamp);
};
datacenter/bizService/F_UpdateRecord.cpp
+281
View File
@@ -0,0 +1,281 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-25 11:02:37
* @LastEditors: why
* @LastEditTime: 2025-02-27 11:59:30
*/
#include <iostream>
#include <chrono>
#include <iomanip>
#include <sstream>
#include <string>
#include <ctime>
#include "json.hpp"
#include "F_UpdateRecord.h"
#include "sqlite3/sqlite3.h"
using namespace std;
using json = nlohmann::json; // for convenience
typedef unsigned char BYTE;
typedef struct topic{
std::string src;
std::string des;
std::string sid;
std::string fid;
} topic_type;
REGISTER_BIZSERVICE(UpdateRecord);
UpdateRecord::UpdateRecord(void){
handpara = nullptr;
initpara = nullptr;
}
void hexstr_to_str(void *dst, const void *src, int src_len)
{
int i;
BYTE low, high, ch;
BYTE *dst_ptr = (BYTE *) dst;
const BYTE *src_ptr = (const BYTE *) src;
for (i = 0; i < src_len / 2; i++)
{
ch = *src_ptr++;
if (ch >= '0' && ch <= '9')
high = ch - '0';
else if (ch >= 'A' && ch <= 'F')
high = ch - 'A' + 10;
else if (ch >= 'a' && ch <= 'f')
high = ch - 'a' + 10;
else
break;
ch = *src_ptr++;
if (ch >= '0' && ch <= '9')
low = ch - '0';
else if (ch >= 'A' && ch <= 'F')
low = ch - 'A' + 10;
else if (ch >= 'a' && ch <= 'f')
low = ch - 'a' + 10;
else
break;
*dst_ptr++ = high * 16 + low; //hex
}
}
void str_to_hexstr(void *dst, const void *src, int src_len)
{
int i;
BYTE low, high, ch;
char *dst_ptr = (char *) dst;
const char *src_ptr = (const char *) src;
const char *str = "0123456789abcdef";
for (i = 0; i < src_len; i++)
{
ch = *src_ptr++;
high = (ch >> 4) & 0x0f;
low = ch & 0x0f;
*dst_ptr++ = str[high];
*dst_ptr++ = str[low];
}
}
int encode_blob(void *dst, int max_len, const void *src, int src_len)
{
int dst_len = 0;
// may be changed to base64 encode
if (max_len >= src_len * 2 + 1)
{
dst_len = src_len * 2;
str_to_hexstr(dst, src, src_len);
}
((BYTE *) dst)[dst_len] = 0x00;
return dst_len;
}
int decode_blob(void *dst, int max_len, const void *src)
{
// may be changed to base64 decode
int src_len, dst_len = 0;
src_len = strlen((char *) src);
std::cout << "decode_blob src len "<< src_len<<std::endl;
if (max_len >= src_len / 2)
{
dst_len = src_len / 2;
hexstr_to_str(dst, src, src_len);
}
return dst_len;
}
UpdateRecord::~UpdateRecord(void){
}
std::string search_table_db(int afn,unsigned int dataid){
std::string table;
return table;
}
void UpdateRecord::Init(initparam_t *param){
initpara = param;
}
std::string UpdateRecord::now_to_string() {
// 获取系统时钟时间点
auto time_now = std::chrono::system_clock::now();
// 将时间点转成time_t类型的值 (秒)
std::time_t tt = std::chrono::system_clock::to_time_t(time_now);
// 转换成tm结构体形式
struct tm* ptm = gmtime(&tt); // 或者 localtime(&tt) 如果想要本地时间
ptm->tm_hour += 8; //utc -> cst
// 计算出毫秒部分
long milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(
time_now.time_since_epoch()
).count() % 1000;
char buffer[40];
strftime(buffer, sizeof(buffer), "%Y-%m-%dT%H:%M:%S", ptm);
// 添加上毫秒和时区信息
sprintf(buffer + strlen(buffer), ".%03ld+0800", milliseconds);
return std::string(buffer);
}
time_t UpdateRecord::string_to_now(std::string stamp) {
struct tm t;
std::istringstream ss(stamp);
ss >> std::get_time(&t,"%Y-%m-%dT%H:%M:%S");
if(ss.fail())return -1;
return mktime(&t);
}
int UpdateRecord::SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result){
sqlite3 *db = (sqlite3 *)handpara;
char *errMsg = nullptr;
const char *sql;
int rc;
int resp = 0;
// 开启事务
if (sqlite3_exec(db, "BEGIN TRANSACTION;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务开启失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
return -1;
}
// 执行批量插入操作
sqlite3_stmt *stmt;
int idx = 0;
for(const auto &k:sql_list){
std::cout << "cmd:"<<k.c_str()<<std::endl;
if (sqlite3_prepare_v2(db, k.c_str(), -1, &stmt, nullptr) != SQLITE_OK) {
std::cerr << "准备语句失败: " << sqlite3_errmsg(db) << std::endl;
continue;
}
rc = sqlite3_step(stmt);
if (rc != SQLITE_DONE) {
std::cerr << "插入失败: " << sqlite3_errmsg(db) << std::endl;
result[idx] = "-1";
}else
result[idx] = "0"; //SUCCESS
sqlite3_reset(stmt); // 重置语句
}
// 提交事务
if (sqlite3_exec(db, "COMMIT;", nullptr, nullptr, &errMsg) != SQLITE_OK) {
std::cerr << "事务提交失败: " << errMsg << std::endl;
sqlite3_free(errMsg);
resp = -1;
}
// 清理
sqlite3_finalize(stmt);
return resp;
}
int UpdateRecord::HandleMsg(const msg_t msg, msg_t &outMsg,handparam_t *param){
topic.clear();
topic_split(msg.topic,'/');
json value,resp;
int ret = 0;
int afn_flag = 0, update_flag = 0;;
std::vector<std::string> sql_list;
std::vector<std::string> result;
sql_list.clear();
result.clear();
std::string table_name ;
try{
value = json::parse(msg.payload);
}
catch(const std::exception &e){
std::cout << "topic "<<msg.topic.c_str()<<" prase error:"<<e.what();
return -1;//format fail
}
if(!value.is_object())return -1;
if(value.contains("token")){
resp["token"] = value["token"];
}
if(value.contains("time")){
;// value["time"]
}
if(value.contains("data")){
json subvalue = value["data"];
int afn = 0;
int dataid = 0;
decode_blob(&dataid,4, subvalue["dataid"].get<std::string>().c_str());
decode_blob(&afn,4,subvalue["afn"].get<std::string>().c_str());
std::stringstream ss;
//table_name =search_table_db(afn,dataid);
table_name = "frozen";
std::cout << "afn:"<<afn<<std::endl;
if(!(afn==0x0d | afn==0x0e |afn == 0x12 |afn == 0x13))
{
afn_flag = -1;
goto resp_process;
}
if(std::stoi(subvalue["updatetype"].get<std::string>()) !=1){
update_flag = -1;
goto resp_process;
}
std::cout << __func__<<","<<__LINE__<<std::endl;
ss << "replace into " <<table_name.c_str() <<" (idx,metid,dataid,frozentime,createdtime,source,crcaux,databuff) values ("\
<< 0 << ","\
<<std::stoi(subvalue["metid"].get<std::string>())<<","\
<<subvalue["dataid"].get<std::string>().c_str()<<","\
<<string_to_now(subvalue["updatefilter"].get<std::string>())<<","\
<<time(NULL)<<","\
<<std::stoi(subvalue["crcaux"].get<std::string>())<<","\
<<subvalue["databuff"].get<std::string>().c_str()<<");";
printf("update sql:%s \n",ss.str().c_str());
sql_list.push_back(ss.str());
ss.clear();
}
ret = SQLBatch_exc(sql_list,result);
resp_process:
resp["time"]=now_to_string();
if(ret == -1 || afn_flag == -1 || update_flag == -1)
resp["ack"]="false";
else if(ret == 0)
resp["ack"]="true";
resp["data"]["dataidex"]=result;
std::stringstream ss;
ss << topic[1].c_str()<<"/"<<topic[0].c_str()<<"/"\
<<topic[2].c_str()<<"/"<<topic[3].c_str();
outMsg.topic = ss.str();
outMsg.timestamp = time(NULL);;
outMsg.payload = resp.dump();
std::cout << "resp UpdateRecord payload:"<<resp.dump().c_str()<<std::endl;
return 1;
}
std::string UpdateRecord::GetFID(){
std::string tmp("F-UpdateRecord");
return tmp;
}
void UpdateRecord::topic_split(const std::string& str, char delimiter) {
std::string token; // 临时存储每个子字符串
std::istringstream tokenStream(str); // 使用字符串流处理输入字符串
// 按分隔符逐个读取子字符串
while (std::getline(tokenStream, token, delimiter)) {
topic.push_back(token); // 将子字符串添加到结果中
}
}
datacenter/bizService/F_UpdateRecord.h
+27
View File
@@ -0,0 +1,27 @@
/*
* @Description: file content
* @Author: why
* @version:
* @Date: 2025-02-25 11:03:17
* @LastEditors: why
* @LastEditTime: 2025-02-27 11:13:51
*/
#include <string>
#include <vector>
#include "bizservice.h"
class UpdateRecord:public IBizService {
public:
void Init(initparam_t *param);
int HandleMsg(const msg_t msg, msg_t &outMsg,handparam_t *param = NULL);
std::string GetFID();
UpdateRecord();
~UpdateRecord();
private:
handparam_t * handpara;
initparam_t * initpara;
std::vector<std::string> topic;//0：dst 1:src 2:sid 3:fid
void topic_split(const std::string& str, char delimiter);
int SQLBatch_exc(std::vector<std::string> &sql_list,std::vector<std::string> &result);
std::string now_to_string();
time_t string_to_now(std::string stamp);
};