zhangwen/front_linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add recall reply

Browse Source
This commit is contained in:
lnk
2025-09-15 16:36:21 +08:00
parent a298ff847a
commit f27d208959
6 changed files with 271 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

243
LFtid1056/cloudfront/code/cfg_parser.cpp
View File

@@ -1092,25 +1092,79 @@ void Get_Recall_Time_Char(const std::string& start_time_str,
}
//mq调用将补招信息写入补招列表
int recall_json_handle(const std::string& jstr) {
// 不指定稳态/暂态则全部补招
int stat = 0;
int voltage = 0;
int recall_json_handle_from_mq(const std::string& body)
{
try {
// 1. 解析 JSON 数组
auto json_root = nlohmann::json::parse(jstr);
if (!json_root.is_array()) {
std::cout << "json root解析错误" << std::endl;
return 10000;
// ====== 解析外层 JSON ======
nlohmann::json root;
try {
root = nlohmann::json::parse(body);
} catch (const std::exception& e) {
std::cerr << "Error parsing JSON: " << e.what() << std::endl;
// ★与原逻辑等价：无法解析，不再进入 recall_json_handle
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,消息的json结构不正确",
g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
return 10004;
}
// 2. 遍历每个补招项
for (auto& item : json_root) {
// 提取 "messageBody"（字符串）
if (!root.contains("messageBody") || !root["messageBody"].is_string()) {
std::cerr << "'messageBody' is missing or is not a string" << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,没有messageBody字段",
g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
return 10004;
}
std::string messageBodyStr = root["messageBody"].get<std::string>();
if (messageBodyStr.empty()) {
std::cerr << "'messageBody' is empty" << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,messageBody为空",
g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
return 10004;
}
// 解析 messageBody 内层 JSON
nlohmann::json messageBody;
try {
messageBody = nlohmann::json::parse(messageBodyStr);
} catch (const std::exception& e) {
std::cerr << "Failed to parse 'messageBody' JSON: " << e.what() << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,messageBody的json结构不正确",
g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
return 10004;
}
// 提取 guid 并立即回执
if (!messageBody.contains("guid") || !messageBody["guid"].is_string()) {
std::cerr << "'guid' is missing or is not a string" << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,没有guid字段",
g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
return 10004;
}
std::string guid = messageBody["guid"].get<std::string>();
send_reply_to_queue(guid, "1", "收到补招指令");
// 提取 data 数组
if (!messageBody.contains("data") || !messageBody["data"].is_array()) {
std::cerr << "'data' is missing or is not an array" << std::endl;
DIY_ERRORLOG("process","【ERROR】前置的%d号进程处理topic:%s_%s的补招触发消息失败,没有data字段",
g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_RC.c_str());
return 10004;
}
// 仅用于保留你原先的调试输出
std::string data_dump;
try { data_dump = messageBody["data"].dump(); } catch (...) { data_dump.clear(); }
std::cout << "parseJsonMessageRC: " << data_dump << std::endl;
// 不指定稳态/暂态则全部补招
int stat = 0;
int voltage = 0;
// 2. 遍历每个补招项（这里直接用已解析的 messageBody["data"]）
for (auto& item : messageBody["data"]) {
// 获取必需字段
// ★修改：强制要求 terminalId；
if (!item.contains("terminalId") ||
!item.contains("monitor") ||
!item.contains("monitor") ||
!item.contains("timeInterval") ||
!item.contains("dataType"))
{
@@ -1125,10 +1179,9 @@ int recall_json_handle(const std::string& jstr) {
continue;
}
// 2.1 解析 dataType
// 2.1 解析 dataType（仅保留稳态/暂态）
std::string datatype = item["dataType"].get<std::string>();
if (!datatype.empty()) {
// ★修改：仅保留稳态/暂态
if (datatype == "0" || datatype == "稳态" || datatype == "steady" || datatype == "STEADY") {
stat = 1; voltage = 0; // 稳态
} else if (datatype == "1" || datatype == "暂态" || datatype == "voltage" || datatype == "VOLTAGE") {
@@ -1143,11 +1196,12 @@ int recall_json_handle(const std::string& jstr) {
// ★新增：定位并校验该 terminal 是否归属当前进程
std::lock_guard<std::mutex> lock(ledgermtx);
const terminal_dev* targetDev = nullptr;
for (const auto& dev : terminal_devlist) {
// 只处理本进程对应的终端
if (dev.terminal_id == terminalId) {
targetDev = &dev;
const terminal_dev* targetDev = NULL;
for (std::vector<terminal_dev>::const_iterator it = terminal_devlist.begin();
it != terminal_devlist.end(); ++it)
{
if (it->terminal_id == terminalId) {
targetDev = &(*it);
break;
}
}
@@ -1156,8 +1210,18 @@ int recall_json_handle(const std::string& jstr) {
continue;
}
// 添加判断装置在线，不注册guid，异步补招
if (ClientManager::instance().get_dev_status(targetDev->terminal_id) != 1) {
std::cout << "terminalId对应装置不在线: " << targetDev->terminal_id << std::endl;
// 响应 web
std::string msg = std::string("装置：") + targetDev->terminal_name + " 不在线，无法补招";
send_reply_to_kafka_recall("12345", "2", static_cast<int>(ResponseCode::INTERNAL_ERROR), msg, targetDev->terminal_id, "", "", "");
continue;//处理下一个装置的补招记录
}
// ★新增：按新结构解析 monitor 层级
auto& monitors = item["monitor"];
nlohmann::json& monitors = item["monitor"];
if (!monitors.is_array() || monitors.empty()) {
std::cout << "monitor数组为空或非数组类型" << std::endl;
continue;
@@ -1168,15 +1232,19 @@ int recall_json_handle(const std::string& jstr) {
std::cout << "monitor项缺少 monitorId 或 timeInterval" << std::endl;
continue;
}
std::string monitorId = mobj["monitorId"].get<std::string>();
if (monitorId.empty()) continue;
// 不只是判断存在，还要拿到指针 lm 以便后续 push_back
ledger_monitor* lm = nullptr;
for (auto& mon : const_cast<terminal_dev*>(targetDev)->line) {
if (!mon.monitor_id.empty() && mon.monitor_id == monitorId) {
lm = &mon;
ledger_monitor* lm = NULL;
// 注意：这里需要非常量指针，取 const_cast 后遍历
terminal_dev* dev_nc = const_cast<terminal_dev*>(targetDev);
for (std::vector<ledger_monitor>::iterator itLm = dev_nc->line.begin();
itLm != dev_nc->line.end(); ++itLm)
{
if (!itLm->monitor_id.empty() && itLm->monitor_id == monitorId) {
lm = &(*itLm);
break;
}
}
@@ -1185,50 +1253,52 @@ int recall_json_handle(const std::string& jstr) {
<< " @ " << terminalId << std::endl;
continue;
}
auto& tiArr = mobj["timeInterval"];
nlohmann::json& tiArr = mobj["timeInterval"];
if (!tiArr.is_array() || tiArr.empty()) {
std::cout << "timeInterval为空或非数组类型: monitorId=" << monitorId << std::endl;
continue;
}
// 这里拆分时间段并 push 到 lm->recall_list
// 这里拆分时间段并 push 到 lm->recall_list / lm->recall_list_static
for (auto& timeItem : tiArr) {
std::string ti = timeItem.get<std::string>();
auto pos = ti.find('~');
std::string::size_type pos = ti.find('~');
if (pos == std::string::npos) {
std::cout << "timeInterval格式错误: " << ti << std::endl;
continue;
}
std::string start = ti.substr(0, pos);
std::string end = ti.substr(pos + 1);
// 仅对 recall_list（事件）进行 1 小时拆分；recall_list_stat（稳态）不拆分
// 仅对 recall_list（事件）进行 1 小时拆分；recall_list_static（稳态）不拆分
{
// 公共字段（整体区间，不拆分）用于 recall_list_stat
RecallFile rm_all;
rm_all.recall_status = 0; // 初始状态：未补招
rm_all.StartTime = start;
rm_all.EndTime = end;
rm_all.STEADY = std::to_string(stat);
rm_all.VOLTAGE = std::to_string(voltage);
// 公共字段（整体区间，不拆分）用于 recall_list_static
RecallFile rm_all; // ★类型正确：稳态列表的元素
rm_all.recall_status = 0; // 初始状态：未补招
rm_all.StartTime = start; // 直接使用字符串
rm_all.EndTime = end;
rm_all.STEADY = std::to_string(stat);
rm_all.VOLTAGE = std::to_string(voltage);
// 仅当需要事件补招（voltage==1）时，才进行 1 小时拆分并压入 recall_list
if (voltage == 1) {
// 拆分时间段为 1 小时一段，并存入 recall_list
std::vector<RecallInfo> recallinfo_list_hour;
Get_Recall_Time_Char(start, end, recallinfo_list_hour);
for (auto& info : recallinfo_list_hour) {
RecallMonitor rm;
rm.recall_status = 0; // 初始状态：未补招
rm.StartTime = epoch_to_datetime_str(info.starttime);
rm.EndTime = epoch_to_datetime_str(info.endtime);
rm.STEADY = std::to_string(stat);
rm.VOLTAGE = std::to_string(voltage);
lm->recall_list.push_back(std::move(rm));
for (std::size_t i = 0; i < recallinfo_list_hour.size(); ++i) {
const RecallInfo& info = recallinfo_list_hour[i];
RecallMonitor rm; // ★类型正确：事件列表的元素
rm.recall_status = 0; // 初始状态：未补招
rm.StartTime = epoch_to_datetime_str(info.starttime);
rm.EndTime = epoch_to_datetime_str(info.endtime);
rm.STEADY = std::to_string(stat);
rm.VOLTAGE = std::to_string(voltage);
lm->recall_list.push_back(rm);
// 事件补招列表（recall_list）调试打印
std::cout << "[recall_json_handle] terminal=" << terminalId
<< " monitor=" << monitorId
@@ -1239,12 +1309,12 @@ int recall_json_handle(const std::string& jstr) {
<< std::endl;
}
}
// 仅当需要稳态补招（stat==1）时，不拆分，直接压入 recall_list_static
if (stat == 1) {
lm->recall_list_static.push_back(rm_all); // 不拆分，整体区间
// 稳态补招列表（recall_list_stat）调试打印
// 稳态补招列表（recall_list_static）调试打印
std::cout << "[recall_json_handle] terminal=" << terminalId
<< " monitor=" << monitorId
<< " [recall_list_static] start=" << lm->recall_list_static.back().StartTime
@@ -1253,15 +1323,15 @@ int recall_json_handle(const std::string& jstr) {
<< " voltage="<< lm->recall_list_static.back().VOLTAGE
<< std::endl;
}
// 非法输入保护
// 非法输入保护（保留你原来的保护与返回码）
if (stat == 0 && voltage == 0) {
std::cout << "[recall_json_handle] skip: stat=0 && voltage=0, monitor=" << monitorId
<< " terminal=" << terminalId
<< " start=" << rm_all.StartTime
<< " end=" << rm_all.EndTime
<< std::endl;
return 10003; // 不可能进入这个逻辑，错误退出
return 10003;
}
}
}
@@ -3227,14 +3297,14 @@ int get_type_by_state(int state) {
case DeviceState::READING_INTERFIXEDVALUEDES:
case DeviceState::READING_CONTROLWORD:
case DeviceState::SET_INTERFIXEDVALUE:
return 0x2106;
return 0x2106; //读数据
case DeviceState::READING_FILEMENU:
return 0x2131;
return 0x2131; //读目录
case DeviceState::READING_EVENTFILE:
case DeviceState::READING_FILEDATA:
return 0x2132;
return 0x2132; //读文件
default:
return 0; // 没有对应的type
@@ -3245,6 +3315,7 @@ int get_type_by_state(int state) {
void check_device_busy_timeout()
{
std::lock_guard<std::mutex> lock(ledgermtx);
for (auto &dev : terminal_devlist)
{
if (dev.isbusy != 0) // 有业务在进行
@@ -3260,7 +3331,7 @@ void check_device_busy_timeout()
<< dev.busytimecount << "s)" << std::endl;
//发送超时响应
send_reply_to_cloud(static_cast<int>(ResponseCode::BAD_REQUEST),dev.terminal_id,get_type_by_state(dev.busytype));
send_reply_to_cloud(static_cast<int>(ResponseCode::TIMEOUT),dev.terminal_id,get_type_by_state(dev.busytype));
// 超时清空状态
dev.guid.clear(); // 清空进行中的 guid
@@ -3276,6 +3347,10 @@ void check_device_busy_timeout()
std::cout << "[Timeout] Device " << dev.terminal_id
<< " busytype=" << dev.busytype
<< " 超时(" << dev.busytimecount << "s)" << std::endl;
//发送超时响应
send_reply_to_cloud(static_cast<int>(ResponseCode::TIMEOUT),dev.terminal_id,get_type_by_state(dev.busytype));
// 超时清空状态
dev.guid.clear();
dev.busytype = 0;
@@ -3821,6 +3896,36 @@ void clear_terminal_runtime_state(const std::string& id) {
}
//////////////////////////////////////////////////////////////////////////////////////////////////////////////处理补招逻辑
//发送补招响应给web
void send_reply_to_kafka_recall(const std::string& guid, const std::string& step,int code, const std::string& result,const std::string& terminalId,const std::string& lineIndex,const std::string& recallStartDate,const std::string& recallEndDate){
// 构造 JSON 字符串
std::ostringstream oss;
oss << "{"
<< "\"guid\":\"" << guid << "\","
<< "\"step\":\"" << step << "\","
<< "\"code\":" << code << ","
<< "\"result\":\"" << result << "\","
<< "\"terminalId\":\"" << terminalId << "\","
<< "\"lineIndex\":\"" << lineIndex << "\","
<< "\"recallStartDate\":\"" << recallStartDate << "\","
<< "\"recallEndDate\":\"" << recallEndDate << "\","
<< "\"processNo\":\"" << g_front_seg_index << "\","
<< "\"nodeId\":\"" << FRONT_INST << "\""
<< "}";
std::string jsonString = oss.str();
// 封装 Kafka 消息
queue_data_t connect_info;
connect_info.strTopic = Topic_Reply_Topic;
connect_info.strText = jsonString;
// 加入发送队列（带互斥锁保护）
queue_data_list_mutex.lock();
queue_data_list.push_back(connect_info);
queue_data_list_mutex.unlock();
}
// ===== 一次遍历可下发“多个终端的一条” =====
void check_recall_event() {
@@ -3847,7 +3952,11 @@ void check_recall_event() {
<< " " << front.StartTime << " ~ " << front.EndTime << std::endl;
//调用reply接口通知web端该时间段补招完成
std::string msg = std::string("监测点：") + lm.monitor_name
+ " 补招时间范围：" + front.StartTime
+ " ~ " + front.EndTime
+ " 补招执行完成";
send_reply_to_kafka_recall("12345","2",static_cast<int>(ResponseCode::OK),msg,dev.terminal_id,lm.logical_device_seq,front.StartTime,front.EndTime);
lm.recall_list.pop_front(); // 弹掉首条
} else if (front.recall_status == static_cast<int>(RecallStatus::FAILED)) {
@@ -3856,7 +3965,11 @@ void check_recall_event() {
<< " " << front.StartTime << " ~ " << front.EndTime << std::endl;
//调用reply接口通知web端该时间段补招失败
std::string msg = std::string("监测点：") + lm.monitor_name
+ " 补招时间范围：" + front.StartTime
+ " ~ " + front.EndTime
+ " 补招执行失败";
send_reply_to_kafka_recall("12345","2",static_cast<int>(ResponseCode::BAD_REQUEST),msg,dev.terminal_id,lm.logical_device_seq,front.StartTime,front.EndTime);
lm.recall_list.pop_front(); // 弹掉首条
} else {