[bilibili.playback] Add support for uploading playback videos

Author: SpriteOvO

Cargo.lock

@@ -10,8 +10,10 @@ bytes = "1.7.2"
 chrono = "0.4.39"
 clap = { version = "4.5.19", features = ["derive"] }
 const_format = "0.2.34"
-futures = "0.3.31"
 headless_chrome = "1.0.15"
+http = "1.3.1"
+http-serde = "2.1.1"
+humansize = "2.1.3"
 humantime-serde = "1.1.1"
 image = "0.25.2"
 itertools = "0.13.0"
@@ -23,8 +25,10 @@ serde = { version = "1.0.210", features = ["derive", "rc"] }
 serde_json = "1.0.128"
 shadow-rs = "0.38.0"
 spdlog-rs = { version = "0.4.0", features = ["source-location"] }
-tokio = { version = "1.40.0", features = ["rt-multi-thread", "macros", "fs", "time", "sync"] }
+tempfile = "3.19.1"
+tokio = { version = "1.40.0", features = ["rt-multi-thread", "macros", "fs", "time", "sync", "process"] }
 toml = "0.8.19"
+warp = "0.3.7"
 
 [build-dependencies]
 shadow-rs = "0.38.0"

Cargo.toml

@@ -147,6 +147,8 @@ pub struct PlatformGlobal {
     pub telegram: Accessor<Option<notify::platform::telegram::ConfigGlobal>>,
     #[serde(rename = "Twitter")]
     pub twitter: Accessor<Option<source::platform::twitter::ConfigGlobal>>,
+    #[serde(rename = "bilibili")]
+    pub bilibili: Accessor<Option<source::platform::bilibili::ConfigGlobal>>,
 }
 
 impl Validator for PlatformGlobal {
@@ -155,6 +157,7 @@ impl Validator for PlatformGlobal {
         self.qq.validate()?;
         self.telegram.validate()?;
         self.twitter.validate()?;
+        self.bilibili.validate()?;
         Ok(())
     }
 }
@@ -185,6 +188,10 @@ pub struct Notifications {
     pub post: bool,
     #[serde(default = "helper::refl_bool::<true>")]
     pub log: bool,
+    #[serde(default = "helper::refl_bool::<true>")]
+    pub playback: bool,
+    #[serde(default = "helper::refl_bool::<true>")]
+    pub document: bool,
 }
 
 serde_impl_default_for!(Notifications);
@@ -201,6 +208,8 @@ impl Overridable for Notifications {
             live_title: new.live_title.unwrap_or(self.live_title),
             post: new.post.unwrap_or(self.post),
             log: new.log.unwrap_or(self.log),
+            playback: new.playback.unwrap_or(self.playback),
+            document: new.document.unwrap_or(self.document),
         }
     }
 }
@@ -211,6 +220,8 @@ pub struct NotificationsOverride {
     pub live_title: Option<bool>,
     pub post: Option<bool>,
     pub log: Option<bool>,
+    pub playback: Option<bool>,
+    pub document: Option<bool>,
 }
 
 #[derive(Debug, PartialEq, Deserialize)]
@@ -288,6 +299,9 @@ token = "ttt"
 [platform.Twitter]
 auth = { cookies = "a=b;c=d;ct0=blah" }
 
+[platform.bilibili]
+playback = { bililive_recorder = { listen_webhook = { host = "127.0.0.1", port = 8888 }, working_directory = "/brec/" } }
+
 [notify]
 meow = { platform = "Telegram", id = 1234, thread_id = 123, token = "xxx" }
 woof = { platform = "Telegram", id = 5678, thread_id = 900, notifications = { post = false } }
@@ -330,11 +344,23 @@ notify = ["meow", "woof", { ref = "woof", id = 123 }]
                     })),
                     telegram: Accessor::new(Some(notify::platform::telegram::ConfigGlobal {
                         token: Some(notify::platform::telegram::ConfigToken::with_raw("ttt")),
+                        api_server: None,
                         experimental: Default::default()
                     })),
                     twitter: Accessor::new(Some(source::platform::twitter::ConfigGlobal {
                         auth: source::platform::twitter::ConfigCookies::with_raw("a=b;c=d;ct0=blah")
-                    }))
+                    })),
+                    bilibili: Accessor::new(Some(source::platform::bilibili::ConfigGlobal {
+                        playback: Accessor::new(Some(source::platform::bilibili::playback::ConfigGlobal {
+                            bililive_recorder: Accessor::new(source::platform::bilibili::playback::bililive_recorder::ConfigBililiveRecorder {
+                                listen_webhook: source::platform::bilibili::playback::bililive_recorder::ConfigListen {
+                                    host: "127.0.0.1".into(),
+                                    port: 8888
+                                },
+                                working_directory: "/brec/".into()
+                            })
+                        }))
+                    })),
                 }),
                 notify_map: Accessor::new(NotifyMap(HashMap::from_iter([
                     (
@@ -354,6 +380,8 @@ notify = ["meow", "woof", { ref = "woof", id = 123 }]
                                 live_title: false,
                                 post: false,
                                 log: true,
+                                playback: true,
+                                document: true
                             },
                             chat: notify::platform::telegram::ConfigChat::Id(5678),
                             thread_id: Some(900),

src/config/mod.rs

@@ -1,8 +1,9 @@
-use std::{convert::identity, time::Duration};
+use std::{convert::identity, path::Path, time::Duration};
 
-use anyhow::anyhow;
+use anyhow::{anyhow, ensure};
 use humantime_serde::re::humantime;
 use reqwest::header::{self, HeaderMap, HeaderValue};
+use tokio::process::Command;
 
 use crate::prop;
 
@@ -65,6 +66,54 @@ pub fn format_duration_in_min(dur: Duration) -> String {
     humantime::format_duration(Duration::from_secs(mins * 60)).to_string()
 }
 
+pub async fn ffmpeg_copy(from: &Path, to: &Path) -> anyhow::Result<()> {
+    let mut cmd = Command::new("ffmpeg");
+    cmd.arg("-i").arg(from).arg("-c").arg("copy").arg(to);
+
+    let output = cmd
+        .output()
+        .await
+        .map_err(|err| anyhow!("failed to run ffmpeg-copy: {err}"))?;
+    ensure!(
+        output.status.success(),
+        "ffmpeg-copy failed with status: {}. stderr: {}",
+        output.status,
+        String::from_utf8_lossy(&output.stderr)
+    );
+    Ok(())
+}
+
+pub async fn ffprob_resolution(video: &Path) -> anyhow::Result<(u32, u32)> {
+    let mut cmd = Command::new("ffprobe");
+    cmd.arg("-v")
+        .arg("error")
+        .arg("-select_streams")
+        .arg("v:0")
+        .arg("-show_entries")
+        .arg("stream=width,height")
+        .arg("-of")
+        .arg("csv=p=0:s=x")
+        .arg(video);
+
+    let output = cmd
+        .output()
+        .await
+        .map_err(|err| anyhow!("failed to run ffprobe: {err}"))?;
+    ensure!(
+        output.status.success(),
+        "ffprobe failed with status: {}. stderr: {}",
+        output.status,
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let resolution = String::from_utf8_lossy(&output.stdout);
+    let (w, h) = resolution
+        .trim()
+        .split_once('x')
+        .and_then(|(w, h)| w.parse().ok().zip(h.parse().ok()))
+        .ok_or_else(|| anyhow!("failed to parse ffprobe resolution '{resolution}'"))?;
+    Ok((w, h))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/helper.rs

@@ -141,6 +141,8 @@ impl Notifier {
             }
             NotificationKind::Posts(posts) => self.notify_posts(posts, notification.source).await,
             NotificationKind::Log(message) => self.notify_log(message).await,
+            NotificationKind::Playback(_) => unimplemented!(),
+            NotificationKind::Document(_) => unimplemented!(),
         }
     }
 

src/notify/platform/qq/mod.rs

@@ -3,6 +3,7 @@ mod request;
 use std::{borrow::Cow, collections::VecDeque, fmt, future::Future, pin::Pin, time::SystemTime};
 
 use anyhow::{anyhow, bail, ensure};
+use http::Uri;
 use request::*;
 use serde::Deserialize;
 use serde_json as json;
@@ -16,8 +17,9 @@ use crate::{
     platform::{PlatformMetadata, PlatformTrait},
     secret_enum, serde_impl_default_for,
     source::{
-        LiveStatus, LiveStatusKind, Notification, NotificationKind, Post, PostAttachment, PostUrl,
-        PostsRef, RepostFrom, StatusSource,
+        DocumentRef, FileRef, LiveStatus, LiveStatusKind, Notification, NotificationKind,
+        PlaybackFormat, PlaybackRef, Post, PostAttachment, PostUrl, PostsRef, RepostFrom,
+        StatusSource,
     },
 };
 
@@ -26,6 +28,8 @@ pub struct ConfigGlobal {
     #[serde(flatten)]
     pub token: Option<ConfigToken>,
     #[serde(default)]
+    pub api_server: Option<ConfigApiServer>,
+    #[serde(default)]
     pub experimental: ConfigExperimental,
 }
 
@@ -42,6 +46,17 @@ impl config::Validator for ConfigGlobal {
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Deserialize)]
+#[serde(untagged)]
+pub enum ConfigApiServer {
+    Url(#[serde(with = "http_serde::uri")] Uri),
+    UrlOpts {
+        #[serde(with = "http_serde::uri")]
+        url: Uri,
+        as_necessary: bool,
+    },
+}
+
 #[derive(Clone, Debug, PartialEq, Deserialize)]
 pub struct ConfigExperimental {
     #[deprecated = "enabled by default"]
@@ -213,6 +228,12 @@ impl Notifier {
             }
             NotificationKind::Posts(posts) => self.notify_posts(posts, notification.source).await,
             NotificationKind::Log(message) => self.notify_log(message).await,
+            NotificationKind::Playback(playback) => {
+                self.notify_playback(playback, notification.source).await
+            }
+            NotificationKind::Document(document) => {
+                self.notify_document(document, notification.source).await
+            }
         }
     }
 
@@ -566,6 +587,110 @@ impl Notifier {
 
         Ok(())
     }
+
+    // TODO: Parallel notify
+    async fn notify_playback(
+        &self,
+        playback: &PlaybackRef<'_>,
+        source: &StatusSource,
+    ) -> anyhow::Result<()> {
+        if !self.params.notifications.playback {
+            info!("playback notification is disabled, skip notifying");
+            return Ok(());
+        }
+
+        const FORMAT: PlaybackFormat = PlaybackFormat::Mp4;
+
+        let file = playback.get(FORMAT).await?;
+
+        let token = self.token()?;
+
+        // Send "uploading" message
+
+        let resp = Request::new(&token)
+            .send_message(&self.params.chat, make_file_text("⏳", &file, source))
+            .thread_id_opt(self.params.thread_id)
+            .link_preview(LinkPreview::Disabled)
+            // .disable_notification() // TODO: Make it configurable
+            .send()
+            .await
+            .map_err(|err| anyhow!("failed to send request to Telegram: {err}"))?;
+        ensure!(
+            resp.ok,
+            "response contains error, description '{}'",
+            resp.description
+                .unwrap_or_else(|| "*no description*".into())
+        );
+
+        // Edit the media
+
+        trace!("uploading playback to Telegram '{file}'");
+
+        let resp = Request::new(&token)
+            .edit_message_media(
+                &self.params.chat,
+                resp.result.unwrap().message_id,
+                Media::Video(MediaVideo {
+                    input: MediaInput::Memory {
+                        data: file.data.clone(),
+                        filename: Some(&file.name),
+                    },
+                    resolution: Some(file.resolution),
+                    has_spoiler: false,
+                }),
+            )
+            .text(make_file_text("🎥", &file, source))
+            .prefer_self_host()
+            .send()
+            .await
+            .map_err(|err| anyhow!("failed to send request to Telegram: {err}"))?;
+        ensure!(
+            resp.ok,
+            "response contains error, description '{}'",
+            resp.description
+                .unwrap_or_else(|| "*no description*".into())
+        );
+
+        Ok(())
+    }
+
+    async fn notify_document(
+        &self,
+        document: &DocumentRef<'_>,
+        source: &StatusSource,
+    ) -> anyhow::Result<()> {
+        if !self.params.notifications.document {
+            info!("document notification is disabled, skip notifying");
+            return Ok(());
+        }
+
+        let token = self.token()?;
+
+        let resp = Request::new(&token)
+            .send_document(
+                &self.params.chat,
+                MediaDocument {
+                    input: MediaInput::Memory {
+                        data: document.file.data.clone(),
+                        filename: Some(&document.file.name),
+                    },
+                },
+            )
+            .text(make_file_text("📊", &document.file, source))
+            .thread_id_opt(self.params.thread_id)
+            // .disable_notification() // TODO: Make it configurable
+            .send()
+            .await
+            .map_err(|err| anyhow!("failed to send request to Telegram: {err}"))?;
+        ensure!(
+            resp.ok,
+            "response contains error, description '{}'",
+            resp.description
+                .unwrap_or_else(|| "*no description*".into())
+        );
+
+        Ok(())
+    }
 }
 
 fn make_live_text<'a>(
@@ -597,6 +722,15 @@ fn make_live_text<'a>(
     Text::link(text, &live_status.live_url)
 }
 
+fn make_file_text<'a>(emoji: &'a str, file: &FileRef<'a>, source: &'a StatusSource) -> Text<'a> {
+    Text::plain(format!(
+        "[{}] {emoji} {} ({})",
+        source.platform.display_name,
+        file.name,
+        humansize::format_size(file.size, humansize::BINARY)
+    ))
+}
+
 struct CurrentLive {
     start_time: SystemTime,
     message_id: i64,