bsky: remux HLS video playlists to MP4 via ffmpeg

bsky video embeds expose only an m3u8 playlist, which Telegram cannot
fetch. Download the master/variant playlists and TS segments through the
shared client (proxy-aware, size-capped), then concat-remux locally;
keep the temp dir alive via Fetched._keep_alive like the ugoira path.
Also adds site::download_media_limited (streaming size cap), status
checks on media_size, and moves the ffmpeg probe to site/mod.rs for
pixiv/bsky to share.
This commit is contained in:
2026-08-08 19:56:36 +08:00
parent 9a96f78177
commit 7998114dc3
3 changed files with 214 additions and 29 deletions
+155 -1
View File
@@ -22,7 +22,161 @@ pub async fn fetch_from_url(url: &str) -> Result<Fetched, FetchError> {
.get(2) .get(2)
.map(|m| m.as_str()) .map(|m| m.as_str())
.ok_or(FetchError::NotFound)?; .ok_or(FetchError::NotFound)?;
Ok(fetch(handle, rkey).await?.into()) let post = fetch(handle, rkey).await?;
let mut fetched: Fetched = post.into();
// bsky video embeds expose only an HLS playlist URL, which Telegram
// cannot fetch; remux it to a single MP4 (mirrors the pixiv ugoira
// encode path — the temp file stays alive via `_keep_alive`). On any
// failure the video item is dropped and the post degrades to its text.
let mut media = Vec::with_capacity(fetched.media.len());
for item in fetched.media {
let is_hls = matches!(&item, Media::Video { url, .. }
if url.contains("playlist") || url.ends_with(".m3u8"));
if !is_hls {
media.push(item);
continue;
}
let url = item.url().to_string();
match resolve_bsky_video(&url).await {
Ok(Some((mp4_path, keep_alive))) => {
let thumbnail_url = match &item {
Media::Video { thumbnail_url, .. } => thumbnail_url.clone(),
_ => String::new(),
};
media.push(Media::Video {
title: None,
url: mp4_path.to_string_lossy().into_owned(),
thumbnail_url,
});
fetched._keep_alive = Some(keep_alive);
}
Ok(None) => log::warn!("bsky video remux unavailable for {url}"),
Err(e) => log::warn!("bsky video remux failed for {url}: {e}"),
}
}
fetched.media = media;
Ok(fetched)
}
/// Downloads an HLS playlist (master or media) and remuxes its segments to a
/// single MP4 via ffmpeg. Returns the MP4 path plus the temp dir that must
/// stay alive until the file is uploaded. `Ok(None)` when ffmpeg is missing.
///
/// Verified live (2026-08): bsky master playlists carry `#EXT-X-STREAM-INF`
/// variant lines (e.g. `720p/video.m3u8?session_id=…`), and the media
/// playlists are VOD MPEG-TS segments (`videoN.ts?…`) without EXT-X-MAP, so
/// a plain `-f concat -c copy` remux is valid.
async fn resolve_bsky_video(
playlist_url: &str,
) -> Result<Option<(std::path::PathBuf, tempfile::TempDir)>, String> {
if !crate::site::ffmpeg_available() {
crate::site::log_once_ffmpeg_missing();
return Ok(None);
}
let master = crate::site::download_media_limited(playlist_url, 1_048_576)
.await
.map_err(|e| format!("bsky video master playlist: {e}"))?;
let master = String::from_utf8_lossy(&master);
// Master playlist: pick the variant with the highest declared bandwidth.
let playlist_url = if master.contains("#EXT-X-STREAM-INF") {
let mut best: Option<(u64, String)> = None;
let mut lines = master.lines();
while let Some(line) = lines.next() {
if !line.starts_with("#EXT-X-STREAM-INF") {
continue;
}
let bandwidth = line
.split_once("BANDWIDTH=")
.and_then(|(_, rest)| rest.split(|c: char| !c.is_ascii_digit()).next())
.and_then(|n| n.parse::<u64>().ok())
.unwrap_or(0);
if let Some(uri) = lines.next().filter(|u| !u.starts_with('#')) {
if bandwidth >= best.as_ref().map(|(b, _)| *b).unwrap_or(0) {
best = Some((bandwidth, uri.to_string()));
}
}
}
let Some((_, uri)) = best else {
return Err("bsky video master playlist has no variants".to_string());
};
url::Url::parse(playlist_url)
.and_then(|base| base.join(&uri))
.map_err(|e| format!("bsky video variant URL: {e}"))?
.to_string()
} else {
playlist_url.to_string()
};
let variant = crate::site::download_media_limited(&playlist_url, 1_048_576)
.await
.map_err(|e| format!("bsky video media playlist: {e}"))?;
let variant = String::from_utf8_lossy(&variant);
// Segment URIs: non-#, non-empty lines, resolved relative to the playlist.
let base = url::Url::parse(&playlist_url).map_err(|e| format!("bsky playlist URL: {e}"))?;
let segments: Vec<String> = variant
.lines()
.map(str::trim)
.filter(|l| !l.is_empty() && !l.starts_with('#'))
.map(|l| base.join(l).map(|u| u.to_string()))
.collect::<Result<_, _>>()
.map_err(|e| format!("bsky segment URL: {e}"))?;
if segments.is_empty() {
return Err("bsky video playlist has no segments".to_string());
}
if segments.len() > 500 {
return Err("bsky video has too many segments".to_string());
}
let frames_dir = tempfile::tempdir().map_err(|e| e.to_string())?;
let out_dir = tempfile::tempdir().map_err(|e| e.to_string())?;
let mut total: u64 = 0;
let mut list = String::new();
for (i, seg) in segments.iter().enumerate() {
let bytes = crate::site::download_media_limited(seg, 20 * 1024 * 1024)
.await
.map_err(|e| format!("bsky segment {i}: {e}"))?;
total += bytes.len() as u64;
if total > 256 * 1024 * 1024 {
return Err("bsky video exceeds total size cap".to_string());
}
let path = frames_dir.path().join(format!("seg_{i:04}.ts"));
std::fs::write(&path, &bytes).map_err(|e| e.to_string())?;
list.push_str(&format!("file '{}'\n", path.to_string_lossy()));
}
let list_path = frames_dir.path().join("list.txt");
std::fs::write(&list_path, &list).map_err(|e| e.to_string())?;
let output = out_dir.path().join("video.mp4");
let list_str = list_path.to_string_lossy().into_owned();
let output_str = output.to_string_lossy().into_owned();
let status = tokio::task::spawn_blocking(move || {
std::process::Command::new("ffmpeg")
.args([
"-y",
"-f",
"concat",
"-safe",
"0",
"-i",
&list_str,
"-c",
"copy",
"-movflags",
"+faststart",
&output_str,
])
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
})
.await
.map_err(|e| format!("bsky remux worker panicked: {e}"))?;
match status {
Ok(s) if s.success() => Ok(Some((output, out_dir))),
Ok(s) => Err(format!("ffmpeg exited with {s}")),
Err(e) => Err(format!("ffmpeg spawn failed: {e}")),
}
} }
/// Fetches a post thread by handle or DID (`at://` URIs work for both). /// Fetches a post thread by handle or DID (`at://` URIs work for both).
+57 -4
View File
@@ -6,6 +6,7 @@
use std::fmt; use std::fmt;
use std::sync::LazyLock; use std::sync::LazyLock;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration; use std::time::Duration;
pub mod bsky; pub mod bsky;
@@ -147,6 +148,8 @@ pub enum FetchError {
/// The post exists but its content is withheld (twitter NSFW / /// The post exists but its content is withheld (twitter NSFW /
/// age-restricted tweets come back as an empty `{}` from syndication). /// age-restricted tweets come back as an empty `{}` from syndication).
Sensitive, Sensitive,
/// A download exceeded the caller's size cap (see [`download_media_limited`]).
TooLarge,
} }
impl fmt::Display for FetchError { impl fmt::Display for FetchError {
@@ -158,6 +161,7 @@ impl fmt::Display for FetchError {
FetchError::NotFound => write!(f, "not found"), FetchError::NotFound => write!(f, "not found"),
FetchError::Blocked => write!(f, "blocked"), FetchError::Blocked => write!(f, "blocked"),
FetchError::Sensitive => write!(f, "content withheld (sensitive)"), FetchError::Sensitive => write!(f, "content withheld (sensitive)"),
FetchError::TooLarge => write!(f, "media too large"),
} }
} }
} }
@@ -169,6 +173,7 @@ impl std::error::Error for FetchError {
FetchError::Json(e) => Some(e), FetchError::Json(e) => Some(e),
FetchError::Pixiv(e) => Some(e), FetchError::Pixiv(e) => Some(e),
FetchError::NotFound | FetchError::Blocked | FetchError::Sensitive => None, FetchError::NotFound | FetchError::Blocked | FetchError::Sensitive => None,
FetchError::TooLarge => None,
} }
} }
} }
@@ -204,6 +209,30 @@ pub(crate) static CLIENT: LazyLock<reqwest::Client> = LazyLock::new(|| {
builder.build().expect("failed to build HTTP client") builder.build().expect("failed to build HTTP client")
}); });
/// Whether a usable `ffmpeg` binary is on PATH (probed once). Shared by the
/// pixiv ugoira encoder and the bsky HLS remuxer.
static FFMPEG_AVAILABLE: LazyLock<bool> = LazyLock::new(|| {
std::process::Command::new("ffmpeg")
.arg("-version")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
});
static FFMPEG_MISSING_LOGGED: AtomicBool = AtomicBool::new(false);
pub(crate) fn ffmpeg_available() -> bool {
*FFMPEG_AVAILABLE
}
pub(crate) fn log_once_ffmpeg_missing() {
if !FFMPEG_MISSING_LOGGED.swap(true, Ordering::Relaxed) {
log::warn!("ffmpeg not found; ugoira and bsky video posts stay unsupported");
}
}
/// Fetches a post from its URL. Returns `Ok(None)` when no site pattern /// Fetches a post from its URL. Returns `Ok(None)` when no site pattern
/// matches (unsupported links are silently ignored by the bot). /// matches (unsupported links are silently ignored by the bot).
/// ///
@@ -262,18 +291,42 @@ pub async fn media_size(url: &str) -> Result<Option<u64>, FetchError> {
if lower.contains("pximg.net") { if lower.contains("pximg.net") {
request = request.header("Referer", "https://www.pixiv.net/"); request = request.header("Referer", "https://www.pixiv.net/");
} }
let response = request.send().await?; let response = request.send().await?.error_for_status()?;
Ok(response.content_length()) Ok(response.content_length())
} }
pub async fn download_media(url: &str) -> Result<bytes::Bytes, FetchError> { /// Downloads a media file with a hard size cap: the body is streamed and the
/// download aborts with [`FetchError::TooLarge`] the moment the cap is
/// crossed (or when a declared Content-Length already exceeds it). Keeps the
/// bot from buffering arbitrarily large bodies into memory.
pub async fn download_media_limited(
url: &str,
max_bytes: u64,
) -> Result<bytes::Bytes, FetchError> {
let mut request = CLIENT.get(url); let mut request = CLIENT.get(url);
let lower = url.to_ascii_lowercase(); let lower = url.to_ascii_lowercase();
if lower.contains("pximg.net") { if lower.contains("pximg.net") {
request = request.header("Referer", "https://www.pixiv.net/"); request = request.header("Referer", "https://www.pixiv.net/");
} }
let response = request.send().await?; let response = request.send().await?.error_for_status()?;
Ok(response.bytes().await?) if let Some(len) = response.content_length()
&& len > max_bytes
{
return Err(FetchError::TooLarge);
}
let mut response = response;
let mut buf = Vec::new();
while let Some(chunk) = response.chunk().await? {
buf.extend_from_slice(&chunk);
if buf.len() as u64 > max_bytes {
return Err(FetchError::TooLarge);
}
}
Ok(bytes::Bytes::from(buf))
}
pub async fn download_media(url: &str) -> Result<bytes::Bytes, FetchError> {
download_media_limited(url, u64::MAX).await
} }
#[cfg(test)] #[cfg(test)]
+2 -24
View File
@@ -207,8 +207,8 @@ impl PixivAPI {
&self, &self,
illust_id: u64, illust_id: u64,
) -> Result<Option<(String, tempfile::TempDir)>, PixivError> { ) -> Result<Option<(String, tempfile::TempDir)>, PixivError> {
if !ffmpeg_available() { if !crate::site::ffmpeg_available() {
log_once_ffmpeg_missing(); crate::site::log_once_ffmpeg_missing();
return Ok(None); return Ok(None);
} }
let metadata = self.ugoira_metadata(illust_id).await?; let metadata = self.ugoira_metadata(illust_id).await?;
@@ -315,28 +315,6 @@ impl PixivAPI {
} }
} }
static FFMPEG_AVAILABLE: LazyLock<bool> = LazyLock::new(|| {
std::process::Command::new("ffmpeg")
.arg("-version")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.map(|s| s.success())
.unwrap_or(false)
});
static FFMPEG_MISSING_LOGGED: AtomicBool = AtomicBool::new(false);
fn ffmpeg_available() -> bool {
*FFMPEG_AVAILABLE
}
fn log_once_ffmpeg_missing() {
if !FFMPEG_MISSING_LOGGED.swap(true, Ordering::Relaxed) {
log::warn!("ffmpeg not found; pixiv ugoira posts stay unsupported");
}
}
/// pixiv3-rs replacement: `None` when `PIXIV_REFRESH_TOKEN` is unset. /// pixiv3-rs replacement: `None` when `PIXIV_REFRESH_TOKEN` is unset.
static PIXIV_CLIENT: LazyLock<Option<PixivAPI>> = static PIXIV_CLIENT: LazyLock<Option<PixivAPI>> =
LazyLock::new(|| env::var("PIXIV_REFRESH_TOKEN").ok().map(PixivAPI::new)); LazyLock::new(|| env::var("PIXIV_REFRESH_TOKEN").ok().map(PixivAPI::new));