Add recording policy and time-range checks

This change adds a recording policy layer that enforces time-window eligibility and excluded title keywords before probing and before recording. It updates the streamer API to surface reasons like OutOfSchedule and TitleExcluded, clamps ffmpeg/segment durations to the configured window, and exposes the last rejection reason from the worker state for the frontend.
This commit is contained in:
ForgQi committed 2026-08-22 19:15:48 +08:00
1 parent 1affe97d87
commit eb8699b203
14 files changed
+994 -16

No files matched your search

Generated
+4 -4
View File
@@ -387,7 +387,7 @@ checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06"
[[package]]
name = "biliup"
version = "1.2.2"
version = "1.2.3"
dependencies = [
"async-trait",
"base64",
@@ -426,7 +426,7 @@ dependencies = [
[[package]]
name = "biliup-cli"
version = "1.2.2"
version = "1.2.3"
dependencies = [
"async-channel",
"async-trait",
@@ -1015,7 +1015,7 @@ dependencies = [
[[package]]
name = "danmaku"
version = "1.2.2"
version = "1.2.3"
dependencies = [
"async-trait",
"base64",
@@ -4109,7 +4109,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
[[package]]
name = "stream-gears"
version = "1.2.2"
version = "1.2.3"
dependencies = [
"async-channel",
"async-trait",
+1 -1
View File
@@ -8,7 +8,7 @@ members = [
resolver = "3"
[workspace.package]
version = "1.2.2"
version = "1.2.3"
[workspace.dependencies]
serde = { version = "1.0", features = ["derive"] }
+3
View File
@@ -69,6 +69,9 @@ export default function Home() {
case 'OutOfSchedule':
statusTag = <Tag color="green">非录播时间</Tag>
break
case 'TitleExcluded':
statusTag = <Tag color="orange">标题已排除</Tag>
break
case 'Pause':
statusTag = <Tag color="pink">暂停中</Tag>
break
+200 -1
View File
@@ -1,10 +1,11 @@
use crate::server::common::recording_policy::{self, Rejection};
use crate::server::common::upload::{build_studio, submit_to_bilibili, upload};
use crate::server::common::util::Recorder;
use crate::server::config::Config;
use crate::server::core::download_manager::DownloadManager;
use crate::server::errors::{AppError, report_to_response};
use crate::server::infrastructure::connection_pool::ConnectionPool;
use crate::server::infrastructure::context::{Stage, WorkerStatus};
use crate::server::infrastructure::context::{Stage, Worker, WorkerStatus};
use crate::server::infrastructure::dto::LiveStreamerResponse;
use crate::server::infrastructure::models::live_streamer::{InsertLiveStreamer, LiveStreamer};
use crate::server::infrastructure::models::upload_streamer::{
@@ -35,6 +36,20 @@ use tokio::fs;
use tracing::info;
use tracing_subscriber::EnvFilter;
/// 主播当前被录制策略挡下的原因,供 `/v1/streamers` 覆盖状态字段。
///
/// 探测前就能判定的条件(录制时间范围)按当前时钟实时计算,不受监控循环节奏影响;
/// 需要房间标题才能判定的条件取最近一次探测的结论。
fn rejection_status(streamer: &LiveStreamer, worker: Option<&Worker>) -> Option<&'static str> {
if let Some(rejection) = recording_policy::reject_before_probe(streamer) {
return Some(rejection.status());
}
worker
.and_then(Worker::rejection)
.as_ref()
.map(Rejection::status)
}
pub async fn get_streamers_endpoint(
State(pool): State<ConnectionPool>,
State(managers): State<Arc<DownloadManager>>,
@@ -52,6 +67,14 @@ pub async fn get_streamers_endpoint(
Some(t) => format!("{:?}", *t.downloader_status.read().unwrap()),
None => String::new(),
};
// 被录制策略挡下时报告具体原因,否则「不在时间范围内」和「单纯没开播」
// 在界面上都是「空闲」,用户无从判断配置有没有生效。
// 时间类条件按当前时钟实时算,不依赖监控循环轮到这个房间,因此不会过期;
// 依赖房间标题的结论来自最近一次探测。正在录制时不覆盖。
let status = match rejection_status(&x, option.as_deref()) {
Some(reason) if status != "Working" => reason.to_string(),
_ => status,
};
results.push(LiveStreamerResponse {
status,
@@ -692,3 +715,179 @@ pub async fn post_uploads(
Ok(Json(serde_json::json!({})))
}
#[cfg(test)]
mod recording_policy_status_tests {
use super::*;
use crate::server::infrastructure::connection_pool::ConnectionManager;
use chrono::{Duration as ChronoDuration, SecondsFormat};
use serde_json::Value;
/// 以当前时刻为基准造窗口,形态与前端 `Date.toISOString()` 写出的一致
fn window(starts_in: i64, ends_in: i64) -> String {
let now = Utc::now();
let iso = |offset: i64| {
(now + ChronoDuration::seconds(offset)).to_rfc3339_opts(SecondsFormat::Millis, true)
};
format!(r#"["{}","{}"]"#, iso(starts_in), iso(ends_in))
}
fn insert(url: &str, time_range: Option<String>) -> InsertLiveStreamer {
InsertLiveStreamer {
url: url.to_string(),
remark: url.to_string(),
filename_prefix: None,
time_range,
upload_streamers_id: None,
format: None,
override_cfg: None,
preprocessor: None,
segment_processor: None,
downloaded_processor: None,
postprocessor: None,
opt_args: None,
excluded_keywords: None,
}
}
fn worker_for(streamer: LiveStreamer) -> Worker {
Worker::new(
streamer,
None,
Arc::new(RwLock::new(Config::default())),
Default::default(),
)
}
/// 走完整的 数据库 -> /v1/streamers 处理函数 -> JSON 这条路
#[tokio::test]
async fn the_streamers_endpoint_reports_why_a_streamer_is_not_recording() {
let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("data.sqlite3");
let pool = ConnectionManager::new_pool(db.to_str().unwrap())
.await
.unwrap();
// 窗口尚未开始 -> 应报告 OutOfSchedule
insert("https://live.example.com/closed", Some(window(3600, 7200)))
.insert(&pool)
.await
.unwrap();
// 窗口正开着 -> 不该被覆盖
insert("https://live.example.com/open", Some(window(-60, 3600)))
.insert(&pool)
.await
.unwrap();
// 没配时间范围 -> 不该被覆盖
insert("https://live.example.com/always", None)
.insert(&pool)
.await
.unwrap();
let managers = Arc::new(DownloadManager::new(1, 0, pool.clone()));
let Json(responses) = get_streamers_endpoint(State(pool), State(managers))
.await
.expect("接口应返回成功");
let status_of = |url: &str| {
responses
.iter()
.find(|r| r.inner.url == url)
.unwrap_or_else(|| panic!("响应里应有 {url}"))
.status
.clone()
};
assert_eq!(
status_of("https://live.example.com/closed"),
"OutOfSchedule"
);
assert_eq!(status_of("https://live.example.com/open"), "");
assert_eq!(status_of("https://live.example.com/always"), "");
}
#[tokio::test]
async fn the_endpoint_response_serialises_the_status_frontend_switches_on() {
let dir = tempfile::tempdir().unwrap();
let db = dir.path().join("data.sqlite3");
let pool = ConnectionManager::new_pool(db.to_str().unwrap())
.await
.unwrap();
insert("https://live.example.com/closed", Some(window(3600, 7200)))
.insert(&pool)
.await
.unwrap();
let managers = Arc::new(DownloadManager::new(1, 0, pool.clone()));
let Json(responses) = get_streamers_endpoint(State(pool), State(managers))
.await
.unwrap();
// 前端 app/(app)/streamers/page.tsx 就是对这个字符串做 switch
let json: Value = serde_json::to_value(&responses[0]).unwrap();
assert_eq!(json["status"], "OutOfSchedule");
}
#[test]
fn a_title_excluded_worker_reports_title_excluded() {
let streamer = LiveStreamer {
id: 1,
url: "https://live.example.com/x".to_string(),
remark: "x".to_string(),
filename_prefix: None,
time_range: None,
upload_streamers_id: None,
format: None,
override_cfg: None,
preprocessor: None,
segment_processor: None,
downloaded_processor: None,
postprocessor: None,
opt_args: None,
excluded_keywords: None,
};
let worker = worker_for(streamer.clone());
// 没有任何拦截时不覆盖状态
assert_eq!(rejection_status(&streamer, Some(&worker)), None);
// 探测后记录下的标题拦截会被报告出来
worker.set_rejection(Some(Rejection::ExcludedKeyword("录像".to_string())));
assert_eq!(
rejection_status(&streamer, Some(&worker)),
Some("TitleExcluded")
);
// 探测结果推翻后清空
worker.set_rejection(None);
assert_eq!(rejection_status(&streamer, Some(&worker)), None);
}
#[test]
fn the_time_window_outranks_a_stale_title_rejection() {
let streamer = LiveStreamer {
id: 1,
url: "https://live.example.com/x".to_string(),
remark: "x".to_string(),
filename_prefix: None,
time_range: Some(window(3600, 7200)),
upload_streamers_id: None,
format: None,
override_cfg: None,
preprocessor: None,
segment_processor: None,
downloaded_processor: None,
postprocessor: None,
opt_args: None,
excluded_keywords: None,
};
let worker = worker_for(streamer.clone());
worker.set_rejection(Some(Rejection::ExcludedKeyword("录像".to_string())));
// 时间范围按当前时钟实时算、不会过期,优先于上一次探测留下的结论
assert_eq!(
rejection_status(&streamer, Some(&worker)),
Some("OutOfSchedule")
);
}
}
@@ -1,3 +1,4 @@
use crate::server::common::recording_policy;
use crate::server::common::upload::UploaderMessage;
use crate::server::common::util::FileValidator;
use crate::server::core::downloader::cover_downloader;
@@ -203,6 +204,19 @@ impl DownloadTask {
}
}
// 录制策略:条件已不成立就不再续录,把房间交回监控循环。
// 少了这一步,下载器按边界收尾后循环会立刻重开一段,等于策略形同虚设。
// 每轮都重新判定,对齐 Python 版每轮 `run()` 前重新调用 `should_record()`。
//
// 放在 check_stream 之后有两个原因:用得到刚刷新的房间标题;且能同时覆盖
// 探测失败的分支——那条路径只加重试计数,否则会带着已失效的策略绕回去重录。
if let Some(rejection) =
recording_policy::reject_before_record(ctx.live_streamer(), &stream.title)
{
info!(url = url, reason = %rejection, "停止录制");
break components;
}
if retry_count >= max_retries {
warn!(
url = url,
@@ -3,6 +3,10 @@ use std::collections::HashMap;
use std::str::FromStr;
pub mod download;
/// 录制准入策略
pub mod recording_policy;
/// 录制时间范围判定
pub mod timerange;
pub mod upload;
/// 通用工具函数
pub mod util;
@@ -0,0 +1,248 @@
//! 录制准入策略:集中回答「现在该不该录这个主播」。
//!
//! Python 版把这些条件放在 `should_record()` 里(`biliup/engine/download.py`)。
//! Rust 重写时这个概念整体消失了,于是 `time_range` 和 `excluded_keywords`
//! 虽然一路存到了数据库和界面,运行时却没有任何代码读它们(issue #1654)。
//! 这个模块把「录制前提」重新变成一个有名字、有测试、可枚举的东西:
//! 新增条件时只要加一个 [`Rejection`] 分支和一行判定,三个调用点会一起生效。
//!
//! 判定分两级,依据是**这个条件需要哪些信息**,而这决定了它最早能在什么时候求值:
//!
//! * [`reject_before_probe`] —— 只看主播配置和时钟,开播探测之前就能判定,
//! 因此可以在不占用下载槽位、不发任何请求的情况下把房间筛掉。
//! * [`reject_before_record`] —— 还要看直播流信息(房间标题),只能在
//! `check_stream` 拿到流之后判定;它是前者的超集。
//!
//! Python 版把两者混在一个探测之后的 `should_record()` 里,所以不在录制时间范围内
//! 的主播照样要发一次网络请求;这里按信息依赖分级,避免了那次无谓的请求。
use std::fmt;
use serde_json::Value;
use crate::server::common::timerange;
use crate::server::infrastructure::models::live_streamer::LiveStreamer;
/// 不予录制的原因。
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Rejection {
/// 当前不在配置的录制时间范围内
OutOfTimeRange,
/// 房间标题命中了排除关键词
ExcludedKeyword(String),
}
impl Rejection {
/// `/v1/streamers` 返回给前端的状态字符串。
///
/// `OutOfSchedule` 沿用 Python 版 `biliup/web/__init__.py` 的取值,
/// 前端 `app/(app)/streamers/page.tsx` 一直保留着对应的「非录播时间」标签。
pub fn status(&self) -> &'static str {
match self {
Rejection::OutOfTimeRange => "OutOfSchedule",
Rejection::ExcludedKeyword(_) => "TitleExcluded",
}
}
}
impl fmt::Display for Rejection {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Rejection::OutOfTimeRange => f.write_str("不在录制时间范围内"),
Rejection::ExcludedKeyword(keyword) => {
write!(f, "房间标题命中排除关键词 {keyword:?}")
}
}
}
}
/// 只依赖主播配置和时钟的条件,开播探测之前即可判定。
///
/// 监控循环在占用下载槽位、发出任何请求之前调用它。
pub fn reject_before_probe(streamer: &LiveStreamer) -> Option<Rejection> {
if !timerange::is_within(streamer.time_range.as_deref()) {
return Some(Rejection::OutOfTimeRange);
}
None
}
/// 需要直播流信息才能判定的条件,是 [`reject_before_probe`] 的超集。
///
/// 在 `check_stream` 取得流之后、真正开录之前调用,续录前每轮也要重新判定
/// ——房间标题和当前时间都可能在录制过程中变化。
pub fn reject_before_record(streamer: &LiveStreamer, title: &str) -> Option<Rejection> {
if let Some(rejection) = reject_before_probe(streamer) {
return Some(rejection);
}
if let Some(keyword) = matched_excluded_keyword(streamer, title) {
return Some(Rejection::ExcludedKeyword(keyword));
}
None
}
/// 房间标题命中的第一个排除关键词。
///
/// 与 Python 版一致地做大小写敏感的子串匹配并去掉关键词首尾空白;
/// 但会跳过空关键词——Python 版 `"" in title` 恒真,一个空字符串就会让该主播完全录不到。
fn matched_excluded_keyword(streamer: &LiveStreamer, title: &str) -> Option<String> {
if title.is_empty() {
return None;
}
streamer
.excluded_keywords
.as_ref()
.and_then(Value::as_array)?
.iter()
.filter_map(Value::as_str)
.map(str::trim)
.filter(|keyword| !keyword.is_empty())
.find(|keyword| title.contains(*keyword))
.map(str::to_owned)
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::{Duration, SecondsFormat, Utc};
/// 以当前时刻为基准造窗口,形态与前端 `Date.toISOString()` 写出的一致
fn window(starts_in: i64, ends_in: i64) -> String {
let now = Utc::now();
let iso = |offset: i64| {
(now + Duration::seconds(offset)).to_rfc3339_opts(SecondsFormat::Millis, true)
};
format!(r#"["{}","{}"]"#, iso(starts_in), iso(ends_in))
}
fn streamer(time_range: Option<String>, excluded_keywords: Option<Value>) -> LiveStreamer {
LiveStreamer {
id: 1,
url: "https://www.douyu.com/5146843".to_string(),
remark: "test".to_string(),
filename_prefix: None,
time_range,
upload_streamers_id: None,
format: None,
override_cfg: None,
preprocessor: None,
segment_processor: None,
downloaded_processor: None,
postprocessor: None,
opt_args: None,
excluded_keywords,
}
}
fn keywords(values: &[&str]) -> Option<Value> {
Some(Value::Array(
values
.iter()
.map(|k| Value::String((*k).to_string()))
.collect(),
))
}
#[test]
fn a_streamer_without_any_condition_is_always_recorded() {
let streamer = streamer(None, None);
assert_eq!(reject_before_probe(&streamer), None);
assert_eq!(reject_before_record(&streamer, "随便什么标题"), None);
}
#[test]
fn an_open_window_and_a_clean_title_are_recorded() {
let streamer = streamer(Some(window(-60, 3600)), keywords(&["录像", "重播"]));
assert_eq!(reject_before_probe(&streamer), None);
assert_eq!(reject_before_record(&streamer, "今天来点好活"), None);
}
#[test]
fn a_closed_window_is_rejected_before_the_probe() {
// 关键性质:这个条件在探测之前就能判定,才能省掉下载槽位和网络请求
let streamer = streamer(Some(window(3600, 7200)), None);
assert_eq!(
reject_before_probe(&streamer),
Some(Rejection::OutOfTimeRange)
);
}
#[test]
fn an_excluded_keyword_is_only_visible_after_the_probe() {
// 标题类条件探测前无从判断,探测后才拒绝
let streamer = streamer(None, keywords(&["录像"]));
assert_eq!(reject_before_probe(&streamer), None);
assert_eq!(
reject_before_record(&streamer, "昨日录像回放"),
Some(Rejection::ExcludedKeyword("录像".to_string()))
);
}
#[test]
fn the_post_probe_check_still_enforces_the_pre_probe_conditions() {
// reject_before_record 必须是 reject_before_probe 的超集,
// 否则新增一个「探测前条件」时会在开录那一步漏掉
let streamer = streamer(Some(window(3600, 7200)), None);
assert_eq!(
reject_before_record(&streamer, "无关标题"),
Some(Rejection::OutOfTimeRange)
);
}
#[test]
fn keywords_are_matched_as_trimmed_substrings() {
let streamer = streamer(None, keywords(&[" 重播 "]));
assert_eq!(
reject_before_record(&streamer, "这是重播内容"),
Some(Rejection::ExcludedKeyword("重播".to_string()))
);
}
#[test]
fn an_empty_keyword_does_not_block_everything() {
// Python 版 `"" in title` 恒真,一个空关键词会让主播彻底录不到
let streamer = streamer(None, keywords(&["", " "]));
assert_eq!(reject_before_record(&streamer, "正常直播"), None);
}
#[test]
fn an_empty_title_skips_the_keyword_check() {
let streamer = streamer(None, keywords(&["录像"]));
assert_eq!(reject_before_record(&streamer, ""), None);
}
#[test]
fn malformed_keywords_fall_open_instead_of_blocking() {
for raw in [
Value::Null,
Value::String("录像".to_string()),
Value::Number(1.into()),
Value::Array(vec![Value::Number(1.into())]),
] {
let streamer = streamer(None, Some(raw.clone()));
assert_eq!(
reject_before_record(&streamer, "昨日录像回放"),
None,
"raw={raw}"
);
}
}
#[test]
fn keyword_matching_is_case_sensitive_like_python() {
let streamer = streamer(None, keywords(&["Replay"]));
assert_eq!(reject_before_record(&streamer, "replay time"), None);
assert_eq!(
reject_before_record(&streamer, "Replay time"),
Some(Rejection::ExcludedKeyword("Replay".to_string()))
);
}
#[test]
fn rejections_carry_the_status_string_the_frontend_expects() {
assert_eq!(Rejection::OutOfTimeRange.status(), "OutOfSchedule");
assert_eq!(
Rejection::ExcludedKeyword("x".to_string()).status(),
"TitleExcluded"
);
}
}
@@ -0,0 +1,331 @@
//! 录制时间范围(`time_range`)的解析与判定。
//!
//! 存储格式是一个含两个 ISO 8601 时刻的 JSON 数组字符串,例如
//! `["2025-03-26T16:00:00.000Z","2025-03-27T15:59:59.000Z"]`。
//!
//! 前端时间选择器拿到的是用户本地时刻,写库前经 `Date.toISOString()` 转成 UTC
//! (`app/ui/TemplateModal.tsx`),所以两端存的都是 **UTC 时刻**,日期部分只是
//! 选择当天的残留、没有意义。判定时统一取 UTC 当日秒数比较,正好抵消前端写入时
//! 的时区偏移,与 Python 版 `check_timerange()` 用 `datetime.now(timezone.utc)`
//! 比较的行为一致。若改用本地时间判定,UTC+8 用户配置的 16:00–20:00 会被当成
//! 08:00–12:00,整整错开 8 小时。
use chrono::{DateTime, NaiveDateTime, Timelike, Utc};
use tracing::debug;
const SECONDS_PER_DAY: u32 = 24 * 60 * 60;
/// 一个录制窗口,端点为 UTC 当日秒数的闭区间。
/// `start > end` 表示跨午夜区间(如 23:00 → 04:00)。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct TimeRange {
start: u32,
end: u32,
}
impl TimeRange {
/// 解析存储的 `time_range` 字符串,任何形式的非法输入都返回 `None`。
pub fn parse(raw: &str) -> Option<Self> {
let bounds: Vec<String> = serde_json::from_str(raw)
.inspect_err(|e| debug!(raw = raw, error = %e, "time_range 不是字符串数组,忽略"))
.ok()?;
let [start, end] = <[String; 2]>::try_from(bounds)
.inspect_err(|_| debug!(raw = raw, "time_range 需要恰好两个时刻,忽略"))
.ok()?;
Some(Self {
start: utc_seconds_from_midnight(&start)?,
end: utc_seconds_from_midnight(&end)?,
})
}
/// `now`(UTC 当日秒数)是否落在窗口内。
fn contains(&self, now: u32) -> bool {
if self.start <= self.end {
// 普通区间,如 16:00 → 20:00
self.start <= now && now <= self.end
} else {
// 跨午夜区间,如 23:00 → 04:00
now >= self.start || now <= self.end
}
}
/// 距窗口结束还有多少秒;不在窗口内返回 0。
///
/// 跨午夜窗口在午夜之前的那段里 `end < now`,必须跨到次日的结束时刻才对。
/// 但「已经错过窗口」同样满足 `end < now`,不先判定在不在窗口内就会算出
/// 「还剩将近 24 小时」——正好是这个功能要防的那种失控录制。
fn seconds_until_end(&self, now: u32) -> u32 {
if !self.contains(now) {
return 0;
}
if self.end >= now {
self.end - now
} else {
SECONDS_PER_DAY - now + self.end
}
}
}
/// 取 ISO 8601 时刻换算到 UTC 后的当日秒数。
///
/// 前端写入的一定带 `Z`;手写配置若带其他偏移量(如 `+08:00`)同样先换算到 UTC,
/// 完全不带时区信息时按 UTC 解释。
fn utc_seconds_from_midnight(raw: &str) -> Option<u32> {
let time = match DateTime::parse_from_rfc3339(raw) {
Ok(dt) => dt.with_timezone(&Utc).time(),
Err(_) => raw
.parse::<NaiveDateTime>()
.inspect_err(|e| debug!(raw = raw, error = %e, "time_range 时刻无法解析,忽略"))
.ok()?
.time(),
};
Some(time.num_seconds_from_midnight())
}
fn now_seconds_from_midnight() -> u32 {
Utc::now().time().num_seconds_from_midnight()
}
/// 当前是否处于配置的录制时间范围内。
///
/// 未配置或配置无法解析时一律放行:坏配置只应让时间范围失效,不该让主播完全录不到。
/// 与 Python 版 `check_timerange()` 出错即 `return True` 的取舍一致。
pub fn is_within(time_range: Option<&str>) -> bool {
is_within_at(time_range, now_seconds_from_midnight())
}
fn is_within_at(time_range: Option<&str>, now: u32) -> bool {
match time_range.and_then(TimeRange::parse) {
Some(range) => range.contains(now),
None => true,
}
}
/// 距录制时间范围结束还剩多久(`"HH:MM:SS"`);未配置或配置非法时为 `None`。
///
/// 给「自己不会退出」的录制方式收尾用(如 ffmpeg 内部分段由 segment muxer 持续切片),
/// 需要一个总时长上限把进程截停在窗口边界。
pub fn remaining_until_end(time_range: Option<&str>) -> Option<String> {
remaining_until_end_at(time_range, now_seconds_from_midnight())
}
fn remaining_until_end_at(time_range: Option<&str>, now: u32) -> Option<String> {
let range = time_range.and_then(TimeRange::parse)?;
Some(format_hms(range.seconds_until_end(now)))
}
/// 计算本次录制块允许的最长时长(`"HH:MM:SS"`)。
///
/// 在 `segment_time` 的基础上按窗口结束时刻裁剪:快到结束时间时缩短本段,
/// 让录制正好停在窗口边界而不是冲出去一整段。等价于 Python 版 `get_duration()`,
/// 但额外覆盖了「配了时间范围却没配分段时长」的情况——Python 那种情形下
/// `-to` 不会下发,录制会一直冲到直播结束。
pub fn clamp_segment_time(segment_time: Option<&str>, time_range: Option<&str>) -> Option<String> {
clamp_segment_time_at(segment_time, time_range, now_seconds_from_midnight())
}
fn clamp_segment_time_at(
segment_time: Option<&str>,
time_range: Option<&str>,
now: u32,
) -> Option<String> {
let Some(range) = time_range.and_then(TimeRange::parse) else {
return segment_time.map(str::to_owned);
};
let remaining = range.seconds_until_end(now);
match segment_time {
// 分段时长无法解析时原样下发,交给下游工具自己报错
Some(configured) => match parse_hms(configured) {
Some(seconds) if seconds > remaining => Some(format_hms(remaining)),
Some(_) | None => Some(configured.to_owned()),
},
None => Some(format_hms(remaining)),
}
}
/// 解析 `"HH:MM:SS"` 为秒数。
fn parse_hms(raw: &str) -> Option<u32> {
let [hours, minutes, seconds] =
<[&str; 3]>::try_from(raw.split(':').collect::<Vec<_>>()).ok()?;
Some(
hours.parse::<u32>().ok()? * 3600
+ minutes.parse::<u32>().ok()? * 60
+ seconds.parse::<u32>().ok()?,
)
}
fn format_hms(total: u32) -> String {
format!(
"{:02}:{:02}:{:02}",
total / 3600,
(total % 3600) / 60,
total % 60
)
}
#[cfg(test)]
mod tests {
use super::*;
/// 把 "HH:MM:SS" 写成当日秒数,让用例读起来贴近配置里的时刻
fn at(hms: &str) -> u32 {
parse_hms(hms).expect("测试时刻格式应合法")
}
/// 前端 `Date.toISOString()` 写出的形态
fn range(start: &str, end: &str) -> String {
format!(r#"["2025-03-26T{start}.000Z","2025-03-27T{end}.000Z"]"#)
}
#[test]
fn missing_time_range_does_not_restrict_recording() {
assert!(is_within_at(None, at("03:00:00")));
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), None, at("03:00:00")),
Some("01:00:00".to_string())
);
}
#[test]
fn malformed_time_range_falls_open_instead_of_blocking() {
// 坏配置只让时间范围失效,不能把主播整个录制掐掉
for raw in [
"",
"null",
"not json",
"[]",
r#"["2025-03-26T16:00:00.000Z"]"#,
r#"["a","b"]"#,
r#"["2025-03-26T16:00:00.000Z","2025-03-26T16:00:00.000Z","x"]"#,
] {
assert!(is_within_at(Some(raw), at("03:00:00")), "raw={raw}");
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), Some(raw), at("03:00:00")),
Some("01:00:00".to_string()),
"raw={raw}"
);
}
}
#[test]
fn normal_interval_admits_only_inside_the_window() {
let tr = range("16:00:00", "20:00:00");
assert!(!is_within_at(Some(&tr), at("15:59:59")));
assert!(is_within_at(Some(&tr), at("16:00:00"))); // 起点闭区间
assert!(is_within_at(Some(&tr), at("18:00:00")));
assert!(is_within_at(Some(&tr), at("20:00:00"))); // 终点闭区间
assert!(!is_within_at(Some(&tr), at("20:00:01")));
}
#[test]
fn cross_midnight_interval_admits_both_sides_of_midnight() {
let tr = range("23:00:00", "04:00:00");
assert!(is_within_at(Some(&tr), at("23:30:00")));
assert!(is_within_at(Some(&tr), at("00:00:00")));
assert!(is_within_at(Some(&tr), at("03:59:59")));
assert!(!is_within_at(Some(&tr), at("04:00:01")));
assert!(!is_within_at(Some(&tr), at("12:00:00")));
assert!(!is_within_at(Some(&tr), at("22:59:59")));
}
/// 前端存的是 UTC,判定也必须按 UTC,否则 UTC+8 用户会整体错开 8 小时
#[test]
fn offset_bearing_instants_are_converted_to_utc() {
// 16:00+08:00 == 08:00Z
let tr = r#"["2025-03-26T16:00:00+08:00","2025-03-26T20:00:00+08:00"]"#;
assert!(is_within_at(Some(tr), at("09:00:00")));
assert!(!is_within_at(Some(tr), at("17:00:00")));
}
#[test]
fn naive_instants_are_read_as_utc() {
let tr = r#"["2025-03-26T16:00:00","2025-03-26T20:00:00"]"#;
assert!(is_within_at(Some(tr), at("17:00:00")));
assert!(!is_within_at(Some(tr), at("09:00:00")));
}
#[test]
fn segment_time_is_untouched_while_the_window_end_is_far() {
let tr = range("16:00:00", "20:00:00");
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), Some(&tr), at("16:30:00")),
Some("01:00:00".to_string())
);
}
#[test]
fn segment_time_is_shortened_near_the_window_end() {
let tr = range("16:00:00", "20:00:00");
// 19:40 距结束还有 20 分钟,小于 1 小时的分段时长
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), Some(&tr), at("19:40:00")),
Some("00:20:00".to_string())
);
}
#[test]
fn segment_time_matching_the_remaining_time_is_not_shortened() {
let tr = range("16:00:00", "20:00:00");
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), Some(&tr), at("19:00:00")),
Some("01:00:00".to_string())
);
}
#[test]
fn missing_segment_time_still_ends_at_the_window_end() {
// Python 版这种情况不下发 -to,会一直录到直播结束;这里补上
let tr = range("16:00:00", "20:00:00");
assert_eq!(
clamp_segment_time_at(None, Some(&tr), at("19:40:00")),
Some("00:20:00".to_string())
);
}
#[test]
fn remaining_time_of_a_cross_midnight_window_wraps_past_zero() {
let tr = range("23:00:00", "04:00:00");
// 23:30 距次日 04:00 还有 4 小时 30 分
assert_eq!(
clamp_segment_time_at(Some("10:00:00"), Some(&tr), at("23:30:00")),
Some("04:30:00".to_string())
);
}
#[test]
fn unparsable_segment_time_is_passed_through() {
let tr = range("16:00:00", "20:00:00");
for configured in ["", "abc", "1:2", "01:00:00:00", "-1:00:00"] {
assert_eq!(
clamp_segment_time_at(Some(configured), Some(&tr), at("19:40:00")),
Some(configured.to_string()),
"configured={configured}"
);
}
}
#[test]
fn sitting_exactly_on_the_window_end_leaves_zero_remaining() {
let tr = range("16:00:00", "20:00:00");
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), Some(&tr), at("20:00:00")),
Some("00:00:00".to_string())
);
}
/// 窗口外 `end < now` 与跨午夜窗口的形态相同,若不先判定在不在窗口内,
/// 会算出「还剩近 24 小时」,反而放行一整天的录制
#[test]
fn outside_the_window_nothing_remains_instead_of_wrapping_to_tomorrow() {
let tr = range("16:00:00", "20:00:00");
assert_eq!(
clamp_segment_time_at(Some("01:00:00"), Some(&tr), at("20:00:01")),
Some("00:00:00".to_string())
);
assert_eq!(
remaining_until_end_at(Some(&tr), at("06:00:00")),
Some("00:00:00".to_string())
);
}
}
@@ -6,6 +6,7 @@ pub mod stream_gears;
pub mod streamlink;
pub mod ytdlp;
use crate::server::common::timerange;
use crate::server::common::util::Recorder;
use crate::server::core::downloader::ffmpeg_downloader::FfmpegDownloader;
use crate::server::core::downloader::stream_gears::StreamGears;
@@ -29,6 +30,10 @@ pub struct DownloadConfig {
/// 分段时长 (格式: "HH:MM:SS")
pub segment_time: Option<String>,
/// 录制时间范围,两个 ISO 8601 时刻的 JSON 数组字符串。
/// 见 [`crate::server::common::timerange`]。
pub time_range: Option<String>,
/// 分段文件大小限制 (字节)
pub file_size: Option<u64>,
@@ -52,6 +57,20 @@ impl DownloadConfig {
fn generate_output_filename(&self, suffix: &str) -> PathBuf {
self.output_dir.join(self.recorder.generate_path(suffix))
}
/// 本次录制块允许的最长时长(`"HH:MM:SS"`)。
///
/// 即 `segment_time` 按录制时间范围的结束时刻裁剪后的结果:快到窗口结束时缩短本段,
/// 让录制停在窗口边界。各下载器都以「本段录多久」的语义使用它,
/// 因此必须用这个方法而不是直接读 [`Self::segment_time`]。
pub fn segment_duration(&self) -> Option<String> {
timerange::clamp_segment_time(self.segment_time.as_deref(), self.time_range.as_deref())
}
/// 距录制时间范围结束还剩多久(`"HH:MM:SS"`);未配置录制时间范围时为 `None`。
pub fn time_range_remaining(&self) -> Option<String> {
timerange::remaining_until_end(self.time_range.as_deref())
}
}
/// 下载器类型枚举
@@ -81,6 +81,12 @@ impl FfmpegDownloader {
args.extend(["-segment_time".to_string(), seconds.to_string()]);
}
// -t: 录制总时长上限。内部分段由 segment muxer 自己切片、进程不会自行退出,
// 所以要用总时长把录制截停在录制时间范围的结束时刻。
if let Some(remaining) = download_config.time_range_remaining() {
args.extend(["-t".to_string(), remaining]);
}
// 添加通用输出参数
self.append_common_output_args(&mut args, "segment");
@@ -99,9 +105,9 @@ impl FfmpegDownloader {
self.append_common_input_args(&mut args, download_config);
// 外部分段特定的输出参数
// -to: 限制录制时长
if let Some(segment_time) = &download_config.segment_time {
args.extend(["-to".to_string(), segment_time.clone()]);
// -to: 限制录制时长,快到录制时间范围结束时会被裁短,使录制停在窗口边界
if let Some(segment_time) = download_config.segment_duration() {
args.extend(["-to".to_string(), segment_time]);
}
// -fs: 限制文件大小(字节)
@@ -390,3 +396,108 @@ async fn spawn_log(
};
Ok(status)
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::{Duration as ChronoDuration, SecondsFormat, Utc};
/// 以「当前时刻」为基准造一个录制时间范围,形态与前端 `Date.toISOString()` 写出的一致。
/// 相对当前时刻取值,因此走的是真实时钟,也顺带覆盖了窗口跨过零点的情形。
fn window(starts_in: i64, ends_in: i64) -> String {
let now = Utc::now();
let iso = |offset: i64| {
(now + ChronoDuration::seconds(offset)).to_rfc3339_opts(SecondsFormat::Millis, true)
};
format!(r#"["{}","{}"]"#, iso(starts_in), iso(ends_in))
}
fn config(segment_time: Option<&str>, time_range: Option<String>) -> DownloadConfig {
DownloadConfig {
segment_time: segment_time.map(str::to_owned),
time_range,
suffix: "flv".to_string(),
..Default::default()
}
}
fn value_of(args: &[String], flag: &str) -> Option<String> {
let index = args.iter().position(|arg| arg == flag)?;
args.get(index + 1).cloned()
}
fn seconds_of(args: &[String], flag: &str) -> u32 {
let raw = value_of(args, flag).unwrap_or_else(|| panic!("命令行里应有 {flag}"));
let parts: Vec<u32> = raw
.split(':')
.map(|p| p.parse().expect("时长应为 HH:MM:SS"))
.collect();
parts[0] * 3600 + parts[1] * 60 + parts[2]
}
fn external() -> FfmpegDownloader {
FfmpegDownloader::new(Vec::new(), DownloaderType::FfmpegExternal)
}
fn internal() -> FfmpegDownloader {
FfmpegDownloader::new(Vec::new(), DownloaderType::FfmpegInternal)
}
#[test]
fn without_a_time_range_the_segment_time_reaches_ffmpeg_unchanged() {
let args = external().build_ffmpeg_args_external_segment(&config(Some("01:00:00"), None));
assert_eq!(value_of(&args, "-to"), Some("01:00:00".to_string()));
}
#[test]
fn without_a_segment_time_or_window_ffmpeg_gets_no_duration_limit() {
let args = external().build_ffmpeg_args_external_segment(&config(None, None));
assert_eq!(value_of(&args, "-to"), None);
}
#[test]
fn a_far_away_window_end_leaves_the_segment_time_alone() {
// 窗口还剩 2 小时,1 小时的分段时长不该被动
let args = external()
.build_ffmpeg_args_external_segment(&config(Some("01:00:00"), Some(window(-60, 7200))));
assert_eq!(value_of(&args, "-to"), Some("01:00:00".to_string()));
}
#[test]
fn a_near_window_end_shortens_the_segment_so_recording_stops_on_the_boundary() {
// 窗口只剩 10 分钟,1 小时的分段必须被裁到 10 分钟,否则会冲出窗口 50 分钟
let args = external()
.build_ffmpeg_args_external_segment(&config(Some("01:00:00"), Some(window(-60, 600))));
let to = seconds_of(&args, "-to");
assert!((595..=600).contains(&to), "-to 应约为 600 秒,实际 {to}");
}
#[test]
fn a_window_bounds_recording_even_when_no_segment_time_is_configured() {
// Python 版这种情况根本不下发 -to,会一直录到直播结束
let args =
external().build_ffmpeg_args_external_segment(&config(None, Some(window(-60, 600))));
let to = seconds_of(&args, "-to");
assert!((595..=600).contains(&to), "-to 应约为 600 秒,实际 {to}");
}
#[test]
fn internal_segmentation_caps_total_duration_at_the_window_end() {
// 内部分段的 -segment_time 只是切片间隔,进程不会自己退出,必须靠 -t 截停
let args = internal()
.build_ffmpeg_args_internal_segment(&config(Some("01:00:00"), Some(window(-60, 600))));
assert_eq!(value_of(&args, "-segment_time"), Some("3600".to_string()));
let total = seconds_of(&args, "-t");
assert!(
(595..=600).contains(&total),
"-t 应约为 600 秒,实际 {total}"
);
}
#[test]
fn internal_segmentation_has_no_total_cap_without_a_window() {
let args = internal().build_ffmpeg_args_internal_segment(&config(Some("01:00:00"), None));
assert_eq!(value_of(&args, "-segment_time"), Some("3600".to_string()));
assert_eq!(value_of(&args, "-t"), None);
}
}
@@ -49,7 +49,11 @@ impl StreamGears {
let headers_in = construct_headers(&download_config.headers).map_err(AppError::Custom)?;
let proxy = self.proxy.clone();
let segment = Segmentable::new(
download_config.segment_time.as_deref().map(parse_time),
// 快到录制时间范围结束时会被裁短,使录制停在窗口边界
download_config
.segment_duration()
.as_deref()
.map(parse_time),
download_config.file_size,
);
@@ -157,9 +157,10 @@ impl StreamlinkDownloader {
args.push("--http-header".to_string());
args.push(format!("{}={}", key, value));
}
if let Some(segment_time) = &download_config.segment_time {
// 快到录制时间范围结束时会被裁短,使录制停在窗口边界
if let Some(segment_time) = download_config.segment_duration() {
args.push("--hls-duration".to_string());
args.push(segment_time.clone());
args.push(segment_time);
}
args.push("--force".to_string());
args.push("--output".to_string());
+30 -4
View File
@@ -1,4 +1,5 @@
use crate::server::common::download::start_download_workflow;
use crate::server::common::recording_policy;
use crate::server::common::upload::UploaderMessage;
use crate::server::core::live::{batch_check_request, live_request, streamer_info};
use crate::server::infrastructure::connection_pool::ConnectionPool;
@@ -34,9 +35,10 @@ enum BatchVerdict {
Fallback,
}
/// 批量结果为未开播时的快速轮换间隔:缓存命中无网络开销,
/// 短暂停即可在一个检测周期内扫完整个队列(网络频率仍由缓存 TTL 限制)。
const BATCH_ROTATE_SLEEP: Duration = Duration::from_secs(1);
/// 无网络开销的跳过路径(批量结果判定未开播、不在录制时间范围内)使用的快速轮换间隔:
/// 这些判定不发请求,短暂停即可在一个检测周期内扫完整个队列,
/// 也让窗口一开就能立刻开录(网络频率仍由批量缓存 TTL 限制)。
const QUICK_ROTATE_SLEEP: Duration = Duration::from_secs(1);
/// 房间处理器
/// 管理多个直播间的状态和操作
@@ -118,6 +120,15 @@ impl Monitor {
.await;
let url = room.get_streamer().url.clone();
let interval = room.get_config().event_loop_interval;
// 探测前的录制策略:不满足就直接放回队列,既不占用下载槽位也不发任何请求。
// 这类条件只看配置和时钟、判定无开销,所以用快速轮换而不是整个检测周期,
// 条件一满足即可开录。
if let Some(rejection) = recording_policy::reject_before_probe(room.get_streamer()) {
self.wake_waker(room.id()).await;
debug!(url = url, reason = %rejection, "跳过检测");
tokio::time::sleep(QUICK_ROTATE_SLEEP).await;
continue;
}
// 批量检测平台:先用一个检测周期内共享的批量结果快速判定是否开播。
// 未开播直接跳过(不占用下载槽位、不做逐间请求),并以较短间隔轮换到
// 下一个房间,使整条队列在一个检测周期内扫完;批量请求本身由缓存 TTL 限流。
@@ -129,7 +140,7 @@ impl Monitor {
BatchVerdict::Offline => {
self.wake_waker(room.id()).await;
debug!(url = url, "批量检测未开播");
tokio::time::sleep(BATCH_ROTATE_SLEEP).await;
tokio::time::sleep(QUICK_ROTATE_SLEEP).await;
continue;
}
// Live / Fallback 都继续走下面的逐间完整检测
@@ -145,6 +156,18 @@ impl Monitor {
// 检查直播状态
match plugin.check_stream(request).await {
Ok(LiveStatus::Live { stream }) => {
// 依赖房间标题的策略要拿到流信息才能判定。命中就按「本轮不录」处理:
// 不建 StreamerInfo 记录、不启动下载,等下个检测周期再看。
if let Some(rejection) =
recording_policy::reject_before_record(room.get_streamer(), &stream.title)
{
room.set_rejection(Some(rejection.clone()));
self.wake_waker(room.id()).await;
info!(url = url, title = stream.title, reason = %rejection, "开播但不录制");
tokio::time::sleep(Duration::from_secs(interval)).await;
continue;
}
room.set_rejection(None);
let sql_no_id = streamer_info(&stream);
let insert = match StreamerInfo::builder()
.url(sql_no_id.url.clone())
@@ -179,10 +202,13 @@ impl Monitor {
info!("成功开始录制 {}", url);
}
Ok(LiveStatus::Offline) => {
// 探测结果推翻了上一轮的策略判定,清掉以免界面停在旧原因上
room.set_rejection(None);
self.wake_waker(room.id()).await;
debug!(url = room.get_streamer().url, "未开播")
}
Err(e) => {
room.set_rejection(None);
self.wake_waker(room.id()).await;
error!(e=?e, ctx=room.get_streamer().url,"检查直播间出错")
}
@@ -1,4 +1,5 @@
use crate::server::common::download::DownloadTask;
use crate::server::common::recording_policy::Rejection;
use crate::server::common::util::Recorder;
use crate::server::config::Config;
use crate::server::core::downloader::DownloadConfig;
@@ -121,6 +122,7 @@ impl Context {
// 流URL
url: stream.raw_stream_url.to_string(),
segment_time: config.segment_time,
time_range: self.live_streamer().time_range.clone(),
file_size: config.file_size,
headers: stream.stream_headers.clone(),
recorder: self.recorder(stream_info),
@@ -146,6 +148,11 @@ pub struct Worker {
config: Arc<RwLock<Config>>,
/// HTTP客户端
pub client: StatelessClient,
/// 最近一次开播探测被录制策略挡下的原因,仅用于向界面解释「为什么没在录」。
///
/// 不参与任何控制流。只存需要流信息才能判定的结论(如标题命中排除关键词);
/// 探测前就能判定的条件(如录制时间范围)由接口侧按当前时钟实时算,不会过期。
last_rejection: RwLock<Option<Rejection>>,
}
impl Worker {
@@ -169,6 +176,7 @@ impl Worker {
upload_streamer,
config,
client,
last_rejection: RwLock::new(None),
}
}
@@ -176,6 +184,16 @@ impl Worker {
self.live_streamer.id
}
/// 记录本轮开播探测的策略判定结果(`None` 表示未被挡下)。
pub fn set_rejection(&self, rejection: Option<Rejection>) {
*self.last_rejection.write().unwrap() = rejection;
}
/// 最近一次探测被挡下的原因。
pub fn rejection(&self) -> Option<Rejection> {
self.last_rejection.read().unwrap().clone()
}
/// 获取主播信息
/// 返回当前工作器关联的直播主播信息
pub fn get_streamer(&self) -> &LiveStreamer {