lnk/microser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add function:upload and download device file ,modify interface function fix memleak

Browse Source
This commit is contained in:
lnk
2026-03-12 15:28:17 +08:00
parent d1ed49412c
commit 0acc58bbe1
8 changed files with 980 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
json/create_json.cpp
View File

@@ -2931,6 +2931,7 @@ void Set_xml_databaseinfo(char* MODEL_ID, char* TMNL_TYPE, char* FILE_PATH, char
{
Xmldata* config2 = new Xmldata();
xmlinfo_list2.insert(type, config2);
xmlinfo_list2[type]->updataflag = true;
}
else
{
@@ -2952,35 +2953,30 @@ void Set_xml_databaseinfo(char* MODEL_ID, char* TMNL_TYPE, char* FILE_PATH, char
char file_name[256];
memset(file_name, 0, 256);
sprintf(file_name, "%s", FILE_NAME);
snprintf(file_name, sizeof(file_name), "%s", FILE_NAME);
file_name[sizeof(file_name) - 1] = '\0';
QString Qsavename;
Qsavename.append("/FeProject/dat/").append(id).append(".xml"); //本地保存路径
char save_name[256];
memset(save_name, 0, 256);
sprintf(save_name, "%s", Qsavename.toAscii().data());
snprintf(save_name, sizeof(save_name), "%s", Qsavename.toAscii().data());
save_name[sizeof(save_name) - 1] = '\0';
cout << file_name << "!!!!!!!!!!!!!!!!!!!!!!!!!!" << save_name << endl;
//mq日志
DIY_WARNLOG_CODE("process",LOG_CODE_ICD_AND_DOWNLOAD,"【WARN】前置获取到终端类型%s,该终端类型对应的映射文件为%s,映射文件将下载并保存在本地为%s",TMNL_TYPE,FILE_PATH,save_name);
//20241028 lnk 替换为文件下载web接口
//构造文件下载接口参数
//接口示例http://192.168.1.125:10215/file/download?filePath=/path/xxx.txt
// 调用web获取文件内容
char* fileContent = NULL;
//测试下载
//char downpath[128] = {"/home/pq/FeProject/src/pt61850netd_pqfe_lnk/download/123.txt"};
//char download[128] = {"{\"filename\":\"file_test.txt\"}"};
//SendJsonAPI_web("http://192.168.1.149:8091/file/download", "", download, &fileContent);
std::string fullPath = std::string("filePath=") + std::string(FILE_PATH);
//调试用
std::cout << "fullpath" << fullPath << std::endl;
SendJsonAPI_web(WEB_FILEDOWNLOAD, fullPath.c_str(), "", &fileContent);
if (fileContent != NULL) {
if (fileContent != NULL && fileContent[0] != '\0') {
// 创建并打开文件
//判断返回的是不是错误json响应
@@ -3459,18 +3455,19 @@ static void scanAndResendOfflineFiles(const std::string &dirPath)
// 尝试发送
char* ptr = NULL; // 接收返回
SendJsonAPI_web(WEB_EVENT, "", jsonContent.c_str(), &ptr);
if (ptr != NULL) {
if (ptr != NULL && ptr[0] != '\0') {
cJSON* j_r = cJSON_Parse(ptr);
if (j_r == NULL) {
std::cout << "old file send fail" << std::endl;
// 表示有响应，则可视为成功；根据项目需要可加更精细的判断
handleCommentResponse(std::string(ptr));
DIY_WARNLOG_CODE("process",LOG_CODE_TRANSIENT_COMM,"【WARN】前置重发暂态事件失败");
}
else{
// 表示有响应，则可视为成功；根据项目需要可加更精细的判断
handleCommentResponse(std::string(ptr));
DIY_WARNLOG_CODE("process",LOG_CODE_TRANSIENT_COMM,"【WARN】前置重发暂态事件成功");
@@ -3478,7 +3475,7 @@ static void scanAndResendOfflineFiles(const std::string &dirPath)
// 删除文件
remove(fileList[i].fileName.c_str());
free(j_r);
cJSON_Delete(j_r);
}
}
@@ -3524,7 +3521,7 @@ char* mp_id,char* Qvvr_rptname,char* devtype)
c_xmlcfg = xmlcfg;
}
if (strlen(mp_id) == 0) {
if (NULL == mp_id || strlen(mp_id) == 0 ) {
std::cout << "mp_id is null" << std::endl;
return 0;
}
@@ -3567,6 +3564,12 @@ char* mp_id,char* Qvvr_rptname,char* devtype)
}
char* json_string = cJSON_Print(root);
if (json_string == NULL) {
DIY_ERRORLOG_CODE(full_key_m_d,LOG_CODE_TRANSIENT_COMM,"【ERROR】监测点：%s暂态事件生成JSON字符串失败",mp_id);
std::cerr << "Failed to print JSON object." << std::endl;
cJSON_Delete(root);
return 0;
}
printf("%s\n", json_string); // 输出 JSON 字符串
// 发送到暂态接口
@@ -3575,7 +3578,7 @@ char* mp_id,char* Qvvr_rptname,char* devtype)
// ================ 插入新功能 =========================
// ********** 新增功能开始 **********
if(ptr != NULL)
if(ptr != NULL && ptr[0] != '\0')
{
cJSON* j_r = cJSON_Parse(ptr);
// 如果发送失败(j_r == NULL)，则把当前 json 存入指定目录(/FeProject/dat/qvvr/)
@@ -3604,13 +3607,13 @@ char* mp_id,char* Qvvr_rptname,char* devtype)
// 把 json_string 写入文件
if(!writeJsonToFile(fileName.c_str(), json_string)){
DIY_ERRORLOG_CODE(full_key_m_d,LOG_CODE_TRANSIENT_COMM,"【ERROR】监测点：%s无法将暂态时间为%lld的暂态事件写入本地缓存",start_tm,mp_id);
DIY_ERRORLOG_CODE(full_key_m_d,LOG_CODE_TRANSIENT_COMM,"【ERROR】监测点：%s无法将暂态时间为%lld的暂态事件写入本地缓存",mp_id,start_tm);
}
checkAndRemoveOldestIfNeeded(qvvrDir, 10LL * 1024 * 1024);
}
else{
free(j_r);
cJSON_Delete(j_r);
//后续处理
}
}
@@ -3623,10 +3626,11 @@ char* mp_id,char* Qvvr_rptname,char* devtype)
// ********** 新增功能结束 **********
// 下面继续原逻辑，不动，处理本次发送
if (ptr != NULL) {
if (ptr != NULL && ptr[0] != '\0') {
std::cout << "current qvvr handle response" << std::endl;
handleCommentResponse(std::string(ptr));
free(ptr);
ptr = NULL;
} else {
// 处理 ptr 为 NULL 的情况，例如日志记录或错误处理
std::cout << "Error: Received NULL response" << std::endl;
@@ -3652,7 +3656,7 @@ char* mp_id,char* Qvvr_rptname,char* devtype)
fileName += ".txt";
// 把 json_string 写入文件
if(!writeJsonToFile(fileName.c_str(), json_string)){
DIY_ERRORLOG_CODE(full_key_m_d,LOG_CODE_TRANSIENT_COMM,"【ERROR】监测点：%s无法将暂态时间为%lld的暂态事件写入本地缓存",start_tm,mp_id);
DIY_ERRORLOG_CODE(full_key_m_d,LOG_CODE_TRANSIENT_COMM,"【ERROR】监测点：%s无法将暂态时间为%lld的暂态事件写入本地缓存",mp_id,start_tm);
}
checkAndRemoveOldestIfNeeded(qvvrDir, 10LL * 1024 * 1024);

595
json/save2json.cpp
View File

@@ -46,6 +46,18 @@ int StringToInt(const std::string& str);
extern pthread_mutex_t mtx;//lnk20250115
extern void SendFileWeb(const std::string& strUrl,
const char* localpath,
const char* cloudpath,
char* wavepath);
extern int DownloadFileWeb(const std::string& strUrl,
const char* remotePath,
const char* localpath);
#ifdef __cplusplus
extern "C" {
@@ -64,6 +76,13 @@ extern "C" {
extern node_t* g_node; //lnk20241223
extern LD_info_t* find_LD_info_only_from_mp_id(char* mp_id);//lnk20241223
extern void print_terminal(const terminal* tmnl);
extern ST_RET mms_mvla_obtfile(MVL_NET_INFO *net_info,
ST_CHAR *srcfilename,
ST_CHAR *destfilename,
int iTimeout);
extern pt61850app_t *g_pt61850app;
#ifdef __cplusplus
}
#endif
@@ -121,13 +140,21 @@ extern std::string G_MQCONSUMER_KEY_RC;//key
extern std::string G_MQCONSUMER_TOPIC_SET;//topie_recall
extern std::string G_MQCONSUMER_TAG_SET;//tag
extern std::string G_MQCONSUMER_KEY_SET;//key
extern std::string G_MQCONSUMER_TOPIC_LOG;//topie_log
extern std::string G_MQCONSUMER_TAG_LOG;//tag
extern std::string G_MQCONSUMER_KEY_LOG;//key
extern std::string G_LOG_TOPIC;//topie
extern std::string G_LOG_TAG;//tag
extern std::string G_LOG_KEY;//key
extern std::string G_MQCONSUMER_TOPIC_FILE;//topie_file
extern std::string G_MQCONSUMER_TAG_FILE;//tag
extern std::string G_MQCONSUMER_KEY_FILE;//key
extern std::string Topic_Reply_Topic;
extern std::string Topic_Reply_Tag;
extern std::string Topic_Reply_Key;
extern std::string WEB_FILEUPLOAD;
extern std::string WEB_FILEDOWNLOAD;
bool showinshellflag =false;
@@ -143,6 +170,10 @@ static QMap<QString, json_block_data*> json_data_map;//CZY 2023-08-17 ww 2023年
static QMap<QString, json_block_data*> json_flicker_data_map;//CZY 2023-09-11 展Map，用于保存各条线路的闪变数据
static QMap<QString, json_block_data*> json_pst_data_map;//CZY 2023-09-11 展Map，用于保存各条线路的闪变数据
//////////////////////////////////////////////////////////////////////////////lnk20260310文件控制
pthread_mutex_t g_file_req_mutex = PTHREAD_MUTEX_INITIALIZER;
std::list<file_dir_req_t*> g_file_dir_req_list;
bool is_blank(const std::string& str)
{
for (std::string::const_iterator it = str.begin(); it != str.end(); ++it)
@@ -581,7 +612,8 @@ std::string extractDataJson(const char* inputJson) {
// 提取 "guid" 部分
cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
if (guidstr == NULL || guidstr->type != cJSON_String) {
std::cerr << "'guid' is missing or is not an array" << std::endl;
std::cerr << "'guid' is missing or is not an string" << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return "";
}
@@ -593,12 +625,19 @@ std::string extractDataJson(const char* inputJson) {
cJSON* data = cJSON_GetObjectItem(messageBody, "data");
if (data == NULL || data->type != cJSON_Array) {
std::cerr << "'data' is missing or is not an array" << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return "";
}
// 创建新的 JSON 数组对象，只包含 "data" 部分
cJSON* newJson = cJSON_CreateArray(); // 创建一个新的数组
if (newJson == NULL) {
std::cerr << "Failed to create new JSON array" << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return "";
}
// 将 "data" 数组中的元素逐个添加到新数组中
cJSON* dataItem = NULL;
@@ -610,6 +649,7 @@ std::string extractDataJson(const char* inputJson) {
char* newJsonString = cJSON_Print(newJson);
if (newJsonString == NULL) {
std::cerr << "Error printing new JSON" << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
cJSON_Delete(newJson);
return "";
@@ -620,6 +660,7 @@ std::string extractDataJson(const char* inputJson) {
// 清理内存
free(newJsonString);
cJSON_Delete(messageBody);
cJSON_Delete(root);
cJSON_Delete(newJson);
@@ -684,10 +725,12 @@ bool parseJsonMessageRT(const std::string& body, std::string& devSeries, std::st
} else {
std::cerr << "Missing expected fields in JSON message." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return false;
}
cJSON_Delete(messageBody);
cJSON_Delete(root); // 清理 JSON 对象
return true;
}
@@ -917,6 +960,7 @@ int parse_set(const std::string& json_str) {
cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
if (guidstr == nullptr) {
std::cout << "Missing 'guid' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -928,6 +972,7 @@ int parse_set(const std::string& json_str) {
cJSON* code = cJSON_GetObjectItem(messageBody, "code");
if (code == nullptr) {
std::cout << "Missing 'code' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -938,6 +983,7 @@ int parse_set(const std::string& json_str) {
cJSON* processNo = cJSON_GetObjectItem(messageBody, "processNo");
if (processNo == nullptr) {
std::cout << "Missing 'processNo' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -948,6 +994,7 @@ int parse_set(const std::string& json_str) {
cJSON* funtion = cJSON_GetObjectItem(messageBody, "fun");
if (funtion == nullptr) {
std::cout << "Missing 'fun' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -957,6 +1004,7 @@ int parse_set(const std::string& json_str) {
cJSON* front = cJSON_GetObjectItem(messageBody, "frontType");
if (front == nullptr) {
std::cout << "Missing 'frontType' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -965,6 +1013,7 @@ int parse_set(const std::string& json_str) {
if (index_value != g_front_seg_index && g_front_seg_index != 0) {
std::cout << "msg index:"<< index_value <<"doesnt match self index:" << g_front_seg_index << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
@@ -979,6 +1028,7 @@ int parse_set(const std::string& json_str) {
cJSON* num = cJSON_GetObjectItem(messageBody, "processNum");
if (num == nullptr) {
std::cout << "Missing 'processNum' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1037,6 +1087,7 @@ int parse_set(const std::string& json_str) {
cJSON* onlyip = cJSON_GetObjectItem(messageBody, "ip");
if (onlyip == nullptr) {
std::cout << "Missing 'ip' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1046,6 +1097,7 @@ int parse_set(const std::string& json_str) {
cJSON* index_item = cJSON_GetObjectItem(messageBody, "proindex");
if (index_item == nullptr) {
std::cout << "Missing 'proindex' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1078,6 +1130,7 @@ int parse_set(const std::string& json_str) {
}
// 释放 JSON 对象
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
@@ -1316,6 +1369,7 @@ int parse_log(const std::string& json_str) {
cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
if (guidstr == nullptr) {
std::cout << "Missing 'guid' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1327,6 +1381,7 @@ int parse_log(const std::string& json_str) {
cJSON* code = cJSON_GetObjectItem(messageBody, "code");
if (code == nullptr) {
std::cout << "Missing 'code' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1338,6 +1393,7 @@ int parse_log(const std::string& json_str) {
cJSON* process = cJSON_GetObjectItem(messageBody, "processNo");
if (process == nullptr) {
std::cout << "Missing 'processNo' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1349,6 +1405,7 @@ int parse_log(const std::string& json_str) {
cJSON* idstr = cJSON_GetObjectItem(messageBody, "id");
if (idstr == nullptr) {
std::cout << "Missing 'id' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1359,6 +1416,7 @@ int parse_log(const std::string& json_str) {
cJSON* levelstr = cJSON_GetObjectItem(messageBody, "level");
if (levelstr == nullptr) {
std::cout << "Missing 'level' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1369,6 +1427,7 @@ int parse_log(const std::string& json_str) {
cJSON* gradestr = cJSON_GetObjectItem(messageBody, "grade");
if (gradestr == nullptr) {
std::cout << "Missing 'grade' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1379,6 +1438,7 @@ int parse_log(const std::string& json_str) {
cJSON* logtypestr = cJSON_GetObjectItem(messageBody, "logtype");
if (logtypestr == nullptr) {
std::cout << "Missing 'logtype' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1389,6 +1449,7 @@ int parse_log(const std::string& json_str) {
cJSON* frontTypestr = cJSON_GetObjectItem(messageBody, "frontType");
if (frontTypestr == nullptr) {
std::cout << "Missing 'frontType' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1397,12 +1458,14 @@ int parse_log(const std::string& json_str) {
if (processNo != g_front_seg_index) {
std::cout << "msg index:"<< processNo <<"doesnt match self index:" << g_front_seg_index << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
if (frontType != subdir) {
std::cout << "msg frontType:"<< frontType <<"doesnt match self frontType:" << subdir << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
@@ -1439,6 +1502,7 @@ int parse_log(const std::string& json_str) {
}
// 释放 JSON 对象
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
@@ -1479,6 +1543,7 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
cJSON* code = cJSON_GetObjectItem(messageBody, "code");
if (code == nullptr) {
std::cout << "Missing 'code' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1490,6 +1555,7 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
cJSON* process = cJSON_GetObjectItem(messageBody, "processNo");
if (process == nullptr) {
std::cout << "Missing 'processNo' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1501,6 +1567,7 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
cJSON* guidstr = cJSON_GetObjectItem(messageBody, "guid");
if (guidstr == nullptr) {
std::cout << "Missing 'guid' in JSON." << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 1;
}
@@ -1511,6 +1578,7 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
//进程号为0的进程处理所有台账更新消息
if (process_No != g_front_seg_index && g_front_seg_index !=0) {
std::cout << "msg index:"<< process_No <<"doesnt match self index:" << g_front_seg_index << std::endl;
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
@@ -1773,10 +1841,492 @@ int parse_control(const std::string& json_str, const std::string& output_dir) {
}
// 释放 JSON 对象
cJSON_Delete(messageBody);
cJSON_Delete(root);
return 0;
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////
//处理文件请求
static int ParseFileDirReq(const char *body, file_dir_req_t *req)
{
if (body == NULL || req == NULL)
return -1;
cJSON *root = cJSON_Parse(body);
if (root == NULL)
return -1;
cJSON *guid = cJSON_GetObjectItem(root, "guid");
cJSON *frontid = cJSON_GetObjectItem(root, "frontid");
cJSON *processNo = cJSON_GetObjectItem(root, "ProcessNo");
cJSON *devid = cJSON_GetObjectItem(root, "devid");
cJSON *type = cJSON_GetObjectItem(root, "type");
cJSON *path = cJSON_GetObjectItem(root, "Path");
if (!guid || guid->type != cJSON_String ||
!frontid || frontid->type != cJSON_String ||
!processNo || processNo->type != cJSON_Number ||
!devid || devid->type != cJSON_String ||
!type || type->type != cJSON_Number ||
!path || path->type != cJSON_String)
{
cJSON_Delete(root);
return -1;
}
memset(req, 0, sizeof(file_dir_req_t));
snprintf(req->guid, sizeof(req->guid), "%s", guid->valuestring);
snprintf(req->frontid, sizeof(req->frontid), "%s", frontid->valuestring);
req->processNo = processNo->valueint;
snprintf(req->devid, sizeof(req->devid), "%s", devid->valuestring);
req->type = type->valueint;
snprintf(req->path, sizeof(req->path), "%s", path->valuestring);
req->create_time = time(NULL);
cJSON_Delete(root);
return 0;
}
static void PushFileDirReq(const file_dir_req_t *req)
{
if (req == NULL)
return;
file_dir_req_t *node = new file_dir_req_t;
*node = *req;
pthread_mutex_lock(&g_file_req_mutex);
g_file_dir_req_list.push_back(node);
pthread_mutex_unlock(&g_file_req_mutex);
}
static file_dir_req_t* PopMatchedFileDirReq(const char *terminal_id)
{
if (terminal_id == NULL || terminal_id[0] == 0)
return NULL;
file_dir_req_t *match = NULL;
pthread_mutex_lock(&g_file_req_mutex);
for (std::list<file_dir_req_t*>::iterator it = g_file_dir_req_list.begin();
it != g_file_dir_req_list.end();
++it)
{
file_dir_req_t *node = *it;
if (node != NULL && strcmp(node->devid, terminal_id) == 0)
{
match = node;
g_file_dir_req_list.erase(it);
break;
}
}
pthread_mutex_unlock(&g_file_req_mutex);
return match;
}
static std::string BuildFileDirRespJsonEx(const file_dir_req_t *req,
char **names,
const char **itemTypes,
int *itemSizes,
int itemNum,
int result)
{
cJSON *root = cJSON_CreateObject();
cJSON *dirInfo = cJSON_CreateArray();
cJSON_AddStringToObject(root, "guid", req ? req->guid : "");
cJSON_AddStringToObject(root, "frontid", req ? req->frontid : "");
cJSON_AddNumberToObject(root, "processNo", req ? req->processNo : 0);
cJSON_AddStringToObject(root, "devid", req ? req->devid : "");
cJSON_AddNumberToObject(root, "type", req ? req->type : 0);
for (int i = 0; i < itemNum; ++i)
{
cJSON *item = cJSON_CreateObject();
cJSON_AddStringToObject(item, "name", (names && names[i]) ? names[i] : "");
cJSON_AddStringToObject(item, "type", (itemTypes && itemTypes[i]) ? itemTypes[i] : "file");
cJSON_AddNumberToObject(item, "size", (itemSizes ? itemSizes[i] : 1));
cJSON_AddItemToArray(dirInfo, item);
}
cJSON_AddItemToObject(root, "dirInfo", dirInfo);
cJSON_AddNumberToObject(root, "result", result);
char *json = cJSON_PrintUnformatted(root);
std::string jsonStr = json ? json : "";
if (json) free(json);
cJSON_Delete(root);
return jsonStr;
}
static std::string BuildFileDirRespJson(const file_dir_req_t *req,
char **filenames,
int filenum,
int result)
{
if (filenum <= 0)
return BuildFileDirRespJsonEx(req, NULL, NULL, NULL, 0, result);
const char **types = new const char*[filenum];
int *sizes = new int[filenum];
for (int i = 0; i < filenum; ++i)
{
types[i] = "dir";
sizes[i] = 1;
}
std::string jsonStr = BuildFileDirRespJsonEx(req, filenames, types, sizes, filenum, result);
delete [] types;
delete [] sizes;
return jsonStr;
}
static std::string BuildSingleFileRespJson(const file_dir_req_t *req,
const char *name,
const char *itemType,
int size,
int result)
{
char *names[1];
const char *types[1];
int sizes[1];
if (name == NULL || result != 0)
{
return BuildFileDirRespJsonEx(req, NULL, NULL, NULL, 0, result);
}
names[0] = (char *)name;
types[0] = (itemType ? itemType : "file");
sizes[0] = size;
return BuildFileDirRespJsonEx(req, names, types, sizes, 1, result);
}
////////////////////////下载
static const char* GetFileNameOnly(const char* fullpath)
{
if (fullpath == NULL)
return "";
const char* p1 = strrchr(fullpath, '/');
const char* p2 = strrchr(fullpath, '\\');
const char* p = p1 > p2 ? p1 : p2;
return (p ? p + 1 : fullpath);
}
static int MakeDirRecursive(const char* dirPath)
{
if (dirPath == NULL || dirPath[0] == '\0')
return -1;
char tmp[512] = {0};
snprintf(tmp, sizeof(tmp), "%s", dirPath);
int len = strlen(tmp);
if (len <= 0)
return -1;
/* 去掉末尾 '/' */
if (tmp[len - 1] == '/')
tmp[len - 1] = '\0';
for (char* p = tmp + 1; *p; ++p)
{
if (*p == '/')
{
*p = '\0';
if (access(tmp, F_OK) != 0)
{
if (mkdir(tmp, 0777) != 0 && errno != EEXIST)
{
printf("mkdir failed: %s, err=%s\n", tmp, strerror(errno));
return -1;
}
}
*p = '/';
}
}
if (access(tmp, F_OK) != 0)
{
if (mkdir(tmp, 0777) != 0 && errno != EEXIST)
{
printf("mkdir failed: %s, err=%s\n", tmp, strerror(errno));
return -1;
}
}
return 0;
}
static void SafePathName(const char* src, char* dst, size_t dstSize)
{
if (dst == NULL || dstSize == 0)
return;
dst[0] = '\0';
if (src == NULL)
return;
size_t j = 0;
for (size_t i = 0; src[i] != '\0' && j + 1 < dstSize; ++i)
{
char c = src[i];
if ((c >= '0' && c <= '9') ||
(c >= 'a' && c <= 'z') ||
(c >= 'A' && c <= 'Z') ||
c == '.' || c == '_' || c == '-')
{
dst[j++] = c;
}
else
{
dst[j++] = '_';
}
}
dst[j] = '\0';
}
static int BuildTempLocalPath(char* outPath,
size_t outSize,
chnl_usr_t* chnl_usr,
const char* remotePath)
{
if (outPath == NULL || outSize == 0 || chnl_usr == NULL)
return -1;
const char* fileName = GetFileNameOnly(remotePath);
if (fileName == NULL || fileName[0] == '\0')
fileName = "tmp_file.dat";
const char* ipStr = chnl_usr->ip_str;
if (ipStr == NULL || ipStr[0] == '\0')
ipStr = "unknown_ip";
char safeIp[128] = {0};
SafePathName(ipStr, safeIp, sizeof(safeIp));
char dirPath[512] = {0};
snprintf(dirPath, sizeof(dirPath), "/tmp/%s", safeIp);
/* 目录不存在则创建 */
if (MakeDirRecursive(dirPath) != 0)
{
printf("BuildTempLocalPath mkdir failed: %s\n", dirPath);
return -1;
}
snprintf(outPath, outSize, "%s/%s", dirPath, fileName);
return 0;
}
static int HandleTypeDownloadAndUpload(chnl_usr_t* chnl_usr,
file_dir_req_t* req,
std::string& jsonString)
{
if (chnl_usr == NULL || req == NULL || chnl_usr->net_info == NULL)
return -1;
char localpath[512] = {0};
if (BuildTempLocalPath(localpath, sizeof(localpath), chnl_usr, req->path) != 0)
{
DIY_ERRORLOG_CODE("process", LOG_CODE_TRANSIENT_COMM,
"【ERROR】构造本地临时路径失败 devid=%s path=%s",
req->devid, req->path);
jsonString = BuildSingleFileRespJson(req, NULL, "file", 1, -1);
return -1;
}
ST_RET ret = mms_getFile(chnl_usr->net_info,
(ST_CHAR*)localpath,
(ST_CHAR*)req->path,
3 * g_pt61850app->mmsOpTimeout);
if (ret != SD_SUCCESS)
{
DIY_ERRORLOG_CODE("process", LOG_CODE_TRANSIENT_COMM,
"【ERROR】装置文件下载失败 devid=%s, rem=%s, ret=0x%X",
req->devid, req->path, ret);
jsonString = BuildSingleFileRespJson(req, NULL, "file", 1, ret);
return -1;
}
DIY_INFOLOG("process",
"【NORMAL】装置文件下载成功 devid=%s, rem=%s, local=%s",
req->devid, req->path, localpath);
char wavepath[512] = {0};
SendFileWeb(WEB_FILEUPLOAD, localpath, req->path, wavepath);
jsonString = BuildSingleFileRespJson(req,
(wavepath[0] != 0 ? wavepath : req->path),
"file",
1,
0);
remove(localpath);
return 0;
}
///////////传输
static int HandleTypeTransferToDevice(chnl_usr_t* chnl_usr,
file_dir_req_t* req,
std::string& jsonString)
{
if (chnl_usr == NULL || req == NULL || chnl_usr->net_info == NULL)
return -1;
char localpath[512] = {0};
if (BuildTempLocalPath(localpath, sizeof(localpath), chnl_usr, req->path) != 0)
{
DIY_ERRORLOG_CODE("process", LOG_CODE_TRANSIENT_COMM,
"【ERROR】构造本地临时路径失败 devid=%s path=%s",
req->devid, req->path);
jsonString = BuildSingleFileRespJson(req, NULL, "file", 1, -1);
return -1;
}
int dlRet = DownloadFileWeb(WEB_FILEDOWNLOAD, req->path, localpath);
if (dlRet != 0)
{
DIY_ERRORLOG_CODE("process", LOG_CODE_TRANSIENT_COMM,
"【ERROR】Web 文件下载失败 devid=%s, path=%s",
req->devid, req->path);
jsonString = BuildSingleFileRespJson(req, NULL, "file", 1, -1);
return -1;
}
DIY_INFOLOG("process",
"【NORMAL】Web 文件下载成功 devid=%s, webpath=%s, local=%s",
req->devid, req->path, localpath);
char destfilename[512] = {0};
const char* fileName = GetFileNameOnly(req->path);
snprintf(destfilename, sizeof(destfilename), "/etc/%s",
(fileName && fileName[0]) ? fileName : "tmp_file.dat");
ST_RET ret = mms_mvla_obtfile(chnl_usr->net_info,
(ST_CHAR*)localpath,
(ST_CHAR*)destfilename,
3 * g_pt61850app->mmsOpTimeout);
if (ret != SD_SUCCESS)
{
DIY_ERRORLOG_CODE("process", LOG_CODE_TRANSIENT_COMM,
"【ERROR】文件传送到装置失败 devid=%s, src=%s, dest=%s, ret=0x%X",
req->devid, localpath, destfilename, ret);
jsonString = BuildSingleFileRespJson(req, NULL, "file", 1, ret);
return -1;
}
DIY_INFOLOG("process",
"【NORMAL】文件传送到装置成功 devid=%s, src=%s, dest=%s",
req->devid, localpath, destfilename);
jsonString = BuildSingleFileRespJson(req,
destfilename,
"file",
1,
0);
remove(localpath);
return 0;
}
void HandleFileDirReqForChannel(chnl_usr_t *chnl_usr)
{
if (chnl_usr == NULL || chnl_usr->chnl == NULL || chnl_usr->chnl->ied == NULL)
return;
ied_t *ied = chnl_usr->chnl->ied;
ied_usr_t *ied_usr = GET_IEDEXT_ADDR(ied);
if (ied_usr == NULL)
return;
if (ied_usr->terminal_id[0] == 0)
return;
file_dir_req_t *req = PopMatchedFileDirReq(ied_usr->terminal_id);
if (req == NULL)
return; // 当前连接没有文件请求
DIY_INFOLOG("process",
"【NORMAL】处理文件请求 terminal_id=%s type=%d path=%s",
req->devid, req->type, req->path);
std::string jsonString;
int handleRet = -1;
if (req->type == 0)
{
/* 目录查询 */
char **filenames = NULL;
int filenum = 0;
ST_RET ret = mms_mvla_fdir(chnl_usr->net_info,
(ST_CHAR*)req->path,
3 * g_pt61850app->mmsOpTimeout,
&filenames,
&filenum,
g_pt61850app->tmp_pool);
jsonString = BuildFileDirRespJson(req,
filenames,
filenum,
(ret == SD_SUCCESS) ? 0 : ret);
DIY_INFOLOG("process",
"【NORMAL】目录请求处理完成 terminal_id=%s ret=0x%X filenum=%d",
req->devid, ret, filenum);
handleRet = (ret == SD_SUCCESS) ? 0 : -1;
}
else if (req->type == 1)
{
/* 装置下载到本地，再上传 Web */
handleRet = HandleTypeDownloadAndUpload(chnl_usr, req, jsonString);
}
else if (req->type == 2)
{
/* Web 下载到本地，再传送到装置 /etc */
handleRet = HandleTypeTransferToDevice(chnl_usr, req, jsonString);
}
else
{
DIY_WARNLOG("process",
"【WARN】未知文件请求类型 type=%d devid=%s path=%s",
req->type, req->devid, req->path);
jsonString = BuildSingleFileRespJson(req, NULL, "file", 1, 1);//1是失败
handleRet = -1;
}
/* 统一回 Kafka */
Ckafka_data_t dir_info;
dir_info.strTopic = QString::fromStdString(Topic_Reply_Topic);
dir_info.strText = QString::fromStdString(jsonString);
kafka_data_list_mutex.lock();
kafka_data_list.append(dir_info);
kafka_data_list_mutex.unlock();
delete req;
}
/////////////////////////////////////////////////////////////////////////////////////////////////////////
@@ -2048,6 +2598,44 @@ int myMessageCallbackrecall(CPushConsumer* consumer, CMessageExt* msg)
return E_CONSUME_SUCCESS;
}
int myMessageCallbackfile(CPushConsumer* consumer, CMessageExt* msg)
{
if (INITFLAG != 1) return 1;
if (msg == NULL) {
std::cerr << "Received null message." << std::endl;
return E_RECONSUME_LATER;
}
const char* body = GetMessageBody(msg);
const char* key = GetMessageKeys(msg);
if (body == NULL) {
std::cerr << "Message body is NULL." << std::endl;
return E_RECONSUME_LATER;
}
DIY_INFOLOG("process","【NORMAL】前置消费topic:%s_%s的文件控制消息",
FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_FILE.c_str());
std::cout << "file Callback received message: " << body << std::endl;
std::cout << "Message Key: " << (key ? key : "N/A") << std::endl;
file_dir_req_t req;
if (ParseFileDirReq(body, &req) != 0)
{
DIY_WARNLOG("process", "【WARN】文件控制消息解析失败: %s", body);
return E_CONSUME_SUCCESS;
}
PushFileDirReq(&req);
DIY_INFOLOG("process",
"【NORMAL】文件目录请求已入队 guid=%s devid=%s path=%s",
req.guid, req.devid, req.path);
return E_CONSUME_SUCCESS;
}
void mqconsumerThread::run()
{
@@ -2060,6 +2648,9 @@ void mqconsumerThread::run()
std::vector<Subscription> subscriptions;
// 初始化消费者1 //lnk20241230只有实时进程会订阅实时topic，不订阅实时topic的进程无法触发实时数据
if(g_node_id == THREE_SECS_DATA_BASE_NODE_ID){
//lnk20260310添加文件管理
subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_FILE, G_MQCONSUMER_TAG_FILE, myMessageCallbackfile));
subscriptions.push_back(Subscription(std::string(FRONT_INST) + "_" + G_MQCONSUMER_TOPIC_RT, G_MQCONSUMER_TAG_RT, myMessageCallbackrtdata));
}