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

1""" 

2Typing stuff 

3""" 

4 

5import contextlib 

6import warnings 

7from collections.abc import AsyncIterator, Iterable 

8from dataclasses import dataclass, fields 

9from datetime import datetime 

10from enum import IntEnum 

11from pathlib import Path 

12from typing import ( 

13 IO, 

14 TYPE_CHECKING, 

15 Any, 

16 Literal, 

17 NamedTuple, 

18 TypeAlias, 

19 TypedDict, 

20 TypeVar, 

21 Union, 

22) 

23 

24import aiohttp 

25from slixmpp import Message, Presence 

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

27 

28if TYPE_CHECKING: 

29 from ..contact import LegacyContact, LegacyRoster 

30 from ..core.gateway import BaseGateway 

31 from ..core.session import BaseSession 

32 from ..group import LegacyBookmarks, LegacyMUC 

33 from ..group.participant import LegacyParticipant 

34 

35AnySession: TypeAlias = "BaseSession[Any]" 

36AnyGateway: TypeAlias = "BaseGateway[AnySession]" 

37AnyMUC: TypeAlias = "LegacyMUC[Any]" 

38AnyBookmarks: TypeAlias = "LegacyBookmarks[Any]" 

39AnyRoster: TypeAlias = "LegacyRoster[Any]" 

40AnyParticipant: TypeAlias = "LegacyParticipant[Any]" 

41 

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

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

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

45 

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

47AnyRecipient = Union["LegacyContact", AnyMUC] 

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

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

50 

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

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

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

54FieldType = Literal[ 

55 "boolean", 

56 "fixed", 

57 "text-single", 

58 "text-multi", 

59 "jid-single", 

60 "jid-multi", 

61 "list-single", 

62 "list-multi", 

63 "text-private", 

64] 

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

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

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

68ClientType = Literal[ 

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

70] 

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

72 

73 

74@dataclass 

75class MessageReference: 

76 """ 

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

78 

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

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

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

82 message). 

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

84 of that failed to find the referenced message. 

85 """ 

86 

87 legacy_id: str 

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

89 body: str | None = None 

90 

91 

92@dataclass 

93class LegacyAttachment: 

94 """ 

95 A file attachment to a message 

96 

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

98 has to be set 

99 

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

101 :meth:`.LegacyParticipant.send_files` 

102 """ 

103 

104 path: Path | str | None = None 

105 name: str | None = None 

106 stream: IO[bytes] | None = None 

107 aio_stream: AsyncIterator[bytes] | None = None 

108 data: bytes | None = None 

109 content_type: str | None = None 

110 legacy_file_id: str | None = None 

111 url: str | None = None 

112 caption: str | None = None 

113 disposition: AttachmentDisposition | None = None 

114 is_sticker: bool = False 

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 

120 def __post_init__(self) -> None: 

121 if all( 

122 x is None 

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

124 ): 

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

126 if isinstance(self.path, str): 

127 self.path = Path(self.path) 

128 if self.is_sticker: 

129 if self.disposition == "attachment": 

130 warnings.warn( 

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

132 ) 

133 self.disposition = "inline" 

134 

135 def format_for_user(self) -> str: 

136 if self.name: 

137 name = self.name 

138 elif self.path: 

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

140 elif self.url: 

141 name = self.url 

142 else: 

143 name = "" 

144 

145 if self.caption: 

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

147 

148 return name 

149 

150 def __str__(self) -> str: 

151 attrs = ", ".join( 

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

153 for f in fields(self) 

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

155 ) 

156 if self.data is not None: 

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

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

159 attrs = ", ".join(to_join) 

160 return f"Attachment({attrs})" 

161 

162 

163class MucType(IntEnum): 

164 """ 

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

166 """ 

167 

168 GROUP = 0 

169 """ 

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

171 """ 

172 CHANNEL = 1 

173 """ 

174 A public group, aka an anonymous channel. 

175 """ 

176 CHANNEL_NON_ANONYMOUS = 2 

177 """ 

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

179 """ 

180 

181 

182PseudoPresenceShow = PresenceShows | Literal[""] 

183 

184 

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

186 

187 

188class LinkPreview(NamedTuple): 

189 """ 

190 Embedded metadata from :xep:`0511`. 

191 

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

193 """ 

194 

195 about: str 

196 """ 

197 URL of the link. 

198 """ 

199 title: str | None 

200 """ 

201 Title of the linked page. 

202 """ 

203 description: str | None 

204 """ 

205 A description of the page. 

206 """ 

207 url: str | None 

208 """ 

209 The canonical URL of the link. 

210 """ 

211 image: str | Path | bytes | None 

212 """ 

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

214 """ 

215 type: str | None 

216 """ 

217 Type of the link destination. 

218 """ 

219 site_name: str | None 

220 """ 

221 Name of the web site. 

222 """ 

223 

224 @property 

225 def is_empty(self) -> bool: 

226 return not any(x for x in self) 

227 

228 

229class Mention(NamedTuple): 

230 contact: "LegacyContact" 

231 start: int 

232 end: int 

233 

234 

235class Hat(NamedTuple): 

236 uri: str 

237 title: str 

238 hue: float | None = None 

239 

240 

241class UserPreferences(TypedDict): 

242 sync_avatar: bool 

243 sync_presence: bool 

244 

245 

246class MamMetadata(NamedTuple): 

247 id: str 

248 sent_on: datetime 

249 

250 

251class HoleBound(NamedTuple): 

252 id: str 

253 timestamp: datetime 

254 

255 

256class CachedPresence(NamedTuple): 

257 last_seen: datetime | None = None 

258 ptype: PresenceTypes | None = None 

259 pstatus: str | None = None 

260 pshow: PresenceShows | None = None 

261 

262 

263class Avatar(NamedTuple): 

264 path: Path | None = None 

265 unique_id: str | None = None 

266 url: str | None = None 

267 data: bytes | None = None 

268 

269 

270class SpaceMetadata(NamedTuple): 

271 creator_legacy_id: str | None = None 

272 name: str | None = None 

273 description: str | None = None 

274 member_count: int | None = None 

275 owner_legacy_ids: Iterable[str] = [] 

276 

277 

278@dataclass 

279class Reply: 

280 msg_id: str 

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

282 fallback: str | None = None 

283 

284 

285@dataclass 

286class XMPPAttachment: 

287 url: str 

288 is_sticker: bool = False 

289 cid: str | None = None 

290 content_type: str | None = None 

291 

292 @contextlib.asynccontextmanager 

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

294 async with ( 

295 aiohttp.ClientSession() as session, 

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

297 ): 

298 yield response 

299 

300 

301@dataclass 

302class XMPPMessage: 

303 body: str | None = None 

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

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

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

307 replace: str | None = None 

308 reply: Reply | None = None 

309 thread: str | None = None 

310 

311 

312@dataclass 

313class Sticker: 

314 path: Path 

315 content_type: str | None 

316 hashes: dict[str, str] 

317 fallback: str | None = None 

318 reply: Reply | None = None 

319 thread: str | None = None