Coverage for slidge/util/types.py: 96%

1""" 

2Typing stuff 

3""" 

4 

5import contextlib 

6import re 

7import warnings 

8from collections.abc import AsyncIterator, Iterable 

9from dataclasses import dataclass, fields 

10from datetime import datetime 

11from enum import IntEnum 

12from functools import cached_property 

13from pathlib import Path 

14from typing import ( 

15 IO, 

16 TYPE_CHECKING, 

17 Any, 

18 Literal, 

19 NamedTuple, 

20 TypeAlias, 

21 TypedDict, 

22 TypeVar, 

23 Union, 

24) 

25 

26import aiohttp 

27from slixmpp import Message, Presence 

28from slixmpp.types import PresenceShows, PresenceTypes, ResourceDict # noqa: F401 

29 

30if TYPE_CHECKING: 

31 from ..contact import LegacyContact, LegacyRoster 

32 from ..core.gateway import BaseGateway 

33 from ..core.session import BaseSession 

34 from ..group import LegacyBookmarks, LegacyMUC 

35 from ..group.participant import LegacyParticipant 

36 

37AnySession: TypeAlias = "BaseSession[Any]" 

38AnyGateway: TypeAlias = "BaseGateway[AnySession]" 

39AnyMUC: TypeAlias = "LegacyMUC[Any]" 

40AnyBookmarks: TypeAlias = "LegacyBookmarks[Any]" 

41AnyRoster: TypeAlias = "LegacyRoster[Any]" 

42AnyParticipant: TypeAlias = "LegacyParticipant[Any]" 

43 

44LegacyContactType = TypeVar("LegacyContactType", bound="LegacyContact") 

45LegacyMUCType = TypeVar("LegacyMUCType", bound=AnyMUC) 

46LegacyParticipantType = TypeVar("LegacyParticipantType", bound=AnyParticipant) 

47 

48SessionType = TypeVar("SessionType", bound=AnySession) 

49AnyRecipient = Union["LegacyContact", AnyMUC] 

50RecipientType = TypeVar("RecipientType", bound=AnyRecipient) 

51Sender = Union["LegacyContact", "AnyParticipant"] 

52 

53ChatState = Literal["active", "composing", "gone", "inactive", "paused"] 

54ProcessingHint = Literal["no-store", "markable", "store"] 

55Marker = Literal["acknowledged", "received", "displayed"] 

56FieldType = Literal[ 

57 "boolean", 

58 "fixed", 

59 "text-single", 

60 "text-multi", 

61 "jid-single", 

62 "jid-multi", 

63 "list-single", 

64 "list-multi", 

65 "text-private", 

66] 

67MucAffiliation = Literal["owner", "admin", "member", "outcast", "none"] 

68MucRole = Literal["visitor", "participant", "moderator", "none"] 

69# https://xmpp.org/registrar/disco-categories.html#client 

70ClientType = Literal[ 

71 "bot", "console", "game", "handheld", "pc", "phone", "sms", "tablet", "web" 

72] 

73AttachmentDisposition = Literal["attachment", "inline"] 

74 

75 

76@dataclass 

77class MessageReference: 

78 """ 

79 A "message reply", ie a "quoted message" (:xep:`0461`) 

80 

81 At the very minimum, the legacy message ID attribute must be set, but to 

82 ensure that the quote is displayed in all XMPP clients, the author must also 

83 be set (use the string "user" if the slidge user is the author of the referenced 

84 message). 

85 The body is used as a fallback for XMPP clients that do not support :xep:`0461` 

86 of that failed to find the referenced message. 

87 """ 

88 

89 legacy_id: str 

90 author: Union[Literal["user"], AnyParticipant, "LegacyContact"] | None = None 

91 body: str | None = None 

92 

93 

94@dataclass 

95class LegacyAttachment: 

96 """ 

97 A file attachment to a message 

98 

99 At the minimum, one of the ``path``, ``steam``, ``data`` or ``url`` attribute 

100 has to be set 

101 

102 To be used with :meth:`.LegacyContact.send_files` or 

103 :meth:`.LegacyParticipant.send_files` 

104 """ 

105 

106 path: Path | str | None = None 

107 name: str | None = None 

108 stream: IO[bytes] | None = None 

109 aio_stream: AsyncIterator[bytes] | None = None 

110 data: bytes | None = None 

111 content_type: str | None = None 

112 legacy_file_id: str | None = None 

113 url: str | None = None 

114 caption: str | None = None 

115 """ 

116 A caption for this specific image. For a global caption for a list of attachments, 

117 use the ``body`` parameter of :meth:`.AttachmentMixin.send_files` 

118 """ 

119 disposition: AttachmentDisposition | None = None 

120 is_sticker: bool = False 

121 size: int | None = None 

122 

123 def __post_init__(self) -> None: 

124 if all( 

125 x is None 

126 for x in (self.path, self.stream, self.data, self.url, self.aio_stream) 

127 ): 

128 raise TypeError("There is not data in this attachment", self) 

129 

130 if isinstance(self.path, str): 

131 self.path = Path(self.path) 

132 

133 if self.is_sticker: 

134 if self.disposition == "attachment": 

135 warnings.warn( 

136 "Sticker declared as 'attachment' disposition, changing it to 'inline'" 

137 ) 

138 self.disposition = "inline" 

139 

140 def format_for_user(self) -> str: 

141 if self.name: 

142 name = self.name 

143 elif self.path: 

144 name = self.path.name # type:ignore[union-attr] 

145 elif self.url: 

146 name = self.url 

147 else: 

148 name = "" 

149 

150 if self.caption: 

151 name = f"{name}: {self.caption}" if name else self.caption 

152 

153 return name 

154 

155 def __str__(self) -> str: 

156 attrs = ", ".join( 

157 f"{f.name}={getattr(self, f.name)!r}" 

158 for f in fields(self) 

159 if getattr(self, f.name) is not None and f.name != "data" 

160 ) 

161 if self.data is not None: 

162 data_str = f"data=<{len(self.data)} bytes>" 

163 to_join = (attrs, data_str) if attrs else (data_str,) 

164 attrs = ", ".join(to_join) 

165 return f"Attachment({attrs})" 

166 

167 

168class MucType(IntEnum): 

169 """ 

170 The type of group, private, public, anonymous or not. 

171 """ 

172 

173 GROUP = 0 

174 """ 

175 A private group, members-only and non-anonymous, eg a family group. 

176 """ 

177 CHANNEL = 1 

178 """ 

179 A public group, aka an anonymous channel. 

180 """ 

181 CHANNEL_NON_ANONYMOUS = 2 

182 """ 

183 A public group where participants' legacy IDs are visible to everybody. 

184 """ 

185 

186 

187PseudoPresenceShow = PresenceShows | Literal[""] 

188 

189 

190MessageOrPresenceTypeVar = TypeVar("MessageOrPresenceTypeVar", bound=Message | Presence) 

191 

192 

193class LinkPreview(NamedTuple): 

194 """ 

195 Embedded metadata from :xep:`0511`. 

196 

197 See <https://ogp.me/>_. 

198 """ 

199 

200 about: str 

201 """ 

202 URL of the link. 

203 """ 

204 title: str | None 

205 """ 

206 Title of the linked page. 

207 """ 

208 description: str | None 

209 """ 

210 A description of the page. 

211 """ 

212 url: str | None 

213 """ 

214 The canonical URL of the link. 

215 """ 

216 image: str | Path | bytes | None 

217 """ 

218 An image representing the link. If it is a string, it should represent a URL to an image. 

219 """ 

220 type: str | None 

221 """ 

222 Type of the link destination. 

223 """ 

224 site_name: str | None 

225 """ 

226 Name of the web site. 

227 """ 

228 

229 @property 

230 def is_empty(self) -> bool: 

231 return not any(x for x in self) 

232 

233 

234class Mention(NamedTuple): 

235 contact: "LegacyContact" 

236 start: int 

237 end: int 

238 

239 

240class Hat(NamedTuple): 

241 uri: str 

242 title: str 

243 hue: float | None = None 

244 

245 

246class UserPreferences(TypedDict): 

247 sync_avatar: bool 

248 sync_presence: bool 

249 

250 

251class MamMetadata(NamedTuple): 

252 id: str 

253 sent_on: datetime 

254 

255 

256class HoleBound(NamedTuple): 

257 id: str 

258 timestamp: datetime 

259 

260 

261class CachedPresence(NamedTuple): 

262 last_seen: datetime | None = None 

263 ptype: PresenceTypes | None = None 

264 pstatus: str | None = None 

265 pshow: PresenceShows | None = None 

266 

267 

268class Avatar(NamedTuple): 

269 path: Path | None = None 

270 unique_id: str | None = None 

271 url: str | None = None 

272 data: bytes | None = None 

273 

274 

275class SpaceMetadata(NamedTuple): 

276 creator_legacy_id: str | None = None 

277 name: str | None = None 

278 description: str | None = None 

279 member_count: int | None = None 

280 owner_legacy_ids: Iterable[str] = [] 

281 

282 

283@dataclass 

284class Reply: 

285 msg_id: str 

286 to: "LegacyContact | LegacyParticipant[Any] | Literal['user'] | None" 

287 fallback: str | None = None 

288 

289 @cached_property 

290 def fallback_no_quote_mark(self) -> str | None: 

291 """ 

292 Return multi-line text without leading quote marks (i.e. the ">" character). 

293 """ 

294 if not self.fallback: 

295 return None 

296 return re.sub(_STRIP_QUOTE_RE, "", self.fallback).strip() 

297 

298 

299_STRIP_QUOTE_RE = re.compile(r"^>\s*", flags=re.MULTILINE) 

300 

301 

302@dataclass 

303class XMPPAttachment: 

304 url: str 

305 is_sticker: bool = False 

306 cid: str | None = None 

307 content_type: str | None = None 

308 

309 @contextlib.asynccontextmanager 

310 async def get(self) -> AsyncIterator[aiohttp.ClientResponse]: 

311 async with ( 

312 aiohttp.ClientSession() as session, 

313 session.get(self.url) as response, 

314 ): 

315 yield response 

316 

317 

318@dataclass 

319class _AbstractXMPPMessage: 

320 body: str | None = None 

321 link_previews: tuple[LinkPreview, ...] = () 

322 attachments: tuple[XMPPAttachment, ...] = () 

323 mentions: tuple[Mention, ...] = () 

324 replace: str | None = None 

325 reply: Reply | None = None 

326 thread: str | None = None 

327 

328 

329class XMPPAttachmentMessage(_AbstractXMPPMessage): 

330 body: str | None 

331 attachments: tuple[XMPPAttachment, *tuple[XMPPAttachment, ...]] 

332 

333 

334class XMPPTextMessage(_AbstractXMPPMessage): 

335 body: str 

336 attachments: tuple[()] 

337 

338 

339XMPPMessage = XMPPAttachmentMessage | XMPPTextMessage 

340 

341 

342@dataclass 

343class Sticker: 

344 path: Path 

345 content_type: str | None 

346 hashes: dict[str, str] 

347 fallback: str | None = None 

348 reply: Reply | None = None 

349 thread: str | None = None