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