汤浩
282 lines
8.3 KiB
C++
282 lines
8.3 KiB
C++
/*
|
|
* @Description: file content
|
|
* @Author: why
|
|
* @version:
|
|
* @Date: 2025-02-25 11:02:37
|
|
* @LastEditors: why
|
|
* @LastEditTime: 2025-02-28 17:56:00
|
|
*/
|
|
#include <iostream>
|
|
#include <chrono>
|
|
#include <iomanip>
|
|
#include <sstream>
|
|
#include <string>
|
|
#include <ctime>
|
|
#include "common/inc/json.hpp"
|
|
#include "F_UpdateRecord.h"
|
|
#include "common/inc/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("F-UpdateRecord",UpdateRecord);
|
|
UpdateRecord::UpdateRecord(void){
|
|
handpara = nullptr;
|
|
initpara = nullptr;
|
|
}
|
|
static 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
|
|
}
|
|
}
|
|
|
|
static 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];
|
|
}
|
|
}
|
|
static 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;
|
|
}
|
|
|
|
static 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); // 将子字符串添加到结果中
|
|
}
|
|
|
|
}
|