feat: pull rss

Cargo.lock

@@ -2814,6 +2814,7 @@ dependencies = [
  "futures",
  "include_dir",
  "insta",
+ "itertools",
  "loco-rs",
  "reqwest",
  "rss",

crates/recorder/Cargo.toml

@@ -35,6 +35,7 @@ thiserror = "1.0.57"
 rss = "2.0.7"
 bytes = "1.5.0"
 futures = "0.3.30"
+itertools = "0.12.1"
 
 [lib]
 name = "recorder"

crates/recorder/src/migrations/defs.rs

@@ -62,6 +62,7 @@ pub enum Downloads {
     CurrSize,
     AllSize,
     Mime,
+    Url,
 }
 
 #[async_trait::async_trait]

crates/recorder/src/migrations/m20240224_082543_add_downloads.rs

@@ -51,6 +51,10 @@ impl MigrationTrait for Migration {
                     ))
                     .col(big_unsigned(Downloads::AllSize))
                     .col(big_unsigned(Downloads::CurrSize))
+                    .col(text(Downloads::Url))
+                    .index(
+                        Index::create().table(Downloads::Table).col(Downloads::Url).name("idx_download_url")
+                    )
                     .foreign_key(
                         ForeignKey::create()
                             .name("fk_download_subscription_id")

crates/recorder/src/models/_entities/downloads.rs

@@ -46,6 +46,7 @@ pub struct Model {
     pub subscription_id: i32,
     pub status: DownloadStatus,
     pub mime: DownloadMime,
+    pub url: String,
     pub all_size: u64,
     pub curr_size: u64,
 }

crates/recorder/src/models/downloads.rs

@@ -1,6 +1,67 @@
-use sea_orm::ActiveModelBehavior;
+use std::collections::HashMap;
+use sea_orm::{prelude::*, ActiveValue, Condition, QuerySelect, SelectColumns};
 
 use crate::models::_entities::downloads::*;
+use crate::models::prelude::{SubscriptionCategory, subscriptions};
+use crate::subscriptions::mikan::MikanSubscriptionEngine;
 
 #[async_trait::async_trait]
 impl ActiveModelBehavior for ActiveModel {}
+
+impl Model {
+    pub async fn pull_subscription(
+        db: &DatabaseConnection,
+        item: &subscriptions::Model,
+    ) -> eyre::Result<()> {
+        match &item.category {
+            SubscriptionCategory::Mikan => {
+                let items =
+                    MikanSubscriptionEngine::subscription_items_from_rss_url(&item.source_url).
+                        await?;
+                let items = items.collect::<Vec<_>>();
+                let mut all_items = items
+                    .into_iter()
+                    .map(|item| (item.url.clone(), item))
+                    .collect::<HashMap<_, _>>();
+
+                let existed_items = {
+                    Entity::find()
+                        .filter(
+                            Condition::all()
+                                .add(Column::SubscriptionId.eq(item.id))
+                                .add(Column::Url.is_in(all_items.keys().cloned()))
+                        )
+                        .select_only()
+                        .select_column(Column::Url)
+                        .all(db).await?
+                };
+
+                for dl in existed_items {
+                    all_items.remove(&dl.url as &str);
+                }
+
+                let new_items = all_items.into_values().map(|i| {
+                    ActiveModel {
+                        origin_name: ActiveValue::Set(i.title.clone()),
+                        display_name: ActiveValue::Set(i.title),
+                        subscription_id: ActiveValue::Set(item.id),
+                        status: ActiveValue::Set(DownloadStatus::Pending),
+                        mime: ActiveValue::Set(DownloadMime::BitTorrent),
+                        url: ActiveValue::Set(i.url),
+                        all_size: ActiveValue::Set(i.content_length.unwrap_or_default()),
+                        curr_size: ActiveValue::Set(0),
+                        ..Default::default()
+                    }
+                });
+
+                Entity::insert_many(new_items)
+                    .exec(db)
+                    .await?;
+            }
+            _ => {
+                todo!("other subscription categories")
+            }
+        }
+        Ok(())
+    }
+}

crates/recorder/src/models/subscriptions.rs

@@ -48,32 +48,4 @@ impl Model {
             .await?;
         Ok(())
     }
-
-    // pub async fn pull_rss (
-    //     db: &DatabaseConnection,
-    //     item: &Self,
-    // ) -> eyre::Result<()> {
-    //     match &item.category {
-    //         SubscriptionCategory::Mikan => {
-    //             let items =
-    // MikanSubscriptionEngine::subscription_items_from_rss_url(&item.source_url).
-    // await?;             let items = items.collect::<Vec<_>>();
-    //             let torrent_urls = items.iter().map(|item| item.torrent_url());
-    //
-    //             let new_torrents = Entity::find()
-    //                 .filter(
-    //                     Column::SourceUrl
-    //                 )
-    //                 .all(db).await?;
-    //
-    //             for item in items {
-    //                 println!("{:?}", item);
-    //             }
-    //         }
-    //         _ => {
-    //             todo!("other subscription categories")
-    //         }
-    //     }
-    //     Ok(())
-    // }
 }

crates/recorder/src/subscriptions/mikan.rs

@@ -3,34 +3,33 @@ use crate::downloader::defs::BITTORRENT_MIME_TYPE;
 
 #[derive(Debug, Clone)]
 pub struct MikanSubscriptionItem {
-    pub item: rss::Item,
+    pub title: String,
-}
+    pub home_page: Option<String>,
-
+    pub url: String,
-impl From<rss::Item> for MikanSubscriptionItem {
+    pub content_length: Option<u64>,
-    fn from(item: rss::Item) -> Self {
+    pub mime: String,
-        MikanSubscriptionItem {
+    pub pub_date: Option<String>,
-            item
-        }
-    }
 }
 
 impl MikanSubscriptionItem {
-    pub fn title(&self) -> &str {
+    pub fn from_rss_item(item: rss::Item) -> Option<Self> {
-        self.item.title().unwrap_or_default()
+        let mime_match = item.enclosure()
-    }
+            .map(|x| x.mime_type == BITTORRENT_MIME_TYPE)
-
+            .unwrap_or_default();
-    pub fn homepage(&self) -> Option<&str> {
+        if mime_match {
-        self.item.link()
+            let enclosure = item.enclosure.unwrap();
-    }
+            let content_length = enclosure.length.parse().ok();
-
+            Some(MikanSubscriptionItem {
-    pub fn torrent_url (&self) -> Option<&str> {
+                title: item.title.unwrap_or_default(),
-        self.item.enclosure().and_then(|en| {
+                home_page: item.link,
-            if en.mime_type == BITTORRENT_MIME_TYPE {
+                url: enclosure.url,
-                Some(en.url.as_str())
+                content_length,
+                mime: enclosure.mime_type,
+                pub_date: item.pub_date,
+            })
         } else {
             None
         }
-        })
     }
 }
 
@@ -44,6 +43,6 @@ impl MikanSubscriptionEngine {
 
         let channel = rss::Channel::read_from(&bytes[..])?;
 
-        Ok(channel.items.into_iter().map(MikanSubscriptionItem::from))
+        Ok(channel.items.into_iter().flat_map(MikanSubscriptionItem::from_rss_item))
     }
 }