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 mean_mess_ratio: float = sum(md_ratios) / len(md_ratios) if md_ratios else 0.0
472 # We might want to check the sequence again with the whole content,
473 # but only if initial MD tests passed.
474 if (
475 not lazy_str_hard_failure
476 and is_too_large_sequence
477 and not is_multi_byte_decoder
478 and mean_mess_ratio < threshold
479 and early_stop_count < max_chunk_gave_up
480 ):
481 try:
482 sequences[int(50e3) :].decode(encoding_iana, errors="strict")
483 except UnicodeDecodeError as e:
484 logger.log(
485 TRACE,
486 "LazyStr Loading: After final lookup, code page %s does not fit given bytes sequence at ALL. %s",
487 encoding_iana,
488 str(e),
489 )
490 tested_but_hard_failure.append(encoding_iana)
491 continue
493 if mean_mess_ratio >= threshold or early_stop_count >= max_chunk_gave_up:
494 tested_but_soft_failure.append(encoding_iana)
495 if encoding_iana in IANA_SUPPORTED_SIMILAR:
496 soft_failure_skip.update(IANA_SUPPORTED_SIMILAR[encoding_iana])
497 logger.log(
498 TRACE,
499 "%s was excluded because of initial chaos probing. Gave up %i time(s). "
500 "Computed mean chaos is %f %%.",
501 encoding_iana,
502 early_stop_count,
503 round(mean_mess_ratio * 100, ndigits=3),
504 )
505 # Preparing those fallbacks in case we got nothing.
506 if (
507 enable_fallback
508 and encoding_iana
509 in ["ascii", "utf_8", specified_encoding, "utf_16", "utf_32"]
510 and not lazy_str_hard_failure
511 ):
512 # Always fully decode payload before.
513 # We've missed a UnicodeDecodeError proof
514 # while issuing release 3.4.8
515 # see https://github.com/jawah/charset_normalizer/issues/771
516 if decoded_payload is None:
517 try:
518 decoded_payload = str(
519 (
520 sequences
521 if not strip_sig_or_bom
522 else sequences[len(sig_payload) :]
523 ),
524 encoding=encoding_iana,
525 )
526 except (UnicodeDecodeError, LookupError):
527 logger.log(
528 TRACE,
529 "%s does not decode the whole payload: fallback entry withheld.",
530 encoding_iana,
531 )
532 continue
533 if is_too_large_sequence:
534 # Don't retain huge payload in RAM.
535 decoded_payload = None
537 fallback_entry = CharsetMatch(
538 sequences,
539 encoding_iana,
540 threshold,
541 bom_or_sig_available,
542 [],
543 decoded_payload,
544 preemptive_declaration=specified_encoding,
545 )
546 if encoding_iana == specified_encoding:
547 fallback_specified = fallback_entry
548 elif encoding_iana == "ascii":
549 fallback_ascii = fallback_entry
550 else:
551 fallback_u8 = fallback_entry
552 continue
554 if deferred_decoding:
555 # The candidate passed chaos probing: perform the whole payload
556 # decode (validation + payload reuse) that was deferred earlier.
557 try:
558 decoded_payload = str(
559 (
560 sequences
561 if not strip_sig_or_bom
562 else sequences[len(sig_payload) :]
563 ),
564 encoding=encoding_iana,
565 )
566 except (UnicodeDecodeError, LookupError) as e:
567 logger.log(
568 TRACE,
569 "Code page %s does not fit given bytes sequence at ALL. %s",
570 encoding_iana,
571 str(e),
572 )
573 tested_but_hard_failure.append(encoding_iana)
574 continue
576 logger.log(
577 TRACE,
578 "%s passed initial chaos probing. Mean measured chaos is %f %%",
579 encoding_iana,
580 round(mean_mess_ratio * 100, ndigits=3),
581 )
583 if not is_multi_byte_decoder:
584 target_languages: list[str] = encoding_languages(encoding_iana)
585 else:
586 target_languages = mb_encoding_languages(encoding_iana)
588 if target_languages:
589 logger.log(
590 TRACE,
591 "%s should target any language(s) of %s",
592 encoding_iana,
593 target_languages,
594 )
596 cd_ratios = []
598 # Run coherence detection on all chunks. We previously tried limiting to
599 # 1-2 chunks for post-definitive encodings to save time, but this caused
600 # coverage regressions by producing unrepresentative coherence scores.
601 # The SB cap and language-family skip optimizations provide sufficient
602 # speedup without sacrificing coherence accuracy.
603 if encoding_iana != "ascii":
604 # We shall skip the CD when its about ASCII
605 # Most of the time its not relevant to run "language-detection" on it.
606 lg_inclusion: str | None = (
607 ",".join(target_languages) if target_languages else None
608 )
610 for chunk in md_chunks:
611 chunk_languages = cached_coherence_ratio(
612 chunk,
613 language_threshold,
614 lg_inclusion,
615 )
617 cd_ratios.append(chunk_languages)
619 cd_ratios_merged = merge_coherence_ratios(cd_ratios)
621 if cd_ratios_merged:
622 logger.log(
623 TRACE,
624 "We detected language %s using %s",
625 cd_ratios_merged,
626 encoding_iana,
627 )
629 current_match = CharsetMatch(
630 sequences,
631 encoding_iana,
632 mean_mess_ratio,
633 bom_or_sig_available,
634 cd_ratios_merged,
635 (
636 decoded_payload
637 if (
638 not is_too_large_sequence
639 or encoding_iana in [specified_encoding, "ascii", "utf_8"]
640 )
641 else None
642 ),
643 preemptive_declaration=specified_encoding,
644 )
646 results.append(current_match)
648 # Count post-definitive same-family SB successes for the early termination cap.
649 # Only count low-mess encodings (< 2%) toward the cap. High-mess encodings are
650 # marginal results that shouldn't prevent better-quality candidates from being
651 # tested. For example, iso8859_4 (mess=0%) should not be skipped just because
652 # 7 high-mess Latin encodings (cp1252 at 8%, etc.) were tried first.
653 if (
654 definitive_match_found
655 and not is_multi_byte_decoder
656 and mean_mess_ratio < 0.02
657 ):
658 post_definitive_sb_success_count += 1
660 if (
661 encoding_iana in [specified_encoding, "ascii", "utf_8"]
662 and mean_mess_ratio < 0.1
663 ):
664 # If md says nothing to worry about, then... stop immediately!
665 if mean_mess_ratio == 0.0:
666 logger.debug(
667 "Encoding detection: %s is most likely the one.",
668 current_match.encoding,
669 )
670 if explain: # Defensive: ensure exit path clean handler
671 logger.removeHandler(explain_handler)
672 logger.setLevel(previous_logger_level)
673 return CharsetMatches([current_match])
675 early_stop_results.append(current_match)
677 if (
678 len(early_stop_results)
679 and (specified_encoding is None or specified_encoding in tested)
680 and "ascii" in tested
681 and "utf_8" in tested
682 ):
683 probable_result = early_stop_results.best()
684 assert probable_result is not None
685 logger.debug(
686 "Encoding detection: %s is most likely the one.",
687 probable_result.encoding,
688 )
689 if explain: # Defensive: ensure exit path clean handler
690 logger.removeHandler(explain_handler)
691 logger.setLevel(previous_logger_level)
693 return CharsetMatches([probable_result])
695 # Once we find a result with good coherence (>= 0.5) after testing the
696 # prioritized encodings (ascii, utf_8), activate "definitive mode": skip
697 # encodings that target completely different language families. This avoids
698 # running expensive mess_ratio + coherence_ratio on clearly unrelated
699 # candidates (e.g., Cyrillic encodings when the match is Latin-based).
700 # We require coherence >= 0.5 to avoid false positives (e.g., cp1251 decoding
701 # Hebrew text with 0.0 chaos but wrong language detection at coherence 0.33).
702 if not definitive_match_found and not is_multi_byte_decoder:
703 best_coherence = (
704 max((v for _, v in cd_ratios_merged), default=0.0)
705 if cd_ratios_merged
706 else 0.0
707 )
708 if best_coherence >= 0.5 and "ascii" in tested and "utf_8" in tested:
709 definitive_match_found = True
710 definitive_target_languages.update(target_languages)
711 logger.log(
712 TRACE,
713 "Definitive match found: %s (chaos=%.3f, coherence=%.2f). Encodings targeting different language families will be skipped.",
714 encoding_iana,
715 mean_mess_ratio,
716 best_coherence,
717 )
719 # When a non-UTF multibyte encoding passes chaos probing with significant
720 # multibyte content (decoded < 98% of raw), activate mb_definitive_match.
721 # This skips all remaining single-byte encodings which would either soft-fail
722 # (running expensive mess_ratio for nothing) or produce inferior results.
723 if (
724 not mb_definitive_match_found
725 and is_multi_byte_decoder
726 and multi_byte_bonus
727 and decoded_payload is not None
728 and len(decoded_payload) < length * 0.98
729 and encoding_iana
730 not in {
731 "utf_8",
732 "utf_8_sig",
733 "utf_16",
734 "utf_16_be",
735 "utf_16_le",
736 "utf_32",
737 "utf_32_be",
738 "utf_32_le",
739 "utf_7",
740 }
741 and "ascii" in tested
742 and "utf_8" in tested
743 ):
744 mb_definitive_match_found = True
745 logger.log(
746 TRACE,
747 "Multi-byte definitive match: %s (chaos=%.3f, decoded=%d/%d=%.1f%%). Single-byte encodings will be skipped.",
748 encoding_iana,
749 mean_mess_ratio,
750 len(decoded_payload),
751 length,
752 len(decoded_payload) / length * 100,
753 )
755 if encoding_iana == sig_encoding:
756 logger.debug(
757 "Encoding detection: %s is most likely the one as we detected a BOM or SIG within "
758 "the beginning of the sequence.",
759 encoding_iana,
760 )
761 if explain: # Defensive: ensure exit path clean handler
762 logger.removeHandler(explain_handler)
763 logger.setLevel(previous_logger_level)
764 return CharsetMatches([results[encoding_iana]])
766 if len(results) == 0:
767 if fallback_u8 or fallback_ascii or fallback_specified:
768 logger.log(
769 TRACE,
770 "Nothing got out of the detection process. Using ASCII/UTF-8/Specified fallback.",
771 )
773 if fallback_specified:
774 logger.debug(
775 "Encoding detection: %s will be used as a fallback match",
776 fallback_specified.encoding,
777 )
778 results.append(fallback_specified)
779 elif (
780 (fallback_u8 and fallback_ascii is None)
781 or (
782 fallback_u8
783 and fallback_ascii
784 and fallback_u8.fingerprint != fallback_ascii.fingerprint
785 )
786 or (fallback_u8 is not None)
787 ):
788 logger.debug("Encoding detection: utf_8 will be used as a fallback match")
789 results.append(fallback_u8)
790 elif fallback_ascii:
791 logger.debug("Encoding detection: ascii will be used as a fallback match")
792 results.append(fallback_ascii)
794 if results:
795 logger.debug(
796 "Encoding detection: Found %s as plausible (best-candidate) for content. With %i alternatives.",
797 results.best().encoding, # type: ignore
798 len(results) - 1,
799 )
800 else:
801 logger.debug("Encoding detection: Unable to determine any suitable charset.")
803 if explain:
804 logger.removeHandler(explain_handler)
805 logger.setLevel(previous_logger_level)
807 return results
810def from_fp(
811 fp: BinaryIO,
812 steps: int = 5,
813 chunk_size: int = 512,
814 threshold: float = 0.20,
815 cp_isolation: list[str] | None = None,
816 cp_exclusion: list[str] | None = None,
817 preemptive_behaviour: bool = True,
818 explain: bool = False,
819 language_threshold: float = 0.1,
820 enable_fallback: bool = True,
821) -> CharsetMatches:
822 """
823 Same thing than the function from_bytes but using a file pointer that is already ready.
824 Will not close the file pointer.
825 """
826 return from_bytes(
827 fp.read(),
828 steps,
829 chunk_size,
830 threshold,
831 cp_isolation,
832 cp_exclusion,
833 preemptive_behaviour,
834 explain,
835 language_threshold,
836 enable_fallback,
837 )
840def from_path(
841 path: str | bytes | PathLike, # type: ignore[type-arg]
842 steps: int = 5,
843 chunk_size: int = 512,
844 threshold: float = 0.20,
845 cp_isolation: list[str] | None = None,
846 cp_exclusion: list[str] | None = None,
847 preemptive_behaviour: bool = True,
848 explain: bool = False,
849 language_threshold: float = 0.1,
850 enable_fallback: bool = True,
851) -> CharsetMatches:
852 """
853 Same thing than the function from_bytes but with one extra step. Opening and reading given file path in binary mode.
854 Can raise IOError.
855 """
856 with open(path, "rb") as fp:
857 return from_fp(
858 fp,
859 steps,
860 chunk_size,
861 threshold,
862 cp_isolation,
863 cp_exclusion,
864 preemptive_behaviour,
865 explain,
866 language_threshold,
867 enable_fallback,
868 )
871def is_binary(
872 fp_or_path_or_payload: PathLike | str | BinaryIO | bytes, # type: ignore[type-arg]
873 steps: int = 5,
874 chunk_size: int = 512,
875 threshold: float = 0.20,
876 cp_isolation: list[str] | None = None,
877 cp_exclusion: list[str] | None = None,
878 preemptive_behaviour: bool = True,
879 explain: bool = False,
880 language_threshold: float = 0.1,
881 enable_fallback: bool = False,
882) -> bool:
883 """
884 Detect if the given input (file, bytes, or path) points to a binary file. aka. not a string.
885 Based on the same main heuristic algorithms and default kwargs at the sole exception that fallbacks match
886 are disabled to be stricter around ASCII-compatible but unlikely to be a string.
887 """
888 if isinstance(fp_or_path_or_payload, (str, PathLike)):
889 guesses = from_path(
890 fp_or_path_or_payload,
891 steps=steps,
892 chunk_size=chunk_size,
893 threshold=threshold,
894 cp_isolation=cp_isolation,
895 cp_exclusion=cp_exclusion,
896 preemptive_behaviour=preemptive_behaviour,
897 explain=explain,
898 language_threshold=language_threshold,
899 enable_fallback=enable_fallback,
900 )
901 elif isinstance(
902 fp_or_path_or_payload,
903 (
904 bytes,
905 bytearray,
906 ),
907 ):
908 guesses = from_bytes(
909 fp_or_path_or_payload,
910 steps=steps,
911 chunk_size=chunk_size,
912 threshold=threshold,
913 cp_isolation=cp_isolation,
914 cp_exclusion=cp_exclusion,
915 preemptive_behaviour=preemptive_behaviour,
916 explain=explain,
917 language_threshold=language_threshold,
918 enable_fallback=enable_fallback,
919 )
920 else:
921 guesses = from_fp(
922 fp_or_path_or_payload,
923 steps=steps,
924 chunk_size=chunk_size,
925 threshold=threshold,
926 cp_isolation=cp_isolation,
927 cp_exclusion=cp_exclusion,
928 preemptive_behaviour=preemptive_behaviour,
929 explain=explain,
930 language_threshold=language_threshold,
931 enable_fallback=enable_fallback,
932 )
934 return not guesses