utils.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452
  1. from __future__ import annotations
  2. import importlib
  3. import logging
  4. import unicodedata
  5. from bisect import bisect_right
  6. from codecs import IncrementalDecoder
  7. from functools import lru_cache
  8. from typing import Generator
  9. from .constant import (
  10. ENCODING_MARKS,
  11. IANA_SUPPORTED_SIMILAR,
  12. RE_POSSIBLE_ENCODING_INDICATION,
  13. UNICODE_RANGES_COMBINED,
  14. _SECONDARY_RANGE_NAMES,
  15. COMMON_CJK_CHARACTERS,
  16. _LATIN,
  17. _CJK,
  18. _HANGUL,
  19. _KATAKANA,
  20. _HIRAGANA,
  21. _HALFWIDTH_KATAKANA,
  22. _THAI,
  23. _ARABIC,
  24. _ARABIC_ISOLATED_FORM,
  25. _LIGATURE,
  26. _SUPERSCRIPT,
  27. _SENTENCE_OPEN_PUNCTUATION,
  28. _ACCENT_KEYWORDS,
  29. _ACCENTUATED,
  30. _KNOWN_MB_DECODERS,
  31. _KNOWN_MB_CLASSES,
  32. _IANA_NAMES,
  33. _MULTIBYTE_SEARCH_RADIUS,
  34. )
  35. def _character_flags(character: str) -> int:
  36. """Compute all name-based classification flags with a single unicodedata.name() call."""
  37. try:
  38. desc: str = unicodedata.name(character)
  39. except ValueError:
  40. return 0
  41. flags: int = 0
  42. if "LATIN" in desc:
  43. flags |= _LATIN
  44. if "CJK" in desc:
  45. flags |= _CJK
  46. if "HANGUL" in desc:
  47. flags |= _HANGUL
  48. if "KATAKANA" in desc:
  49. flags |= _KATAKANA
  50. if "HALFWIDTH" in desc:
  51. flags |= _HALFWIDTH_KATAKANA
  52. if "HIRAGANA" in desc:
  53. flags |= _HIRAGANA
  54. if "THAI" in desc:
  55. flags |= _THAI
  56. if "ARABIC" in desc:
  57. flags |= _ARABIC
  58. if "ISOLATED FORM" in desc:
  59. flags |= _ARABIC_ISOLATED_FORM
  60. if "LIGATURE" in desc or desc.endswith("LETTER AE"):
  61. flags |= _LIGATURE
  62. if "SUPERSCRIPT" in desc:
  63. flags |= _SUPERSCRIPT
  64. if desc in {"INVERTED QUESTION MARK", "INVERTED EXCLAMATION MARK"}:
  65. flags |= _SENTENCE_OPEN_PUNCTUATION
  66. for kw in _ACCENT_KEYWORDS:
  67. if kw in desc:
  68. flags |= _ACCENTUATED
  69. break
  70. return flags
  71. def is_accentuated(character: str) -> bool:
  72. return bool(_character_flags(character) & _ACCENTUATED)
  73. def remove_accent(character: str) -> str:
  74. decomposed: str = unicodedata.decomposition(character)
  75. if not decomposed:
  76. return character
  77. codes: list[str] = decomposed.split(" ")
  78. return chr(int(codes[0], 16))
  79. # Pre-built sorted lookup table for O(log n) binary search in unicode_range().
  80. # Each entry is (range_start, range_end_exclusive, range_name).
  81. _UNICODE_RANGES_SORTED: list[tuple[int, int, str]] = sorted(
  82. (ord_range.start, ord_range.stop, name)
  83. for name, ord_range in UNICODE_RANGES_COMBINED.items()
  84. )
  85. _UNICODE_RANGE_STARTS: list[int] = [e[0] for e in _UNICODE_RANGES_SORTED]
  86. def unicode_range(character: str) -> str | None:
  87. """
  88. Retrieve the Unicode range official name from a single character.
  89. """
  90. character_ord: int = ord(character)
  91. if character_ord < 32:
  92. return "Control character"
  93. if character_ord < 128:
  94. return "Basic Latin"
  95. # Binary search: find the rightmost range whose start <= character_ord
  96. idx = bisect_right(_UNICODE_RANGE_STARTS, character_ord) - 1
  97. if idx >= 0:
  98. start, stop, name = _UNICODE_RANGES_SORTED[idx]
  99. if character_ord < stop:
  100. return name
  101. return None
  102. def is_latin(character: str) -> bool:
  103. return bool(_character_flags(character) & _LATIN)
  104. def is_punctuation(character: str) -> bool:
  105. character_category: str = unicodedata.category(character)
  106. if "P" in character_category:
  107. return True
  108. character_range: str | None = unicode_range(character)
  109. if character_range is None:
  110. return False
  111. return "Punctuation" in character_range
  112. def is_symbol(character: str) -> bool:
  113. character_category: str = unicodedata.category(character)
  114. if "S" in character_category or "N" in character_category:
  115. return True
  116. character_range: str | None = unicode_range(character)
  117. if character_range is None:
  118. return False
  119. return "Forms" in character_range and character_category != "Lo"
  120. def is_emoticon(character: str) -> bool:
  121. character_range: str | None = unicode_range(character)
  122. if character_range is None:
  123. return False
  124. return "Emoticons" in character_range or "Pictographs" in character_range
  125. def is_separator(character: str) -> bool:
  126. if character.isspace() or character in {"|", "+", "<", ">"}:
  127. return True
  128. character_category: str = unicodedata.category(character)
  129. return "Z" in character_category or character_category in {"Po", "Pd", "Pc"}
  130. def is_case_variable(character: str) -> bool:
  131. return character.islower() != character.isupper()
  132. def is_cjk(character: str) -> bool:
  133. return bool(_character_flags(character) & _CJK)
  134. def is_hiragana(character: str) -> bool:
  135. return bool(_character_flags(character) & _HIRAGANA)
  136. def is_katakana(character: str) -> bool:
  137. return bool(_character_flags(character) & _KATAKANA)
  138. def is_hangul(character: str) -> bool:
  139. return bool(_character_flags(character) & _HANGUL)
  140. def is_thai(character: str) -> bool:
  141. return bool(_character_flags(character) & _THAI)
  142. def is_arabic(character: str) -> bool:
  143. return bool(_character_flags(character) & _ARABIC)
  144. def is_arabic_isolated_form(character: str) -> bool:
  145. return bool(_character_flags(character) & _ARABIC_ISOLATED_FORM)
  146. def is_cjk_uncommon(character: str) -> bool:
  147. return character not in COMMON_CJK_CHARACTERS
  148. def is_unicode_range_secondary(range_name: str) -> bool:
  149. return range_name in _SECONDARY_RANGE_NAMES
  150. def is_unprintable(character: str) -> bool:
  151. return (
  152. not character.isspace() # includes \n \t \r \v
  153. and not character.isprintable()
  154. and character != "\x1a" # Why? Its the ASCII substitute character.
  155. and character != "\ufeff" # bug discovered in Python,
  156. # Zero Width No-Break Space located in Arabic Presentation Forms-B, Unicode 1.1 not acknowledged as space.
  157. )
  158. def any_specified_encoding(
  159. sequence: bytes | bytearray, search_zone: int = 8192
  160. ) -> str | None:
  161. """
  162. Extract using ASCII-only decoder any specified encoding in the first n-bytes.
  163. """
  164. if not isinstance(sequence, (bytes, bytearray)):
  165. raise TypeError
  166. seq_len: int = len(sequence)
  167. # Cheap literal pre-filter.
  168. search_bytes = sequence[: min(seq_len, search_zone)]
  169. lowered_bytes = search_bytes.lower()
  170. if b"coding" not in lowered_bytes and b"charset" not in lowered_bytes:
  171. return None
  172. decoded_zone: str = search_bytes.decode("ascii", errors="ignore")
  173. for match in RE_POSSIBLE_ENCODING_INDICATION.finditer(decoded_zone):
  174. specified_encoding = match.group(1).lower().replace("-", "_")
  175. encoding_iana = _IANA_NAMES.get(specified_encoding)
  176. if encoding_iana is not None:
  177. return encoding_iana
  178. return None
  179. @lru_cache(maxsize=None)
  180. def is_multi_byte_encoding(name: str) -> bool:
  181. """
  182. Verify is a specific encoding is a multi byte one based on it IANA name
  183. """
  184. if name in _KNOWN_MB_DECODERS:
  185. return True
  186. # Besides the Unicode family above, every multibyte codec shipped with
  187. # Python is implemented by _multibytecodec through exactly one of the six
  188. # cjkcodecs providers below. Probing those providers directly (getcodec)
  189. # classifies a name without importing its "encodings.<name>" module:
  190. # classifying the whole IANA_SUPPORTED list would otherwise import many
  191. # modules and dominate "import charset_normalizer" wall time.
  192. # see https://github.com/jawah/charset_normalizer/issues/742
  193. for provider in _KNOWN_MB_CLASSES:
  194. try:
  195. importlib.import_module(provider).getcodec(name) # type: ignore[attr-defined]
  196. except (ImportError, AttributeError, LookupError): # Defensive: edge cases
  197. continue
  198. return True
  199. return False
  200. def identify_sig_or_bom(sequence: bytes | bytearray) -> tuple[str | None, bytes]:
  201. """
  202. Identify and extract SIG/BOM in given sequence.
  203. """
  204. for iana_encoding in ENCODING_MARKS:
  205. marks: bytes | list[bytes] = ENCODING_MARKS[iana_encoding]
  206. if isinstance(marks, bytes):
  207. marks = [marks]
  208. for mark in marks:
  209. if sequence.startswith(mark):
  210. return iana_encoding, mark
  211. return None, b""
  212. def should_strip_sig_or_bom(iana_encoding: str) -> bool:
  213. return iana_encoding not in {"utf_16", "utf_32"}
  214. def iana_name(cp_name: str, strict: bool = True) -> str:
  215. """Returns the Python normalized encoding name (Not the IANA official name)."""
  216. cp_name = cp_name.lower().replace("-", "_")
  217. encoding_iana = _IANA_NAMES.get(cp_name)
  218. if encoding_iana is not None:
  219. return encoding_iana
  220. if strict:
  221. raise ValueError(f"Unable to retrieve IANA for '{cp_name}'")
  222. return cp_name
  223. def cp_similarity(iana_name_a: str, iana_name_b: str) -> float:
  224. if is_multi_byte_encoding(iana_name_a) or is_multi_byte_encoding(iana_name_b):
  225. return 0.0
  226. decoder_a = importlib.import_module(f"encodings.{iana_name_a}").IncrementalDecoder
  227. decoder_b = importlib.import_module(f"encodings.{iana_name_b}").IncrementalDecoder
  228. id_a: IncrementalDecoder = decoder_a(errors="ignore")
  229. id_b: IncrementalDecoder = decoder_b(errors="ignore")
  230. character_match_count: int = 0
  231. for i in range(256):
  232. to_be_decoded: bytes = bytes([i])
  233. if id_a.decode(to_be_decoded) == id_b.decode(to_be_decoded):
  234. character_match_count += 1
  235. return character_match_count / 256
  236. def is_cp_similar(iana_name_a: str, iana_name_b: str) -> bool:
  237. """
  238. Determine if two code page are at least 80% similar. IANA_SUPPORTED_SIMILAR dict was generated using
  239. the function cp_similarity.
  240. """
  241. return (
  242. iana_name_a in IANA_SUPPORTED_SIMILAR
  243. and iana_name_b in IANA_SUPPORTED_SIMILAR[iana_name_a]
  244. )
  245. def set_logging_handler(
  246. name: str = "charset_normalizer",
  247. level: int = logging.INFO,
  248. format_string: str = "%(asctime)s | %(levelname)s | %(message)s",
  249. ) -> None:
  250. logger = logging.getLogger(name)
  251. logger.setLevel(level)
  252. handler = logging.StreamHandler()
  253. handler.setFormatter(logging.Formatter(format_string))
  254. logger.addHandler(handler)
  255. def cut_sequence_chunks(
  256. sequences: bytes | bytearray,
  257. encoding_iana: str,
  258. offsets: range,
  259. chunk_size: int,
  260. bom_or_sig_available: bool,
  261. strip_sig_or_bom: bool,
  262. sig_payload: bytes,
  263. is_multi_byte_decoder: bool,
  264. decoded_payload: str | None = None,
  265. deferred_decoding: bool = False,
  266. ) -> Generator[str, None, None]:
  267. # iso2022 codec is stateful, generic mb cuter isn't going to cut it!
  268. if decoded_payload and encoding_iana.startswith("iso2022_"):
  269. decoded_length = len(decoded_payload)
  270. sequence_length = len(sequences)
  271. for i in offsets:
  272. decoded_offset = i * decoded_length // sequence_length
  273. chunk = decoded_payload[decoded_offset : decoded_offset + chunk_size]
  274. if not chunk:
  275. break
  276. yield chunk
  277. elif decoded_payload and not is_multi_byte_decoder:
  278. for i in offsets:
  279. chunk = decoded_payload[i : i + chunk_size]
  280. if not chunk:
  281. break
  282. yield chunk
  283. elif deferred_decoding:
  284. # Deferred single-byte probing: the whole payload is not decoded
  285. # yet. Single-byte codecs are stateless (1 byte == 1 char), hence
  286. # decode(base)[i:j] == decode(base[i:j]): slicing the raw bytes
  287. # yields exactly the chunks the branch above would have produced,
  288. # short trailing chunks included, and raises UnicodeDecodeError on
  289. # invalid bytes just like the whole-payload decode would.
  290. base_bytes = (
  291. sequences if not strip_sig_or_bom else sequences[len(sig_payload) :]
  292. )
  293. for i in offsets:
  294. cut_sequence = base_bytes[i : i + chunk_size]
  295. if not cut_sequence:
  296. break
  297. yield str(cut_sequence, encoding_iana)
  298. else:
  299. for i in offsets:
  300. chunk_end = i + chunk_size
  301. if chunk_end > len(sequences) + 8:
  302. continue
  303. cut_sequence = sequences[i : i + chunk_size]
  304. if bom_or_sig_available and not strip_sig_or_bom:
  305. cut_sequence = sig_payload + cut_sequence
  306. chunk = cut_sequence.decode(
  307. encoding_iana,
  308. errors="ignore" if is_multi_byte_decoder else "strict",
  309. )
  310. # multi-byte bad cutting detector and adjustment
  311. # not the cleanest way to perform that fix but clever enough for now.
  312. if is_multi_byte_decoder and i > 0:
  313. chunk_partial_size_chk: int = min(chunk_size, 16)
  314. chunk_prefix = chunk[:chunk_partial_size_chk]
  315. found_nearby = False
  316. if decoded_payload:
  317. decoded_length = len(decoded_payload)
  318. expected_offset = i * decoded_length // len(sequences)
  319. search_start = max(0, expected_offset - _MULTIBYTE_SEARCH_RADIUS)
  320. search_end = min(
  321. decoded_length, expected_offset + _MULTIBYTE_SEARCH_RADIUS
  322. )
  323. found_nearby = (
  324. decoded_payload.find(chunk_prefix, search_start, search_end)
  325. >= 0
  326. )
  327. if (
  328. decoded_payload
  329. and not found_nearby
  330. and chunk_prefix not in decoded_payload
  331. ):
  332. for j in range(i, i - 4, -1):
  333. cut_sequence = sequences[j:chunk_end]
  334. if bom_or_sig_available and not strip_sig_or_bom:
  335. cut_sequence = sig_payload + cut_sequence
  336. chunk = cut_sequence.decode(encoding_iana, errors="ignore")
  337. if chunk[:chunk_partial_size_chk] in decoded_payload:
  338. break
  339. yield chunk