injectlib/bili-shadowreplay
1
0
Fork 0
mirror of https://github.com/Xinrea/bili-shadowreplay.git synced 2025-11-25 04:22:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: bilibili stream pathway not update (close #117）

Browse Source
This commit is contained in:
Xinrea
2025-07-26 22:10:03 +08:00
parent 3088500c8d
commit c4fdcf86d4
4 changed files with 296 additions and 208 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src-tauri/src/main.rs
View File

@@ -117,6 +117,9 @@ async fn setup_logging(log_dir: &Path) -> Result<(), Box<dyn std::error::Error>>
),
])?;
// logging current package version
log::info!("Current version: {}", env!("CARGO_PKG_VERSION"));
Ok(())
}

263
src-tauri/src/recorder/bilibili.rs
View File

@@ -13,7 +13,6 @@ use crate::subtitle_generator::item_to_srt;
use super::danmu::{DanmuEntry, DanmuStorage};
use super::entry::TsEntry;
use std::path::Path;
use chrono::Utc;
use client::{BiliClient, BiliStream, RoomInfo, StreamType, UserInfo};
use danmu_stream::danmu_stream::DanmuStream;
@@ -22,11 +21,12 @@ use danmu_stream::DanmuMessageType;
use errors::BiliClientError;
use m3u8_rs::{Playlist, QuotedOrUnquoted, VariantStream};
use regex::Regex;
use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
use std::path::Path;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::sync::{broadcast, Mutex, RwLock};
use tokio::task::JoinHandle;
use url::Url;
@@ -73,6 +73,7 @@ pub struct BiliRecorder {
header_changed_recently: Arc<AtomicBool>, // 标记最近是否发生了 header 变化
danmu_task: Arc<Mutex<Option<JoinHandle<()>>>>,
record_task: Arc<Mutex<Option<JoinHandle<()>>>>,
master_manifest: Arc<RwLock<Option<String>>>,
}
impl From<DatabaseError> for super::errors::RecorderError {
@@ -147,6 +148,7 @@ impl BiliRecorder {
header_changed_recently: Arc::new(AtomicBool::new(false)),
danmu_task: Arc::new(Mutex::new(None)),
record_task: Arc::new(Mutex::new(None)),
master_manifest: Arc::new(RwLock::new(None)),
};
log::info!("Recorder for room {} created.", options.room_id);
Ok(recorder)
@@ -161,7 +163,6 @@ impl BiliRecorder {
*self.last_segment_offset.write().await = None;
*self.current_header_url.write().await = None;
self.header_changed_recently.store(false, Ordering::Relaxed);
}
async fn should_record(&self) -> bool {
@@ -267,10 +268,12 @@ impl BiliRecorder {
return true;
}
let master_manifest =
m3u8_rs::parse_playlist_res(master_manifest.as_ref().unwrap().as_bytes())
let master_manifest = master_manifest.unwrap();
*self.master_manifest.write().await = Some(master_manifest.clone());
let master_manifest = m3u8_rs::parse_playlist_res(master_manifest.as_bytes())
.map_err(|_| super::errors::RecorderError::M3u8ParseFailed {
content: master_manifest.as_ref().unwrap().clone(),
content: master_manifest.clone(),
});
if master_manifest.is_err() {
log::error!(
@@ -312,6 +315,8 @@ impl BiliRecorder {
let variant = variant.unwrap();
log::info!("Variant: {:?}", variant);
let new_stream = self.stream_from_variant(variant).await;
if new_stream.is_err() {
log::error!(
@@ -325,26 +330,13 @@ impl BiliRecorder {
let stream = new_stream.unwrap();
let should_update_stream = self.live_stream.read().await.is_none()
|| !self
.live_stream
.read()
.await
.as_ref()
.unwrap()
.is_same(&stream)
|| self.force_update.load(Ordering::Relaxed);
|| self.force_update.load(Ordering::Relaxed)
|| self.header_changed_recently.load(Ordering::Relaxed);
if should_update_stream {
log::info!(
"[{}]Update to a new stream: {:?} => {}",
self.room_id,
self.live_stream.read().await.clone(),
stream
);
self.force_update.store(false, Ordering::Relaxed);
let new_stream = self.fetch_real_stream(stream).await;
let new_stream = self.fetch_real_stream(&stream).await;
if new_stream.is_err() {
log::error!(
"[{}]Fetch real stream failed: {}",
@@ -357,6 +349,13 @@ impl BiliRecorder {
let new_stream = new_stream.unwrap();
*self.live_stream.write().await = Some(new_stream);
*self.last_update.write().await = Utc::now().timestamp();
log::info!(
"[{}]Update to a new stream: {:?} => {}",
self.room_id,
self.live_stream.read().await.clone(),
stream
);
}
true
@@ -463,6 +462,10 @@ impl BiliRecorder {
}
Err(e) => {
log::error!("Failed fetching index content from {}", stream.index());
log::error!(
"Master manifest: {}",
self.master_manifest.read().await.as_ref().unwrap()
);
Err(super::errors::RecorderError::BiliClientError { err: e })
}
}
@@ -504,7 +507,7 @@ impl BiliRecorder {
async fn fetch_real_stream(
&self,
stream: BiliStream,
stream: &BiliStream,
) -> Result<BiliStream, super::errors::RecorderError> {
let index_content = self
.client
@@ -513,7 +516,9 @@ impl BiliRecorder {
.get_index_content(&self.account, &stream.index())
.await?;
if index_content.is_empty() {
return Err(super::errors::RecorderError::InvalidStream { stream });
return Err(super::errors::RecorderError::InvalidStream {
stream: stream.clone(),
});
}
if index_content.contains("Not Found") {
return Err(super::errors::RecorderError::IndexNotFound {
@@ -538,9 +543,9 @@ impl BiliRecorder {
+ "/";
let new_stream = BiliStream::new(StreamType::FMP4, base_url.as_str(), host, extra);
return Box::pin(self.fetch_real_stream(new_stream)).await;
return Box::pin(self.fetch_real_stream(&new_stream)).await;
}
Ok(stream)
Ok(stream.clone())
}
async fn get_work_dir(&self, live_id: &str) -> String {
@@ -560,6 +565,13 @@ impl BiliRecorder {
}
let current_stream = current_stream.unwrap();
let parsed = self.get_playlist().await;
if parsed.is_err() {
self.force_update.store(true, Ordering::Relaxed);
return Err(parsed.err().unwrap());
}
let playlist = parsed.unwrap();
let mut timestamp: i64 = self.live_id.read().await.parse::<i64>().unwrap_or(0);
let mut work_dir;
let mut is_first_record = false;
@@ -604,9 +616,6 @@ impl BiliRecorder {
return Err(super::errors::RecorderError::HeaderChanged);
}
// Save the new header URL
*self.current_header_url.write().await = Some(header_url.clone());
timestamp = Utc::now().timestamp_millis();
*self.live_id.write().await = timestamp.to_string();
work_dir = self.get_work_dir(timestamp.to_string().as_str()).await;
@@ -625,9 +634,9 @@ impl BiliRecorder {
};
// Create work directory before download
tokio::fs::create_dir_all(&work_dir).await.map_err(|e| {
super::errors::RecorderError::IoError { err: e }
})?;
tokio::fs::create_dir_all(&work_dir)
.await
.map_err(|e| super::errors::RecorderError::IoError { err: e })?;
// Download header
match self
@@ -642,7 +651,11 @@ impl BiliRecorder {
log::error!("Download header failed: {}", full_header_url);
// Clean up empty directory since header download failed
if let Err(cleanup_err) = tokio::fs::remove_dir_all(&work_dir).await {
log::warn!("Failed to cleanup empty work directory {}: {}", work_dir, cleanup_err);
log::warn!(
"Failed to cleanup empty work directory {}: {}",
work_dir,
cleanup_err
);
}
return Err(super::errors::RecorderError::InvalidStream {
stream: current_stream,
@@ -667,7 +680,8 @@ impl BiliRecorder {
// danmu file
let danmu_file_path = format!("{}{}", work_dir, "danmu.txt");
*self.danmu_storage.write().await = DanmuStorage::new(&danmu_file_path).await;
*self.danmu_storage.write().await =
DanmuStorage::new(&danmu_file_path).await;
self.entry_store
.write()
@@ -676,12 +690,19 @@ impl BiliRecorder {
.unwrap()
.add_entry(header)
.await;
// Save the new header URL
*self.current_header_url.write().await = Some(header_url.clone());
}
Err(e) => {
log::error!("Download header failed: {}", e);
// Clean up empty directory since header download failed
if let Err(cleanup_err) = tokio::fs::remove_dir_all(&work_dir).await {
log::warn!("Failed to cleanup empty work directory {}: {}", work_dir, cleanup_err);
log::warn!(
"Failed to cleanup empty work directory {}: {}",
work_dir,
cleanup_err
);
}
return Err(e.into());
}
@@ -701,9 +722,9 @@ impl BiliRecorder {
}
}
match parsed {
Ok(Playlist::MasterPlaylist(pl)) => log::debug!("Master playlist:\n{:?}", pl),
Ok(Playlist::MediaPlaylist(pl)) => {
match playlist {
Playlist::MasterPlaylist(pl) => log::debug!("Master playlist:\n{:?}", pl),
Playlist::MediaPlaylist(pl) => {
let mut new_segment_fetched = false;
let last_sequence = self
.entry_store
@@ -736,9 +757,19 @@ impl BiliRecorder {
// Extract stream start timestamp from header if available for FMP4
let stream_start_timestamp = if current_stream.format == StreamType::FMP4 {
if let Some(header_entry) = self.entry_store.read().await.as_ref().and_then(|store| store.get_header()) {
if let Some(header_entry) = self
.entry_store
.read()
.await
.as_ref()
.and_then(|store| store.get_header())
{
// Parse timestamp from header filename like "h1753276580.m4s"
if let Some(timestamp_str) = header_entry.url.strip_prefix("h").and_then(|s| s.strip_suffix(".m4s")) {
if let Some(timestamp_str) = header_entry
.url
.strip_prefix("h")
.and_then(|s| s.strip_suffix(".m4s"))
{
timestamp_str.parse::<i64>().unwrap_or(0)
} else {
0
@@ -766,7 +797,10 @@ impl BiliRecorder {
}
// Calculate precise timestamp from stream start + BILI-AUX offset for FMP4
let ts_mili = if current_stream.format == StreamType::FMP4 && stream_start_timestamp > 0 && i < segment_offsets.len() {
let ts_mili = if current_stream.format == StreamType::FMP4
&& stream_start_timestamp > 0
&& i < segment_offsets.len()
{
let seg_offset = segment_offsets[i];
stream_start_timestamp * 1000 + seg_offset
@@ -780,7 +814,8 @@ impl BiliRecorder {
let ts_length = pl.target_duration as f64;
// Calculate precise duration from BILI-AUX offsets for FMP4
let precise_length_from_aux = if current_stream.format == StreamType::FMP4 && i < segment_offsets.len() {
let precise_length_from_aux =
if current_stream.format == StreamType::FMP4 && i < segment_offsets.len() {
let current_offset = segment_offsets[i];
// Get the previous offset for duration calculation
@@ -813,9 +848,9 @@ impl BiliRecorder {
// For non-FMP4 streams, create record on first successful ts download
if is_first_record && current_stream.format != StreamType::FMP4 {
// Create work directory before first ts download
tokio::fs::create_dir_all(&work_dir).await.map_err(|e| {
super::errors::RecorderError::IoError { err: e }
})?;
tokio::fs::create_dir_all(&work_dir)
.await
.map_err(|e| super::errors::RecorderError::IoError { err: e })?;
work_dir_created_for_non_fmp4 = true;
}
@@ -824,9 +859,17 @@ impl BiliRecorder {
log::error!("Download ts failed after retry");
// Clean up empty directory if first ts download failed for non-FMP4
if is_first_record && current_stream.format != StreamType::FMP4 && work_dir_created_for_non_fmp4 {
if let Err(cleanup_err) = tokio::fs::remove_dir_all(&work_dir).await {
log::warn!("Failed to cleanup empty work directory {}: {}", work_dir, cleanup_err);
if is_first_record
&& current_stream.format != StreamType::FMP4
&& work_dir_created_for_non_fmp4
{
if let Err(cleanup_err) = tokio::fs::remove_dir_all(&work_dir).await
{
log::warn!(
"Failed to cleanup empty work directory {}: {}",
work_dir,
cleanup_err
);
}
}
@@ -843,9 +886,18 @@ impl BiliRecorder {
log::error!("Segment with size 0, stream might be corrupted");
// Clean up empty directory if first ts download failed for non-FMP4
if is_first_record && current_stream.format != StreamType::FMP4 && work_dir_created_for_non_fmp4 {
if let Err(cleanup_err) = tokio::fs::remove_dir_all(&work_dir).await {
log::warn!("Failed to cleanup empty work directory {}: {}", work_dir, cleanup_err);
if is_first_record
&& current_stream.format != StreamType::FMP4
&& work_dir_created_for_non_fmp4
{
if let Err(cleanup_err) =
tokio::fs::remove_dir_all(&work_dir).await
{
log::warn!(
"Failed to cleanup empty work directory {}: {}",
work_dir,
cleanup_err
);
}
}
@@ -872,20 +924,31 @@ impl BiliRecorder {
// danmu file
let danmu_file_path = format!("{}{}", work_dir, "danmu.txt");
*self.danmu_storage.write().await = DanmuStorage::new(&danmu_file_path).await;
*self.danmu_storage.write().await =
DanmuStorage::new(&danmu_file_path).await;
is_first_record = false;
}
// Get precise duration - prioritize BILI-AUX for FMP4, fallback to ffprobe if needed
let precise_length = if let Some(aux_duration) = precise_length_from_aux {
let precise_length = if let Some(aux_duration) =
precise_length_from_aux
{
aux_duration
} else if current_stream.format != StreamType::FMP4 {
// For regular TS segments, use direct ffprobe
let file_path = format!("{}/{}", work_dir, file_name);
match crate::ffmpeg::get_segment_duration(std::path::Path::new(&file_path)).await {
match crate::ffmpeg::get_segment_duration(std::path::Path::new(
&file_path,
))
.await
{
Ok(duration) => {
log::debug!("Precise TS segment duration: {}s (original: {}s)", duration, ts_length);
log::debug!(
"Precise TS segment duration: {}s (original: {}s)",
duration,
ts_length
);
duration
}
Err(e) => {
@@ -915,7 +978,9 @@ impl BiliRecorder {
.await;
// Update last offset for next segment calculation
if current_stream.format == StreamType::FMP4 && i < segment_offsets.len() {
if current_stream.format == StreamType::FMP4
&& i < segment_offsets.len()
{
last_offset = Some(segment_offsets[i]);
}
@@ -927,9 +992,19 @@ impl BiliRecorder {
log::warn!("Download ts failed, retry {}: {}", retry, e);
// If this is the last retry and it's the first record for non-FMP4, clean up
if retry > 3 && is_first_record && current_stream.format != StreamType::FMP4 && work_dir_created_for_non_fmp4 {
if let Err(cleanup_err) = tokio::fs::remove_dir_all(&work_dir).await {
log::warn!("Failed to cleanup empty work directory {}: {}", work_dir, cleanup_err);
if retry > 3
&& is_first_record
&& current_stream.format != StreamType::FMP4
&& work_dir_created_for_non_fmp4
{
if let Err(cleanup_err) =
tokio::fs::remove_dir_all(&work_dir).await
{
log::warn!(
"Failed to cleanup empty work directory {}: {}",
work_dir,
cleanup_err
);
}
}
}
@@ -978,10 +1053,6 @@ impl BiliRecorder {
}
}
}
Err(e) => {
self.force_update.store(true, Ordering::Relaxed);
return Err(e);
}
}
// check stream is nearly expired
@@ -1031,11 +1102,7 @@ impl BiliRecorder {
};
if let Some(entry_store) = self.entry_store.read().await.as_ref() {
entry_store.manifest(
!live_status || range.is_some(),
true,
range,
)
entry_store.manifest(!live_status || range.is_some(), true, range)
} else {
// Return empty manifest if entry_store is not initialized yet
"#EXTM3U\n#EXT-X-VERSION:3\n".to_string()
@@ -1083,7 +1150,9 @@ impl super::Recorder for BiliRecorder {
std::cmp::min(5, connection_fail_count + 1);
} else if let RecorderError::HeaderChanged = e {
// Mark that exit was triggered by header change
self_clone.header_changed_recently.store(true, Ordering::Relaxed);
self_clone
.header_changed_recently
.store(true, Ordering::Relaxed);
}
break;
}
@@ -1092,9 +1161,12 @@ impl super::Recorder for BiliRecorder {
*self_clone.is_recording.write().await = false;
// Check if exit was triggered by header change for faster recovery
let sleep_duration = if self_clone.header_changed_recently.load(Ordering::Relaxed) {
let sleep_duration =
if self_clone.header_changed_recently.load(Ordering::Relaxed) {
// Clear the flag after checking
self_clone.header_changed_recently.store(false, Ordering::Relaxed);
self_clone
.header_changed_recently
.store(false, Ordering::Relaxed);
// Quick recovery for header change - only 500ms
Duration::from_millis(500)
} else {
@@ -1196,7 +1268,11 @@ impl super::Recorder for BiliRecorder {
Ok(if live_id == *self.live_id.read().await {
// just return current cache content
match self.danmu_storage.read().await.as_ref() {
Some(storage) => storage.get_entries(self.first_segment_ts(live_id).await).await,
Some(storage) => {
storage
.get_entries(self.first_segment_ts(live_id).await)
.await
}
None => Vec::new(),
}
} else {
@@ -1214,7 +1290,9 @@ impl super::Recorder for BiliRecorder {
return Ok(Vec::new());
}
let storage = storage.unwrap();
storage.get_entries(self.first_segment_ts(live_id).await).await
storage
.get_entries(self.first_segment_ts(live_id).await)
.await
})
}
@@ -1222,7 +1300,10 @@ impl super::Recorder for BiliRecorder {
*self.live_id.read().await == live_id && *self.live_status.read().await
}
async fn get_archive_subtitle(&self, live_id: &str) -> Result<String, super::errors::RecorderError> {
async fn get_archive_subtitle(
&self,
live_id: &str,
) -> Result<String, super::errors::RecorderError> {
// read subtitle file under work_dir
let work_dir = self.get_work_dir(live_id).await;
let subtitle_file_path = format!("{}/{}", work_dir, "subtitle.srt");
@@ -1239,7 +1320,10 @@ impl super::Recorder for BiliRecorder {
Ok(subtitle_content)
}
async fn generate_archive_subtitle(&self, live_id: &str) -> Result<String, super::errors::RecorderError> {
async fn generate_archive_subtitle(
&self,
live_id: &str,
) -> Result<String, super::errors::RecorderError> {
// generate subtitle file under work_dir
let work_dir = self.get_work_dir(live_id).await;
let subtitle_file_path = format!("{}/{}", work_dir, "subtitle.srt");
@@ -1252,7 +1336,13 @@ impl super::Recorder for BiliRecorder {
log::info!("M3U8 index file generated: {}", m3u8_index_file_path);
// generate a tmp clip file
let clip_file_path = format!("{}/{}", work_dir, "tmp.mp4");
if let Err(e) = crate::ffmpeg::clip_from_m3u8(None::<&crate::progress_reporter::ProgressReporter>, Path::new(&m3u8_index_file_path), Path::new(&clip_file_path)).await {
if let Err(e) = crate::ffmpeg::clip_from_m3u8(
None::<&crate::progress_reporter::ProgressReporter>,
Path::new(&m3u8_index_file_path),
Path::new(&clip_file_path),
)
.await
{
return Err(super::errors::RecorderError::SubtitleGenerationFailed {
error: e.to_string(),
});
@@ -1260,7 +1350,17 @@ impl super::Recorder for BiliRecorder {
log::info!("Temp clip file generated: {}", clip_file_path);
// generate subtitle file
let config = self.config.read().await;
let result = crate::ffmpeg::generate_video_subtitle(None, Path::new(&clip_file_path), "whisper", &config.whisper_model, &config.whisper_prompt, &config.openai_api_key, &config.openai_api_endpoint, &config.whisper_language).await;
let result = crate::ffmpeg::generate_video_subtitle(
None,
Path::new(&clip_file_path),
"whisper",
&config.whisper_model,
&config.whisper_prompt,
&config.openai_api_key,
&config.openai_api_endpoint,
&config.whisper_language,
)
.await;
// write subtitle file
if let Err(e) = result {
return Err(super::errors::RecorderError::SubtitleGenerationFailed {
@@ -1269,7 +1369,12 @@ impl super::Recorder for BiliRecorder {
}
log::info!("Subtitle generated");
let result = result.unwrap();
let subtitle_content = result.subtitle_content.iter().map(item_to_srt).collect::<Vec<String>>().join("");
let subtitle_content = result
.subtitle_content
.iter()
.map(item_to_srt)
.collect::<Vec<String>>()
.join("");
subtitle_file.write_all(subtitle_content.as_bytes()).await?;
log::info!("Subtitle file written");
// remove tmp file

13
src-tauri/src/recorder/bilibili/client.rs
View File

@@ -138,19 +138,6 @@ impl BiliStream {
}
})
}
pub fn is_same(&self, other: &BiliStream) -> bool {
// Extract live_id part from path (e.g., live_1848752274_71463808)
let get_live_id = |path: &str| {
path.split('/')
.find(|part| part.starts_with("live_"))
.unwrap_or("")
.to_string()
};
let self_live_id = get_live_id(&self.path);
let other_live_id = get_live_id(&other.path);
self_live_id == other_live_id
}
}
impl BiliClient {

7
src-tauri/src/recorder/entry.rs
View File

@@ -212,7 +212,6 @@ impl EntryStore {
/// `vod` indicates the manifest is for stream or video.
/// `force_time` adds DATE-TIME tag for each entry.
pub fn manifest(&self, vod: bool, force_time: bool, range: Option<Range>) -> String {
log::debug!("Generate manifest for range: {:?} with vod: {} and force_time: {}", range, vod, force_time);
let mut m3u8_content = "#EXTM3U\n".to_string();
m3u8_content += "#EXT-X-VERSION:6\n";
m3u8_content += if vod {
@@ -240,12 +239,6 @@ impl EntryStore {
// Collect entries in range
let first_entry = self.entries.first().unwrap();
let first_entry_ts = first_entry.ts_seconds();
log::debug!("First entry ts: {}", first_entry_ts);
let last_entry = self.entries.last().unwrap();
let last_entry_ts = last_entry.ts_seconds();
log::debug!("Last entry ts: {}", last_entry_ts);
log::debug!("Full length: {}", last_entry_ts - first_entry_ts);
log::debug!("Range: {:?}", range);
let mut entries_in_range = vec![];
for e in &self.entries {
// ignore header, cause it's already in EXT-X-MAP