/
/
/
1"""Helpers for Audiobookshelf provider."""
2
3import time
4from dataclasses import dataclass, field
5from typing import TYPE_CHECKING
6
7from mashumaro.mixins.dict import DataClassDictMixin
8
9if TYPE_CHECKING:
10 from aioaudiobookshelf.schema.media_progress import MediaProgress
11
12
13@dataclass(kw_only=True)
14class LibraryHelper(DataClassDictMixin):
15 """Lib name + media items' uuids."""
16
17 name: str
18 item_ids: set[str] = field(default_factory=set)
19
20
21@dataclass(kw_only=True)
22class NarratorHelper(DataClassDictMixin):
23 """Store narrator's name and id."""
24
25 id_: str
26 name: str
27
28 def __hash__(self) -> int:
29 """Hash."""
30 return hash(self.id_)
31
32
33@dataclass(kw_only=True)
34class LibrariesHelper(DataClassDictMixin):
35 """
36 Helper class to store ABSLibrary name, id and the uuids of its media items.
37
38 Dictionary is lib_id:LibraryHelper or lib_id:set[playlist_ids/narrator_ids/author_ids].
39 """
40
41 audiobooks: dict[str, LibraryHelper] = field(default_factory=dict)
42 podcasts: dict[str, LibraryHelper] = field(default_factory=dict)
43 playlists_audiobooks: dict[str, set[str]] = field(default_factory=dict)
44 playlists_podcasts: dict[str, set[str]] = field(default_factory=dict)
45 authors: dict[str, set[str]] = field(default_factory=dict)
46 narrators: dict[str, set[str]] = field(default_factory=dict)
47 # audiobook_id is key. Abs does not have a dedicated narrator endpoint.
48 audiobook_narrators: dict[str, set[NarratorHelper]] = field(default_factory=dict)
49
50
51@dataclass(kw_only=True)
52class SessionHelper:
53 """Helper class to store some session information."""
54
55 abs_session_id: str
56 last_sync_time: float
57 failed_sync_count: int = 0
58
59
60@dataclass(kw_only=True)
61class _ProgressHelper:
62 id_: str # audiobook or podcast id
63 episode_id: str | None = None
64 last_update_ms: int # last update in ms epoch (same as last_update in abs)
65
66
67class ProgressGuard:
68 """
69 Class used to avoid ping pong between abs and mass.
70
71 We continuously update the progress from mass to abs with the provider's on_played function.
72 We also register callbacks for progress reports from abs to mass. This is not only triggered
73 on external updates, but also on our own update. To avoid messages going back and forth, this
74 class is used.
75 """
76
77 def __init__(self) -> None:
78 """Init."""
79 self._progresses: list[_ProgressHelper] = []
80 self._max_progresses = 100
81 # 8s have to have passed before we accept an external progress update
82 # abs updates every 10 s
83 self._min_time_between_updates_ms = 8000
84
85 def _get_progress(self, item_id: str, episode_id: str | None = None) -> _ProgressHelper | None:
86 """Get a helper progress."""
87 for x in self._progresses:
88 if x.id_ == item_id and x.episode_id == episode_id:
89 return x
90 return None
91
92 def _remove_oldest(self) -> None:
93 """Remove oldest helper progress."""
94 progresses = sorted(self._progresses, key=lambda x: x.last_update_ms)
95 if len(progresses) > 0:
96 self._progresses.remove(progresses[0])
97
98 def remove_progress(self, item_id: str, episode_id: str | None = None) -> None:
99 """Remove a helper progress."""
100 progress = self._get_progress(item_id=item_id, episode_id=episode_id)
101 if progress is not None:
102 self._progresses.remove(progress)
103
104 def add_progress(self, item_id: str, episode_id: str | None = None) -> None:
105 """Store a timestamp for the last update of an audiobook or podcast episode, mass ids."""
106 if len(self._progresses) > self._max_progresses:
107 self._remove_oldest()
108 self.remove_progress(item_id=item_id, episode_id=episode_id)
109 progress = _ProgressHelper(
110 id_=item_id, episode_id=episode_id, last_update_ms=int(time.time() * 1000)
111 )
112 self._progresses.append(progress)
113
114 def guard_ok_abs(self, abs_progress: MediaProgress) -> bool:
115 """
116 Check, if we may update against an abs media progress.
117
118 The abs media progress has a property last_update_ms, which also reflects non
119 mass external updates. Here, we compare this property against a potential
120 stored one.
121 """
122 return self.guard_ok_mass(abs_progress.library_item_id, abs_progress.episode_id)
123
124 def guard_ok_mass(self, item_id: str, episode_id: str | None = None) -> bool:
125 """
126 Check, if we may update against a mass internal item.
127
128 Here, we use the current time and compare it against the stored time.
129 """
130 stored_progress = self._get_progress(item_id=item_id, episode_id=episode_id)
131 if stored_progress is None:
132 return True
133 return (
134 int(time.time() * 1000) - stored_progress.last_update_ms
135 >= self._min_time_between_updates_ms
136 )
137