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
1"""Pipeline orchestrator — runs all detection stages in sequence.
3Note: ``from __future__ import annotations`` is intentionally omitted because
4this module is compiled with mypyc, which does not support PEP 563 string
5annotations.
6"""
8import warnings
10from chardet._utils import DEFAULT_MAX_BYTES, EVIDENCE_CAP_BYTES, decodes_without_error
11from chardet.enums import EncodingEra
12from chardet.pipeline import (
13 _NONE_RESULT,
14 DETERMINISTIC_CONFIDENCE,
15 HIGH_BYTES,
16 DetectionResult,
17 PipelineContext,
18)
19from chardet.pipeline.ascii import detect_ascii
20from chardet.pipeline.binary import is_binary
21from chardet.pipeline.bom import detect_bom
22from chardet.pipeline.escape import detect_escape_encoding
23from chardet.pipeline.language import fill_languages
24from chardet.pipeline.magic import detect_magic
25from chardet.pipeline.markup import detect_markup_charset, promote_markup_superset
26from chardet.pipeline.postprocess import postprocess_results
27from chardet.pipeline.statistical import score_candidates
28from chardet.pipeline.structural import (
29 compute_lead_byte_diversity,
30 compute_multibyte_byte_coverage,
31 compute_structural_score,
32)
33from chardet.pipeline.utf8 import scan_utf8
34from chardet.pipeline.utf1632 import detect_utf1632_patterns
35from chardet.pipeline.validity import filter_by_validity
36from chardet.registry import EncodingInfo, get_candidates
38_BINARY_RESULT = DetectionResult(
39 encoding=None,
40 confidence=DETERMINISTIC_CONFIDENCE,
41 language=None,
42 mime_type="application/octet-stream",
43)
44# Threshold at which a CJK structural score is confident enough to trigger
45# combined structural+statistical ranking rather than purely statistical.
46_STRUCTURAL_CONFIDENCE_THRESHOLD = 0.85
48# Maximum bytes used for statistical bigram scoring. Bigram models
49# converge quickly — 16 KB is sufficient for discrimination across all
50# language models (single-byte and multi-byte alike) while avoiding
51# unnecessary work on large files. Experimentally verified: 0 real
52# accuracy losses across 835 test files at this threshold.
53_STAT_SCORE_MAX_BYTES = 16384
56def _make_fallback_or_none(
57 encoding: str,
58 allowed: frozenset[str],
59 param_name: str,
60) -> list[DetectionResult]:
61 """Return a low-confidence result for *encoding*, or ``encoding=None`` if filtered out.
63 ``stacklevel=5`` targets the public caller:
64 detect() -> run_pipeline() -> _run_pipeline_core() -> _make_fallback_or_none().
65 """
66 if encoding not in allowed:
67 warnings.warn(
68 f"{param_name} {encoding!r} is excluded by "
69 f"include_encodings/exclude_encodings; returning encoding=None",
70 UserWarning,
71 stacklevel=5,
72 )
73 return [_NONE_RESULT]
74 return [DetectionResult(encoding=encoding, confidence=0.10, language=None)]
77def _hold_validity_past_cap(
78 data: bytes,
79 evidence: bytes,
80 results: list[DetectionResult],
81 allowed: frozenset[str],
82 no_match_encoding: str,
83) -> list[DetectionResult]:
84 """Keep the validity contract for the part of the window past the evidence cap.
86 Byte-validity filtering converges on the first ``EVIDENCE_CAP_BYTES``
87 of the window (ADR-0006), so on a longer window a candidate that
88 cannot decode the rest can still reach the top. What callers rely on
89 is that the answer decodes the window chardet examined: they pass
90 ``max_bytes`` precisely to say how much of the input the verdict is
91 about. So walk the corrected ranking and drop every entry ahead of
92 the first one that decodes the whole window, under the same tolerant
93 judgment the validity filter passes on the evidence slice. Those
94 entries are what validity would have removed had it seen the bytes;
95 the survivors keep their own ranks and confidences, since nothing
96 about the evidence they were scored on has changed.
98 The walk is bounded by the ranking's length, each step is one C-speed
99 decode that stops at the first bad byte, and it costs nothing at all
100 when the window fits inside the cap, which every default call does.
101 Entries below the first decodable one are not re-examined, so a
102 ``detect_all`` ranking on a long window is exact down to its winner
103 and unverified past it. When no listed entry decodes, the answer is
104 the no-match fallback, as it is when validity leaves nothing.
105 """
106 if len(data) <= len(evidence):
107 return results
108 for i, r in enumerate(results):
109 if r.encoding is not None and decodes_without_error(data, r.encoding):
110 return results if i == 0 else results[i:]
111 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
114# Minimum structural score (valid multi-byte sequences / lead bytes) required
115# to keep a CJK multi-byte candidate. Below this threshold the encoding is
116# eliminated as a false positive (e.g. Shift_JIS matching Latin data where
117# scattered high bytes look like lead bytes but rarely form valid pairs).
118_CJK_MIN_MB_RATIO = 0.05
119# Minimum number of non-ASCII bytes required for a CJK candidate to survive
120# gating. Very short inputs are validated by the other gates (structural
121# pair ratio, byte coverage) and by coverage-aware boosting in statistical
122# scoring — so we keep this threshold low to let even 1-character CJK
123# inputs compete.
124_CJK_MIN_NON_ASCII = 2
125# Minimum ratio of non-ASCII bytes that must participate in valid multi-byte
126# sequences for a CJK candidate to survive gating. Genuine CJK text has
127# nearly all non-ASCII bytes in valid pairs (coverage >= 0.95); Latin text
128# with scattered high bytes has many orphan bytes (coverage often < 0.5).
129# The lowest true-positive coverage in the test suite is ~0.39 (a CP932 HTML
130# file with many half-width katakana).
131_CJK_MIN_BYTE_COVERAGE = 0.35
132# Minimum number of distinct lead byte values for a CJK candidate to
133# survive gating. Genuine CJK text uses a wide range of lead bytes;
134# European false positives cluster in a narrow band. Only applied when
135# there are enough non-ASCII bytes to expect diversity (see
136# _CJK_DIVERSITY_MIN_NON_ASCII).
137_CJK_MIN_LEAD_DIVERSITY = 4
138# Minimum non-ASCII byte count before applying the lead diversity gate.
139# Very small files (e.g. 8 non-ASCII bytes) may genuinely have low
140# diversity even for real CJK text (e.g. repeated katakana).
141_CJK_DIVERSITY_MIN_NON_ASCII = 16
144def _gate_cjk_candidates(
145 data: bytes,
146 valid_candidates: tuple[EncodingInfo, ...],
147 ctx: PipelineContext,
148) -> tuple[EncodingInfo, ...]:
149 """Eliminate CJK multi-byte candidates that lack genuine multi-byte structure.
151 Four checks are applied in order to each multi-byte candidate:
153 1. **Structural pair ratio** (valid_pairs / lead_bytes) must be
154 >= ``_CJK_MIN_MB_RATIO``. Catches files with many orphan lead bytes.
156 2. **Minimum non-ASCII byte count**: the data must contain at least
157 ``_CJK_MIN_NON_ASCII`` bytes > 0x7F. Tiny files with 1-5 high bytes
158 can accidentally form perfect pairs and score 1.0 structurally.
160 3. **Byte coverage** (non-ASCII bytes in valid multi-byte sequences /
161 total non-ASCII bytes) must be >= ``_CJK_MIN_BYTE_COVERAGE``. Latin
162 text has many high bytes that are NOT consumed by multi-byte pairs;
163 genuine CJK text has nearly all high bytes accounted for.
165 4. **Lead byte diversity**: the number of distinct lead byte values in
166 valid pairs must be >= ``_CJK_MIN_LEAD_DIVERSITY``. Genuine CJK text
167 draws from a wide repertoire of lead bytes; European false positives
168 cluster in a narrow band (e.g. 0xC0-0xDF for accented Latin).
170 Returns the filtered candidate list. Structural scores are cached in
171 ``ctx.mb_scores`` for reuse in Stage 2b.
172 """
173 gated: list[EncodingInfo] = []
174 for enc in valid_candidates:
175 if enc.is_multibyte:
176 mb_score = compute_structural_score(data, enc, ctx)
177 ctx.mb_scores[enc.name] = mb_score
178 if mb_score < _CJK_MIN_MB_RATIO:
179 continue # No multi-byte structure -> eliminate
180 if ctx.non_ascii_count is None:
181 ctx.non_ascii_count = len(data) - len(data.translate(None, HIGH_BYTES))
182 if ctx.non_ascii_count < _CJK_MIN_NON_ASCII:
183 continue # Too few high bytes to trust the score
184 byte_coverage = compute_multibyte_byte_coverage(
185 data, enc, ctx, non_ascii_count=ctx.non_ascii_count
186 )
187 ctx.mb_coverage[enc.name] = byte_coverage
188 if byte_coverage < _CJK_MIN_BYTE_COVERAGE:
189 continue # Most high bytes are orphans -> not CJK
190 if ctx.non_ascii_count >= _CJK_DIVERSITY_MIN_NON_ASCII:
191 lead_diversity = compute_lead_byte_diversity(data, enc, ctx)
192 if lead_diversity < _CJK_MIN_LEAD_DIVERSITY:
193 continue # Too few distinct lead bytes -> not CJK
194 gated.append(enc)
195 return tuple(gated)
198def _score_structural_candidates(
199 data: bytes,
200 structural_scores: list[tuple[str, float]],
201 valid_candidates: tuple[EncodingInfo, ...],
202 ctx: PipelineContext,
203 *,
204 full_ranking: bool = False,
205) -> list[DetectionResult]:
206 """Score structurally-valid CJK candidates using statistical bigrams.
208 When multiple CJK encodings score equally high structurally, statistical
209 scoring differentiates them (e.g. euc-jp vs big5 for Japanese data).
210 Single-byte candidates are also scored and included so that the caller
211 can compare CJK vs single-byte confidence.
213 Multi-byte candidates with high byte coverage (>= 0.95) receive a
214 confidence boost proportional to coverage. When nearly all non-ASCII
215 bytes form valid multi-byte pairs, the structural evidence is strong
216 and should increase the candidate's ranking relative to single-byte
217 alternatives whose bigram models may score higher on small samples.
219 Note: boosted confidence values may exceed 1.0 and are used only for
220 relative ranking among candidates. ``run_pipeline`` clamps all
221 confidence values to [0.0, 1.0] before returning to callers.
222 """
223 enc_lookup: dict[str, EncodingInfo] = {
224 e.name: e for e in valid_candidates if e.is_multibyte
225 }
226 valid_mb = tuple(
227 enc_lookup[name] for name, _sc in structural_scores if name in enc_lookup
228 )
229 single_byte = tuple(e for e in valid_candidates if not e.is_multibyte)
230 results = list(
231 score_candidates(
232 data[:_STAT_SCORE_MAX_BYTES],
233 (*valid_mb, *single_byte),
234 full_ranking=full_ranking,
235 )
236 )
238 # Boost multi-byte candidates with high byte coverage.
239 boosted: list[DetectionResult] = []
240 for r in results:
241 coverage = ctx.mb_coverage.get(r.encoding, 0.0) if r.encoding else 0.0
242 if coverage >= 0.95:
243 boosted.append(
244 DetectionResult(
245 r.encoding, r.confidence * (1 + coverage), r.language, r.mime_type
246 )
247 )
248 else:
249 boosted.append(r)
250 boosted.sort(key=lambda x: x.confidence, reverse=True)
251 return boosted
254def _with_default_mime(result: DetectionResult) -> DetectionResult:
255 """Default ``mime_type`` to ``text/plain`` (text) or ``application/octet-stream`` (binary)."""
256 if result.mime_type is not None:
257 return result
258 mime = "text/plain" if result.encoding is not None else "application/octet-stream"
259 return DetectionResult(result.encoding, result.confidence, result.language, mime)
262def _run_pipeline_core( # noqa: PLR0913
263 data: bytes,
264 encoding_era: EncodingEra,
265 max_bytes: int = DEFAULT_MAX_BYTES,
266 *,
267 include_encodings: frozenset[str] | None = None,
268 exclude_encodings: frozenset[str] | None = None,
269 no_match_encoding: str = "cp1252",
270 empty_input_encoding: str = "utf-8",
271 full_ranking: bool = False,
272 input_truncated: bool = False,
273) -> list[DetectionResult]:
274 """Core pipeline logic. Returns list of results sorted by confidence."""
275 ctx = PipelineContext()
276 input_truncated = input_truncated or len(data) > max_bytes
277 data = data[:max_bytes]
279 # Build candidate set once — used for both early-exit gating and
280 # statistical scoring. The set incorporates encoding_era, include, and
281 # exclude filters so all pipeline stages are gated consistently.
282 candidates = get_candidates(encoding_era, include_encodings, exclude_encodings)
283 allowed: frozenset[str] = frozenset(enc.name for enc in candidates)
285 if not data:
286 return _make_fallback_or_none(
287 empty_input_encoding, allowed, "empty_input_encoding"
288 )
290 # Stage 1a: BOM detection (runs first — BOMs are definitive and
291 # UTF-16/32 data looks binary due to null bytes)
292 bom_result = detect_bom(data)
293 if bom_result is not None and bom_result.encoding in allowed:
294 return [bom_result]
296 # Stage 1a+: UTF-16/32 null-byte pattern detection (for files without
297 # BOMs — must run before binary detection since these encodings contain
298 # many null bytes that would trigger the binary check)
299 utf1632_result = detect_utf1632_patterns(data)
300 if utf1632_result is not None and utf1632_result.encoding in allowed:
301 return [utf1632_result]
303 # Escape-sequence encodings (ISO-2022, HZ-GB-2312, UTF-7): must run
304 # before binary detection (ESC is a control byte) and before ASCII
305 # detection (HZ-GB-2312 uses only printable ASCII plus tildes).
306 escape_result = detect_escape_encoding(data)
307 if (
308 escape_result is not None
309 and escape_result.encoding is not None
310 and escape_result.encoding in allowed
311 ):
312 return [escape_result]
314 # Magic number detection for known binary formats — runs before
315 # UTF-8/ASCII prechecks to avoid unnecessary analysis on binary data.
316 magic_result = detect_magic(data)
317 if magic_result is not None:
318 return [magic_result]
320 # Pre-check UTF-8 to prevent false binary classification. Valid UTF-8
321 # with multi-byte sequences can contain control bytes (e.g. ESC for ANSI
322 # codes) that would otherwise exceed the binary threshold. We compute
323 # the result now but return it at the normal pipeline position (after
324 # markup) so that explicit charset declarations still take precedence.
325 utf8_valid, utf8_precheck = scan_utf8(data)
327 # Pre-check ASCII to prevent false binary classification. ASCII text
328 # with null byte separators (e.g. find -print0 output) would exceed the
329 # binary threshold due to the null bytes. Like the UTF-8 precheck, we
330 # compute the result now but return it at the normal position (after
331 # markup) so explicit charset declarations still take precedence.
332 ascii_precheck = detect_ascii(data)
334 # Stage 0: Binary detection (skip when data is valid UTF-8 or ASCII)
335 # Binary detection (encoding=None) is NOT gated by filters.
336 if (
337 utf8_precheck is None
338 and ascii_precheck is None
339 and is_binary(data, max_bytes=max_bytes)
340 ):
341 return [_BINARY_RESULT]
343 # Stage 1b: Markup charset extraction (before ASCII/UTF-8 so explicit
344 # declarations like <?xml encoding="iso-8859-1"?> are honoured even
345 # when the bytes happen to be pure ASCII).
346 markup_result = detect_markup_charset(data)
347 if markup_result is not None and markup_result.encoding in allowed:
348 # A declaration is honoured over pure ASCII (the declared encoding
349 # decodes those bytes identically), but not over genuine UTF-8
350 # structure: data containing valid multi-byte UTF-8 sequences is
351 # UTF-8 regardless of what the (frequently stale) declaration
352 # claims, and decoding it as the declared encoding would produce
353 # mojibake. Keep the markup mime type; only the encoding wins.
354 if (
355 utf8_precheck is not None
356 and utf8_precheck.encoding != markup_result.encoding
357 and utf8_precheck.encoding in allowed
358 ):
359 return [
360 DetectionResult(
361 utf8_precheck.encoding,
362 utf8_precheck.confidence,
363 utf8_precheck.language,
364 markup_result.mime_type,
365 )
366 ]
367 markup_result = promote_markup_superset(data, markup_result, allowed)
368 return [markup_result]
370 # Stage 1c: ASCII (use pre-computed result)
371 if ascii_precheck is not None and ascii_precheck.encoding in allowed:
372 return [ascii_precheck]
374 # Stage 1d: UTF-8 structural validation (use pre-computed result)
375 if utf8_precheck is not None and utf8_precheck.encoding in allowed:
376 return [utf8_precheck]
378 # The filtering, gating, and probing stages below converge on bounded
379 # evidence (ADR-0006). One slice, taken here, feeds them all: the
380 # structural analysis cache is keyed by encoding name only, so every
381 # consumer must see the same view of the data. For the decode-safety
382 # flip in postprocess, a window that outruns the cap counts as
383 # truncation — the slice's tail is a chardet-made cut, not the end of
384 # what the caller will decode.
385 evidence = data[:EVIDENCE_CAP_BYTES]
386 evidence_truncated = input_truncated or len(data) > len(evidence)
388 # Stage 2a: Byte validity filtering
389 valid_candidates = filter_by_validity(evidence, candidates)
391 # The exhaustive UTF-8 check saw the whole window; when it *rejected*
392 # the data but the window extends past the evidence cap, the slice
393 # alone may still look like valid UTF-8 to the statistical path. Honor
394 # the proof: chardet never calls data UTF-8 that is not valid UTF-8
395 # throughout the window. A missing precheck result is not a rejection
396 # — valid UTF-8 with no multi-byte evidence (pure ASCII, ASCII plus
397 # control bytes, ASCII plus a truncated tail) also returns None, and
398 # ruling utf-8 out for those would be wrong. (Within the cap the slice
399 # is the window, so validity already agrees and this never fires.)
400 if not utf8_valid and len(data) > len(evidence):
401 valid_candidates = tuple(e for e in valid_candidates if e.name != "utf-8")
403 if not valid_candidates:
404 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
406 # Gate: eliminate CJK multi-byte candidates that lack genuine
407 # multi-byte structure. Cache structural scores for Stage 2b.
408 valid_candidates = _gate_cjk_candidates(evidence, valid_candidates, ctx)
410 if not valid_candidates:
411 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
413 # Stage 2b: Structural probing for multi-byte encodings
414 # Reuse scores already computed during the CJK gate above.
415 structural_scores: list[tuple[str, float]] = []
416 for enc in valid_candidates:
417 if enc.is_multibyte:
418 score = ctx.mb_scores.get(enc.name)
419 if score is None: # pragma: no cover - gate always populates cache
420 score = compute_structural_score(evidence, enc, ctx)
421 if score > 0.0:
422 structural_scores.append((enc.name, score))
424 # If a multi-byte encoding scored very high, score all candidates
425 # (CJK + single-byte) statistically.
426 if structural_scores:
427 structural_scores.sort(key=lambda x: x[1], reverse=True)
428 _, best_score = structural_scores[0]
429 if best_score >= _STRUCTURAL_CONFIDENCE_THRESHOLD:
430 results = _score_structural_candidates(
431 evidence,
432 structural_scores,
433 valid_candidates,
434 ctx,
435 full_ranking=full_ranking,
436 )
437 if results:
438 results = postprocess_results(
439 evidence, results, input_truncated=evidence_truncated
440 )
441 return _hold_validity_past_cap(
442 data, evidence, results, allowed, no_match_encoding
443 )
445 # Stage 3: Statistical scoring for all remaining candidates.
446 # Bigram models converge quickly and don't benefit from scanning
447 # beyond 16 KB — cap the data to avoid unnecessary work on large files.
448 stat_data = evidence[:_STAT_SCORE_MAX_BYTES]
449 results = list(
450 score_candidates(stat_data, tuple(valid_candidates), full_ranking=full_ranking)
451 )
452 if not results:
453 return _make_fallback_or_none(no_match_encoding, allowed, "no_match_encoding")
455 # Rank corrections reason about the same evidence window the ranking
456 # came from, which also bounds their byte-presence scans.
457 results = postprocess_results(evidence, results, input_truncated=evidence_truncated)
458 return _hold_validity_past_cap(data, evidence, results, allowed, no_match_encoding)
461def run_pipeline( # noqa: PLR0913
462 data: bytes,
463 encoding_era: EncodingEra,
464 max_bytes: int = DEFAULT_MAX_BYTES,
465 *,
466 include_encodings: frozenset[str] | None = None,
467 exclude_encodings: frozenset[str] | None = None,
468 no_match_encoding: str = "cp1252",
469 empty_input_encoding: str = "utf-8",
470 full_ranking: bool = False,
471 input_truncated: bool = False,
472) -> list[DetectionResult]:
473 """Run the full detection pipeline.
475 :param data: The raw byte data to analyze.
476 :param encoding_era: Filter candidates to a specific era of encodings.
477 :param max_bytes: Maximum number of bytes to process.
478 :param include_encodings: If not ``None``, only return these encodings.
479 :param exclude_encodings: If not ``None``, never return these encodings.
480 :param no_match_encoding: Encoding returned when no candidate survives.
481 :param empty_input_encoding: Encoding returned for empty input.
482 :param full_ranking: When ``True``, statistical scoring evaluates every
483 candidate so the returned list is complete (``detect_all`` needs
484 this). When ``False``, candidates that provably cannot affect the
485 top of the ranking may be omitted from the tail.
486 :param input_truncated: Pass ``True`` when *data* is already a truncated
487 view of the caller's input (``UniversalDetector`` caps its buffer at
488 ``max_bytes``, which this function cannot see from ``len(data)``
489 alone). Truncation by the ``max_bytes`` slice here is detected
490 either way; the flag only ever widens it.
491 :returns: A list of :class:`DetectionResult` sorted by confidence descending.
492 """
493 results = _run_pipeline_core(
494 data,
495 encoding_era,
496 max_bytes,
497 input_truncated=input_truncated,
498 include_encodings=include_encodings,
499 exclude_encodings=exclude_encodings,
500 no_match_encoding=no_match_encoding,
501 empty_input_encoding=empty_input_encoding,
502 full_ranking=full_ranking,
503 )
504 # Same slice the core ran on: ``max_bytes`` is the caller's cap on what
505 # chardet may look at, and language fill is not exempt from it just
506 # because it applies a smaller cap of its own.
507 results = fill_languages(data[:max_bytes], results)
508 # The ANSI-art model is keyed under the "zxx" pseudo-language (ISO 639
509 # for "no linguistic content"). Kept internal so language fill does not
510 # overwrite it; callers see language=None.
511 results = [
512 DetectionResult(r.encoding, r.confidence, None, r.mime_type)
513 if r.language == "zxx"
514 else r
515 for r in results
516 ]
517 results = [_with_default_mime(r) for r in results]
518 if not results: # pragma: no cover
519 msg = "pipeline must always return at least one result"
520 raise RuntimeError(msg)
521 # Clamp confidence to [0.0, 1.0] at the public API boundary. Internal
522 # stages may boost confidence above 1.0 for ranking purposes (e.g.
523 # CJK byte-coverage boost), but callers expect a probability-like value.
524 return [
525 DetectionResult(r.encoding, min(r.confidence, 1.0), r.language, r.mime_type)
526 if r.confidence > 1.0
527 else r
528 for r in results
529 ]