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, Awaitable, Callable, Mapping 

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 TypeIs, 

24 TypeVar, 

25 Union, 

26 runtime_checkable, 

27) 

28 

29import aiohttp 

30from slixmpp import JID, Message, Presence 

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

32from typing_extensions import TypedDict 

33 

34if TYPE_CHECKING: 

35 from ..contact import LegacyContact, LegacyRoster 

36 from ..core.gateway import BaseGateway 

37 from ..core.session import BaseSession 

38 from ..room import LegacyBookmarks, LegacyRoom 

39 from ..room.participant import LegacyParticipant 

40 from ..space import LegacySpace, LegacySpaceNodes 

41 

42type AnySession = "BaseSession[Any, Any]" 

43type AnyGateway = "BaseGateway[Any, Any, Any]" 

44type AnyRoom = "LegacyRoom[Any]" 

45type AnySpace = "LegacySpace[Any]" 

46type AnyBookmarks = "LegacyBookmarks[Any]" 

47 

48type AnySpaceBookmarks = "LegacySpaceNodes[Any]" 

49type AnyRoster = "LegacyRoster[Any]" 

50type AnyParticipant = "LegacyParticipant[Any]" 

51 

52 

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

54LegacyRoomType = TypeVar("LegacyRoomType", bound=AnyRoom) 

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

56 

57 

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

59AnyRecipient = Union["LegacyContact", AnyRoom] 

60Sender = Union["LegacyContact[Any]", "AnyParticipant"] 

61 

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

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

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

65FieldType = Literal[ 

66 "boolean", 

67 "fixed", 

68 "text-single", 

69 "text-multi", 

70 "jid-single", 

71 "jid-multi", 

72 "list-single", 

73 "list-multi", 

74 "text-private", 

75] 

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

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

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

79ClientType = Literal[ 

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

81] 

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

83 

84 

85@dataclass 

86class MessageReference: 

87 """ 

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

89 

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

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

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

93 message). 

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

95 of that failed to find the referenced message. 

96 """ 

97 

98 legacy_id: str 

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

100 body: str | None = None 

101 

102 

103@dataclass 

104class LegacyAttachment: 

105 """ 

106 A file attachment to a message 

107 

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

109 has to be set 

110 

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

112 :meth:`.LegacyParticipant.send_files` 

113 """ 

114 

115 path: Path | str | None = None 

116 name: str | None = None 

117 stream: IO[bytes] | None = None 

118 aio_stream: AsyncIterator[bytes] | None = None 

119 data: bytes | None = None 

120 content_type: str | None = None 

121 legacy_file_id: str | None = None 

122 url: str | None = None 

123 caption: str | None = None 

124 """ 

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

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

127 """ 

128 disposition: AttachmentDisposition | None = None 

129 is_sticker: bool = False 

130 size: int | None = None 

131 date: datetime | None = None 

132 sha256: str | None = None 

133 """ 

134 Base64-encoded sha256 hash of the file. NB: not the hexdigest. 

135 """ 

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 RoomType(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, total=False): 

261 sync_avatar: bool 

262 sync_presence: bool 

263 bookmarks_push: bool 

264 bookmarks_push_public: bool 

265 autojoin_public: bool 

266 last_seen_fallback: bool 

267 roster_push: bool 

268 reaction_fallback: bool 

269 

270 

271class MamMetadata(NamedTuple): 

272 id: str 

273 sent_on: datetime 

274 

275 

276class HoleBound(NamedTuple): 

277 id: str 

278 timestamp: datetime 

279 

280 

281class Avatar(NamedTuple): 

282 path: Path | None = None 

283 unique_id: str | None = None 

284 url: str | None = None 

285 data: bytes | None = None 

286 

287 

288class AvatarMetadata(NamedTuple): 

289 id: str 

290 type: str 

291 bytes: int 

292 url: str 

293 height: int 

294 width: int 

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 RoomReplyProtocol(_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 http: aiohttp.ClientSession | None = None 

351 is_sticker: bool = False 

352 cid: str | None = None 

353 content_type: str | None = None 

354 

355 @contextlib.asynccontextmanager 

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

357 http_context = ( 

358 aiohttp.ClientSession() 

359 if self.http is None 

360 else contextlib.nullcontext(self.http) 

361 ) 

362 async with http_context as http, http.get(self.url) as response: 

363 yield response 

364 

365 

366@dataclass 

367class XMPPMessage[LegacyParticipantType: AnyParticipant]: 

368 body: str | None = None 

369 """ 

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

371 """ 

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

373 """ 

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

375 Is never empty if there is no body. 

376 """ 

377 reply: Reply | None = None 

378 """ 

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

380 """ 

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

382 """ 

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

384 """ 

385 replace: str | None = None 

386 """ 

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

388 """ 

389 thread: str | None = None 

390 """ 

391 The thread this message is part of. 

392 """ 

393 mentions: tuple[Mention[LegacyParticipantType], ...] = () 

394 """ 

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

396 nicknames of Room participants). 

397 Is always empty for 1:1 messages. 

398 """ 

399 

400 

401class XMPPMessageProtocol[LegacyParticipantType: AnyParticipant](Protocol): 

402 body: str | None 

403 attachments: tuple[XMPPAttachment, ...] 

404 reply: _ReplyProtocol | None 

405 link_previews: tuple[LinkPreview, ...] 

406 replace: str | None 

407 thread: str | None 

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

409 

410 

411@runtime_checkable 

412class ContactMessageProtocol(XMPPMessageProtocol[Any], Protocol): 

413 reply: ContactReplyProtocol | None 

414 mentions: tuple[()] 

415 

416 

417@runtime_checkable 

418class RoomMessageProtocol[T: AnyParticipant](XMPPMessageProtocol[T], Protocol): 

419 reply: RoomReplyProtocol | None 

420 mentions: tuple[Mention[T], ...] 

421 

422 

423class AttachmentMessageProtocol[T: AnyParticipant](XMPPMessageProtocol[T]): 

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

425 

426 

427class XMPPTextMessageProtocol[T: AnyParticipant](XMPPMessageProtocol[T]): 

428 body: str 

429 attachments: tuple[()] 

430 

431 

432class ContactAttachmentMessage[T: AnyParticipant]( 

433 ContactMessageProtocol, AttachmentMessageProtocol[T] 

434): 

435 pass 

436 

437 

438class RoomAttachmentMessage[T: AnyParticipant]( 

439 RoomMessageProtocol[T], 

440 AttachmentMessageProtocol[T], 

441): 

442 pass 

443 

444 

445class ContactTextMessage[T: AnyParticipant]( 

446 ContactMessageProtocol, XMPPTextMessageProtocol[T] 

447): 

448 pass 

449 

450 

451class RoomTextMessage[T: AnyParticipant]( 

452 RoomMessageProtocol[T], XMPPTextMessageProtocol[T] 

453): 

454 pass 

455 

456 

457type RoomMessage[T] = "RoomAttachmentMessage[T]|RoomTextMessage[T]" 

458ContactMessage = ContactAttachmentMessage[Any] | ContactTextMessage[Any] 

459 

460 

461def has_attachments[T: AnyParticipant]( 

462 message: XMPPMessageProtocol[T], 

463) -> TypeIs[AttachmentMessageProtocol[T]]: 

464 return bool(message.attachments) 

465 

466 

467@dataclass 

468class Sticker: 

469 path: Path 

470 content_type: str | None 

471 hashes: dict[str, str] 

472 fallback: str | None = None 

473 reply: Reply | None = None 

474 thread: str | None = None 

475 

476 

477class StickerProtocol(Protocol): 

478 path: Path 

479 content_type: str | None 

480 hashes: dict[str, str] 

481 fallback: str | None 

482 reply: _ReplyProtocol | None 

483 thread: str | None 

484 

485 

486@runtime_checkable 

487class ContactSticker(StickerProtocol, Protocol): 

488 reply: ContactReplyProtocol | None 

489 

490 

491@runtime_checkable 

492class RoomSticker(StickerProtocol, Protocol): 

493 reply: RoomReplyProtocol | None 

494 

495 

496type RegistrationValidationCoroutine[ 

497 RegistrationFormT: Any = Any, 

498 ModuleDataT: Mapping[str, object] = Any, 

499] = Callable[[JID, RegistrationFormT], Awaitable[ModuleDataT | None]] 

500 

501 

502LocalpartType = Literal["contact", "room"] 

503 

504 

505_BaseAttrT = str | float | bool | None 

506_AttrT = _BaseAttrT | list[str | float] 

507ExtraAttrMapping = Mapping[str, _AttrT]