Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import logging
4from functools import lru_cache
5from os import PathLike
6from typing import BinaryIO
8from .cd import (
9 coherence_ratio,
10 encoding_languages,
11 mb_encoding_languages,
12 merge_coherence_ratios,
13)
14from .constant import (
15 IANA_SUPPORTED,
16 IANA_SUPPORTED_SIMILAR,
17 TOO_BIG_SEQUENCE,
18 TOO_SMALL_SEQUENCE,
19 TRACE,
20)
21from .md import mess_ratio
22from .models import CharsetMatch, CharsetMatches
23from .utils import (
24 any_specified_encoding,
25 cut_sequence_chunks,
26 iana_name,
27 identify_sig_or_bom,
28 is_multi_byte_encoding,
29 should_strip_sig_or_bom,
30)
32logger = logging.getLogger("charset_normalizer")
33explain_handler = logging.StreamHandler()
34explain_handler.setFormatter(
35 logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")
36)
38# Pre-compute a reordered encoding list: multibyte first, then single-byte.
39# This allows the mb_definitive_match optimization to fire earlier, skipping
40# all single-byte encodings for genuine CJK content. Multibyte codecs
41# hard-fail (UnicodeDecodeError) on single-byte data almost instantly, so
42# testing them first costs negligible time for non-CJK files.
43# Stable sort on a boolean key: multibyte (False) first, IANA order kept
44# within each group.
45IANA_SUPPORTED_MB_FIRST: list[str] = sorted(
46 IANA_SUPPORTED, key=lambda encoding: not is_multi_byte_encoding(encoding)
47)
50def from_bytes(
51 sequences: bytes | bytearray,
52 steps: int = 5,
53 chunk_size: int = 512,
54 threshold: float = 0.2,
55 cp_isolation: list[str] | None = None,
56 cp_exclusion: list[str] | None = None,
57 preemptive_behaviour: bool = True,
58 explain: bool = False,
59 language_threshold: float = 0.1,
60 enable_fallback: bool = True,
61) -> CharsetMatches:
62 """
63 Given a raw bytes sequence, return the best possibles charset usable to render str objects.
64 If there is no results, it is a strong indicator that the source is binary/not text.
65 By default, the process will extract 5 blocks of 512o each to assess the mess and coherence of a given sequence.
66 And will give up a particular code page after 20% of measured mess. Those criteria are customizable at will.
68 The preemptive behavior DOES NOT replace the traditional detection workflow, it prioritize a particular code page
69 but never take it for granted. Can improve the performance.
71 You may want to focus your attention to some code page or/and not others, use cp_isolation and cp_exclusion for that
72 purpose.
74 This function will strip the SIG in the payload/sequence every time except on UTF-16, UTF-32.
75 By default the library does not setup any handler other than the NullHandler, if you choose to set the 'explain'
76 toggle to True it will alter the logger configuration to add a StreamHandler that is suitable for debugging.
77 Custom logging format and handler can be set manually.
78 """
80 if not isinstance(sequences, (bytearray, bytes)):
81 raise TypeError(
82 "Expected object of type bytes or bytearray, got: {}".format(
83 type(sequences)
84 )
85 )
87 if explain:
88 previous_logger_level: int = logger.level
89 logger.addHandler(explain_handler)
90 logger.setLevel(TRACE)
92 length: int = len(sequences)
94 if length == 0:
95 logger.debug("Encoding detection on empty bytes, assuming utf_8 intention.")
96 if explain: # Defensive: ensure exit path clean handler
97 logger.removeHandler(explain_handler)
98 logger.setLevel(previous_logger_level)
99 return CharsetMatches([CharsetMatch(sequences, "utf_8", 0.0, False, [], "")])
101 if cp_isolation is not None:
102 logger.log(
103 TRACE,
104 "cp_isolation is set. use this flag for debugging purpose. "
105 "limited list of encoding allowed : %s.",
106 ", ".join(cp_isolation),
107 )
108 cp_isolation = [iana_name(cp, False) for cp in cp_isolation]
109 else:
110 cp_isolation = []
112 if cp_exclusion is not None:
113 logger.log(
114 TRACE,
115 "cp_exclusion is set. use this flag for debugging purpose. "
116 "limited list of encoding excluded : %s.",
117 ", ".join(cp_exclusion),
118 )
119 cp_exclusion = [iana_name(cp, False) for cp in cp_exclusion]
120 else:
121 cp_exclusion = []
123 if length <= (chunk_size * steps):
124 logger.log(
125 TRACE,
126 "override steps (%i) and chunk_size (%i) as content does not fit (%i byte(s) given) parameters.",
127 steps,
128 chunk_size,
129 length,
130 )
131 steps = 1
132 chunk_size = length
134 if steps > 1 and length / steps < chunk_size:
135 chunk_size = int(length / steps)
137 is_too_small_sequence: bool = len(sequences) < TOO_SMALL_SEQUENCE
138 is_too_large_sequence: bool = len(sequences) >= TOO_BIG_SEQUENCE
140 if is_too_small_sequence:
141 logger.log(
142 TRACE,
143 "Trying to detect encoding from a tiny portion of (%i) byte(s).",
144 length,
145 )
146 elif is_too_large_sequence:
147 logger.log(
148 TRACE,
149 "Using lazy str decoding because the payload is quite large, (%i) byte(s).",
150 length,
151 )
153 prioritized_encodings: list[str] = []
155 specified_encoding: str | None = (
156 any_specified_encoding(sequences) if preemptive_behaviour else None
157 )
159 if specified_encoding is not None:
160 prioritized_encodings.append(specified_encoding)
161 logger.log(
162 TRACE,
163 "Detected declarative mark in sequence. Priority +1 given for %s.",
164 specified_encoding,
165 )
167 tested: set[str] = set()
168 tested_but_hard_failure: list[str] = []
169 tested_but_soft_failure: list[str] = []
170 soft_failure_skip: set[str] = set()
172 # Avoid unoptimized RSS usage.
173 # this cache is mostly interesting for
174 # local usage. Garbage collected at the
175 # end. Like it should.
176 cached_mess_ratio = lru_cache(maxsize=None)(mess_ratio)
177 cached_coherence_ratio = lru_cache(maxsize=None)(coherence_ratio)
179 # When a definitive result (chaos=0.0 and good coherence) is found after testing
180 # the prioritized encodings (ascii, utf_8), we can significantly reduce the remaining
181 # work. Encodings that target completely different language families (e.g., Cyrillic
182 # when the definitive match is Latin) are skipped entirely.
183 # Additionally, for same-family encodings that pass chaos probing, we reuse the
184 # definitive match's coherence ratios instead of recomputing them — a major savings
185 # since coherence_ratio accounts for ~30% of total time on slow Latin files.
186 definitive_match_found: bool = False
187 definitive_target_languages: set[str] = set()
188 # After the definitive match fires, we cap the number of additional same-family
189 # single-byte encodings that pass chaos probing. Once we've accumulated enough
190 # good candidates (N), further same-family SB encodings are unlikely to produce
191 # a better best() result and just waste mess_ratio + coherence_ratio time.
192 # The first encoding to trigger the definitive match is NOT counted (it's already in).
193 post_definitive_sb_success_count: int = 0
194 POST_DEFINITIVE_SB_CAP: int = 7
196 # When a non-UTF multibyte encoding passes chaos probing with significant multibyte
197 # content (decoded length < 98% of raw length), skip all remaining single-byte encodings.
198 # Rationale: multi-byte decoders (CJK) have strict byte-sequence validation — if they
199 # decode without error AND pass chaos probing with substantial multibyte content, the
200 # data is genuinely multibyte encoded. Single-byte encodings will always decode (every
201 # byte maps to something) but waste time on mess_ratio before failing.
202 # The 98% threshold prevents false triggers on files that happen to have a few valid
203 # multibyte pairs (e.g., cp424/_ude_1.txt where big5 decodes with 99% ratio).
204 mb_definitive_match_found: bool = False
206 fallback_ascii: CharsetMatch | None = None
207 fallback_u8: CharsetMatch | None = None
208 fallback_specified: CharsetMatch | None = None
210 results: CharsetMatches = CharsetMatches()
212 early_stop_results: CharsetMatches = CharsetMatches()
214 sig_encoding, sig_payload = identify_sig_or_bom(sequences)
216 if sig_encoding is not None:
217 prioritized_encodings.insert(0, sig_encoding)
218 logger.log(
219 TRACE,
220 "Detected a SIG or BOM mark on first %i byte(s). Priority +1 given for %s.",
221 len(sig_payload),
222 sig_encoding,
223 )
225 prioritized_encodings.append("ascii")
227 if "utf_8" not in prioritized_encodings:
228 prioritized_encodings.append("utf_8")
230 for encoding_iana in prioritized_encodings + IANA_SUPPORTED_MB_FIRST:
231 if cp_isolation and encoding_iana not in cp_isolation:
232 continue
234 if cp_exclusion and encoding_iana in cp_exclusion:
235 continue
237 if encoding_iana in tested:
238 continue
240 tested.add(encoding_iana)
242 decoded_payload: str | None = None
243 bom_or_sig_available: bool = sig_encoding == encoding_iana
244 strip_sig_or_bom: bool = bom_or_sig_available and should_strip_sig_or_bom(
245 encoding_iana
246 )
248 if encoding_iana in {"utf_16", "utf_32"} and not bom_or_sig_available:
249 logger.log(
250 TRACE,
251 "Encoding %s won't be tested as-is because it require a BOM. Will try some sub-encoder LE/BE.",
252 encoding_iana,
253 )
254 continue
255 if encoding_iana in {"utf_7"} and not bom_or_sig_available:
256 logger.log(
257 TRACE,
258 "Encoding %s won't be tested as-is because detection is unreliable without BOM/SIG.",
259 encoding_iana,
260 )
261 continue
263 # Skip encodings similar to ones that already soft-failed (high mess ratio).
264 # Checked BEFORE the expensive decode attempt.
265 if encoding_iana in soft_failure_skip:
266 logger.log(
267 TRACE,
268 "%s is deemed too similar to a code page that was already considered unsuited. Continuing!",
269 encoding_iana,
270 )
271 continue
273 try:
274 is_multi_byte_decoder: bool = is_multi_byte_encoding(encoding_iana)
275 except (ModuleNotFoundError, ImportError): # Defensive:
276 logger.log(
277 TRACE,
278 "Encoding %s does not provide an IncrementalDecoder",
279 encoding_iana,
280 )
281 continue
283 # When we've already found a definitive match (chaos=0.0 with good coherence)
284 # after testing the prioritized encodings, skip encodings that target
285 # completely different language families. This avoids running expensive
286 # mess_ratio + coherence_ratio on clearly unrelated candidates (e.g., Cyrillic
287 # when the definitive match is Latin-based).
288 if definitive_match_found:
289 if not is_multi_byte_decoder:
290 enc_languages = set(encoding_languages(encoding_iana))
291 else:
292 enc_languages = set(mb_encoding_languages(encoding_iana))
293 if not enc_languages.intersection(definitive_target_languages):
294 logger.log(
295 TRACE,
296 "Skipping %s: definitive match already found, this encoding targets different languages (%s vs %s).",
297 encoding_iana,
298 enc_languages,
299 definitive_target_languages,
300 )
301 continue
303 # After the definitive match, cap the number of additional same-family
304 # single-byte encodings that pass chaos probing. This avoids testing the
305 # tail of rare, low-value same-family encodings (mac_iceland, cp860, etc.)
306 # that almost never change best() but each cost ~1-2ms of mess_ratio + coherence.
307 if (
308 definitive_match_found
309 and not is_multi_byte_decoder
310 and post_definitive_sb_success_count >= POST_DEFINITIVE_SB_CAP
311 ):
312 logger.log(
313 TRACE,
314 "Skipping %s: already accumulated %d same-family results after definitive match (cap=%d).",
315 encoding_iana,
316 post_definitive_sb_success_count,
317 POST_DEFINITIVE_SB_CAP,
318 )
319 continue
321 # When a multibyte encoding with significant multibyte content has already
322 # passed chaos probing, skip all single-byte encodings. They will either fail
323 # chaos probing (wasting mess_ratio time) or produce inferior results.
324 if mb_definitive_match_found and not is_multi_byte_decoder:
325 logger.log(
326 TRACE,
327 "Skipping single-byte %s: multi-byte definitive match already found.",
328 encoding_iana,
329 )
330 continue
332 # Single-byte candidates of regular size defer the expensive whole
333 # payload decode until after chunk probing: single-byte codecs are
334 # stateless (1 byte == 1 char) so decoding chunk slices is provably
335 # identical to slicing the decoded payload, and candidates rejected
336 # by chaos probing (the common case) never pay the full decode nor
337 # the payload hash.
338 deferred_decoding: bool = (
339 not is_multi_byte_decoder and not is_too_large_sequence
340 )
342 try:
343 if is_too_large_sequence and not is_multi_byte_decoder:
344 str(
345 (
346 sequences[: int(50e4)]
347 if not strip_sig_or_bom
348 else sequences[len(sig_payload) : int(50e4)]
349 ),
350 encoding=encoding_iana,
351 )
352 elif not deferred_decoding:
353 # UTF-7 BOM is encoded in modified Base64 whose byte boundary
354 # can overlap with the next character. Stripping raw SIG bytes
355 # before decoding may leave stray bytes that decode as garbage.
356 # Decode the full sequence and remove the leading BOM char instead.
357 # see https://github.com/jawah/charset_normalizer/issues/718
358 # and https://github.com/jawah/charset_normalizer/issues/716
359 if encoding_iana == "utf_7" and bom_or_sig_available:
360 decoded_payload = str(
361 sequences,
362 encoding=encoding_iana,
363 )
364 if decoded_payload and decoded_payload[0] == "\ufeff":
365 decoded_payload = decoded_payload[1:]
366 else:
367 decoded_payload = str(
368 (
369 sequences
370 if not strip_sig_or_bom
371 else sequences[len(sig_payload) :]
372 ),
373 encoding=encoding_iana,
374 )
375 except (UnicodeDecodeError, LookupError) as e:
376 if not isinstance(e, LookupError):
377 logger.log(
378 TRACE,
379 "Code page %s does not fit given bytes sequence at ALL. %s",
380 encoding_iana,
381 str(e),
382 )
383 tested_but_hard_failure.append(encoding_iana)
384 continue
386 r_ = range(
387 0 if not bom_or_sig_available else len(sig_payload),
388 length,
389 int(length / steps),
390 )
392 multi_byte_bonus: bool = (
393 is_multi_byte_decoder
394 and decoded_payload is not None
395 and len(decoded_payload) < length
396 )
398 if multi_byte_bonus:
399 logger.log(
400 TRACE,
401 "Code page %s is a multi byte encoding table and it appear that at least one character "
402 "was encoded using n-bytes.",
403 encoding_iana,
404 )
406 max_chunk_gave_up: int = int(len(r_) / 4)
408 max_chunk_gave_up = max(max_chunk_gave_up, 2)
409 early_stop_count: int = 0
410 lazy_str_hard_failure = False
412 md_chunks: list[str] = []
413 md_ratios = []
415 try:
416 for chunk in cut_sequence_chunks(
417 sequences,
418 encoding_iana,
419 r_,
420 chunk_size,
421 bom_or_sig_available,
422 strip_sig_or_bom,
423 sig_payload,
424 is_multi_byte_decoder,
425 decoded_payload,
426 deferred_decoding,
427 ):
428 md_chunks.append(chunk)
430 md_ratios.append(
431 cached_mess_ratio(
432 chunk,
433 threshold,
434 explain and 1 <= len(cp_isolation) <= 2,
435 )
436 )
438 if md_ratios[-1] >= threshold:
439 early_stop_count += 1
441 if (early_stop_count >= max_chunk_gave_up) or (
442 bom_or_sig_available and not strip_sig_or_bom
443 ):
444 break
445 except (
446 UnicodeDecodeError,
447 LookupError,
448 ) as e: # Lazy str loading may have missed something there
449 if deferred_decoding:
450 # Deferred single-byte validation failed on a chunk (or the
451 # codec is unavailable on this interpreter build): identical
452 # outcome and bookkeeping to the eager full-decode failure.
453 logger.log(
454 TRACE,
455 "Code page %s does not fit given bytes sequence at ALL. %s",
456 encoding_iana,
457 str(e),
458 )
459 tested_but_hard_failure.append(encoding_iana)
460 continue
461 logger.log(
462 TRACE,
463 "LazyStr Loading: After MD chunk decode, code page %s does not fit given bytes sequence at ALL. %s",
464 encoding_iana,
465 str(e),
466 )
467 early_stop_count = max_chunk_gave_up
468 lazy_str_hard_failure = True
470 # We might want to check the sequence again with the whole content
471 # Only if initial MD tests passes
472 if (
473 not lazy_str_hard_failure
474 and is_too_large_sequence
475 and not is_multi_byte_decoder
476 ):
477 try:
478 sequences[int(50e3) :].decode(encoding_iana, errors="strict")
479 except UnicodeDecodeError as e:
480 logger.log(
481 TRACE,
482 "LazyStr Loading: After final lookup, code page %s does not fit given bytes sequence at ALL. %s",
483 encoding_iana,
484 str(e),
485 )
486 tested_but_hard_failure.append(encoding_iana)
487 continue
489 mean_mess_ratio: float = sum(md_ratios) / len(md_ratios) if md_ratios else 0.0
490 if mean_mess_ratio >= threshold or early_stop_count >= max_chunk_gave_up:
491 tested_but_soft_failure.append(encoding_iana)
492 if encoding_iana in IANA_SUPPORTED_SIMILAR:
493 soft_failure_skip.update(IANA_SUPPORTED_SIMILAR[encoding_iana])
494 logger.log(
495 TRACE,
496 "%s was excluded because of initial chaos probing. Gave up %i time(s). "
497 "Computed mean chaos is %f %%.",
498 encoding_iana,
499 early_stop_count,
500 round(mean_mess_ratio * 100, ndigits=3),
501 )
502 # Preparing those fallbacks in case we got nothing.
503 if (
504 enable_fallback
505 and encoding_iana
506 in ["ascii", "utf_8", specified_encoding, "utf_16", "utf_32"]
507 and not lazy_str_hard_failure
508 ):
509 # Always fully decode payload before.
510 # We've missed a UnicodeDecodeError proof
511 # while issuing release 3.4.8
512 # see https://github.com/jawah/charset_normalizer/issues/771
513 if decoded_payload is None:
514 try:
515 decoded_payload = str(
516 (
517 sequences
518 if not strip_sig_or_bom
519 else sequences[len(sig_payload) :]
520 ),
521 encoding=encoding_iana,
522 )
523 except (UnicodeDecodeError, LookupError):
524 logger.log(
525 TRACE,
526 "%s does not decode the whole payload: fallback entry withheld.",
527 encoding_iana,
528 )
529 continue
530 if is_too_large_sequence:
531 # Don't retain huge payload in RAM.
532 decoded_payload = None
534 fallback_entry = CharsetMatch(
535 sequences,
536 encoding_iana,
537 threshold,
538 bom_or_sig_available,
539 [],
540 decoded_payload,
541 preemptive_declaration=specified_encoding,
542 )
543 if encoding_iana == specified_encoding:
544 fallback_specified = fallback_entry
545 elif encoding_iana == "ascii":
546 fallback_ascii = fallback_entry
547 else:
548 fallback_u8 = fallback_entry
549 continue
551 if deferred_decoding:
552 # The candidate passed chaos probing: perform the whole payload
553 # decode (validation + payload reuse) that was deferred earlier.
554 try:
555 decoded_payload = str(
556 (
557 sequences
558 if not strip_sig_or_bom
559 else sequences[len(sig_payload) :]
560 ),
561 encoding=encoding_iana,
562 )
563 except (UnicodeDecodeError, LookupError) as e:
564 logger.log(
565 TRACE,
566 "Code page %s does not fit given bytes sequence at ALL. %s",
567 encoding_iana,
568 str(e),
569 )
570 tested_but_hard_failure.append(encoding_iana)
571 continue
573 logger.log(
574 TRACE,
575 "%s passed initial chaos probing. Mean measured chaos is %f %%",
576 encoding_iana,
577 round(mean_mess_ratio * 100, ndigits=3),
578 )
580 if not is_multi_byte_decoder:
581 target_languages: list[str] = encoding_languages(encoding_iana)
582 else:
583 target_languages = mb_encoding_languages(encoding_iana)
585 if target_languages:
586 logger.log(
587 TRACE,
588 "%s should target any language(s) of %s",
589 encoding_iana,
590 target_languages,
591 )
593 cd_ratios = []
595 # Run coherence detection on all chunks. We previously tried limiting to
596 # 1-2 chunks for post-definitive encodings to save time, but this caused
597 # coverage regressions by producing unrepresentative coherence scores.
598 # The SB cap and language-family skip optimizations provide sufficient
599 # speedup without sacrificing coherence accuracy.
600 if encoding_iana != "ascii":
601 # We shall skip the CD when its about ASCII
602 # Most of the time its not relevant to run "language-detection" on it.
603 lg_inclusion: str | None = (
604 ",".join(target_languages) if target_languages else None
605 )
607 for chunk in md_chunks:
608 chunk_languages = cached_coherence_ratio(
609 chunk,
610 language_threshold,
611 lg_inclusion,
612 )
614 cd_ratios.append(chunk_languages)
616 cd_ratios_merged = merge_coherence_ratios(cd_ratios)
618 if cd_ratios_merged:
619 logger.log(
620 TRACE,
621 "We detected language %s using %s",
622 cd_ratios_merged,
623 encoding_iana,
624 )
626 current_match = CharsetMatch(
627 sequences,
628 encoding_iana,
629 mean_mess_ratio,
630 bom_or_sig_available,
631 cd_ratios_merged,
632 (
633 decoded_payload
634 if (
635 not is_too_large_sequence
636 or encoding_iana in [specified_encoding, "ascii", "utf_8"]
637 )
638 else None
639 ),
640 preemptive_declaration=specified_encoding,
641 )
643 results.append(current_match)
645 # Count post-definitive same-family SB successes for the early termination cap.
646 # Only count low-mess encodings (< 2%) toward the cap. High-mess encodings are
647 # marginal results that shouldn't prevent better-quality candidates from being
648 # tested. For example, iso8859_4 (mess=0%) should not be skipped just because
649 # 7 high-mess Latin encodings (cp1252 at 8%, etc.) were tried first.
650 if (
651 definitive_match_found
652 and not is_multi_byte_decoder
653 and mean_mess_ratio < 0.02
654 ):
655 post_definitive_sb_success_count += 1
657 if (
658 encoding_iana in [specified_encoding, "ascii", "utf_8"]
659 and mean_mess_ratio < 0.1
660 ):
661 # If md says nothing to worry about, then... stop immediately!
662 if mean_mess_ratio == 0.0:
663 logger.debug(
664 "Encoding detection: %s is most likely the one.",
665 current_match.encoding,
666 )
667 if explain: # Defensive: ensure exit path clean handler
668 logger.removeHandler(explain_handler)
669 logger.setLevel(previous_logger_level)
670 return CharsetMatches([current_match])
672 early_stop_results.append(current_match)
674 if (
675 len(early_stop_results)
676 and (specified_encoding is None or specified_encoding in tested)
677 and "ascii" in tested
678 and "utf_8" in tested
679 ):
680 probable_result = early_stop_results.best()
681 assert probable_result is not None
682 logger.debug(
683 "Encoding detection: %s is most likely the one.",
684 probable_result.encoding,
685 )
686 if explain: # Defensive: ensure exit path clean handler
687 logger.removeHandler(explain_handler)
688 logger.setLevel(previous_logger_level)
690 return CharsetMatches([probable_result])
692 # Once we find a result with good coherence (>= 0.5) after testing the
693 # prioritized encodings (ascii, utf_8), activate "definitive mode": skip
694 # encodings that target completely different language families. This avoids
695 # running expensive mess_ratio + coherence_ratio on clearly unrelated
696 # candidates (e.g., Cyrillic encodings when the match is Latin-based).
697 # We require coherence >= 0.5 to avoid false positives (e.g., cp1251 decoding
698 # Hebrew text with 0.0 chaos but wrong language detection at coherence 0.33).
699 if not definitive_match_found and not is_multi_byte_decoder:
700 best_coherence = (
701 max((v for _, v in cd_ratios_merged), default=0.0)
702 if cd_ratios_merged
703 else 0.0
704 )
705 if best_coherence >= 0.5 and "ascii" in tested and "utf_8" in tested:
706 definitive_match_found = True
707 definitive_target_languages.update(target_languages)
708 logger.log(
709 TRACE,
710 "Definitive match found: %s (chaos=%.3f, coherence=%.2f). Encodings targeting different language families will be skipped.",
711 encoding_iana,
712 mean_mess_ratio,
713 best_coherence,
714 )
716 # When a non-UTF multibyte encoding passes chaos probing with significant
717 # multibyte content (decoded < 98% of raw), activate mb_definitive_match.
718 # This skips all remaining single-byte encodings which would either soft-fail
719 # (running expensive mess_ratio for nothing) or produce inferior results.
720 if (
721 not mb_definitive_match_found
722 and is_multi_byte_decoder
723 and multi_byte_bonus
724 and decoded_payload is not None
725 and len(decoded_payload) < length * 0.98
726 and encoding_iana
727 not in {
728 "utf_8",
729 "utf_8_sig",
730 "utf_16",
731 "utf_16_be",
732 "utf_16_le",
733 "utf_32",
734 "utf_32_be",
735 "utf_32_le",
736 "utf_7",
737 }
738 and "ascii" in tested
739 and "utf_8" in tested
740 ):
741 mb_definitive_match_found = True
742 logger.log(
743 TRACE,
744 "Multi-byte definitive match: %s (chaos=%.3f, decoded=%d/%d=%.1f%%). Single-byte encodings will be skipped.",
745 encoding_iana,
746 mean_mess_ratio,
747 len(decoded_payload),
748 length,
749 len(decoded_payload) / length * 100,
750 )
752 if encoding_iana == sig_encoding:
753 logger.debug(
754 "Encoding detection: %s is most likely the one as we detected a BOM or SIG within "
755 "the beginning of the sequence.",
756 encoding_iana,
757 )
758 if explain: # Defensive: ensure exit path clean handler
759 logger.removeHandler(explain_handler)
760 logger.setLevel(previous_logger_level)
761 return CharsetMatches([results[encoding_iana]])
763 if len(results) == 0:
764 if fallback_u8 or fallback_ascii or fallback_specified:
765 logger.log(
766 TRACE,
767 "Nothing got out of the detection process. Using ASCII/UTF-8/Specified fallback.",
768 )
770 if fallback_specified:
771 logger.debug(
772 "Encoding detection: %s will be used as a fallback match",
773 fallback_specified.encoding,
774 )
775 results.append(fallback_specified)
776 elif (
777 (fallback_u8 and fallback_ascii is None)
778 or (
779 fallback_u8
780 and fallback_ascii
781 and fallback_u8.fingerprint != fallback_ascii.fingerprint
782 )
783 or (fallback_u8 is not None)
784 ):
785 logger.debug("Encoding detection: utf_8 will be used as a fallback match")
786 results.append(fallback_u8)
787 elif fallback_ascii:
788 logger.debug("Encoding detection: ascii will be used as a fallback match")
789 results.append(fallback_ascii)
791 if results:
792 logger.debug(
793 "Encoding detection: Found %s as plausible (best-candidate) for content. With %i alternatives.",
794 results.best().encoding, # type: ignore
795 len(results) - 1,
796 )
797 else:
798 logger.debug("Encoding detection: Unable to determine any suitable charset.")
800 if explain:
801 logger.removeHandler(explain_handler)
802 logger.setLevel(previous_logger_level)
804 return results
807def from_fp(
808 fp: BinaryIO,
809 steps: int = 5,
810 chunk_size: int = 512,
811 threshold: float = 0.20,
812 cp_isolation: list[str] | None = None,
813 cp_exclusion: list[str] | None = None,
814 preemptive_behaviour: bool = True,
815 explain: bool = False,
816 language_threshold: float = 0.1,
817 enable_fallback: bool = True,
818) -> CharsetMatches:
819 """
820 Same thing than the function from_bytes but using a file pointer that is already ready.
821 Will not close the file pointer.
822 """
823 return from_bytes(
824 fp.read(),
825 steps,
826 chunk_size,
827 threshold,
828 cp_isolation,
829 cp_exclusion,
830 preemptive_behaviour,
831 explain,
832 language_threshold,
833 enable_fallback,
834 )
837def from_path(
838 path: str | bytes | PathLike, # type: ignore[type-arg]
839 steps: int = 5,
840 chunk_size: int = 512,
841 threshold: float = 0.20,
842 cp_isolation: list[str] | None = None,
843 cp_exclusion: list[str] | None = None,
844 preemptive_behaviour: bool = True,
845 explain: bool = False,
846 language_threshold: float = 0.1,
847 enable_fallback: bool = True,
848) -> CharsetMatches:
849 """
850 Same thing than the function from_bytes but with one extra step. Opening and reading given file path in binary mode.
851 Can raise IOError.
852 """
853 with open(path, "rb") as fp:
854 return from_fp(
855 fp,
856 steps,
857 chunk_size,
858 threshold,
859 cp_isolation,
860 cp_exclusion,
861 preemptive_behaviour,
862 explain,
863 language_threshold,
864 enable_fallback,
865 )
868def is_binary(
869 fp_or_path_or_payload: PathLike | str | BinaryIO | bytes, # type: ignore[type-arg]
870 steps: int = 5,
871 chunk_size: int = 512,
872 threshold: float = 0.20,
873 cp_isolation: list[str] | None = None,
874 cp_exclusion: list[str] | None = None,
875 preemptive_behaviour: bool = True,
876 explain: bool = False,
877 language_threshold: float = 0.1,
878 enable_fallback: bool = False,
879) -> bool:
880 """
881 Detect if the given input (file, bytes, or path) points to a binary file. aka. not a string.
882 Based on the same main heuristic algorithms and default kwargs at the sole exception that fallbacks match
883 are disabled to be stricter around ASCII-compatible but unlikely to be a string.
884 """
885 if isinstance(fp_or_path_or_payload, (str, PathLike)):
886 guesses = from_path(
887 fp_or_path_or_payload,
888 steps=steps,
889 chunk_size=chunk_size,
890 threshold=threshold,
891 cp_isolation=cp_isolation,
892 cp_exclusion=cp_exclusion,
893 preemptive_behaviour=preemptive_behaviour,
894 explain=explain,
895 language_threshold=language_threshold,
896 enable_fallback=enable_fallback,
897 )
898 elif isinstance(
899 fp_or_path_or_payload,
900 (
901 bytes,
902 bytearray,
903 ),
904 ):
905 guesses = from_bytes(
906 fp_or_path_or_payload,
907 steps=steps,
908 chunk_size=chunk_size,
909 threshold=threshold,
910 cp_isolation=cp_isolation,
911 cp_exclusion=cp_exclusion,
912 preemptive_behaviour=preemptive_behaviour,
913 explain=explain,
914 language_threshold=language_threshold,
915 enable_fallback=enable_fallback,
916 )
917 else:
918 guesses = from_fp(
919 fp_or_path_or_payload,
920 steps=steps,
921 chunk_size=chunk_size,
922 threshold=threshold,
923 cp_isolation=cp_isolation,
924 cp_exclusion=cp_exclusion,
925 preemptive_behaviour=preemptive_behaviour,
926 explain=explain,
927 language_threshold=language_threshold,
928 enable_fallback=enable_fallback,
929 )
931 return not guesses