Coverage for slidge/util/util.py: 75%

1import logging 

2import mimetypes 

3import re 

4import uuid 

5from collections.abc import Awaitable, Callable, Collection, Coroutine 

6from contextlib import AbstractContextManager 

7from functools import wraps 

8from inspect import iscoroutinefunction 

9from pathlib import Path 

10from time import time 

11from typing import ( 

12 Any, 

13 Concatenate, 

14 NamedTuple, 

15 ParamSpec, 

16 Protocol, 

17 TypeVar, 

18 get_args, 

19 get_origin, 

20 overload, 

21) 

22from xml.etree import ElementTree as ET 

23 

24try: 

25 import emoji 

26except ImportError: 

27 EMOJI_LIB_AVAILABLE = False 

28else: 

29 EMOJI_LIB_AVAILABLE = True 

30 

31from slixmpp.types import ExtPresenceShows, ResourceDict 

32 

33from .types import AnyParticipant, Mention 

34 

35try: 

36 import magic 

37except ImportError as e: 

38 magic = None # type:ignore 

39 logging.warning( # noqa: LOG015 

40 ( 

41 "Libmagic is not available: %s. " 

42 "It's OK if you don't use fix-filename-suffix-mime-type." 

43 ), 

44 e, 

45 ) 

46 

47 

48def fix_suffix( 

49 path: Path, mime_type: str | None, file_name: str | None 

50) -> tuple[str, str]: 

51 guessed = magic.from_file(path, mime=True) 

52 if guessed == mime_type: 

53 log.debug("Magic and given MIME match") 

54 else: 

55 log.debug("Magic (%s) and given MIME (%s) differ", guessed, mime_type) 

56 mime_type = guessed 

57 

58 valid_suffix_list = mimetypes.guess_all_extensions(mime_type, strict=False) 

59 

60 name = Path(file_name) if file_name else Path(path.name) 

61 

62 suffix = name.suffix 

63 

64 if suffix in valid_suffix_list: 

65 log.debug("Suffix %s is in %s", suffix, valid_suffix_list) 

66 return str(name), guessed 

67 

68 valid_suffix = mimetypes.guess_extension(mime_type.split(";")[0], strict=False) 

69 if valid_suffix is None: 

70 log.debug("No valid suffix found") 

71 return str(name), guessed 

72 

73 log.debug("Changing suffix of %s to %s", file_name or path.name, valid_suffix) 

74 return str(name.with_suffix(valid_suffix)), guessed 

75 

76 

77def is_valid_phone_number(phone: str | None) -> bool: 

78 if phone is None: 

79 return False 

80 match = re.match(r"\+\d.*", phone) 

81 if match is None: 

82 return False 

83 return match[0] == phone 

84 

85 

86def strip_illegal_chars(s: str, repl: str = "") -> str: 

87 return ILLEGAL_XML_CHARS_RE.sub(repl, s) 

88 

89 

90# from https://stackoverflow.com/a/64570125/5902284 and Link Mauve 

91ILLEGAL = [ 

92 (0x00, 0x08), 

93 (0x0B, 0x0C), 

94 (0x0E, 0x1F), 

95 (0x7F, 0x84), 

96 (0x86, 0x9F), 

97 (0xFDD0, 0xFDDF), 

98 (0xFFFE, 0xFFFF), 

99 (0x1FFFE, 0x1FFFF), 

100 (0x2FFFE, 0x2FFFF), 

101 (0x3FFFE, 0x3FFFF), 

102 (0x4FFFE, 0x4FFFF), 

103 (0x5FFFE, 0x5FFFF), 

104 (0x6FFFE, 0x6FFFF), 

105 (0x7FFFE, 0x7FFFF), 

106 (0x8FFFE, 0x8FFFF), 

107 (0x9FFFE, 0x9FFFF), 

108 (0xAFFFE, 0xAFFFF), 

109 (0xBFFFE, 0xBFFFF), 

110 (0xCFFFE, 0xCFFFF), 

111 (0xDFFFE, 0xDFFFF), 

112 (0xEFFFE, 0xEFFFF), 

113 (0xFFFFE, 0xFFFFF), 

114 (0x10FFFE, 0x10FFFF), 

115] 

116 

117ILLEGAL_RANGES = [rf"{chr(low)}-{chr(high)}" for (low, high) in ILLEGAL] 

118XML_ILLEGAL_CHARACTER_REGEX = "[" + "".join(ILLEGAL_RANGES) + "]" 

119ILLEGAL_XML_CHARS_RE = re.compile(XML_ILLEGAL_CHARACTER_REGEX) 

120 

121 

122# from https://stackoverflow.com/a/35804945/5902284 

123def addLoggingLevel( 

124 levelName: str = "TRACE", 

125 levelNum: int = logging.DEBUG - 5, 

126 methodName: str | None = None, 

127) -> None: 

128 """ 

129 Comprehensively adds a new logging level to the `logging` module and the 

130 currently configured logging class. 

131 

132 `levelName` becomes an attribute of the `logging` module with the value 

133 `levelNum`. `methodName` becomes a convenience method for both `logging` 

134 itself and the class returned by `logging.getLoggerClass()` (usually just 

135 `logging.Logger`). If `methodName` is not specified, `levelName.lower()` is 

136 used. 

137 

138 To avoid accidental clobberings of existing attributes, this method will 

139 raise an `AttributeError` if the level name is already an attribute of the 

140 `logging` module or if the method name is already present 

141 

142 Example 

143 ------- 

144 >>> addLoggingLevel('TRACE', logging.DEBUG - 5) 

145 >>> logging.getLogger(__name__).setLevel("TRACE") 

146 >>> logging.getLogger(__name__).trace('that worked') 

147 >>> logging.trace('so did this') 

148 >>> logging.TRACE 

149 5 

150 

151 """ 

152 if not methodName: 

153 methodName = levelName.lower() 

154 

155 if hasattr(logging, levelName): 

156 log.debug(f"{levelName} already defined in logging module") 

157 return 

158 if hasattr(logging, methodName): 

159 log.debug(f"{methodName} already defined in logging module") 

160 return 

161 if hasattr(logging.getLoggerClass(), methodName): 

162 log.debug(f"{methodName} already defined in logger class") 

163 return 

164 

165 # This method was inspired by the answers to Stack Overflow post 

166 # http://stackoverflow.com/q/2183233/2988730, especially 

167 # http://stackoverflow.com/a/13638084/2988730 

168 def logForLevel(self, message, *args, **kwargs) -> None: # type:ignore[no-untyped-def] # noqa 

169 if self.isEnabledFor(levelNum): 

170 self._log(levelNum, message, args, **kwargs) 

171 

172 def logToRoot(message, *args, **kwargs) -> None: # type:ignore[no-untyped-def] # noqa 

173 logging.log(levelNum, message, *args, **kwargs) # noqa: LOG015 

174 

175 logging.addLevelName(levelNum, levelName) 

176 setattr(logging, levelName, levelNum) 

177 setattr(logging.getLoggerClass(), methodName, logForLevel) 

178 setattr(logging, methodName, logToRoot) 

179 

180 

181class SlidgeLogger(logging.Logger): 

182 def trace(self) -> None: 

183 pass 

184 

185 

186log = logging.getLogger(__name__) 

187 

188 

189def merge_resources(resources: dict[str, ResourceDict]) -> ResourceDict | None: 

190 if len(resources) == 0: 

191 return None 

192 

193 if len(resources) == 1: 

194 return next(iter(resources.values())) 

195 

196 by_priority = sorted(resources.values(), key=lambda r: r["priority"], reverse=True) 

197 

198 if any(r["show"] == "" for r in resources.values()): 

199 # if a client is "available", we're "available" 

200 show: ExtPresenceShows = "" 

201 else: 

202 for r in by_priority: 

203 if r["show"]: 

204 show = r["show"] 

205 break 

206 else: 

207 raise RuntimeError() 

208 

209 # if there are different statuses, we use the highest priority one, 

210 # but we ignore resources without status, even with high priority 

211 status = "" 

212 for r in by_priority: 

213 if r["status"]: 

214 status = r["status"] 

215 break 

216 

217 return { 

218 "show": show, 

219 "status": status, 

220 "priority": 0, 

221 } 

222 

223 

224_EMOJI_VARIATION_SELECTOR = "\ufe0f" 

225 

226 

227def remove_emoji_variation_selector_16(emoji: str) -> str: 

228 # this is required for compatibility with dino, and maybe other future clients? 

229 return emoji.rstrip(_EMOJI_VARIATION_SELECTOR) 

230 

231 

232NamedTupleT = TypeVar("NamedTupleT", bound=NamedTuple) 

233 

234 

235def dict_to_named_tuple[NamedTupleT: NamedTuple]( 

236 data: dict[str, Any], cls: type[NamedTupleT] 

237) -> NamedTupleT: 

238 return cls(*(data.get(f) for f in cls._fields)) # type:ignore[arg-type] 

239 

240 

241def replace_mentions[T: AnyParticipant]( 

242 text: str, 

243 mentions: Collection[Mention[T]] | None, 

244 mapping: Callable[[Mention[T]], str], 

245) -> str: 

246 if not mentions: 

247 return text 

248 

249 cursor = 0 

250 pieces = [] 

251 for mention in mentions: 

252 new_text = mapping(mention) 

253 pieces.extend([text[cursor : mention.start], new_text]) 

254 cursor = mention.end 

255 pieces.append(text[cursor:]) 

256 return "".join(pieces) 

257 

258 

259class HasLogger(Protocol): 

260 log: logging.Logger 

261 

262 

263P = ParamSpec("P") 

264T = TypeVar("T") 

265Self = TypeVar("Self", bound=HasLogger) 

266TimeItWrapped = Callable[Concatenate[Self, P], Coroutine[Any, Any, T]] 

267 

268 

269def derive_wired_class(cls: type[Any], origin: type[Any], *attrs: str) -> None: 

270 """Derive a factory class's associated type(s) from its generic parameter(s). 

271 

272 E.g.: for ``class Roster(LegacyRoster[Contact])``, set ``Roster.contact_cls`` to 

273 ``Contact``. Attribute names are passed in the same order as the generic 

274 parameters. 

275 """ 

276 for attr in attrs: 

277 if attr in cls.__dict__: 

278 raise TypeError( 

279 f"{cls.__name__} must not declare '{attr}' explicitly: it is" 

280 " derived from the generic parameter.", 

281 ) 

282 for base in getattr(cls, "__orig_bases__", ()): 

283 if get_origin(base) is not origin: 

284 continue 

285 for attr, arg in zip(attrs, get_args(base)): 

286 if isinstance(arg, TypeVar) or type(arg).__name__ == "TypeVar": 

287 # Still generic? Intermediate subclass? Probably? 

288 continue 

289 if not isinstance(arg, type): 

290 # e.g. LegacyBookmarks[LegacyRoom[LegacyParticipant]]: the runtime 

291 # class to instantiate is the alias' origin, LegacyRoom 

292 arg = get_origin(arg) 

293 if isinstance(arg, type): 

294 setattr(cls, attr, arg) 

295 continue 

296 raise TypeError( 

297 f"Cannot derive {cls.__name__}.{attr} from a forward reference. " 

298 f"Parameterize {origin.__name__} with the actual class." 

299 ) 

300 return 

301 

302 

303def timeit[Self: HasLogger, **P, T]( 

304 func: TimeItWrapped[Self, P, T], 

305) -> TimeItWrapped[Self, P, T]: 

306 @wraps(func) 

307 async def wrapped(self: Self, /, *args: P.args, **kwargs: P.kwargs) -> T: 

308 start = time() 

309 r = await func(self, *args, **kwargs) 

310 self.log.debug("%s took %s ms", func.__name__, round((time() - start) * 1000)) 

311 return r 

312 

313 return wrapped 

314 

315 

316def strip_leading_emoji(text: str) -> str: 

317 if not EMOJI_LIB_AVAILABLE: 

318 return text 

319 words = text.split(" ") 

320 # is_emoji returns False for 🛷️ for obscure reasons, 

321 # purely_emoji seems better 

322 if len(words) > 1 and emoji.purely_emoji(words[0]): 

323 return " ".join(words[1:]) 

324 return text 

325 

326 

327async def noop_coro() -> None: 

328 pass 

329 

330 

331def add_quote_prefix(text: str) -> str: 

332 """ 

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

334 """ 

335 return "\n".join(("> " + x).strip() for x in text.split("\n")).strip() 

336 

337 

338def fix_namespaces( 

339 xml: ET.Element, 

340 old: str, 

341 new: str, 

342) -> None: 

343 """ 

344 Hack to fix namespaces between jabber:component and jabber:client 

345 

346 Acts in-place. 

347 

348 :param xml: 

349 :param old: 

350 :param new: 

351 """ 

352 xml.tag = xml.tag.replace(f"{{{old}}}", f"{{{new}}}") 

353 for child in xml: 

354 fix_namespaces(child, old, new) 

355 

356 

357class _HasBatchSetAttributes(Protocol): 

358 def batch_set_attributes(self) -> AbstractContextManager[None]: ... 

359 

360 

361@overload 

362def batch_set_attributes[S: _HasBatchSetAttributes, **P, R]( 

363 func: Callable[Concatenate[S, P], Awaitable[R]], 

364) -> Callable[Concatenate[S, P], Awaitable[R]]: ... 

365 

366 

367@overload 

368def batch_set_attributes[S: _HasBatchSetAttributes, **P, R]( 

369 func: Callable[Concatenate[S, P], R], 

370) -> Callable[Concatenate[S, P], R]: ... 

371 

372 

373def batch_set_attributes(func: Callable[..., Any]) -> Callable[..., Any]: 

374 if iscoroutinefunction(func): 

375 

376 @wraps(func) 

377 async def wrapper( 

378 self: _HasBatchSetAttributes, 

379 *args: Any, # noqa: ANN401 

380 **kwargs: Any, # noqa: ANN401 

381 ) -> Any: # noqa: ANN401 

382 with self.batch_set_attributes(): 

383 return await func(self, *args, **kwargs) 

384 

385 return wrapper 

386 

387 else: 

388 

389 @wraps(func) 

390 def wrapper(self: _HasBatchSetAttributes, *args: Any, **kwargs: Any) -> Any: # noqa: ANN401 

391 with self.batch_set_attributes(): 

392 return func(self, *args, **kwargs) 

393 

394 return wrapper 

395 

396 

397def uniquify_replace(legacy_msg_id: str) -> str: 

398 return f"--slidge--correct--{uuid.uuid4()}--{legacy_msg_id}" 

399 

400 

401def recover_replaced_legacy_msg_id(xmpp_id: str) -> str: 

402 if not xmpp_id.startswith("--slidge--correct--"): 

403 return xmpp_id 

404 return xmpp_id.removeprefix("--slidge--correct--")[_uuid4_len + 2 :] 

405 

406 

407_uuid4_len = 36