cxh/kunlang_video
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
cxh 2025-11-14 09:42:17 +08:00
parent 6eafa6ccc8
commit 0b9c927e3b
3 changed files with 340 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
include/record_manager.hpp Normal file
View File

@ -0,0 +1,49 @@
#pragma once
#include <chrono>
#include <filesystem>
#include <string>
#include <unordered_map>
#include <vector>
struct RecordFileInfo
{
std::string stream; // AHD1_main
std::string path; // /sata/record/AHD1_main/2025-11-13/10/10-23-17.mp4
int64_t start_ms; // 文件起始时间（ms）
int64_t end_ms; // 文件结束时间（ms）
};
struct RecordSegment
{
int index; // 段号
std::string segment_id;
int64_t start_ms;
int64_t end_ms;
std::vector<RecordFileInfo> files;
};
class RecordManager
{
public:
explicit RecordManager(const std::string& base_dir);
// 执行一次全扫描
void scanAll();
// 查询时间段的录像段
std::vector<RecordSegment> querySegments(const std::string& stream, int64_t start_ms, int64_t end_ms);
// 根据 segmentId 获取可播放文件列表
RecordSegment getSegment(const std::string& segmentId);
private:
std::string base_dir_; // /sata/record/
std::unordered_map<std::string, std::vector<RecordFileInfo>> index_;
// 解析文件名得到 start_ms 和 end_ms
RecordFileInfo parseFile(const std::filesystem::path& p);
// 给一个 stream 的文件列表排序
void sortStream(const std::string& stream);
};

138
src/main.cpp
View File

@ -9,6 +9,7 @@
#include "app_config.hpp"
#include "logger.hpp"
#include "mqtt_client_wrapper.hpp"
#include "record_manager.hpp"
#include "rtmp_manager.hpp"
std::atomic<bool> g_running(true);
@ -35,54 +36,107 @@ int main()
Logger::init(get_executable_dir_file_path("logs"), 7);
LOG_INFO("[MAIN] ===== Vehicle Video Service Starting =====");
try
std::string base = "/sata/record/";
RecordManager rm(base);
std::cout << "[RecordManager] scanning " << base << " ..." << std::endl;
rm.scanAll();
std::cout << "\n=== 全部扫描结果 ===\n";
// 遍历所有 stream
for (const auto& kv : rm.index_)
{
// 加载配置
g_app_config = AppConfig::load_from_file(get_executable_dir_file_path("config.json"));
LOG_INFO("[MAIN] Loaded config from config.json");
}
catch (const std::exception& e)
{
LOG_ERROR(std::string("[MAIN] Failed to load config: ") + e.what());
return -1;
}
const std::string& stream = kv.first;
const auto& files = kv.second;
// ---------- 初始化 GStreamer ----------
RTMPManager::init();
std::cout << "\n>>> Stream = " << stream << "\n";
std::cout << "文件数量 = " << files.size() << "\n";
// ---------- 自动推流（8 路录像守护） ----------
LOG_INFO("[MAIN] Starting all record streams...");
RTMPManager::start_all();
// 启动 MQTT 线程
std::thread mqtt_thread(
[]
for (auto& f : files)
{
try
{
LOG_INFO("[MAIN] MQTT thread started.");
mqtt_client_thread_func(); // 在回调里执行推流控制
}
catch (const std::exception& e)
{
LOG_ERROR(std::string("[MAIN] MQTT thread crashed: ") + e.what());
}
LOG_INFO("[MAIN] MQTT thread exiting...");
});
// 主循环，仅等待退出信号
while (g_running.load(std::memory_order_relaxed)) std::this_thread::sleep_for(std::chrono::milliseconds(200));
// ---------- 退出清理 ----------
LOG_INFO("[MAIN] Shutdown requested. Stopping RTMP streams...");
RTMPManager::stop_all();
if (mqtt_thread.joinable())
{
mqtt_thread.join();
LOG_INFO("[MAIN] MQTT thread joined.");
std::cout << " - " << f.path << "\n start_ms=" << f.start_ms << ", end_ms=" << f.end_ms << "\n";
}
}
// 测试一次查询：你可以换成真实时间戳
std::cout << "\n=== 测试 querySegments() ===\n";
// 用你的 AHD1_main 做示例
std::string stream = "AHD1_main";
// 随便选一个时间区间，比如 2025-11-13 10:23:00 ~ 11:00:00
// 你可以换成真实值
long long t1 = 1731464600000;
long long t2 = 1731467400000;
auto segments = rm.querySegments(stream, t1, t2);
std::cout << "找到录像段数量 = " << segments.size() << "\n";
for (auto& seg : segments)
{
std::cout << "\n--- Segment " << seg.index << " ---\n";
std::cout << "segmentId = " << seg.segment_id << "\n";
std::cout << "start_ms = " << seg.start_ms << "\n";
std::cout << "end_ms = " << seg.end_ms << "\n";
std::cout << "files = " << seg.files.size() << "\n";
for (auto& f : seg.files)
{
std::cout << " file: " << f.path << "\n [" << f.start_ms << " ~ " << f.end_ms << "]\n";
}
}
// try
// {
// // 加载配置
// g_app_config = AppConfig::load_from_file(get_executable_dir_file_path("config.json"));
// LOG_INFO("[MAIN] Loaded config from config.json");
// }
// catch (const std::exception& e)
// {
// LOG_ERROR(std::string("[MAIN] Failed to load config: ") + e.what());
// return -1;
// }
// // ---------- 初始化 GStreamer ----------
// RTMPManager::init();
// // ---------- 自动推流（8 路录像守护） ----------
// LOG_INFO("[MAIN] Starting all record streams...");
// RTMPManager::start_all();
// // 启动 MQTT 线程
// std::thread mqtt_thread(
// []
// {
// try
// {
// LOG_INFO("[MAIN] MQTT thread started.");
// mqtt_client_thread_func(); // 在回调里执行推流控制
// }
// catch (const std::exception& e)
// {
// LOG_ERROR(std::string("[MAIN] MQTT thread crashed: ") + e.what());
// }
// LOG_INFO("[MAIN] MQTT thread exiting...");
// });
// // 主循环，仅等待退出信号
// while (g_running.load(std::memory_order_relaxed)) std::this_thread::sleep_for(std::chrono::milliseconds(200));
// // ---------- 退出清理 ----------
// LOG_INFO("[MAIN] Shutdown requested. Stopping RTMP streams...");
// RTMPManager::stop_all();
// if (mqtt_thread.joinable())
// {
// mqtt_thread.join();
// LOG_INFO("[MAIN] MQTT thread joined.");
// }
LOG_INFO("[MAIN] ===== Vehicle Video Service Exited Cleanly =====");
return 0;
}

195
src/record_manager.cpp Normal file
View File

@ -0,0 +1,195 @@
#include "record_manager.hpp"
#include <algorithm>
#include <ctime>
#include <iostream>
namespace fs = std::filesystem;
RecordManager::RecordManager(const std::string& base_dir) : base_dir_(base_dir) {}
//
// 扫描所有文件
//
void RecordManager::scanAll()
{
index_.clear();
if (!fs::exists(base_dir_)) return;
for (auto const& entry : fs::recursive_directory_iterator(base_dir_))
{
if (!entry.is_regular_file()) continue;
auto p = entry.path();
if (p.extension() != ".mp4") continue;
auto info = parseFile(p);
if (info.start_ms <= 0) continue;
index_[info.stream].push_back(info);
}
// 各 stream 排序
for (auto& kv : index_)
{
sortStream(kv.first);
}
}
//
// 排序函数：按 start_ms 升序
//
void RecordManager::sortStream(const std::string& stream)
{
auto& files = index_[stream];
std::sort(files.begin(), files.end(),
[](const RecordFileInfo& a, const RecordFileInfo& b) { return a.start_ms < b.start_ms; });
}
//
// 解析路径得到文件时间戳
//
// 格式：
/// sata/record/AHD1_main/2025-11-13/10/10-23-17.mp4
//
RecordFileInfo RecordManager::parseFile(const fs::path& p)
{
RecordFileInfo info;
info.path = p.string();
try
{
// 目录结构倒推
auto filename = p.filename().string(); // 10-23-17.mp4
auto hour_dir = p.parent_path().filename().string(); // 10
auto date_dir = p.parent_path().parent_path().filename().string(); // 2025-11-13
auto stream_dir = p.parent_path().parent_path().parent_path().filename().string();
info.stream = stream_dir;
// 日期解析
int y = 0, mon = 0, d = 0;
sscanf(date_dir.c_str(), "%d-%d-%d", &y, &mon, &d);
// 文件名解析 H-M-S
int h = 0, m = 0, s = 0;
sscanf(filename.c_str(), "%d-%d-%d", &h, &m, &s);
std::tm t{};
t.tm_year = y - 1900;
t.tm_mon = mon - 1;
t.tm_mday = d;
t.tm_hour = h;
t.tm_min = m;
t.tm_sec = s;
int64_t epoch = std::mktime(&t) * 1000LL;
info.start_ms = epoch;
info.end_ms = epoch + 60 * 1000; // 默认 1 分钟
}
catch (...)
{
info.start_ms = -1;
info.end_ms = -1;
}
return info;
}
//
// 查询某段时间的录像段
//
std::vector<RecordSegment> RecordManager::querySegments(const std::string& stream, int64_t start_ms, int64_t end_ms)
{
std::vector<RecordSegment> result;
if (!index_.count(stream)) return result;
const auto& files = index_[stream];
std::vector<RecordFileInfo> hits;
// 先找到落在时间区间内的文件
for (const auto& f : files)
{
bool overlap = !(f.end_ms <= start_ms || f.start_ms >= end_ms);
if (overlap) hits.push_back(f);
}
if (hits.empty()) return result;
// 合并连续文件为段
std::sort(hits.begin(), hits.end(), [](auto& a, auto& b) { return a.start_ms < b.start_ms; });
int idx = 1;
RecordSegment seg{};
seg.index = idx;
seg.start_ms = hits[0].start_ms;
seg.files.push_back(hits[0]);
seg.end_ms = hits[0].end_ms;
seg.segment_id = hits[0].stream + "_" + std::to_string(seg.start_ms) + "_" + std::to_string(seg.end_ms);
for (size_t i = 1; i < hits.size(); ++i)
{
auto& prev = hits[i - 1];
auto& cur = hits[i];
bool continuous = (cur.start_ms <= prev.end_ms + 1500); // 允许 1.5 秒误差
if (continuous)
{
seg.files.push_back(cur);
seg.end_ms = cur.end_ms;
}
else
{
// 关闭上一段
seg.segment_id =
seg.files[0].stream + "_" + std::to_string(seg.start_ms) + "_" + std::to_string(seg.end_ms);
result.push_back(seg);
// 开新段
idx++;
seg = RecordSegment{};
seg.index = idx;
seg.files.push_back(cur);
seg.start_ms = cur.start_ms;
seg.end_ms = cur.end_ms;
}
}
// 最后一段补进去
seg.segment_id = seg.files[0].stream + "_" + std::to_string(seg.start_ms) + "_" + std::to_string(seg.end_ms);
result.push_back(seg);
return result;
}
//
// 根据 segmentId 反查 segment
//
RecordSegment RecordManager::getSegment(const std::string& segmentId)
{
// segmentId 格式：
// AHD1_main_1731465600000_1731467400000
auto pos1 = segmentId.find('_');
auto pos2 = segmentId.find('_', pos1 + 1);
if (pos1 == std::string::npos || pos2 == std::string::npos) return {};
std::string stream = segmentId.substr(0, pos1);
int64_t start_ms = std::stoll(segmentId.substr(pos1 + 1, pos2 - pos1 - 1));
int64_t end_ms = std::stoll(segmentId.substr(pos2 + 1));
auto segs = querySegments(stream, start_ms, end_ms);
for (auto& s : segs)
{
if (s.start_ms == start_ms && s.end_ms == end_ms) return s;
}
return {};
}