Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/charset_normalizer/api.py: 6%

1from __future__ import annotations 

2 

3import logging 

4from functools import lru_cache 

5from os import PathLike 

6from typing import BinaryIO 

7 

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) 

31 

32logger = logging.getLogger("charset_normalizer") 

33explain_handler = logging.StreamHandler() 

34explain_handler.setFormatter( 

35 logging.Formatter("%(asctime)s | %(levelname)s | %(message)s") 

36) 

37 

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) 

48 

49 

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. 

67 

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. 

70 

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. 

73 

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

79 

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 ) 

86 

87 if explain: 

88 previous_logger_level: int = logger.level 

89 logger.addHandler(explain_handler) 

90 logger.setLevel(TRACE) 

91 

92 length: int = len(sequences) 

93 

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, [], "")]) 

100 

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 = [] 

111 

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 = [] 

122 

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 

133 

134 if steps > 1 and length / steps < chunk_size: 

135 chunk_size = int(length / steps) 

136 

137 is_too_small_sequence: bool = len(sequences) < TOO_SMALL_SEQUENCE 

138 is_too_large_sequence: bool = len(sequences) >= TOO_BIG_SEQUENCE 

139 

140 if is_too_small_sequence: 

141 logger.log( 

142 TRACE, 

143 "Trying to detect encoding from a tiny portion of ({}) byte(s).".format( 

144 length 

145 ), 

146 ) 

147 elif is_too_large_sequence: 

148 logger.log( 

149 TRACE, 

150 "Using lazy str decoding because the payload is quite large, ({}) byte(s).".format( 

151 length 

152 ), 

153 ) 

154 

155 prioritized_encodings: list[str] = [] 

156 

157 specified_encoding: str | None = ( 

158 any_specified_encoding(sequences) if preemptive_behaviour else None 

159 ) 

160 

161 if specified_encoding is not None: 

162 prioritized_encodings.append(specified_encoding) 

163 logger.log( 

164 TRACE, 

165 "Detected declarative mark in sequence. Priority +1 given for %s.", 

166 specified_encoding, 

167 ) 

168 

169 tested: set[str] = set() 

170 tested_but_hard_failure: list[str] = [] 

171 tested_but_soft_failure: list[str] = [] 

172 soft_failure_skip: set[str] = set() 

173 success_fast_tracked: set[str] = set() 

174 

175 # Cache for decoded payload deduplication: hash(decoded_payload) -> (mean_mess_ratio, cd_ratios_merged, passed) 

176 # When multiple encodings decode to the exact same string, we can skip the expensive 

177 # mess_ratio and coherence_ratio analysis and reuse the results from the first encoding. 

178 payload_result_cache: dict[int, tuple[float, list[tuple[str, float]], bool]] = {} 

179 

180 # Avoid unoptimized RSS usage. 

181 # this cache is mostly interesting for 

182 # local usage. Garbage collected at the 

183 # end. Like it should. 

184 cached_mess_ratio = lru_cache(maxsize=None)(mess_ratio) 

185 cached_coherence_ratio = lru_cache(maxsize=None)(coherence_ratio) 

186 

187 # When a definitive result (chaos=0.0 and good coherence) is found after testing 

188 # the prioritized encodings (ascii, utf_8), we can significantly reduce the remaining 

189 # work. Encodings that target completely different language families (e.g., Cyrillic 

190 # when the definitive match is Latin) are skipped entirely. 

191 # Additionally, for same-family encodings that pass chaos probing, we reuse the 

192 # definitive match's coherence ratios instead of recomputing them — a major savings 

193 # since coherence_ratio accounts for ~30% of total time on slow Latin files. 

194 definitive_match_found: bool = False 

195 definitive_target_languages: set[str] = set() 

196 # After the definitive match fires, we cap the number of additional same-family 

197 # single-byte encodings that pass chaos probing. Once we've accumulated enough 

198 # good candidates (N), further same-family SB encodings are unlikely to produce 

199 # a better best() result and just waste mess_ratio + coherence_ratio time. 

200 # The first encoding to trigger the definitive match is NOT counted (it's already in). 

201 post_definitive_sb_success_count: int = 0 

202 POST_DEFINITIVE_SB_CAP: int = 7 

203 

204 # When a non-UTF multibyte encoding passes chaos probing with significant multibyte 

205 # content (decoded length < 98% of raw length), skip all remaining single-byte encodings. 

206 # Rationale: multi-byte decoders (CJK) have strict byte-sequence validation — if they 

207 # decode without error AND pass chaos probing with substantial multibyte content, the 

208 # data is genuinely multibyte encoded. Single-byte encodings will always decode (every 

209 # byte maps to something) but waste time on mess_ratio before failing. 

210 # The 98% threshold prevents false triggers on files that happen to have a few valid 

211 # multibyte pairs (e.g., cp424/_ude_1.txt where big5 decodes with 99% ratio). 

212 mb_definitive_match_found: bool = False 

213 

214 fallback_ascii: CharsetMatch | None = None 

215 fallback_u8: CharsetMatch | None = None 

216 fallback_specified: CharsetMatch | None = None 

217 

218 results: CharsetMatches = CharsetMatches() 

219 

220 early_stop_results: CharsetMatches = CharsetMatches() 

221 

222 sig_encoding, sig_payload = identify_sig_or_bom(sequences) 

223 

224 if sig_encoding is not None: 

225 prioritized_encodings.append(sig_encoding) 

226 logger.log( 

227 TRACE, 

228 "Detected a SIG or BOM mark on first %i byte(s). Priority +1 given for %s.", 

229 len(sig_payload), 

230 sig_encoding, 

231 ) 

232 

233 prioritized_encodings.append("ascii") 

234 

235 if "utf_8" not in prioritized_encodings: 

236 prioritized_encodings.append("utf_8") 

237 

238 for encoding_iana in prioritized_encodings + IANA_SUPPORTED_MB_FIRST: 

239 if cp_isolation and encoding_iana not in cp_isolation: 

240 continue 

241 

242 if cp_exclusion and encoding_iana in cp_exclusion: 

243 continue 

244 

245 if encoding_iana in tested: 

246 continue 

247 

248 tested.add(encoding_iana) 

249 

250 decoded_payload: str | None = None 

251 bom_or_sig_available: bool = sig_encoding == encoding_iana 

252 strip_sig_or_bom: bool = bom_or_sig_available and should_strip_sig_or_bom( 

253 encoding_iana 

254 ) 

255 

256 if encoding_iana in {"utf_16", "utf_32"} and not bom_or_sig_available: 

257 logger.log( 

258 TRACE, 

259 "Encoding %s won't be tested as-is because it require a BOM. Will try some sub-encoder LE/BE.", 

260 encoding_iana, 

261 ) 

262 continue 

263 if encoding_iana in {"utf_7"} and not bom_or_sig_available: 

264 logger.log( 

265 TRACE, 

266 "Encoding %s won't be tested as-is because detection is unreliable without BOM/SIG.", 

267 encoding_iana, 

268 ) 

269 continue 

270 

271 # Skip encodings similar to ones that already soft-failed (high mess ratio). 

272 # Checked BEFORE the expensive decode attempt. 

273 if encoding_iana in soft_failure_skip: 

274 logger.log( 

275 TRACE, 

276 "%s is deemed too similar to a code page that was already considered unsuited. Continuing!", 

277 encoding_iana, 

278 ) 

279 continue 

280 

281 # Skip encodings that were already fast-tracked from a similar successful encoding. 

282 if encoding_iana in success_fast_tracked: 

283 logger.log( 

284 TRACE, 

285 "Skipping %s: already fast-tracked from a similar successful encoding.", 

286 encoding_iana, 

287 ) 

288 continue 

289 

290 try: 

291 is_multi_byte_decoder: bool = is_multi_byte_encoding(encoding_iana) 

292 except (ModuleNotFoundError, ImportError): # Defensive: 

293 logger.log( 

294 TRACE, 

295 "Encoding %s does not provide an IncrementalDecoder", 

296 encoding_iana, 

297 ) 

298 continue 

299 

300 # When we've already found a definitive match (chaos=0.0 with good coherence) 

301 # after testing the prioritized encodings, skip encodings that target 

302 # completely different language families. This avoids running expensive 

303 # mess_ratio + coherence_ratio on clearly unrelated candidates (e.g., Cyrillic 

304 # when the definitive match is Latin-based). 

305 if definitive_match_found: 

306 if not is_multi_byte_decoder: 

307 enc_languages = set(encoding_languages(encoding_iana)) 

308 else: 

309 enc_languages = set(mb_encoding_languages(encoding_iana)) 

310 if not enc_languages.intersection(definitive_target_languages): 

311 logger.log( 

312 TRACE, 

313 "Skipping %s: definitive match already found, this encoding targets different languages (%s vs %s).", 

314 encoding_iana, 

315 enc_languages, 

316 definitive_target_languages, 

317 ) 

318 continue 

319 

320 # After the definitive match, cap the number of additional same-family 

321 # single-byte encodings that pass chaos probing. This avoids testing the 

322 # tail of rare, low-value same-family encodings (mac_iceland, cp860, etc.) 

323 # that almost never change best() but each cost ~1-2ms of mess_ratio + coherence. 

324 if ( 

325 definitive_match_found 

326 and not is_multi_byte_decoder 

327 and post_definitive_sb_success_count >= POST_DEFINITIVE_SB_CAP 

328 ): 

329 logger.log( 

330 TRACE, 

331 "Skipping %s: already accumulated %d same-family results after definitive match (cap=%d).", 

332 encoding_iana, 

333 post_definitive_sb_success_count, 

334 POST_DEFINITIVE_SB_CAP, 

335 ) 

336 continue 

337 

338 # When a multibyte encoding with significant multibyte content has already 

339 # passed chaos probing, skip all single-byte encodings. They will either fail 

340 # chaos probing (wasting mess_ratio time) or produce inferior results. 

341 if mb_definitive_match_found and not is_multi_byte_decoder: 

342 logger.log( 

343 TRACE, 

344 "Skipping single-byte %s: multi-byte definitive match already found.", 

345 encoding_iana, 

346 ) 

347 continue 

348 

349 # Single-byte candidates of regular size defer the expensive whole 

350 # payload decode until after chunk probing: single-byte codecs are 

351 # stateless (1 byte == 1 char) so decoding chunk slices is provably 

352 # identical to slicing the decoded payload, and candidates rejected 

353 # by chaos probing (the common case) never pay the full decode nor 

354 # the payload hash. 

355 deferred_decoding: bool = ( 

356 not is_multi_byte_decoder and not is_too_large_sequence 

357 ) 

358 

359 try: 

360 if is_too_large_sequence and not is_multi_byte_decoder: 

361 str( 

362 ( 

363 sequences[: int(50e4)] 

364 if not strip_sig_or_bom 

365 else sequences[len(sig_payload) : int(50e4)] 

366 ), 

367 encoding=encoding_iana, 

368 ) 

369 elif not deferred_decoding: 

370 # UTF-7 BOM is encoded in modified Base64 whose byte boundary 

371 # can overlap with the next character. Stripping raw SIG bytes 

372 # before decoding may leave stray bytes that decode as garbage. 

373 # Decode the full sequence and remove the leading BOM char instead. 

374 # see https://github.com/jawah/charset_normalizer/issues/718 

375 # and https://github.com/jawah/charset_normalizer/issues/716 

376 if encoding_iana == "utf_7" and bom_or_sig_available: 

377 decoded_payload = str( 

378 sequences, 

379 encoding=encoding_iana, 

380 ) 

381 if decoded_payload and decoded_payload[0] == "\ufeff": 

382 decoded_payload = decoded_payload[1:] 

383 else: 

384 decoded_payload = str( 

385 ( 

386 sequences 

387 if not strip_sig_or_bom 

388 else sequences[len(sig_payload) :] 

389 ), 

390 encoding=encoding_iana, 

391 ) 

392 except (UnicodeDecodeError, LookupError) as e: 

393 if not isinstance(e, LookupError): 

394 logger.log( 

395 TRACE, 

396 "Code page %s does not fit given bytes sequence at ALL. %s", 

397 encoding_iana, 

398 str(e), 

399 ) 

400 tested_but_hard_failure.append(encoding_iana) 

401 continue 

402 

403 r_ = range( 

404 0 if not bom_or_sig_available else len(sig_payload), 

405 length, 

406 int(length / steps), 

407 ) 

408 

409 multi_byte_bonus: bool = ( 

410 is_multi_byte_decoder 

411 and decoded_payload is not None 

412 and len(decoded_payload) < length 

413 ) 

414 

415 if multi_byte_bonus: 

416 logger.log( 

417 TRACE, 

418 "Code page %s is a multi byte encoding table and it appear that at least one character " 

419 "was encoded using n-bytes.", 

420 encoding_iana, 

421 ) 

422 

423 max_chunk_gave_up: int = int(len(r_) / 4) 

424 

425 max_chunk_gave_up = max(max_chunk_gave_up, 2) 

426 early_stop_count: int = 0 

427 lazy_str_hard_failure = False 

428 

429 md_chunks: list[str] = [] 

430 md_ratios = [] 

431 

432 try: 

433 for chunk in cut_sequence_chunks( 

434 sequences, 

435 encoding_iana, 

436 r_, 

437 chunk_size, 

438 bom_or_sig_available, 

439 strip_sig_or_bom, 

440 sig_payload, 

441 is_multi_byte_decoder, 

442 decoded_payload, 

443 deferred_decoding, 

444 ): 

445 md_chunks.append(chunk) 

446 

447 md_ratios.append( 

448 cached_mess_ratio( 

449 chunk, 

450 threshold, 

451 explain and 1 <= len(cp_isolation) <= 2, 

452 ) 

453 ) 

454 

455 if md_ratios[-1] >= threshold: 

456 early_stop_count += 1 

457 

458 if (early_stop_count >= max_chunk_gave_up) or ( 

459 bom_or_sig_available and not strip_sig_or_bom 

460 ): 

461 break 

462 except ( 

463 UnicodeDecodeError, 

464 LookupError, 

465 ) as e: # Lazy str loading may have missed something there 

466 if deferred_decoding: 

467 # Deferred single-byte validation failed on a chunk (or the 

468 # codec is unavailable on this interpreter build): identical 

469 # outcome and bookkeeping to the eager full-decode failure. 

470 logger.log( 

471 TRACE, 

472 "Code page %s does not fit given bytes sequence at ALL. %s", 

473 encoding_iana, 

474 str(e), 

475 ) 

476 tested_but_hard_failure.append(encoding_iana) 

477 continue 

478 logger.log( 

479 TRACE, 

480 "LazyStr Loading: After MD chunk decode, code page %s does not fit given bytes sequence at ALL. %s", 

481 encoding_iana, 

482 str(e), 

483 ) 

484 early_stop_count = max_chunk_gave_up 

485 lazy_str_hard_failure = True 

486 

487 # We might want to check the sequence again with the whole content 

488 # Only if initial MD tests passes 

489 if ( 

490 not lazy_str_hard_failure 

491 and is_too_large_sequence 

492 and not is_multi_byte_decoder 

493 ): 

494 try: 

495 sequences[int(50e3) :].decode(encoding_iana, errors="strict") 

496 except UnicodeDecodeError as e: 

497 logger.log( 

498 TRACE, 

499 "LazyStr Loading: After final lookup, code page %s does not fit given bytes sequence at ALL. %s", 

500 encoding_iana, 

501 str(e), 

502 ) 

503 tested_but_hard_failure.append(encoding_iana) 

504 continue 

505 

506 mean_mess_ratio: float = sum(md_ratios) / len(md_ratios) if md_ratios else 0.0 

507 if mean_mess_ratio >= threshold or early_stop_count >= max_chunk_gave_up: 

508 tested_but_soft_failure.append(encoding_iana) 

509 if encoding_iana in IANA_SUPPORTED_SIMILAR: 

510 soft_failure_skip.update(IANA_SUPPORTED_SIMILAR[encoding_iana]) 

511 # Cache this soft-failure so identical decoding from other encodings 

512 # can be skipped immediately. 

513 if decoded_payload is not None and not is_multi_byte_decoder: 

514 payload_result_cache.setdefault( 

515 hash(decoded_payload), (mean_mess_ratio, [], False) 

516 ) 

517 logger.log( 

518 TRACE, 

519 "%s was excluded because of initial chaos probing. Gave up %i time(s). " 

520 "Computed mean chaos is %f %%.", 

521 encoding_iana, 

522 early_stop_count, 

523 round(mean_mess_ratio * 100, ndigits=3), 

524 ) 

525 # Preparing those fallbacks in case we got nothing. 

526 if ( 

527 enable_fallback 

528 and encoding_iana 

529 in ["ascii", "utf_8", specified_encoding, "utf_16", "utf_32"] 

530 and not lazy_str_hard_failure 

531 ): 

532 # Always fully decode payload before. 

533 # We've missed a UnicodeDecodeError proof 

534 # while issuing release 3.4.8 

535 # see https://github.com/jawah/charset_normalizer/issues/771 

536 if decoded_payload is None: 

537 try: 

538 decoded_payload = str( 

539 ( 

540 sequences 

541 if not strip_sig_or_bom 

542 else sequences[len(sig_payload) :] 

543 ), 

544 encoding=encoding_iana, 

545 ) 

546 except (UnicodeDecodeError, LookupError): 

547 logger.log( 

548 TRACE, 

549 "%s does not decode the whole payload: fallback entry withheld.", 

550 encoding_iana, 

551 ) 

552 continue 

553 if is_too_large_sequence: 

554 # Don't retain huge payload in RAM. 

555 decoded_payload = None 

556 

557 fallback_entry = CharsetMatch( 

558 sequences, 

559 encoding_iana, 

560 threshold, 

561 bom_or_sig_available, 

562 [], 

563 decoded_payload, 

564 preemptive_declaration=specified_encoding, 

565 ) 

566 if encoding_iana == specified_encoding: 

567 fallback_specified = fallback_entry 

568 elif encoding_iana == "ascii": 

569 fallback_ascii = fallback_entry 

570 else: 

571 fallback_u8 = fallback_entry 

572 continue 

573 

574 if deferred_decoding: 

575 # The candidate passed chaos probing: perform the whole payload 

576 # decode (validation + payload reuse) that was deferred earlier. 

577 try: 

578 decoded_payload = str( 

579 ( 

580 sequences 

581 if not strip_sig_or_bom 

582 else sequences[len(sig_payload) :] 

583 ), 

584 encoding=encoding_iana, 

585 ) 

586 except (UnicodeDecodeError, LookupError) as e: 

587 logger.log( 

588 TRACE, 

589 "Code page %s does not fit given bytes sequence at ALL. %s", 

590 encoding_iana, 

591 str(e), 

592 ) 

593 tested_but_hard_failure.append(encoding_iana) 

594 continue 

595 

596 # Payload-hash deduplication: if another encoding already decoded to the 

597 # exact same string, reuse its mess_ratio and coherence results entirely. 

598 # This is strictly more general than the old IANA_SUPPORTED_SIMILAR approach 

599 # because it catches ALL identical decoding, not just pre-mapped ones. 

600 if decoded_payload is not None and not is_multi_byte_decoder: 

601 payload_hash: int = hash(decoded_payload) 

602 cached = payload_result_cache.get(payload_hash) 

603 if cached is not None: 

604 cached_mess, cached_cd, cached_passed = cached 

605 if cached_passed: 

606 # The previous encoding with identical output passed chaos probing. 

607 fast_match = CharsetMatch( 

608 sequences, 

609 encoding_iana, 

610 cached_mess, 

611 bom_or_sig_available, 

612 cached_cd, 

613 ( 

614 decoded_payload 

615 if ( 

616 not is_too_large_sequence 

617 or encoding_iana 

618 in [specified_encoding, "ascii", "utf_8"] 

619 ) 

620 else None 

621 ), 

622 preemptive_declaration=specified_encoding, 

623 ) 

624 results.append(fast_match) 

625 success_fast_tracked.add(encoding_iana) 

626 logger.log( 

627 TRACE, 

628 "%s fast-tracked (identical decoded payload to a prior encoding, chaos=%f %%).", 

629 encoding_iana, 

630 round(cached_mess * 100, ndigits=3), 

631 ) 

632 

633 if ( 

634 encoding_iana in [specified_encoding, "ascii", "utf_8"] 

635 and cached_mess < 0.1 

636 ): 

637 if cached_mess == 0.0: 

638 logger.debug( 

639 "Encoding detection: %s is most likely the one.", 

640 fast_match.encoding, 

641 ) 

642 if explain: 

643 logger.removeHandler(explain_handler) 

644 logger.setLevel(previous_logger_level) 

645 return CharsetMatches([fast_match]) 

646 early_stop_results.append(fast_match) 

647 

648 if ( 

649 len(early_stop_results) 

650 and (specified_encoding is None or specified_encoding in tested) 

651 and "ascii" in tested 

652 and "utf_8" in tested 

653 ): 

654 probable_result: CharsetMatch = early_stop_results.best() # type: ignore[assignment] 

655 logger.debug( 

656 "Encoding detection: %s is most likely the one.", 

657 probable_result.encoding, 

658 ) 

659 if explain: 

660 logger.removeHandler(explain_handler) 

661 logger.setLevel(previous_logger_level) 

662 return CharsetMatches([probable_result]) 

663 

664 continue 

665 else: 

666 # The previous encoding with identical output failed chaos 

667 # probing. Unreachable when the current candidate passed 

668 # probing on the identical payload (deterministic ratios), 

669 # kept for structural parity with the historic flow. 

670 tested_but_soft_failure.append(encoding_iana) 

671 logger.log( 

672 TRACE, 

673 "%s fast-skipped (identical decoded payload to a prior encoding that failed chaos probing).", 

674 encoding_iana, 

675 ) 

676 # Prepare fallbacks for special encodings even when skipped. 

677 if enable_fallback and encoding_iana in [ 

678 "ascii", 

679 "utf_8", 

680 specified_encoding, 

681 "utf_16", 

682 "utf_32", 

683 ]: 

684 fallback_entry = CharsetMatch( 

685 sequences, 

686 encoding_iana, 

687 threshold, 

688 bom_or_sig_available, 

689 [], 

690 decoded_payload, 

691 preemptive_declaration=specified_encoding, 

692 ) 

693 if encoding_iana == specified_encoding: 

694 fallback_specified = fallback_entry 

695 elif encoding_iana == "ascii": 

696 fallback_ascii = fallback_entry 

697 else: 

698 fallback_u8 = fallback_entry 

699 continue 

700 

701 logger.log( 

702 TRACE, 

703 "%s passed initial chaos probing. Mean measured chaos is %f %%", 

704 encoding_iana, 

705 round(mean_mess_ratio * 100, ndigits=3), 

706 ) 

707 

708 if not is_multi_byte_decoder: 

709 target_languages: list[str] = encoding_languages(encoding_iana) 

710 else: 

711 target_languages = mb_encoding_languages(encoding_iana) 

712 

713 if target_languages: 

714 logger.log( 

715 TRACE, 

716 "{} should target any language(s) of {}".format( 

717 encoding_iana, str(target_languages) 

718 ), 

719 ) 

720 

721 cd_ratios = [] 

722 

723 # Run coherence detection on all chunks. We previously tried limiting to 

724 # 1-2 chunks for post-definitive encodings to save time, but this caused 

725 # coverage regressions by producing unrepresentative coherence scores. 

726 # The SB cap and language-family skip optimizations provide sufficient 

727 # speedup without sacrificing coherence accuracy. 

728 if encoding_iana != "ascii": 

729 # We shall skip the CD when its about ASCII 

730 # Most of the time its not relevant to run "language-detection" on it. 

731 lg_inclusion: str | None = ( 

732 ",".join(target_languages) if target_languages else None 

733 ) 

734 

735 for chunk in md_chunks: 

736 chunk_languages = cached_coherence_ratio( 

737 chunk, 

738 language_threshold, 

739 lg_inclusion, 

740 ) 

741 

742 cd_ratios.append(chunk_languages) 

743 

744 cd_ratios_merged = merge_coherence_ratios(cd_ratios) 

745 

746 if cd_ratios_merged: 

747 logger.log( 

748 TRACE, 

749 "We detected language {} using {}".format( 

750 cd_ratios_merged, encoding_iana 

751 ), 

752 ) 

753 

754 current_match = CharsetMatch( 

755 sequences, 

756 encoding_iana, 

757 mean_mess_ratio, 

758 bom_or_sig_available, 

759 cd_ratios_merged, 

760 ( 

761 decoded_payload 

762 if ( 

763 not is_too_large_sequence 

764 or encoding_iana in [specified_encoding, "ascii", "utf_8"] 

765 ) 

766 else None 

767 ), 

768 preemptive_declaration=specified_encoding, 

769 ) 

770 

771 results.append(current_match) 

772 

773 # Cache the successful result for payload-hash deduplication. 

774 if decoded_payload is not None and not is_multi_byte_decoder: 

775 payload_result_cache.setdefault( 

776 hash(decoded_payload), 

777 (mean_mess_ratio, cd_ratios_merged, True), 

778 ) 

779 

780 # Count post-definitive same-family SB successes for the early termination cap. 

781 # Only count low-mess encodings (< 2%) toward the cap. High-mess encodings are 

782 # marginal results that shouldn't prevent better-quality candidates from being 

783 # tested. For example, iso8859_4 (mess=0%) should not be skipped just because 

784 # 7 high-mess Latin encodings (cp1252 at 8%, etc.) were tried first. 

785 if ( 

786 definitive_match_found 

787 and not is_multi_byte_decoder 

788 and mean_mess_ratio < 0.02 

789 ): 

790 post_definitive_sb_success_count += 1 

791 

792 if ( 

793 encoding_iana in [specified_encoding, "ascii", "utf_8"] 

794 and mean_mess_ratio < 0.1 

795 ): 

796 # If md says nothing to worry about, then... stop immediately! 

797 if mean_mess_ratio == 0.0: 

798 logger.debug( 

799 "Encoding detection: %s is most likely the one.", 

800 current_match.encoding, 

801 ) 

802 if explain: # Defensive: ensure exit path clean handler 

803 logger.removeHandler(explain_handler) 

804 logger.setLevel(previous_logger_level) 

805 return CharsetMatches([current_match]) 

806 

807 early_stop_results.append(current_match) 

808 

809 if ( 

810 len(early_stop_results) 

811 and (specified_encoding is None or specified_encoding in tested) 

812 and "ascii" in tested 

813 and "utf_8" in tested 

814 ): 

815 probable_result = early_stop_results.best() # type: ignore[assignment] 

816 logger.debug( 

817 "Encoding detection: %s is most likely the one.", 

818 probable_result.encoding, # type: ignore[union-attr] 

819 ) 

820 if explain: # Defensive: ensure exit path clean handler 

821 logger.removeHandler(explain_handler) 

822 logger.setLevel(previous_logger_level) 

823 

824 return CharsetMatches([probable_result]) 

825 

826 # Once we find a result with good coherence (>= 0.5) after testing the 

827 # prioritized encodings (ascii, utf_8), activate "definitive mode": skip 

828 # encodings that target completely different language families. This avoids 

829 # running expensive mess_ratio + coherence_ratio on clearly unrelated 

830 # candidates (e.g., Cyrillic encodings when the match is Latin-based). 

831 # We require coherence >= 0.5 to avoid false positives (e.g., cp1251 decoding 

832 # Hebrew text with 0.0 chaos but wrong language detection at coherence 0.33). 

833 if not definitive_match_found and not is_multi_byte_decoder: 

834 best_coherence = ( 

835 max((v for _, v in cd_ratios_merged), default=0.0) 

836 if cd_ratios_merged 

837 else 0.0 

838 ) 

839 if best_coherence >= 0.5 and "ascii" in tested and "utf_8" in tested: 

840 definitive_match_found = True 

841 definitive_target_languages.update(target_languages) 

842 logger.log( 

843 TRACE, 

844 "Definitive match found: %s (chaos=%.3f, coherence=%.2f). Encodings targeting different language families will be skipped.", 

845 encoding_iana, 

846 mean_mess_ratio, 

847 best_coherence, 

848 ) 

849 

850 # When a non-UTF multibyte encoding passes chaos probing with significant 

851 # multibyte content (decoded < 98% of raw), activate mb_definitive_match. 

852 # This skips all remaining single-byte encodings which would either soft-fail 

853 # (running expensive mess_ratio for nothing) or produce inferior results. 

854 if ( 

855 not mb_definitive_match_found 

856 and is_multi_byte_decoder 

857 and multi_byte_bonus 

858 and decoded_payload is not None 

859 and len(decoded_payload) < length * 0.98 

860 and encoding_iana 

861 not in { 

862 "utf_8", 

863 "utf_8_sig", 

864 "utf_16", 

865 "utf_16_be", 

866 "utf_16_le", 

867 "utf_32", 

868 "utf_32_be", 

869 "utf_32_le", 

870 "utf_7", 

871 } 

872 and "ascii" in tested 

873 and "utf_8" in tested 

874 ): 

875 mb_definitive_match_found = True 

876 logger.log( 

877 TRACE, 

878 "Multi-byte definitive match: %s (chaos=%.3f, decoded=%d/%d=%.1f%%). Single-byte encodings will be skipped.", 

879 encoding_iana, 

880 mean_mess_ratio, 

881 len(decoded_payload), 

882 length, 

883 len(decoded_payload) / length * 100, 

884 ) 

885 

886 if encoding_iana == sig_encoding: 

887 logger.debug( 

888 "Encoding detection: %s is most likely the one as we detected a BOM or SIG within " 

889 "the beginning of the sequence.", 

890 encoding_iana, 

891 ) 

892 if explain: # Defensive: ensure exit path clean handler 

893 logger.removeHandler(explain_handler) 

894 logger.setLevel(previous_logger_level) 

895 return CharsetMatches([results[encoding_iana]]) 

896 

897 if len(results) == 0: 

898 if fallback_u8 or fallback_ascii or fallback_specified: 

899 logger.log( 

900 TRACE, 

901 "Nothing got out of the detection process. Using ASCII/UTF-8/Specified fallback.", 

902 ) 

903 

904 if fallback_specified: 

905 logger.debug( 

906 "Encoding detection: %s will be used as a fallback match", 

907 fallback_specified.encoding, 

908 ) 

909 results.append(fallback_specified) 

910 elif ( 

911 (fallback_u8 and fallback_ascii is None) 

912 or ( 

913 fallback_u8 

914 and fallback_ascii 

915 and fallback_u8.fingerprint != fallback_ascii.fingerprint 

916 ) 

917 or (fallback_u8 is not None) 

918 ): 

919 logger.debug("Encoding detection: utf_8 will be used as a fallback match") 

920 results.append(fallback_u8) 

921 elif fallback_ascii: 

922 logger.debug("Encoding detection: ascii will be used as a fallback match") 

923 results.append(fallback_ascii) 

924 

925 if results: 

926 logger.debug( 

927 "Encoding detection: Found %s as plausible (best-candidate) for content. With %i alternatives.", 

928 results.best().encoding, # type: ignore 

929 len(results) - 1, 

930 ) 

931 else: 

932 logger.debug("Encoding detection: Unable to determine any suitable charset.") 

933 

934 if explain: 

935 logger.removeHandler(explain_handler) 

936 logger.setLevel(previous_logger_level) 

937 

938 return results 

939 

940 

941def from_fp( 

942 fp: BinaryIO, 

943 steps: int = 5, 

944 chunk_size: int = 512, 

945 threshold: float = 0.20, 

946 cp_isolation: list[str] | None = None, 

947 cp_exclusion: list[str] | None = None, 

948 preemptive_behaviour: bool = True, 

949 explain: bool = False, 

950 language_threshold: float = 0.1, 

951 enable_fallback: bool = True, 

952) -> CharsetMatches: 

953 """ 

954 Same thing than the function from_bytes but using a file pointer that is already ready. 

955 Will not close the file pointer. 

956 """ 

957 return from_bytes( 

958 fp.read(), 

959 steps, 

960 chunk_size, 

961 threshold, 

962 cp_isolation, 

963 cp_exclusion, 

964 preemptive_behaviour, 

965 explain, 

966 language_threshold, 

967 enable_fallback, 

968 ) 

969 

970 

971def from_path( 

972 path: str | bytes | PathLike, # type: ignore[type-arg] 

973 steps: int = 5, 

974 chunk_size: int = 512, 

975 threshold: float = 0.20, 

976 cp_isolation: list[str] | None = None, 

977 cp_exclusion: list[str] | None = None, 

978 preemptive_behaviour: bool = True, 

979 explain: bool = False,