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

1import logging 

2import mimetypes 

3import re 

4from collections.abc import Callable, Collection, Coroutine 

5from functools import wraps 

6from pathlib import Path 

7from time import time 

8from typing import ( 

9 Any, 

10 Concatenate, 

11 NamedTuple, 

12 ParamSpec, 

13 Protocol, 

14 TypeVar, 

15 get_args, 

16 get_origin, 

17) 

18from xml.etree import ElementTree as ET 

19 

20try: 

21 import emoji 

22except ImportError: 

23 EMOJI_LIB_AVAILABLE = False 

24else: 

25 EMOJI_LIB_AVAILABLE = True 

26 

27from slixmpp.types import ExtPresenceShows, ResourceDict 

28 

29from .types import LegacyParticipantType, Mention 

30 

31try: 

32 import magic 

33except ImportError as e: 

34 magic = None # type:ignore 

35 logging.warning( # noqa: LOG015 

36 ( 

37 "Libmagic is not available: %s. " 

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

39 ), 

40 e, 

41 ) 

42 

43 

44def fix_suffix( 

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

46) -> tuple[str, str]: 

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

48 if guessed == mime_type: 

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

50 else: 

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

52 mime_type = guessed 

53 

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

55 

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

57 

58 suffix = name.suffix 

59 

60 if suffix in valid_suffix_list: 

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

62 return str(name), guessed 

63 

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

65 if valid_suffix is None: 

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

67 return str(name), guessed 

68 

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

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

71 

72 

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

74 if phone is None: 

75 return False 

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

77 if match is None: 

78 return False 

79 return match[0] == phone 

80 

81 

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

83 return ILLEGAL_XML_CHARS_RE.sub(repl, s) 

84 

85 

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

87ILLEGAL = [ 

88 (0x00, 0x08), 

89 (0x0B, 0x0C), 

90 (0x0E, 0x1F), 

91 (0x7F, 0x84), 

92 (0x86, 0x9F), 

93 (0xFDD0, 0xFDDF), 

94 (0xFFFE, 0xFFFF), 

95 (0x1FFFE, 0x1FFFF), 

96 (0x2FFFE, 0x2FFFF), 

97 (0x3FFFE, 0x3FFFF), 

98 (0x4FFFE, 0x4FFFF), 

99 (0x5FFFE, 0x5FFFF), 

100 (0x6FFFE, 0x6FFFF), 

101 (0x7FFFE, 0x7FFFF), 

102 (0x8FFFE, 0x8FFFF), 

103 (0x9FFFE, 0x9FFFF), 

104 (0xAFFFE, 0xAFFFF), 

105 (0xBFFFE, 0xBFFFF), 

106 (0xCFFFE, 0xCFFFF), 

107 (0xDFFFE, 0xDFFFF), 

108 (0xEFFFE, 0xEFFFF), 

109 (0xFFFFE, 0xFFFFF), 

110 (0x10FFFE, 0x10FFFF), 

111] 

112 

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

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

115ILLEGAL_XML_CHARS_RE = re.compile(XML_ILLEGAL_CHARACTER_REGEX) 

116 

117 

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

119def addLoggingLevel( 

120 levelName: str = "TRACE", 

121 levelNum: int = logging.DEBUG - 5, 

122 methodName: str | None = None, 

123) -> None: 

124 """ 

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

126 currently configured logging class. 

127 

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

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

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

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

132 used. 

133 

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

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

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

137 

138 Example 

139 ------- 

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

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

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

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

144 >>> logging.TRACE 

145 5 

146 

147 """ 

148 if not methodName: 

149 methodName = levelName.lower() 

150 

151 if hasattr(logging, levelName): 

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

153 return 

154 if hasattr(logging, methodName): 

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

156 return 

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

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

159 return 

160 

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

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

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

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

165 if self.isEnabledFor(levelNum): 

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

167 

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

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

170 

171 logging.addLevelName(levelNum, levelName) 

172 setattr(logging, levelName, levelNum) 

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

174 setattr(logging, methodName, logToRoot) 

175 

176 

177class SlidgeLogger(logging.Logger): 

178 def trace(self) -> None: 

179 pass 

180 

181 

182log = logging.getLogger(__name__) 

183 

184 

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

186 if len(resources) == 0: 

187 return None 

188 

189 if len(resources) == 1: 

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

191 

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

193 

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

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

196 show: ExtPresenceShows = "" 

197 else: 

198 for r in by_priority: 

199 if r["show"]: 

200 show = r["show"] 

201 break 

202 else: 

203 raise RuntimeError() 

204 

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

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

207 status = "" 

208 for r in by_priority: 

209 if r["status"]: 

210 status = r["status"] 

211 break 

212 

213 return { 

214 "show": show, 

215 "status": status, 

216 "priority": 0, 

217 } 

218 

219 

220_EMOJI_VARIATION_SELECTOR = "\ufe0f" 

221 

222 

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

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

225 return emoji.rstrip(_EMOJI_VARIATION_SELECTOR) 

226 

227 

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

229 

230 

231def dict_to_named_tuple[NamedTupleT: NamedTuple]( 

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

233) -> NamedTupleT: 

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

235 

236 

237def replace_mentions( 

238 text: str, 

239 mentions: Collection[Mention[LegacyParticipantType]] | None, 

240 mapping: Callable[[Mention[LegacyParticipantType]], str], 

241) -> str: 

242 if not mentions: 

243 return text 

244 

245 cursor = 0 

246 pieces = [] 

247 for mention in mentions: 

248 new_text = mapping(mention) 

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

250 cursor = mention.end 

251 pieces.append(text[cursor:]) 

252 return "".join(pieces) 

253 

254 

255class HasLogger(Protocol): 

256 log: logging.Logger 

257 

258 

259P = ParamSpec("P") 

260T = TypeVar("T") 

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

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

263 

264 

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

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

267 

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

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

270 parameters. 

271 """ 

272 for attr in attrs: 

273 if attr in cls.__dict__: 

274 raise TypeError( 

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

276 " derived from the generic parameter.", 

277 ) 

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

279 if get_origin(base) is not origin: 

280 continue 

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

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

283 # Still generic? Intermediate subclass? Probably? 

284 continue 

285 if not isinstance(arg, type): 

286 # e.g. LegacyBookmarks[LegacyMUC[LegacyParticipant]]: the runtime 

287 # class to instantiate is the alias' origin, LegacyMUC 

288 arg = get_origin(arg) 

289 if isinstance(arg, type): 

290 setattr(cls, attr, arg) 

291 continue 

292 raise TypeError( 

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

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

295 ) 

296 return 

297 

298 

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

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

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

302 @wraps(func) 

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

304 start = time() 

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

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

307 return r 

308 

309 return wrapped 

310 

311 

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

313 if not EMOJI_LIB_AVAILABLE: 

314 return text 

315 words = text.split(" ") 

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

317 # purely_emoji seems better 

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

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

320 return text 

321 

322 

323async def noop_coro() -> None: 

324 pass 

325 

326 

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

328 """ 

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

330 """ 

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

332 

333 

334def fix_namespaces( 

335 xml: ET.Element, 

336 old: str, 

337 new: str, 

338) -> None: 

339 """ 

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

341 

342 Acts in-place. 

343 

344 :param xml: 

345 :param old: 

346 :param new: 

347 """ 

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

349 for child in xml: 

350 fix_namespaces(child, old, new)