plugin transcode

Author: neckaros

Cargo.lock

@@ -70,7 +70,7 @@ sha256 = "1.5.0"
 axum-extra = { version = "0.9.2", features = ["query"] }
 http = "1.1.0"
 extism = "1.10.0"
-rs-plugin-common-interfaces = { version = "0.19.0", features = ["rusqlite",] }
+rs-plugin-common-interfaces = { version = "0.20.3", features = ["rusqlite",] }
 async-recursion = "1.1.0"
 async-compression = { version = "0.4.6", features = ["tokio"] }
 youtube_dl = { version = "0.10.0", features = ["tokio", "downloader-rustls-tls"] }

Cargo.toml

@@ -1,13 +1,13 @@
 use std::str::FromStr;
 
 use nanoid::nanoid;
-use rs_plugin_common_interfaces::{request::{RsCookie, RsRequest, RsRequestStatus}, url::RsLink};
+use rs_plugin_common_interfaces::{request::{RsCookie, RsRequest, RsRequestStatus}, url::RsLink, video::VideoConvertRequest};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 use strum_macros::EnumString;
 
 
-use crate::{plugins::sources::SourceRead, tools::video_tools::VideoConvertRequest};
+use crate::plugins::sources::SourceRead;
 
 use super::{progress::RsProgress, ElementAction};
 

src/domain/media.rs

@@ -13,16 +13,16 @@ use mime_guess::get_mime_extensions_str;
 use nanoid::nanoid;
 use query_external_ip::SourceError;
 use regex::Regex;
-use rs_plugin_common_interfaces::{request::{RsRequest, RsRequestStatus}, url::{RsLink, RsLinkType}, PluginType, RsCookie};
+use rs_plugin_common_interfaces::{request::{RsRequest, RsRequestStatus}, url::{RsLink, RsLinkType}, video::{RsVideoTranscodeJob, RsVideoTranscodeJobPluginRequest, RsVideoTranscodeStatus, VideoConvertRequest, VideoOverlayType}, PluginType, RsCookie};
 use rusqlite::{types::{FromSql, FromSqlError, FromSqlResult, ToSqlOutput, ValueRef}, ToSql};
 use serde::{Deserialize, Serialize};
 use strum_macros::EnumString;
-use tokio::{fs::File, io::{copy, AsyncRead, AsyncReadExt, AsyncWriteExt}, sync::mpsc};
+use tokio::{fs::File, io::{copy, AsyncRead, AsyncReadExt, AsyncWriteExt}, sync::mpsc, time::sleep, time::Duration};
 use tokio_stream::StreamExt;
 use tokio_util::{compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}, io::{ReaderStream, StreamReader, SyncIoBridge}};
 use zip::ZipWriter;
 
-use crate::{domain::{deleted::RsDeleted, library::LibraryType, media::{self, ConvertMessage, ConvertProgress, RsGpsPosition, DEFAULT_MIME}, plugin, MediaElement}, error::RsError, model::store::sql::SqlOrder, plugins::sources::{path_provider::PathProvider, Source}, routes::infos, tools::{file_tools::{filename_from_path, remove_extension}, image_tools::{convert_image_reader, image_infos, IMAGES_MIME_FULL_BROWSER_SUPPORT}, recognition, video_tools::{VideoCommandBuilder, VideoConvertRequest, VideoOverlayPosition}}};
+use crate::{domain::{deleted::RsDeleted, library::LibraryType, media::{self, ConvertMessage, ConvertProgress, RsGpsPosition, DEFAULT_MIME}, plugin, MediaElement}, error::RsError, model::store::sql::SqlOrder, plugins::sources::{path_provider::PathProvider, Source}, routes::infos, tools::{file_tools::{filename_from_path, remove_extension}, image_tools::{convert_image_reader, image_infos, IMAGES_MIME_FULL_BROWSER_SUPPORT}, recognition, video_tools::{VideoCommandBuilder}}};
 
 use crate::{domain::{library::LibraryRole, media::{FileType, GroupMediaDownload, Media, MediaDownloadUrl, MediaForAdd, MediaForInsert, MediaForUpdate, MediaItemReference, MediaWithAction, MediasMessage, ProgressMessage}, progress::{RsProgress, RsProgressType}, ElementAction}, error::RsResult, plugins::{get_plugin_fodler, sources::{async_reader_progress::ProgressReader, error::SourcesError, AsyncReadPinBox, FileStreamResult, SourceRead}}, routes::mw_range::RangeDefinition, server::get_server_port, tools::{auth::{sign_local, ClaimsLocal}, file_tools::{file_type_from_mime, get_extension_from_mime}, image_tools::{self, resize_image_reader, ImageSize}, log::{log_error, log_info, LogServiceType}, prediction::{predict_net, preload_model, PredictionTagResult}, video_tools::{self, probe_video, VideoTime}}};
 
@@ -1050,16 +1050,16 @@ impl ModelController {
         let uri = if let Some(local_path) = local_path {
             local_path.to_str().unwrap().to_string()
         } else {
-            ModelController::get_temporary_local_read_url(library_id, media_id).await?
+            ModelController::get_temporary_local_read_url(library_id, media_id, None).await?
         };
         let thumb = video_tools::thumb_video(&uri, time).await?;
         let mut cursor = std::io::Cursor::new(thumb);
         let thumb = resize_image_reader(Box::pin(cursor), 512, format, quality, false).await?;
         Ok(thumb)
     }
 
-    pub async  fn get_temporary_local_read_url(library_id: &str, media_id: &str) -> Result<String> {
-        let exp = ClaimsLocal::generate_seconds(240);
+    pub async  fn get_temporary_local_read_url(library_id: &str, media_id: &str, delay: Option<u64>) -> Result<String> {
+        let exp = ClaimsLocal::generate_seconds(delay.unwrap_or(240));
         let claims = ClaimsLocal {
             cr: "service::get_video_thumb".to_string(),
             kind: crate::tools::auth::ClaimsLocalType::File(library_id.to_string(), media_id.to_string()),
@@ -1127,20 +1127,124 @@ impl ModelController {
         }
     }
 
-    pub async fn convert(&self, library_id: &str, media_id: &str, request: VideoConvertRequest, requesting_user: &ConnectedUser) -> crate::Result<()> {
+
+    pub async fn convert(&self, library_id: &str, media_id: &str, request: VideoConvertRequest, plugin_id: Option<String>, requesting_user: &ConnectedUser) -> crate::Result<()> {
         requesting_user.check_file_role(library_id, media_id, LibraryRole::Write)?;
         let store = self.store.get_library_store(library_id)?;
         let media = store.get_media(media_id, requesting_user.user_id().ok()).await?.ok_or(SourcesError::UnableToFindMedia(library_id.to_string(), media_id.to_string(), "convert".to_string()))?;
 
 
         let filename = format!("{}.{}", remove_extension(&media.name), request.format);
-        let queue_element = VideoConvertQueueElement::new(library_id.to_string(), media_id.to_string(), filename, requesting_user.clone(), request);
+        let queue_element = VideoConvertQueueElement::new(library_id.to_string(), plugin_id.clone(), media_id.to_string(), filename.clone(), requesting_user.clone(), request.clone());
         let message = ConvertMessage {
             library: library_id.to_string(),
             progress: queue_element.status.clone(),
         };
         self.send_convert_progress(message);
 
+
+        // If plugin convert is requested
+        if let Some(plugin_id) = &plugin_id {
+            let status = self.convert_submit_media(library_id, media_id, request.clone(), &plugin_id).await?;
+            let job_id = status.id.clone();
+            
+            // Clone what you need for the background task
+            let plugin_id_owned = plugin_id.clone();
+            let mc_progress = self.clone(); // or however you get your progress sender
+            let request_clone = request.clone();
+            let lib_progress = library_id.to_string(); // adjust based on your actual type
+            let name_progress = filename.clone();
+            let media_progress: MediaForUpdate = media.into();
+            // Spawn background task
+            tokio::spawn(async move {
+    let mut poll_interval = Duration::from_secs(2);
+    
+    loop {
+        sleep(poll_interval).await;
+        
+        match mc_progress.convert_status(&job_id, &plugin_id_owned).await {
+                        Ok(current_status) => {
+                            let is_terminal = matches!(
+                                current_status.status,
+                                RsVideoTranscodeStatus::Completed 
+                                | RsVideoTranscodeStatus::Failed 
+                                | RsVideoTranscodeStatus::Canceled
+                            );
+                            
+                            let done = is_terminal;
+                            let percent = current_status.progress;
+                            
+                            let message = ConvertMessage {
+                                library: lib_progress.clone(),
+                                progress: ConvertProgress {
+                                    percent: percent.into(),
+                                    converted_id: if done { Some(job_id.clone()) } else { None },
+                                    filename: name_progress.clone(),
+                                    done,
+                                    id: request_clone.id.clone(),
+                                    request: Some(request_clone.clone()),
+                                    estimated_remaining_seconds: None,
+                                },
+                            };
+                            mc_progress.send_convert_progress(message);
+                            
+                            if matches!(current_status.status, RsVideoTranscodeStatus::Completed) {
+                                // Handle all errors here instead of using ?
+                                match mc_progress.convert_link(&job_id, &plugin_id_owned).await {
+                                    Ok(link) => {
+                                        let source = SourceRead::Request(link);
+                                        
+                                        match source.into_reader(
+                                            Some(&lib_progress), 
+                                            None, 
+                                            None, 
+                                            Some((mc_progress.clone(), &ConnectedUser::ServerAdmin)), 
+                                            None
+                                        ).await {
+                                            Ok(reader) => {
+                                                // Note: Can't use self here, need to use mc_progress
+                                                match mc_progress.add_library_file(
+                                                    &lib_progress, 
+                                                    &name_progress, 
+                                                    Some(media_progress.clone()), 
+                                                    reader.stream, 
+                                                    &ConnectedUser::ServerAdmin  // Or pass requesting_user as owned
+                                                ).await {
+                                                    Ok(media) => {
+                                                        // Success - maybe send final progress update
+                                                        tracing::info!("Successfully added media: {:?}", media);
+                                                    }
+                                                    Err(e) => {
+                                                        tracing::error!("Failed to add library file: {:?}", e);
+                                                        // Send error progress update
+                                                    }
+                                                }
+                                            }
+                                            Err(e) => {
+                                                tracing::error!("Failed to create reader: {:?}", e);
+                                            }
+                                        }
+                                    }
+                                    Err(e) => {
+                                        tracing::error!("Failed to get convert link: {:?}", e);
+                                    }
+                                }
+                            }
+                            
+                            if is_terminal {
+                                break;
+                            }
+                        }
+                        Err(e) => {
+                            tracing::error!("Error polling status: {:?}", e);
+                            break;
+                        }
+                    }
+                }
+            });
+            return Ok(());
+        }
+
         let converting = self.convert_current.read().await;
         let mut queue = self.convert_queue.write().await;
         queue.push_back(queue_element);
@@ -1233,11 +1337,11 @@ impl ModelController {
 
         if let Some(overlay) = &mut element.request.overlay {
             match overlay.kind {
-                video_tools::VideoOverlayType::Watermark => {
+                VideoOverlayType::Watermark => {
                     let name = if overlay.path.is_empty() { ".watermark.png".to_owned() } else { format!(".watermark.{}.png", &overlay.path)};
                     overlay.path = local.get_full_path(&name).to_str().ok_or(Error::ServiceError("Convert".to_owned(), Some("Invalid watermark path".to_owned())))?.to_string();
                 },
-                video_tools::VideoOverlayType::File => todo!(),
+                VideoOverlayType::File => todo!(),
             }
 
         }
@@ -1301,7 +1405,7 @@ impl ModelController {
         let uri = if let Some(local_path) = local_path {
             local_path.to_str().unwrap().to_string()
         } else {
-            ModelController::get_temporary_local_read_url(library_id, media_id).await?
+            ModelController::get_temporary_local_read_url(library_id, media_id, Some(240)).await?
         };
 
         let videos_infos = probe_video(&uri).await?;

src/model/medias.rs

@@ -21,10 +21,10 @@ pub mod player;
 use std::{collections::{HashMap, VecDeque}, io::Read, path::PathBuf, pin::Pin, sync::Arc, thread::JoinHandle};
 use futures::lock::Mutex;
 use nanoid::nanoid;
-use rs_plugin_common_interfaces::{ImageType, RsRequest};
+use rs_plugin_common_interfaces::{video::VideoConvertRequest, ImageType, RsRequest};
 use serde::{Deserialize, Serialize};
 use strum::IntoEnumIterator;
-use crate::{domain::{backup::BackupProcessStatus, library::{LibraryMessage, LibraryRole, ServerLibrary}, media::ConvertProgress, player::RsPlayerAvailable}, error::{RsError, RsResult}, plugins::{medias::{fanart::FanArtContext, imdb::ImdbContext, tmdb::TmdbContext, trakt::TraktContext}, sources::{error::SourcesError, local_provider_for_library, path_provider::PathProvider, AsyncReadPinBox, FileStreamResult, Source, SourceRead}, PluginManager}, tools::{clock::SECONDS_IN_HOUR, image_tools::{resize_image_reader, ImageSize}, log::log_info, scheduler::{self, ip::RefreshIpTask, refresh::RefreshTask, RsScheduler, RsTaskType}, video_tools::VideoConvertRequest}};
+use crate::{domain::{backup::BackupProcessStatus, library::{LibraryMessage, LibraryRole, ServerLibrary}, media::ConvertProgress, player::RsPlayerAvailable}, error::{RsError, RsResult}, plugins::{medias::{fanart::FanArtContext, imdb::ImdbContext, tmdb::TmdbContext, trakt::TraktContext}, sources::{error::SourcesError, local_provider_for_library, path_provider::PathProvider, AsyncReadPinBox, FileStreamResult, Source, SourceRead}, PluginManager}, tools::{clock::SECONDS_IN_HOUR, image_tools::{resize_image_reader, ImageSize}, log::log_info, scheduler::{self, ip::RefreshIpTask, refresh::RefreshTask, RsScheduler, RsTaskType}}};
 
 use self::{medias::CRYPTO_HEADER_SIZE, store::SqliteStore, users::{ConnectedUser, ServerUser, UserRole}};
 use error::{Result, Error};
@@ -38,13 +38,15 @@ pub struct VideoConvertQueueElement {
     media: String,
 	user: ConnectedUser,
 	id: String,
+	plugin_id: Option<String>,
 	status: ConvertProgress
 
 }
 
 impl VideoConvertQueueElement {
-	pub fn new(library: String, media: String, filename: String, user: ConnectedUser, request: VideoConvertRequest) -> VideoConvertQueueElement {
+	pub fn new(library: String, plugin_id: Option<String>, media: String, filename: String, user: ConnectedUser, request: VideoConvertRequest) -> VideoConvertQueueElement {
 		VideoConvertQueueElement {id: request.id.clone(), 
+			plugin_id,
 			status: ConvertProgress { id: request.id.clone(), filename, converted_id: None, done: false, percent: 0f64, estimated_remaining_seconds: None, request: Some(request.clone()) },
 			request, library, media, user, 
 			 }

src/model/mod.rs

@@ -15,6 +15,8 @@ use crate::{domain::{backup::Backup, library::LibraryRole, plugin::{Plugin, Plug
 
 use super::{error::{Error, Result}, users::{ConnectedUser, UserRole}, ModelController};
 
+pub mod video_convert_plugin;
+
 #[derive(Debug, Serialize, Deserialize, Clone, Default)]
 pub struct PluginQuery {
     pub kind: Option<PluginType>,
@@ -55,7 +57,7 @@ impl ModelController {
 		Ok(plugins)
 	}
 
-    async fn get_plugins_with_credential(&self, query: PluginQuery) -> Result<impl Iterator<Item = PluginWithCredential>> {
+    pub async fn get_plugins_with_credential(&self, query: PluginQuery) -> Result<impl Iterator<Item = PluginWithCredential>> {
 		let plugins = self.store.get_plugins(query).await?.into_iter();
 		let credentials = self.store.get_credentials().await?;
         let iter = plugins.map(move |p| {
@@ -70,6 +72,14 @@ impl ModelController {
 		let credential = self.store.get_plugin(&plugin_id).await?.ok_or(SourcesError::UnableToFindPlugin(plugin_id.to_string(), "get_plugin".to_string()))?;
 		Ok(credential)
 	}
+    
+    pub async fn get_plugin_with_credential(&self, id: &str) -> Result<PluginWithCredential> {
+		let plugin = self.store.get_plugin(id).await?.ok_or(SourcesError::UnableToFindPlugin(id.to_string(), "get_plugin_with_credential".to_string()))?;
+		let credentials = self.store.get_credentials().await?;
+      
+        let credential = credentials.iter().find(|c| Some(&c.id) == plugin.credential.as_ref()).cloned();
+        Ok(PluginWithCredential { plugin: plugin, credential })
+	}
 
     pub async fn reload_plugins(&self, requesting_user: &ConnectedUser) -> RsResult<()> {
         requesting_user.check_role(&UserRole::Admin)?;