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

1""" 

2Typing stuff 

3""" 

4 

5from __future__ import annotations 

6 

7import contextlib 

8import re 

9import warnings 

10from collections.abc import AsyncIterator, Iterable 

11from dataclasses import dataclass, fields 

12from datetime import datetime 

13from enum import IntEnum 

14from functools import cached_property 

15from pathlib import Path 

16from typing import ( 

17 IO, 

18 TYPE_CHECKING, 

19 Any, 

20 Literal, 

21 NamedTuple, 

22 Protocol, 

23 TypedDict, 

24 TypeVar, 

25 Union, 

26 runtime_checkable, 

27) 

28 

29import aiohttp 

30from slixmpp import Message, Presence 

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

32 

33if TYPE_CHECKING: 

34 from ..contact import LegacyContact, LegacyRoster 

35 from ..core.gateway import BaseGateway 

36 from ..core.session import BaseSession 

37 from ..group import LegacyBookmarks, LegacyMUC 

38 from ..group.participant import LegacyParticipant 

39 

40type AnySession = "BaseSession" 

41type AnyGateway = "BaseGateway" 

42type AnyMUC = "LegacyMUC[Any]" 

43type AnyBookmarks = "LegacyBookmarks[Any]" 

44type AnyRoster = "LegacyRoster[Any]" 

45type AnyParticipant = "LegacyParticipant[Any]" 

46 

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

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

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

50# Covariant with the bound as default, so that a bare `BaseSession` reference is typed 

51# `BaseSession[LegacyRoster[Any], LegacyBookmarks[Any]]`` instead of `BaseSession[Any]`. 

52LegacyRosterType_co = TypeVar( 

53 "LegacyRosterType_co", bound=AnyRoster, default=AnyRoster, covariant=True 

54) 

55LegacyBookmarksType_co = TypeVar( 

56 "LegacyBookmarksType_co", bound=AnyBookmarks, default=AnyBookmarks, covariant=True 

57) 

58SessionType_co = TypeVar( 

59 "SessionType_co", bound=AnySession, default=AnySession, covariant=True 

60) 

61 

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

63AnyRecipient = Union["LegacyContact", AnyMUC] 

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

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

66 

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

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

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

70FieldType = Literal[ 

71 "boolean", 

72 "fixed", 

73 "text-single", 

74 "text-multi", 

75 "jid-single", 

76 "jid-multi", 

77 "list-single", 

78 "list-multi", 

79 "text-private", 

80] 

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

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

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

84ClientType = Literal[ 

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

86] 

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

88 

89 

90@dataclass 

91class MessageReference: 

92 """ 

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

94 

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

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

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

98 message). 

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

100 of that failed to find the referenced message. 

101 """ 

102 

103 legacy_id: str 

104 author: Literal["user"] | AnyParticipant | LegacyContact | None = None 

105 body: str | None = None 

106 

107 

108@dataclass 

109class LegacyAttachment: 

110 """ 

111 A file attachment to a message 

112 

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

114 has to be set 

115 

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

117 :meth:`.LegacyParticipant.send_files` 

118 """ 

119 

120 path: Path | str | None = None 

121 name: str | None = None 

122 stream: IO[bytes] | None = None 

123 aio_stream: AsyncIterator[bytes] | None = None 

124 data: bytes | None = None 

125 content_type: str | None = None 

126 legacy_file_id: str | None = None 

127 url: str | None = None 

128 caption: str | None = None 

129 """ 

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

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

132 """ 

133 disposition: AttachmentDisposition | None = None 

134 is_sticker: bool = False 

135 size: int | None = None 

136 

137 def __post_init__(self) -> None: 

138 if all( 

139 x is None 

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

141 ): 

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

143 

144 if isinstance(self.path, str): 

145 self.path = Path(self.path) 

146 

147 if self.is_sticker: 

148 if self.disposition == "attachment": 

149 warnings.warn( 

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

151 ) 

152 self.disposition = "inline" 

153 

154 def format_for_user(self) -> str: 

155 if self.name: 

156 name = self.name 

157 elif self.path: 

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

159 elif self.url: 

160 name = self.url 

161 else: 

162 name = "" 

163 

164 if self.caption: 

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

166 

167 return name 

168 

169 def __str__(self) -> str: 

170 attrs = ", ".join( 

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

172 for f in fields(self) 

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

174 ) 

175 if self.data is not None: 

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

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

178 attrs = ", ".join(to_join) 

179 return f"Attachment({attrs})" 

180 

181 

182class MucType(IntEnum): 

183 """ 

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

185 """ 

186 

187 GROUP = 0 

188 """ 

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

190 """ 

191 CHANNEL = 1 

192 """ 

193 A public group, aka an anonymous channel. 

194 """ 

195 CHANNEL_NON_ANONYMOUS = 2 

196 """ 

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

198 """ 

199 

200 

201PseudoPresenceShow = PresenceShows | Literal[""] 

202 

203 

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

205 

206 

207class LinkPreview(NamedTuple): 

208 """ 

209 Embedded metadata from :xep:`0511`. 

210 

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

212 """ 

213 

214 about: str 

215 """ 

216 URL of the link. 

217 """ 

218 title: str | None 

219 """ 

220 Title of the linked page. 

221 """ 

222 description: str | None 

223 """ 

224 A description of the page. 

225 """ 

226 url: str | None 

227 """ 

228 The canonical URL of the link. 

229 """ 

230 image: str | Path | bytes | None 

231 """ 

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

233 """ 

234 type: str | None 

235 """ 

236 Type of the link destination. 

237 """ 

238 site_name: str | None 

239 """ 

240 Name of the web site. 

241 """ 

242 

243 @property 

244 def is_empty(self) -> bool: 

245 return not any(x for x in self) 

246 

247 

248class Mention[LegacyParticipantType: AnyParticipant](NamedTuple): 

249 participant: LegacyParticipantType 

250 start: int 

251 end: int 

252 

253 

254class Hat(NamedTuple): 

255 uri: str 

256 title: str 

257 hue: float | None = None 

258 

259 

260class UserPreferences(TypedDict): 

261 sync_avatar: bool 

262 sync_presence: bool 

263 

264 

265class MamMetadata(NamedTuple): 

266 id: str 

267 sent_on: datetime 

268 

269 

270class HoleBound(NamedTuple): 

271 id: str 

272 timestamp: datetime 

273 

274 

275class CachedPresence(NamedTuple): 

276 last_seen: datetime | None = None 

277 ptype: PresenceTypes | None = None 

278 pstatus: str | None = None 

279 pshow: PresenceShows | None = None 

280 

281 

282class Avatar(NamedTuple): 

283 path: Path | None = None 

284 unique_id: str | None = None 

285 url: str | None = None 

286 data: bytes | None = None 

287 

288 

289class SpaceMetadata(NamedTuple): 

290 creator_legacy_id: str | None = None 

291 name: str | None = None 

292 description: str | None = None 

293 member_count: int | None = None 

294 owner_legacy_ids: Iterable[str] = [] 

295 

296 

297@dataclass 

298class Reply: 

299 """ 

300 Represents a message referenced (replied to) via :xep:`0461` 

301 """ 

302 

303 msg_id: str 

304 """ 

305 The ID of the message being replied to. 

306 """ 

307 fallback: str | None 

308 """ 

309 A fallback (:xep:`0428`) text for clients not supporting :xep:`0461`, 

310 usually consisting of the text of the referenced message. 

311 NB: some XMPP clients might not fill this, so it can be empty. 

312 """ 

313 to: Literal["self", "contact"] | LegacyParticipant[Any] = "contact" 

314 """ 

315 Author of the referenced message. 

316 """ 

317 

318 @cached_property 

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

320 """ 

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

322 """ 

323 if not self.fallback: 

324 return None 

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

326 

327 

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

329 

330 

331class _ReplyProtocol(Protocol): 

332 msg_id: str 

333 fallback: str | None 

334 to: Literal["self", "contact"] | LegacyParticipant[Any] 

335 

336 

337@runtime_checkable 

338class MUCReplyProtocol(_ReplyProtocol, Protocol): 

339 to: LegacyParticipant[Any] 

340 

341 

342@runtime_checkable 

343class ContactReplyProtocol(_ReplyProtocol, Protocol): 

344 to: Literal["self", "contact"] 

345 

346 

347@dataclass 

348class XMPPAttachment: 

349 url: str 

350 is_sticker: bool = False 

351 cid: str | None = None 

352 content_type: str | None = None 

353 

354 @contextlib.asynccontextmanager 

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

356 async with ( 

357 aiohttp.ClientSession() as session, 

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

359 ): 

360 yield response 

361 

362 

363@dataclass 

364class XMPPMessage: 

365 body: str | None = None 

366 """ 

367 Text content of the message. Can be empty if there are attachments. 

368 """ 

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

370 """ 

371 Attachments to this message (:xep:`0066`, :xep:`0385` or :xep:`0447`). 

372 Is never empty if there is no body. 

373 """ 

374 reply: Reply | None = None 

375 """ 

376 A reference to message being replied-to (:xep:`0461`), if applicable. 

377 """ 

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

379 """ 

380 A list of link metadata (:xep:`0511`), attached to this message. 

381 """ 

382 replace: str | None = None 

383 """ 

384 The message this message is a correction for (:xep:`0308`). 

385 """ 

386 thread: str | None = None 

387 """ 

388 The thread this message is part of. 

389 """ 

390 mentions: tuple[Mention[Any], ...] = () 

391 """ 

392 A list of mentions parsed in the body of this message (by searching for 

393 nicknames of MUC participants). 

394 Is always empty for 1:1 messages. 

395 """ 

396 

397 

398class XMPPMessageProtocol(Protocol): 

399 body: str | None 

400 attachments: tuple[XMPPAttachment, ...] 

401 reply: _ReplyProtocol | None 

402 link_previews: tuple[LinkPreview, ...] 

403 replace: str | None 

404 thread: str | None 

405 

406 

407@runtime_checkable 

408class ContactMessageProtocol(XMPPMessageProtocol, Protocol): 

409 reply: ContactReplyProtocol | None 

410 

411 

412@runtime_checkable 

413class MUCMessageProtocol[LegacyParticipantType: AnyParticipant]( 

414 XMPPMessageProtocol, Protocol 

415): 

416 reply: MUCReplyProtocol | None 

417 mentions: tuple[Mention[LegacyParticipantType], ...] 

418 

419 

420class AttachmentMessageProtocol(XMPPMessageProtocol): 

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

422 

423 

424class XMPPTextMessageProtocol(XMPPMessageProtocol): 

425 body: str 

426 attachments: tuple[()] 

427 

428 

429class ContactAttachmentMessage(ContactMessageProtocol, AttachmentMessageProtocol): 

430 pass 

431 

432 

433class MUCAttachmentMessage[LegacyParticipantType: AnyParticipant]( 

434 MUCMessageProtocol[LegacyParticipantType], 

435 AttachmentMessageProtocol, 

436): 

437 pass 

438 

439 

440class ContactTextMessage(ContactMessageProtocol, XMPPTextMessageProtocol): 

441 pass 

442 

443 

444class MUCTextMessage[LegacyParticipantType: AnyParticipant]( 

445 MUCMessageProtocol[LegacyParticipantType], 

446 XMPPTextMessageProtocol, 

447): 

448 pass 

449 

450 

451MUCMessage = ( 

452 MUCAttachmentMessage[LegacyParticipantType] | MUCTextMessage[LegacyParticipantType] 

453) 

454ContactMessage = ContactAttachmentMessage | ContactTextMessage 

455 

456 

457@dataclass 

458class Sticker: 

459 path: Path 

460 content_type: str | None 

461 hashes: dict[str, str] 

462 fallback: str | None = None 

463 reply: Reply | None = None 

464 thread: str | None = None 

465 

466 

467class StickerProtocol(Protocol): 

468 path: Path 

469 content_type: str | None 

470 hashes: dict[str, str] 

471 fallback: str | None 

472 reply: _ReplyProtocol | None 

473 thread: str | None 

474 

475 

476@runtime_checkable 

477class ContactSticker(StickerProtocol, Protocol): 

478 reply: ContactReplyProtocol | None 

479 

480 

481@runtime_checkable 

482class MUCSticker(StickerProtocol, Protocol): 

483 reply: MUCReplyProtocol | None