recall stat

LFtid1056/pqdif_thread_processor.cpp

@@ -28,6 +28,17 @@
 #include "pqdif/include/pqdif_lg.h"
 #include "pqdif_semantic_ids.h"
 
+#include "cloudfront/code/log4.h" //lnk20260526
+
+extern void enqueue_stat_pq(const std::string& max_base64Str,
+                     const std::string& min_base64Str,
+                     const std::string& avg_base64Str,
+                     const std::string& cp95_base64Str,
+                     time_t data_time,
+                     const std::string& mac,
+                     short cid);
+extern std::string extract_filename1(const std::string& path);
+
 namespace fs = std::experimental::filesystem;
 
 namespace {
@@ -7877,6 +7888,96 @@ void ClearReadyPqdifStatBase64Queue()
     g_pqdif_stat_base64_ready_queue.clear();
 }
 
+static bool GetBase64ByKind(const PqdifStatBase64TimePointPacket& tp,   //从序列中获取指定 kind 的 Base64 内容
+                            StatValueKind kind,
+                            std::string& out)
+{
+    for (const auto& r : tp.records) {
+        if (r.value_kind == kind) {
+            out = r.base64_payload;
+            return !out.empty();
+        }
+    }
+    return false;
+}
+
+static bool extract_monitor_seq_from_local_pqdif_path(const std::string& path,
+                                                      short& point_name)
+{
+    point_name = 0;
+
+    std::cout << "[extract_monitor_seq] begin path="
+              << path << std::endl;
+
+    // 取纯文件名，例如：
+    // download/192.168.1.10/M1_xxx.pqd
+    // -> M1_xxx.pqd
+    std::string fname = extract_filename1(path);
+
+    std::cout << "[extract_monitor_seq] filename="
+              << fname << std::endl;
+
+    if (fname.size() < 3) {
+        std::cout << "[extract_monitor_seq] filename too short"
+                  << std::endl;
+        return false;
+    }
+
+    if (fname[0] != 'M' && fname[0] != 'm') {
+        std::cout << "[extract_monitor_seq] filename not start with M/m"
+                  << std::endl;
+        return false;
+    }
+
+    size_t pos = fname.find('_');
+
+    std::cout << "[extract_monitor_seq] underscore pos="
+              << pos << std::endl;
+
+    if (pos == std::string::npos || pos <= 1) {
+        std::cout << "[extract_monitor_seq] invalid underscore position"
+                  << std::endl;
+        return false;
+    }
+
+    // M1_xxx -> 1
+    std::string seq_str = fname.substr(1, pos - 1);
+
+    std::cout << "[extract_monitor_seq] seq_str="
+              << seq_str << std::endl;
+
+    for (char c : seq_str) {
+        if (!std::isdigit(static_cast<unsigned char>(c))) {
+            std::cout << "[extract_monitor_seq] non-digit char="
+                      << c << std::endl;
+            return false;
+        }
+    }
+
+    try {
+        point_name = static_cast<short>(std::stoi(seq_str));
+
+        std::cout << "[extract_monitor_seq] success point_name="
+                  << point_name << std::endl;
+
+        return point_name > 0;
+    }
+    catch (const std::exception& e) {
+        std::cout << "[extract_monitor_seq] exception="
+                  << e.what() << std::endl;
+
+        point_name = 0;
+        return false;
+    }
+    catch (...) {
+        std::cout << "[extract_monitor_seq] unknown exception"
+                  << std::endl;
+
+        point_name = 0;
+        return false;
+    }
+}
+
 void RunPqdifScanLoop()
 {
     std::cout << "[PQDIF] scan loop started, root=" << kPqdRootDir
@@ -7910,6 +8011,54 @@ void RunPqdifScanLoop()
             if (PopReadyPqdifStatBase64FileBatch(batch)) {
                 // batch 就是一个 PQDIF 文件完整的 Base64 组装结果
                 // 在此处处理上送逻辑
+                const std::string& mac = batch.mac;
+
+                short point_name = 0;
+
+                if (!extract_monitor_seq_from_local_pqdif_path(batch.pqdif_file_path, point_name)) {
+                    std::cout << "[PQDIF_UPLOAD] failed to extract monitor seq from file="
+                              << batch.pqdif_file_path << std::endl;
+                    continue;
+                }
+
+                for (const auto& tp : batch.time_points) {
+                    std::string max_base64;
+                    std::string min_base64;
+                    std::string avg_base64;
+                    std::string p95_base64;
+                
+                    bool has_max = GetBase64ByKind(tp, StatValueKind::Max, max_base64);
+                    bool has_min = GetBase64ByKind(tp, StatValueKind::Min, min_base64);
+                    bool has_avg = GetBase64ByKind(tp, StatValueKind::Avg, avg_base64);
+                    bool has_p95 = GetBase64ByKind(tp, StatValueKind::P95, p95_base64);
+                
+                    if (!has_max || !has_min || !has_avg || !has_p95) {
+                        std::cout << "[PQDIF_UPLOAD] skip incomplete time point, file="
+                                  << batch.pqdif_file_path
+                                  << " time=" << tp.timestamp_text
+                                  << " has_max=" << has_max
+                                  << " has_min=" << has_min
+                                  << " has_avg=" << has_avg
+                                  << " has_p95=" << has_p95
+                                  << std::endl;
+                        continue;
+                    }
+                
+                    enqueue_stat_pq(max_base64,
+                                    min_base64,
+                                    avg_base64,
+                                    p95_base64,
+                                    tp.timestamp,
+                                    mac,
+                                    point_name);
+                    
+                    std::cout << "[PQDIF_UPLOAD] enqueue_stat_pq ok, file="
+                              << batch.pqdif_file_path
+                              << " time=" << tp.timestamp_text
+                              << " mac=" << mac
+                              << " point=" << point_name
+                              << std::endl;
+                }
             }
         }
         catch (const std::exception& ex)