feat: Seek log WIP

Signed-off-by: Lachezar Lechev <[email protected]>

Cargo.toml

@@ -59,6 +59,8 @@ chrono = { version = "0.4", features = ["serde"] }
 semver = { version = "1", features = ["serde"] }
 base64 = "0.21"
 sha1 = "0.10"
+sha2 = "0.10"
+
 either = "1.6"
 enclose = "1.1"
 derivative = "2.2"

src/constants.rs

@@ -50,6 +50,9 @@ pub const URI_COMPONENT_ENCODE_SET: &AsciiSet = &NON_ALPHANUMERIC
     .remove(b'(')
     .remove(b')');
 
+/// In milliseconds
+pub const PLAYER_IGNORE_SEEK_AFTER: u64 = 600000;
+
 pub static BASE64: base64::engine::general_purpose::GeneralPurpose =
     base64::engine::general_purpose::STANDARD;
 

src/models/player.rs

@@ -1,19 +1,25 @@
 use std::marker::PhantomData;
 
+use base64::Engine;
+use futures::{future, FutureExt, TryFutureExt};
+
 use crate::constants::{
-    CREDITS_THRESHOLD_COEF, VIDEO_HASH_EXTRA_PROP, VIDEO_SIZE_EXTRA_PROP, WATCHED_THRESHOLD_COEF,
+    BASE64, CREDITS_THRESHOLD_COEF, PLAYER_IGNORE_SEEK_AFTER, VIDEO_HASH_EXTRA_PROP,
+    VIDEO_SIZE_EXTRA_PROP, WATCHED_THRESHOLD_COEF,
 };
 use crate::models::common::{
-    eq_update, resource_update, resources_update_with_vector_content, Loadable, ResourceAction,
-    ResourceLoadable, ResourcesAction,
+    eq_update, resource_update, resource_update_with_vector_content,
+    resources_update_with_vector_content, Loadable, ResourceAction, ResourceLoadable,
+    ResourcesAction,
 };
-use crate::models::ctx::Ctx;
+use crate::models::ctx::{Ctx, CtxError};
 use crate::runtime::msg::{Action, ActionLoad, ActionPlayer, Event, Internal, Msg};
-use crate::runtime::{Effects, Env, UpdateWithCtx};
+use crate::runtime::{Effect, EffectFuture, Effects, Env, EnvFutureExt, UpdateWithCtx};
 use crate::types::addon::{AggrRequest, Descriptor, ExtraExt, ResourcePath, ResourceRequest};
+use crate::types::api::{fetch_api, APIRequest, APIResult, SeekLogRequest, SuccessResponse};
 use crate::types::library::{LibraryBucket, LibraryItem};
 use crate::types::profile::Settings as ProfileSettings;
-use crate::types::resource::{MetaItem, SeriesInfo, Stream, Subtitles, Video};
+use crate::types::resource::{MetaItem, SeriesInfo, Stream, StreamSource, Subtitles, Video};
 
 use stremio_watched_bitfield::WatchedBitField;
 
@@ -24,8 +30,6 @@ use serde::{Deserialize, Serialize};
 
 use lazy_static::lazy_static;
 
-use super::common::resource_update_with_vector_content;
-
 lazy_static! {
     /// The duration that must have passed in order for a library item to be updated.
     pub static ref PUSH_TO_LIBRARY_EVERY: Duration = Duration::seconds(30);
@@ -57,6 +61,9 @@ pub struct AnalyticsContext {
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct VideoParams {
+    /// Opensubtitles hash usually retrieved from a streaming server endpoint.
+    ///
+    /// It's used for requesting subtitles from Opensubtitles.
     pub hash: Option<String>,
     pub size: Option<u64>,
 }
@@ -99,6 +106,17 @@ pub struct Player {
     pub ended: bool,
     #[serde(skip_serializing)]
     pub paused: Option<bool>,
+    #[serde(skip_serializing)]
+    pub seek_history: Vec<SeekLog>,
+}
+
+#[derive(Debug, Clone, Copy, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct SeekLog {
+    /// in milliseconds
+    pub from: u64,
+    /// in milliseconds
+    pub to: u64,
 }
 
 impl<E: Env + 'static> UpdateWithCtx<E> for Player {
@@ -326,7 +344,18 @@ impl<E: Env + 'static> UpdateWithCtx<E> for Player {
                     Some(library_item),
                 ) => {
                     let seeking = library_item.state.time_offset.abs_diff(*time) > 1000;
-                    // library_item.state.last_watched = Some(E::now() - chrono::Duration::days(1));
+
+                    // seek logging
+                    if seeking
+                        && library_item.r#type == "series"
+                        && time < &PLAYER_IGNORE_SEEK_AFTER
+                    {
+                        self.seek_history.push(SeekLog {
+                            from: *time,
+                            to: library_item.state.time_offset,
+                        });
+                    }
+
                     library_item.state.last_watched = Some(E::now());
                     if library_item.state.video_id != Some(video_id.to_owned()) {
                         library_item.state.video_id = Some(video_id.to_owned());
@@ -451,6 +480,69 @@ impl<E: Env + 'static> UpdateWithCtx<E> for Player {
                 };
                 trakt_event_effects.join(update_library_item_effects)
             }
+            Msg::Action(Action::Player(ActionPlayer::NextVideo)) => {
+                let seek_outro_effects = match (
+                    self.selected.as_ref(),
+                    self.video_params.as_ref(),
+                    self.series_info.as_ref(),
+                    self.library_item.as_ref(),
+                ) {
+                    (Some(selected), Some(video_params), Some(series_info), Some(library_item)) => {
+                        match (
+                            &selected.stream.source,
+                            selected.stream.name.as_ref(),
+                            video_params.hash.as_ref(),
+                        ) {
+                            (
+                                StreamSource::Torrent { .. },
+                                Some(stream_name),
+                                Some(opensubtitles_hash),
+                            ) => {
+                                let filename_hash = {
+                                    use sha2::Digest;
+                                    let mut sha256 = sha2::Sha256::new();
+                                    sha256.update(&stream_name);
+                                    let sha256_encoded = sha256.finalize();
+
+                                    BASE64.encode(sha256_encoded)
+                                };
+
+                                let seek_log_req = SeekLogRequest {
+                                    opensubtitles_hash: opensubtitles_hash.clone(),
+                                    item_id: library_item.id.clone(),
+                                    series_info: series_info.clone(),
+                                    filename_hash,
+                                    // library_item.state.video_id.to_owned(),
+                                    // Some(library_item.state.time_offset),
+                                    // Some(library_item.state.duration),
+                                    duration: library_item.state.duration,
+                                    seek_history: self.seek_history.clone(),
+                                    skip_outro: library_item.state.time_offset,
+                                };
+                                // TODO: FIX
+
+                                Effects::one(send_seek_log_api::<E>(seek_log_req)).unchanged()
+                            }
+                            _ => Effects::none().unchanged(),
+                        }
+                    }
+                    _ => Effects::none().unchanged(),
+                };
+
+                let next_video_effects =
+                    switch_to_next_video(&mut self.library_item, &self.next_video);
+
+                seek_outro_effects
+                    .join(
+                        Effects::msg(Msg::Event(Event::PlayerNextVideo {
+                            context: self.analytics_context.as_ref().cloned().unwrap_or_default(),
+                            is_binge_enabled: ctx.profile.settings.binge_watching,
+                            is_playing_next_video: self.next_video.is_some(),
+                        }))
+                        .unchanged(),
+                    )
+                    .join(next_video_effects)
+            }
             Msg::Action(Action::Player(ActionPlayer::Ended)) if self.selected.is_some() => {
                 self.ended = true;
                 Effects::msg(Msg::Event(Event::PlayerEnded {
@@ -870,6 +962,22 @@ fn subtitles_update<E: Env + 'static>(
     }
 }
 
+fn send_seek_log_api<E: Env + 'static>(seek_log_req: SeekLogRequest) -> Effect {
+    let api_request = APIRequest::SeekLog(seek_log_req.clone());
+
+    EffectFuture::Concurrent(
+        fetch_api::<E, _, _, SuccessResponse>(&api_request)
+            .map_err(CtxError::from)
+            .and_then(|result| match result {
+                APIResult::Ok { result } => future::ok(result),
+                APIResult::Err { error } => future::err(CtxError::from(error)),
+            })
+            .map(move |result| Msg::Internal(Internal::SeekLogsResult(seek_log_req, result)))
+            .boxed_env(),
+    )
+    .into()
+}
+
 #[cfg(test)]
 mod test {
     use chrono::{TimeZone, Utc};

src/models/streaming_server.rs

@@ -589,9 +589,10 @@ fn get_torrent_statistics<E: Env + 'static>(url: &Url, request: &StatisticsReque
         .expect("request builder failed");
     EffectFuture::Concurrent(
         E::fetch::<_, Statistics>(request)
-            .map(enclose!((url) move |result|
+            .map(enclose!((url) move |result| {
+                tracing::info!("result for statistics: {result:?}");
                 Msg::Internal(Internal::StreamingServerStatisticsResult((url, statistics_request), result))
-            ))
+            }))
             .boxed_env(),
     )
     .into()

src/runtime/msg/action.rs

@@ -138,6 +138,13 @@ pub enum ActionPlayer {
     PausedChanged {
         paused: bool,
     },
+    /// Play next video, if there is one, applicable to e.g.
+    /// movie series and playing the next episode.
+    NextVideo,
+    /// Video player has ended.
+    /// 2 scenarios are possible:
+    /// - We've watched a movie to the last second
+    /// - We've watched a movie series to the last episode and the last second
     Ended,
 }
 

src/runtime/msg/event.rs

@@ -20,6 +20,11 @@ pub enum Event {
     PlayerStopped {
         context: PlayerAnalyticsContext,
     },
+    PlayerNextVideo {
+        context: PlayerAnalyticsContext,
+        is_binge_enabled: bool,
+        is_playing_next_video: bool,
+    },
     PlayerEnded {
         context: PlayerAnalyticsContext,
         is_binge_enabled: bool,

src/runtime/msg/internal.rs

@@ -3,14 +3,15 @@ use url::Url;
 use crate::models::ctx::CtxError;
 use crate::models::link::LinkError;
 use crate::models::local_search::Searchable;
+use crate::models::player::SeekLog;
 use crate::models::streaming_server::{
     PlaybackDevice, Settings as StreamingServerSettings, StatisticsRequest,
 };
 use crate::runtime::EnvError;
 use crate::types::addon::{Descriptor, Manifest, ResourceRequest, ResourceResponse};
 use crate::types::api::{
     APIRequest, AuthRequest, DataExportResponse, DatastoreRequest, LinkCodeResponse,
-    LinkDataResponse,
+    LinkDataResponse, SuccessResponse, SeekLogRequest,
 };
 use crate::types::library::{LibraryBucket, LibraryItem, LibraryItemId};
 use crate::types::profile::{Auth, AuthKey, Profile, User};
@@ -96,6 +97,10 @@ pub enum Internal {
     NotificationsRequestResult(ResourceRequest, Box<Result<ResourceResponse, EnvError>>),
     /// Result for requesting a `dataExport` of user data.
     DataExportResult(AuthKey, Result<DataExportResponse, CtxError>),
+    /// Result for submitting SeekLogs request for a played stream.
+    ///
+    /// Applicable only to movie series and torrents.
+    SeekLogsResult(SeekLogRequest, Result<SuccessResponse, CtxError>),
     /// The result of querying the data for LocalSearch
     LoadLocalSearchResult(Url, Result<Vec<Searchable>, EnvError>),
 }

src/types/api/request.rs

@@ -1,7 +1,9 @@
 use crate::constants::{API_URL, LINK_API_URL};
+use crate::models::player::SeekLog;
 use crate::types::addon::Descriptor;
 use crate::types::library::LibraryItem;
 use crate::types::profile::{AuthKey, GDPRConsent, User};
+use crate::types::resource::SeriesInfo;
 #[cfg(test)]
 use derivative::Derivative;
 use http::Method;
@@ -53,6 +55,28 @@ pub enum APIRequest {
         auth_key: AuthKey,
         events: Vec<serde_json::Value>,
     },
+    #[serde(rename_all = "camelCase")]
+    SeekLog(SeekLogRequest),
+}
+
+#[derive(Clone, PartialEq, Eq, Serialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct SeekLogRequest {
+    /// Opensubtitles hash returned by the server
+    #[serde(alias = "osId")]
+    pub opensubtitles_hash: String,
+    pub item_id: String,
+    #[serde(flatten)]
+    pub series_info: SeriesInfo,
+    /// Filename hash
+    ///
+    /// base64 encoded SHA-256 hash of the Stream filename.
+    #[serde(alias = "stHash")]
+    pub filename_hash: String,
+    pub duration: u64,
+    pub seek_history: Vec<SeekLog>,
+    /// The time (in milliseconds) when the user decided to play the next video/episode
+    pub skip_outro: u64,
 }
 
 impl FetchRequestParams<APIRequest> for APIRequest {
@@ -74,6 +98,7 @@ impl FetchRequestParams<APIRequest> for APIRequest {
             APIRequest::SaveUser { .. } => "saveUser".to_owned(),
             APIRequest::DataExport { .. } => "dataExport".to_owned(),
             APIRequest::Events { .. } => "events".to_owned(),
+            APIRequest::SeekLog { .. } => "seekLog".to_owned(),
         }
     }
     fn query(&self) -> Option<String> {