feat: 支持使用动态 api 获取投稿，该 api 会返回动态视频 (#485)

crates/bili_sync/src/adapter/collection.rs

@@ -44,7 +44,12 @@ impl VideoSource for collection::Model {
         })
     }
 
-    fn should_take(&self, _release_datetime: &chrono::DateTime<Utc>, _latest_row_at: &chrono::DateTime<Utc>) -> bool {
+    fn should_take(
+        &self,
+        _idx: usize,
+        _release_datetime: &chrono::DateTime<Utc>,
+        _latest_row_at: &chrono::DateTime<Utc>,
+    ) -> bool {
         // collection（视频合集/视频列表）返回的内容似乎并非严格按照时间排序，并且不同 collection 的排序方式也不同
         // 为了保证程序正确性，collection 不根据时间提前 break，而是每次都全量拉取
         true
@@ -52,6 +57,7 @@ impl VideoSource for collection::Model {
 
     fn should_filter(
         &self,
+        _idx: usize,
         video_info: Result<VideoInfo, anyhow::Error>,
         latest_row_at: &chrono::DateTime<Utc>,
     ) -> Option<VideoInfo> {
@@ -61,7 +67,6 @@ impl VideoSource for collection::Model {
         {
             return Some(video_info);
         }
-
         None
     }
 

crates/bili_sync/src/adapter/mod.rs

@@ -56,12 +56,18 @@ pub trait VideoSource {
     fn update_latest_row_at(&self, datetime: DateTime) -> _ActiveModel;
 
     // 判断是否应该继续拉取视频
-    fn should_take(&self, release_datetime: &chrono::DateTime<Utc>, latest_row_at: &chrono::DateTime<Utc>) -> bool {
+    fn should_take(
+        &self,
+        _idx: usize,
+        release_datetime: &chrono::DateTime<Utc>,
+        latest_row_at: &chrono::DateTime<Utc>,
+    ) -> bool {
         release_datetime > latest_row_at
     }
 
     fn should_filter(
         &self,
+        _idx: usize,
         video_info: Result<VideoInfo, anyhow::Error>,
         _latest_row_at: &chrono::DateTime<Utc>,
     ) -> Option<VideoInfo> {

crates/bili_sync/src/adapter/submission.rs

@@ -11,7 +11,7 @@ use sea_orm::sea_query::SimpleExpr;
 use sea_orm::{DatabaseConnection, Unchanged};
 
 use crate::adapter::{_ActiveModel, VideoSource, VideoSourceEnum};
-use crate::bilibili::{BiliClient, Submission, VideoInfo};
+use crate::bilibili::{BiliClient, Dynamic, Submission, VideoInfo};
 
 impl VideoSource for submission::Model {
     fn display_name(&self) -> std::borrow::Cow<'static, str> {
@@ -42,6 +42,42 @@ impl VideoSource for submission::Model {
         })
     }
 
+    fn should_take(
+        &self,
+        idx: usize,
+        release_datetime: &chrono::DateTime<chrono::Utc>,
+        latest_row_at: &chrono::DateTime<chrono::Utc>,
+    ) -> bool {
+        // 如果使用动态 API，那么可能出现用户置顶了一个很久以前的视频在动态顶部的情况
+        // 这种情况应该继续拉取下去，不能因为第一条不满足条件就停止
+        // 后续的非置顶内容是正常由新到旧排序的，可以继续使用常规方式处理
+        if idx == 0 && self.use_dynamic_api {
+            return true;
+        }
+        release_datetime > latest_row_at
+    }
+
+    fn should_filter(
+        &self,
+        idx: usize,
+        video_info: Result<VideoInfo, anyhow::Error>,
+        latest_row_at: &chrono::DateTime<chrono::Utc>,
+    ) -> Option<VideoInfo> {
+        if idx == 0 && self.use_dynamic_api {
+            // 同理，动态 API 的第一条内容可能是置顶的老视频，单独做个过滤
+            // 其实不过滤也不影响逻辑正确性，因为后续 insert 发生冲突仍然会忽略掉
+            // 此处主要是出于性能考虑，减少不必要的数据库操作
+            if let Ok(video_info) = video_info
+                && video_info.release_datetime() > latest_row_at
+            {
+                return Some(video_info);
+            }
+            None
+        } else {
+            video_info.ok()
+        }
+    }
+
     fn rule(&self) -> &Option<Rule> {
         &self.rule
     }
@@ -69,6 +105,12 @@ impl VideoSource for submission::Model {
         }
         .update(connection)
         .await?;
-        Ok((updated_model.into(), Box::pin(submission.into_video_stream())))
+        let video_stream = if self.use_dynamic_api {
+            // 必须显式写出 dyn，否则 rust 会自动推导到 impl 从而认为 if else 返回类型不一致
+            Box::pin(Dynamic::from(submission).into_video_stream()) as Pin<Box<dyn Stream<Item = _> + Send + 'a>>
+        } else {
+            Box::pin(submission.into_video_stream())
+        };
+        Ok((updated_model.into(), video_stream))
     }
 }

crates/bili_sync/src/api/request.rs

@@ -83,9 +83,11 @@ pub struct InsertSubmissionRequest {
 }
 
 #[derive(Deserialize, Validate)]
+#[serde(rename_all = "camelCase")]
 pub struct UpdateVideoSourceRequest {
     #[validate(custom(function = "crate::utils::validation::validate_path"))]
     pub path: String,
     pub enabled: bool,
     pub rule: Option<Rule>,
+    pub use_dynamic_api: Option<bool>,
 }

crates/bili_sync/src/api/response.rs

@@ -179,6 +179,8 @@ pub struct VideoSourceDetail {
     pub rule: Option<Rule>,
     #[serde(default)]
     pub rule_display: Option<String>,
+    #[serde(default)]
+    pub use_dynamic_api: Option<bool>,
     pub enabled: bool,
 }
 

crates/bili_sync/src/api/routes/video_sources/mod.rs

@@ -111,7 +111,8 @@ pub async fn get_video_sources_details(
                 submission::Column::Id,
                 submission::Column::Path,
                 submission::Column::Enabled,
-                submission::Column::Rule
+                submission::Column::Rule,
+                submission::Column::UseDynamicApi
             ])
             .into_model::<VideoSourceDetail>()
             .all(&db),
@@ -134,6 +135,7 @@ pub async fn get_video_sources_details(
             path: String::new(),
             rule: None,
             rule_display: None,
+            use_dynamic_api: None,
             enabled: false,
         })
     }
@@ -179,6 +181,9 @@ pub async fn update_video_source(
             active_model.path = Set(request.path);
             active_model.enabled = Set(request.enabled);
             active_model.rule = Set(request.rule);
+            if let Some(use_dynamic_api) = request.use_dynamic_api {
+                active_model.use_dynamic_api = Set(use_dynamic_api);
+            }
             _ActiveModel::Submission(active_model)
         }),
         "watch_later" => match watch_later::Entity::find_by_id(id).one(&db).await? {

crates/bili_sync/src/bilibili/dynamic.rs

@@ -0,0 +1,88 @@
+use anyhow::{Context, Result, anyhow};
+use async_stream::try_stream;
+use chrono::serde::ts_seconds;
+use chrono::{DateTime, Utc};
+use futures::Stream;
+use reqwest::Method;
+use serde_json::Value;
+
+use crate::bilibili::credential::encoded_query;
+use crate::bilibili::{BiliClient, MIXIN_KEY, Validate, VideoInfo};
+
+pub struct Dynamic<'a> {
+    client: &'a BiliClient,
+    pub upper_id: String,
+}
+
+#[derive(Debug, serde::Deserialize)]
+pub struct DynamicItemPublished {
+    #[serde(with = "ts_seconds")]
+    pub_ts: DateTime<Utc>,
+}
+
+impl<'a> Dynamic<'a> {
+    pub fn new(client: &'a BiliClient, upper_id: String) -> Self {
+        Self { client, upper_id }
+    }
+
+    pub async fn get_dynamics(&self, offset: Option<String>) -> Result<Value> {
+        self.client
+            .request(
+                Method::GET,
+                "https://api.bilibili.com/x/polymer/web-dynamic/v1/feed/all",
+            )
+            .await
+            .query(&encoded_query(
+                vec![
+                    ("host_mid", self.upper_id.as_str()),
+                    ("offset", offset.as_deref().unwrap_or("")),
+                ],
+                MIXIN_KEY.load().as_deref(),
+            ))
+            .send()
+            .await?
+            .error_for_status()?
+            .json::<serde_json::Value>()
+            .await?
+            .validate()
+    }
+
+    pub fn into_video_stream(self) -> impl Stream<Item = Result<VideoInfo>> + 'a {
+        try_stream! {
+            let mut offset = None;
+            loop {
+                let mut res = self
+                    .get_dynamics(offset.take())
+                    .await
+                    .with_context(|| "failed to get dynamics")?;
+                let items = res["data"]["items"].as_array_mut().context("items not exist")?;
+                for item in items.iter_mut() {
+                    if item["type"].as_str().is_none_or(|t| t != "DYNAMIC_TYPE_AV") {
+                        continue;
+                    }
+                    let published: DynamicItemPublished = serde_json::from_value(item["modules"]["module_author"].take())
+                        .with_context(|| "failed to parse published time")?;
+                    let mut video_info: VideoInfo =
+                        serde_json::from_value(item["modules"]["module_dynamic"]["major"]["archive"].take())?;
+                    // 这些地方不使用 let else 是因为 try_stream! 宏不支持
+                    if let VideoInfo::Dynamic { ref mut pubtime, .. } = video_info {
+                        *pubtime = published.pub_ts;
+                        yield video_info;
+                    } else {
+                        Err(anyhow!("video info is not dynamic"))?;
+                    }
+                }
+                if let (Some(has_more), Some(new_offset)) =
+                    (res["data"]["has_more"].as_bool(), res["data"]["offset"].as_str())
+                {
+                    if !has_more {
+                        break;
+                    }
+                    offset = Some(new_offset.to_string());
+                } else {
+                    Err(anyhow!("no has_more or offset found"))?;
+                }
+            }
+        }
+    }
+}

crates/bili_sync/src/bilibili/mod.rs

@@ -9,6 +9,7 @@ pub use client::{BiliClient, Client};
 pub use collection::{Collection, CollectionItem, CollectionType};
 pub use credential::Credential;
 pub use danmaku::DanmakuOption;
+pub use dynamic::Dynamic;
 pub use error::BiliError;
 pub use favorite_list::FavoriteList;
 use favorite_list::Upper;
@@ -23,6 +24,7 @@ mod client;
 mod collection;
 mod credential;
 mod danmaku;
+mod dynamic;
 mod error;
 mod favorite_list;
 mod me;
@@ -135,18 +137,32 @@ pub enum VideoInfo {
         #[serde(rename = "created", with = "ts_seconds")]
         ctime: DateTime<Utc>,
     },
+    // 从动态获取的视频信息（此处 pubtime 未在结构中，因此使用 default + 手动赋值）
+    Dynamic {
+        title: String,
+        bvid: String,
+        desc: String,
+        cover: String,
+        #[serde(default)]
+        pubtime: DateTime<Utc>,
+    },
 }
 
 #[cfg(test)]
 mod tests {
+    use std::path::Path;
+
     use futures::StreamExt;
 
     use super::*;
+    use crate::config::VersionedConfig;
+    use crate::database::setup_database;
     use crate::utils::init_logger;
 
     #[ignore = "only for manual test"]
     #[tokio::test]
-    async fn test_video_info_type() {
+    async fn test_video_info_type() -> Result<()> {
+        VersionedConfig::init_for_test(&setup_database(Path::new("./test.sqlite")).await?).await?;
         init_logger("None,bili_sync=debug", None);
         let bili_client = BiliClient::new();
         // 请求 UP 主视频必须要获取 mixin key，使用 key 计算请求参数的签名，否则直接提示权限不足返回空
@@ -200,6 +216,17 @@ mod tests {
             .await;
         assert!(videos.iter().all(|v| matches!(v, VideoInfo::Submission { .. })));
         assert!(videos.iter().rev().is_sorted_by_key(|v| v.release_datetime()));
+        // 测试动态
+        let dynamic = Dynamic::new(&bili_client, "659898".to_string());
+        let videos = dynamic
+            .into_video_stream()
+            .take(20)
+            .filter_map(|v| futures::future::ready(v.ok()))
+            .collect::<Vec<_>>()
+            .await;
+        assert!(videos.iter().all(|v| matches!(v, VideoInfo::Dynamic { .. })));
+        assert!(videos.iter().skip(1).rev().is_sorted_by_key(|v| v.release_datetime()));
+        Ok(())
     }
 
     #[ignore = "only for manual test"]

crates/bili_sync/src/bilibili/submission.rs

@@ -6,12 +6,18 @@ use serde_json::Value;
 
 use crate::bilibili::credential::encoded_query;
 use crate::bilibili::favorite_list::Upper;
-use crate::bilibili::{BiliClient, MIXIN_KEY, Validate, VideoInfo};
+use crate::bilibili::{BiliClient, Dynamic, MIXIN_KEY, Validate, VideoInfo};
 pub struct Submission<'a> {
     client: &'a BiliClient,
     pub upper_id: String,
 }
 
+impl<'a> From<Submission<'a>> for Dynamic<'a> {
+    fn from(submission: Submission<'a>) -> Self {
+        Dynamic::new(submission.client, submission.upper_id)
+    }
+}
+
 impl<'a> Submission<'a> {
     pub fn new(client: &'a BiliClient, upper_id: String) -> Self {
         Self { client, upper_id }

crates/bili_sync/src/config/versioned_config.rs

@@ -53,11 +53,16 @@ impl VersionedConfig {
     }
 
     #[cfg(test)]
-    /// 单元测试直接使用测试专用的配置即可
-    pub fn get() -> &'static VersionedConfig {
-        use std::sync::LazyLock;
-        static TEST_CONFIG: LazyLock<VersionedConfig> = LazyLock::new(|| VersionedConfig::new(Config::test_default()));
-        return &TEST_CONFIG;
+    /// 仅在测试环境使用，该方法会尝试从测试数据库中加载配置并写入到全局的 VERSIONED_CONFIG
+    pub async fn init_for_test(connection: &DatabaseConnection) -> Result<()> {
+        let Some(Ok(config)) = Config::load_from_database(&connection).await? else {
+            bail!("no config found in test database");
+        };
+        let versioned_config = VersionedConfig::new(config);
+        VERSIONED_CONFIG
+            .set(versioned_config)
+            .map_err(|e| anyhow!("VERSIONED_CONFIG has already been initialized: {}", e))?;
+        Ok(())
     }
 
     #[cfg(not(test))]
@@ -66,6 +71,16 @@ impl VersionedConfig {
         VERSIONED_CONFIG.get().expect("VERSIONED_CONFIG is not initialized")
     }
 
+    #[cfg(test)]
+    /// 尝试获取全局的 `VersionedConfig`，如果未初始化则退回测试环境的默认配置
+    pub fn get() -> &'static VersionedConfig {
+        use std::sync::LazyLock;
+        static FALLBACK_CONFIG: LazyLock<VersionedConfig> =
+            LazyLock::new(|| VersionedConfig::new(Config::test_default()));
+        // 优先从全局变量获取，未初始化则返回测试环境的默认配置
+        return VERSIONED_CONFIG.get().unwrap_or_else(|| &FALLBACK_CONFIG);
+    }
+
     pub fn new(config: Config) -> Self {
         Self {
             inner: ArcSwap::from_pointee(config),

crates/bili_sync/src/database.rs

@@ -1,3 +1,4 @@
+use std::path::Path;
 use std::time::Duration;
 
 use anyhow::{Context, Result};
@@ -6,14 +7,12 @@ use sea_orm::sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqliteSynch
 use sea_orm::sqlx::{ConnectOptions as SqlxConnectOptions, Sqlite};
 use sea_orm::{ConnectOptions, Database, DatabaseConnection, SqlxSqliteConnector};
 
-use crate::config::CONFIG_DIR;
-
-fn database_url() -> String {
-    format!("sqlite://{}?mode=rwc", CONFIG_DIR.join("data.sqlite").to_string_lossy())
+fn database_url(path: &Path) -> String {
+    format!("sqlite://{}?mode=rwc", path.to_string_lossy())
 }
 
-async fn database_connection() -> Result<DatabaseConnection> {
-    let mut option = ConnectOptions::new(database_url());
+async fn database_connection(database_url: &str) -> Result<DatabaseConnection> {
+    let mut option = ConnectOptions::new(database_url);
     option
         .max_connections(50)
         .min_connections(5)
@@ -35,18 +34,25 @@ async fn database_connection() -> Result<DatabaseConnection> {
     ))
 }
 
-async fn migrate_database() -> Result<()> {
+async fn migrate_database(database_url: &str) -> Result<()> {
     // 注意此处使用内部构造的 DatabaseConnection，而不是通过 database_connection() 获取
     // 这是因为使用多个连接的 Connection 会导致奇怪的迁移顺序问题，而使用默认的连接选项不会
-    let connection = Database::connect(database_url()).await?;
+    let connection = Database::connect(database_url).await?;
     Ok(Migrator::up(&connection, None).await?)
 }
 
 /// 进行数据库迁移并获取数据库连接，供外部使用
-pub async fn setup_database() -> Result<DatabaseConnection> {
-    tokio::fs::create_dir_all(CONFIG_DIR.as_path()).await.context(
-        "Failed to create config directory. Please check if you have granted necessary permissions to your folder.",
-    )?;
-    migrate_database().await.context("Failed to migrate database")?;
-    database_connection().await.context("Failed to connect to database")
+pub async fn setup_database(path: &Path) -> Result<DatabaseConnection> {
+    if let Some(parent) = path.parent() {
+        tokio::fs::create_dir_all(parent).await.context(
+            "Failed to create config directory. Please check if you have granted necessary permissions to your folder.",
+        )?;
+    }
+    let database_url = database_url(path);
+    migrate_database(&database_url)
+        .await
+        .context("Failed to migrate database")?;
+    database_connection(&database_url)
+        .await
+        .context("Failed to connect to database")
 }

crates/bili_sync/src/main.rs

@@ -25,7 +25,7 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 
 use crate::api::{LogHelper, MAX_HISTORY_LOGS};
-use crate::config::{ARGS, VersionedConfig};
+use crate::config::{ARGS, CONFIG_DIR, VersionedConfig};
 use crate::database::setup_database;
 use crate::utils::init_logger;
 use crate::utils::signal::terminate;
@@ -85,7 +85,9 @@ async fn init() -> (DatabaseConnection, LogHelper) {
     init_logger(&ARGS.log_level, Some(log_writer.clone()));
     info!("欢迎使用 Bili-Sync，当前程序版本：{}", config::version());
     info!("项目地址：https://github.com/amtoaer/bili-sync");
-    let connection = setup_database().await.expect("数据库初始化失败");
+    let connection = setup_database(&CONFIG_DIR.join("data.sqlite"))
+        .await
+        .expect("数据库初始化失败");
     info!("数据库初始化完成");
     VersionedConfig::init(&connection).await.expect("配置初始化失败");
     info!("配置初始化完成");

crates/bili_sync/src/utils/convert.rs

@@ -97,7 +97,23 @@ impl VideoInfo {
                 valid: Set(true),
                 ..default
             },
-            _ => unreachable!(),
+            VideoInfo::Dynamic {
+                title,
+                bvid,
+                desc,
+                cover,
+                pubtime,
+            } => bili_sync_entity::video::ActiveModel {
+                bvid: Set(bvid),
+                name: Set(title),
+                intro: Set(desc),
+                cover: Set(cover),
+                pubtime: Set(pubtime.naive_utc()),
+                category: Set(2), // 动态里的视频内容类型肯定是视频
+                valid: Set(true),
+                ..default
+            },
+            VideoInfo::Detail { .. } => unreachable!(),
         }
     }
 
@@ -145,8 +161,9 @@ impl VideoInfo {
             VideoInfo::Collection { pubtime: time, .. }
             | VideoInfo::Favorite { fav_time: time, .. }
             | VideoInfo::WatchLater { fav_time: time, .. }
-            | VideoInfo::Submission { ctime: time, .. } => time,
-            _ => unreachable!(),
+            | VideoInfo::Submission { ctime: time, .. }
+            | VideoInfo::Dynamic { pubtime: time, .. } => time,
+            VideoInfo::Detail { .. } => unreachable!(),
         }
     }
 }

crates/bili_sync/src/workflow.rs

@@ -58,7 +58,8 @@ pub async fn refresh_video_source<'a>(
     let mut max_datetime = latest_row_at;
     let mut error = Ok(());
     let mut video_streams = video_streams
-        .take_while(|res| {
+        .enumerate()
+        .take_while(|(idx, res)| {
             match res {
                 Err(e) => {
                     error = Err(anyhow!(e.to_string()));
@@ -72,11 +73,11 @@ pub async fn refresh_video_source<'a>(
                     if release_datetime > &max_datetime {
                         max_datetime = *release_datetime;
                     }
-                    futures::future::ready(video_source.should_take(release_datetime, &latest_row_at))
+                    futures::future::ready(video_source.should_take(*idx, release_datetime, &latest_row_at))
                 }
             }
         })
-        .filter_map(|res| futures::future::ready(video_source.should_filter(res, &latest_row_at)))
+        .filter_map(|(idx, res)| futures::future::ready(video_source.should_filter(idx, res, &latest_row_at)))
         .chunks(10);
     let mut count = 0;
     while let Some(videos_info) = video_streams.next().await {