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