Coverage for slidge/core/mixins/presence.py: 90%

1import contextlib 

2import re 

3from asyncio import Task, sleep 

4from datetime import UTC, datetime, timedelta 

5from functools import partial 

6from typing import Literal 

7 

8from slixmpp import Presence 

9from slixmpp.types import PresenceShows, PresenceTypes 

10 

11from ...db.models import Contact, Participant 

12from ...util.types import AnySession 

13from .base import BaseSender 

14from .db import UpdateInfoMixin 

15 

16_UPDATE_LAST_SEEN_FALLBACK_TASKS = dict[int, Task[None]]() 

17_ONE_WEEK_SECONDS = 3600 * 24 * 7 

18 

19 

20async def _update_last_seen_fallback(session: AnySession, contact_pk: int) -> None: 

21 await sleep(_ONE_WEEK_SECONDS) 

22 with session.xmpp.store.session() as orm: 

23 stored = orm.get(Contact, contact_pk) 

24 if stored is None: 

25 return 

26 contact = session.contacts._from_store(stored) 

27 contact.send_last_presence(force=True) 

28 

29 

30def _clear_last_seen_task(contact_pk: int, _task: Task[None]) -> None: 

31 with contextlib.suppress(KeyError): 

32 del _UPDATE_LAST_SEEN_FALLBACK_TASKS[contact_pk] 

33 

34 

35class PresenceMixin(BaseSender, UpdateInfoMixin): 

36 _stored: Contact | Participant 

37 __batch_set_presence = False 

38 

39 @property 

40 def online(self) -> bool: 

41 return self.__stored.online 

42 

43 @online.setter 

44 def online(self, online: bool) -> None: 

45 

46 if self._update_stored_attribute(online=online): 

47 self._send_presence() 

48 

49 @property 

50 def pshow(self) -> PresenceShows | None: 

51 return self.__stored.pshow 

52 

53 @pshow.setter 

54 def pshow(self, pshow: PresenceShows | Literal["available"] | None) -> None: 

55 if pshow == "available": 

56 pshow = None 

57 self.online = True 

58 if self._update_stored_attribute(pshow=pshow): 

59 self._send_presence() 

60 

61 @property 

62 def pstatus(self) -> str | None: 

63 return self.__stored.pstatus 

64 

65 @pstatus.setter 

66 def pstatus(self, pstatus: str | None) -> None: 

67 if self._update_stored_attribute(pstatus=pstatus): 

68 self._send_presence() 

69 

70 @property 

71 def last_seen(self) -> datetime | None: 

72 return self.__stored.last_seen 

73 

74 @last_seen.setter 

75 def last_seen(self, last_seen: datetime | None) -> None: 

76 if self._update_stored_attribute(last_seen=last_seen): 

77 self._send_presence() 

78 

79 def set_presence( 

80 self, 

81 online: bool = True, 

82 show: PresenceShows | None = None, 

83 last_seen: datetime | None = None, 

84 status: str | None = None, 

85 ) -> None: 

86 self.__batch_set_presence = True 

87 with self.batch_set_attributes(): 

88 self.online = online 

89 self.pshow = show 

90 self.last_seen = last_seen 

91 self.pstatus = status 

92 self.__batch_set_presence = False 

93 self._send_presence() 

94 

95 def _send_presence(self) -> None: 

96 if self.__batch_set_presence: 

97 return 

98 self._send( 

99 self._make_presence( 

100 pstatus=self.pstatus, 

101 last_seen=self.last_seen, 

102 ptype=None if self.online else "unavailable", 

103 pshow=self.pshow, 

104 ) 

105 ) 

106 

107 @property 

108 def __has_cache(self) -> bool: 

109 self._stored.ensure_loaded("presence", self.xmpp.store.session) 

110 return any( 

111 x is not None 

112 for x in (self._stored.pshow, self._stored.pstatus, self._stored.last_seen) 

113 ) 

114 

115 @property 

116 def __stored(self) -> Contact | Participant: 

117 if isinstance(self._stored, Contact): 

118 self._stored.ensure_loaded("presence", self.xmpp.store.session) 

119 return self._stored 

120 elif isinstance(self._stored, Participant): 

121 if self.__has_cache: 

122 return self._stored 

123 self._stored.ensure_relation("contact", self.xmpp.store.session) 

124 if self._stored.contact is None: 

125 return self._stored 

126 else: 

127 self._stored.contact.ensure_loaded("presence", self.xmpp.store.session) 

128 return self._stored.contact 

129 

130 @property 

131 def __contact_pk(self) -> int | None: 

132 return self._stored.id if isinstance(self._stored, Contact) else None 

133 

134 def _make_presence( 

135 self, 

136 *, 

137 last_seen: datetime | None = None, 

138 bare: bool = False, 

139 ptype: PresenceTypes | None = None, 

140 pstatus: str | None = None, 

141 pshow: PresenceShows | None = None, 

142 ) -> Presence: 

143 if last_seen and last_seen.tzinfo is None: 

144 last_seen = last_seen.astimezone(UTC) 

145 

146 p = self.xmpp.make_presence( 

147 pfrom=self.jid.bare if bare else self.jid, 

148 ptype=ptype, 

149 pshow=pshow, 

150 pstatus=pstatus, 

151 ) 

152 if last_seen: 

153 # it's ugly to check for the presence of this string, but a better fix is more work 

154 if not re.match( 

155 ".*Last seen .*", p["status"] 

156 ) and self._session.preferences.get("last_seen_fallback", True): 

157 last_seen_fallback, recent = get_last_seen_fallback(last_seen) 

158 if p["status"]: 

159 p["status"] = p["status"] + " -- " + last_seen_fallback 

160 else: 

161 p["status"] = last_seen_fallback 

162 pk = self.__contact_pk 

163 if recent and pk is not None: 

164 # if less than a week, we use sth like 'Last seen: Monday, 8:05", 

165 # but if lasts more than a week, this is not very informative, so 

166 # we need to force resend an updated presence status 

167 task = _UPDATE_LAST_SEEN_FALLBACK_TASKS.get(pk) 

168 if task is not None: 

169 task.cancel() 

170 task = self._session.create_task( 

171 _update_last_seen_fallback(self._session, pk), 

172 name=f"update last seen fallback of {self}", 

173 ) 

174 _UPDATE_LAST_SEEN_FALLBACK_TASKS[pk] = task 

175 task.add_done_callback(partial(_clear_last_seen_task, pk)) 

176 p["idle"]["since"] = last_seen 

177 return p 

178 

179 def send_last_presence(self, force: bool = False) -> None: 

180 if self.__has_cache: 

181 if force: 

182 self._send(self._make_presence(ptype=None)) 

183 else: 

184 self._send(self._make_last_presence()) 

185 

186 def _make_last_presence(self) -> Presence: 

187 stored = self.__stored 

188 return self._make_presence( 

189 last_seen=stored.last_seen, 

190 ptype=None if stored.online else "unavailable", 

191 pshow=stored.pshow, 

192 pstatus=stored.pstatus, 

193 ) 

194 

195 

196def get_last_seen_fallback(last_seen: datetime) -> tuple[str, bool]: 

197 now = datetime.now(tz=UTC) 

198 if now - last_seen < timedelta(days=7): 

199 return f"Last seen {last_seen:%A %H:%M %p GMT}", True 

200 else: 

201 return f"Last seen {last_seen:%b %-d %Y}", False