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( 

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"]: 

90 for base in cls.__bases__: 

91 if issubclass(base, SubclassableOnce): 

92 return base 

93 else: 

94 raise RuntimeError("wut") 

95 

96 @classmethod 

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

98 try: 

99 return cls.get_unique_subclass() 

100 except AttributeError: 

101 return cls 

102 

103 @classmethod 

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

105 existing = SubclassableOnce.__subclasses.get(cls) 

106 if existing is None: 

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

108 return existing 

109 

110 @classmethod 

111 def reset_subclass(cls) -> None: 

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

113 cls.__subclasses[cls] = None 

114 

115 

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

117 if phone is None: 

118 return False 

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

120 if match is None: 

121 return False 

122 return match[0] == phone 

123 

124 

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

126 return ILLEGAL_XML_CHARS_RE.sub(repl, s) 

127 

128 

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

130ILLEGAL = [ 

131 (0x00, 0x08), 

132 (0x0B, 0x0C), 

133 (0x0E, 0x1F), 

134 (0x7F, 0x84), 

135 (0x86, 0x9F), 

136 (0xFDD0, 0xFDDF), 

137 (0xFFFE, 0xFFFF), 

138 (0x1FFFE, 0x1FFFF), 

139 (0x2FFFE, 0x2FFFF), 

140 (0x3FFFE, 0x3FFFF), 

141 (0x4FFFE, 0x4FFFF), 

142 (0x5FFFE, 0x5FFFF), 

143 (0x6FFFE, 0x6FFFF), 

144 (0x7FFFE, 0x7FFFF), 

145 (0x8FFFE, 0x8FFFF), 

146 (0x9FFFE, 0x9FFFF), 

147 (0xAFFFE, 0xAFFFF), 

148 (0xBFFFE, 0xBFFFF), 

149 (0xCFFFE, 0xCFFFF), 

150 (0xDFFFE, 0xDFFFF), 

151 (0xEFFFE, 0xEFFFF), 

152 (0xFFFFE, 0xFFFFF), 

153 (0x10FFFE, 0x10FFFF), 

154] 

155 

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

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

158ILLEGAL_XML_CHARS_RE = re.compile(XML_ILLEGAL_CHARACTER_REGEX) 

159 

160 

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

162def addLoggingLevel( 

163 levelName: str = "TRACE", 

164 levelNum: int = logging.DEBUG - 5, 

165 methodName: str | None = None, 

166) -> None: 

167 """ 

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

169 currently configured logging class. 

170 

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

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

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

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

175 used. 

176 

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

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

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

180 

181 Example 

182 ------- 

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

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

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

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

187 >>> logging.TRACE 

188 5 

189 

190 """ 

191 if not methodName: 

192 methodName = levelName.lower() 

193 

194 if hasattr(logging, levelName): 

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

196 return 

197 if hasattr(logging, methodName): 

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

199 return 

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

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

202 return 

203 

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

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

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

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

208 if self.isEnabledFor(levelNum): 

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

210 

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

212 logging.log(levelNum, message, *args, **kwargs) 

213 

214 logging.addLevelName(levelNum, levelName) 

215 setattr(logging, levelName, levelNum) 

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

217 setattr(logging, methodName, logToRoot) 

218 

219 

220class SlidgeLogger(logging.Logger): 

221 def trace(self) -> None: 

222 pass 

223 

224 

225log = logging.getLogger(__name__) 

226 

227 

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

229 if len(resources) == 0: 

230 return None 

231 

232 if len(resources) == 1: 

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

234 

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

236 

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

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

239 show: ExtPresenceShows = "" 

240 else: 

241 for r in by_priority: 

242 if r["show"]: 

243 show = r["show"] 

244 break 

245 else: 

246 raise RuntimeError() 

247 

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

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

250 status = "" 

251 for r in by_priority: 

252 if r["status"]: 

253 status = r["status"] 

254 break 

255 

256 return { 

257 "show": show, 

258 "status": status, 

259 "priority": 0, 

260 } 

261 

262 

263_EMOJI_VARIATION_SELECTOR = "\ufe0f" 

264 

265 

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

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

268 return emoji.removesuffix(_EMOJI_VARIATION_SELECTOR) 

269 

270 

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

272 

273 

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

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

276 

277 

278def replace_mentions( 

279 text: str, 

280 mentions: Collection[Mention] | None, 

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

282) -> str: 

283 if not mentions: 

284 return text 

285 

286 cursor = 0 

287 pieces = [] 

288 for mention in mentions: 

289 try: 

290 new_text = mapping(mention) 

291 except Exception as exc: 

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

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

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

295 cursor = mention.end 

296 pieces.append(text[cursor:]) 

297 return "".join(pieces) 

298 

299 

300class HasLogger(Protocol): 

301 log: logging.Logger 

302 

303 

304P = ParamSpec("P") 

305T = TypeVar("T") 

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

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

308 

309 

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

311 @wraps(func) 

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

313 start = time() 

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

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

316 return r 

317 

318 return wrapped 

319 

320 

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

322 if not EMOJI_LIB_AVAILABLE: 

323 return text 

324 words = text.split(" ") 

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

326 # purely_emoji seems better 

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

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

329 return text 

330 

331 

332async def noop_coro() -> None: 

333 pass 

334 

335 

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

337 """ 

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

339 """ 

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

341 

342 

343def fix_namespaces( 

344 xml: ET.Element, 

345 old: str, 

346 new: str, 

347) -> None: 

348 """ 

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

350 

351 Acts in-place. 

352 

353 :param xml: 

354 :param old: 

355 :param new: 

356 """ 

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

358 for child in xml: 

359 fix_namespaces(child, old, new)