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

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

11 Concatenate, 

12 NamedTuple, 

13 ParamSpec, 

14 Protocol, 

15 TypeVar, 

16) 

17from xml.etree import ElementTree as ET 

18 

19try: 

20 import emoji 

21except ImportError: 

22 EMOJI_LIB_AVAILABLE = False 

23else: 

24 EMOJI_LIB_AVAILABLE = True 

25 

26from slixmpp.types import ExtPresenceShows, ResourceDict 

27 

28from .types import Mention 

29 

30try: 

31 import magic 

32except ImportError as e: 

33 magic = None # type:ignore 

34 logging.warning( # noqa: LOG015 

35 ( 

36 "Libmagic is not available: %s. " 

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

38 ), 

39 e, 

40 ) 

41 

42 

43def fix_suffix( 

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

45) -> tuple[str, str]: 

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

47 if guessed == mime_type: 

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

49 else: 

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

51 mime_type = guessed 

52 

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

54 

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

56 

57 suffix = name.suffix 

58 

59 if suffix in valid_suffix_list: 

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

61 return str(name), guessed 

62 

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

64 if valid_suffix is None: 

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

66 return str(name), guessed 

67 

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

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

70 

71 

72class SubclassableOnce: 

73 # To allow importing everything, including plugins, during tests 

74 TEST_MODE: bool = False 

75 __subclasses: ClassVar[ 

76 dict[type["SubclassableOnce"], type["SubclassableOnce"] | None] 

77 ] = {} 

78 

79 def __init_subclass__(cls, **kwargs: object) -> None: 

80 if SubclassableOnce not in cls.__bases__: 

81 base = SubclassableOnce.__find_direct_child(cls) 

82 existing = SubclassableOnce.__subclasses.get(base) 

83 if existing is not None and not SubclassableOnce.TEST_MODE: 

84 raise RuntimeError("This class must be subclassed once at most!") 

85 cls.__subclasses[base] = cls 

86 super().__init_subclass__(**kwargs) 

87 

88 @staticmethod 

89 def __find_direct_child(cls: type["SubclassableOnce"]) -> type["SubclassableOnce"]: # noqa:PLW0211 

90 for base in cls.__bases__: 

91 if issubclass(base, SubclassableOnce): 

92 return base 

93 raise RuntimeError("wut") 

94 

95 @classmethod 

96 def get_self_or_unique_subclass(cls) -> "type[SubclassableOnce]": 

97 try: 

98 return cls.get_unique_subclass() 

99 except AttributeError: 

100 return cls 

101 

102 @classmethod 

103 def get_unique_subclass(cls) -> "type[SubclassableOnce]": 

104 existing = SubclassableOnce.__subclasses.get(cls) 

105 if existing is None: 

106 raise AttributeError("Could not find any subclass", cls) 

107 return existing 

108 

109 @classmethod 

110 def reset_subclass(cls) -> None: 

111 log.debug("Resetting subclass of %s", cls) 

112 cls.__subclasses[cls] = None 

113 

114 

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

116 if phone is None: 

117 return False 

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

119 if match is None: 

120 return False 

121 return match[0] == phone 

122 

123 

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

125 return ILLEGAL_XML_CHARS_RE.sub(repl, s) 

126 

127 

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

129ILLEGAL = [ 

130 (0x00, 0x08), 

131 (0x0B, 0x0C), 

132 (0x0E, 0x1F), 

133 (0x7F, 0x84), 

134 (0x86, 0x9F), 

135 (0xFDD0, 0xFDDF), 

136 (0xFFFE, 0xFFFF), 

137 (0x1FFFE, 0x1FFFF), 

138 (0x2FFFE, 0x2FFFF), 

139 (0x3FFFE, 0x3FFFF), 

140 (0x4FFFE, 0x4FFFF), 

141 (0x5FFFE, 0x5FFFF), 

142 (0x6FFFE, 0x6FFFF), 

143 (0x7FFFE, 0x7FFFF), 

144 (0x8FFFE, 0x8FFFF), 

145 (0x9FFFE, 0x9FFFF), 

146 (0xAFFFE, 0xAFFFF), 

147 (0xBFFFE, 0xBFFFF), 

148 (0xCFFFE, 0xCFFFF), 

149 (0xDFFFE, 0xDFFFF), 

150 (0xEFFFE, 0xEFFFF), 

151 (0xFFFFE, 0xFFFFF), 

152 (0x10FFFE, 0x10FFFF), 

153] 

154 

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

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

157ILLEGAL_XML_CHARS_RE = re.compile(XML_ILLEGAL_CHARACTER_REGEX) 

158 

159 

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

161def addLoggingLevel( 

162 levelName: str = "TRACE", 

163 levelNum: int = logging.DEBUG - 5, 

164 methodName: str | None = None, 

165) -> None: 

166 """ 

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

168 currently configured logging class. 

169 

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

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

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

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

174 used. 

175 

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

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

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

179 

180 Example 

181 ------- 

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

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

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

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

186 >>> logging.TRACE 

187 5 

188 

189 """ 

190 if not methodName: 

191 methodName = levelName.lower() 

192 

193 if hasattr(logging, levelName): 

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

195 return 

196 if hasattr(logging, methodName): 

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

198 return 

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

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

201 return 

202 

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

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

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

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

207 if self.isEnabledFor(levelNum): 

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

209 

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

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

212 

213 logging.addLevelName(levelNum, levelName) 

214 setattr(logging, levelName, levelNum) 

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

216 setattr(logging, methodName, logToRoot) 

217 

218 

219class SlidgeLogger(logging.Logger): 

220 def trace(self) -> None: 

221 pass 

222 

223 

224log = logging.getLogger(__name__) 

225 

226 

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

228 if len(resources) == 0: 

229 return None 

230 

231 if len(resources) == 1: 

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

233 

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

235 

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

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

238 show: ExtPresenceShows = "" 

239 else: 

240 for r in by_priority: 

241 if r["show"]: 

242 show = r["show"] 

243 break 

244 else: 

245 raise RuntimeError() 

246 

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

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

249 status = "" 

250 for r in by_priority: 

251 if r["status"]: 

252 status = r["status"] 

253 break 

254 

255 return { 

256 "show": show, 

257 "status": status, 

258 "priority": 0, 

259 } 

260 

261 

262_EMOJI_VARIATION_SELECTOR = "\ufe0f" 

263 

264 

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

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

267 return emoji.rstrip(_EMOJI_VARIATION_SELECTOR) 

268 

269 

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

271 

272 

273def dict_to_named_tuple(data: dict[str, Any], cls: type[NamedTupleT]) -> NamedTupleT: 

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

275 

276 

277def replace_mentions( 

278 text: str, 

279 mentions: Collection[Mention] | None, 

280 mapping: Callable[[Mention], str], 

281) -> str: 

282 if not mentions: 

283 return text 

284 

285 cursor = 0 

286 pieces = [] 

287 for mention in mentions: 

288 try: 

289 new_text = mapping(mention) 

290 except Exception as exc: # noqa: BLE001 

291 log.debug("Attempting slidge <= 0.3.3 compatibility: %s", exc) 

292 new_text = mapping(mention.contact) # type:ignore 

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

294 cursor = mention.end 

295 pieces.append(text[cursor:]) 

296 return "".join(pieces) 

297 

298 

299class HasLogger(Protocol): 

300 log: logging.Logger 

301 

302 

303P = ParamSpec("P") 

304T = TypeVar("T") 

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

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

307 

308 

309def timeit(func: TimeItWrapped[Self, P, T]) -> TimeItWrapped[Self, P, T]: 

310 @wraps(func) 

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

312 start = time() 

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

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

315 return r 

316 

317 return wrapped 

318 

319 

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

321 if not EMOJI_LIB_AVAILABLE: 

322 return text 

323 words = text.split(" ") 

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

325 # purely_emoji seems better 

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

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

328 return text 

329 

330 

331async def noop_coro() -> None: 

332 pass 

333 

334 

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

336 """ 

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

338 """ 

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

340 

341 

342def fix_namespaces( 

343 xml: ET.Element, 

344 old: str, 

345 new: str, 

346) -> None: 

347 """ 

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

349 

350 Acts in-place. 

351 

352 :param xml: 

353 :param old: 

354 :param new: 

355 """ 

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

357 for child in xml: 

358 fix_namespaces(child, old, new)